fix(state): serialize state.db schema surgery across processes
`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 `<db>.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 <noreply@anthropic.com>
This commit is contained in:
Ernst Bablick
Teknium
parent
12a7a46578
commit
923d86e099
159
hermes_state.py
159
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")
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue