From 3c3f7feef8cd622b72c38a73d72a6377352186b7 Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Sun, 18 Oct 2026 01:58:38 +0800 Subject: [PATCH] fix: avoid redundant checkpoint packing reads --- CHANGELOG.md | 2 + docs/guides/private-checkpoints.md | 8 +- scripts/private_discord_checkpoint.py | 94 +++++++----- .../test_private_discord_checkpoint_save.py | 134 +++++++++++++++++- 4 files changed, 202 insertions(+), 36 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0300efc4..abd95b0f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- Avoid redundant checkpoint packing reads by hashing the existing byte streams, and separate transfer and restore verification timings while retaining full authenticated disk and SQLite checks. + - Pin the reviewed private acquisition policy for manual runs and prepare member refresh within the existing shared acquisition deadline; automatic cadence remains disabled pending measured publication qualification. - Let targeted Discord syncs opt into stored direct threads of selected text and announcement roots, so recurring bounded acquisition covers newly discovered threads without changing ordinary channel selection. - Bound checkpoint saves by one shared deadline, report safe phase timings, and remove a redundant integrity scan while preserving authenticated full-state verification. diff --git a/docs/guides/private-checkpoints.md b/docs/guides/private-checkpoints.md index 2cd71ba0..92de4ffe 100644 --- a/docs/guides/private-checkpoints.md +++ b/docs/guides/private-checkpoints.md @@ -39,7 +39,13 @@ Ordinary saves share one 840-second deadline inside the unchanged 15-minute step. Source integrity, backup, compression, hashing, transfer and authenticated readback use that deadline. The intermediate backup is not scanned again: its hash and pinned source facts are both checked against the downloaded image. -Fixed phase names and elapsed times identify a slow phase without exposing data. +Packing carries the database hash from its raw input traversal and compressed +part hashes from successful output writes, including the gzip trailer. It does +not reread those local files only to hash them. The authenticated restored image +still receives an independent disk hash and full SQLite checks. Fixed phase +names and elapsed times separate download, reconstruction, disk verification +and SQLite verification without exposing data. These reductions do not establish +publication headroom until a complete run is measured. Promotion requires at least one minute remaining. Deadline checks are cooperative for filesystem and library calls; the outer minute is cleanup margin, not a hard completion guarantee. Any error after an attempted mutation holds the retained diff --git a/scripts/private_discord_checkpoint.py b/scripts/private_discord_checkpoint.py index db32c0e2..9a66308b 100644 --- a/scripts/private_discord_checkpoint.py +++ b/scripts/private_discord_checkpoint.py @@ -37,7 +37,8 @@ MAX_POLICY = 64 * 1024 MAX_TARGETS = 4096 SAVE_TIMEOUT = 840 -SAVE_PHASES = {"source_facts", "backup", "pack", "upload", "restore_verify", "promote"} +CHECKPOINT_PHASES = {"source_facts", "backup", "pack", "upload", "restore_verify", "promote", + "download", "reconstruct", "disk_verify", "sqlite_verify"} API_PHASES = {"api_read", "api_write", "repository_read", "archive_ref_read", "release_list", "latest_read", "release_read", "asset_list", "manifest_download", "part_download", "release_create", "asset_upload", "release_promote", "release_delete", "source_ref_read", "source_run_read", @@ -67,15 +68,15 @@ def check_deadline(deadline): @contextlib.contextmanager -def save_phase(name): - require(name in SAVE_PHASES, "checkpoint_phase") +def checkpoint_phase(name): + require(name in CHECKPOINT_PHASES, "checkpoint_phase") started, state = time.monotonic(), "failed" - print(f"checkpoint save: phase={name} state=begin", flush=True) + print(f"checkpoint: phase={name} state=begin", flush=True) try: yield state = "complete" finally: - print(f"checkpoint save: phase={name} state={state} elapsed_seconds={time.monotonic() - started:.3f}", flush=True) + print(f"checkpoint: phase={name} state={state} elapsed_seconds={time.monotonic() - started:.3f}", flush=True) def file_digest(path, deadline=None): @@ -167,9 +168,9 @@ def progress(_status, _remaining, _total): # The main file can be tiny while committed pages are still in WAL. # Reserve the standalone image and worst-case compressed parts together. require(shutil.disk_usage(destination.parent).free > logical_bytes * 2 + BLOCK * MAX_PARTS, "checkpoint_capacity") - with save_phase("source_facts"): + with checkpoint_phase("source_facts"): facts = database_facts(src, deadline) - with save_phase("backup"): + with checkpoint_phase("backup"): with contextlib.closing(sqlite3.connect(destination)) as out: src.backup(out, pages=256, progress=progress, sleep=0.05) out.execute("PRAGMA journal_mode=DELETE") @@ -180,8 +181,32 @@ def progress(_status, _remaining, _total): return facts +class HashingWriter: + # gzip writes its header and closing trailer through the same sink. Hash + # only accepted bytes; a short or failed write must never mint a descriptor. + def __init__(self, output, deadline): + self.output, self.deadline = output, deadline + self.sha256, self.size = hashlib.sha256(), 0 + + def write(self, data): + remaining = memoryview(data) + require(self.size + len(remaining) <= MAX_PART, "part_size") + while remaining: + check_deadline(self.deadline) + written = self.output.write(remaining) + require(type(written) is int and 0 < written <= len(remaining), "part_write") + self.sha256.update(remaining[:written]) + self.size += written + remaining = remaining[written:] + return len(data) + + def flush(self): + check_deadline(self.deadline) + self.output.flush() + + def pack(database, directory, deadline=None): - parts = [] + parts, database_hash, database_bytes = [], hashlib.sha256(), 0 with open(database, "rb") as source: while source.tell() < database.stat().st_size: check_deadline(deadline) @@ -189,19 +214,22 @@ def pack(database, directory, deadline=None): name = f"state-{len(parts):03d}.sqlite.gz" target = directory / name raw_bytes = 0 - with target.open("xb") as raw, gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as compressed: - while raw_bytes < RAW_PART: - check_deadline(deadline) - chunk = source.read(min(BLOCK, RAW_PART - raw_bytes)) - if not chunk: - break - compressed.write(chunk) - raw_bytes += len(chunk) - size = target.stat().st_size - require(0 < size <= MAX_PART, "part_size") - parts.append({"name": name, "bytes": size, "raw_bytes": raw_bytes, "sha256": file_digest(target, deadline)}) + with target.open("xb") as raw: + sink = HashingWriter(raw, deadline) + with gzip.GzipFile(filename="", mode="wb", fileobj=sink, mtime=0) as compressed: + while raw_bytes < RAW_PART: + check_deadline(deadline) + chunk = source.read(min(BLOCK, RAW_PART - raw_bytes)) + if not chunk: + break + compressed.write(chunk) + database_hash.update(chunk) + raw_bytes += len(chunk) + require(0 < sink.size == target.stat().st_size <= MAX_PART, "part_size") + database_bytes += raw_bytes + parts.append({"name": name, "bytes": sink.size, "raw_bytes": raw_bytes, "sha256": sink.sha256.hexdigest()}) check_deadline(deadline) - return parts + return {"bytes": database_bytes, "sha256": database_hash.hexdigest()}, parts def validate_manifest(value): @@ -248,7 +276,7 @@ def validate_manifest(value): def unpack(manifest, directory, destination, deadline=None): check_deadline(deadline) - with destination.open("xb") as output: + with checkpoint_phase("reconstruct"), destination.open("xb") as output: for part in manifest["parts"]: file = directory / part["name"] require(file.stat().st_size == part["bytes"] and file_digest(file, deadline) == part["sha256"], "part_checksum") @@ -263,8 +291,9 @@ def unpack(manifest, directory, destination, deadline=None): output.flush() os.fsync(output.fileno()) expected = manifest["database"] - require(destination.stat().st_size == expected["bytes"] and file_digest(destination, deadline) == expected["sha256"], "database_checksum") - with contextlib.closing(readonly(destination)) as db: + with checkpoint_phase("disk_verify"): + require(destination.stat().st_size == expected["bytes"] and file_digest(destination, deadline) == expected["sha256"], "database_checksum") + with checkpoint_phase("sqlite_verify"), contextlib.closing(readonly(destination)) as db: require(database_facts(db, deadline) == {k: expected[k] for k in ("user_version", "schema_sha256", "tables")}, "restore_parity") @@ -468,9 +497,10 @@ def download_database(github, release, directory, draft=False, deadline=None): manifest, assets, ref = release_manifest(github, release, directory, draft) require(shutil.disk_usage(directory).free > manifest["database"]["bytes"] + sum(p["bytes"] for p in manifest["parts"]), "checkpoint_capacity") - for part in manifest["parts"]: - check_deadline(deadline) - github.download(assets[part["name"]], directory / part["name"]) + with checkpoint_phase("download"): + for part in manifest["parts"]: + check_deadline(deadline) + github.download(assets[part["name"]], directory / part["name"]) unpack(manifest, directory, directory / "restored.sqlite", deadline) inventory = [{"name": name, "id": asset["id"], "bytes": asset["size"], "digest": asset["digest"]} for name, asset in sorted(assets.items())] @@ -587,17 +617,17 @@ def save(github, database, receipt, outcome, migration=None, deadline=None): source = source_identity(outcome, archive_commit) standalone = directory / "checkpoint.sqlite" facts = snapshot(database, standalone, deadline=deadline) - with save_phase("pack"): + with checkpoint_phase("pack"): + packed_database, parts = pack(standalone, directory, deadline) manifest = {"schema": SCHEMA, "boundary": "native-ingestion-before-publication", "repository": github.identity, "source": source, "parent": parent["reference"] if parent else None, "migration": migration, - "database": {**facts, "bytes": standalone.stat().st_size, "sha256": file_digest(standalone, deadline)}, - "parts": pack(standalone, directory, deadline)} + "database": {**facts, **packed_database}, "parts": parts} validate_manifest(manifest) (directory / MANIFEST).write_bytes(canonical(manifest)) tag = f'discrawl-state-{source["run_id"]}-{source["run_attempt"]}' try: - with save_phase("upload"): + with checkpoint_phase("upload"): check_deadline(deadline) attempted = True draft = github.create(tag, archive_commit) @@ -609,7 +639,7 @@ def save(github, database, receipt, outcome, migration=None, deadline=None): (directory / part["name"]).unlink() verification = directory / "verification" verification.mkdir() - with save_phase("restore_verify"): + with checkpoint_phase("restore_verify"): current = download_database(github, github.api('/releases/' + str(draft["id"])), verification, draft=True, deadline=deadline) require(current["manifest"] == manifest, "uploaded_manifest") @@ -619,7 +649,7 @@ def save(github, database, receipt, outcome, migration=None, deadline=None): require(read_parent(github, receipt, recheck) == parent, "parent_changed") else: require(github.api("/releases/latest", missing=True) is None, "parent_changed") - with save_phase("promote"): + with checkpoint_phase("promote"): require(deadline - time.monotonic() >= 60, "checkpoint_publication_budget") published = github.api('/releases/' + str(draft["id"]), {"draft": False, "make_latest": "true"}) require(published.get("id") == draft["id"] and published.get("draft") is False diff --git a/scripts/test_private_discord_checkpoint_save.py b/scripts/test_private_discord_checkpoint_save.py index d9815ee6..b2af0407 100644 --- a/scripts/test_private_discord_checkpoint_save.py +++ b/scripts/test_private_discord_checkpoint_save.py @@ -1,5 +1,7 @@ import contextlib +import gzip import io +from pathlib import Path import sqlite3 import sys import unittest @@ -22,7 +24,7 @@ def test_corrupted_snapshot_is_rejected_after_authentication_without_promotion(s original = checkpoint.snapshot def corrupt(*args, **kwargs): facts = original(*args, **kwargs) - with sqlite3.connect(args[1]) as db: + with contextlib.closing(sqlite3.connect(args[1])) as db, db: db.execute("UPDATE messages SET value='different retained bytes'") db.execute("INSERT INTO messages VALUES('extra','unexpected row')") return facts @@ -107,11 +109,137 @@ def test_phase_output_contains_only_fixed_labels_and_elapsed_time(self): with contextlib.redirect_stdout(output): self.seed() for line in output.getvalue().splitlines(): - self.assertRegex(line, r"^checkpoint save: phase=(source_facts|backup|pack|upload|restore_verify|promote) state=(begin|complete)( elapsed_seconds=[0-9.]+)?$") + self.assertRegex(line, r"^checkpoint: phase=(source_facts|backup|pack|upload|restore_verify|promote|download|reconstruct|disk_verify|sqlite_verify) state=(begin|complete)( elapsed_seconds=[0-9.]+)?$") + for phase in ("download", "reconstruct", "disk_verify", "sqlite_verify"): + self.assertIn(f"phase={phase} state=complete", output.getvalue()) with self.assertRaisesRegex(checkpoint.CheckpointError, "checkpoint_phase"): - with checkpoint.save_phase("private-secret"): + with checkpoint.checkpoint_phase("private-secret"): pass + def test_pack_carries_exact_disk_hashes_and_unchanged_gzip_bytes_without_rereads(self): + for size in (0, 1, 17, 18, 34, 39): + with self.subTest(size=size): + directory = self.root / str(size) + directory.mkdir() + source = directory / "input" + data = bytes(range(size)) + source.write_bytes(data) + with patch.object(checkpoint, "RAW_PART", 17), patch.object(checkpoint, "BLOCK", 5), \ + patch.object(checkpoint, "file_digest", side_effect=AssertionError("redundant hash read")): + database, parts = checkpoint.pack(source, directory) + self.assertEqual(database, {"bytes": size, "sha256": checkpoint.digest(data)}) + self.assertEqual(len(parts), (size + 16) // 17) + for index, part in enumerate(parts): + expected = io.BytesIO() + chunk = data[index * 17:(index + 1) * 17] + with gzip.GzipFile(filename="", mode="wb", fileobj=expected, mtime=0) as compressed: + for start in range(0, len(chunk), 5): + compressed.write(chunk[start:start + 5]) + actual = (directory / part["name"]).read_bytes() + self.assertEqual(actual, expected.getvalue()) + self.assertEqual(gzip.decompress(actual), chunk) + self.assertEqual(part, {"name": f"state-{index:03d}.sqlite.gz", "bytes": len(actual), + "raw_bytes": len(chunk), "sha256": checkpoint.digest(actual)}) + + def test_short_writes_include_header_flush_and_trailer_in_carried_hash(self): + class ShortWriter(io.BytesIO): + def write(self, data): + return super().write(data[:3]) + raw = ShortWriter() + sink = checkpoint.HashingWriter(raw, None) + with gzip.GzipFile(filename="", mode="wb", fileobj=sink, mtime=0) as compressed: + compressed.write(b"first native bytes") + compressed.flush() + compressed.write(b"second native bytes") + actual = raw.getvalue() + self.assertEqual(gzip.decompress(actual), b"first native bytessecond native bytes") + self.assertEqual(sink.size, len(actual)) + self.assertEqual(sink.sha256.hexdigest(), checkpoint.digest(actual)) + + def test_invalid_write_counts_never_hash_unaccepted_bytes(self): + for count in (0, None, -1, 5, True, 1.5): + with self.subTest(count=count): + class InvalidWriter: + def write(self, _data): + return count + sink = checkpoint.HashingWriter(InvalidWriter(), None) + with self.assertRaisesRegex(checkpoint.CheckpointError, "part_write"): + sink.write(b"data") + self.assertEqual(sink.size, 0) + self.assertEqual(sink.sha256.hexdigest(), checkpoint.digest(b"")) + + def test_partial_write_failure_hashes_only_the_accepted_prefix(self): + class FailedWriter(io.BytesIO): + def write(self, data): + if self.tell(): + raise OSError("write failed") + return super().write(data[:2]) + raw = FailedWriter() + sink = checkpoint.HashingWriter(raw, None) + self.assertEqual(sink.write(b""), 0) + self.assertEqual(raw.getvalue(), b"") + with self.assertRaisesRegex(OSError, "write failed"): + sink.write(b"data") + self.assertEqual(raw.getvalue(), b"da") + self.assertEqual(sink.size, 2) + self.assertEqual(sink.sha256.hexdigest(), checkpoint.digest(b"da")) + + def test_sink_deadline_and_capacity_refuse_before_write(self): + raw = io.BytesIO() + sink = checkpoint.HashingWriter(raw, 100) + with patch.object(checkpoint.time, "monotonic", return_value=101): + with self.assertRaisesRegex(checkpoint.CheckpointError, "checkpoint_timeout"): + sink.write(b"data") + with patch.object(checkpoint, "MAX_PART", 3): + with self.assertRaisesRegex(checkpoint.CheckpointError, "part_size"): + sink.write(b"data") + self.assertEqual(raw.getvalue(), b"") + + def test_pack_hash_failure_never_creates_a_draft(self): + parent = self.seed() + class FailedHash: + def update(self, _data): + raise OSError("hash failed") + original_pack, original_hash = checkpoint.pack, checkpoint.hashlib.sha256 + for failed in (0, 1): + def broken_pack(*args, **kwargs): + hashes = [original_hash(), original_hash()] + hashes[failed] = FailedHash() + with patch.object(checkpoint.hashlib, "sha256", side_effect=hashes): + return original_pack(*args, **kwargs) + with self.subTest(hash=failed), patch.object(checkpoint, "pack", broken_pack): + with self.assertRaisesRegex(OSError, "hash failed"): + checkpoint.save(self.github, self.path, self.receipt, "success") + self.assertEqual(self.github.latest, parent["release_id"]) + self.assertEqual(len(self.github.releases), 1) + + def test_gzip_or_file_close_failure_never_creates_a_draft(self): + parent = self.seed() + gzip_close, path_open = checkpoint.gzip.GzipFile.close, Path.open + def failed_gzip_close(compressed): + opened = compressed.fileobj is not None + gzip_close(compressed) + if opened: + raise OSError("gzip close failed") + class FailedClose: + def __init__(self, output): + self.output = output + def __enter__(self): + return self.output + def __exit__(self, *args): + self.output.__exit__(*args) + raise OSError("file close failed") + def failed_file_close(path, *args, **kwargs): + output = path_open(path, *args, **kwargs) + return FailedClose(output) if path.name.startswith("state-") and args == ("xb",) else output + for owner, name, replacement in ((checkpoint.gzip.GzipFile, "close", failed_gzip_close), + (Path, "open", failed_file_close)): + with self.subTest(close=name), patch.object(owner, name, replacement): + with self.assertRaisesRegex(OSError, "close failed"): + checkpoint.save(self.github, self.path, self.receipt, "success") + self.assertEqual(self.github.latest, parent["release_id"]) + self.assertEqual(len(self.github.releases), 1) + def test_cli_deadline_starts_before_private_admission(self): self.seed() completion = self.root / "writer.json"