diff --git a/CHANGELOG/v4.md b/CHANGELOG/v4.md index 2ba59edc2..2f50fbffd 100644 --- a/CHANGELOG/v4.md +++ b/CHANGELOG/v4.md @@ -47,6 +47,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **Opening a store while another process opened the same store could raise `database schema has changed` ([#1310](https://github.com/robotrocketscience/aelfrice/issues/1310)).** SQLite raises `SQLITE_SCHEMA` when the schema cookie moves between a statement's prepare and its step, which is exactly what two processes running the open-time `CREATE TABLE IF NOT EXISTS` battery against one `.git/aelfrice/memory.db` do to each other. `busy_timeout=5000` does not cover it — that pragma retries `database is locked`, a different error — so the constructor failed outright and took a required CI check down at random. The whole open-time window (the stale-`ingest_log` probe, `_SCHEMA`, `_MIGRATIONS`, the post-migration indexes, the `schema_meta` seed reads and writes, and the local scope-id resolve) now runs under one bounded re-prepare retry rather than only the DDL loop the traceback named: the reads are exposed to the same race, and a per-call-site patch would leave gaps as the constructor grows. The window is idempotent by construction (`IF NOT EXISTS` / `OR IGNORE` / marker-gated), so re-running it is a no-op. Only `schema has changed` is retried — every other `OperationalError` still propagates, and a persistently changing schema fails after a bounded number of attempts instead of spinning. Contrary to the issue's reasoning, the `_MIGRATIONS` catch three lines below did **not** already tolerate this class: it admits only `duplicate column name` and `no such column`, both of which are re-raised for anything else. - **`[relationship_detector]` was half-honoured at ingest: the flag took effect, the thresholds did not ([#1299](https://github.com/robotrocketscience/aelfrice/issues/1299)).** `ingest.py` resolved `auto_detect` from `.aelfrice.toml` and then called `write_semantic_edges(store, new_belief_ids=...)` with no threshold arguments, so `jaccard_min` / `confidence_min` / `max_candidate_pairs` reached the read-only `aelf doctor --relationships` audits and were silently ignored on the one path that actually mutates the graph. Adjacent keys in the same section, opposite reach, no trace on stderr — and the asymmetry ran the risky direction, with the audit tunable and the writer pinned at the module defaults. Ingest now threads the resolved config through. The three keys are parsed from the **same** `.aelfrice.toml` read that resolves `auto_detect` (new `resolve_ingest_relationship_config`), so the per-turn config-probe count is unchanged — measured 11 probes with no config file and 4 with one four directories up, before and after — rather than adding a second filesystem walk to a hot path ([#1289](https://github.com/robotrocketscience/aelfrice/issues/1289)/[#1298](https://github.com/robotrocketscience/aelfrice/issues/1298)). That measurement is env-unset; the env var is the other half of the precedence and it moves the count, so it is stated separately. `AELFRICE_AUTO_RELATIONSHIPS=0` decides the question without needing any threshold, and the flag-only resolver short-circuited on env before touching the filesystem — so that install paid **0** probes per turn and still does, the resolver returning before the walk rather than reading a config whose only consumer will not run. With the var set truthy the walk is real and new (0 -> 11 on a deep tree): the thresholds are then actually used, which is the point of the fix, so that one is a cost and not a regression. Precedence for `auto_detect` is unchanged (env > TOML > default-off), and default-off means a fresh install is byte-identical. `residual_overlap_min` and `max_edges_per_belief` still have no TOML key at all; `docs/user/CONFIG.md` now documents the section with per-key reach so which is which is readable. - **PHILOSOPHY and the write-log memo claimed `edges` are a materialized projection of the log; they never have been ([#1283](https://github.com/robotrocketscience/aelfrice/issues/1283)).** All six `derive()` return paths emit `edges=[]`, so `ingest_log.derived_edge_ids` is NULL on **every** row (0 of 139,592 on the development store), and every real edge is written outside the log by `ingest.py`, `temporal_spine.py` and the relationship / contradiction detectors. The substrate claim therefore overreached for the graph: a replay from empty cannot reconstruct the L3 BFS graph, the temporal spine or the `CONTRADICTS` substrate, and `replay.py` reports edge divergence only in the bucket documented as *"never promoted into `has_drift`"*. Both documents now state the ratified contract (edges are **log-derived**, recompute keyed on `(created_at, ingest_log ULID)`, operator ruling 2026-08-01) separately from the shipped state, and name what is still missing. The key is the log's ULID rather than anything read off the belief table because the writer actually orders by `(created_at, rowid)` and `rowid` is implicit here — VACUUM may renumber it; measured, the ULID key reproduces **93.7%** of the live `TEMPORAL_NEXT` set against **7.4%** for a belief-table key (`benchmarks/spine_order_provenance.py`). Docs only — no code, no defaults, and the recompute itself is not built. - **LIMITATIONS claimed a residual exposure-as-evidence path that has not existed since #1162 ([#1267](https://github.com/robotrocketscience/aelfrice/issues/1267)).** The sharp-edges entry said retrieval "still enqueues each surfaced belief" and that `aelf sweep-feedback` "applies a small alpha bump (default +0.05)", citing [#1091](https://github.com/robotrocketscience/aelfrice/issues/1091) as having only *flagged* the sweep for audit-only treatment. All three were stale: the sweeper has written nothing since [#1162](https://github.com/robotrocketscience/aelfrice/issues/1162) (it classifies what it *would* have applied and returns `mutated=False`), and the enqueue inside `retrieve()` is gated on `AELFRICE_IMPLICIT_FEEDBACK_ENQUEUE`, default off. A reader auditing where their posteriors come from was pointed at a mutation path that no longer fires. New `benchmarks/posterior_channel_audit.py` drives all three `apply_feedback` routes against a fresh store and fails non-zero if any default moves, so the entry cannot go stale silently again. diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py index 71d6231b0..0d7ed56b5 100644 --- a/src/aelfrice/store.py +++ b/src/aelfrice/store.py @@ -20,7 +20,15 @@ import sqlite3 from contextlib import contextmanager from datetime import datetime, timezone -from typing import TYPE_CHECKING, Callable, Final, Iterable, Iterator, Sequence +from typing import ( + TYPE_CHECKING, + Callable, + Final, + Iterable, + Iterator, + Sequence, + TypeVar, +) if TYPE_CHECKING: # #1126 belief categories. Imported lazily at runtime inside the @@ -1125,6 +1133,61 @@ def _drop_stale_ingest_log(conn: sqlite3.Connection) -> None: conn.execute("DROP TABLE ingest_log") +_SCHEMA_CHANGED_MSG: Final[str] = "schema has changed" +_SCHEMA_RETRY_ATTEMPTS: Final[int] = 3 + +_RetryT = TypeVar("_RetryT") + + +def _retry_on_schema_change( + op: Callable[[], _RetryT], *, attempts: int = _SCHEMA_RETRY_ATTEMPTS +) -> _RetryT: + """Run `op`, retrying only `SQLITE_SCHEMA` ("schema has changed"). + + #1310. SQLite raises `OperationalError: database schema has changed` + when the schema cookie moves between a statement's prepare and its + step — i.e. another connection committed DDL in between. Two + processes opening the same store both run the open-time + `CREATE TABLE IF NOT EXISTS` battery, so this is reachable on every + multi-worktree open. `busy_timeout` does not cover it: that pragma + retries `database is locked`, a different error. + + Re-preparing against the new cookie is the entire fix, so `op` must + be idempotent. Every other `OperationalError` propagates unchanged — + a malformed statement must stay loud rather than be retried into + silence. `attempts` is bounded so a schema that keeps changing + fails instead of spinning forever. + """ + for i in range(attempts): + try: + return op() + except sqlite3.OperationalError as e: + if _SCHEMA_CHANGED_MSG not in str(e) or i == attempts - 1: + raise + # Unreachable: the loop either returns or raises on the last pass. + raise AssertionError("attempts must be >= 1") + + +def _execute_reprepare( + conn: sqlite3.Connection, + stmt: str, + *, + attempts: int = _SCHEMA_RETRY_ATTEMPTS, +) -> sqlite3.Cursor: + """Execute one parameterless statement, re-preparing on SQLITE_SCHEMA. + + #1310. Statement-level sibling of `_retry_on_schema_change`, used by + the open-time DDL loops so the common case re-runs one `CREATE TABLE + IF NOT EXISTS` rather than restarting the whole battery. The battery + is wrapped as well — reads and parameterised writes are exposed to + the same race, and a per-call-site patch would leave gaps as the + constructor grows. + """ + return _retry_on_schema_change( + lambda: conn.execute(stmt), attempts=attempts + ) + + def _ingest_row_to_dict(row: sqlite3.Row) -> dict[str, object]: """Decode an `ingest_log` sqlite row into a Python dict. @@ -1223,68 +1286,18 @@ def __init__(self, path: str, *, project_context_default: str = "") -> None: # aelfrice/memory.db. Per the v1.1.0 #89 concurrency tests. self._conn.execute("PRAGMA busy_timeout=5000") self._conn.execute("PRAGMA foreign_keys=ON") - _drop_stale_ingest_log(self._conn) - for stmt in _SCHEMA: - self._conn.execute(stmt) - for stmt in _MIGRATIONS: - try: - self._conn.execute(stmt) - except sqlite3.OperationalError as e: - # Idempotency catches: - # - "duplicate column name: X" — ADD COLUMN already - # present (fresh v1.2 DB or prior migration pass). - # - "no such column: X" — DROP COLUMN already done - # (fresh DB that never had the column, or prior - # migration pass). - msg = str(e) - if ( - "duplicate column name" not in msg - and "no such column" not in msg - ): - raise - for stmt in _POST_MIGRATION_INDEXES: - self._conn.execute(stmt) - # #1135: one-shot. Ran unguarded on every open pre-v4.2; the - # marker matches the other schema_meta-gated passes. Rides the - # single open commit below. - # #1135: seed the durable mutation counter. Read-first: an - # unconditional INSERT OR IGNORE would take the write lock on - # every open even when the row exists, blocking behind any - # concurrent writer's open transaction. Rides the single open - # commit below. Two connections racing the first-ever seed both - # pass the SELECT; OR IGNORE makes the second insert a no-op. - seeded = self._conn.execute( - "SELECT 1 FROM schema_meta WHERE key = ?", - (SCHEMA_META_STORE_GENERATION,), - ).fetchone() - if seeded is None: - self._conn.execute( - "INSERT OR IGNORE INTO schema_meta (key, value) " - "VALUES (?, '0')", - (SCHEMA_META_STORE_GENERATION,), - ) - marker = self._conn.execute( - "SELECT value FROM schema_meta WHERE key = ?", - (SCHEMA_META_ORIGIN_BACKFILL,), - ).fetchone() - if marker is None: - for stmt in _BACKFILL_STATEMENTS: - self._conn.execute(stmt) - self._conn.execute( - "INSERT OR REPLACE INTO schema_meta (key, value) " - "VALUES (?, ?)", - ( - SCHEMA_META_ORIGIN_BACKFILL, - datetime.now(timezone.utc).isoformat(), - ), - ) - self._commit() self._invalidation_callbacks: list[Callable[[], None]] = [] - # v1.5.0 #204 federation forward-compat. Resolve (or - # generate) the local scope id BEFORE any belief/edge - # write path runs — write hooks consume `_local_scope_id` - # to bump the version-vector counter. - self._local_scope_id: str = self._resolve_local_scope_id() + # #1310: the whole open-time schema window runs under one + # schema-cookie retry, not just the DDL loops. Every statement + # in it — the `_drop_stale_ingest_log` reads, the schema_meta + # queries, the backfill writes, and `_resolve_local_scope_id` — + # is exposed to a concurrent opener committing DDL between + # prepare and step. The window is idempotent by construction + # (IF NOT EXISTS / OR IGNORE / marker-gated), so re-running it + # is safe. See `_retry_on_schema_change`. + self._local_scope_id: str = _retry_on_schema_change( + self._apply_open_schema + ) # #1161: every one-shot below runs through `_run_guarded_migration` # so a raising pass cannot make the store unopenable. Ordering is # unchanged and the guard does not alter the success path — see @@ -1356,6 +1369,80 @@ def __init__(self, path: str, *, project_context_default: str = "") -> None: self._peer_handles: dict[str, sqlite3.Connection] = {} self._peer_deps_loaded: bool = False + def _apply_open_schema(self) -> str: + """Run the open-time DDL + seed + backfill window; return scope id. + + Extracted from `__init__` (#1310) so the whole window can be + re-run as a unit when a concurrent opener moves the schema + cookie — see `_retry_on_schema_change`, which is the only + intended caller. Idempotent: every statement here is + `IF NOT EXISTS`, `OR IGNORE`/`OR REPLACE`, or gated on a + `schema_meta` marker, so a second pass is a no-op. + """ + _drop_stale_ingest_log(self._conn) + for stmt in _SCHEMA: + _execute_reprepare(self._conn, stmt) + for stmt in _MIGRATIONS: + try: + _execute_reprepare(self._conn, stmt) + except sqlite3.OperationalError as e: + # Idempotency catches: + # - "duplicate column name: X" — ADD COLUMN already + # present (fresh v1.2 DB or prior migration pass). + # - "no such column: X" — DROP COLUMN already done + # (fresh DB that never had the column, or prior + # migration pass). + # It does NOT catch "database schema has changed" — that + # class is handled by the retry above/around, not here. + msg = str(e) + if ( + "duplicate column name" not in msg + and "no such column" not in msg + ): + raise + for stmt in _POST_MIGRATION_INDEXES: + _execute_reprepare(self._conn, stmt) + # #1135: one-shot. Ran unguarded on every open pre-v4.2; the + # marker matches the other schema_meta-gated passes. Rides the + # single open commit below. + # #1135: seed the durable mutation counter. Read-first: an + # unconditional INSERT OR IGNORE would take the write lock on + # every open even when the row exists, blocking behind any + # concurrent writer's open transaction. Rides the single open + # commit below. Two connections racing the first-ever seed both + # pass the SELECT; OR IGNORE makes the second insert a no-op. + seeded = self._conn.execute( + "SELECT 1 FROM schema_meta WHERE key = ?", + (SCHEMA_META_STORE_GENERATION,), + ).fetchone() + if seeded is None: + self._conn.execute( + "INSERT OR IGNORE INTO schema_meta (key, value) " + "VALUES (?, '0')", + (SCHEMA_META_STORE_GENERATION,), + ) + marker = self._conn.execute( + "SELECT value FROM schema_meta WHERE key = ?", + (SCHEMA_META_ORIGIN_BACKFILL,), + ).fetchone() + if marker is None: + for stmt in _BACKFILL_STATEMENTS: + self._conn.execute(stmt) + self._conn.execute( + "INSERT OR REPLACE INTO schema_meta (key, value) " + "VALUES (?, ?)", + ( + SCHEMA_META_ORIGIN_BACKFILL, + datetime.now(timezone.utc).isoformat(), + ), + ) + self._commit() + # v1.5.0 #204 federation forward-compat. Resolve (or generate) + # the local scope id BEFORE any belief/edge write path runs — + # write hooks consume `_local_scope_id` to bump the + # version-vector counter. + return self._resolve_local_scope_id() + def close(self) -> None: for conn in self._peer_handles.values(): try: diff --git a/tests/test_store_schema_race.py b/tests/test_store_schema_race.py new file mode 100644 index 000000000..2e66ca7b3 --- /dev/null +++ b/tests/test_store_schema_race.py @@ -0,0 +1,259 @@ +"""SQLITE_SCHEMA ("database schema has changed") retry on store open (#1310). + +`MemoryStore.__init__` runs a DDL + seed + backfill battery. When a +second process commits DDL between one of those statements' prepare and +step, SQLite raises `OperationalError: database schema has changed`. +`busy_timeout` does not cover it — that pragma retries `database is +locked`, a different error — so the open failed outright and took a +required CI check down with it at random. + +What these tests prove, and what they do not: + +- They prove the retry is **wired**: an injected SQLITE_SCHEMA error at + a DDL statement, at a bare `schema_meta` read, and at the scope-id + write all leave `MemoryStore(...)` constructing successfully, and a + non-schema `OperationalError` still propagates. +- They do **not** prove the CI flake is gone. Real-concurrency + reproduction was attempted and did not fire (0/180 rounds locally), so + a race-based guard would assert nothing. Fault injection is the only + arm that distinguishes the fixed code from the broken code + deterministically. + +Injection works by wrapping the real connection. Two mechanics matter: + +- `__getattr__` is not consulted for dunder lookups on new-style + classes, so the context-manager protocol (`with self._conn:`, used by + a one-shot migration) is forwarded explicitly. +- `aelfrice.store.sqlite3` *is* the stdlib `sqlite3` module, so patching + `connect` through it is process-global. Every patch here is scoped to + a single `MemoryStore(...)` call and restored in a `finally`. +""" +from __future__ import annotations + +import sqlite3 +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from pathlib import Path + +import pytest + +from aelfrice import store as store_mod +from aelfrice.store import ( + MemoryStore, + _execute_reprepare, + _retry_on_schema_change, +) + +_SCHEMA_ERR = "database schema has changed" + + +class _FaultingConnection: + """Wrap a real connection; raise `error` when `match` hits. + + Fires once by default (or on every matching call when `once=False`), + so a retry that re-prepares the same statement sees a success on its + second attempt. `fired` is the arm that keeps a test from passing + vacuously because the injection never triggered. + """ + + def __init__( + self, + conn: sqlite3.Connection, + match: Callable[[str], bool], + error: Exception, + *, + once: bool = True, + ) -> None: + # Bypass __setattr__, which forwards to the wrapped connection. + object.__setattr__( + self, + "_state", + { + "conn": conn, + "match": match, + "error": error, + "once": once, + "fired": 0, + }, + ) + + @property + def fired(self) -> int: + return int(self._state["fired"]) + + def execute(self, sql: str, *args: object, **kwargs: object) -> object: + state = self._state + already = int(state["fired"]) + if state["match"](sql) and not (state["once"] and already): + state["fired"] = already + 1 + raise state["error"] + return state["conn"].execute(sql, *args, **kwargs) + + # Dunders are looked up on the type, so __getattr__ below never sees + # them. `with self._conn:` in a one-shot migration needs these. + def __enter__(self) -> object: + return self._state["conn"].__enter__() + + def __exit__(self, *exc: object) -> object: + return self._state["conn"].__exit__(*exc) + + def __getattr__(self, name: str) -> object: + return getattr(self._state["conn"], name) + + def __setattr__(self, name: str, value: object) -> None: + setattr(self._state["conn"], name, value) + + +@contextmanager +def _faulting_connect( + match: Callable[[str], bool], + error: Exception, + *, + once: bool = True, +) -> Iterator[list[_FaultingConnection]]: + """Patch `sqlite3.connect` to hand back a faulting wrapper. + + PROCESS-GLOBAL: `aelfrice.store.sqlite3` is the stdlib module, not a + module-local alias. Keep the body to the one `MemoryStore(...)` call + under test; the original is restored unconditionally. + """ + real = sqlite3.connect + made: list[_FaultingConnection] = [] + + def fake(*args: object, **kwargs: object) -> _FaultingConnection: + wrapper = _FaultingConnection( + real(*args, **kwargs), match, error, once=once + ) + made.append(wrapper) + return wrapper + + store_mod.sqlite3.connect = fake # type: ignore[assignment] + try: + yield made + finally: + store_mod.sqlite3.connect = real # type: ignore[assignment] + + +def _startswith(prefix: str) -> Callable[[str], bool]: + return lambda sql: sql.strip().upper().startswith(prefix.upper()) + + +# --- the helpers themselves ------------------------------------------- + + +def test_retry_on_schema_change_retries_then_returns() -> None: + calls: list[int] = [] + + def op() -> str: + calls.append(1) + if len(calls) == 1: + raise sqlite3.OperationalError(_SCHEMA_ERR) + return "ok" + + assert _retry_on_schema_change(op) == "ok" + assert len(calls) == 2 + + +def test_retry_on_schema_change_reraises_other_operational_errors() -> None: + """A malformed statement must stay loud, not be retried into silence.""" + calls: list[int] = [] + + def op() -> None: + calls.append(1) + raise sqlite3.OperationalError('near "CRATE": syntax error') + + with pytest.raises(sqlite3.OperationalError, match="CRATE"): + _retry_on_schema_change(op) + assert len(calls) == 1, "non-schema errors must not be retried" + + +def test_retry_on_schema_change_is_bounded() -> None: + """A schema that keeps changing fails rather than spinning forever.""" + calls: list[int] = [] + + def op() -> None: + calls.append(1) + raise sqlite3.OperationalError(_SCHEMA_ERR) + + with pytest.raises(sqlite3.OperationalError, match="schema has changed"): + _retry_on_schema_change(op, attempts=3) + assert len(calls) == 3 + + +def test_execute_reprepare_reruns_the_statement(tmp_path: Path) -> None: + conn = sqlite3.connect(str(tmp_path / "x.db")) + wrapper = _FaultingConnection( + conn, + _startswith("CREATE TABLE"), + sqlite3.OperationalError(_SCHEMA_ERR), + ) + _execute_reprepare( # type: ignore[arg-type] + wrapper, "CREATE TABLE t (a INTEGER)" + ) + assert wrapper.fired == 1 + rows = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ).fetchall() + assert [r[0] for r in rows] == ["t"] + conn.close() + + +# --- the wiring, through MemoryStore.__init__ -------------------------- + + +@pytest.mark.parametrize( + ("label", "match"), + [ + # The _SCHEMA loop — the site the traceback in #1310 named. + ("schema_ddl", _startswith("CREATE TABLE")), + # A bare read the issue's narrower patch would have left exposed. + ("generation_read", _startswith("SELECT 1 FROM schema_meta")), + # _resolve_local_scope_id's write, the last statement in the window. + ("scope_id_write", _startswith("INSERT OR REPLACE INTO schema_meta")), + ], +) +def test_open_survives_injected_schema_change( + tmp_path: Path, label: str, match: Callable[[str], bool] +) -> None: + db = str(tmp_path / f"{label}.db") + err = sqlite3.OperationalError(_SCHEMA_ERR) + with _faulting_connect(match, err) as made: + store = MemoryStore(db) + assert made, "connect was never patched" + assert made[0].fired == 1, "injection never fired — test is vacuous" + assert store.local_scope_id + store.close() + + # The store is usable and the battery left the canonical schema. + plain = MemoryStore(db) + assert plain.get_schema_meta("local_scope_id") + plain.close() + + +def test_open_still_raises_on_a_non_schema_error(tmp_path: Path) -> None: + """AC2: a genuinely broken statement is not retried into silence.""" + db = str(tmp_path / "bad.db") + with _faulting_connect( + _startswith("CREATE TABLE"), + sqlite3.OperationalError('near "CRATE": syntax error'), + ): + with pytest.raises(sqlite3.OperationalError, match="CRATE"): + MemoryStore(db) + + +def test_open_gives_up_on_a_persistently_changing_schema( + tmp_path: Path, +) -> None: + """Bounded attempts: a permanent SQLITE_SCHEMA fails, it does not hang.""" + db = str(tmp_path / "spin.db") + with _faulting_connect( + _startswith("CREATE TABLE"), + sqlite3.OperationalError(_SCHEMA_ERR), + once=False, + ) as made: + with pytest.raises( + sqlite3.OperationalError, match="schema has changed" + ): + MemoryStore(db) + # 3 outer window attempts x 3 statement-level attempts, not unbounded. + assert made[0].fired <= 9