diff --git a/hermes_state_schema.py b/hermes_state_schema.py index 3b331711ee..b8a078c455 100644 --- a/hermes_state_schema.py +++ b/hermes_state_schema.py @@ -288,7 +288,10 @@ class SessionSchemaMixin: """Replace FTS triggers without rebuilding historical indexes. Existing rows keep their full-content token stream; the durable high-water id makes new tool rows use the bounded prefix in INSERT and the matching external-content delete/update. One savepoint, so no - concurrent writer lands in a trigger gap.""" + concurrent writer lands in a trigger gap. A fresh store has no historical index to migrate; + its FTS family is created later under rebuild admission.""" + if not self._sqlite_table_exists(cursor, "messages_fts"): + return marker = cursor.execute( "SELECT 1 FROM state_meta WHERE key = ? LIMIT 1", (FTS_TOOL_FULL_CONTENT_HIGH_WATER_KEY,), ).fetchone() @@ -1033,38 +1036,58 @@ class SessionSchemaMixin: self._fts_enabled = self._trigram_available = self._fts_cjk_available = False else: base_sql, trigram_sql = _FTS_DDL[legacy_fts] - # Measure BEFORE the DDL below runs (pre-repair state). Whether the trigram half is - # creatable is only known AFTER _ensure_fts_schema, hence the halves combine at the `if`. + # Measure before any DDL. Publishing missing base triggers before rebuild admission lets + # another process write through an index whose bootstrap/repair has no owner (#105790). base_triggers_missing = self._fts_triggers_missing(cursor, _FTS_BASE_TRIGGERS) or getattr( self, "_fts_tool_prefix_migration_requires_rebuild", False) trigram_triggers_missing = self._fts_triggers_missing(cursor, _FTS_TRIGRAM_TRIGGERS) - self._fts_enabled = self._ensure_fts_schema(cursor, "messages_fts", base_sql) - if self._fts_enabled: - # Trigram is optional; without it CJK search falls back to LIKE. - trigram_enabled = self._ensure_fts_schema(cursor, "messages_fts_trigram", trigram_sql) - self._trigram_available = trigram_enabled - if base_triggers_missing or (trigram_enabled and trigram_triggers_missing): - self._run_admitted_startup_rebuild( - cursor, - lambda: self._rebuild_fts_indexes(cursor, legacy=legacy_fts, include_trigram=trigram_enabled), - ) + + def ensure_and_rebuild() -> None: + self._fts_enabled = self._ensure_fts_schema(cursor, "messages_fts", base_sql) + if not self._fts_enabled: + return + self._trigram_available = self._ensure_fts_schema(cursor, "messages_fts_trigram", trigram_sql) + self._rebuild_fts_indexes( + cursor, legacy=legacy_fts, include_trigram=self._trigram_available, + ) if not legacy_fts: - # CJK-bigram index: strictly additive, gated on the loadable tokenizer. self._ensure_fts_cjk_schema(cursor) + + if base_triggers_missing: + # The authority covers the whole first-publication sequence, not merely the final rebuild. + # ``executescript`` commits DDL statement-by-statement, so acquiring after ensure exposed a + # partially initialized FTS family while another opener held the rebuild lock. + self._run_admitted_startup_rebuild(cursor, ensure_and_rebuild) + else: + self._fts_enabled = self._ensure_fts_schema(cursor, "messages_fts", base_sql) + if self._fts_enabled: + # Trigram is optional; without it CJK search falls back to LIKE. + trigram_enabled = self._ensure_fts_schema(cursor, "messages_fts_trigram", trigram_sql) + self._trigram_available = trigram_enabled + if trigram_enabled and trigram_triggers_missing: + self._run_admitted_startup_rebuild( + cursor, + lambda: self._rebuild_fts_indexes( + cursor, legacy=legacy_fts, include_trigram=trigram_enabled, + ), + ) + if self._fts_enabled and not legacy_fts and not base_triggers_missing: + # CJK-bigram index: strictly additive, gated on the loadable tokenizer. + self._ensure_fts_cjk_schema(cursor) # IF NOT EXISTS cannot rewrite pre-existing broad AFTER UPDATE triggers. if self._fts_enabled: self._migrate_broad_fts_update_triggers(cursor) def _run_admitted_startup_rebuild(self, cursor, rebuild_fn) -> None: - """Run a full trigger-repair FTS rebuild under cross-process admission (the sync triggers - were missing and the DDL just recreated them: the index has a gap of unknown - extent). Two processes opening the same DB after an update commonly hit this - simultaneously (the interleaving that corrupted state.db in production), so this - FAILS CLOSED: on deferral the just-repaired triggers are dropped again and the - stale breadcrumb persisted — triggers must never be live over an unrebuilt gap - (``_enter_fts_fail_open``'s ordering contract); a later recovery path restores both. + """Run FTS bootstrap or trigger-repair rebuild under cross-process admission. + + The fresh/base-missing path includes DDL in ``rebuild_fn`` so no process can publish + triggers before it owns the rebuild. Other repair paths may already have recreated an + optional trigger; deferral therefore still drops every trigger and persists the stale + breadcrumb. A later recovery path restores the complete family. See #93200. + See #105790. """ with fts_rebuild_admission(self.db_path) as admitted: if admitted: diff --git a/tests/state/test_fts_fresh_bootstrap_admission.py b/tests/state/test_fts_fresh_bootstrap_admission.py new file mode 100644 index 0000000000..e3e91a2ffc --- /dev/null +++ b/tests/state/test_fts_fresh_bootstrap_admission.py @@ -0,0 +1,60 @@ +"""Fresh state.db FTS bootstrap must honor cross-process rebuild admission.""" + +import subprocess +import sys +from pathlib import Path + +import hermes_state_common +from hermes_state import SessionDB + + +_HOLD_ADMISSION_SCRIPT = """ +import sys +import time +from pathlib import Path + +from hermes_state_common import fts_rebuild_admission + +with fts_rebuild_admission(Path(sys.argv[1]), timeout_seconds=0) as acquired: + assert acquired + print("locked", flush=True) + time.sleep(30) +""" + + +def test_fresh_fts_bootstrap_does_not_publish_schema_without_admission(tmp_path, monkeypatch): + """A fresh opener that loses admission leaves no partially initialized FTS surface.""" + db_path = tmp_path / "state.db" + proc = subprocess.Popen( + [sys.executable, "-c", _HOLD_ADMISSION_SCRIPT, str(db_path)], + cwd=Path(__file__).resolve().parents[2], + stdout=subprocess.PIPE, + text=True, + ) + try: + assert proc.stdout is not None + assert proc.stdout.readline().strip() == "locked" + monkeypatch.setattr(hermes_state_common, "_FTS_REBUILD_LOCK_TIMEOUT_SECONDS", 0.1) + + deferred = SessionDB(db_path=db_path) + try: + assert deferred._fts_stale is True + assert deferred._fts_enabled is False + assert deferred._conn.execute( + "SELECT 1 FROM sqlite_master WHERE name = 'messages_fts'" + ).fetchone() is None + finally: + deferred.close() + finally: + proc.kill() + proc.wait(timeout=10) + + recovered = SessionDB(db_path=db_path) + try: + assert recovered._fts_stale is False + assert recovered._fts_enabled is True + recovered.create_session("s1", source="test") + recovered.append_message("s1", "user", "fresh bootstrap recovered") + assert recovered.search_messages("recovered") + finally: + recovered.close()