From fc72d6c71691a48743dc641aaa6679916c3c8eb0 Mon Sep 17 00:00:00 2001 From: fangliquanflq Date: Thu, 20 Aug 2026 21:04:39 +0800 Subject: [PATCH] fix(state): defer FTS rebuild under foreign WAL holders --- hermes_state.py | 62 +++++++++++++++ hermes_state_schema.py | 9 +++ tests/state/test_fts_runtime_rebuild.py | 100 ++++++++++++++++++++++++ 3 files changed, 171 insertions(+) diff --git a/hermes_state.py b/hermes_state.py index 4f4916761e..a7d706e296 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -4211,6 +4211,59 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) msg = str(exc).lower() return "fts5" in msg and "corrupt" in msg + def _foreign_state_db_holders(self) -> List[Tuple[int, str]]: + """Return foreign processes holding this DB or its WAL sidecars. + + Automatic FTS repair is structural maintenance, not an ordinary WAL + write. It must not run while another process remains attached: a + sidecar reset under that holder can leave the two processes writing + through different WAL inodes. ``psutil`` reads the kernel's open-file + table, including Linux ``(deleted)`` descriptors, so this also catches + a split brain already in progress. + + A scan failure is represented as an unknown holder. Skipping optional + automatic maintenance is safer than assuming quiescence; canonical + writes continue through the stale-FTS fail-open path. + """ + # The split-brain mechanism requires POSIX unlink semantics: Windows + # refuses to replace SQLite sidecars while another process has them + # open. Avoid psutil.open_files() there; querying arbitrary Windows + # processes can block for minutes on device-backed handles. + if _IS_WINDOWS: + return [] + if psutil is None: + return [(-1, "open-file scan unavailable")] + + def _canonical(path: str) -> str: + clean = path.removesuffix(" (deleted)") + return os.path.normcase(os.path.abspath(clean)) + + db_path = os.path.abspath(os.fspath(self.db_path)) + watched = { + _canonical(db_path), + _canonical(db_path + "-wal"), + _canonical(db_path + "-shm"), + } + holders: List[Tuple[int, str]] = [] + try: + for process in psutil.process_iter(["pid", "open_files"]): + info = process.info + pid = int(info["pid"]) + if pid == os.getpid(): + continue + for opened in info.get("open_files") or (): + path = getattr(opened, "path", "") + if path and _canonical(path) in watched: + holders.append((pid, path)) + except Exception as exc: + logger.warning( + "Could not prove state.db has no foreign holders; deferring " + "automatic FTS maintenance: %s", + exc, + ) + return holders or [(-1, f"open-file scan failed: {exc}")] + return holders + def _try_runtime_fts_rebuild(self, exc: sqlite3.DatabaseError) -> bool: """One-shot in-place FTS rebuild after a corrupt-index write failure. @@ -4234,6 +4287,15 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) if not self._is_fts_write_corruption_error(exc): return False self._fts_runtime_rebuild_attempted = True + foreign_holders = self._foreign_state_db_holders() + if foreign_holders: + logger.warning( + "Skipping automatic state.db FTS rebuild while foreign " + "processes hold the database or WAL sidecars (%s); detaching " + "FTS sync so canonical writes can continue.", + foreign_holders, + ) + return False logger.warning( "state.db write failed with an FTS-corruption error (%s) — " "attempting one-shot in-place FTS rebuild; canonical message " diff --git a/hermes_state_schema.py b/hermes_state_schema.py index 3fa101a772..ac996758df 100644 --- a/hermes_state_schema.py +++ b/hermes_state_schema.py @@ -368,6 +368,15 @@ class SessionSchemaMixin: def _recover_stale_fts(self, cursor: sqlite3.Cursor, *, legacy: bool) -> bool: """Atomically rebuild stale base/trigram indexes and resume syncing.""" + foreign_holders = self._foreign_state_db_holders() + if foreign_holders: + logger.warning( + "Deferred stale state.db FTS rebuild while foreign processes " + "hold the database or WAL sidecars (%s); canonical writes and " + "LIKE search remain available.", + foreign_holders, + ) + return False try: trigram_status = self._fts_table_probe(cursor, "messages_fts_trigram") except sqlite3.DatabaseError: diff --git a/tests/state/test_fts_runtime_rebuild.py b/tests/state/test_fts_runtime_rebuild.py index e9c14ecd54..d8bc8fb887 100644 --- a/tests/state/test_fts_runtime_rebuild.py +++ b/tests/state/test_fts_runtime_rebuild.py @@ -14,9 +14,11 @@ until a later open atomically rebuilds the index and restores the triggers. """ import sqlite3 +from types import SimpleNamespace import pytest +import hermes_state from hermes_state import ( FTS_STALE_KEY, LEGACY_FTS_SQL, @@ -84,6 +86,47 @@ def _base_fts_triggers(db_path): class TestRuntimeFtsRebuild: + def test_foreign_holder_detection_includes_deleted_wal( + self, db, tmp_path, monkeypatch + ): + db_path = tmp_path / "state.db" + + class FakePsutil: + @staticmethod + def process_iter(_attrs): + return iter( + ( + SimpleNamespace( + info={ + "pid": 111, + "open_files": [SimpleNamespace(path=str(db_path))], + } + ), + SimpleNamespace( + info={ + "pid": 222, + "open_files": [ + SimpleNamespace(path=f"{db_path}-wal (deleted)") + ], + } + ), + SimpleNamespace( + info={ + "pid": 333, + "open_files": [SimpleNamespace(path=str(tmp_path / "other.db"))], + } + ), + ) + ) + + monkeypatch.setattr(hermes_state, "psutil", FakePsutil) + monkeypatch.setattr(hermes_state, "_IS_WINDOWS", False) + monkeypatch.setattr(hermes_state.os, "getpid", lambda: 111) + + assert db._foreign_state_db_holders() == [ + (222, f"{db_path}-wal (deleted)") + ] + def test_corruption_error_classification_covers_both_sqlite_messages(self): """SQLite's message for a corrupt FTS index varies by version: older builds raise the generic malformed-image error, newer builds raise an @@ -242,6 +285,30 @@ class TestRuntimeFtsRebuild: assert _meta_value(db_path, FTS_STALE_KEY) == "1" assert _base_fts_triggers(db_path) == set() + def test_foreign_holder_skips_runtime_rebuild_and_fails_open( + self, db, tmp_path, monkeypatch + ): + if not db._fts_enabled: + pytest.skip("FTS5 unavailable in this build") + db_path = tmp_path / "state.db" + db.create_session("s1", source="test") + db.append_message("s1", "user", "seed") + _corrupt_fts(db_path) + + monkeypatch.setattr( + db, + "_foreign_state_db_holders", + lambda: [(4242, str(db_path) + "-wal")], + raising=False, + ) + + db.append_message("s1", "user", "canonical survives foreign holder") + + assert _message_contents(db_path)[-1] == "canonical survives foreign holder" + assert db._fts_stale is True + assert _meta_value(db_path, FTS_STALE_KEY) == "1" + assert _base_fts_triggers(db_path) == set() + def test_stale_search_preserves_not_semantics(self, db, tmp_path, monkeypatch): if not db._fts_enabled: pytest.skip("FTS5 unavailable in this build") @@ -324,6 +391,39 @@ class TestRuntimeFtsRebuild: finally: reopened.close() + def test_foreign_holder_defers_startup_stale_rebuild( + self, db, tmp_path, monkeypatch + ): + if not db._fts_enabled: + pytest.skip("FTS5 unavailable in this build") + db_path = tmp_path / "state.db" + db.create_session("s1", source="test") + db.append_message("s1", "user", "seed") + _corrupt_fts(db_path) + monkeypatch.setattr( + db, + "rebuild_fts", + lambda: (_ for _ in ()).throw(sqlite3.DatabaseError("still corrupt")), + ) + db.append_message("s1", "user", "before restart") + db.close() + + monkeypatch.setattr( + SessionDB, + "_foreign_state_db_holders", + lambda self: [(4242, str(db_path) + "-wal")], + raising=False, + ) + reopened = SessionDB(db_path=db_path) + try: + assert reopened._fts_stale is True + assert _meta_value(db_path, FTS_STALE_KEY) == "1" + assert _base_fts_triggers(db_path) == set() + reopened.append_message("s1", "user", "after deferred recovery") + assert _message_contents(db_path)[-1] == "after deferred recovery" + finally: + reopened.close() + def test_legacy_inline_fts_fails_open_and_recovers(self, tmp_path, monkeypatch): db_path = tmp_path / "legacy-state.db" raw = sqlite3.connect(str(db_path))