fix(state): self-heal FTS corruption on the SessionDB write path (#66296)
Complements the #65637 salvage (53d358838+a9cc17fd8): the gateway session store now retries transcript appends through its own queue, but cron and CLI writers call SessionDB directly — a corrupt FTS index still hard-failed their appends until the next process restart triggered the offline repair. _execute_write now detects the FTS-corruption error class (both the generic 'database disk image is malformed' and newer SQLite's 'fts5: corrupt structure record' variant), performs a one-shot in-place rebuild by delegating to the existing rebuild_fts(), and retries the failed write. One-shot per instance so an unrecoverable database cannot loop; lock/busy jitter-retry path untouched. E2E-verified: corrupted messages_fts_data rejects appends; with this fix the same append self-heals, persists, and FTS search works again.
This commit is contained in:
Teknium
GitHub
parent
0bf44d557f
commit
9e1b1d7536
2 changed files with 218 additions and 0 deletions
|
|
@ -1016,6 +1016,12 @@ class SessionDB:
|
|||
|
||||
self._lock = threading.Lock()
|
||||
self._write_count = 0
|
||||
# One-shot guard for the runtime FTS rebuild recovery on the write
|
||||
# path. A corrupt FTS shadow table makes EVERY message write raise
|
||||
# the malformed/corrupt error class via the sync triggers; we repair
|
||||
# in place at most once per SessionDB instance so a genuinely
|
||||
# unrecoverable database can't put writers into a rebuild loop.
|
||||
self._fts_runtime_rebuild_attempted = False
|
||||
self._fts_enabled = False
|
||||
self._trigram_available = False
|
||||
self._fts_unavailable_warned = False
|
||||
|
|
@ -1297,11 +1303,89 @@ class SessionDB:
|
|||
continue
|
||||
# Non-lock error or retries exhausted — propagate.
|
||||
raise
|
||||
except sqlite3.DatabaseError as exc:
|
||||
# Corrupt FTS shadow tables make every write raise the
|
||||
# malformed/corrupt error class through the FTS sync triggers
|
||||
# while the canonical messages table is intact. The gateway
|
||||
# session store has its own retry queue for transcript
|
||||
# appends (#65637 salvage), but cron and CLI writers call
|
||||
# SessionDB directly — without this, their writes hard-fail
|
||||
# until the next process restart triggers the offline repair.
|
||||
# Rebuild the FTS index in place (once per instance) via
|
||||
# rebuild_fts() and retry the failed write immediately.
|
||||
if not self._try_runtime_fts_rebuild(exc):
|
||||
raise
|
||||
continue
|
||||
# Retries exhausted (shouldn't normally reach here).
|
||||
raise last_err or sqlite3.OperationalError(
|
||||
"database is locked after max retries"
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _is_fts_write_corruption_error(exc: sqlite3.DatabaseError) -> bool:
|
||||
"""True for the error class a corrupt FTS index raises on writes.
|
||||
|
||||
The message varies by SQLite version: older builds raise the generic
|
||||
``database disk image is malformed`` (covered by
|
||||
``is_malformed_db_error``); newer builds (e.g. ubuntu-latest CI)
|
||||
raise the FTS5-specific ``fts5: corrupt structure record for table
|
||||
"messages_fts"``. Both mean the same thing for the write path: the
|
||||
canonical rows are fine, the FTS shadow tables are not.
|
||||
"""
|
||||
if is_malformed_db_error(exc):
|
||||
return True
|
||||
msg = str(exc).lower()
|
||||
return "fts5" in msg and "corrupt" in msg
|
||||
|
||||
def _try_runtime_fts_rebuild(self, exc: sqlite3.DatabaseError) -> bool:
|
||||
"""One-shot in-place FTS rebuild after a corrupt-index write failure.
|
||||
|
||||
Returns True when a rebuild was performed and the failed write should
|
||||
be retried; False when the error isn't the FTS-corruption class, FTS
|
||||
is disabled, or a rebuild was already attempted for this instance.
|
||||
|
||||
Delegates to :meth:`rebuild_fts` (the FTS5 ``'rebuild'`` command —
|
||||
index rewritten from the canonical messages table, zero message-row
|
||||
mutation). Safe to call from ``_execute_write``'s except path: the
|
||||
failed transaction was rolled back and ``self._lock`` released before
|
||||
the exception propagated, and ``rebuild_fts`` re-acquires it.
|
||||
E2E-verified: a corrupted ``messages_fts_data`` shadow table rejects
|
||||
every append; after the in-place rebuild the same append succeeds and
|
||||
search works again.
|
||||
"""
|
||||
if self._fts_runtime_rebuild_attempted:
|
||||
return False
|
||||
if not self._fts_enabled:
|
||||
return False
|
||||
if not self._is_fts_write_corruption_error(exc):
|
||||
return False
|
||||
self._fts_runtime_rebuild_attempted = True
|
||||
logger.warning(
|
||||
"state.db write failed with an FTS-corruption error (%s) — "
|
||||
"attempting one-shot in-place FTS rebuild; canonical message "
|
||||
"rows are preserved.", exc,
|
||||
)
|
||||
try:
|
||||
rebuilt = self.rebuild_fts()
|
||||
except Exception as rebuild_exc:
|
||||
logger.error(
|
||||
"In-place FTS rebuild failed (%s); the database needs the "
|
||||
"full offline repair path (repair_state_db_schema).",
|
||||
rebuild_exc,
|
||||
)
|
||||
return False
|
||||
if not rebuilt:
|
||||
logger.error(
|
||||
"In-place FTS rebuild made no progress; the database needs "
|
||||
"the full offline repair path (repair_state_db_schema)."
|
||||
)
|
||||
return False
|
||||
logger.warning(
|
||||
"state.db FTS indexes rebuilt in place (%d); retrying the failed write.",
|
||||
rebuilt,
|
||||
)
|
||||
return True
|
||||
|
||||
def _try_wal_checkpoint(self) -> None:
|
||||
"""Best-effort PASSIVE WAL checkpoint. Never raises.
|
||||
|
||||
|
|
|
|||
134
tests/state/test_fts_runtime_rebuild.py
Normal file
134
tests/state/test_fts_runtime_rebuild.py
Normal file
|
|
@ -0,0 +1,134 @@
|
|||
"""Runtime FTS-corruption self-heal on the SessionDB write path (#65637 class).
|
||||
|
||||
A corrupted FTS5 shadow table (``messages_fts_data``) makes every message
|
||||
write raise ``sqlite3.DatabaseError: database disk image is malformed``
|
||||
through the FTS sync triggers, while the canonical ``messages`` rows stay
|
||||
intact. Before this fix the gateway swallowed the failure at debug level and
|
||||
the in-memory session advanced while disk silently fell behind — surfacing
|
||||
later as "Persisted transcript lagged live cached history" amnesia.
|
||||
|
||||
The fix: ``_execute_write`` detects the malformed-image class, performs a
|
||||
one-shot in-place FTS rebuild (FTS5 ``'rebuild'`` command — index rewritten
|
||||
from canonical rows, no messages touched), and retries the failed write.
|
||||
"""
|
||||
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_state import SessionDB
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db(tmp_path):
|
||||
d = SessionDB(db_path=tmp_path / "state.db")
|
||||
yield d
|
||||
try:
|
||||
d.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _corrupt_fts(db_path):
|
||||
raw = sqlite3.connect(str(db_path))
|
||||
raw.execute(
|
||||
"UPDATE messages_fts_data SET block = X'DEADBEEFDEADBEEFDEADBEEFDEADBEEF'"
|
||||
)
|
||||
raw.commit()
|
||||
raw.close()
|
||||
|
||||
|
||||
def _message_contents(db_path):
|
||||
raw = sqlite3.connect(str(db_path))
|
||||
rows = raw.execute("SELECT content FROM messages ORDER BY id").fetchall()
|
||||
raw.close()
|
||||
return [r[0] for r in rows]
|
||||
|
||||
|
||||
class TestRuntimeFtsRebuild:
|
||||
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
|
||||
FTS5-specific one. Both must trigger the self-heal."""
|
||||
assert SessionDB._is_fts_write_corruption_error(
|
||||
sqlite3.DatabaseError("database disk image is malformed")
|
||||
)
|
||||
assert SessionDB._is_fts_write_corruption_error(
|
||||
sqlite3.DatabaseError(
|
||||
'fts5: corrupt structure record for table "messages_fts"'
|
||||
)
|
||||
)
|
||||
assert not SessionDB._is_fts_write_corruption_error(
|
||||
sqlite3.DatabaseError("no such table: nothing_fts_related")
|
||||
)
|
||||
|
||||
def test_append_self_heals_after_fts_corruption(self, db, tmp_path):
|
||||
if not db._fts_enabled:
|
||||
pytest.skip("FTS5 unavailable in this build")
|
||||
db.create_session("s1", source="test")
|
||||
db.append_message("s1", "user", "hello world")
|
||||
|
||||
_corrupt_fts(tmp_path / "state.db")
|
||||
|
||||
# Before the fix this raised DatabaseError and the row was lost.
|
||||
msg_id = db.append_message("s1", "user", "healed append")
|
||||
assert msg_id is not None
|
||||
assert _message_contents(tmp_path / "state.db") == [
|
||||
"hello world",
|
||||
"healed append",
|
||||
]
|
||||
|
||||
def test_search_works_after_self_heal(self, db, tmp_path):
|
||||
if not db._fts_enabled:
|
||||
pytest.skip("FTS5 unavailable in this build")
|
||||
db.create_session("s1", source="test")
|
||||
db.append_message("s1", "user", "before corruption")
|
||||
_corrupt_fts(tmp_path / "state.db")
|
||||
db.append_message("s1", "user", "searchable needle text")
|
||||
|
||||
raw = sqlite3.connect(str(tmp_path / "state.db"))
|
||||
hits = raw.execute(
|
||||
"SELECT rowid FROM messages_fts WHERE messages_fts MATCH 'needle'"
|
||||
).fetchall()
|
||||
raw.close()
|
||||
assert len(hits) == 1
|
||||
|
||||
def test_rebuild_is_one_shot_per_instance(self, db, tmp_path):
|
||||
if not db._fts_enabled:
|
||||
pytest.skip("FTS5 unavailable in this build")
|
||||
db.create_session("s1", source="test")
|
||||
db.append_message("s1", "user", "seed")
|
||||
_corrupt_fts(tmp_path / "state.db")
|
||||
db.append_message("s1", "user", "first heal") # consumes the one shot
|
||||
assert db._fts_runtime_rebuild_attempted is True
|
||||
|
||||
# Corrupt again: the guard must NOT loop — the write now propagates.
|
||||
_corrupt_fts(tmp_path / "state.db")
|
||||
with pytest.raises(sqlite3.DatabaseError):
|
||||
db.append_message("s1", "user", "second corruption")
|
||||
|
||||
def test_non_fts_errors_still_propagate(self, db):
|
||||
db.create_session("s1", source="test")
|
||||
|
||||
def _bad(conn):
|
||||
raise sqlite3.IntegrityError("NOT NULL constraint failed: x.y")
|
||||
|
||||
with pytest.raises(sqlite3.IntegrityError):
|
||||
db._execute_write(_bad)
|
||||
# The guard must not have been consumed by an unrelated error class.
|
||||
assert db._fts_runtime_rebuild_attempted is False
|
||||
|
||||
def test_lock_retry_path_unchanged(self, db):
|
||||
"""A locked error still follows the jitter-retry path, untouched by
|
||||
the DatabaseError handler (OperationalError is caught first)."""
|
||||
calls = {"n": 0}
|
||||
|
||||
def _flaky(conn):
|
||||
calls["n"] += 1
|
||||
if calls["n"] < 3:
|
||||
raise sqlite3.OperationalError("database is locked")
|
||||
return "ok"
|
||||
|
||||
assert db._execute_write(_flaky) == "ok"
|
||||
assert calls["n"] == 3
|
||||
assert db._fts_runtime_rebuild_attempted is False
|
||||
Loading…
Add table
Add a link
Reference in a new issue