From f0fd360804afbe6aeddb2bc12f6c86d0a2b2f165 Mon Sep 17 00:00:00 2001 From: Eva Date: Mon, 3 Aug 2026 22:20:58 +0700 Subject: [PATCH] fix: serialize concurrent FTS bootstrap repair --- db_bootstrap.py | 200 +++++++++++++++++++----- tests/test_db_bootstrap_fts.py | 271 +++++++++++++++++++++++++++++++++ 2 files changed, 431 insertions(+), 40 deletions(-) diff --git a/db_bootstrap.py b/db_bootstrap.py index f89476e7..9115e9c7 100644 --- a/db_bootstrap.py +++ b/db_bootstrap.py @@ -15,6 +15,7 @@ import sqlite3 import threading import time +from contextlib import contextmanager from typing import Iterable, Sequence logger = logging.getLogger(__name__) @@ -67,6 +68,32 @@ def __init__( self.trigger_sqls = tuple(trigger_sqls) +def _is_sqlite_lock_error(exc: BaseException) -> bool: + """Return True when an exception chain represents SQLite lock contention.""" + lock_codes = {sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED} + lock_messages = ( + "database is locked", + "database table is locked", + "database schema is locked", + "database is busy", + "database table is busy", + "database schema is busy", + ) + seen: set[int] = set() + current: BaseException | None = exc + while current is not None and id(current) not in seen: + seen.add(id(current)) + if isinstance(current, sqlite3.Error): + error_code = getattr(current, "sqlite_errorcode", None) + if isinstance(error_code, int) and (error_code & 0xFF) in lock_codes: + return True + detail = str(current).lower() + if any(message in detail for message in lock_messages): + return True + current = current.__cause__ or current.__context__ + return False + + def configure_connection(conn: sqlite3.Connection) -> None: """Configure SQLite connection for WAL durability and hygiene. @@ -1798,7 +1825,12 @@ def _fts_needs_rebuild_structural(conn: sqlite3.Connection, spec: ExternalConten ).fetchone()[0] if int(content_count or 0) != int(fts_count or 0): return True - except sqlite3.DatabaseError: + except sqlite3.DatabaseError as exc: + # A busy/locked snapshot is an availability problem, not FTS + # corruption. Let the bounded caller transaction report it instead of + # turning lock exhaustion into destructive repair. + if _is_sqlite_lock_error(exc): + raise return True return False @@ -2306,6 +2338,53 @@ def external_content_fts_needs_repair(conn: sqlite3.Connection, spec: ExternalCo return _fts_needs_rebuild_structural(conn, spec) or _fts_missing_triggers(conn, spec) +@contextmanager +def _fts_repair_ownership(conn: sqlite3.Connection): + """Own the short FTS structural-repair transaction. + + Fresh startup connections are normally outside a transaction, so + ``BEGIN IMMEDIATE`` serializes the structural recheck and all FTS DDL + across independent processes. A caller that already owns a transaction + gets a savepoint instead; committing that caller-owned transaction remains + the historical behavior of the repair helper. + """ + if conn.in_transaction: + savepoint = quote_sql_identifier("lcm_fts_repair_ownership") + conn.execute(f"SAVEPOINT {savepoint}") + try: + yield False + except BaseException: + try: + conn.execute(f"ROLLBACK TO {savepoint}") + finally: + conn.execute(f"RELEASE {savepoint}") + raise + else: + conn.execute(f"RELEASE {savepoint}") + return + + previous_timeout = conn.execute("PRAGMA busy_timeout").fetchone() + previous_timeout_ms = int(previous_timeout[0]) if previous_timeout else 0 + conn.execute(f"PRAGMA busy_timeout={SQLITE_BUSY_TIMEOUT_MS}") + try: + # A loser waits for the winner, then takes a current write snapshot and + # rechecks the complete FTS state before deciding whether to repair. + conn.execute("BEGIN IMMEDIATE") + try: + yield True + except BaseException: + conn.rollback() + raise + else: + try: + conn.commit() + except BaseException: + conn.rollback() + raise + finally: + conn.execute(f"PRAGMA busy_timeout={previous_timeout_ms}") + + def repair_external_content_fts( conn: sqlite3.Connection, spec: ExternalContentFtsSpec, @@ -2315,51 +2394,92 @@ def repair_external_content_fts( ) -> dict[str, bool]: rebuilt = False degraded = False - if _fts_needs_rebuild(conn, spec, now=now, throttle=throttle): - db_path = conn.execute("PRAGMA database_list").fetchone() - if db_path: - db_file = db_path[2] - if db_file and not _check_disk_space(db_file): + structural_repair_needed = external_content_fts_needs_repair(conn, spec) + deep_repair_needed = False + if not structural_repair_needed: + # Preserve the cheap startup path and its background integrity-scan + # behavior. Only a caller that observed repair-worthy state enters the + # write-ownership boundary below. + deep_repair_needed = _fts_needs_rebuild(conn, spec, now=now, throttle=throttle) + if not deep_repair_needed: + # A trigger can disappear after the initial complete-state check. + # Return on the healthy fast path only while it is still complete; + # any observed trigger repair must pass through write ownership below. + if not _fts_missing_triggers(conn, spec): + _clear_integrity_failed(conn, spec) + conn.commit() + return { + "rebuilt": False, + "degraded": False, + "triggers_recreated": False, + } + + with _fts_repair_ownership(conn) as owns_transaction: + # This is the decisive cross-process recheck. If another process won + # while we waited, accept its complete table/shadow/trigger state and do + # not drop or recreate it. + winner_state_needs_repair = external_content_fts_needs_repair(conn, spec) + owner_rebuild_needed = ( + _fts_needs_rebuild_structural(conn, spec) if winner_state_needs_repair else False + ) + if not owner_rebuild_needed and not structural_repair_needed and deep_repair_needed: + # Explicit repair (and synchronous startup when background scans are + # disabled) must revalidate same-row-count token drift while owning + # the write boundary. A repaired winner is accepted without a second + # destructive rebuild. + owner_rebuild_needed = _fts_needs_rebuild( + conn, spec, now=now, throttle=False + ) + if owner_rebuild_needed: + db_path = conn.execute("PRAGMA database_list").fetchone() + low_disk = False + if db_path: + db_file = db_path[2] + low_disk = bool(db_file and not _check_disk_space(db_file)) + if low_disk: logger.warning( "Low disk space for FTS rebuild of '%s' (%d MB needed), degrading to LIKE search", spec.table_name, _MIN_DISK_SPACE_BYTES // (1024 * 1024), ) _drop_fts_artifacts(conn, spec) - # The corrupt index is gone (degraded to LIKE search); a stale - # integrity-failed flag would otherwise keep `/lcm doctor` - # reporting issues-found for an index that no longer exists. - _clear_integrity_failed(conn, spec) - conn.commit() - return {"rebuilt": False, "degraded": True, "triggers_recreated": False} - _drop_fts_table(conn, spec.table_name) - conn.execute( - f""" - CREATE VIRTUAL TABLE {quote_sql_identifier(spec.table_name)} USING fts5( - {quote_sql_identifier(spec.indexed_column)}, - content={quote_sql_identifier(spec.content_table)}, - content_rowid={quote_sql_identifier(spec.content_rowid)} - ) - """ - ) - conn.execute( - f"INSERT INTO {quote_sql_identifier(spec.table_name)}({quote_sql_identifier(spec.table_name)}) VALUES('rebuild')" - ) - rebuilt = True - - triggers_were_missing = _fts_missing_triggers(conn, spec) - for trigger_sql in spec.trigger_sqls: - conn.execute(trigger_sql) - if rebuilt: - # A freshly rebuilt index is known-consistent; record the marker so the - # next startup can skip the deep integrity-check within the interval. - _record_integrity_checked(conn, spec, now=now) - # A completed repair resolves any prior background-scan corruption flag: clear - # it in the SAME transaction that commits the rebuild so `/lcm doctor` stops - # reporting issues-found (and the next self-healing scan is not pushed out a - # full interval). Without this an explicit `repair apply` left the flag stuck. - _clear_integrity_failed(conn, spec) - conn.commit() + degraded = True + else: + _drop_fts_table(conn, spec.table_name) + conn.execute( + f""" + CREATE VIRTUAL TABLE {quote_sql_identifier(spec.table_name)} USING fts5( + {quote_sql_identifier(spec.indexed_column)}, + content={quote_sql_identifier(spec.content_table)}, + content_rowid={quote_sql_identifier(spec.content_rowid)} + ) + """ + ) + conn.execute( + f"INSERT INTO {quote_sql_identifier(spec.table_name)}({quote_sql_identifier(spec.table_name)}) VALUES('rebuild')" + ) + rebuilt = True + + if degraded: + triggers_were_missing = False + else: + triggers_were_missing = _fts_missing_triggers(conn, spec) + for trigger_sql in spec.trigger_sqls: + conn.execute(trigger_sql) + if rebuilt: + # A freshly rebuilt index is known-consistent; record the marker so + # the next startup can skip the deep integrity-check within the + # interval. + _record_integrity_checked(conn, spec, now=now) + # A completed repair resolves any prior background-scan corruption flag: + # clear it in the SAME transaction that commits the rebuild. + _clear_integrity_failed(conn, spec) + + if not owns_transaction: + # Preserve the helper's historical behavior for callers that supplied an + # already-active transaction while keeping the startup path's ownership + # boundary isolated and rollback-safe. + conn.commit() return {"rebuilt": rebuilt, "degraded": degraded, "triggers_recreated": triggers_were_missing} diff --git a/tests/test_db_bootstrap_fts.py b/tests/test_db_bootstrap_fts.py index 5f3e647f..620a449f 100644 --- a/tests/test_db_bootstrap_fts.py +++ b/tests/test_db_bootstrap_fts.py @@ -12,13 +12,26 @@ the index first, then exercise the existing-index path. """ +import json +import multiprocessing as mp import sqlite3 +import subprocess +import sys import threading import time +import traceback import types +from pathlib import Path import pytest + +if "hermes_lcm" not in sys.modules: + package = types.ModuleType("hermes_lcm") + package.__path__ = [str(Path(__file__).resolve().parents[1])] + package.__package__ = "hermes_lcm" + sys.modules["hermes_lcm"] = package + from hermes_lcm import command, db_bootstrap from hermes_lcm.db_bootstrap import ( ExternalContentFtsSpec, @@ -54,6 +67,218 @@ def _spec(): ) +def _spawn_message_store_worker(db_path, start_barrier, repair_barrier, queue, worker): + """Construct and append after all children observe incomplete FTS state.""" + store = None + try: + start_barrier.wait(timeout=30) + from hermes_lcm import db_bootstrap as worker_db_bootstrap + from hermes_lcm.store import MessageStore + + original_structural_check = worker_db_bootstrap._fts_needs_rebuild_structural + synchronized = False + + def synchronized_structural_check(conn, spec): + nonlocal synchronized + result = original_structural_check(conn, spec) + if result and not synchronized: + synchronized = True + repair_barrier.wait(timeout=30) + return result + + worker_db_bootstrap._fts_needs_rebuild_structural = synchronized_structural_check + store = MessageStore(db_path) + messages = [ + { + "role": "user", + "content": f"ftsbootstrapworker{worker} token{index}", + } + for index in range(4) + ] + ids = store.append_batch( + f"session-{worker}", + messages, + [1] * len(messages), + source="spawn-regression", + conversation_id=f"conversation-{worker}", + ) + queue.put({"ok": True, "worker": worker, "ids": ids}) + except BaseException as exc: # pragma: no cover - exercised in child + queue.put( + { + "ok": False, + "worker": worker, + "error": repr(exc), + "trace": traceback.format_exc(), + } + ) + finally: + if store is not None: + store.close() + + +def _run_spawn_message_store_probe(db_path): + workers = 6 + ctx = mp.get_context("spawn") + start_barrier = ctx.Barrier(workers + 1) + repair_barrier = ctx.Barrier(workers) + queue = ctx.Queue() + processes = [ + ctx.Process( + target=_spawn_message_store_worker, + args=(db_path, start_barrier, repair_barrier, queue, worker), + ) + for worker in range(workers) + ] + started_processes = [] + results = [] + deadline = time.monotonic() + 90 + try: + for process in processes: + process.start() + started_processes.append(process) + start_barrier.wait(timeout=max(0.1, min(30, deadline - time.monotonic()))) + while len(results) < workers: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise AssertionError(f"timed out waiting for workers: {results!r}") + results.append(queue.get(timeout=remaining)) + finally: + join_deadline = time.monotonic() + 10 + for process in started_processes: + process.join(timeout=max(0, join_deadline - time.monotonic())) + alive = [process for process in started_processes if process.is_alive()] + for process in alive: + process.terminate() + terminate_deadline = time.monotonic() + 10 + for process in alive: + process.join(timeout=max(0, terminate_deadline - time.monotonic())) + queue.close() + queue.join_thread() + + return { + "results": results, + "exitcodes": [process.exitcode for process in started_processes], + } + + +if __name__ == "__main__" and sys.argv[1:2] == ["--spawn-fts-bootstrap"]: + print(json.dumps(_run_spawn_message_store_probe(sys.argv[2]))) + raise SystemExit(0) + + +def test_spawned_message_store_startup_serializes_fresh_fts_repair(tmp_path): + """Independent constructors accept one winner's complete FTS state.""" + from hermes_lcm.store import MessageStore, build_message_fts_spec + + workers = 6 + db_path = str(tmp_path / "spawn-fresh-fts.db") + completed = subprocess.run( + [ + sys.executable, + str(Path(__file__).resolve()), + "--spawn-fts-bootstrap", + db_path, + ], + capture_output=True, + text=True, + timeout=180, + check=False, + ) + assert completed.returncode == 0, completed.stderr + probe = json.loads(completed.stdout) + results = probe["results"] + assert all(result["ok"] for result in results), results + assert all(exitcode == 0 for exitcode in probe["exitcodes"]), probe["exitcodes"] + + # This is a fresh product-store reopen after the contention round, not only + # a direct SQLite audit of the winner's file. + store = MessageStore(db_path) + try: + conn = store.connection + assert conn is not None + spec = build_message_fts_spec() + assert db_bootstrap.check_external_content_fts_integrity(conn, spec)["status"] == "pass" + assert conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0] == workers * 4 + assert conn.execute("SELECT COUNT(*) FROM messages_fts").fetchone()[0] == workers * 4 + assert conn.execute("SELECT COUNT(*) FROM messages_fts_docsize").fetchone()[0] == workers * 4 + assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok" + assert conn.execute("PRAGMA foreign_key_check").fetchall() == [] + for worker in range(workers): + matches = store.search(f"ftsbootstrapworker{worker}", limit=10) + assert len(matches) == 4 + assert all(f"ftsbootstrapworker{worker}" in row["content"] for row in matches) + finally: + store.close() + + +def test_trigger_disappearing_on_fast_path_reenters_repair_ownership( + tmp_path, monkeypatch +): + """Trigger DDL observed after the healthy precheck runs only under ownership.""" + from hermes_lcm.store import MessageStore, build_message_fts_spec + + db_path = str(tmp_path / "trigger-race.db") + store = MessageStore(db_path) + try: + conn = store.connection + assert conn is not None + assert conn.in_transaction is False + spec = build_message_fts_spec() + trigger_name = db_bootstrap._extract_trigger_name(spec.trigger_sqls[0]) + assert trigger_name is not None + assert db_bootstrap._fts_missing_triggers(conn, spec) is False + + original_deep_check = db_bootstrap._fts_needs_rebuild + trigger_dropped = False + + def deep_check_then_drop_trigger(conn_arg, spec_arg, *, now=None, throttle=False): + nonlocal trigger_dropped + result = original_deep_check( + conn_arg, spec_arg, now=now, throttle=throttle + ) + if not trigger_dropped: + trigger_dropped = True + other = sqlite3.connect(db_path) + try: + other.execute( + f"DROP TRIGGER {db_bootstrap.quote_sql_identifier(trigger_name)}" + ) + other.commit() + finally: + other.close() + return result + + monkeypatch.setattr( + db_bootstrap, "_fts_needs_rebuild", deep_check_then_drop_trigger + ) + trigger_create_transaction_states = [] + + def trace_trigger_ddl(sql): + if sql.lstrip().upper().startswith("CREATE TRIGGER"): + trigger_create_transaction_states.append(conn.in_transaction) + + conn.set_trace_callback(trace_trigger_ddl) + try: + result = db_bootstrap.repair_external_content_fts( + conn, spec, throttle=True + ) + finally: + conn.set_trace_callback(None) + + assert trigger_dropped is True + assert result == { + "rebuilt": False, + "degraded": False, + "triggers_recreated": True, + } + assert trigger_create_transaction_states + assert all(trigger_create_transaction_states) + assert db_bootstrap._fts_missing_triggers(conn, spec) is False + finally: + store.close() + + def _make_future_schema_db(db_path): conn = sqlite3.connect(str(db_path)) try: @@ -325,6 +550,33 @@ def test_is_fts_corruption_error_classification(): assert not db_bootstrap._is_fts_corruption_error("query timeout expired") +def test_sqlite_lock_error_classification_uses_codes_and_lock_messages(): + coded_busy = sqlite3.OperationalError("synthetic non-lock message") + coded_busy.sqlite_errorcode = sqlite3.SQLITE_BUSY | (1 << 8) + assert db_bootstrap._is_sqlite_lock_error(coded_busy) + + assert db_bootstrap._is_sqlite_lock_error( + sqlite3.OperationalError("database is locked") + ) + assert db_bootstrap._is_sqlite_lock_error( + sqlite3.OperationalError("database table is locked: sqlite_master") + ) + + wrapped = RuntimeError("startup failed") + wrapped.__cause__ = sqlite3.OperationalError("database schema is locked") + assert db_bootstrap._is_sqlite_lock_error(wrapped) + + +def test_sqlite_lock_error_classification_rejects_unrelated_timeout_text(): + assert not db_bootstrap._is_sqlite_lock_error( + sqlite3.IntegrityError("constraint timeout while validating data") + ) + assert not db_bootstrap._is_sqlite_lock_error( + sqlite3.OperationalError("busy parsing application expression") + ) + assert not db_bootstrap._is_sqlite_lock_error(TimeoutError("query timeout")) + + def test_integrity_check_lock_error_is_unchecked_not_corruption(tmp_path, monkeypatch): """A transient lock/busy error classifies as 'unchecked', never 'fail' (F3). @@ -355,6 +607,25 @@ def test_integrity_check_malformed_error_is_fail(tmp_path, monkeypatch): conn.close() +def test_fts_repair_lock_budget_propagates_without_destructive_repair(tmp_path, monkeypatch): + """A bounded ownership failure is not reclassified as FTS corruption.""" + conn = _make_conn(tmp_path) + locker = sqlite3.connect(_db_file(tmp_path), timeout=1.0) + try: + locker.execute("BEGIN IMMEDIATE") + monkeypatch.setattr(db_bootstrap, "SQLITE_BUSY_TIMEOUT_MS", 25) + with pytest.raises(sqlite3.OperationalError, match="locked|busy"): + ensure_external_content_fts(conn, _spec()) + + tables = _table_names(_db_file(tmp_path)) + assert "messages_fts" not in tables + assert "messages_fts_docsize" not in tables + finally: + conn.close() + locker.rollback() + locker.close() + + def test_doctor_repair_apply_joins_background_scans_first(tmp_path, monkeypatch): """Explicit repair joins in-flight background scans before repairing (F3).""" from hermes_lcm.config import LCMConfig