From df841d342ca8a058c8185980ee4d7235b746da4f Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Tue, 28 Jul 2026 12:12:15 -0700 Subject: [PATCH] =?UTF-8?q?fix(state):=20complete=20the=20bounded-merge=20?= =?UTF-8?q?protocol=20=E2=80=94=20usermerge=20floor,=20progress-bounded=20?= =?UTF-8?q?continuation,=20tolerate=20mid-rebuild=20missing=20index?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up on top of #65554 (@the3asic): - Lower FTS5 'usermerge' to its minimum of 2 (persisted in the config shadow table, applied once per SessionDB instance). Without this a positive-rank 'merge' skips any level holding fewer than 4 segments (SQLite FTS5 §6.8), so the fragmented-index case the cadence targets never converges. - Run up to _FTS_MERGE_COMMANDS_PER_PASS (4) bounded merge commands per index per cadence, stopping early on the documented no-progress signal (total_changes delta < 2). Each command is its own implicit transaction, so the write lock is released between commands. - Skip a missing messages_fts instead of raising: the chunked optimize-storage rebuild legitimately drops + backfills FTS tables while writers keep running; warning every 1000 writes for the whole backfill window would be noise, and optimize_fts() has always treated missing tables as skippable. - Replace traced-SQL shape assertions with behavioral tests: real fragmented-index convergence (automerge suppressed, 60 segments) and a 3-segment below-default-usermerge compaction test that fails against a bare positive-rank merge (sabotage-verified). Validation on a sqlite3.backup() copy of a real 10.7 GB production state.db (1.49M messages, 1.3 GB + 2.9 GB FTS shadow tables): worst per-command write-lock hold 41.8 ms (was 9.2 s / 18.1 s per index with 'optimize'), search results byte-identical, fts5 integrity-check and PRAGMA integrity_check clean, steady-state pass 0.0 ms. --- hermes_state.py | 106 ++++++++++++++++++++--------- tests/test_hermes_state.py | 135 +++++++++++++++++++++++++++++++------ 2 files changed, 192 insertions(+), 49 deletions(-) diff --git a/hermes_state.py b/hermes_state.py index 19c1af24dead..743db7ae3944 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -1976,13 +1976,21 @@ class SessionDB: _WRITE_RETRY_MAX_S = 0.150 # 150ms # Attempt a WAL checkpoint every N successful writes (PASSIVE mode). _CHECKPOINT_EVERY_N_WRITES = 50 - # 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. + # Retain the existing coarse 1000-write maintenance cadence, but replace + # the unbounded FTS5 ``'optimize'`` (measured holding the write lock for + # 9-18 s per index on a 10 GB production DB — longer than a competing + # writer's full retry patience, surfacing as "database is locked" / + # session_persistence_failed) with bounded ``'merge'`` commands. A + # positive merge rank is an approximate output-page budget, so each + # command holds the write lock for milliseconds; up to + # ``_FTS_MERGE_COMMANDS_PER_PASS`` commands run per index per cadence, + # stopping early on the documented no-progress signal. ``usermerge`` is + # lowered to 2 so positive merges act on any level with >= 2 segments — + # without that, levels below the default threshold of 4 are skipped and + # a fragmented index never converges (SQLite FTS5 §6.8-6.9). _FTS_MERGE_EVERY_N_WRITES = 1000 _FTS_MERGE_MAX_PAGES_PER_INDEX = 500 + _FTS_MERGE_COMMANDS_PER_PASS = 4 # 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 @@ -2020,6 +2028,9 @@ class SessionDB: # 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 + # One-shot guard for the usermerge-floor config write on the + # incremental FTS merge cadence (see _merge_fts_incrementally). + self._fts_usermerge_floor_applied = False self._fts_enabled = False self._trigram_available = False # CJK-bigram index (cjk_unicode61 loadable tokenizer). _fts_cjk_loaded: @@ -11295,41 +11306,76 @@ 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. + def _merge_fts_incrementally( + self, *, max_pages: int, max_commands: Optional[int] = None + ) -> int: + """Run bounded FTS5 ``'merge'`` commands against each 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. + A positive merge rank tells SQLite to stop after approximately that + many output pages, so each command holds the write lock for + milliseconds regardless of index size — unlike ``'optimize'``, which + rewrites the whole index in one transaction (measured 9-18 s per + index on a 10 GB production DB, long enough to exhaust a competing + writer's entire lock-retry patience). + + Protocol (SQLite FTS5 §6.8-6.9): + + - ``usermerge`` is lowered to its minimum of 2 (persisted in the + ``%_config`` shadow table, applied once per instance) so a + positive merge acts on ANY level holding >= 2 segments. With the + default of 4, levels below that threshold are never merged by a + positive-rank command and a fragmented index cannot converge. + - Up to *max_commands* merge commands run per index, stopping early + on the documented no-progress signal: the delta in + ``total_changes`` is < 2 (the command's own INSERT accounts + for 1 change; >= 2 means real merge work happened). + + Each command is its own implicit transaction (the connection runs + with ``isolation_level=None``), so the SQLite write lock is released + between commands and competing processes can interleave writes + mid-pass. Missing tables are valid schema variants (FTS variants are + optional, and ``optimize_fts_storage`` legitimately drops + backfills + these tables while writers keep running) and are skipped, mirroring + ``optimize_fts``. Other SQLite errors propagate to the caller. + + Returns the number of merge commands executed. """ 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") + if max_commands is None: + max_commands = self._FTS_MERGE_COMMANDS_PER_PASS + if isinstance(max_commands, bool) or not isinstance(max_commands, int): + raise TypeError("max_commands must be an integer") + if max_commands <= 0: + raise ValueError("max_commands must be greater than zero") - merged = 0 + executed = 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: + if not self._fts_table_exists(tbl): continue - self._conn.execute( - f"INSERT INTO {tbl}({tbl}, rank) VALUES('merge', ?)", - (max_pages,), - ) - merged += 1 - return merged + # One-time (per instance) usermerge floor; the value is + # persisted in the index's config shadow table so future + # connections inherit it. Setting config is a metadata-only + # write — it never touches segment data. + if not getattr(self, "_fts_usermerge_floor_applied", False): + self._conn.execute( + f"INSERT INTO {tbl}({tbl}, rank) " + "VALUES('usermerge', 2)" + ) + for _ in range(max_commands): + before = self._conn.total_changes + self._conn.execute( + f"INSERT INTO {tbl}({tbl}, rank) VALUES('merge', ?)", + (max_pages,), + ) + executed += 1 + if self._conn.total_changes - before < 2: + break + self._fts_usermerge_floor_applied = True + return executed 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 17e26bb93029..2b503e0ede9c 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -5474,27 +5474,115 @@ class TestOptimizeFts: # Search still works after repeated optimization. assert len(db.search_messages("repeat")) == 1 - def test_incremental_merge_is_one_bounded_command_per_present_index(self, db): + def test_incremental_merge_bounded_commands_per_present_index(self, db): + """Each pass issues bounded 'merge' commands, never 'optimize'.""" db.create_session(session_id="s1", source="cli") 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 + executed = db._merge_fts_incrementally(max_pages=37) finally: db._conn.set_trace_callback(None) + # At least one merge command per present FTS index, and never more + # than the per-pass command cap per index. + present = [t for t in db._FTS_TABLES if db._fts_table_exists(t)] + assert len(present) >= 2 # messages_fts + trigram on a fresh DB 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 len(merge_sql) == executed + assert len(present) <= executed <= ( + len(present) * db._FTS_MERGE_COMMANDS_PER_PASS + ) + for tbl in present: + n = sum(f"{tbl}({tbl}, rank)" in sql for sql in merge_sql) + assert 1 <= n <= db._FTS_MERGE_COMMANDS_PER_PASS + # The usermerge floor is applied so positive merges can make + # progress on levels with >= 2 segments (SQLite FTS5 §6.8). + assert any("VALUES('usermerge', 2)" in sql for sql in statements) assert not any("'optimize'" in sql for sql in statements) + def test_incremental_merge_converges_on_fragmented_index(self, db): + """Bounded passes make real progress on a fragmented index and + reach a no-more-work steady state — the failure mode of a bare + positive-rank merge (usermerge default 4) is that it never merges + anything and the index stays fragmented forever.""" + db.create_session(session_id="s1", source="cli") + # Suppress automerge so every insert leaves its own level-0 segment + # — a deliberately fragmented index that only explicit merge + # commands can compact. Config is scoped to this test's temp DB. + with db._lock: + for tbl in ("messages_fts", "messages_fts_trigram"): + db._conn.execute( + f"INSERT INTO {tbl}({tbl}, rank) VALUES('automerge', 0)" + ) + for i in range(60): + db.append_message( + session_id="s1", role="user", + content=f"fragment needle {i} lorem ipsum dolor", + ) + + # First pass must do real merge work (shadow-table rows change). + before = db._conn.total_changes + executed_first = db._merge_fts_incrementally(max_pages=500) + assert executed_first >= 2 + assert db._conn.total_changes - before > executed_first # real work + + # Repeated passes converge: eventually a pass issues exactly one + # no-progress command per present index and stops early. + present = sum(1 for t in db._FTS_TABLES if db._fts_table_exists(t)) + for _ in range(50): + if db._merge_fts_incrementally(max_pages=500) == present: + break + else: + pytest.fail("bounded merge passes never converged") + + # Merging is layout-only: every row is still searchable. + assert len(db.search_messages("fragment", limit=100)) == 60 + + def test_incremental_merge_compacts_below_default_usermerge(self, db): + """A level with only 2-3 segments — below FTS5's default usermerge + threshold of 4 — must still get merged. A bare positive-rank + 'merge' skips such levels entirely (SQLite FTS5 §6.8), which is + why the cadence lowers usermerge to 2 first; without the floor, + light fragmentation persists forever and every MATCH pays for it.""" + + def _segment_count(tbl: str) -> int: + # FTS5 structure record: id=10 in the %_data shadow table. + # Format: 4-byte cookie, then varint nlevel, varint nsegment... + # nsegment fits in one varint byte for small indexes; parse the + # second varint (single-byte values < 128 read directly). + blob = db._conn.execute( + f"SELECT block FROM {tbl}_data WHERE id=10" + ).fetchone()[0] + pos = 4 + for _ in range(1): # skip nlevel varint (single byte here) + assert blob[pos] < 0x80 + pos += 1 + assert blob[pos] < 0x80 + return blob[pos] + + db.create_session(session_id="s1", source="cli") + with db._lock: + db._conn.execute( + "INSERT INTO messages_fts(messages_fts, rank) " + "VALUES('automerge', 0)" + ) + # Exactly 3 level-0 segments: below the default usermerge of 4. + for i in range(3): + db.append_message( + session_id="s1", role="user", content=f"sparse token{i}" + ) + assert _segment_count("messages_fts") == 3 + + for _ in range(10): + db._merge_fts_incrementally(max_pages=500) + + assert _segment_count("messages_fts") == 1, ( + "3-segment level was not compacted — usermerge floor missing?" + ) + assert len(db.search_messages("sparse")) == 3 + def test_incremental_merge_skips_absent_optional_index(self, db): with db._lock: for trigger in ( @@ -5508,14 +5596,18 @@ class TestOptimizeFts: statements = [] db._conn.set_trace_callback(statements.append) try: - assert db._merge_fts_incrementally(max_pages=19) == 1 + 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] + assert merge_sql + assert all("messages_fts(messages_fts, rank)" in sql for sql in merge_sql) - def test_incremental_merge_does_not_hide_missing_required_index(self, db): + def test_incremental_merge_skips_missing_primary_index(self, db): + """A missing messages_fts is a valid transient state (chunked + optimize-storage rebuild drops + backfills it while writers keep + running) — the cadence must skip it, not raise a warning every + 1000 writes for the whole backfill window.""" with db._lock: for trigger in ( "messages_fts_insert", @@ -5525,11 +5617,16 @@ class TestOptimizeFts: 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) + statements = [] + db._conn.set_trace_callback(statements.append) + try: + executed = db._merge_fts_incrementally(max_pages=19) + finally: + db._conn.set_trace_callback(None) + assert executed >= 1 # trigram index still present and merged + assert not any( + "messages_fts(messages_fts, rank)" in sql for sql in statements + ) def test_write_path_merges_fts_only_at_cadence_boundary(self, db, monkeypatch): """Routine writes use bounded merge and never full optimize."""