diff --git a/gateway/session.py b/gateway/session.py index c16aa4beba20..02fdd0043aeb 100644 --- a/gateway/session.py +++ b/gateway/session.py @@ -2514,34 +2514,59 @@ class SessionStore: with self._transcript_retry_lock: pending = self._dirty_transcripts.setdefault(session_id, []) pending.append(dict(message)) - while pending: - try: - self._append_transcript_message(session_id, pending[0]) - except Exception as exc: - if self._is_fts_corruption_error(exc) and self._rebuild_fts_once(): - try: - self._append_transcript_message(session_id, pending[0]) - except Exception as retry_exc: - exc = retry_exc - else: - pending.pop(0) - continue + # Cap pending messages per session to avoid unbounded memory + # growth when the DB is persistently broken. Drop the oldest. + if len(pending) > self._MAX_PENDING_PER_SESSION: + dropped = pending.pop(0) + logger.warning( + "Session DB transcript pending queue full for %s " + "(cap=%d); dropping oldest message to make room", + session_id, self._MAX_PENDING_PER_SESSION, + ) + # Snapshot the first pending message, then release the lock + # before the DB write so other sessions are not blocked. + msg = pending[0] + # DB write outside the retry lock — other sessions can append + # concurrently. We re-acquire the lock only to update the queue. + while True: + try: + self._append_transcript_message(session_id, msg) + except Exception as exc: + if self._is_fts_corruption_error(exc) and self._rebuild_fts_once(): + try: + self._append_transcript_message(session_id, msg) + except Exception as retry_exc: + exc = retry_exc + else: + with self._transcript_retry_lock: + if pending and pending[0] is msg: + pending.pop(0) + if not pending: + self._dirty_transcripts.pop(session_id, None) + self._transcript_append_failures.pop(session_id, None) + continue + with self._transcript_retry_lock: failures = self._transcript_append_failures.get(session_id, 0) + 1 self._transcript_append_failures[session_id] = failures - logger.warning( - "Session DB transcript append failed for %s " - "(failure_count=%d, pending=%d); will retry: %s", - session_id, failures, len(pending), exc, - ) - break - else: - pending.pop(0) - if not pending: - self._dirty_transcripts.pop(session_id, None) - self._transcript_append_failures.pop(session_id, None) + logger.warning( + "Session DB transcript append failed for %s " + "(failure_count=%d, pending=%d); will retry: %s", + session_id, failures, len(pending), exc, + ) + return + else: + with self._transcript_retry_lock: + if pending and pending[0] is msg: + pending.pop(0) + if not pending: + self._dirty_transcripts.pop(session_id, None) + self._transcript_append_failures.pop(session_id, None) + return + msg = pending[0] + continue def _append_transcript_message(self, session_id: str, message: Dict[str, Any]) -> None: - """Write one transcript row. Caller serializes retries per session store.""" + """Write one transcript row. Caller handles retry queuing.""" self._db.append_message( session_id=session_id, role=message.get("role", "unknown"), @@ -2559,40 +2584,65 @@ class SessionStore: timestamp=message.get("timestamp"), ) + # Maximum in-memory pending messages per session before dropping the + # oldest. Prevents unbounded growth when the DB is persistently broken. + _MAX_PENDING_PER_SESSION = 200 + @staticmethod def _is_fts_corruption_error(exc: Exception) -> bool: + """True if *exc* looks like an FTS index corruption error. + + Matches the specific SQLite error strings for malformed disk images + and FTS table corruption — not bare ``"fts"`` substrings which match + unrelated words like ``"shifts"`` or ``"gifts"``. + """ text = str(exc).lower() - return "database disk image is malformed" in text or "fts" in text + return any( + marker in text + for marker in ( + "database disk image is malformed", + "malformed database schema", + "messages_fts", + "no such table: messages_fts", + ) + ) def _rebuild_fts_once(self) -> bool: - """Attempt SQLite's documented FTS5 rebuild command once per store.""" + """Attempt FTS5 ``rebuild`` command once per store lifetime. + + Delegates to ``SessionDB.rebuild_fts()`` which handles locking and + table-existence checks internally. Returns ``True`` when at least + one index was rebuilt. + """ if self._fts_rebuild_attempted: return False self._fts_rebuild_attempted = True db = self._db - conn = getattr(db, "_conn", None) - lock = getattr(db, "_lock", None) - if conn is None or lock is None: + if db is None or not hasattr(db, "rebuild_fts"): return False - rebuilt = 0 try: - with lock: - for table in getattr(db, "_FTS_TABLES", ("messages_fts", "messages_fts_trigram")): - try: - if hasattr(db, "_fts_table_exists") and not db._fts_table_exists(table): - continue - conn.execute(f"INSERT INTO {table}({table}) VALUES('rebuild')") - conn.commit() - rebuilt += 1 - except Exception as exc: - conn.rollback() - logger.warning("Session DB FTS rebuild failed for %s: %s", table, exc) + rebuilt = db.rebuild_fts() except Exception as exc: logger.warning("Session DB FTS rebuild failed: %s", exc) return False if rebuilt: - logger.warning("Rebuilt %d Session DB FTS index(es) after append corruption", rebuilt) + logger.warning( + "Rebuilt %d Session DB FTS index(es) after append corruption", + rebuilt, + ) return rebuilt > 0 + + def _clear_dirty_transcript(self, session_id: str) -> None: + """Drop queued pending messages for a session. + + Called by ``rewrite_transcript`` and ``rewind_session`` so that + /retry, /undo, /compress — which replace or truncate the transcript — + don't leave stale messages that would be re-inserted on the next + append. + """ + with self._transcript_retry_lock: + self._dirty_transcripts.pop(session_id, None) + self._transcript_append_failures.pop(session_id, None) def has_platform_message_id( self, session_id: str, platform_message_id: str @@ -2628,6 +2678,7 @@ class SessionStore: """ if not self._db: return True + self._clear_dirty_transcript(session_id) try: self._db.replace_messages(session_id, messages) return True @@ -2670,6 +2721,7 @@ class SessionStore: """ if not self._db: return None + self._clear_dirty_transcript(session_id) if n < 1: n = 1 try: diff --git a/hermes_state.py b/hermes_state.py index 632017b958f5..a463d0de50c5 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -7273,6 +7273,37 @@ class SessionDB: ) return optimized + def rebuild_fts(self) -> int: + """Rebuild FTS5 indexes from the canonical ``messages`` table. + + Uses the FTS5 ``'rebuild'`` command, which rewrites the internal + b-tree segments from the content rows. This is the documented + recovery for a corrupt FTS index that rejects message writes while + reads still succeed (issue #50502). Unlike ``optimize_fts`` (which + merges existing segments), ``rebuild`` discards and recreates the + index data entirely. + + Safe to call when FTS tables don't exist (skips them). + Returns the number of FTS indexes that were rebuilt. + """ + rebuilt = 0 + with self._lock: + for tbl in self._FTS_TABLES: + if not self._fts_table_exists(tbl): + continue + try: + self._conn.execute( + f"INSERT INTO {tbl}({tbl}) VALUES('rebuild')" + ) + self._conn.commit() + rebuilt += 1 + except sqlite3.OperationalError as exc: + self._conn.rollback() + logger.warning( + "FTS rebuild failed for %s: %s", tbl, exc + ) + return rebuilt + def vacuum(self) -> int: """Run VACUUM to reclaim disk space after large deletes. diff --git a/tests/gateway/test_session.py b/tests/gateway/test_session.py index b146bdb2f003..d0bf2913432a 100644 --- a/tests/gateway/test_session.py +++ b/tests/gateway/test_session.py @@ -1663,30 +1663,15 @@ class TestGatewaySessionDbRecovery: def test_transcript_append_rebuilds_fts_and_retries_dirty_rows_in_order(self): import threading - class FakeConnection: - def __init__(self): - self.sql = [] - - def execute(self, sql): - self.sql.append(sql) - - def commit(self): - pass - - def rollback(self): - pass - class FakeDb: - _FTS_TABLES = ("messages_fts",) - def __init__(self): - self._lock = threading.Lock() - self._conn = FakeConnection() self.attempts = [] self.persisted = [] + self.rebuild_calls = 0 - def _fts_table_exists(self, _table): - return True + def rebuild_fts(self): + self.rebuild_calls += 1 + return 1 def append_message(self, **kwargs): content = kwargs["content"] @@ -1704,14 +1689,127 @@ class TestGatewaySessionDbRecovery: store.append_to_transcript("s1", {"role": "user", "content": "first"}) assert [m["content"] for m in store._dirty_transcripts["s1"]] == ["first"] + assert store._db.rebuild_calls == 1 store.append_to_transcript("s1", {"role": "assistant", "content": "second"}) assert store._db.persisted == ["first", "second"] assert "s1" not in store._dirty_transcripts - assert store._db._conn.sql == [ - "INSERT INTO messages_fts(messages_fts) VALUES('rebuild')" - ] + + def test_transcript_append_clears_dirty_on_rewrite(self): + """rewrite_transcript must clear pending dirty messages so /retry + and /compress don't re-insert replaced rows.""" + import threading + + class FakeDb: + def __init__(self): + self.persisted = [] + self.replaced = [] + + def rebuild_fts(self): + return 0 + + def append_message(self, **kwargs): + raise RuntimeError("database disk image is malformed") + + def replace_messages(self, session_id, messages): + self.replaced.append((session_id, messages)) + + store = object.__new__(SessionStore) + store._db = FakeDb() + store._transcript_retry_lock = threading.Lock() + store._dirty_transcripts = {} + store._transcript_append_failures = {} + store._fts_rebuild_attempted = True # prevent rebuild attempt + + # Queue a failed message + store.append_to_transcript("s1", {"role": "user", "content": "stale"}) + assert "s1" in store._dirty_transcripts + + # rewrite_transcript should clear the dirty queue + store.rewrite_transcript("s1", [{"role": "user", "content": "fresh"}]) + assert "s1" not in store._dirty_transcripts + assert len(store._db.replaced) == 1 + + def test_transcript_append_clears_dirty_on_rewind(self): + """rewind_session must clear pending dirty messages so /undo + doesn't re-insert rewound rows.""" + import threading + + class FakeDb: + def __init__(self): + self.persisted = [] + + def rebuild_fts(self): + return 0 + + def append_message(self, **kwargs): + raise RuntimeError("database disk image is malformed") + + def list_recent_user_messages(self, session_id, limit=10): + return [{"id": 1, "content": "old"}] + + def rewind_to_message(self, session_id, target_id): + return {"target_message": {"id": target_id, "content": "old"}} + + store = object.__new__(SessionStore) + store._db = FakeDb() + store._transcript_retry_lock = threading.Lock() + store._dirty_transcripts = {} + store._transcript_append_failures = {} + store._fts_rebuild_attempted = True + + store.append_to_transcript("s1", {"role": "user", "content": "stale"}) + assert "s1" in store._dirty_transcripts + + store.rewind_session("s1", 1) + assert "s1" not in store._dirty_transcripts + + def test_fts_corruption_error_does_not_match_false_positives(self): + """_is_fts_corruption_error must not match unrelated error strings + containing 'fts' as a substring (e.g. 'shifts', 'gifts').""" + assert SessionStore._is_fts_corruption_error( + RuntimeError("database disk image is malformed") + ) + assert SessionStore._is_fts_corruption_error( + RuntimeError("no such table: messages_fts") + ) + assert not SessionStore._is_fts_corruption_error( + RuntimeError("shifts were applied") + ) + assert not SessionStore._is_fts_corruption_error( + RuntimeError("gifts received") + ) + + def test_pending_queue_caps_at_max(self): + """Pending queue should drop oldest messages when exceeding the cap + to prevent unbounded memory growth on persistent DB failure.""" + import threading + + class FakeDb: + def __init__(self): + self.count = 0 + + def rebuild_fts(self): + return 0 + + def append_message(self, **kwargs): + self.count += 1 + raise RuntimeError("database disk image is malformed") + + store = object.__new__(SessionStore) + store._db = FakeDb() + store._transcript_retry_lock = threading.Lock() + store._dirty_transcripts = {} + store._transcript_append_failures = {} + store._fts_rebuild_attempted = True + + # Fill beyond the cap + for i in range(store._MAX_PENDING_PER_SESSION + 10): + store.append_to_transcript("s1", {"role": "user", "content": f"msg{i}"}) + + pending = store._dirty_transcripts.get("s1", []) + assert len(pending) <= store._MAX_PENDING_PER_SESSION def test_new_session_records_gateway_peer_fields(self, tmp_path): store = SessionStore(sessions_dir=tmp_path, config=GatewayConfig())