diff --git a/hermes_state.py b/hermes_state.py index 7e9db6a75257..19c1af24dead 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -1976,16 +1976,13 @@ class SessionDB: _WRITE_RETRY_MAX_S = 0.150 # 150ms # Attempt a WAL checkpoint every N successful writes (PASSIVE mode). _CHECKPOINT_EVERY_N_WRITES = 50 - # Merge fragmented FTS5 segments every N successful writes. The message - # triggers append one segment per insert; left unmaintained these grow - # into tens of thousands of segments, so every MATCH must scan them all - # and every insert pays a growing automerge cost — which lengthens the - # write-lock hold time and starves competing writers (gateway + cron - # processes share one state.db), surfacing as "database is locked". - # 'optimize' is a no-op once the index is already merged, so an idle DB - # pays almost nothing; the cadence is deliberately coarse so the one-off - # merge cost is amortised far below the checkpoint cadence. - _OPTIMIZE_EVERY_N_WRITES = 1000 + # Retain the existing coarse 1000-write maintenance cadence, but give each + # FTS5 index only one positive-rank merge command per pass. SQLite treats + # that rank as an approximate output-page budget. A 500-page budget keeps + # routine work small while automerge handles normal background compaction; + # unlike ``optimize``, it cannot chase every remaining segment in one write. + _FTS_MERGE_EVERY_N_WRITES = 1000 + _FTS_MERGE_MAX_PAGES_PER_INDEX = 500 # Session imports intentionally use a lower cap than exports: import holds # one BEGIN IMMEDIATE transaction, so bounded batches avoid starving live # gateway/CLI writers. The dashboard accepts one exported JSON/JSONL file @@ -2611,8 +2608,8 @@ class SessionDB: self._write_count += 1 if self._write_count % self._CHECKPOINT_EVERY_N_WRITES == 0: self._try_wal_checkpoint() - if self._write_count % self._OPTIMIZE_EVERY_N_WRITES == 0: - self._try_optimize_fts() + if self._write_count % self._FTS_MERGE_EVERY_N_WRITES == 0: + self._try_incremental_merge_fts() return result except sqlite3.OperationalError as exc: err_msg = str(exc).lower() @@ -2740,21 +2737,19 @@ class SessionDB: except Exception as exc: logger.warning("WAL checkpoint (PASSIVE) failed: %s", exc) - def _try_optimize_fts(self) -> None: - """Best-effort FTS5 segment merge. Never raises. - - Runs on the ``_OPTIMIZE_EVERY_N_WRITES`` cadence from the write hot - path (off the lock — ``optimize_fts`` re-acquires ``self._lock`` - itself, mirroring ``_try_wal_checkpoint``). ``read_only`` connections - never reach the write path, so this is implicitly skipped for them. - Once the index is merged the 'optimize' command is close to free, so - the steady-state cost is negligible; the expensive case is only the - first merge of a long-neglected index. - """ + def _try_incremental_merge_fts(self) -> None: + """Run one bounded FTS5 merge pass without failing the completed write.""" + if not self._fts_enabled: + return try: - self.optimize_fts() - except Exception: - pass # Best effort — never fatal. + self._merge_fts_incrementally( + max_pages=self._FTS_MERGE_MAX_PAGES_PER_INDEX + ) + except sqlite3.Error as exc: + # Routine maintenance is best effort, but unexpected SQLite errors + # must remain visible instead of being silently mistaken for an + # optional missing index. + logger.warning("FTS incremental merge failed: %s", exc) def close(self): """Close the database connection. @@ -11300,6 +11295,42 @@ class SessionDB: logger.debug("Could not read logical DB size: %s", exc) return None + def _merge_fts_incrementally(self, *, max_pages: int) -> int: + """Issue one positive-rank FTS5 ``merge`` per present index. + + A positive rank tells SQLite to stop after approximately that many + output pages. This method deliberately does not loop: one invocation + executes at most one merge command for each table in ``_FTS_TABLES``. + Missing tables are valid schema variants; all other SQLite errors + propagate to the caller. + """ + if isinstance(max_pages, bool) or not isinstance(max_pages, int): + raise TypeError("max_pages must be an integer") + if max_pages <= 0: + raise ValueError("max_pages must be greater than zero") + + merged = 0 + with self._lock: + rows = self._conn.execute( + "SELECT name FROM sqlite_master " + "WHERE type = 'table' AND name IN (?, ?)", + self._FTS_TABLES, + ).fetchall() + present = {row[0] for row in rows} + if "messages_fts" not in present: + raise sqlite3.OperationalError( + "required FTS5 index is missing: messages_fts" + ) + for tbl in self._FTS_TABLES: + if tbl not in present: + continue + self._conn.execute( + f"INSERT INTO {tbl}({tbl}, rank) VALUES('merge', ?)", + (max_pages,), + ) + merged += 1 + return merged + def vacuum(self) -> int: """Run VACUUM to reclaim disk space after large deletes. diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index 2ec871f0b7ff..17e26bb93029 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -5414,7 +5414,15 @@ class TestOptimizeFts: """A fresh DB has both FTS indexes; optimize merges both.""" db.create_session(session_id="s1", source="cli") db.append_message(session_id="s1", role="user", content="hello world") - assert db.optimize_fts() == 2 + statements = [] + db._conn.set_trace_callback(statements.append) + try: + assert db.optimize_fts() == 2 + finally: + db._conn.set_trace_callback(None) + optimize_sql = [sql for sql in statements if "'optimize'" in sql] + assert len(optimize_sql) == 2 + assert not any("'merge'" in sql for sql in optimize_sql) def test_optimize_preserves_search_and_snippet(self, db): """Optimize is layout-only: MATCH results + snippets are unchanged.""" @@ -5466,39 +5474,103 @@ class TestOptimizeFts: # Search still works after repeated optimization. assert len(db.search_messages("repeat")) == 1 - def test_write_path_optimizes_fts_on_cadence(self, db, monkeypatch): - """Writes periodically merge FTS segments so they never accumulate - into the tens-of-thousands that lengthen the write-lock hold and - starve competing writers ("database is locked").""" - db._OPTIMIZE_EVERY_N_WRITES = 5 - calls = {"n": 0} - real_optimize = db.optimize_fts - - def _counting_optimize(): - calls["n"] += 1 - return real_optimize() - - monkeypatch.setattr(db, "optimize_fts", _counting_optimize) - # create_session is write #1; appends are #2.. -> #5 and #10 trigger. + def test_incremental_merge_is_one_bounded_command_per_present_index(self, db): db.create_session(session_id="s1", source="cli") - for i in range(9): + db.append_message(session_id="s1", role="user", content="bounded merge") + statements = [] + db._conn.set_trace_callback(statements.append) + try: + assert db._merge_fts_incrementally(max_pages=37) == 2 + finally: + db._conn.set_trace_callback(None) + + merge_sql = [sql for sql in statements if "VALUES('merge', 37)" in sql] + assert len(merge_sql) == 2 + assert sum( + "messages_fts(messages_fts, rank)" in sql for sql in merge_sql + ) == 1 + assert sum( + "messages_fts_trigram(messages_fts_trigram, rank)" in sql + for sql in merge_sql + ) == 1 + assert not any("'optimize'" in sql for sql in statements) + + def test_incremental_merge_skips_absent_optional_index(self, db): + with db._lock: + for trigger in ( + "messages_fts_trigram_insert", + "messages_fts_trigram_delete", + "messages_fts_trigram_update", + ): + db._conn.execute(f"DROP TRIGGER {trigger}") + db._conn.execute("DROP TABLE messages_fts_trigram") + + statements = [] + db._conn.set_trace_callback(statements.append) + try: + assert db._merge_fts_incrementally(max_pages=19) == 1 + finally: + db._conn.set_trace_callback(None) + merge_sql = [sql for sql in statements if "VALUES('merge', 19)" in sql] + assert len(merge_sql) == 1 + assert "messages_fts(messages_fts, rank)" in merge_sql[0] + + def test_incremental_merge_does_not_hide_missing_required_index(self, db): + with db._lock: + for trigger in ( + "messages_fts_insert", + "messages_fts_delete", + "messages_fts_update", + ): + db._conn.execute(f"DROP TRIGGER {trigger}") + db._conn.execute("DROP TABLE messages_fts") + + with pytest.raises( + sqlite3.OperationalError, + match="required FTS5 index is missing: messages_fts", + ): + db._merge_fts_incrementally(max_pages=19) + + def test_write_path_merges_fts_only_at_cadence_boundary(self, db, monkeypatch): + """Routine writes use bounded merge and never full optimize.""" + db._FTS_MERGE_EVERY_N_WRITES = 5 + calls = [] + + def _counting_merge(*, max_pages): + calls.append(max_pages) + return 0 + + def _unexpected_optimize(): + raise AssertionError("routine cadence must not call optimize") + + monkeypatch.setattr(db, "_merge_fts_incrementally", _counting_merge) + monkeypatch.setattr(db, "optimize_fts", _unexpected_optimize) + db.create_session(session_id="s1", source="cli") + for i in range(3): db.append_message(session_id="s1", role="user", content=f"needle {i}") - assert calls["n"] == 2 - # The auto-merge is layout-only: search is unaffected. + assert calls == [] # Four successful writes are below the boundary. + db.append_message(session_id="s1", role="user", content="needle 3") + assert calls == [500] # The fifth write gets the production page budget. + for i in range(4, 8): + db.append_message(session_id="s1", role="user", content=f"needle {i}") + assert calls == [500] + db.append_message(session_id="s1", role="user", content="needle 8") + assert calls == [500, 500] # The tenth write is the next boundary. assert len(db.search_messages("needle")) == 9 - def test_write_path_optimize_failure_never_breaks_write(self, db, monkeypatch): - """A failing periodic optimize must not fail the surrounding write.""" - db._OPTIMIZE_EVERY_N_WRITES = 2 + def test_write_path_merge_failure_is_logged_without_breaking_write( + self, db, monkeypatch, caplog + ): + db._FTS_MERGE_EVERY_N_WRITES = 2 - def _boom(): - raise sqlite3.OperationalError("simulated optimize failure") + def _boom(*, max_pages): + raise sqlite3.OperationalError("simulated merge failure") - monkeypatch.setattr(db, "optimize_fts", _boom) + monkeypatch.setattr(db, "_merge_fts_incrementally", _boom) db.create_session(session_id="s1", source="cli") # write #1 - # write #2 trips the cadence; the swallowed failure must not propagate. db.append_message(session_id="s1", role="user", content="still persists") assert len(db.get_messages("s1")) == 1 + assert "FTS incremental merge failed: simulated merge failure" in caplog.text class TestAutoMaintenance: