diff --git a/arborist/store.py b/arborist/store.py index ddca716..a2ffe19 100644 --- a/arborist/store.py +++ b/arborist/store.py @@ -554,9 +554,17 @@ def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection: - temp_store=MEMORY keeps temp tables in RAM (no /tmp churn). - mmap_size=256 MB lets reads come from page-cache without read() syscalls. - Migration probes (executescript(SCHEMA_SQL) + 7 forward migrations) + Migration probes (executescript(SCHEMA_SQL) + the forward migrations) run once per (physical file, process). Subsequent ``connect()`` calls on the same shard skip migration entirely — see #000026 Phase 1. + + During that one migration pass we raise ``busy_timeout`` so a second + ``connect()`` racing the same fresh shard *waits* for the first to + finish its DDL rather than failing fast with ``database is locked``. + Combined with the ``IF NOT EXISTS`` / ``_add_column_if_missing`` + idempotency, the loser then re-runs its (now no-op) probes cleanly. + The timeout is left in place on that connection; later connections + that skip the migration block keep SQLite's fail-fast default. """ p = Path(db_path) p.parent.mkdir(parents=True, exist_ok=True) @@ -565,6 +573,9 @@ def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection: cache_key = str(p.resolve()) if cache_key not in _MIGRATED_SHARDS: + # Wait, don't fail, if a peer process is mid-migration on this + # shard (the _rebuild_providence_cache_* swaps hold a write txn). + conn.execute("PRAGMA busy_timeout = 5000") conn.executescript(SCHEMA_SQL) _migrate_audit_mode(conn) _migrate_mesh_peer_chains(conn) @@ -590,52 +601,79 @@ def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection: return conn +def _add_column_if_missing( + conn: sqlite3.Connection, existing: set[str], column: str, ddl: str +) -> None: + """Run an ``ALTER TABLE … ADD COLUMN`` idempotently. + + SQLite has no ``ADD COLUMN IF NOT EXISTS``. The PRAGMA-derived + ``existing`` column set is the common-case fast-path skip; the + ``duplicate column name`` catch swallows the TOCTOU loser when two + ``connect()`` calls race a fresh shard (both probe, both ADD — the + second would otherwise raise). Sibling DDL applied by the same + codebase, so the column definition the winner installed is identical + to ours. + """ + if column in existing: + return + try: + conn.execute(ddl) + except sqlite3.OperationalError as exc: + if "duplicate column name" not in str(exc).lower(): + raise + + def _migrate_audit_mode(conn: sqlite3.Connection) -> None: """Forward-migrate pre-v9.8-audit-mode providence_cache shards. Adds audit_mode + n_quotes + n_verified + unverified_quotes + verifier_method columns to DBs that pre-date the faithfulness-classification rollout. SQLite ALTER TABLE ADD COLUMN is O(1) (metadata-only) so this is cheap on every - open. Idempotent — checks PRAGMA before each ADD. + open. Idempotent under repeat-open (PRAGMA fast-path) and under concurrent + ``connect()`` (the duplicate-column catch in ``_add_column_if_missing``). Also handles the VISUAL → UNGROUNDED rename for the audit_mode value space. SQLite cannot ALTER a column's CHECK in place, so legacy tables with the old `CHECK (audit_mode IN ('STRICT','HYBRID','VISUAL'))` get - rebuilt via the standard temp-table dance, with values translated. + rebuilt via the standard temp-table dance, with values translated. Each + ``_rebuild_providence_cache_*`` self-wraps in ``BEGIN IMMEDIATE`` so a + failure rolls back clean; under a concurrent ``connect()`` the loser + waits on that write txn (``busy_timeout`` is raised for the migration + pass — see ``connect()``) and then re-runs its probes harmlessly. """ cols = {row["name"] for row in conn.execute("PRAGMA table_info(providence_cache)")} - if "audit_mode" not in cols: - conn.execute( - "ALTER TABLE providence_cache ADD COLUMN audit_mode TEXT " - "NOT NULL DEFAULT 'UNGROUNDED' " - "CHECK (audit_mode IN ('STRICT','HYBRID','UNGROUNDED'))" - ) - if "n_quotes" not in cols: - conn.execute( - "ALTER TABLE providence_cache ADD COLUMN n_quotes INTEGER NOT NULL DEFAULT 0" - ) - if "n_verified" not in cols: - conn.execute( - "ALTER TABLE providence_cache ADD COLUMN n_verified INTEGER NOT NULL DEFAULT 0" - ) - if "unverified_quotes" not in cols: - conn.execute( - "ALTER TABLE providence_cache ADD COLUMN unverified_quotes TEXT" - ) - if "verifier_method" not in cols: - conn.execute( - "ALTER TABLE providence_cache ADD COLUMN verifier_method TEXT " - "NOT NULL DEFAULT 'none' " - "CHECK (verifier_method IN ('quote','span','entity','paraphrase','none'))" - ) - if "run_dag_root" not in cols: - conn.execute( - "ALTER TABLE providence_cache ADD COLUMN run_dag_root TEXT" - ) - if "run_dag_blob" not in cols: - conn.execute( - "ALTER TABLE providence_cache ADD COLUMN run_dag_blob TEXT" - ) + _add_column_if_missing( + conn, cols, "audit_mode", + "ALTER TABLE providence_cache ADD COLUMN audit_mode TEXT " + "NOT NULL DEFAULT 'UNGROUNDED' " + "CHECK (audit_mode IN ('STRICT','HYBRID','UNGROUNDED'))", + ) + _add_column_if_missing( + conn, cols, "n_quotes", + "ALTER TABLE providence_cache ADD COLUMN n_quotes INTEGER NOT NULL DEFAULT 0", + ) + _add_column_if_missing( + conn, cols, "n_verified", + "ALTER TABLE providence_cache ADD COLUMN n_verified INTEGER NOT NULL DEFAULT 0", + ) + _add_column_if_missing( + conn, cols, "unverified_quotes", + "ALTER TABLE providence_cache ADD COLUMN unverified_quotes TEXT", + ) + _add_column_if_missing( + conn, cols, "verifier_method", + "ALTER TABLE providence_cache ADD COLUMN verifier_method TEXT " + "NOT NULL DEFAULT 'none' " + "CHECK (verifier_method IN ('quote','span','entity','paraphrase','none'))", + ) + _add_column_if_missing( + conn, cols, "run_dag_root", + "ALTER TABLE providence_cache ADD COLUMN run_dag_root TEXT", + ) + _add_column_if_missing( + conn, cols, "run_dag_blob", + "ALTER TABLE providence_cache ADD COLUMN run_dag_blob TEXT", + ) # VISUAL → UNGROUNDED rename. Detect legacy CHECK by inspecting DDL. ddl_row = conn.execute( @@ -692,7 +730,7 @@ def _migrate_document_http_meta(conn: sqlite3.Connection) -> None: ).fetchone() if row is None: conn.execute( - "CREATE TABLE document_http_meta (" + "CREATE TABLE IF NOT EXISTS document_http_meta (" " document_root TEXT PRIMARY KEY," " etag TEXT," " last_modified TEXT," @@ -721,7 +759,7 @@ def _migrate_selfmodel_tables(conn: sqlite3.Connection) -> None: ).fetchone() if row is None: conn.execute( - "CREATE TABLE selfmodel_records (" + "CREATE TABLE IF NOT EXISTS selfmodel_records (" " selfmodel_root TEXT PRIMARY KEY," " schema_version TEXT NOT NULL," " parent_selfmodel_root TEXT," @@ -743,10 +781,10 @@ def _migrate_selfmodel_tables(conn: sqlite3.Connection) -> None: ")" ) conn.execute( - "CREATE INDEX idx_selfmodel_state ON selfmodel_records(state)" + "CREATE INDEX IF NOT EXISTS idx_selfmodel_state ON selfmodel_records(state)" ) conn.execute( - "CREATE INDEX idx_selfmodel_parent " + "CREATE INDEX IF NOT EXISTS idx_selfmodel_parent " "ON selfmodel_records(parent_selfmodel_root)" ) row = conn.execute( @@ -755,7 +793,7 @@ def _migrate_selfmodel_tables(conn: sqlite3.Connection) -> None: ).fetchone() if row is None: conn.execute( - "CREATE TABLE selfmodel_capability_claims (" + "CREATE TABLE IF NOT EXISTS selfmodel_capability_claims (" " claim_hash TEXT PRIMARY KEY," " selfmodel_root TEXT NOT NULL," " metric TEXT NOT NULL," @@ -769,11 +807,11 @@ def _migrate_selfmodel_tables(conn: sqlite3.Connection) -> None: ")" ) conn.execute( - "CREATE INDEX idx_claim_metric " + "CREATE INDEX IF NOT EXISTS idx_claim_metric " "ON selfmodel_capability_claims(metric)" ) conn.execute( - "CREATE INDEX idx_claim_selfmodel " + "CREATE INDEX IF NOT EXISTS idx_claim_selfmodel " "ON selfmodel_capability_claims(selfmodel_root)" ) @@ -793,7 +831,7 @@ def _migrate_capital_ledger(conn: sqlite3.Connection) -> None: ).fetchone() if row is None: conn.execute( - "CREATE TABLE capital_ledger (" + "CREATE TABLE IF NOT EXISTS capital_ledger (" " ledger_id INTEGER PRIMARY KEY AUTOINCREMENT," " audit_event_hash TEXT NOT NULL," " op_type TEXT NOT NULL," @@ -811,11 +849,11 @@ def _migrate_capital_ledger(conn: sqlite3.Connection) -> None: ")" ) conn.execute( - "CREATE INDEX idx_capital_ledger_audit " + "CREATE INDEX IF NOT EXISTS idx_capital_ledger_audit " "ON capital_ledger(audit_event_hash)" ) conn.execute( - "CREATE INDEX idx_capital_ledger_op " + "CREATE INDEX IF NOT EXISTS idx_capital_ledger_op " "ON capital_ledger(op_type)" ) @@ -844,7 +882,7 @@ def _migrate_controller_events(conn: sqlite3.Connection) -> None: ).fetchone() if row is None: conn.execute( - "CREATE TABLE controller_events (" + "CREATE TABLE IF NOT EXISTS controller_events (" " event_id INTEGER PRIMARY KEY AUTOINCREMENT," " organism_root TEXT NOT NULL," " branch_id TEXT," @@ -860,15 +898,15 @@ def _migrate_controller_events(conn: sqlite3.Connection) -> None: ")" ) conn.execute( - "CREATE INDEX idx_controller_events_organism " + "CREATE INDEX IF NOT EXISTS idx_controller_events_organism " "ON controller_events(organism_root)" ) conn.execute( - "CREATE INDEX idx_controller_events_kind " + "CREATE INDEX IF NOT EXISTS idx_controller_events_kind " "ON controller_events(event_kind)" ) conn.execute( - "CREATE INDEX idx_controller_events_at " + "CREATE INDEX IF NOT EXISTS idx_controller_events_at " "ON controller_events(recorded_at)" ) @@ -894,7 +932,7 @@ def _migrate_fork_score_branches(conn: sqlite3.Connection) -> None: ).fetchone() if row is None: conn.execute( - "CREATE TABLE fork_score_branches (" + "CREATE TABLE IF NOT EXISTS fork_score_branches (" " branch_set_id TEXT NOT NULL," " branch_id TEXT NOT NULL," " parent_root TEXT NOT NULL," @@ -909,11 +947,11 @@ def _migrate_fork_score_branches(conn: sqlite3.Connection) -> None: ")" ) conn.execute( - "CREATE INDEX idx_fork_score_branches_set " + "CREATE INDEX IF NOT EXISTS idx_fork_score_branches_set " "ON fork_score_branches(branch_set_id)" ) conn.execute( - "CREATE INDEX idx_fork_score_branches_parent " + "CREATE INDEX IF NOT EXISTS idx_fork_score_branches_parent " "ON fork_score_branches(parent_root)" ) @@ -933,7 +971,7 @@ def _migrate_memory_root(conn: sqlite3.Connection) -> None: ).fetchone() if row is None: conn.execute( - "CREATE TABLE memory_records (" + "CREATE TABLE IF NOT EXISTS memory_records (" " memory_root TEXT PRIMARY KEY," " schema_version TEXT NOT NULL," " parent_memory_root TEXT," @@ -948,7 +986,7 @@ def _migrate_memory_root(conn: sqlite3.Connection) -> None: ")" ) conn.execute( - "CREATE INDEX idx_memory_state ON memory_records(state)" + "CREATE INDEX IF NOT EXISTS idx_memory_state ON memory_records(state)" ) row = conn.execute( "SELECT name FROM sqlite_master " @@ -956,7 +994,7 @@ def _migrate_memory_root(conn: sqlite3.Connection) -> None: ).fetchone() if row is None: conn.execute( - "CREATE TABLE memory_branch_summaries (" + "CREATE TABLE IF NOT EXISTS memory_branch_summaries (" " branch_id TEXT NOT NULL," " memory_root TEXT NOT NULL," " summary_digest TEXT NOT NULL," @@ -967,7 +1005,7 @@ def _migrate_memory_root(conn: sqlite3.Connection) -> None: ")" ) conn.execute( - "CREATE INDEX idx_branch_memory " + "CREATE INDEX IF NOT EXISTS idx_branch_memory " "ON memory_branch_summaries(memory_root)" ) @@ -985,7 +1023,7 @@ def _migrate_adapter_loss_reports(conn: sqlite3.Connection) -> None: ).fetchone() if row is None: conn.execute( - "CREATE TABLE adapter_loss_reports (" + "CREATE TABLE IF NOT EXISTS adapter_loss_reports (" " chunk_id INTEGER NOT NULL," " document_root TEXT NOT NULL," " stage TEXT NOT NULL," @@ -1007,15 +1045,15 @@ def _migrate_adapter_loss_reports(conn: sqlite3.Connection) -> None: ")" ) conn.execute( - "CREATE INDEX idx_adapter_loss_doc " + "CREATE INDEX IF NOT EXISTS idx_adapter_loss_doc " "ON adapter_loss_reports(document_root, stage)" ) conn.execute( - "CREATE INDEX idx_adapter_loss_kind " + "CREATE INDEX IF NOT EXISTS idx_adapter_loss_kind " "ON adapter_loss_reports(loss_kind, stage)" ) conn.execute( - "CREATE INDEX idx_adapter_loss_chunk " + "CREATE INDEX IF NOT EXISTS idx_adapter_loss_chunk " "ON adapter_loss_reports(chunk_id, stage)" ) @@ -1035,7 +1073,7 @@ def _migrate_mesh_peer_chains(conn: sqlite3.Connection) -> None: ).fetchone() if row is None: conn.execute( - "CREATE TABLE mesh_peer_chains (" + "CREATE TABLE IF NOT EXISTS mesh_peer_chains (" " peer_member_id TEXT PRIMARY KEY," " last_event_hash TEXT NOT NULL," " last_seq INTEGER NOT NULL," diff --git a/tests/test_store_migration_concurrency.py b/tests/test_store_migration_concurrency.py new file mode 100644 index 0000000..43a5355 --- /dev/null +++ b/tests/test_store_migration_concurrency.py @@ -0,0 +1,128 @@ +"""Regression: forward-migration DDL must be idempotent (concurrency-safe). + +Surfaced 2026-05-11 during orientation. Two ``arborist`` invocations a +few seconds apart hit shards 000-003 right after #000012 (Phase 1c) +landed — before the new ``fork_score_branches`` table had been committed +on those shards. ``arborist analyze`` crashed:: + + sqlite3.OperationalError: table fork_score_branches already exists + +Root cause: ``store.connect()`` runs check-``sqlite_master``-then-``CREATE`` +forward migrations. Two connections both pass the existence probe (table +absent), both issue ``CREATE TABLE`` — the loser raises. The probe stays +as the common-case fast-path skip, but the DDL it guards must itself be +``CREATE TABLE/INDEX IF NOT EXISTS`` so the TOCTOU loser no-ops. + +The race window (probe → create within one ``connect()``) is microseconds +and not reliably reproducible from threads under the GIL, so this test +pins the invariant at the source level: every ``CREATE TABLE`` / +``CREATE INDEX`` inside a ``_migrate_*`` helper uses ``IF NOT EXISTS``. +Any future sidecar migration that forgets trips this immediately. +""" + +from __future__ import annotations + +import inspect +import re +import threading +from pathlib import Path + +import pytest + +from arborist import store +from arborist.store import _clear_migration_cache, connect + +# Every forward-migration helper that issues CREATE statements (the +# CHECK-constraint table-rebuild helpers are exempt — they intentionally +# CREATE a fresh `providence_cache_new` and rename). +_MIGRATION_HELPERS = ( + store._migrate_audit_mode, + store._migrate_mesh_peer_chains, + store._migrate_document_http_meta, + store._migrate_selfmodel_tables, + store._migrate_capital_ledger, + store._migrate_memory_root, + store._migrate_adapter_loss_reports, + store._migrate_controller_events, + store._migrate_fork_score_branches, +) + +_BARE_CREATE = re.compile( + r"CREATE\s+(?:UNIQUE\s+)?(?:TABLE|INDEX)\s+(?!IF\s+NOT\s+EXISTS)", + re.IGNORECASE, +) + + +@pytest.mark.parametrize("helper", _MIGRATION_HELPERS, ids=lambda h: h.__name__) +def test_forward_migration_ddl_uses_if_not_exists(helper) -> None: + src = inspect.getsource(helper) + offenders = _BARE_CREATE.findall(src) + assert not offenders, ( + f"{helper.__name__} issues CREATE without IF NOT EXISTS — a " + f"concurrent connect() loser will raise 'already exists': {offenders}" + ) + + +def test_migrating_connect_raises_busy_timeout(tmp_path: Path) -> None: + """The one migration pass raises ``busy_timeout`` so a ``connect()`` + racing the same fresh shard waits on a peer's migration write txn + instead of failing fast with ``database is locked``.""" + _clear_migration_cache() + db = tmp_path / "bt.db" + conn = connect(db) + try: + (busy_timeout,) = conn.execute("PRAGMA busy_timeout").fetchone() + assert busy_timeout == 5000 + finally: + conn.close() + + +def test_concurrent_connect_fresh_shard_smoke(tmp_path: Path) -> None: + """Best-effort behavioural smoke: N barrier-synced ``connect()`` calls + on a brand-new shard (per-process migration memo is lock-free, so the + losers run the migration block too) must not raise, and every sidecar + table must land exactly once. Does not guarantee hitting the TOCTOU + window — the source-level test above is the hard regression pin. + """ + _clear_migration_cache() + db = tmp_path / "race.db" + + n = 8 + barrier = threading.Barrier(n) + errors: list[BaseException] = [] + lock = threading.Lock() + + def worker() -> None: + try: + barrier.wait(timeout=15) + connect(db).close() + except BaseException as exc: # noqa: BLE001 — surface it + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker) for _ in range(n)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=30) + + assert not errors, f"connect() raced on a fresh shard: {errors!r}" + + conn = connect(db) + try: + names = { + row[0] + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ).fetchall() + } + finally: + conn.close() + for table in ( + "fork_score_branches", + "controller_events", + "memory_records", + "capital_ledger", + "selfmodel_records", + ): + assert table in names, f"{table} missing after concurrent connect()" diff --git a/tests/test_store_migration_memoization.py b/tests/test_store_migration_memoization.py index e3a7665..266ecda 100644 --- a/tests/test_store_migration_memoization.py +++ b/tests/test_store_migration_memoization.py @@ -1,10 +1,10 @@ """Regression tests for ticket #000026 Phase 1. Per-process migration memoization. ``arborist.store.connect()`` used -to run all 7 forward-migration probes on every open — 588 redundant -probe sequences per typical query. With memoization, the migration -block runs once per (physical file, process) and is skipped on every -subsequent open of the same shard. +to run every forward-migration probe on every open — hundreds of +redundant probe sequences per typical query. With memoization, the +migration block runs once per (physical file, process) and is skipped +on every subsequent open of the same shard. Tests pin: @@ -38,6 +38,9 @@ class _MigrationProbeCounter: counted versions. """ + # Mirrors the forward-migration sequence in ``store.connect()`` — keep + # in sync when a migration is added (otherwise its memoization goes + # unverified, and the count assertions below silently under-count). NAMES = ( "_migrate_audit_mode", "_migrate_mesh_peer_chains", @@ -46,6 +49,8 @@ class _MigrationProbeCounter: "_migrate_capital_ledger", "_migrate_memory_root", "_migrate_adapter_loss_reports", + "_migrate_controller_events", + "_migrate_fork_score_branches", ) def __init__(self, monkeypatch): @@ -64,7 +69,7 @@ class _MigrationProbeCounter: return sum(self.counts.values()) -def test_first_connect_runs_all_seven_migrations(tmp_path: Path, monkeypatch): +def test_first_connect_runs_all_forward_migrations(tmp_path: Path, monkeypatch): _clear_migration_cache() counter = _MigrationProbeCounter(monkeypatch) db = tmp_path / "first.db" @@ -72,8 +77,8 @@ def test_first_connect_runs_all_seven_migrations(tmp_path: Path, monkeypatch): conn = connect(db) conn.close() - # All seven probes ran exactly once. - assert counter.total == 7 + # Every probe ran exactly once. + assert counter.total == len(counter.NAMES) for name in counter.NAMES: assert counter.counts[name] == 1 @@ -112,6 +117,8 @@ def test_second_connect_schema_is_intact(tmp_path: Path): "capital_ledger", # _migrate_capital_ledger "memory_records", # _migrate_memory_root "adapter_loss_reports", # _migrate_adapter_loss_reports + "controller_events", # _migrate_controller_events + "fork_score_branches", # _migrate_fork_score_branches ): row = conn.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name=?", @@ -137,7 +144,7 @@ def test_clear_migration_cache_forces_reprobe(tmp_path: Path, monkeypatch): # After clearing — full re-probe. _clear_migration_cache() connect(db).close() - assert counter.total == 7 + assert counter.total == len(counter.NAMES) def test_invalidate_migration_cache_after_file_replacement( @@ -169,7 +176,7 @@ def test_invalidate_migration_cache_after_file_replacement( counter = _MigrationProbeCounter(monkeypatch) conn = connect(db) conn.close() - assert counter.total == 7 + assert counter.total == len(counter.NAMES) def test_distinct_paths_each_get_one_probe(tmp_path: Path, monkeypatch): @@ -182,8 +189,8 @@ def test_distinct_paths_each_get_one_probe(tmp_path: Path, monkeypatch): counter = _MigrationProbeCounter(monkeypatch) connect(db_a).close() connect(db_b).close() - # Two paths × seven migrations = 14. - assert counter.total == 14 + # Two paths × the full migration sequence. + assert counter.total == 2 * len(counter.NAMES) # Re-open both — cache hit on each, zero new probes. before = counter.total