diff --git a/src/pinky_memory/store.py b/src/pinky_memory/store.py index 7f329aec..92f024b6 100644 --- a/src/pinky_memory/store.py +++ b/src/pinky_memory/store.py @@ -121,17 +121,24 @@ class InvalidQueryEmbeddingError(ValueError): INSERT INTO reflections_fts(reflections_fts, rowid, id, content, context, project) VALUES ('delete', old.rowid, old.id, old.content, old.context, old.project); END; +""" --- Drop+recreate migrates older DBs where the trigger fired on every UPDATE --- (access tracking churned the FTS index); only indexed columns matter here. -DROP TRIGGER IF EXISTS reflections_au; +# The AFTER UPDATE trigger is migrated separately (#366): older DBs have a +# version that fired on every UPDATE (access tracking churned the FTS index), +# so it needs a drop+recreate rather than a CREATE ... IF NOT EXISTS. That +# rewrite must NOT run on every open — see _migrate_fts_update_trigger. +_FTS5_AU_TRIGGER = """\ CREATE TRIGGER reflections_au AFTER UPDATE OF id, content, context, project ON reflections BEGIN INSERT INTO reflections_fts(reflections_fts, rowid, id, content, context, project) VALUES ('delete', old.rowid, old.id, old.content, old.context, old.project); INSERT INTO reflections_fts(rowid, id, content, context, project) VALUES (new.rowid, new.id, new.content, new.context, new.project); -END; -""" +END""" + + +def _normalize_sql(sql: str) -> str: + """Collapse whitespace so stored DDL compares equal to our literal.""" + return " ".join(sql.split()) def _now_iso() -> str: @@ -209,8 +216,15 @@ def _init_schema(self) -> None: self._conn.executescript(_FTS5_SCHEMA) self._conn.executescript(_FTS5_TRIGGERS) self._conn.commit() + self._migrate_fts_update_trigger() self._fts5_available = True except sqlite3.OperationalError as e: + if "no such module" not in str(e).lower(): + # Any other OperationalError (I/O error, corruption, lock + # timeout) is a real fault. Swallowing it here would pin this + # process to LIKE-based search for its whole lifetime with no + # signal at all — let it surface instead. #366 + raise # FTS5 extension not compiled into this sqlite build. Search # falls back to LIKE — log so admins can diagnose slow queries. #295 logger.warning( @@ -221,6 +235,44 @@ def _init_schema(self) -> None: # sqlite-vec virtual table (separate try — graceful if not available) self._init_vec() + def _current_au_trigger_sql(self) -> str | None: + row = self._conn.execute( + "SELECT sql FROM sqlite_master WHERE type = 'trigger' AND name = 'reflections_au'" + ).fetchone() + return row[0] if row is not None else None + + def _migrate_fts_update_trigger(self) -> None: + """Bring reflections_au up to date, writing only when it is stale (#366). + + The old code dropped and recreated this trigger on *every* open. Since + several components construct a ReflectionStore at daemon start, two of + them could overlap in that window: the loser hit "trigger + reflections_au already exists" and — via an over-broad except — spent + its lifetime with keyword search silently degraded to LIKE. + + The read-only fast path removes the window entirely for an up-to-date + DB. A DB that genuinely needs migrating takes a write lock and + re-checks under it, so concurrent openers serialize and exactly one + performs the rewrite. Note executescript() would COMMIT implicitly, + hence execute() statement by statement here. + """ + wanted = _normalize_sql(_FTS5_AU_TRIGGER) + current = self._current_au_trigger_sql() + if current is not None and _normalize_sql(current) == wanted: + return + + with self._lock: + self._conn.execute("BEGIN IMMEDIATE") + try: + current = self._current_au_trigger_sql() + if current is None or _normalize_sql(current) != wanted: + self._conn.execute("DROP TRIGGER IF EXISTS reflections_au") + self._conn.execute(_FTS5_AU_TRIGGER) + self._conn.commit() + except Exception: + self._conn.rollback() + raise + def _migrate_add_column(self, column: str, definition: str) -> None: """Add a column to the reflections table if it doesn't exist.""" try: diff --git a/tests/test_memory_fts_init_race.py b/tests/test_memory_fts_init_race.py new file mode 100644 index 00000000..1e6b50a9 --- /dev/null +++ b/tests/test_memory_fts_init_race.py @@ -0,0 +1,183 @@ +"""Opening a ReflectionStore concurrently must not silently lose FTS5 (#366). + +The FTS trigger migration used to run ``DROP TRIGGER`` + ``CREATE TRIGGER`` on +*every* open. Two processes overlapping in that window made the loser fail with +``trigger reflections_au already exists``, and the over-broad ``except +OperationalError`` read that as "FTS5 is not compiled in" — leaving that process +with keyword search degraded to LIKE for its whole lifetime, silently. +""" +from __future__ import annotations + +import sqlite3 +import threading +from pathlib import Path + +from pinky_memory.store import ReflectionStore +from pinky_memory.types import Reflection, ReflectionType + + +def _seed(db: Path) -> None: + store = ReflectionStore(str(db)) + store.insert(Reflection(type=ReflectionType.fact, content="il daemon riavvia il gateway")) + store.close() + + +def _patch_executescript(monkeypatch, fake): + """Swap executescript on the store's connection. + + sqlite3.Connection is immutable, so the only seam is the connection + factory: hand sqlite3.connect a subclass that overrides the method. + """ + real_connect = sqlite3.connect + + class Patched(sqlite3.Connection): + def executescript(self, script): + return fake(self, script, super().executescript) + + def traced(*args, **kwargs): + kwargs["factory"] = Patched + return real_connect(*args, **kwargs) + + monkeypatch.setattr(sqlite3, "connect", traced) + + +def _make_trigger_legacy(db: Path) -> None: + """Restore the pre-migration trigger so every opener must migrate.""" + conn = sqlite3.connect(str(db)) + conn.executescript( + "DROP TRIGGER IF EXISTS reflections_au;\n" + "CREATE TRIGGER reflections_au AFTER UPDATE ON reflections BEGIN\n" + " INSERT INTO reflections_fts(reflections_fts, rowid, id, content, context, project)\n" + " VALUES ('delete', old.rowid, old.id, old.content, old.context, old.project);\n" + " INSERT INTO reflections_fts(rowid, id, content, context, project)\n" + " VALUES (new.rowid, new.id, new.content, new.context, new.project);\n" + "END;" + ) + conn.commit() + conn.close() + + +def _trace_opens(monkeypatch) -> list[str]: + """Capture every SQL statement the next ReflectionStore(s) execute.""" + seen: list[str] = [] + real_connect = sqlite3.connect + + def traced(*args, **kwargs): + conn = real_connect(*args, **kwargs) + conn.set_trace_callback(seen.append) + return conn + + monkeypatch.setattr(sqlite3, "connect", traced) + return seen + + +class TestFtsInitIsIdempotent: + def test_reopening_an_up_to_date_db_runs_no_trigger_ddl(self, tmp_path, monkeypatch): + """No DDL on reopen means no window for a concurrent opener to lose.""" + db = tmp_path / "memory.db" + _seed(db) + + seen = _trace_opens(monkeypatch) + store = ReflectionStore(str(db)) + store.close() + + # A read of sqlite_master is fine — it takes no lock and opens no + # window. Only writing the trigger does. + ddl = [ + s + for s in seen + if s.strip().upper().startswith(("DROP TRIGGER", "CREATE TRIGGER")) + and "reflections_au" in s + ] + assert ddl == [], f"reopen still rewrites the FTS trigger: {ddl}" + + def test_concurrent_first_opens_migrate_exactly_once(self, tmp_path, monkeypatch): + """Racing openers serialize: one migrates, the rest see it already done. + + Asserting "exactly one rewrite" is what makes this deterministic. The + migration re-checks sqlite_master under the write lock, so however the + 8 threads interleave, only the first one to hold the lock can still + find the trigger stale. + """ + db = tmp_path / "memory.db" + _seed(db) + _make_trigger_legacy(db) + + seen = _trace_opens(monkeypatch) + barrier = threading.Barrier(8) + results: list[bool] = [] + lock = threading.Lock() + + def open_once(): + barrier.wait() + store = ReflectionStore(str(db)) + with lock: + results.append(store._fts5_available) + store.close() + + threads = [threading.Thread(target=open_once) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert results == [True] * 8, f"some opens lost FTS5: {results}" + rewrites = [s for s in seen if s.strip().upper().startswith("DROP TRIGGER")] + assert len(rewrites) == 1, f"expected a single migration, got {len(rewrites)}" + + def test_reopen_keeps_bm25_keyword_search_working(self, tmp_path): + """The user-visible half: FTS still ranks, it has not fallen back to LIKE.""" + db = tmp_path / "memory.db" + _seed(db) + + store = ReflectionStore(str(db)) + try: + assert store._fts5_available is True + hits = store.search_by_keyword_scored("gateway", limit=5) + assert [r.content for _, r in hits] == ["il daemon riavvia il gateway"] + finally: + store.close() + + +class TestFtsInitErrorHandling: + def test_unexpected_operational_error_is_not_mistaken_for_missing_fts5( + self, tmp_path, monkeypatch + ): + """Only "no such module: fts5" means FTS5 is unavailable. + + Anything else is a real fault and must surface instead of silently + switching every later recall() to LIKE. + """ + db = tmp_path / "memory.db" + + def boom(conn, script, real): + if "reflections_fts" in script: + raise sqlite3.OperationalError("disk I/O error") + return real(script) + + _patch_executescript(monkeypatch, boom) + + try: + store = ReflectionStore(str(db)) + except sqlite3.OperationalError as exc: + assert "disk I/O error" in str(exc) + else: + store.close() + raise AssertionError("a disk I/O error was swallowed as 'FTS5 unavailable'") + + def test_missing_fts5_module_still_degrades_gracefully(self, tmp_path, monkeypatch): + """The #295 behaviour we must keep: a build without FTS5 falls back to LIKE.""" + db = tmp_path / "memory.db" + + def no_fts5(conn, script, real): + if "reflections_fts" in script: + raise sqlite3.OperationalError("no such module: fts5") + return real(script) + + _patch_executescript(monkeypatch, no_fts5) + + store = ReflectionStore(str(db)) + try: + assert store._fts5_available is False + finally: + store.close()