diff --git a/hermes_state.py b/hermes_state.py index c19283f94b..8ecc3b5c26 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -979,14 +979,7 @@ class SessionDB( # mutations, not just idempotent UPSERTs. ioerr_begin_retried = False while True: - self._raise_if_db_corrupt() - if storage_state(self.db_path) == STORAGE_CORRUPT: - # Another handle in this process already saw structural damage on this file. - # Quarantine this one before it touches SQLite; the error type is the same - # StateDbCorruptError, so every transcript-diversion owner handles it unchanged. - self._halt_db_corrupt(sqlite3.DatabaseError( - "database disk image is malformed (reported earlier in this process: " - f"{storage_corrupt_reason(self.db_path)})")) + self._raise_if_db_corrupt(storage=True) # NOTE: the replaced/generation live probe runs INSIDE the lock below, # not here. close() mutates _conn and _db_sidecar_identity under that # same lock, ending the WAL generation (SQLite unlinks the -wal/-shm @@ -1397,9 +1390,16 @@ class SessionDB( ) return retire_without_close - def _raise_if_db_corrupt(self) -> None: + def _raise_if_db_corrupt(self, *, storage: bool = False) -> None: if self._db_corrupt: raise self._corrupt_error() + if storage and storage_state(self.db_path) == STORAGE_CORRUPT: + # Another handle in this process already saw structural damage on this file. + # Quarantine this one before it touches SQLite; the error type is the same + # StateDbCorruptError, so every transcript-diversion owner handles it unchanged. + self._halt_db_corrupt(sqlite3.DatabaseError( + "database disk image is malformed (reported earlier in this process: " + f"{storage_corrupt_reason(self.db_path)})")) def _sleep_before_write_retry(self, deadline: float, patience_s: float) -> bool: """Sleep one jitter interval if the budget allows; True = retry, False = deadline passed. Small diff --git a/hermes_state_fts.py b/hermes_state_fts.py index 40a3045722..3ced10cb2c 100644 --- a/hermes_state_fts.py +++ b/hermes_state_fts.py @@ -12,7 +12,7 @@ from typing import Sequence from hermes_constants import get_hermes_home from hermes_state_common import (FTS_CJK_STALE_KEY, FTS_STALE_KEY, _FTS_CJK_TRIGGERS, _FTS_TRIGGERS, routed_sessions_setting) -from hermes_state_errors import is_fts_scoped_corruption_error +from hermes_state_errors import is_fts_scoped_corruption_error, is_sqlite_lock_error # caplog tests pin the "hermes_state" logger name. logger = logging.getLogger("hermes_state") @@ -364,12 +364,14 @@ class SessionFtsSetupMixin: same corrupt index — giving up after 1 s cost that turn's canonical write.""" if not self._fts_enabled or not self._is_fts_write_corruption_error(exc): return False - self._raise_if_db_corrupt() if patience_s is None: patience_s = self._WRITE_PATIENCE_S if deadline is None: deadline = time.monotonic() + patience_s while True: + # Re-checked every attempt: a sibling may quarantine the file while we wait for the + # lock, and nothing may be committed on a quarantined handle. + self._raise_if_db_corrupt(storage=True) try: with self._lock: self._raise_if_db_replaced() @@ -401,9 +403,8 @@ class SessionFtsSetupMixin: raise break except sqlite3.Error as detach_exc: - msg = str(detach_exc).lower() if ( - isinstance(detach_exc, sqlite3.OperationalError) and ("locked" in msg or "busy" in msg) + isinstance(detach_exc, sqlite3.OperationalError) and is_sqlite_lock_error(detach_exc) and self._sleep_before_write_retry(deadline, patience_s) ): continue diff --git a/tests/hermes_state/test_fts_index_fail_open.py b/tests/hermes_state/test_fts_index_fail_open.py index 59bb95ae5f..496ca0f7a2 100644 --- a/tests/hermes_state/test_fts_index_fail_open.py +++ b/tests/hermes_state/test_fts_index_fail_open.py @@ -12,6 +12,7 @@ user-facing guidance: * an FTS-scoped error that still escapes (detach refused) classifies as ``fts_index`` and never quarantines the handle; * a sibling process holding the write lock when the detach runs is waited out, not a lost write; +* a quarantine that lands while the detach waits stops it: nothing is committed on the file; """ import sqlite3 @@ -20,7 +21,8 @@ from types import SimpleNamespace import pytest -from hermes_state import SessionDB +from hermes_state import SessionDB, StateDbCorruptError +from hermes_state_health import mark_storage_corrupt, reset_storage_state from run_agent import AIAgent @@ -178,3 +180,62 @@ def test_detach_waits_out_a_sibling_holding_the_write_lock(tmp_path): assert db._db_corrupt is False finally: db.close() + + +def test_quarantine_while_detach_waits_commits_nothing(tmp_path): + """The detach may now wait up to the write budget for the lock. A sibling that quarantines + this file meanwhile (structural corruption latched process-wide) must stop it: the retry + drops no triggers, commits no stale breadcrumb, and the corrupt error surfaces.""" + db_path = tmp_path / "state.db" + db = SessionDB(db_path=db_path) + try: + _seed(db, rows=5) + _stomp_fts_shadow(db_path) + held, release = threading.Event(), threading.Event() + + def sibling(): + raw = sqlite3.connect(str(db_path), timeout=30, isolation_level=None) + raw.execute("BEGIN IMMEDIATE") + held.set() + release.wait(10) + raw.execute("COMMIT") + raw.close() + + real_check, real_sleep = db._is_fts_write_corruption_error, db._sleep_before_write_retry + holder = [] + + def check_then_contend(exc): + hit = real_check(exc) + if hit and not holder: + holder.append(threading.Thread(target=sibling)) + holder[0].start() + assert held.wait(10) + return hit + + def quarantine_then_sleep(deadline, patience_s): + mark_storage_corrupt(db_path, "database disk image is malformed (sibling handle)") + release.set() + return real_sleep(deadline, patience_s) + + db._is_fts_write_corruption_error = check_then_contend + db._sleep_before_write_retry = quarantine_then_sleep + with pytest.raises(StateDbCorruptError): + db.append_message("s1", "user", "must not land on a quarantined file") + if not holder: + pytest.skip("this SQLite build defers FTS shadow corruption past the insert trigger") + holder[0].join(10) + + raw = sqlite3.connect(str(db_path)) + try: + triggers = raw.execute( + "SELECT count(*) FROM sqlite_master WHERE type = 'trigger' AND name LIKE 'messages_fts%'" + ).fetchone()[0] + stale = raw.execute("SELECT value FROM state_meta WHERE key LIKE 'fts%stale%'").fetchall() + finally: + raw.close() + assert triggers > 0 + assert stale == [] + assert db._fts_stale is False + finally: + db.close() + reset_storage_state(db_path)