fix(sessions): serialize fresh FTS bootstrap
This commit is contained in:
@@ -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:
|
||||
|
||||
60
tests/state/test_fts_fresh_bootstrap_admission.py
Normal file
60
tests/state/test_fts_fresh_bootstrap_admission.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user