From 9d0727d49b185d44af09170cb45e71c1568d577b Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sun, 23 Aug 2026 18:41:59 -0700 Subject: [PATCH] fix(state): single fail-closed cross-process authority for all full FTS rebuilds Follow-up to the salvaged #93200 commit. Factors the portable _cross_process_repair_lock ownership pattern (msvcrt on Windows, flock on POSIX, bounded 120s wait) into a cycle-safe shared primitive, fts_rebuild_admission() in hermes_state_common, and routes EVERY full structural FTS rebuild entry point through it: - SessionSearchMixin.rebuild_fts() (replaces the POSIX-only, fail-open 30s flock from the original commit) - _init_schema's trigger-repair rebuilds (_rebuild_fts_indexes / _rebuild_legacy_fts_indexes) via _run_admitted_startup_rebuild - _recover_stale_fts() Fail closed: a caller that cannot acquire the authority DEFERS the rebuild (FTS detached + durable stale breadcrumb, retried at next startup) instead of proceeding into the exact concurrent-rebuild interleaving that structurally corrupted state.db in production. Chunked deferred backfill (fts_rebuild_step) intentionally stays outside the authority. Adds spawned-process regression tests (real child process holding the real lock file): holder blocks contender, deferral fails closed on both the runtime and schema paths, release/holder-death permits the next owner, and stale recovery completes after contention clears. Sabotage-verified: 4/6 tests fail with the admission forced open. --- hermes_state_common.py | 118 +++++++++++ hermes_state_schema.py | 72 ++++++- hermes_state_search.py | 79 ++------ tests/state/test_fts_rebuild_admission.py | 226 ++++++++++++++++++++++ 4 files changed, 429 insertions(+), 66 deletions(-) create mode 100644 tests/state/test_fts_rebuild_admission.py diff --git a/hermes_state_common.py b/hermes_state_common.py index 44e8b91498..1b0daa290e 100644 --- a/hermes_state_common.py +++ b/hermes_state_common.py @@ -6,6 +6,11 @@ reference them without importing hermes_state (which would be a cycle). hermes_state re-imports every name here for backward compatibility. """ +import contextlib +import logging +import os +import sys +import time from typing import Any from agent.skill_commands import ( @@ -793,3 +798,116 @@ AFTER UPDATE OF content, tool_name, tool_calls ON messages BEGIN ); END; """ + + +# ── Cross-process full-FTS-rebuild admission (single authority) ────────────── +# +# Several independent Hermes processes routinely share one state.db (gateway +# service, the Desktop app's `hermes serve` backend, interactive CLI sessions, +# the TUI slash worker). A full structural FTS rebuild — the FTS5 'rebuild' +# command or the drop/recreate script in `_recover_stale_fts` — must only ever +# run in ONE of them at a time: two concurrent rebuilds collide on write and +# have structurally corrupted state.db in production (PR #93200; the +# 2026-08-15 / 2026-08-23 incidents and issues #89293 / #90950). +# +# This is the single admission authority for every full structural rebuild +# entry point: `SessionSearchMixin.rebuild_fts()`, +# `SessionSchemaMixin._rebuild_fts_indexes()` (via `_init_schema`), and +# `SessionSchemaMixin._recover_stale_fts()`. The chunked deferred backfill +# (`fts_rebuild_step`) is deliberately NOT routed through it — it claims +# progress under `_execute_write`'s SQLite transaction authority and is +# intentionally multi-process. +# +# Semantics mirror `hermes_state._cross_process_repair_lock` (the schema- +# surgery authority): portable (msvcrt on Windows, flock elsewhere), bounded +# wait, and FAIL CLOSED — a caller that cannot acquire the lock must NOT +# rebuild. The kernel drops both lock types when the holder dies, so a crashed +# rebuilder cannot wedge future rebuilds. It lives here (not hermes_state) +# because the search/schema mixins cannot import hermes_state (cycle). +# +# The lock file is `.fts_rebuild.lock`, distinct from `.repair.lock`: +# schema surgery runs on an EXCLUSIVE offline connection and can legitimately +# take minutes in VACUUM, while runtime rebuilds run on live connections. The +# timeout is sized for a full 'rebuild' of both indexes on a large DB. + +logger = logging.getLogger("hermes_state") + +_FTS_REBUILD_LOCK_TIMEOUT_SECONDS = 120.0 +_FTS_REBUILD_LOCK_POLL_SECONDS = 0.1 +_IS_WINDOWS = sys.platform == "win32" + + +@contextlib.contextmanager +def fts_rebuild_admission(db_path): + """Serialize full structural FTS rebuilds on *db_path* across processes. + + Yields True when this process holds the rebuild authority, False when the + bounded acquire timed out. A caller that gets False must NOT perform a + full rebuild — proceeding is exactly the concurrent-rebuild interleaving + this lock exists to prevent (fail closed). The deferred/stale breadcrumb + machinery already guarantees a skipped rebuild is retried later. + + ``db_path`` may be a str or Path; None (in-memory DB / tests without a + file path) yields True — a private in-memory DB has no cross-process + surface. + """ + if db_path is None: + yield True + return + lock_path = f"{db_path}.fts_rebuild.lock" + try: + handle = open(lock_path, "a+b") + except OSError as exc: + # Read-only dir, exhausted fds, exotic filesystem: fall back to the + # pre-lock behaviour rather than refusing a rebuild we could run. + logger.warning( + "Could not open FTS rebuild lock %s (%s) — proceeding with " + "in-process serialisation only.", lock_path, exc, + ) + yield True + return + + acquired = False + try: + deadline = time.monotonic() + _FTS_REBUILD_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(_FTS_REBUILD_LOCK_POLL_SECONDS) + if not acquired: + logger.warning( + "FTS rebuild lock %s held by another process for more than " + "%.0fs — deferring this rebuild to avoid racing the holder " + "(the stale-FTS breadcrumb keeps it retryable).", + lock_path, _FTS_REBUILD_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() diff --git a/hermes_state_schema.py b/hermes_state_schema.py index ac996758df..4a8ce5ca63 100644 --- a/hermes_state_schema.py +++ b/hermes_state_schema.py @@ -28,6 +28,7 @@ from hermes_state_common import ( _FTS_CJK_TRIGGERS, _FTS_TRIGGERS, _ephemeral_child_sql, + fts_rebuild_admission, ) # Moved methods logged under the "hermes_state" logger before the split; @@ -377,6 +378,24 @@ class SessionSchemaMixin: foreign_holders, ) return False + # Full structural rebuild: admit through the single cross-process + # authority (fail closed). Losing the race means another process is + # already performing this exact recovery; the stale breadcrumb stays + # set, so this process simply keeps FTS detached and retries later. + with fts_rebuild_admission(getattr(self, "db_path", None)) as admitted: + if not admitted: + logger.warning( + "Deferred stale state.db FTS rebuild: another process " + "holds the rebuild authority; canonical writes and LIKE " + "search remain available." + ) + return False + return self._recover_stale_fts_locked(cursor, legacy=legacy) + + def _recover_stale_fts_locked( + self, cursor: sqlite3.Cursor, *, legacy: bool + ) -> bool: + """Body of :meth:`_recover_stale_fts`; caller holds rebuild authority.""" try: trigram_status = self._fts_table_probe(cursor, "messages_fts_trigram") except sqlite3.DatabaseError: @@ -1264,8 +1283,11 @@ class SessionSchemaMixin: ) self._trigram_available = trigram_enabled if triggers_need_repair: - self._rebuild_legacy_fts_indexes( - cursor, include_trigram=trigram_enabled + self._run_admitted_startup_rebuild( + cursor, + lambda: self._rebuild_legacy_fts_indexes( + cursor, include_trigram=trigram_enabled + ), ) else: triggers_need_repair = ( @@ -1284,9 +1306,12 @@ class SessionSchemaMixin: ) self._trigram_available = trigram_enabled if triggers_need_repair: - self._rebuild_fts_indexes( + self._run_admitted_startup_rebuild( cursor, - include_trigram=trigram_enabled, + lambda: self._rebuild_fts_indexes( + cursor, + include_trigram=trigram_enabled, + ), ) # CJK-bigram index (cjk_unicode61). Strictly additive to # the surfaces above and gated on the loadable tokenizer: @@ -1299,6 +1324,45 @@ class SessionSchemaMixin: self._conn.commit() + def _run_admitted_startup_rebuild(self, cursor, rebuild_fn) -> None: + """Run a full trigger-repair FTS rebuild under cross-process admission. + + ``_init_schema`` reaches here when the sync triggers were missing and + the DDL just recreated them, so the index has a gap of unknown extent + and must be rebuilt in full. Two processes opening the same DB after + an update commonly hit this path simultaneously — the exact + concurrent-rebuild interleaving that structurally corrupted state.db + in production (PR #93200) — so the rebuild admits through + ``fts_rebuild_admission`` and FAILS CLOSED. + + On deferral (another process holds the rebuild authority) the + just-repaired triggers are dropped again and the durable stale + breadcrumb is persisted, mirroring ``_enter_fts_fail_open``'s + ordering contract: triggers must never be live over an index with an + unrebuilt gap. FTS stays detached for this instance; the winner's + rebuild — or ``_recover_stale_fts`` at the next startup — restores + the index and triggers atomically. + """ + with fts_rebuild_admission(getattr(self, "db_path", None)) as admitted: + if admitted: + rebuild_fn() + return + logger.warning( + "Deferred startup FTS rebuild: another process holds the " + "rebuild authority for this state.db; detaching FTS sync " + "until the stale-index recovery path rebuilds it." + ) + cursor.execute( + "INSERT INTO state_meta (key, value) VALUES (?, '1') " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (FTS_STALE_KEY,), + ) + self._drop_all_fts_triggers(cursor) + self._fts_stale = True + self._fts_enabled = False + self._trigram_available = False + self._fts_cjk_available = False + def _backfill_gateway_metadata_from_sessions_json( self, cursor: sqlite3.Cursor ) -> None: diff --git a/hermes_state_search.py b/hermes_state_search.py index d0c1b4d341..738dbbe6b2 100644 --- a/hermes_state_search.py +++ b/hermes_state_search.py @@ -27,65 +27,13 @@ from hermes_state_common import ( SCHEMA_VERSION, _FTS_CJK_TRIGGERS, escape_like as _escape_like, + fts_rebuild_admission, ) # Moved methods logged under the "hermes_state" logger before the split; # keep that logger identity so log filtering/capture behavior is unchanged. logger = logging.getLogger("hermes_state") - -try: # POSIX-only; used for cross-process FTS rebuild serialization - import fcntl - _HAS_FCNTL = True -except ImportError: # pragma: no cover - non-POSIX fallback - _HAS_FCNTL = False - -_FTS_REBUILD_LOCK_TIMEOUT = 30.0 - - -def _acquire_fts_rebuild_lock(db_path) -> Optional[int]: - """Exclusive flock on .fts_rebuild.lock. Returns fd or None.""" - if not _HAS_FCNTL: - return None - try: - lock_path = f"{db_path}.fts_rebuild.lock" - fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600) - try: - fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) - return fd - except OSError: - pass - deadline = time.monotonic() + _FTS_REBUILD_LOCK_TIMEOUT - while True: - try: - fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) - return fd - except OSError: - if time.monotonic() >= deadline: - logger.warning( - "FTS rebuild lock held by another process beyond " - "%ds timeout; proceeding (SQLite writer lock is the " - "final backstop)", int(_FTS_REBUILD_LOCK_TIMEOUT)) - os.close(fd) - return None - time.sleep(0.1) - except OSError as exc: - logger.warning("FTS rebuild lock unavailable (%s); proceeding", exc) - return None - - -def _release_fts_rebuild_lock(fd: int) -> None: - if not _HAS_FCNTL: - return - try: - fcntl.flock(fd, fcntl.LOCK_UN) - except OSError: - pass - try: - os.close(fd) - except OSError: - pass - # Characters FTS5's query grammar rejects outside a quoted phrase. Anything # missing from this set reaches MATCH raw and raises, which the execute site # swallows into zero results — the failure this strip step exists to prevent. @@ -2453,16 +2401,26 @@ class SessionSearchMixin: merges existing segments), ``rebuild`` discards and recreates the index data entirely. + A full structural rebuild must never run concurrently in two + processes sharing one state.db — that interleaving has structurally + corrupted the database in production (PR #93200) — so this admits + through the cross-process ``fts_rebuild_admission`` authority and + FAILS CLOSED: if another process holds the rebuild lock beyond the + bounded wait, this call defers (returns 0) rather than racing it. + Callers already treat 0 as "rebuild made no progress" and fall back + to the stale-FTS breadcrumb path, which retries at next startup. + Safe to call when FTS tables don't exist (skips them). Returns the number of FTS indexes that were rebuilt. """ rebuilt = 0 - lock_fd = None - if _HAS_FCNTL: - db_path = getattr(self, "db_path", None) - if db_path is not None: - lock_fd = _acquire_fts_rebuild_lock(db_path) - try: + with fts_rebuild_admission(getattr(self, "db_path", None)) as admitted: + if not admitted: + logger.warning( + "Deferred in-place FTS rebuild: another process holds " + "the rebuild authority for this state.db." + ) + return 0 with self._lock: for tbl in self._FTS_TABLES: if not self._fts_table_exists(tbl): @@ -2478,9 +2436,6 @@ class SessionSearchMixin: logger.warning( "FTS rebuild failed for %s: %s", tbl, exc ) - finally: - if lock_fd is not None: - _release_fts_rebuild_lock(lock_fd) return rebuilt def _merge_fts_incrementally( diff --git a/tests/state/test_fts_rebuild_admission.py b/tests/state/test_fts_rebuild_admission.py new file mode 100644 index 0000000000..b6baf087a0 --- /dev/null +++ b/tests/state/test_fts_rebuild_admission.py @@ -0,0 +1,226 @@ +"""Cross-process admission for full structural FTS rebuilds (PR #93200 class). + +Several independent Hermes processes routinely share one state.db (gateway, +Desktop's ``hermes serve`` backend, CLI sessions, the TUI slash worker). Two +of them detecting FTS corruption at once each ran the full FTS5 'rebuild' on +the same file in parallel, colliding on write and structurally corrupting +state.db (two documented production incidents, 2026-08-15 and 2026-08-23). + +The fix: every full structural rebuild entry point — ``rebuild_fts()``, the +``_init_schema`` trigger-repair rebuilds, and ``_recover_stale_fts`` — admits +through one cross-process file lock (``fts_rebuild_admission`` in +hermes_state_common) and FAILS CLOSED: a process that cannot acquire the +authority defers the rebuild instead of racing the holder. These tests use +real spawned processes holding the real lock file, per the review contract +on PR #93200 — the bug is cross-process ownership, so monkeypatched helpers +prove nothing. +""" + +import contextlib +import subprocess +import sqlite3 +import sys +from pathlib import Path + +import pytest + +import hermes_state_common +from hermes_state import FTS_STALE_KEY, SessionDB, _FTS_TRIGGERS + +pytestmark = pytest.mark.skipif( + sys.platform == "win32", reason="POSIX flock child-process harness" +) + + +_HOLD_LOCK_SCRIPT = """ +import sys, time, fcntl, pathlib +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}) +""" + + +def _lock_file(db_path: Path) -> Path: + return db_path.with_name(db_path.name + ".fts_rebuild.lock") + + +@contextlib.contextmanager +def _rebuild_lock_held_by_other_process(db_path: Path, hold_seconds: float = 30.0): + """Hold the FTS rebuild authority for *db_path* in a real child process.""" + script = _HOLD_LOCK_SCRIPT.format( + lock=str(_lock_file(db_path)), hold=hold_seconds + ) + proc = subprocess.Popen( + [sys.executable, "-c", script], stdout=subprocess.PIPE, text=True + ) + try: + assert proc.stdout.readline().strip() == "locked" + yield proc + finally: + proc.kill() + proc.wait(timeout=10) + + +def _fts_docsize_count(db_path: Path) -> int: + raw = sqlite3.connect(str(db_path)) + try: + return raw.execute("SELECT count(*) FROM messages_fts_docsize").fetchone()[0] + finally: + raw.close() + + +def _base_fts_triggers(db_path: Path) -> set: + raw = sqlite3.connect(str(db_path)) + try: + rows = raw.execute( + "SELECT name FROM sqlite_master WHERE type = 'trigger' " + f"AND name IN ({','.join('?' for _ in _FTS_TRIGGERS)})", + _FTS_TRIGGERS, + ).fetchall() + return {r[0] for r in rows} + finally: + raw.close() + + +def _meta_value(db_path: Path, key: str): + raw = sqlite3.connect(str(db_path)) + try: + row = raw.execute( + "SELECT value FROM state_meta WHERE key = ?", (key,) + ).fetchone() + return None if row is None else row[0] + finally: + raw.close() + + +@pytest.fixture +def fast_timeout(monkeypatch): + monkeypatch.setattr( + hermes_state_common, "_FTS_REBUILD_LOCK_TIMEOUT_SECONDS", 0.5 + ) + + +@pytest.fixture +def db(tmp_path): + d = SessionDB(db_path=tmp_path / "state.db") + if not d._fts_enabled: + d.close() + pytest.skip("FTS5 unavailable in this build") + d.create_session("s1", source="test") + for i in range(5): + d.append_message("s1", "user", f"hello world {i}") + yield d + try: + d.close() + except Exception: + pass + + +class TestRebuildFtsAdmission: + def test_rebuild_defers_while_another_process_holds_authority( + self, db, fast_timeout + ): + """Fail closed: the contender must NOT rebuild while the lock is held.""" + with _rebuild_lock_held_by_other_process(db.db_path): + assert db.rebuild_fts() == 0 + + def test_rebuild_proceeds_after_holder_releases(self, db, fast_timeout): + with _rebuild_lock_held_by_other_process(db.db_path): + assert db.rebuild_fts() == 0 + # Holder killed on context exit → kernel drops the flock → the next + # caller acquires the authority and the rebuild really runs. + assert db.rebuild_fts() >= 1 + + def test_rebuild_waits_out_a_short_holder(self, db, monkeypatch): + """A holder that releases within the bounded wait does not cause deferral.""" + monkeypatch.setattr( + hermes_state_common, "_FTS_REBUILD_LOCK_TIMEOUT_SECONDS", 10.0 + ) + with _rebuild_lock_held_by_other_process(db.db_path, hold_seconds=1.0): + # Child exits after 1s; deadline is 10s — this must acquire and rebuild. + assert db.rebuild_fts() >= 1 + + def test_admission_yields_true_for_pathless_db(self): + """In-memory / pathless stores have no cross-process surface.""" + with hermes_state_common.fts_rebuild_admission(None) as admitted: + assert admitted is True + + +class TestSchemaPathAdmission: + def test_startup_trigger_repair_defers_and_fails_closed( + self, tmp_path, fast_timeout + ): + """The _init_schema trigger-repair rebuild is covered by the SAME + authority — deferral must leave FTS detached with the durable stale + breadcrumb, never triggers installed over an unrebuilt index gap.""" + db_path = tmp_path / "state.db" + d = SessionDB(db_path=db_path) + if not d._fts_enabled: + d.close() + pytest.skip("FTS5 unavailable in this build") + d.create_session("s1", source="test") + d.append_message("s1", "user", "hello schema path") + d.close() + + # Drop one sync trigger out-of-band: next open takes the + # triggers_need_repair branch in _init_schema. + raw = sqlite3.connect(str(db_path)) + raw.execute(f"DROP TRIGGER IF EXISTS {sorted(_FTS_TRIGGERS)[0]}") + raw.commit() + raw.close() + + with _rebuild_lock_held_by_other_process(db_path): + d2 = SessionDB(db_path=db_path) + try: + assert d2._fts_enabled is False + finally: + d2.close() + + # Durable state: stale breadcrumb set, no live sync triggers. + assert _meta_value(db_path, FTS_STALE_KEY) == "1" + assert _base_fts_triggers(db_path) == set() + + def test_stale_recovery_defers_then_succeeds_after_release( + self, tmp_path, fast_timeout + ): + """_recover_stale_fts defers under contention and completes once the + authority is free (next open).""" + db_path = tmp_path / "state.db" + d = SessionDB(db_path=db_path) + if not d._fts_enabled: + d.close() + pytest.skip("FTS5 unavailable in this build") + d.create_session("s1", source="test") + d.append_message("s1", "user", "hello recovery path") + d.close() + + raw = sqlite3.connect(str(db_path)) + raw.execute( + "INSERT INTO state_meta (key, value) VALUES (?, '1') " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (FTS_STALE_KEY,), + ) + for trig in _FTS_TRIGGERS: + raw.execute(f"DROP TRIGGER IF EXISTS {trig}") + raw.commit() + raw.close() + + with _rebuild_lock_held_by_other_process(db_path): + d2 = SessionDB(db_path=db_path) + try: + assert d2._fts_enabled is False + finally: + d2.close() + # Deferred: breadcrumb still present, recovery not performed. + assert _meta_value(db_path, FTS_STALE_KEY) == "1" + + d3 = SessionDB(db_path=db_path) + try: + assert d3._fts_enabled is True + finally: + d3.close() + # Recovered: breadcrumb cleared, triggers restored. + assert _meta_value(db_path, FTS_STALE_KEY) is None + assert _base_fts_triggers(db_path) == set(_FTS_TRIGGERS)