diff --git a/devtools/archive_schema_fast_forward.py b/devtools/archive_schema_fast_forward.py index c26a640434..41f8f9ef38 100644 --- a/devtools/archive_schema_fast_forward.py +++ b/devtools/archive_schema_fast_forward.py @@ -21,6 +21,7 @@ from pathlib import Path from typing import Final, cast +from polylogue.storage.index_generation import RebuildLease, RebuildLeaseUnavailableError from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database from polylogue.storage.sqlite.archive_tiers.embeddings import EMBEDDINGS_DDL, EMBEDDINGS_SCHEMA_VERSION from polylogue.storage.sqlite.archive_tiers.index import INDEX_DDL, INDEX_SCHEMA_VERSION @@ -29,6 +30,11 @@ from polylogue.storage.sqlite.sqlite_vec_extension import try_load_sqlite_vec RECEIPT_SCHEMA: Final = "polylogue.archive-schema-fast-forward.v1" +# A smaller receipt written immediately after one index clone proof succeeds +# -- before embeddings/ops work -- so a later prepare attempt can reuse a +# fully-proven clone without re-running its table census, foreign-key check, +# and quick_check against the (potentially tens-of-GiB) clone file. +INDEX_CLONE_CHECKPOINT_SCHEMA: Final = "polylogue.archive-schema-fast-forward.index-clone-checkpoint.v1" _BEADS_ORIGIN: Final = "beads-issue" _INDEX_COPY_FORWARD_TABLES: Final = ("sessions", "session_links") _SQLITE_SIDECARS: Final = ("-wal", "-shm", "-journal") @@ -53,6 +59,7 @@ class CloneForwardResult: clone: DatabaseEvidence foreign_key_declarations_preserved: bool foreign_key_check: tuple[str, ...] + quick_check: tuple[str, ...] = () def _now_ms() -> int: @@ -146,6 +153,20 @@ def _database_evidence(path: Path) -> DatabaseEvidence: ) +def _database_evidence_from_payload(payload: dict[str, object]) -> DatabaseEvidence: + """Reconstruct ``DatabaseEvidence`` from a JSON-decoded receipt field.""" + raw_table_counts = payload.get("table_counts") + if not isinstance(raw_table_counts, dict): + raise SchemaFastForwardError("checkpoint clone evidence is missing table_counts") + return DatabaseEvidence( + path=str(payload.get("path", "")), + user_version=int(cast(int, payload.get("user_version", -1))), + sha256=str(payload.get("sha256", "")), + size_bytes=int(cast(int, payload.get("size_bytes", -1))), + table_counts={str(name): int(cast(int, count)) for name, count in raw_table_counts.items()}, + ) + + def _require_receipt_identity(payload: dict[str, object], key: str, path: Path) -> None: raw_evidence = payload.get(key) if key in {"index", "embeddings", "ops"} and isinstance(raw_evidence, dict): @@ -347,6 +368,7 @@ def fast_forward_index_clone(source: Path, destination: Path) -> CloneForwardRes require_no_beads_evidence(source) reflink_clone(source, destination) canonical_tables, canonical_indexes = _canonical_index_objects() + quick_check: tuple[str, ...] = () try: with sqlite3.connect(destination) as conn: before_fk = _foreign_key_declarations(conn) @@ -379,7 +401,7 @@ def fast_forward_index_clone(source: Path, destination: Path) -> CloneForwardRes if clone_after.table_counts != source_before.table_counts: destination.unlink(missing_ok=True) raise SchemaFastForwardError("index copy-forward changed structural row counts") - return CloneForwardResult(source_before, clone_after, True, ()) + return CloneForwardResult(source_before, clone_after, True, (), quick_check) def fast_forward_embeddings_clone(source: Path, destination: Path) -> CloneForwardResult: @@ -388,6 +410,7 @@ def fast_forward_embeddings_clone(source: Path, destination: Path) -> CloneForwa if source_before.user_version != 1: raise SchemaFastForwardError(f"embeddings clone requires v1, found v{source_before.user_version}") reflink_clone(source, destination) + quick_check: tuple[str, ...] = () try: with sqlite3.connect(destination) as conn: loaded, error = try_load_sqlite_vec(conn) @@ -415,7 +438,114 @@ def fast_forward_embeddings_clone(source: Path, destination: Path) -> CloneForwa if comparable != source_before.table_counts: destination.unlink(missing_ok=True) raise SchemaFastForwardError("embeddings clone changed existing row counts") - return CloneForwardResult(source_before, clone_after, True, ()) + return CloneForwardResult(source_before, clone_after, True, (), quick_check) + + +def _canonical_ddl_sha256() -> str: + """Cheap in-memory identity of the target index copy-forward DDL. + + Guards a checkpoint against reuse across a code change that alters the + canonical target schema between a failed attempt and its retry: no file + I/O beyond the ``:memory:`` DDL execution ``_canonical_index_objects`` + already performs. + """ + tables, indexes = _canonical_index_objects() + body = json.dumps({"tables": tables, "indexes": indexes}, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(body.encode()).hexdigest() + + +def _index_clone_checkpoint_path(clone_path: Path) -> Path: + return clone_path.with_name(f"{clone_path.name}.clone-checkpoint.json") + + +def _lightweight_database_identity(path: Path) -> tuple[str, int, int]: + """Return (sha256, size_bytes, user_version) without a table census. + + ``_database_evidence`` additionally runs ``_table_counts``, which walks + every table's rows -- exactly the duplicate work a checkpoint reuse is + meant to skip. Byte identity (sha256) is still computed in full; only the + derived SQL-level census is omitted, and only when a checkpoint already + recorded it against these same bytes. + """ + with _open_immutable_readonly(path) as conn: + _load_vec_if_required(conn) + user_version = int(conn.execute("PRAGMA user_version").fetchone()[0]) + return _sha256(path), path.stat().st_size, user_version + + +def write_index_clone_checkpoint(result: CloneForwardResult, *, clone_path: Path) -> dict[str, object]: + """Persist a self-checking index clone-proof checkpoint beside ``clone_path``. + + Called immediately after one index clone proof succeeds, before + embeddings/ops work -- so a crash before the run's final receipt still + leaves behind a fully-verified checkpoint a later ``reuse_index_clone`` + call can trust instead of re-deriving. + """ + clone_beads_findings = beads_evidence(clone_path) + if clone_beads_findings: + raise SchemaFastForwardError( + f"index clone checkpoint found Beads evidence in the clone: {clone_beads_findings}" + ) + payload: dict[str, object] = { + "schema": INDEX_CLONE_CHECKPOINT_SCHEMA, + "checkpointed_at_ms": _now_ms(), + "source": asdict(result.source), + "clone": asdict(result.clone), + "foreign_key_declarations_preserved": result.foreign_key_declarations_preserved, + "foreign_key_check": list(result.foreign_key_check), + "quick_check": list(result.quick_check), + "clone_beads_findings": clone_beads_findings, + "canonical_ddl_sha256": _canonical_ddl_sha256(), + } + _write_receipt(_index_clone_checkpoint_path(clone_path), payload) + return payload + + +def _load_valid_index_clone_checkpoint( + clone_path: Path, *, source_evidence: DatabaseEvidence +) -> dict[str, object] | None: + """Return a checkpoint payload only when self-integrity and source identity hold. + + Any failure here (missing file, tampered hash, drifted source, stale DDL, + recorded Beads evidence) returns ``None`` and the caller falls back to a + full reprove -- a checkpoint can only skip work, never launder a tampered + or stale one into acceptance. + """ + checkpoint_path = _index_clone_checkpoint_path(clone_path) + if not checkpoint_path.exists(): + return None + try: + raw = json.loads(checkpoint_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + if not isinstance(raw, dict): + return None + payload = cast(dict[str, object], raw) + if payload.get("schema") != INDEX_CLONE_CHECKPOINT_SCHEMA: + return None + stored_hash = payload.get("receipt_sha256") + body = {key: value for key, value in payload.items() if key != "receipt_sha256"} + recomputed = hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + if not isinstance(stored_hash, str) or stored_hash != recomputed: + return None + if payload.get("canonical_ddl_sha256") != _canonical_ddl_sha256(): + return None + if payload.get("clone_beads_findings"): + return None + source_payload = payload.get("source") + if not isinstance(source_payload, dict): + return None + expected_source = asdict(source_evidence) + if any( + source_payload.get(field) != expected_source.get(field) for field in ("user_version", "sha256", "size_bytes") + ): + return None + clone_payload = payload.get("clone") + if not isinstance(clone_payload, dict) or not isinstance(payload.get("foreign_key_check"), list): + return None + if not isinstance(payload.get("quick_check"), list): + return None + return payload def reuse_index_clone(source: Path, staged_clone: Path, destination: Path) -> CloneForwardResult: @@ -425,6 +555,14 @@ def reuse_index_clone(source: Path, staged_clone: Path, destination: Path) -> Cl It does not resume arbitrary phases: the active source must still be the observed v35 snapshot, and the supplied clone must independently satisfy the same structural and integrity proof before its atomic move. + + When a checkpoint written by the original ``fast_forward_index_clone`` + call is present beside ``staged_clone`` and its recorded source identity + still matches, this trusts the checkpoint's already-proven table census, + FK check, and quick_check instead of re-deriving them -- verifying only + that the clone's bytes (sha256+size+user_version) still match what the + checkpoint recorded. A missing or mismatched checkpoint falls back to + the full reprove unchanged. """ source_before = _database_evidence(source) if source_before.user_version != 35: @@ -437,16 +575,37 @@ def reuse_index_clone(source: Path, staged_clone: Path, destination: Path) -> Cl raise SchemaFastForwardError(f"invalid reused index clone: {staged_clone}") if destination.exists(): raise SchemaFastForwardError(f"reused index destination already exists: {destination}") - _finalize_clone_database(staged_clone) - require_no_beads_evidence(staged_clone) - clone_after = _database_evidence(staged_clone) + + checkpoint = _load_valid_index_clone_checkpoint(staged_clone, source_evidence=source_before) + checkpoint_clone: DatabaseEvidence | None = None + if checkpoint is not None: + _require_no_sidecars(staged_clone) + checkpoint_clone = _database_evidence_from_payload(cast(dict[str, object], checkpoint["clone"])) + clone_sha256, clone_size, clone_version = _lightweight_database_identity(staged_clone) + if (clone_sha256, clone_size, clone_version) != ( + checkpoint_clone.sha256, + checkpoint_clone.size_bytes, + checkpoint_clone.user_version, + ): + checkpoint = None + checkpoint_clone = None + + if checkpoint is not None and checkpoint_clone is not None: + clone_after = checkpoint_clone + foreign_key_check = tuple(str(item) for item in cast(list[object], checkpoint["foreign_key_check"])) + quick_check = tuple(str(item) for item in cast(list[object], checkpoint["quick_check"])) + else: + _finalize_clone_database(staged_clone) + require_no_beads_evidence(staged_clone) + clone_after = _database_evidence(staged_clone) + with _open_immutable_readonly(staged_clone) as conn: + foreign_key_check = tuple(str(row) for row in conn.execute("PRAGMA foreign_key_check")) + quick_check = tuple(str(row[0]) for row in conn.execute("PRAGMA quick_check")) + if clone_after.user_version != INDEX_SCHEMA_VERSION: raise SchemaFastForwardError( f"reused index clone requires v{INDEX_SCHEMA_VERSION}, found v{clone_after.user_version}" ) - with _open_immutable_readonly(staged_clone) as conn: - foreign_key_check = tuple(str(row) for row in conn.execute("PRAGMA foreign_key_check")) - quick_check = tuple(str(row[0]) for row in conn.execute("PRAGMA quick_check")) if foreign_key_check or quick_check != ("ok",): raise SchemaFastForwardError( f"reused index clone integrity contract failed: fk={foreign_key_check!r}, quick={quick_check!r}" @@ -467,7 +626,7 @@ def reuse_index_clone(source: Path, staged_clone: Path, destination: Path) -> Cl except Exception: local_clone.unlink(missing_ok=True) raise - return CloneForwardResult(source_before, clone_after, True, foreign_key_check) + return CloneForwardResult(source_before, clone_after, True, foreign_key_check, quick_check) def atomic_promote(clone: Path, active: Path, rollback: Path) -> dict[str, str]: @@ -586,53 +745,65 @@ def plan_clone_forward( backup_manifest: Path, reuse_index_clone_path: Path | None = None, ) -> dict[str, object]: - """Prepare derived clones and a receipt; it never promotes or migrates durable tiers.""" + """Prepare derived clones and a receipt; it never promotes or migrates durable tiers. + + Holds the same archive-root-scoped ``RebuildLease`` every daemon and + direct maintenance writer honors (``polylogue.storage.index_generation``, + wired into ``ArchiveStore.__init__`` via ``ActiveWriterLease``) for the + entire prepare phase. A writer already holding the archive (installed + daemon, transient unit, or direct writer) makes this fail before its + first read; a writer started after preflight cannot open the archive for + write until this phase releases the lease. + """ source = archive_root / "source.db" user = archive_root / "user.db" index = archive_root / "index.db" embeddings = archive_root / "embeddings.db" ops = archive_root / "ops.db" - for path in (source, user, index, embeddings, ops): - if not path.exists(): - raise SchemaFastForwardError(f"archive tier is missing: {path}") - if reuse_index_clone_path is not None: - _require_service_stopped("polylogued.service") - require_no_beads_evidence(source, index) - if not backup_manifest.exists(): - raise SchemaFastForwardError(f"verified backup manifest is missing: {backup_manifest}") - run_root = staging_root / f"schema-forward-{uuid.uuid4().hex}" - run_root.mkdir(parents=True, exist_ok=False) - index_result = ( - fast_forward_index_clone(index, run_root / "index.db") - if reuse_index_clone_path is None - else reuse_index_clone(index, reuse_index_clone_path, run_root / "index.db") - ) - embeddings_result = fast_forward_embeddings_clone(embeddings, run_root / "embeddings.db") - initialize_archive_database(run_root / "ops.db", ArchiveTier.OPS) - payload: dict[str, object] = { - "schema": RECEIPT_SCHEMA, - "status": "prepared", - "prepared_at_ms": _now_ms(), - "archive_root": str(archive_root), - "backup_manifest": str(backup_manifest), - "staging_root": str(run_root), - "reused_index_clone": str(reuse_index_clone_path) if reuse_index_clone_path is not None else None, - "source": asdict(_database_evidence(source)), - "user": asdict(_database_evidence(user)), - "index": asdict(index_result), - "embeddings": asdict(embeddings_result), - "ops": { - "source": asdict(_database_evidence(ops)), - "clone": asdict(_database_evidence(run_root / "ops.db")), - "rotation": "disposable canonical reset", - }, - "raw_reparse": False, - "fts_rebuild": False, - "vector_reembed": False, - "durable_activation": "stopped-daemon shipped migration runner with retained reflink rollback clones", - } - _write_receipt(receipt_path, payload) - return payload + try: + with RebuildLease(archive_root): + for path in (source, user, index, embeddings, ops): + if not path.exists(): + raise SchemaFastForwardError(f"archive tier is missing: {path}") + _require_service_stopped("polylogued.service") + require_no_beads_evidence(source, index) + if not backup_manifest.exists(): + raise SchemaFastForwardError(f"verified backup manifest is missing: {backup_manifest}") + run_root = staging_root / f"schema-forward-{uuid.uuid4().hex}" + run_root.mkdir(parents=True, exist_ok=False) + if reuse_index_clone_path is None: + index_result = fast_forward_index_clone(index, run_root / "index.db") + write_index_clone_checkpoint(index_result, clone_path=run_root / "index.db") + else: + index_result = reuse_index_clone(index, reuse_index_clone_path, run_root / "index.db") + embeddings_result = fast_forward_embeddings_clone(embeddings, run_root / "embeddings.db") + initialize_archive_database(run_root / "ops.db", ArchiveTier.OPS) + payload: dict[str, object] = { + "schema": RECEIPT_SCHEMA, + "status": "prepared", + "prepared_at_ms": _now_ms(), + "archive_root": str(archive_root), + "backup_manifest": str(backup_manifest), + "staging_root": str(run_root), + "reused_index_clone": str(reuse_index_clone_path) if reuse_index_clone_path is not None else None, + "source": asdict(_database_evidence(source)), + "user": asdict(_database_evidence(user)), + "index": asdict(index_result), + "embeddings": asdict(embeddings_result), + "ops": { + "source": asdict(_database_evidence(ops)), + "clone": asdict(_database_evidence(run_root / "ops.db")), + "rotation": "disposable canonical reset", + }, + "raw_reparse": False, + "fts_rebuild": False, + "vector_reembed": False, + "durable_activation": "stopped-daemon shipped migration runner with retained reflink rollback clones", + } + _write_receipt(receipt_path, payload) + return payload + except RebuildLeaseUnavailableError as exc: + raise SchemaFastForwardError(f"refusing prepare while another writer owns the archive: {exc}") from exc def activate_prepared_forward( @@ -648,6 +819,13 @@ def activate_prepared_forward( Before migration this actuator retains byte-identical source/user rollback clones. No second backup, raw reparse, FTS rebuild, or vector re-embed is required. + + Holds the archive-root-scoped ``RebuildLease`` (see ``plan_clone_forward``) + for the entire migrate-and-promote window, before its first write and + across every subsequent write. If another writer already owns the + archive, this raises before mutating anything and the receipt is left + exactly as ``prepare`` wrote it (not marked ``rolled_back`` -- nothing was + attempted). """ raw_payload = json.loads(receipt_path.read_text(encoding="utf-8")) if not isinstance(raw_payload, dict): @@ -665,60 +843,74 @@ def activate_prepared_forward( index = archive_root / "index.db" embeddings = archive_root / "embeddings.db" ops = archive_root / "ops.db" - _require_service_stopped(service) - for key, path in (("source", source), ("user", user), ("index", index), ("embeddings", embeddings), ("ops", ops)): - _require_receipt_identity(payload, key, path) - require_no_beads_evidence(source, index) - rollback_root = staging_root / "rollback" - rollback_root.mkdir(exist_ok=False) - snapshots: list[tuple[Path, Path]] = [] - promoted: list[dict[str, str]] = [] try: - # Keep exact old durable bytes locally before the runner commits. The - # pre-existing verified manifest still authenticates active paths - # because these clones are not promoted before validation. - for path in (source, user): - clone = rollback_root / path.name - reflink_clone(path, clone) - snapshots.append((clone, path)) - with sqlite3.connect(source) as conn: - migrate_archive_tier(conn, ArchiveTier.SOURCE, backup_manifest=backup_manifest) - with sqlite3.connect(user) as conn: - migrate_archive_tier(conn, ArchiveTier.USER, backup_manifest=backup_manifest) - # Durable migrations can legitimately leave a WAL behind after their - # connection closes. Finalize those files before immutable evidence - # reads: their committed pages must be in the database file, not a - # transient sidecar. This remains inside the rollback boundary. - for path in (source, user): - _finalize_clone_database(path) - promoted.append(_promote_index_generation(staging_root / index.name, index)) - promoted.append(atomic_promote(staging_root / embeddings.name, embeddings, rollback_root / embeddings.name)) - promoted.append(atomic_promote(staging_root / ops.name, ops, rollback_root / ops.name)) - versions = { - "source": _database_evidence(source).user_version, - "user": _database_evidence(user).user_version, - "index": _database_evidence(index).user_version, - "embeddings": _database_evidence(embeddings).user_version, - "ops": _database_evidence(ops).user_version, - } - except Exception as exc: - for item in reversed(promoted): - _restore_promoted(item) - for snapshot, active in reversed(snapshots): - _restore_snapshot(snapshot, active) - payload.update({"status": "rolled_back", "activation_error": f"{type(exc).__name__}: {exc}"}) - _write_receipt(receipt_path, payload) - raise - payload.update( - { - "status": "activated", - "activated_at_ms": _now_ms(), - "promoted": promoted, - "versions": versions, - } - ) - _write_receipt(receipt_path, payload) - return payload + with RebuildLease(archive_root): + _require_service_stopped(service) + for key, path in ( + ("source", source), + ("user", user), + ("index", index), + ("embeddings", embeddings), + ("ops", ops), + ): + _require_receipt_identity(payload, key, path) + require_no_beads_evidence(source, index) + rollback_root = staging_root / "rollback" + rollback_root.mkdir(exist_ok=False) + snapshots: list[tuple[Path, Path]] = [] + promoted: list[dict[str, str]] = [] + try: + # Keep exact old durable bytes locally before the runner + # commits. The pre-existing verified manifest still + # authenticates active paths because these clones are not + # promoted before validation. + for path in (source, user): + clone = rollback_root / path.name + reflink_clone(path, clone) + snapshots.append((clone, path)) + with sqlite3.connect(source) as conn: + migrate_archive_tier(conn, ArchiveTier.SOURCE, backup_manifest=backup_manifest) + with sqlite3.connect(user) as conn: + migrate_archive_tier(conn, ArchiveTier.USER, backup_manifest=backup_manifest) + # Durable migrations can legitimately leave a WAL behind after + # their connection closes. Finalize those files before + # immutable evidence reads: their committed pages must be in + # the database file, not a transient sidecar. This remains + # inside the rollback boundary. + for path in (source, user): + _finalize_clone_database(path) + promoted.append(_promote_index_generation(staging_root / index.name, index)) + promoted.append( + atomic_promote(staging_root / embeddings.name, embeddings, rollback_root / embeddings.name) + ) + promoted.append(atomic_promote(staging_root / ops.name, ops, rollback_root / ops.name)) + versions = { + "source": _database_evidence(source).user_version, + "user": _database_evidence(user).user_version, + "index": _database_evidence(index).user_version, + "embeddings": _database_evidence(embeddings).user_version, + "ops": _database_evidence(ops).user_version, + } + except Exception as exc: + for item in reversed(promoted): + _restore_promoted(item) + for snapshot, active in reversed(snapshots): + _restore_snapshot(snapshot, active) + payload.update({"status": "rolled_back", "activation_error": f"{type(exc).__name__}: {exc}"}) + _write_receipt(receipt_path, payload) + raise + payload.update( + { + "status": "activated", + "activated_at_ms": _now_ms(), + "promoted": promoted, + "versions": versions, + } + ) + _write_receipt(receipt_path, payload) + return payload + except RebuildLeaseUnavailableError as exc: + raise SchemaFastForwardError(f"refusing activation while another writer owns the archive: {exc}") from exc def _parser() -> argparse.ArgumentParser: @@ -767,6 +959,7 @@ def main(argv: list[str] | None = None) -> int: __all__ = [ "CloneForwardResult", "DatabaseEvidence", + "INDEX_CLONE_CHECKPOINT_SCHEMA", "RECEIPT_SCHEMA", "SchemaFastForwardError", "activate_prepared_forward", @@ -778,5 +971,6 @@ def main(argv: list[str] | None = None) -> int: "plan_clone_forward", "reflink_clone", "require_no_beads_evidence", + "write_index_clone_checkpoint", "main", ] diff --git a/polylogue/storage/sqlite/migration_runner.py b/polylogue/storage/sqlite/migration_runner.py index e12ba1e869..f7bfb5549b 100644 --- a/polylogue/storage/sqlite/migration_runner.py +++ b/polylogue/storage/sqlite/migration_runner.py @@ -170,6 +170,65 @@ def _backup_artifact_inventory(backup_root: Path) -> list[dict[str, object]]: return rows +@dataclass(frozen=True, slots=True) +class _BackupInventoryCacheEntry: + stat_signature: tuple[tuple[str, int, int], ...] + artifact_inventory: tuple[dict[str, object], ...] + + +# Process-lifetime cache: a durable migration runs as a short-lived actuator +# invocation (devtools schema fast-forward, `polylogue ops` maintenance, or +# one daemon-startup migration), so this never needs cross-process +# persistence or eviction -- it exists to collapse the four SHA-256 scans of +# one immutable backup tree that `migrate_archive_tier` otherwise performs +# per activation (pre-BEGIN + in-transaction, times two durable tiers) into +# one. +_backup_artifact_inventory_cache: dict[Path, _BackupInventoryCacheEntry] = {} + + +def _backup_root_stat_signature(backup_root: Path) -> tuple[tuple[str, int, int], ...]: + """Cheap, content-free fingerprint used only to decide whether the + expensive SHA-256 scan below can be skipped. + + This is deliberately not evidence by itself: every entry the cache + returns still carries the SHA-256 computed the last time the signature + changed. A stat-identical-but-content-tampered backup (same size and + mtime, different bytes) is outside this actuator's threat model already + -- ``backup_attestation.py`` documents it is "not a privilege boundary + against arbitrary code running as the same Unix user" -- and any + genuine artifact/manifest/receipt mutation changes size or mtime and is + still caught below. + """ + entries: list[tuple[str, int, int]] = [] + for candidate in sorted(backup_root.rglob("*")): + relative = candidate.relative_to(backup_root) + if relative == Path(_VERIFICATION_RECEIPT_FILE): + continue + metadata = candidate.lstat() + entries.append((str(relative), metadata.st_size, metadata.st_mtime_ns)) + return tuple(entries) + + +def _cached_backup_artifact_inventory(backup_root: Path) -> list[dict[str, object]]: + """Reuse a SHA-256'd backup artifact inventory while its bytes are unchanged.""" + resolved = backup_root.resolve(strict=True) + signature = _backup_root_stat_signature(resolved) + cached = _backup_artifact_inventory_cache.get(resolved) + if cached is not None and cached.stat_signature == signature: + return [dict(item) for item in cached.artifact_inventory] + inventory = _backup_artifact_inventory(resolved) + # Guard a scan-time mutation race: only cache a result whose signature is + # still what it was before the (potentially slow) hashing pass began. + if _backup_root_stat_signature(resolved) == signature: + _backup_artifact_inventory_cache[resolved] = _BackupInventoryCacheEntry( + stat_signature=signature, + artifact_inventory=tuple(dict(item) for item in inventory), + ) + else: + _backup_artifact_inventory_cache.pop(resolved, None) + return inventory + + def _canonical_json_sha256(payload: object) -> str: encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") return hashlib.sha256(encoded).hexdigest() @@ -455,7 +514,7 @@ def validate_migration_backup_manifest( raise MigrationError(f"migration backup receipt authentication failed: {exc}") from exc if receipt.get("verdict") != "success": raise MigrationError(f"migration backup receipt is not a successful verification: {receipt_path}") - artifact_inventory = _backup_artifact_inventory(backup_root) + artifact_inventory = _cached_backup_artifact_inventory(backup_root) file_evidence = {str(item["path"]): item for item in artifact_inventory if item.get("type") == "file"} manifest_evidence = file_evidence.get("manifest.json", {}) if _json_int(receipt.get("manifest_size_bytes")) != _json_int(manifest_evidence.get("size_bytes")): diff --git a/tests/unit/devtools/test_archive_schema_fast_forward.py b/tests/unit/devtools/test_archive_schema_fast_forward.py index e5d7a5c4be..1b11039ef1 100644 --- a/tests/unit/devtools/test_archive_schema_fast_forward.py +++ b/tests/unit/devtools/test_archive_schema_fast_forward.py @@ -3,6 +3,7 @@ from __future__ import annotations import errno +import hashlib import json import os import sqlite3 @@ -10,7 +11,9 @@ import pytest +import devtools.archive_schema_fast_forward as fast_forward_module from devtools.archive_schema_fast_forward import ( + INDEX_CLONE_CHECKPOINT_SCHEMA, SchemaFastForwardError, _parser, _promote_index_generation, @@ -20,9 +23,12 @@ beads_evidence, fast_forward_embeddings_clone, fast_forward_index_clone, + plan_clone_forward, reuse_index_clone, + write_index_clone_checkpoint, ) -from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_tier +from polylogue.storage.index_generation import ActiveWriterLease, RebuildLeaseUnavailableError +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database, initialize_archive_tier from polylogue.storage.sqlite.archive_tiers.index import INDEX_DDL from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from polylogue.storage.sqlite.sqlite_vec_extension import try_load_sqlite_vec @@ -84,6 +90,30 @@ def _fk_declarations(path: Path) -> dict[str, tuple[tuple[object, ...], ...]]: } +def _create_v1_embeddings(path: Path) -> None: + with sqlite3.connect(path) as conn: + conn.execute("PRAGMA journal_mode = WAL") + loaded, error = try_load_sqlite_vec(conn) + if not loaded: + pytest.skip(f"sqlite-vec unavailable: {error}") + initialize_archive_tier(conn, ArchiveTier.EMBEDDINGS) + conn.execute("DROP INDEX idx_embedding_failures_active") + conn.execute("DROP TABLE embedding_failures") + conn.execute("PRAGMA user_version = 1") + conn.commit() + _clear_wal_sidecars(path) + + +def _build_v35_archive_root(root: Path) -> None: + """Build a full 5-tier archive at the exact versions ``prepare`` expects.""" + root.mkdir(parents=True, exist_ok=True) + initialize_archive_database(root / "source.db", ArchiveTier.SOURCE) + initialize_archive_database(root / "user.db", ArchiveTier.USER) + initialize_archive_database(root / "ops.db", ArchiveTier.OPS) + _create_v35_index(root / "index.db") + _create_v1_embeddings(root / "embeddings.db") + + def test_index_clone_copy_forward_preserves_fk_graph_ddl_and_rows(tmp_path: Path) -> None: source = tmp_path / "index-v35.db" clone = tmp_path / "index-v36.db" @@ -533,3 +563,372 @@ def test_operator_cli_exposes_only_prepare_and_existing_manifest_activation() -> assert activate.command == "activate" assert str(activate.backup_manifest) == "/tmp/manifest.json" assert activate.service == "polylogued.service" + + +# --- polylogue-qg6x: resumable schema-forward clone proofs ----------------- + + +def test_write_index_clone_checkpoint_records_a_self_checking_receipt(tmp_path: Path) -> None: + source = tmp_path / "index-v35.db" + clone = tmp_path / "index-v36.db" + _create_v35_index(source) + + result = fast_forward_index_clone(source, clone) + checkpoint = write_index_clone_checkpoint(result, clone_path=clone) + + checkpoint_path = clone.with_name(f"{clone.name}.clone-checkpoint.json") + assert checkpoint_path.exists() + on_disk = json.loads(checkpoint_path.read_text(encoding="utf-8")) + assert on_disk == checkpoint + assert checkpoint["schema"] == INDEX_CLONE_CHECKPOINT_SCHEMA + assert on_disk["source"]["user_version"] == 35 + assert on_disk["clone"]["user_version"] == 36 + assert checkpoint["quick_check"] == ["ok"] + assert checkpoint["foreign_key_check"] == [] + assert checkpoint["foreign_key_declarations_preserved"] is True + assert checkpoint["clone_beads_findings"] == {} + assert isinstance(checkpoint["canonical_ddl_sha256"], str) and len(checkpoint["canonical_ddl_sha256"]) == 64 + # Self-checking: the receipt hash covers every other field. + body = {key: value for key, value in checkpoint.items() if key != "receipt_sha256"} + assert ( + checkpoint["receipt_sha256"] + == hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + ) + + +def test_reuse_index_clone_with_valid_checkpoint_skips_clone_census_fk_and_quick_check( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + source = tmp_path / "index-v35.db" + staged = tmp_path / "completed-index-v36.db" + destination = tmp_path / "new-run" / "index.db" + _create_v35_index(source) + result = fast_forward_index_clone(source, staged) + write_index_clone_checkpoint(result, clone_path=staged) + + evidence_calls: list[Path] = [] + original_evidence = fast_forward_module._database_evidence + + def tracking_evidence(path: Path) -> object: + evidence_calls.append(Path(path)) + return original_evidence(path) + + monkeypatch.setattr(fast_forward_module, "_database_evidence", tracking_evidence) + + def fail_finalize(path: Path) -> None: + pytest.fail("checkpoint reuse must not re-finalize the already-finalized clone") + + monkeypatch.setattr(fast_forward_module, "_finalize_clone_database", fail_finalize) + + reused = reuse_index_clone(source, staged, destination) + + assert reused.source == result.source + assert reused.clone == result.clone + assert reused.quick_check == ("ok",) + assert destination.exists() + # The source is always freshly evidenced (it can legitimately drift); + # the clone's table census is trusted from the checkpoint instead of a + # fresh (table-scanning) `_database_evidence` call. + assert evidence_calls == [source] + + +def test_reuse_index_clone_ignores_checkpoint_when_source_has_drifted_since(tmp_path: Path) -> None: + source = tmp_path / "index-v35.db" + staged = tmp_path / "completed-index-v36.db" + destination = tmp_path / "new-run" / "index.db" + _create_v35_index(source) + result = fast_forward_index_clone(source, staged) + write_index_clone_checkpoint(result, clone_path=staged) + + with sqlite3.connect(source) as conn: + conn.execute( + "INSERT INTO sessions(native_id, origin, content_hash) VALUES (?, ?, ?)", + ("drift", "chatgpt-export", b"d" * 32), + ) + conn.commit() + + # Source drift invalidates the checkpoint (source identity mismatch), so + # this falls back to a full reprove against the now-larger source -- and + # correctly rejects the stale clone rather than silently trusting it. + with pytest.raises(SchemaFastForwardError, match="structural row counts"): + reuse_index_clone(source, staged, destination) + + +def test_reuse_index_clone_ignores_checkpoint_when_clone_bytes_tampered(tmp_path: Path) -> None: + source = tmp_path / "index-v35.db" + staged = tmp_path / "completed-index-v36.db" + destination = tmp_path / "new-run" / "index.db" + _create_v35_index(source) + result = fast_forward_index_clone(source, staged) + write_index_clone_checkpoint(result, clone_path=staged) + + with sqlite3.connect(staged) as conn: + conn.execute( + "INSERT INTO sessions(native_id, origin, content_hash) VALUES (?, ?, ?)", + ("tampered", "chatgpt-export", b"t" * 32), + ) + conn.commit() + + # The clone's bytes no longer match what the checkpoint recorded, so the + # lightweight identity check discards the checkpoint and a full reprove + # against the (unchanged) source correctly rejects the tampered clone. + with pytest.raises(SchemaFastForwardError, match="structural row counts"): + reuse_index_clone(source, staged, destination) + + +def test_reuse_index_clone_ignores_a_tampered_checkpoint_receipt(tmp_path: Path) -> None: + source = tmp_path / "index-v35.db" + staged = tmp_path / "completed-index-v36.db" + destination = tmp_path / "new-run" / "index.db" + _create_v35_index(source) + result = fast_forward_index_clone(source, staged) + write_index_clone_checkpoint(result, clone_path=staged) + + checkpoint_path = staged.with_name(f"{staged.name}.clone-checkpoint.json") + payload = json.loads(checkpoint_path.read_text(encoding="utf-8")) + payload["quick_check"] = ["forged"] + checkpoint_path.write_text(json.dumps(payload), encoding="utf-8") + + # A checkpoint whose recorded body no longer matches its own hash cannot + # be trusted, but that must not corrupt reuse of an otherwise-sound + # clone: it just forces the (still-correct) full reprove path. + reused = reuse_index_clone(source, staged, destination) + assert reused.source == result.source + assert reused.clone == result.clone + assert reused.quick_check == ("ok",) + + +def test_reuse_index_clone_checkpoint_fast_path_still_rejects_clone_sidecars(tmp_path: Path) -> None: + source = tmp_path / "index-v35.db" + staged = tmp_path / "completed-index-v36.db" + destination = tmp_path / "new-run" / "index.db" + _create_v35_index(source) + result = fast_forward_index_clone(source, staged) + write_index_clone_checkpoint(result, clone_path=staged) + + # Keep the writer open: SQLite auto-checkpoints (and can remove) the WAL + # sidecar once the last connection to a WAL-mode database closes, so the + # sidecar must stay visible for the duration of this check -- matching + # the existing beads-census WAL fixture pattern above. + writer = sqlite3.connect(staged) + try: + writer.execute("PRAGMA journal_mode = WAL") + writer.execute("CREATE TABLE scratch (value TEXT)") + writer.commit() + assert Path(f"{staged}-wal").exists() + + with pytest.raises(SchemaFastForwardError, match="SQLite sidecars"): + reuse_index_clone(source, staged, destination) + finally: + writer.close() + + +# --- polylogue-b5l.1: writer-exclusive, crash-resumable rebuilds ----------- + + +def test_plan_clone_forward_fails_before_first_write_when_writer_already_owns_archive(tmp_path: Path) -> None: + archive = tmp_path / "archive" + _build_v35_archive_root(archive) + staging = tmp_path / "staging" + receipt = tmp_path / "receipt.json" + manifest = tmp_path / "manifest.json" + manifest.write_text("{}", encoding="utf-8") + + writer = ActiveWriterLease(archive) + writer.acquire() + try: + with pytest.raises(SchemaFastForwardError, match="another writer owns the archive"): + plan_clone_forward( + archive_root=archive, + staging_root=staging, + receipt_path=receipt, + backup_manifest=manifest, + ) + finally: + writer.close() + + assert not receipt.exists() + assert not staging.exists() + + +def test_plan_clone_forward_writes_index_clone_checkpoint_before_later_tier_work( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + archive = tmp_path / "archive" + _build_v35_archive_root(archive) + staging = tmp_path / "staging" + receipt = tmp_path / "receipt.json" + manifest = tmp_path / "manifest.json" + manifest.write_text("{}", encoding="utf-8") + monkeypatch.setattr(fast_forward_module, "_require_service_stopped", lambda _service: None) + + payload = plan_clone_forward( + archive_root=archive, + staging_root=staging, + receipt_path=receipt, + backup_manifest=manifest, + ) + + run_root = Path(str(payload["staging_root"])) + checkpoint_path = run_root / "index.db.clone-checkpoint.json" + assert checkpoint_path.exists() + checkpoint = json.loads(checkpoint_path.read_text(encoding="utf-8")) + assert checkpoint["schema"] == INDEX_CLONE_CHECKPOINT_SCHEMA + assert checkpoint["quick_check"] == ["ok"] + + +def test_plan_clone_forward_reuse_flag_consumes_the_prior_runs_checkpoint( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + archive = tmp_path / "archive" + _build_v35_archive_root(archive) + staging = tmp_path / "staging" + manifest = tmp_path / "manifest.json" + manifest.write_text("{}", encoding="utf-8") + monkeypatch.setattr(fast_forward_module, "_require_service_stopped", lambda _service: None) + + first_receipt = tmp_path / "receipt-1.json" + first_payload = plan_clone_forward( + archive_root=archive, + staging_root=staging, + receipt_path=first_receipt, + backup_manifest=manifest, + ) + first_run_root = Path(str(first_payload["staging_root"])) + first_index_clone = first_run_root / "index.db" + + census_calls: list[Path] = [] + original_evidence = fast_forward_module._database_evidence + + def tracking_evidence(path: Path) -> object: + census_calls.append(Path(path)) + return original_evidence(path) + + monkeypatch.setattr(fast_forward_module, "_database_evidence", tracking_evidence) + + second_receipt = tmp_path / "receipt-2.json" + second_payload = plan_clone_forward( + archive_root=archive, + staging_root=staging, + receipt_path=second_receipt, + backup_manifest=manifest, + reuse_index_clone_path=first_index_clone, + ) + + assert second_payload["reused_index_clone"] == str(first_index_clone) + # This is the exact scenario the operator hit live on 2026-07-13: the + # first attempt's clone survives (fully proven) and a retry consumes it + # via --reuse-index-clone. The retry must never re-census the *staged* + # clone -- only the live archive's index.db (which can legitimately + # drift) is freshly evidenced. + assert first_index_clone not in census_calls + assert (archive / "index.db") in census_calls + + +def test_activate_prepared_forward_holds_writer_exclusion_across_the_whole_migration( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + archive = tmp_path / "archive" + staging = tmp_path / "staging" + archive.mkdir() + staging.mkdir() + source = archive / "source.db" + user = archive / "user.db" + for database in (source, user, archive / "index.db", archive / "embeddings.db", archive / "ops.db"): + with sqlite3.connect(database) as conn: + conn.execute("CREATE TABLE retained (value TEXT)") + conn.commit() + manifest = tmp_path / "manifest.json" + manifest.write_text("{}", encoding="utf-8") + receipt = tmp_path / "receipt.json" + receipt.write_text( + json.dumps( + { + "schema": "polylogue.archive-schema-fast-forward.v1", + "status": "prepared", + "archive_root": str(archive), + "staging_root": str(staging), + "backup_manifest": str(manifest), + } + ), + encoding="utf-8", + ) + monkeypatch.setattr(fast_forward_module, "_require_service_stopped", lambda _service: None) + monkeypatch.setattr(fast_forward_module, "_require_receipt_identity", lambda *_args: None) + monkeypatch.setattr(fast_forward_module, "require_no_beads_evidence", lambda *_paths: None) + + observed_conflict: list[bool] = [] + + def observe_exclusion_then_fail(conn: sqlite3.Connection, tier: object, **_kwargs: object) -> None: + del conn, tier + writer = ActiveWriterLease(archive) + try: + writer.acquire() + except RebuildLeaseUnavailableError: + observed_conflict.append(True) + else: + observed_conflict.append(False) + writer.close() + raise RuntimeError("synthetic stop after observing exclusion") + + monkeypatch.setattr(fast_forward_module, "migrate_archive_tier", observe_exclusion_then_fail) + + with pytest.raises(RuntimeError, match="synthetic stop after observing exclusion"): + activate_prepared_forward(receipt_path=receipt, backup_manifest=manifest) + + assert observed_conflict == [True] + assert json.loads(receipt.read_text(encoding="utf-8"))["status"] == "rolled_back" + + +def test_activate_prepared_forward_fails_before_first_write_when_writer_already_owns_archive( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + archive = tmp_path / "archive" + staging = tmp_path / "staging" + archive.mkdir() + staging.mkdir() + for database in ( + archive / "source.db", + archive / "user.db", + archive / "index.db", + archive / "embeddings.db", + archive / "ops.db", + ): + with sqlite3.connect(database) as conn: + conn.execute("CREATE TABLE retained (value TEXT)") + conn.execute("INSERT INTO retained VALUES ('original')") + conn.commit() + manifest = tmp_path / "manifest.json" + manifest.write_text("{}", encoding="utf-8") + receipt = tmp_path / "receipt.json" + receipt.write_text( + json.dumps( + { + "schema": "polylogue.archive-schema-fast-forward.v1", + "status": "prepared", + "archive_root": str(archive), + "staging_root": str(staging), + "backup_manifest": str(manifest), + } + ), + encoding="utf-8", + ) + monkeypatch.setattr(fast_forward_module, "_require_service_stopped", lambda _service: None) + monkeypatch.setattr( + fast_forward_module, + "migrate_archive_tier", + lambda *_args, **_kwargs: pytest.fail("must fail before migrate_archive_tier is ever called"), + ) + + writer = ActiveWriterLease(archive) + writer.acquire() + try: + with pytest.raises(SchemaFastForwardError, match="another writer owns the archive"): + activate_prepared_forward(receipt_path=receipt, backup_manifest=manifest) + finally: + writer.close() + + # Nothing was attempted -- the receipt is untouched, not marked rolled_back. + assert json.loads(receipt.read_text(encoding="utf-8"))["status"] == "prepared" + with sqlite3.connect(archive / "source.db") as conn: + assert conn.execute("SELECT value FROM retained").fetchall() == [("original",)] diff --git a/tests/unit/storage/test_durable_migrations.py b/tests/unit/storage/test_durable_migrations.py index a2ac88028b..14a9b5794e 100644 --- a/tests/unit/storage/test_durable_migrations.py +++ b/tests/unit/storage/test_durable_migrations.py @@ -1340,3 +1340,96 @@ def test_derived_tiers_do_not_use_migration_runner(tmp_path: Path) -> None: migrate_archive_tier(conn, ArchiveTier.INDEX, backup_manifest=tmp_path / "missing-manifest.json") finally: conn.close() + + +def test_backup_artifact_inventory_scan_is_cached_across_both_durable_tier_migrations( + workspace_env: dict[str, Path], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """One activation authenticates the immutable backup inventory once. + + Without caching, ``migrate_archive_tier`` scans and SHA-256's the whole + backup artifact tree (including the blob inventory) four times for one + source+user activation: once before ``BEGIN`` and once inside the + transaction, for each of the two durable tiers. The backup directory is + asserted immutable for the run, so those are pure duplication -- observed + as 100+ GiB of repeat reads against a 35 GiB index during the 2026-07-13 + v35->v36 cutover. This proves the expensive scan (``_backup_artifact_inventory``) + now runs exactly once and both tier migrations still succeed off the + shared, cached result. + """ + archive_root_path = workspace_env["archive_root"] + source_path = archive_root_path / "source.db" + user_path = archive_root_path / "user.db" + _create_source_v1(source_path) + _create_user_v3(user_path) + # Default "rebuildable_cache_exclude" profile includes source+user+embeddings. + manifest = _verified_backup_manifest(tmp_path / "backup") + + scan_calls: list[Path] = [] + original_scan = migration_runner._backup_artifact_inventory + + def counting_scan(backup_root: Path) -> list[dict[str, object]]: + scan_calls.append(backup_root) + return original_scan(backup_root) + + monkeypatch.setattr(migration_runner, "_backup_artifact_inventory", counting_scan) + + with sqlite3.connect(source_path) as conn: + source_result = migrate_archive_tier(conn, ArchiveTier.SOURCE, backup_manifest=manifest) + with sqlite3.connect(user_path) as conn: + user_result = migrate_archive_tier(conn, ArchiveTier.USER, backup_manifest=manifest) + + assert source_result.to_version == SOURCE_SCHEMA_VERSION + assert user_result.to_version == USER_SCHEMA_VERSION + assert len(scan_calls) == 1 + + +def test_cached_backup_inventory_still_detects_tamper_between_tier_migrations( + workspace_env: dict[str, Path], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A cache hit can only skip work, never launder a mutated backup. + + The source-tier migration populates the cache; tampering the backup + afterward must still be caught by the user-tier migration, proving the + cheap stat-signature check actually invalidates on real mutation instead + of silently trusting a stale scan. + """ + archive_root_path = workspace_env["archive_root"] + source_path = archive_root_path / "source.db" + user_path = archive_root_path / "user.db" + _create_source_v1(source_path) + _create_user_v3(user_path) + manifest = _verified_backup_manifest(tmp_path / "backup") + + with sqlite3.connect(source_path) as conn: + migrate_archive_tier(conn, ArchiveTier.SOURCE, backup_manifest=manifest) + + _tamper_backup_tier(manifest) + _block_migration_sql(monkeypatch) + + with sqlite3.connect(user_path) as conn: + with pytest.raises(MigrationError, match="tier artifact .* mismatch"): + migrate_archive_tier(conn, ArchiveTier.USER, backup_manifest=manifest) + + +def test_backup_inventory_cache_signature_rejects_stale_entry_after_size_change(tmp_path: Path) -> None: + """Unit-level proof that a changed artifact invalidates the cached scan.""" + backup_root = tmp_path / "backup-cache" + backup_root.mkdir() + artifact = backup_root / "example.db" + artifact.write_bytes(b"original-bytes") + migration_runner._backup_artifact_inventory_cache.clear() + + first = migration_runner._cached_backup_artifact_inventory(backup_root) + first_hash = next(item["sha256"] for item in first if item["path"] == "example.db") + + artifact.write_bytes(b"mutated-bytes-are-longer") + second = migration_runner._cached_backup_artifact_inventory(backup_root) + second_hash = next(item["sha256"] for item in second if item["path"] == "example.db") + + assert first_hash != second_hash + assert second_hash == hashlib.sha256(b"mutated-bytes-are-longer").hexdigest()