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" )