From 31a01f373bb520703e26fece6bb106255c2d55b2 Mon Sep 17 00:00:00 2001 From: "Andrex Ibiza, MBA" Date: Sun, 23 Aug 2026 17:39:13 -0500 Subject: [PATCH] fix(state): make automatic repair non-destructive Reproduce the schema-btree failure where the in-place writable_schema/VACUUM ladder can reduce a 3,048-page canonical state.db to 113 pages and still return repaired=False. Move all mutating strategies behind a complete SQLite online-backup snapshot, retain one exclusive SQLite guard from staging through transactional promotion, preserve committed WAL frames and the live inode, fail closed on environmental hazards, and add adversarial regression coverage for failed-repair preservation, post-stage writer races, interrupted copies, stale scratch, disk admission, attempt-ledger semantics, and durability routing. Fixes #93064 Supersedes the delivery mechanics of #87409 while preserving its implementation provenance. Co-authored-by: cervantesh <11169707+cervantesh@users.noreply.github.com> --- hermes_state.py | 523 +++++++++-- tests/test_state_db_repair_non_destructive.py | 877 ++++++++++++++++++ tests/test_state_db_write_durability.py | 52 +- 3 files changed, 1374 insertions(+), 78 deletions(-) create mode 100644 tests/test_state_db_repair_non_destructive.py diff --git a/hermes_state.py b/hermes_state.py index d24b695b0e..55e8145c65 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -1738,6 +1738,11 @@ def _claim_repair_attempt(db_path: Path) -> bool: # just did so on top of the winner's. _REPAIR_LOCK_TIMEOUT_SECONDS = 120.0 _REPAIR_LOCK_POLL_SECONDS = 0.1 +# Copying a multi-GB database through SQLite's online-backup API is a data +# transfer, not an inter-process locking operation. Keep the repair-lock +# bound separate and give a conservative 10 MiB/s budget to each snapshot or +# promotion, with the historical two-minute floor for ordinary state.db files. +_REPAIR_SNAPSHOT_MIN_THROUGHPUT_BYTES_PER_SECOND = 10 * 1024 * 1024 _IS_WINDOWS = sys.platform == "win32" @@ -1922,6 +1927,95 @@ def _repair_backup_headroom_bytes(total_bytes: int) -> int: ) +def _repair_scratch_space_error(db_path: Path) -> Optional[str]: + """Return an error unless snapshot, VACUUM and promotion can fit safely.""" + import shutil + + try: + main_bytes = db_path.stat().st_size + snapshot_bytes = main_bytes + for suffix in _DB_SIDECAR_SUFFIXES: + sidecar = db_path.with_name(db_path.name + suffix) + if sidecar.exists(): + snapshot_bytes += sidecar.stat().st_size + usage = shutil.disk_usage(db_path.parent) + headroom = _repair_backup_headroom_bytes(usage.total) + # Strategy 2 runs VACUUM on the staged database. SQLite documents that + # VACUUM may need up to twice the database size in additional free + # space while it builds the replacement and journals the overwrite. + # Reserve that beyond the snapshot itself; after VACUUM releases its + # temporary files, the same reserve also covers transactional + # promotion into the live database. + required = snapshot_bytes + (2 * snapshot_bytes) + headroom + if usage.free >= required: + return None + return ( + f"only {usage.free / 1e9:.2f}GB free on {db_path.parent}; the " + f"repair snapshot needs up to {snapshot_bytes / 1e9:.2f}GB, " + f"VACUUM may need another {(2 * snapshot_bytes) / 1e9:.2f}GB, and " + f"{headroom / 1e9:.2f}GB must remain as headroom. Free disk space, " + "then retry." + ) + except OSError as exc: + return ( + f"could not determine free space on {db_path.parent} ({exc}); " + "refusing the repair snapshot rather than risk filling the volume" + ) + + +def _repair_snapshot_timeout_seconds(source_path: Path) -> float: + """Bound one SQLite snapshot by source size, including live sidecars. + + A WAL can contain committed canonical rows which are not yet present in + the main database file. Count it (and the rollback journal where + present), both to describe the work honestly and to avoid applying the + repair-lock timeout to an otherwise healthy large-database copy. + """ + source_bytes = 0 + for suffix in ("", *_DB_SIDECAR_SUFFIXES): + candidate = ( + source_path + if not suffix + else source_path.with_name(source_path.name + suffix) + ) + try: + source_bytes += candidate.stat().st_size + except FileNotFoundError: + continue + return max( + _REPAIR_LOCK_TIMEOUT_SECONDS, + source_bytes / _REPAIR_SNAPSHOT_MIN_THROUGHPUT_BYTES_PER_SECOND, + ) + + +def _repair_failure_consumes_attempt(exc: BaseException) -> bool: + """Whether a pre-strategy SQLite failure proves deterministic corruption. + + Lock contention, timeouts, disk-full, I/O and filesystem failures are + environmental aborts: retrying later may succeed and must not exhaust the + repair ledger. Only SQLite's corruption/image result codes prove the + deterministic damage the bounded ledger exists to stop from retrying + forever, even when SQLite cannot stage a snapshot far enough to run a + named strategy. + """ + if not isinstance(exc, sqlite3.DatabaseError): + return False + error_code = getattr(exc, "sqlite_errorcode", None) + if isinstance(error_code, int): + # Extended result codes retain the primary code in the low byte. + primary_code = error_code & 0xFF + return primary_code in (sqlite3.SQLITE_CORRUPT, sqlite3.SQLITE_NOTADB) + + # sqlite3 versions before exception result-code attributes need a narrow, + # conservative compatibility path. Do not turn generic DatabaseError + # messages such as "disk is full" or "readonly" into permanent failures. + message = str(exc).lower() + return ( + "file is not a database" in message + or "database disk image is malformed" in message + ) + + def _repair_ledger_path(db_path: Path) -> Path: return db_path.with_name(db_path.name + ".repair-attempts.json") @@ -2104,6 +2198,19 @@ def _persistent_repair_attempts_exhausted(db_path: Path) -> bool: return int(ledger.get("failed_attempts", 0)) >= _MAX_PERSISTENT_REPAIR_ATTEMPTS +def _persistent_repair_exhausted_error(db_path: Path) -> str: + """The stable operator-facing diagnostic for an exhausted repair budget.""" + return ( + f"automatic repair has already failed " + f"{_MAX_PERSISTENT_REPAIR_ATTEMPTS} times on this exact file — " + "the corruption is beyond the schema/FTS repair strategies " + "(likely b-tree page damage). Manual recovery required: restore " + f"a backup, or salvage with `sqlite3 {db_path} \".recover\"`. " + f"Delete {_repair_ledger_path(db_path).name} to force another " + "automatic attempt." + ) + + def _record_repair_outcome( db_path: Path, *, repaired: bool, fingerprint: "Optional[str]" = None ) -> None: @@ -2185,12 +2292,12 @@ def _prune_malformed_backups(db_path: Path, keep: int = _MAX_MALFORMED_BACKUPS) def _backup_db_file(db_path: Path) -> "Tuple[Optional[Path], Optional[str]]": """Copy a (possibly malformed) DB file to a timestamped backup beside it. Raw file copy on purpose: the DB won't open cleanly, so we preserve the - bytes exactly for forensics / manual restore. WAL and SHM sidecars are - copied too when present. Returns ``(backup_path, None)`` on success or + bytes exactly for forensics / manual restore. WAL, SHM and rollback-journal + sidecars are copied too when present. Returns ``(backup_path, None)`` on success or ``(None, reason)`` on failure — callers on the repair path treat a - refused backup as a HARD STOP (see #69603: proceeding without the - pre-repair backup leaves the writable_schema surgery, FTS deletion and - VACUUM strategies mutating the only remaining copy of the damaged DB). + refused backup as a HARD STOP (see #69603). Repair strategies run on a + scratch snapshot, but the forensic bundle remains the recovery path when + corruption defeats them. Refuses when a connection to this database is still live in the process: reading the file would ``close()`` a descriptor for it and cancel that @@ -2475,7 +2582,9 @@ def preflight_db_writability( _ensure_writable(p) -def _connect_repair_durable(db_path: Path) -> sqlite3.Connection: +def _connect_repair_durable( + db_path: Path, *, timeout: float = 5.0 +) -> sqlite3.Connection: """``sqlite3.connect`` for the repair/probe paths, with macOS write barriers. These paths open ``state.db`` directly rather than through ``SessionDB`` @@ -2506,7 +2615,7 @@ def _connect_repair_durable(db_path: Path) -> sqlite3.Connection: rewrite the whole file call :func:`_reapply_durability_barriers` once the schema parses again, which is the point at which the pragmas can stick. """ - conn = sqlite3.connect(str(db_path), isolation_level=None) + conn = sqlite3.connect(str(db_path), timeout=timeout, isolation_level=None) _reapply_durability_barriers(conn) return conn @@ -2530,6 +2639,111 @@ def _reapply_durability_barriers(conn: sqlite3.Connection) -> bool: return False +@contextmanager +def _exclusive_repair_db_guard(db_path: Path): + """Yield one live connection that excludes writers for repair surgery. + + ``locking_mode=EXCLUSIVE`` retains SQLite's file-level exclusion after the + short ``BEGIN EXCLUSIVE`` transaction is rolled back. That rollback is + essential: ``Connection.backup`` may use the guarded connection as a + *source* while it is transaction-free, and it must be transaction-free + when it is later the promotion *destination*. The connection itself + remains open for the entire snapshot -> strategies -> promotion window, + so another writer cannot commit a change that promotion could overwrite. + + Existing WAL readers make exclusive acquisition fail rather than being + disturbed. In DELETE mode an existing reader similarly prevents + ``BEGIN EXCLUSIVE``; a future reader/writer waits behind the guard. A + repair therefore fails closed whenever this process cannot own that whole + window. + """ + guard: Optional[sqlite3.Connection] = None + try: + # The cross-process repair lock already serializes repairers. Do not + # wait behind an ordinary application connection: a partial repair is + # less safe than an explicit "stop the gateway and retry" result. + guard = _connect_repair_durable(db_path, timeout=0.0) + guard.execute("PRAGMA locking_mode=EXCLUSIVE") + guard.execute("BEGIN EXCLUSIVE") + guard.execute("ROLLBACK") + except (sqlite3.Error, OSError) as exc: + if guard is not None: + try: + guard.execute("PRAGMA locking_mode=NORMAL") + except Exception: + pass + guard.close() + yield None, exc + return + + try: + yield guard, None + finally: + try: + # Let SQLite release the exclusive locks before close; this also + # avoids a connection-close checkpoint being mistaken for a + # repair write in callers that immediately reopen state.db. + guard.execute("PRAGMA locking_mode=NORMAL") + except Exception: + pass + guard.close() + + +def _copy_database_snapshot( + source_path: Path, + destination_path: Path, + *, + source_connection: Optional[sqlite3.Connection] = None, + destination_connection: Optional[sqlite3.Connection] = None, +) -> None: + """Copy one complete SQLite snapshot without replacing either file inode. + + SQLite's online backup API incorporates committed WAL frames into the + source snapshot and writes the destination inside one transaction. This + avoids both the main-file-only staging gap and replacing ``state.db`` from + under handles that already refer to it. If backup is interrupted, SQLite + rolls the destination transaction back. + """ + # Work out the deadline before opening an owned source connection. A + # sidecar disappearing while we stat it is an ordinary staging failure, + # but it must not leak a just-opened SQLite descriptor. + deadline_seconds = _repair_snapshot_timeout_seconds(source_path) + deadline = time.monotonic() + deadline_seconds + source = source_connection or _connect_repair_durable(source_path) + destination = destination_connection + own_source = source_connection is None + own_destination = destination_connection is None + + def _check_deadline(_status: int, _remaining: int, _total: int) -> None: + if time.monotonic() >= deadline: + raise TimeoutError( + "timed out copying SQLite repair snapshot after " + f"{deadline_seconds:.0f}s" + ) + + try: + if destination is None: + destination = _connect_repair_durable(destination_path) + elif destination.in_transaction: + # sqlite3_backup requires a transaction-free destination. The + # exclusive repair guard deliberately retains file exclusion via + # locking_mode, not an active transaction, so it satisfies this. + raise sqlite3.ProgrammingError( + "SQLite repair backup destination has an active transaction" + ) + source.backup( + destination, + pages=256, + progress=_check_deadline, + sleep=_REPAIR_LOCK_POLL_SECONDS, + ) + finally: + if own_destination and destination is not None: + destination.close() + if own_source: + source.close() + + def _db_opens_cleanly(db_path: Path) -> Optional[str]: """Probe a DB on a fresh connection. Returns None if healthy, else a reason. @@ -2683,7 +2897,7 @@ def _live_writer_holds_db(db_path: Path) -> bool: """ probe = None try: - probe = sqlite3.connect(str(db_path), timeout=0.0, isolation_level=None) + probe = _connect_repair_durable(db_path, timeout=0.0) probe.execute("PRAGMA locking_mode=EXCLUSIVE") probe.execute("BEGIN IMMEDIATE") probe.execute("ROLLBACK") @@ -2731,8 +2945,10 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A The next ``SessionDB()`` open rebuilds the FTS indexes from the canonical ``messages`` table. - Canonical ``sessions`` / ``messages`` rows are never modified. A - timestamped raw backup is taken first unless ``backup=False``. + Canonical ``sessions`` / ``messages`` rows are never modified by a failed + attempt. Mutating strategies run against a complete SQLite snapshot and a + successful result is copied back transactionally. A timestamped raw backup + is taken first unless ``backup=False``. The surgery below is serialised across processes (see :func:`_cross_process_repair_lock`): the gateway service, the Desktop @@ -2762,18 +2978,11 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A # After _MAX_PERSISTENT_REPAIR_ATTEMPTS failures against the same # damaged file, stop retrying and surface a terminal, actionable error. if _persistent_repair_attempts_exhausted(db_path): - report["error"] = ( - f"automatic repair has already failed " - f"{_MAX_PERSISTENT_REPAIR_ATTEMPTS} times on this exact file — " - "the corruption is beyond the schema/FTS repair strategies " - "(likely b-tree page damage). Manual recovery required: restore " - f"a backup, or salvage with `sqlite3 {db_path} \".recover\"`. " - f"Delete {_repair_ledger_path(db_path).name} to force another " - "automatic attempt." - ) + report["error"] = _persistent_repair_exhausted_error(db_path) logger.error("state.db repair skipped: %s", report["error"]) return report + result = report with _cross_process_repair_lock(db_path) as holding_lock: if not holding_lock: # Another process is still inside its critical section. It may @@ -2782,39 +2991,52 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A if _db_opens_cleanly(db_path) is None: report["repaired"] = True report["strategy"] = "repaired_by_other_process" - return report - report["error"] = ( - "another process holds the state.db repair lock; skipped " - "schema surgery to avoid racing it" - ) - return report - - # The cross-process lock serialises repairers against each other; it - # says nothing about the gateway, Desktop or a CLI still holding the - # database open. Rewriting b-tree pages under a concurrent writer is - # what spread the 2026-08-18/19 damage out of the FTS shadow tables - # and into the canonical ones. The caller closes only its own - # connection — the incident process held seven descriptors on - # state.db — so probe for the rest before touching anything. - if _live_writer_holds_db(db_path): - report["error"] = ( - "a live writer still holds state.db; skipped schema surgery " - "to avoid tearing b-tree pages under a concurrent writer. " - "Stop the gateway (hermes gateway stop) and retry." - ) - logger.error("state.db repair skipped: %s", report["error"]) - return report - - result = _repair_state_db_schema_locked(db_path, backup=backup, report=report) - # Persist the outcome AFTER surgery, keyed on the post-attempt - # fingerprint — that is the file state the NEXT attempt's exhaustion - # probe will observe. Failures count toward the cross-restart cap; - # success clears the ledger. (A failing strategy that mutates the - # file re-keys the ledger and restarts the count: that keeps a - # genuinely NEW corruption event from inheriting a stale budget, - # while the backup dedupe/cap above bounds the disk cost either way.) - _record_repair_outcome(db_path, repaired=bool(result.get("repaired"))) - return result + else: + report["error"] = ( + "another process holds the state.db repair lock; skipped " + "schema surgery to avoid racing it" + ) + else: + # The fast check above avoids taking the lock for a known-exhausted + # image. Recheck after acquisition: a queued repairer can have + # recorded the final failure while this process waited, and this + # process must not start a fourth attempt. + if _persistent_repair_attempts_exhausted(db_path): + report["error"] = _persistent_repair_exhausted_error(db_path) + logger.error("state.db repair skipped: %s", report["error"]) + # Keep the existing WAL-holder preflight: it preserves the + # established fail-closed behaviour for active readers before we + # create a forensic backup. It is not the race defence; the + # retained exclusive guard inside the locked routine is what + # excludes writers continuously through promotion. DELETE-mode + # readers which this probe cannot see are still rejected by the + # later BEGIN EXCLUSIVE acquisition. + elif _live_writer_holds_db(db_path): + report["error"] = ( + "a live writer still holds state.db; skipped schema surgery " + "to avoid tearing b-tree pages under a concurrent writer. " + "Stop the gateway (hermes gateway stop) and retry." + ) + logger.error("state.db repair skipped: %s", report["error"]) + else: + result = _repair_state_db_schema_locked( + db_path, backup=backup, report=report + ) + # Environmental aborts happen before a strategy gets to mutate the + # isolated snapshot. They are retriable operating conditions, not + # proof that the damaged database exhausted a repair strategy. + # Keep that private signal out of the public report while + # successful health checks still clear a stale persistent failure + # record. This ledger update stays under the same cross-process + # lock as surgery, so two repairers cannot lose each other's + # attempt updates. A queued loser must not record at all: its + # owner is responsible for its outcome. + attempted = bool(result.pop("_repair_attempted", False)) + if attempted or result.get("repaired"): + _record_repair_outcome( + db_path, repaired=bool(result.get("repaired")) + ) + return result def _repair_state_db_schema_locked( @@ -2823,7 +3045,42 @@ def _repair_state_db_schema_locked( """Repair strategies for :func:`repair_state_db_schema`. Caller must hold the cross-process repair lock for *db_path*. + + The strategies run on a SCRATCH COPY and the result is copied back through + SQLite's transactional backup API only once it is proven to open cleanly. + A repair that does not succeed therefore cannot modify or lose committed + canonical data. In WAL mode SQLite may checkpoint already-committed WAL + frames into the main file while the exclusive guard is released; that is + not a repair mutation and does not change the committed database image. + + They used to run in place, and Strategy 2 ends in ``VACUUM``. VACUUM does + not preserve what it cannot parse: it rebuilds the file from the schema + SQLite can still read, so when the damage IS in the schema b-tree — page + 1's child pointers resolving to data pages, which is exactly the + ``malformed database schema ()`` class this function exists to handle — + every table hanging off the unreadable part is silently dropped. The probe + afterwards then correctly reports the file is STILL malformed, so the + function returns ``repaired=False`` and advises a manual restore, having + already destroyed the thing it was asked to save. Destroying the data and + reporting the repair failed are not mutually exclusive outcomes, and + nothing here treated them as a contradiction. + + The pre-repair backup (#69603) does not close this: it is a forensic + artefact that nothing reads back, so recovery still depends on a human + noticing a ``.malformed-backup-*`` file and knowing what to do with it. + Not mutating the original in the first place is the property that holds + without a human in the loop. """ + scratch = db_path.with_name(f"{db_path.name}.repair-scratch") + cleanup_error = _unlink_db_triple(scratch) + if cleanup_error is not None: + report["error"] = ( + "could not remove a stale repair snapshot before probing state.db: " + f"{cleanup_error}" + ) + logger.error("state.db repair aborted: %s", report["error"]) + return report + # Re-probe under the lock: a process we queued behind may have just # repaired the file, in which case redoing the surgery would undo its # work on a now-healthy DB (the repair/re-corrupt cascade this lock @@ -2837,12 +3094,9 @@ def _repair_state_db_schema_locked( bpath, backup_error = _backup_db_file(db_path) report["backup_path"] = str(bpath) if bpath else None if bpath is None: - # HARD STOP (#69603): every strategy below mutates the damaged - # file in place (FTS rebuild, REINDEX, writable_schema surgery, - # VACUUM). Without the pre-repair backup, the damaged DB is the - # only copy of the user's data — a failed or interrupted repair - # would then be unrecoverable. Abort and surface the reason - # instead of proceeding fail-open. + # HARD STOP (#69603). The forensic recovery image remains required + # when corruption defeats every strategy, even though the + # strategies themselves now run against an isolated snapshot. report["error"] = ( "pre-repair backup refused; aborting schema repair to avoid " f"mutating the only copy of the damaged DB: {backup_error}" @@ -2850,6 +3104,144 @@ def _repair_state_db_schema_locked( logger.error("state.db repair aborted: %s", report["error"]) return report + # The forensic copy intentionally happens before this guard: its raw-copy + # safety checks inspect real live holders and would be poisoned by our + # exclusive connection. Everything that can affect the repair image or + # live promotion happens only after writer exclusion is held. + with _exclusive_repair_db_guard(db_path) as (live_guard, guard_error): + if live_guard is None: + report["error"] = ( + "could not acquire exclusive state.db repair ownership; " + "skipped schema surgery to avoid overwriting a concurrent " + f"writer. Stop the gateway and retry: {guard_error}" + ) + if guard_error is not None and _repair_failure_consumes_attempt( + guard_error + ): + report["_repair_attempted"] = True + logger.error("state.db repair skipped: %s", report["error"]) + return report + + space_error = _repair_scratch_space_error(db_path) + if space_error is not None: + report["error"] = space_error + logger.error("state.db repair aborted: %s", report["error"]) + return report + + try: + # Reuse live_guard rather than opening a second source connection: + # the guard owns the exclusion, so a second connection could be + # blocked by our own EXCLUSIVE lock on some SQLite builds. + _copy_database_snapshot( + db_path, scratch, source_connection=live_guard + ) + except (OSError, sqlite3.Error, TimeoutError) as exc: + report["error"] = ( + f"could not stage a complete SQLite repair snapshot of {db_path}: {exc}" + ) + if _repair_failure_consumes_attempt(exc): + report["_repair_attempted"] = True + logger.error("state.db repair aborted: %s", report["error"]) + _unlink_db_triple(scratch) + return report + + try: + # This private marker is consumed by the outer wrapper. A strategy + # failure is a genuine repair outcome and consumes the persistent + # budget. A later promotion failure is classified separately: + # full disks, I/O, permission and lock failures are environmental + # aborts, not evidence that the strategy cannot repair this image. + report["_repair_attempted"] = True + _run_repair_strategies(scratch, report) + if report.get("repaired"): + try: + # Do not os.replace the live DB: Windows rejects + # replacement under open handles, while POSIX would leave + # those handles on the old inode. The same transaction- + # free guard that staged the live image receives the + # promotion, retaining writer exclusion throughout. + _copy_database_snapshot( + scratch, + db_path, + destination_connection=live_guard, + ) + except (OSError, sqlite3.Error, TimeoutError) as exc: + report["repaired"] = False + report["strategy"] = None + report["_repair_attempted"] = _repair_failure_consumes_attempt( + exc + ) + report["error"] = ( + "repaired snapshot could not be promoted transactionally: " + f"{exc}" + ) + logger.error("state.db repair promotion failed: %s", exc) + else: + logger.warning( + "state.db repaired via '%s' and promoted transactionally: %s", + report.get("strategy"), + db_path, + ) + if not report.get("repaired"): + # Logged HERE, not inside the strategies: they run against the + # scratch copy, and naming that throwaway path in the one + # message a human is meant to act on would send them to a file + # that no longer exists by the time they read it. + logger.error( + "state.db schema repair could not recover %s automatically " + "(no committed canonical data was modified or lost; backup: %s); " + "manual restore from backup may be required.", + db_path, + report["backup_path"], + ) + return report + finally: + # Never leave a half-repaired file beside the DB for a later probe + # — or a later human — to mistake for the real thing. + cleanup_error = _unlink_db_triple(scratch) + if cleanup_error is not None: + logger.warning( + "Could not remove state.db repair snapshot after repair: %s", + cleanup_error, + ) + + +def _unlink_db_triple(path: Path) -> Optional[str]: + """Remove *path* and every SQLite sidecar; return any cleanup failure.""" + failures: List[str] = [] + for suffix in ("", *_DB_SIDECAR_SUFFIXES): + victim = path if not suffix else path.with_name(path.name + suffix) + for attempt in range(10): + try: + victim.unlink() + break + except FileNotFoundError: + break + except PermissionError as exc: + # Windows may retain a just-closed SQLite handle for a few + # scheduler ticks. Bound the retry; a later backup open still + # fails safely if the handle truly remains live. + if _IS_WINDOWS and attempt < 9: + time.sleep(0.05) + continue + failures.append(f"{victim}: {exc}") + break + except OSError as exc: + failures.append(f"{victim}: {exc}") + break + return "; ".join(failures) or None + + +def _run_repair_strategies( + db_path: Path, report: Dict[str, Any] +) -> Dict[str, Any]: + """Escalating repair attempts, applied to *db_path* IN PLACE. + + Every strategy here mutates its argument — FTS rebuild, REINDEX, + ``writable_schema`` surgery, ``VACUUM``. It is therefore only ever called + by :func:`_repair_state_db_schema_locked` on a scratch copy that nothing + else holds open, never on the user's database. + """ # ── Strategy 0: rebuild FTS indexes in place (FTS write-corruption) ── # The FTS5 'rebuild' command rewrites the internal index from the canonical # content table. This is the recommended, least-destructive recovery for a @@ -2943,6 +3335,13 @@ def _repair_state_db_schema_locked( logger.warning("state.db dedup repair pass failed: %s", exc) # ── Strategy 2: drop all FTS schema, VACUUM, rebuild on next open ── + # + # The destructive one, and the reason this whole path now runs on a + # scratch copy. VACUUM rebuilds the file from the schema SQLite can still + # parse, so on a damaged schema b-tree it silently drops every table + # hanging off the unreadable part — and the probe below then correctly + # reports the result is still malformed. On a scratch copy that is merely + # a discarded attempt; on the live file it was data loss. try: conn = _connect_repair_durable(db_path) try: @@ -2971,12 +3370,8 @@ def _repair_state_db_schema_locked( except sqlite3.DatabaseError as exc: report["error"] = str(exc) - if not report["repaired"]: - logger.error( - "state.db schema repair could not recover %s automatically " - "(backup: %s); manual restore from backup may be required.", - db_path, report["backup_path"], - ) + # The "could not recover" log lives in the caller: it must name the user's + # database, not the scratch copy these strategies were handed. return report diff --git a/tests/test_state_db_repair_non_destructive.py b/tests/test_state_db_repair_non_destructive.py new file mode 100644 index 0000000000..0b00e3699c --- /dev/null +++ b/tests/test_state_db_repair_non_destructive.py @@ -0,0 +1,877 @@ +"""A failed state.db schema repair must preserve the recovery image. + +For rollback-journal databases the failed-repair invariant is byte identity: +the main file and its sidecars must be unchanged. WAL mode has an important +qualification: SQLite may checkpoint committed frames when a connection is +opened or closed, so the main file's bytes are not themselves the recovery +image. WAL tests therefore assert committed-row and recovery semantics, +while the byte-identity assertions remain on the DELETE/rollback-journal +path. + +Reported incident: the automatic repair path deleted a user's transcripts and +then reported that it had failed. + +``_repair_state_db_schema_locked`` ran its strategies on the live file, and +Strategy 2 ends in ``VACUUM``:: + + PRAGMA writable_schema=ON + DELETE FROM sqlite_master WHERE name LIKE 'messages_fts%' + PRAGMA writable_schema=OFF + VACUUM + +VACUUM does not preserve what it cannot parse — it rebuilds the file from the +schema SQLite can still read. When the damage IS in the schema b-tree (page +1's child pointers aimed at data pages, which is exactly the ``malformed +database schema ()`` class this function exists to handle), the rebuild drops +every table hanging off the unreadable part. Measured on the reporting +install: ``state.db`` went from 3048 pages / 29 sessions / 2537 messages to +113 pages, in place. + +The probe that follows then correctly reported the file was *still* malformed, +so the function returned ``repaired=False`` with "manual restore from backup +may be required" — after the only live copy had already been gutted. +Destroying the data and reporting the repair failed are not mutually exclusive +outcomes, and nothing in the code treated them as a contradiction. + +The pre-repair backup (#69603) is a forensic artefact, not a recovery path: +nothing reads it back. So the invariant under test is the stronger one — a +repair that does not succeed must not change the file at all — plus the +structural assertion that makes it hold: the strategies never receive the live +database. + +Fix under test: every strategy runs on a ``.repair-scratch`` snapshot and +is copied back through SQLite's transactional backup API only once the result +is proven to open cleanly. + +Mutation-checked: pointing ``_run_repair_strategies`` back at ``db_path`` +instead of the scratch copy fails +``test_failed_repair_leaves_the_original_byte_identical`` and +``test_strategies_never_receive_the_live_database``. +""" + +from __future__ import annotations + +import contextlib +import hashlib +import multiprocessing +import os +import shutil +import sqlite3 +import struct +import threading +import time +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import hermes_state +from hermes_state import repair_state_db_schema + +PAGE_SIZE = 4096 + + +def _writer_after_stage( + db_path: str, + ready: "multiprocessing.synchronize.Event", + start: "multiprocessing.synchronize.Event", + result: "multiprocessing.queues.Queue", +) -> None: + """Try one real cross-process write after staging has begun. + + The repair process owns the SQLite exclusion for the complete + stage/strategy/promotion interval. A writer is allowed to fail or to + wait until that interval ends and commit afterwards; what is forbidden is + a successful commit that promotion silently overwrites. + """ + conn = sqlite3.connect(db_path, timeout=0.75, isolation_level=None) + try: + ready.set() + if not start.wait(20): + result.put(("not-started", "repair did not reach staging")) + return + try: + conn.execute( + "INSERT INTO messages (body) VALUES ('committed-after-stage')" + ) + result.put(("committed", None)) + except sqlite3.Error as exc: + result.put(("failed", str(exc))) + finally: + conn.close() + + +def _make_repair_test_db(path: Path, *, journal_mode: str = "delete") -> None: + conn = sqlite3.connect(str(path), isolation_level=None) + try: + conn.execute("CREATE TABLE sessions (id INTEGER PRIMARY KEY, name TEXT)") + conn.execute("CREATE TABLE messages (id INTEGER PRIMARY KEY, body TEXT)") + conn.execute("INSERT INTO sessions (name) VALUES ('seed')") + conn.execute("INSERT INTO messages (body) VALUES ('seed')") + actual = conn.execute(f"PRAGMA journal_mode={journal_mode}").fetchone()[0] + if journal_mode == "wal" and actual != "wal": + pytest.skip("SQLite build/filesystem does not support WAL") + finally: + conn.close() + + +def _leave_hot_wal_row(db_path: str) -> None: + """Commit a WAL frame, then die without letting sqlite3 close the DB.""" + conn = sqlite3.connect(db_path, isolation_level=None) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA wal_autocheckpoint=0") + conn.execute("INSERT INTO messages (body) VALUES ('committed-wal-row')") + conn.commit() + # Deliberately bypass Connection.close(): the next process must recover + # the committed row from the hot WAL image, not from a pre-checkpoint main. + os._exit(0) + + +def _probe_repair_lock_from_child(db_path: str, result) -> None: + """Attempt the repair lock with a short timeout from another process.""" + hermes_state._REPAIR_LOCK_TIMEOUT_SECONDS = 0.5 + with hermes_state._cross_process_repair_lock(Path(db_path)) as holding: + result.put(holding) + + +def _write_populated_db(path: Path, *, sessions: int = 3, messages: int = 25) -> None: + conn = sqlite3.connect(str(path)) + conn.execute(f"PRAGMA page_size={PAGE_SIZE}") + conn.execute("CREATE TABLE sessions (id INTEGER PRIMARY KEY, name TEXT)") + conn.execute("CREATE TABLE messages (id INTEGER PRIMARY KEY, body TEXT)") + conn.executemany( + "INSERT INTO sessions (name) VALUES (?)", + [(f"session-{i}",) for i in range(sessions)], + ) + conn.executemany( + "INSERT INTO messages (body) VALUES (?)", + [(f"message body {i}" * 20,) for i in range(messages)], + ) + conn.commit() + conn.close() + + +def _break_the_schema_btree(path: Path) -> None: + """Aim page 1's rightmost child at a data page. + + This is the shape of the reported corruption: ``sqlite_master``'s b-tree + resolves to pages holding table content, so SQLite reports "malformed + database schema ()" — the parentheses empty because the bogus row's name + is not text. + """ + data = bytearray(path.read_bytes()) + page_count = struct.unpack(">I", data[28:32])[0] + assert page_count >= 3, "fixture needs a multi-page database" + # Byte 100 is page 1's b-tree header; offset 108 is the rightmost pointer + # on an interior page. Force page 1 to be interior and point it at the + # last page, which holds table data rather than schema records. + data[100] = 0x05 + struct.pack_into(">H", data, 103, 1) # one cell + struct.pack_into(">I", data, 108, page_count) # rightmost -> data page + struct.pack_into(">H", data, 112, PAGE_SIZE - 6) # cell pointer + struct.pack_into(">I", data, PAGE_SIZE - 6, page_count) + path.write_bytes(bytes(data)) + + +@pytest.fixture +def corrupt_db(tmp_path: Path) -> Path: + path = tmp_path / "state.db" + _write_populated_db(path) + _break_the_schema_btree(path) + with pytest.raises(sqlite3.DatabaseError): + conn = sqlite3.connect(str(path)) + try: + conn.execute("SELECT * FROM sessions").fetchall() + finally: + conn.close() + return path + + +# --------------------------------------------------------------------------- +# The structural guarantee +# --------------------------------------------------------------------------- + + +def test_strategies_never_receive_the_live_database(corrupt_db, monkeypatch): + """Every strategy mutates its argument in place, so the property that + makes them safe is simply that the argument is never the real file.""" + seen: list[Path] = [] + real = hermes_state._run_repair_strategies + + def spy(path, report): + seen.append(path) + return real(path, report) + + monkeypatch.setattr(hermes_state, "_run_repair_strategies", spy) + repair_state_db_schema(corrupt_db) + + assert seen, "the repair path did not run at all" + for path in seen: + assert path != corrupt_db + assert path.name.endswith(".repair-scratch") + + +# --------------------------------------------------------------------------- +# The regression +# --------------------------------------------------------------------------- + + +def test_failed_repair_leaves_the_original_byte_identical(corrupt_db): + before = hashlib.sha256(corrupt_db.read_bytes()).hexdigest() + + report = repair_state_db_schema(corrupt_db) + + after = hashlib.sha256(corrupt_db.read_bytes()).hexdigest() + assert not report.get("repaired"), ( + "fixture precondition: this corruption is not automatically repairable" + ) + assert before == after, ( + "a repair that FAILED rewrote the database anyway — this is the " + "reported data loss: 29 sessions / 2537 messages became 113 pages " + "while the function reported 'manual restore may be required'" + ) + + +def test_failed_repair_leaves_no_scratch_file_behind(corrupt_db): + """A half-repaired file beside the DB is a trap for the next probe.""" + repair_state_db_schema(corrupt_db) + leftovers = sorted(p.name for p in corrupt_db.parent.glob("*repair-scratch*")) + assert leftovers == [] + + +def test_failed_repair_still_takes_the_forensic_backup(corrupt_db): + """Non-destructive repair does not make the #69603 backup redundant.""" + report = repair_state_db_schema(corrupt_db) + assert report["backup_path"], "the pre-repair forensic copy is still required" + assert Path(report["backup_path"]).exists() + + +# --------------------------------------------------------------------------- +# ...and a repair that DOES succeed must still land on the original path +# --------------------------------------------------------------------------- + + +def test_successful_repair_is_promoted_over_the_original(tmp_path, monkeypatch): + """The scratch copy is a staging area, not a detour: a strategy that + heals the copy must leave the healed bytes at ``db_path``.""" + db = tmp_path / "state.db" + _write_populated_db(db) + # Force the "already healthy" short-circuit off so the staging path runs, + # and have the strategy pass mark a repair after writing a marker row. + monkeypatch.setattr( + hermes_state, "_db_opens_cleanly", lambda path: "forced-unhealthy" + ) + + def fake_strategies(scratch_path, report): + conn = sqlite3.connect(str(scratch_path)) + conn.execute("INSERT INTO sessions (name) VALUES ('healed-on-scratch')") + conn.commit() + conn.close() + report["repaired"] = True + report["strategy"] = "test_strategy" + return report + + monkeypatch.setattr(hermes_state, "_run_repair_strategies", fake_strategies) + + report = repair_state_db_schema(db) + assert report["repaired"] is True + + conn = sqlite3.connect(str(db)) + try: + names = [r[0] for r in conn.execute("SELECT name FROM sessions")] + finally: + conn.close() + assert "healed-on-scratch" in names, ( + "the repaired copy was never promoted over the original" + ) + assert not list(db.parent.glob("*repair-scratch*")) + + +@pytest.mark.parametrize("journal_mode", ("delete", "wal")) +def test_committed_writer_after_staging_is_never_lost( + tmp_path, monkeypatch, journal_mode +): + """A writer racing the repair lifecycle cannot be overwritten. + + This uses the real orchestration and both SQLite online-backup calls. The + strategy hook is only a deterministic latch: it writes the repaired + marker to the real scratch database, then gives a separate process a + chance to attempt its commit before promotion. A successful writer must + still be present after repair; a locked/timeout writer is also valid. + """ + db = tmp_path / "state.db" + _make_repair_test_db(db, journal_mode=journal_mode) + + monkeypatch.setattr( + hermes_state, "_db_opens_cleanly", lambda _path: "forced-unhealthy" + ) + ready = multiprocessing.get_context("spawn").Event() + start = multiprocessing.get_context("spawn").Event() + result = multiprocessing.get_context("spawn").Queue() + writer = multiprocessing.get_context("spawn").Process( + target=_writer_after_stage, + args=(str(db), ready, start, result), + ) + writer.start() + assert ready.wait(20), "writer process did not initialize" + + def staged_strategy(scratch_path, report): + with sqlite3.connect(str(scratch_path)) as conn: + conn.execute( + "INSERT INTO sessions (name) VALUES ('repaired-before-race')" + ) + conn.commit() + start.set() + # Make the race deterministic: an unguarded implementation lets the + # child commit here, after which promotion silently erases its row. + time.sleep(1.0) + report["repaired"] = True + report["strategy"] = "race_test" + return report + + monkeypatch.setattr(hermes_state, "_run_repair_strategies", staged_strategy) + report = repair_state_db_schema(db, backup=False) + writer.join(20) + if writer.is_alive(): + writer.terminate() + writer.join(5) + pytest.fail("writer process did not finish") + + outcome, detail = result.get(timeout=5) + assert outcome in {"failed", "committed"}, (outcome, detail) + assert report["repaired"] is True + + with sqlite3.connect(str(db)) as conn: + rows = { + body for (body,) in conn.execute("SELECT body FROM messages") + } + names = { + name for (name,) in conn.execute("SELECT name FROM sessions") + } + assert "repaired-before-race" in names + if outcome == "committed": + assert "committed-after-stage" in rows, ( + "a writer reported a successful commit, but transactional repair " + "promotion silently overwrote it" + ) + + +def test_environmental_aborts_do_not_burn_repair_ledger(tmp_path, monkeypatch): + """Three disk/staging aborts leave the actual strategy budget untouched.""" + db = tmp_path / "state.db" + _make_repair_test_db(db) + monkeypatch.setattr( + hermes_state, "_db_opens_cleanly", lambda _path: "forced-unhealthy" + ) + monkeypatch.setattr( + hermes_state, + "_repair_scratch_space_error", + lambda _path: "temporary disk pressure", + ) + + for _ in range(3): + report = repair_state_db_schema(db, backup=False) + assert report["repaired"] is False + assert report["error"] == "temporary disk pressure" + + ledger_path = hermes_state._repair_ledger_path(db) + assert not ledger_path.exists(), "environmental aborts must not consume attempts" + + monkeypatch.setattr(hermes_state, "_repair_scratch_space_error", lambda _path: None) + + def successful_strategy(scratch_path, report): + with sqlite3.connect(str(scratch_path)) as conn: + conn.execute("INSERT INTO sessions (name) VALUES ('after-aborts')") + conn.commit() + report["repaired"] = True + report["strategy"] = "after_environmental_aborts" + return report + + monkeypatch.setattr(hermes_state, "_run_repair_strategies", successful_strategy) + report = repair_state_db_schema(db, backup=False) + assert report["repaired"] is True + assert report["strategy"] == "after_environmental_aborts" + + +def test_actual_strategy_failure_still_consumes_one_attempt(tmp_path, monkeypatch): + db = tmp_path / "state.db" + _make_repair_test_db(db) + monkeypatch.setattr( + hermes_state, "_db_opens_cleanly", lambda _path: "forced-unhealthy" + ) + + def failed_strategy(_scratch_path, report): + report["repaired"] = False + report["strategy"] = None + return report + + monkeypatch.setattr(hermes_state, "_run_repair_strategies", failed_strategy) + report = repair_state_db_schema(db, backup=False) + assert report["repaired"] is False + ledger = hermes_state._read_repair_ledger(db) + assert ledger["failed_attempts"] == 1 + + +def test_repair_outcome_is_recorded_while_cross_process_lock_is_held( + tmp_path, monkeypatch +): + """Ledger publication must remain inside the repairer's critical section.""" + db = tmp_path / "state.db" + _make_repair_test_db(db) + monkeypatch.setattr( + hermes_state, "_db_opens_cleanly", lambda _path: "forced-unhealthy" + ) + + def failed_strategy(_scratch_path, report): + report["repaired"] = False + report["strategy"] = None + return report + + monkeypatch.setattr(hermes_state, "_run_repair_strategies", failed_strategy) + observed = [] + lock_released = threading.Event() + real_repair_lock = hermes_state._cross_process_repair_lock + + @contextlib.contextmanager + def tracking_repair_lock(path): + with real_repair_lock(path) as holding: + yield holding + lock_released.set() + + monkeypatch.setattr( + hermes_state, "_cross_process_repair_lock", tracking_repair_lock + ) + + def record_outcome(_db_path, *, repaired, fingerprint=None): + assert not lock_released.is_set(), ( + "repair outcome was recorded after the cross-process lock released" + ) + context = multiprocessing.get_context("spawn") + result = context.Queue() + probe = context.Process( + target=_probe_repair_lock_from_child, + args=(str(db), result), + ) + probe.start() + try: + observed.append(result.get(timeout=5)) + finally: + probe.join(5) + if probe.is_alive(): + probe.terminate() + probe.join(5) + assert repaired is False + + monkeypatch.setattr(hermes_state, "_record_repair_outcome", record_outcome) + report = repair_state_db_schema(db, backup=False) + + assert report["repaired"] is False + assert observed == [False], ( + "repair outcome was recorded after releasing the cross-process lock" + ) + + +def test_exhaustion_is_rechecked_after_acquiring_repair_lock(tmp_path, monkeypatch): + """A queued repairer must not start surgery on a newly exhausted ledger.""" + db = tmp_path / "state.db" + _make_repair_test_db(db) + exhaustion_checks = [] + + def exhaustion_probe(_db_path): + exhaustion_checks.append(True) + # The first caller's pre-lock view is stale; the queued view is + # terminal because another repairer just recorded the final failure. + return len(exhaustion_checks) >= 2 + + monkeypatch.setattr( + hermes_state, "_persistent_repair_attempts_exhausted", exhaustion_probe + ) + monkeypatch.setattr(hermes_state, "_live_writer_holds_db", lambda _path: False) + surgery_calls = [] + + def unexpected_surgery(_db_path, *, backup, report): + surgery_calls.append(True) + return report + + monkeypatch.setattr( + hermes_state, "_repair_state_db_schema_locked", unexpected_surgery + ) + + report = repair_state_db_schema(db, backup=False) + + assert report["repaired"] is False + assert "automatic repair has already failed" in report["error"] + assert len(exhaustion_checks) == 2 + assert surgery_calls == [], "surgery ran despite the under-lock exhaustion recheck" + + +def test_scratch_budget_counts_sidecars_in_vacuum_multiplier(tmp_path, monkeypatch): + """A WAL-heavy snapshot must reserve 3x snapshot bytes, not 2x main.""" + db = tmp_path / "state.db" + main_bytes = 100 + wal_bytes = 900 + db.write_bytes(b"m" * main_bytes) + db.with_name(db.name + "-wal").write_bytes(b"w" * wal_bytes) + total = 10_000_000_000 + headroom = hermes_state._repair_backup_headroom_bytes(total) + # This exactly satisfies the obsolete ``snapshot + 2*main + headroom`` + # calculation, but is below the corrected ``3*snapshot + headroom``. + old_required = main_bytes + wal_bytes + (2 * main_bytes) + headroom + monkeypatch.setattr( + shutil, + "disk_usage", + lambda _path: SimpleNamespace(total=total, free=old_required), + ) + + error = hermes_state._repair_scratch_space_error(db) + + assert error is not None + assert "VACUUM may need another" in error + + +def test_environmental_promotion_failures_do_not_burn_ledger( + tmp_path, monkeypatch +): + """Transient promotion disk errors remain retriable across three passes.""" + db = tmp_path / "state.db" + _make_repair_test_db(db) + monkeypatch.setattr( + hermes_state, "_db_opens_cleanly", lambda _path: "forced-unhealthy" + ) + + def successful_strategy(_scratch_path, report): + report["repaired"] = True + report["strategy"] = "promotion_environment_test" + return report + + monkeypatch.setattr(hermes_state, "_run_repair_strategies", successful_strategy) + real_copy = hermes_state._copy_database_snapshot + copy_calls = 0 + + def fail_promotion_three_times(source, destination, **kwargs): + nonlocal copy_calls + copy_calls += 1 + if copy_calls in (2, 4, 6): + raise sqlite3.OperationalError("database or disk is full") + return real_copy(source, destination, **kwargs) + + monkeypatch.setattr( + hermes_state, "_copy_database_snapshot", fail_promotion_three_times + ) + for _ in range(3): + report = repair_state_db_schema(db, backup=False) + assert report["repaired"] is False + assert "could not be promoted" in report["error"] + + ledger = hermes_state._read_repair_ledger(db) + assert ledger.get("failed_attempts", 0) == 0 + + monkeypatch.setattr(hermes_state, "_copy_database_snapshot", real_copy) + report = repair_state_db_schema(db, backup=False) + assert report["repaired"] is True + assert report["strategy"] == "promotion_environment_test" + + +def test_corrupt_promotion_failure_consumes_one_attempt(tmp_path, monkeypatch): + """A deterministic malformed-image promotion failure is budgeted.""" + db = tmp_path / "state.db" + _make_repair_test_db(db) + monkeypatch.setattr( + hermes_state, "_db_opens_cleanly", lambda _path: "forced-unhealthy" + ) + + def successful_strategy(_scratch_path, report): + report["repaired"] = True + report["strategy"] = "corrupt-promotion-test" + return report + + monkeypatch.setattr(hermes_state, "_run_repair_strategies", successful_strategy) + real_copy = hermes_state._copy_database_snapshot + calls = 0 + + def fail_promotion_with_corruption(source, destination, **kwargs): + nonlocal calls + calls += 1 + if calls == 2: + raise sqlite3.DatabaseError("database disk image is malformed") + return real_copy(source, destination, **kwargs) + + monkeypatch.setattr( + hermes_state, "_copy_database_snapshot", fail_promotion_with_corruption + ) + report = repair_state_db_schema(db, backup=False) + + assert report["repaired"] is False + assert hermes_state._read_repair_ledger(db)["failed_attempts"] == 1 + + +def test_snapshot_includes_committed_wal_frames(tmp_path): + """The recovery image includes commits not checkpointed to the WAL main.""" + db = tmp_path / "state.db" + conn = sqlite3.connect(str(db), isolation_level=None) + try: + conn.execute("CREATE TABLE messages (body TEXT)") + assert conn.execute("PRAGMA journal_mode=WAL").fetchone()[0] == "wal" + conn.execute("PRAGMA wal_autocheckpoint=0") + conn.execute("INSERT INTO messages VALUES ('committed-only-in-wal')") + assert db.with_name(db.name + "-wal").exists() + + scratch = tmp_path / "state.db.repair-scratch" + hermes_state._copy_database_snapshot(db, scratch) + with sqlite3.connect(str(scratch)) as check: + assert check.execute("SELECT body FROM messages").fetchall() == [ + ("committed-only-in-wal",) + ] + finally: + conn.close() + + +@pytest.mark.requires_wal +def test_failed_wal_repair_preserves_committed_rows_semantically( + tmp_path, monkeypatch +): + """WAL correctness is about committed recovery state, not main-file bytes.""" + db = tmp_path / "state.db" + _make_repair_test_db(db) + context = multiprocessing.get_context("spawn") + writer = context.Process(target=_leave_hot_wal_row, args=(str(db),)) + writer.start() + writer.join(20) + if writer.is_alive(): + writer.terminate() + writer.join(5) + pytest.fail("hot-WAL fixture process did not finish") + assert writer.exitcode == 0, "hot-WAL fixture process failed" + if not db.with_name(db.name + "-wal").exists(): + pytest.skip("SQLite/filesystem did not retain a hot WAL sidecar") + + monkeypatch.setattr( + hermes_state, "_db_opens_cleanly", lambda _path: "forced-unhealthy" + ) + + strategy_calls = [] + + def failed_strategy(_scratch_path, report): + strategy_calls.append(True) + report["repaired"] = False + report["strategy"] = None + return report + + monkeypatch.setattr(hermes_state, "_run_repair_strategies", failed_strategy) + report = repair_state_db_schema(db, backup=False) + assert report["repaired"] is False + assert strategy_calls == [True], "the test must exercise strategy failure" + + with sqlite3.connect(str(db)) as conn: + bodies = { + body for (body,) in conn.execute("SELECT body FROM messages") + } + assert {"seed", "committed-wal-row"} <= bodies + + +def test_snapshot_deadline_has_a_floor_and_scales_with_source_size( + tmp_path, monkeypatch +): + """Large snapshots get more than the lock floor without huge fixtures.""" + deadline = getattr(hermes_state, "_repair_snapshot_timeout_seconds", None) + assert callable(deadline), "repair snapshots need a size-scaled deadline" + # Lower the throughput only for this arithmetic test so a 72 MiB fixture + # crosses the floor without allocating a multi-GB file. + monkeypatch.setattr( + hermes_state, "_REPAIR_SNAPSHOT_MIN_THROUGHPUT_BYTES_PER_SECOND", 256 * 1024 + ) + + db = tmp_path / "state.db" + db.write_bytes(b"x" * PAGE_SIZE) + small = deadline(db) + # A modest sparse-ish fixture is enough to cross the 120-second floor on + # the production throughput constant; no multi-GB allocation is needed. + with db.open("ab") as fh: + fh.truncate(64 * 1024 * 1024) + db.with_name(db.name + "-wal").write_bytes(b"w" * (8 * 1024 * 1024)) + large = deadline(db) + assert small >= hermes_state._REPAIR_LOCK_TIMEOUT_SECONDS + assert large > small + + +def test_transactional_promotion_preserves_a_live_wal_reader(tmp_path): + """Promotion must not replace the inode behind an existing reader.""" + live = tmp_path / "state.db" + scratch = tmp_path / "scratch.db" + for path, value in ((live, "old"), (scratch, "repaired")): + with sqlite3.connect(str(path)) as conn: + conn.execute("CREATE TABLE messages (body TEXT)") + conn.execute("INSERT INTO messages VALUES (?)", (value,)) + conn.commit() + if path == live: + assert conn.execute("PRAGMA journal_mode=WAL").fetchone()[0] == "wal" + + reader = sqlite3.connect(str(live), isolation_level=None) + try: + reader.execute("BEGIN") + assert reader.execute("SELECT body FROM messages").fetchall() == [("old",)] + + hermes_state._copy_database_snapshot(scratch, live) + + assert reader.execute("SELECT body FROM messages").fetchall() == [("old",)] + with sqlite3.connect(str(live)) as fresh: + assert fresh.execute("SELECT body FROM messages").fetchall() == [ + ("repaired",) + ] + finally: + reader.execute("ROLLBACK") + reader.close() + + +def test_interrupted_snapshot_rolls_back_destination(tmp_path, monkeypatch): + source = tmp_path / "source.db" + destination = tmp_path / "destination.db" + with sqlite3.connect(str(source)) as conn: + conn.execute("CREATE TABLE payloads (body BLOB)") + conn.executemany( + "INSERT INTO payloads VALUES (?)", + [(b"x" * PAGE_SIZE,) for _ in range(400)], + ) + conn.commit() + with sqlite3.connect(str(destination)) as conn: + conn.execute("CREATE TABLE marker (value TEXT)") + conn.execute("INSERT INTO marker VALUES ('original')") + conn.commit() + + ticks = iter((0.0, hermes_state._REPAIR_LOCK_TIMEOUT_SECONDS + 1.0)) + monkeypatch.setattr(hermes_state.time, "monotonic", lambda: next(ticks)) + + with pytest.raises(TimeoutError): + hermes_state._copy_database_snapshot(source, destination) + + with sqlite3.connect(str(destination)) as conn: + assert conn.execute("SELECT value FROM marker").fetchall() == [("original",)] + + +def test_failed_promotion_returns_failure_and_preserves_original(tmp_path, monkeypatch): + db = tmp_path / "state.db" + _write_populated_db(db) + before = hashlib.sha256(db.read_bytes()).hexdigest() + real_copy = hermes_state._copy_database_snapshot + calls = 0 + + def fail_second_copy(source, destination, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + return real_copy(source, destination, **kwargs) + raise sqlite3.OperationalError("destination is busy") + + def fake_strategies(_scratch, report): + report["repaired"] = True + report["strategy"] = "test_strategy" + return report + + monkeypatch.setattr( + hermes_state, "_db_opens_cleanly", lambda _path: "forced-unhealthy" + ) + monkeypatch.setattr(hermes_state, "_copy_database_snapshot", fail_second_copy) + monkeypatch.setattr(hermes_state, "_run_repair_strategies", fake_strategies) + + report = repair_state_db_schema(db, backup=False) + + assert report["repaired"] is False + assert report["strategy"] is None + assert "could not be promoted" in report["error"] + assert hashlib.sha256(db.read_bytes()).hexdigest() == before + assert not list(tmp_path.glob("*repair-scratch*")) + + +def test_scratch_space_guard_accounts_for_snapshot_and_vacuum(tmp_path, monkeypatch): + db = tmp_path / "state.db" + db.write_bytes(b"x" * 4096) + headroom = hermes_state._repair_backup_headroom_bytes(10_000_000_000) + free = (3 * db.stat().st_size) + headroom - 1 + usage = SimpleNamespace(total=10_000_000_000, free=free) + monkeypatch.setattr(shutil, "disk_usage", lambda _path: usage) + + error = hermes_state._repair_scratch_space_error(db) + + assert error is not None + assert "VACUUM may need" in error + + +def test_stale_scratch_is_removed_before_health_check(tmp_path, monkeypatch): + db = tmp_path / "state.db" + _write_populated_db(db) + scratch = tmp_path / "state.db.repair-scratch" + scratch.write_bytes(b"crash debris" * 1000) + checks: list[str] = [] + + def fake_health(_path): + checks.append("health") + assert not scratch.exists() + return None + + monkeypatch.setattr(hermes_state, "_db_opens_cleanly", fake_health) + + report = repair_state_db_schema(db, backup=False) + + assert report["repaired"] is True + assert report["strategy"] == "already_healthy" + assert checks == ["health"] + assert not scratch.exists() + + +def test_stale_scratch_is_removed_before_space_check(tmp_path, monkeypatch): + db = tmp_path / "state.db" + _write_populated_db(db) + scratch = tmp_path / "state.db.repair-scratch" + scratch.write_bytes(b"crash debris" * 1000) + checked_space = False + + monkeypatch.setattr( + hermes_state, "_db_opens_cleanly", lambda _path: "forced-unhealthy" + ) + + def fake_space_check(_path): + nonlocal checked_space + checked_space = True + assert not scratch.exists() + return "forced low space" + + monkeypatch.setattr( + hermes_state, "_repair_scratch_space_error", fake_space_check + ) + + report = repair_state_db_schema(db, backup=False) + + assert report["repaired"] is False + assert report["error"] == "forced low space" + assert checked_space is True + assert not scratch.exists() + + +def test_stale_scratch_cleanup_failure_aborts_before_probe(tmp_path, monkeypatch): + db = tmp_path / "state.db" + _write_populated_db(db) + probed = False + + def fake_health(_path): + nonlocal probed + probed = True + return "forced-unhealthy" + + monkeypatch.setattr(hermes_state, "_db_opens_cleanly", fake_health) + monkeypatch.setattr( + hermes_state, "_unlink_db_triple", lambda _path: "scratch is locked" + ) + report = { + "repaired": False, + "strategy": None, + "backup_path": None, + "error": None, + } + + result = hermes_state._repair_state_db_schema_locked( + db, backup=False, report=report + ) + + assert result["repaired"] is False + assert "stale repair snapshot" in result["error"] + assert probed is False diff --git a/tests/test_state_db_write_durability.py b/tests/test_state_db_write_durability.py index ac306d8254..4ba56e49f5 100644 --- a/tests/test_state_db_write_durability.py +++ b/tests/test_state_db_write_durability.py @@ -29,7 +29,7 @@ only the repair-connection durability half.) from __future__ import annotations -import re +import ast import sqlite3 import sys from pathlib import Path @@ -102,23 +102,47 @@ def test_repair_path_has_no_bare_connects() -> None: Source-level guard: the bare form is exactly what regressed, and a unit test on the helper alone would not notice a sixth site being added. """ - source = Path(hermes_state.__file__).read_text() - pattern = r"^\s*conn = sqlite3\.connect\(str\(db_path\), isolation_level=None\)" + source = Path(hermes_state.__file__).read_text(encoding="utf-8") + tree = ast.parse(source, filename=str(hermes_state.__file__)) - # The one legitimate bare connect is inside the helper itself; everything - # after that definition must go through it. - helper = source.index("def _connect_repair_durable(") - body_end = source.index("\ndef ", helper + 1) - inside_helper = re.findall(pattern, source[helper:body_end], flags=re.MULTILINE) - assert len(inside_helper) == 1, ( - "_connect_repair_durable no longer opens the connection itself" + def is_db_path_connect(node: ast.AST) -> bool: + if not isinstance(node, ast.Call): + return False + callee = node.func + if not ( + isinstance(callee, ast.Attribute) + and isinstance(callee.value, ast.Name) + and callee.value.id == "sqlite3" + and callee.attr == "connect" + ): + return False + if not node.args: + return False + first = node.args[0] + return ( + isinstance(first, ast.Call) + and isinstance(first.func, ast.Name) + and first.func.id == "str" + and len(first.args) == 1 + and isinstance(first.args[0], ast.Name) + and first.args[0].id == "db_path" + ) + + helper = next( + node + for node in tree.body + if isinstance(node, ast.FunctionDef) + and node.name == "_connect_repair_durable" + ) + helper_calls = [node for node in ast.walk(helper) if is_db_path_connect(node)] + assert len(helper_calls) == 1, ( + "_connect_repair_durable must own exactly one sqlite3.connect(str(db_path), ...)" ) - elsewhere = re.findall( - pattern, source[:helper] + source[body_end:], flags=re.MULTILINE - ) + all_calls = [node for node in ast.walk(tree) if is_db_path_connect(node)] + elsewhere = [node for node in all_calls if node not in helper_calls] assert elsewhere == [], ( - f"{len(elsewhere)} repair-path connection(s) still bypass " + f"{len(elsewhere)} repair/probe connection(s) still bypass " "_connect_repair_durable() and write state.db without the macOS " "fsync barriers" )