engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

store: idempotent forward migrations — survive concurrent connect()

Browse source 
`arborist analyze` crashed during orientation with
`sqlite3.OperationalError: table fork_score_branches already exists`:
check-sqlite_master-then-CREATE forward migrations let two connect()
calls racing a fresh shard both pass the probe and both issue CREATE.

- CREATE TABLE/INDEX inside the _migrate_* helpers now IF NOT EXISTS
  (probe stays as the fast-path skip).
- _migrate_audit_mode's ALTER ADD COLUMN routes through
  _add_column_if_missing (PRAGMA fast-path + duplicate-column catch).
- connect() raises busy_timeout=5000 for the one migration pass so a
  racing connect() waits on a peer's _rebuild_* write txn.
- test_store_migration_concurrency.py: source-level "all CREATE is
  IF NOT EXISTS" pin, busy_timeout assertion, 8-thread connect smoke.
- test_store_migration_memoization.py: _MigrationProbeCounter.NAMES
  was missing two migrations; synced + de-hardcoded counts.
This commit is contained in:
russell@unturf.com 2026-05-11 07:28:35 -04:00
parent e894634406
commit 71c98487b7
No known key found for this signature in database
3 changed files with 245 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

160
arborist/store.py
View file

@ -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,"

128
tests/test_store_migration_concurrency.py Normal file
View file

@ -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()"

29
tests/test_store_migration_memoization.py
View file

@ -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