From 923d86e09953ed6785eb5b24d218438c60092d79 Mon Sep 17 00:00:00 2001 From: Ernst Bablick Date: Wed, 22 Jul 2026 22:37:51 +0200 Subject: [PATCH] fix(state): serialize state.db schema surgery across processes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `repair_state_db_schema()` performs `PRAGMA writable_schema=ON` + `sqlite_master` surgery + `VACUUM` on a private connection. The only guard around it is `_repair_attempt_lock`, a `threading.Lock`, whose docstring claims it "serialises concurrent web_server / gateway opens" — but a threading lock covers threads inside one interpreter, not processes. A normal host runs four independent processes against the same state.db: the gateway service, the Desktop app's own `hermes serve` backend (it spawns one per launch, not a thin client), interactive CLI sessions, and the TUI slash worker. When two of them hit a malformed DB, both entered the critical section and each ran the full surgery while the other was mid-rewrite. Observed as a repair/re-corrupt cascade: the DB is repaired, then re-corrupts minutes later, repeatedly. Two fixes: 1. Wrap the surgery in a bounded `flock` on `.repair.lock`. `flock` is the right primitive — the kernel drops it when the holder dies, so a crashed repairer cannot wedge future repairs the way a pidfile would. The acquire is bounded (#36644's failure shape) and, unlike the kanban init lock, a caller that times out must NOT proceed: here "proceed anyway" is exactly the unsafe interleaving. It re-probes instead, and reports success if the holder already healed the file. Under the lock, the existing `_db_opens_cleanly()` check becomes a double-check: a queued process finds the DB healthy and returns `already_healthy` rather than re-running surgery on a repaired DB. 2. Bump the schema cookie after direct `sqlite_master` edits. Ordinary DDL bumps it for free and every other connection compares it before running a prepared statement — that is how they learn to drop a cached schema. Editing `sqlite_master` under `writable_schema=ON` does not, so live connections in other processes kept writing `messages` rows through triggers into `messages_fts*` shadow tables the surgery had just deleted. SQLite's writable_schema docs call out incrementing `schema_version` as the required companion to such an edit. Tests: four new cases in tests/test_state_db_malformed_repair.py, all using real child processes and a real flock. All four fail on main and pass with this change; the concurrency case asserts exactly one `malformed-backup-*` file is produced by two simultaneous repairers (two on main). Full state suite: 558 passed. Complements #43742, which makes the *in-process* claim loser retry rather than raise; it explicitly leaves `repair_state_db_schema()` unchanged and does nothing cross-process. The two are independent and compose. Co-Authored-By: Claude Opus 4.8 --- hermes_state.py | 159 +++++++++++++++++++++++ tests/test_state_db_malformed_repair.py | 162 ++++++++++++++++++++++++ 2 files changed, 321 insertions(+) diff --git a/hermes_state.py b/hermes_state.py index 7b6871023868b..b68f7992f4f18 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -16,6 +16,7 @@ Key design decisions: import asyncio import atexit +import contextlib import errno import hashlib import json @@ -1459,6 +1460,127 @@ def _claim_repair_attempt(db_path: Path) -> bool: return True +# Cross-process serialisation for the schema-surgery paths below. The +# ``_repair_attempt_lock`` above is a ``threading.Lock`` — it only covers +# threads inside ONE interpreter, yet a normal Hermes host runs several +# independent processes against the same ``state.db``: the gateway service, +# the Desktop app's own ``hermes serve`` backend, interactive CLI sessions, +# and the TUI slash worker. Two of those hitting a malformed DB at once each +# ran the full ``writable_schema`` surgery + ``VACUUM`` on their own private +# connection, with nothing serialising them. +# +# The timeout is sized for the slowest legitimate holder — a ``VACUUM`` over a +# multi-GB DB in strategy 2. Waiting that long is not a new stall: before this +# lock the losing caller spent the same minutes running its own surgery, it +# just did so on top of the winner's. +_REPAIR_LOCK_TIMEOUT_SECONDS = 120.0 +_REPAIR_LOCK_POLL_SECONDS = 0.1 +_IS_WINDOWS = sys.platform == "win32" + + +@contextlib.contextmanager +def _cross_process_repair_lock(db_path: Path): + """Serialize state.db schema surgery across processes. + + Yields True when this process holds the repair lock for *db_path*, False + when the bounded acquire timed out. Unlike the kanban init lock — whose + critical section is idempotent, so proceeding without the lock is merely + redundant work — proceeding here would be exactly the unsafe interleaving + we are trying to prevent, so a caller that gets False must NOT do surgery. + + ``flock`` is the right primitive for this: the kernel drops the lock when + the holding process dies, so a crashed repairer cannot leave a stale lock + that wedges every future repair (a pidfile would). The acquire is still + bounded because a *live* repairer can legitimately sit in ``VACUUM`` for + minutes on a large DB, and an unbounded wait would hang the caller's open + with no traceback (the failure shape of #36644). + """ + lock_path = db_path.with_name(db_path.name + ".repair.lock") + try: + lock_path.parent.mkdir(parents=True, exist_ok=True) + handle = lock_path.open("a+b") + except OSError as exc: + # Read-only dir, exhausted fds, exotic filesystem: fall back to the + # in-process behaviour that shipped before this lock existed rather + # than refusing to repair a DB we could otherwise heal. + logger.warning( + "Could not open state.db repair lock %s (%s) — proceeding with " + "in-process serialisation only.", lock_path, exc, + ) + yield True + return + + acquired = False + try: + deadline = time.monotonic() + _REPAIR_LOCK_TIMEOUT_SECONDS + while True: + try: + if _IS_WINDOWS: + import msvcrt + + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + acquired = True + break + except (BlockingIOError, OSError): + if time.monotonic() >= deadline: + break + time.sleep(_REPAIR_LOCK_POLL_SECONDS) + if not acquired: + logger.warning( + "state.db repair lock %s held by another process for more " + "than %.0fs — skipping schema surgery in this process to " + "avoid racing the repairer.", + lock_path, _REPAIR_LOCK_TIMEOUT_SECONDS, + ) + yield acquired + finally: + try: + if acquired: + if _IS_WINDOWS: + import msvcrt + + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + except OSError: # pragma: no cover - best effort release + pass + finally: + handle.close() + + +def _bump_schema_cookie(conn: sqlite3.Connection) -> None: + """Increment the schema cookie after direct ``sqlite_master`` surgery. + + Ordinary DDL bumps this counter for free, and every other connection + compares it before running a prepared statement — that is how they learn + to discard a cached schema. Editing ``sqlite_master`` under + ``PRAGMA writable_schema=ON`` does NOT bump it, so live connections in + other processes keep compiling statements against the schema we just + deleted objects from — e.g. writing ``messages`` rows through triggers + into ``messages_fts*`` shadow tables that no longer exist. SQLite's + writable_schema documentation calls out incrementing ``schema_version`` + as the required companion to such an edit. + + Best-effort and never raises: a failed bump leaves exactly the + pre-existing behaviour, and the repair itself is still worth completing. + """ + try: + current = conn.execute("PRAGMA schema_version").fetchone()[0] + # Wraps within the 32-bit signed range SQLite stores this in; the + # comparison other connections make is equality, not ordering. + conn.execute(f"PRAGMA schema_version={(int(current) + 1) & 0x7FFFFFFF}") + except (sqlite3.DatabaseError, TypeError, IndexError) as exc: + logger.warning("Could not bump state.db schema cookie: %s", exc) + + def _backup_db_file(db_path: Path) -> Optional[Path]: """Copy a (possibly malformed) DB file to a timestamped backup beside it. @@ -1743,6 +1865,12 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A Canonical ``sessions`` / ``messages`` rows are never modified. 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 + app's backend and interactive CLI sessions all open the same file, and + two of them running ``writable_schema`` surgery concurrently is itself a + corruption source. + Returns a report dict: ``{repaired: bool, strategy: str|None, backup_path: str|None, error: str|None}``. """ @@ -1758,6 +1886,34 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A report["error"] = f"{db_path} does not exist" return 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 + # nonetheless have healed the file already (long VACUUM after a + # successful strategy), so re-probe before reporting failure. + 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 + return _repair_state_db_schema_locked(db_path, backup=backup, report=report) + + +def _repair_state_db_schema_locked( + db_path: Path, *, backup: bool, report: Dict[str, Any] +) -> Dict[str, Any]: + """Repair strategies for :func:`repair_state_db_schema`. + + Caller must hold the cross-process repair lock for *db_path*. + """ + # 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 + # exists to break). if _db_opens_cleanly(db_path) is None: report["repaired"] = True report["strategy"] = "already_healthy" @@ -1839,6 +1995,8 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A "WHERE type IS ? AND name IS ? AND rowid <> ?", (type_, name, keep), ) + if dupes: + _bump_schema_cookie(conn) conn.execute("PRAGMA writable_schema=OFF") conn.commit() finally: @@ -1860,6 +2018,7 @@ def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, A try: conn.execute("PRAGMA writable_schema=ON") conn.execute("DELETE FROM sqlite_master WHERE name LIKE 'messages_fts%'") + _bump_schema_cookie(conn) conn.execute("PRAGMA writable_schema=OFF") conn.commit() conn.execute("VACUUM") diff --git a/tests/test_state_db_malformed_repair.py b/tests/test_state_db_malformed_repair.py index 49a5e24d43ca2..1002794710570 100644 --- a/tests/test_state_db_malformed_repair.py +++ b/tests/test_state_db_malformed_repair.py @@ -12,7 +12,11 @@ journal_mode in apply_wal_with_fallback), before _init_schema runs — so it cannot be handled at the FTS-rebuild layer. These tests verify the sqlite_master surgery path recovers the canonical data and self-heals on open. """ +import contextlib +import json import sqlite3 +import subprocess +import sys import uuid from pathlib import Path @@ -395,3 +399,161 @@ def test_repair_stale_btree_index_preserves_rows(tmp_path): db.close() +# --------------------------------------------------------------------------- +# Cross-process serialisation of the schema surgery +# --------------------------------------------------------------------------- +# A normal host runs several independent processes against one state.db: the +# gateway service, the Desktop app's own `hermes serve` backend, interactive +# CLI sessions and the TUI slash worker. `_repair_attempt_lock` is a +# threading.Lock and covers none of that, so two of them hitting a malformed +# DB at once each ran the full writable_schema surgery + VACUUM on a private +# connection — one repairing while the other was mid-surgery. + + +_HOLD_LOCK_SCRIPT = """ +import sys, time, fcntl, pathlib +sys.path.insert(0, {root!r}) +lock_path = pathlib.Path({lock!r}) +handle = lock_path.open("a+b") +fcntl.flock(handle.fileno(), fcntl.LOCK_EX) +print("locked", flush=True) +time.sleep({hold}) +""" + + +@contextlib.contextmanager +def _lock_held_by_other_process(db_path: Path, hold_seconds: float = 30.0): + """Hold the repair flock for *db_path* in a real child process.""" + script = _HOLD_LOCK_SCRIPT.format( + root=str(Path(hermes_state.__file__).parent), + lock=str(db_path.with_name(db_path.name + ".repair.lock")), + hold=hold_seconds, + ) + proc = subprocess.Popen( + [sys.executable, "-c", script], + stdout=subprocess.PIPE, text=True, + ) + try: + # Wait for the child to actually own the lock before yielding. + assert proc.stdout.readline().strip() == "locked" + yield + finally: + proc.kill() + proc.wait(timeout=10) + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX flock test") +def test_repair_skips_surgery_while_another_process_holds_the_lock( + tmp_path, monkeypatch +): + """The losing process must NOT run writable_schema surgery in parallel.""" + db_path = tmp_path / "state.db" + _build_healthy_db(db_path) + _corrupt_duplicate_fts(db_path) + monkeypatch.setattr(hermes_state, "_REPAIR_LOCK_TIMEOUT_SECONDS", 0.5) + + with _lock_held_by_other_process(db_path): + report = repair_state_db_schema(db_path) + + assert report["repaired"] is False + assert "repair lock" in (report["error"] or "") + # No surgery ran: no backup was taken and the DB is still malformed. + assert report["backup_path"] is None + assert not list(tmp_path.glob("state.db.malformed-backup-*")) + assert hermes_state._db_opens_cleanly(db_path) is not None + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX flock test") +def test_repair_reports_success_when_the_holder_already_healed_the_db( + tmp_path, monkeypatch +): + """Timing out against a healthy DB is a success, not an error.""" + db_path = tmp_path / "state.db" + _build_healthy_db(db_path) + monkeypatch.setattr(hermes_state, "_REPAIR_LOCK_TIMEOUT_SECONDS", 0.5) + + with _lock_held_by_other_process(db_path): + report = repair_state_db_schema(db_path) + + assert report["repaired"] is True + assert report["strategy"] == "repaired_by_other_process" + + +_REPAIR_SCRIPT = """ +import sys, json +sys.path.insert(0, {root!r}) +from hermes_state import repair_state_db_schema +print(json.dumps(repair_state_db_schema({db!r})), flush=True) +""" + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX flock test") +def test_two_processes_repairing_at_once_perform_surgery_once(tmp_path): + """Concurrent repairers serialise; the loser sees a healed DB and stops. + + Without the cross-process lock both processes back up and operate on + sqlite_master, i.e. one runs surgery on a database the other is + simultaneously rewriting. The backup count is the observable proxy for + "how many processes entered the critical section". + """ + db_path = tmp_path / "state.db" + _build_healthy_db(db_path) + _corrupt_duplicate_fts(db_path) + + script = _REPAIR_SCRIPT.format( + root=str(Path(hermes_state.__file__).parent), db=str(db_path) + ) + procs = [ + subprocess.Popen( + [sys.executable, "-c", script], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + ) + for _ in range(2) + ] + reports = [] + for proc in procs: + out, err = proc.communicate(timeout=120) + assert proc.returncode == 0, err + reports.append(json.loads(out.strip().splitlines()[-1])) + + assert all(r["repaired"] for r in reports), reports + # Exactly one process did the work; the other found the DB already healthy. + strategies = sorted(r["strategy"] for r in reports) + assert "already_healthy" in strategies or "repaired_by_other_process" in strategies + assert len(list(tmp_path.glob("state.db.malformed-backup-*"))) == 1 + + conn = sqlite3.connect(str(db_path)) + try: + assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok" + assert conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0] == 10 + finally: + conn.close() + + +def test_schema_surgery_bumps_the_schema_cookie(tmp_path): + """Live connections in other processes must be told to reload the schema. + + Editing sqlite_master under writable_schema=ON does not bump the cookie + that every other connection checks before running a prepared statement, + so they keep compiling against objects the surgery just deleted. + """ + db_path = tmp_path / "state.db" + _build_healthy_db(db_path) + _corrupt_duplicate_fts(db_path) + + probe = sqlite3.connect(str(db_path)) + try: + probe.execute("PRAGMA writable_schema=ON") + before = probe.execute("PRAGMA schema_version").fetchone()[0] + finally: + probe.close() + + report = repair_state_db_schema(db_path) + assert report["repaired"] is True + + probe = sqlite3.connect(str(db_path)) + try: + after = probe.execute("PRAGMA schema_version").fetchone()[0] + finally: + probe.close() + assert after != before