diff --git a/arborist/store.py b/arborist/store.py index 62b6af8..da24dc8 100644 --- a/arborist/store.py +++ b/arborist/store.py @@ -129,9 +129,11 @@ CREATE TABLE IF NOT EXISTS edges ( PRIMARY KEY (src_root, edge_type, dst_root, dst_uri, anchor) ) WITHOUT ROWID; CREATE INDEX IF NOT EXISTS idx_edges_dst_root ON edges(dst_root) WHERE dst_root <> ''; --- idx_edges_dst_uri intentionally omitted: only the gravity_top_inbound --- analytical query in cli.py filters on dst_uri alone, and a full scan + --- sort over edges is acceptable for that one-shot reporting path. +-- No idx_edges_dst_uri: the gravity_top_inbound analytical query in cli.py +-- counts inbound links per resolved destination *document* (dst_root), which +-- this partial index already serves with a streaming GROUP BY. (An earlier +-- formulation grouped by the raw dst_uri link string with no index and +-- hash-aggregated over every target -> unbounded RSS at corpus scale.) -- Distillation: core_root <- src_root with Merkle-signed proof binding. CREATE TABLE IF NOT EXISTS derivations ( @@ -558,24 +560,22 @@ def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection: 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. + ``busy_timeout`` is set on every connection (before the migration + pass, so it covers that too): a peer mid-write — migration DDL, a + ``transaction()`` block, ``append_audit``'s own ``BEGIN IMMEDIATE`` — + makes us *wait* rather than fail fast with ``database is locked``. + Without it, concurrent appenders that fail-and-retry can re-read a + stale chain head and fork the audit chain (qa.db seq 7724/7725 was + that bug); waiting + the ``BEGIN IMMEDIATE`` serialization fixes it. """ p = Path(db_path) p.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(p, isolation_level=None) # autocommit; we'll BEGIN manually conn.row_factory = sqlite3.Row + conn.execute("PRAGMA busy_timeout = 5000") # wait on a peer's write lock, don't fail-fast 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) @@ -1587,25 +1587,47 @@ def append_audit( ) -> str: """Append one event to the audit chain. Returns the new event_hash (hex). + Atomic head-read + insert. If the connection is not already inside a + transaction, the read of the current chain head and the INSERT run + inside this call's own ``BEGIN IMMEDIATE`` / ``COMMIT`` — so two + concurrent appenders serialize on the write lock instead of both + reading the same head and chaining off it (which forks the chain; + qa.db seq 7724/7725 was exactly that, from two concurrent + ``providence_burn`` writes). A caller already inside a ``transaction()`` + gets the append folded into that unit. With ``connect()``'s + ``busy_timeout`` the loser waits rather than failing ``database is + locked``. + Convenience wrapper for one-off events. Bulk inserts should use - chain_audit_events() + executemany() for ~10x throughput on large batches. + chain_audit_events() + executemany() inside a ``transaction()`` for + ~10x throughput on large batches (same serialization guarantee). """ import hashlib if ts is None: ts = int(time.time()) - prev = latest_event_hash(conn) body_json = _canonical_json(body) - h = hashlib.sha256() - if prev is not None: - h.update(bytes.fromhex(prev)) - h.update(body_json.encode("utf-8", errors="surrogatepass")) - event_hash = h.hexdigest() - conn.execute( - "INSERT INTO audit_events (event_hash, prev_event_hash, event_type, subject_root, body, ts) " - "VALUES (?, ?, ?, ?, ?, ?)", - (event_hash, prev, event_type, subject_root, body_json, ts), - ) + own_txn = not conn.in_transaction + if own_txn: + conn.execute("BEGIN IMMEDIATE") + try: + prev = latest_event_hash(conn) + h = hashlib.sha256() + if prev is not None: + h.update(bytes.fromhex(prev)) + h.update(body_json.encode("utf-8", errors="surrogatepass")) + event_hash = h.hexdigest() + conn.execute( + "INSERT INTO audit_events (event_hash, prev_event_hash, event_type, subject_root, body, ts) " + "VALUES (?, ?, ?, ?, ?, ?)", + (event_hash, prev, event_type, subject_root, body_json, ts), + ) + except BaseException: + if own_txn: + conn.execute("ROLLBACK") + raise + if own_txn: + conn.execute("COMMIT") return event_hash diff --git a/tests/test_audit_chain_concurrency.py b/tests/test_audit_chain_concurrency.py new file mode 100644 index 0000000..9429946 --- /dev/null +++ b/tests/test_audit_chain_concurrency.py @@ -0,0 +1,121 @@ +"""Regression: ``append_audit`` must not fork the chain under concurrency. + +qa.db carried a 2-way fork at ``seq`` 7724/7725 — two ``providence_burn`` +events whose ``prev_event_hash`` both pointed at row 7723's ``event_hash``. +Cause: ``append_audit`` did ``prev = latest_event_hash(conn)`` then +``INSERT`` as two separate autocommit statements (the burn caller ran it +*outside* its ``transaction()`` block). Two concurrent appenders read the +same head, both chained off it; ``seq AUTOINCREMENT`` serialized the rows +but not the hash linkage. + +Fix: ``append_audit`` wraps its head-read + insert in its own +``BEGIN IMMEDIATE``/``COMMIT`` when the connection isn't already in a +transaction, and ``connect()`` sets a ``busy_timeout`` so the loser waits +on the write lock instead of failing ``database is locked`` (and then +re-reading a stale head). This test fans many ``append_audit`` calls +across barrier-synced threads against one shard and asserts the result is +a clean linear chain — every non-NULL ``prev_event_hash`` claimed by +exactly one row, exactly one genesis. +""" + +from __future__ import annotations + +import threading +from pathlib import Path + +from arborist.store import _clear_migration_cache, append_audit, connect + + +def test_concurrent_append_audit_keeps_chain_linear(tmp_path: Path) -> None: + _clear_migration_cache() + db = tmp_path / "fork.db" + connect(db).close() # bootstrap schema (no audit rows written) + + n_threads = 8 + per_thread = 20 + barrier = threading.Barrier(n_threads) + errors: list[BaseException] = [] + lock = threading.Lock() + + def worker(tid: int) -> None: + try: + conn = connect(db) + try: + barrier.wait(timeout=20) + for j in range(per_thread): + append_audit( + conn, event_type="test_event", body={"tid": tid, "j": j} + ) + finally: + conn.close() + except BaseException as exc: # noqa: BLE001 — surface it + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=60) + + assert not errors, f"concurrent append_audit raised: {errors!r}" + + conn = connect(db) + try: + forks = conn.execute( + "SELECT prev_event_hash, COUNT(*) AS k FROM audit_events " + "WHERE prev_event_hash IS NOT NULL GROUP BY prev_event_hash HAVING k > 1" + ).fetchall() + genesis = conn.execute( + "SELECT COUNT(*) FROM audit_events WHERE prev_event_hash IS NULL" + ).fetchone()[0] + total = conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] + # Walk by seq: every row chains off the immediately-prior row. + rows = conn.execute( + "SELECT event_hash, prev_event_hash FROM audit_events ORDER BY seq" + ).fetchall() + finally: + conn.close() + + assert not forks, f"audit chain forked: {[dict(r) for r in forks]}" + assert genesis == 1, f"expected one genesis row, got {genesis}" + assert total == n_threads * per_thread + + prev = None + for r in rows: + assert r["prev_event_hash"] == prev, "chain not linear in seq order" + prev = r["event_hash"] + + +def test_append_audit_inside_transaction_folds_in(tmp_path: Path) -> None: + """When the caller already holds a ``transaction()``, ``append_audit`` + must NOT open a nested ``BEGIN IMMEDIATE`` (SQLite forbids it) — it + folds the head-read + insert into the caller's unit, and a rollback + drops the audit row with the rest of the work.""" + from arborist.store import transaction + + _clear_migration_cache() + db = tmp_path / "txn.db" + conn = connect(db) + try: + before = conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] + # Commit path: audit row persists. + with transaction(conn): + append_audit(conn, event_type="t", body={"k": 1}) + assert ( + conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] + == before + 1 + ) + # Rollback path: audit row is dropped with the rest. + try: + with transaction(conn): + append_audit(conn, event_type="t", body={"k": 2}) + raise RuntimeError("boom") + except RuntimeError: + pass + assert ( + conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0] + == before + 1 + ) + finally: + conn.close()