store: append_audit atomic head-read+insert — no audit-chain forks
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 read the chain head and INSERTed as two separate autocommit statements (the burn caller ran it OUTSIDE its transaction() block), so two concurrent appenders both chained off the same head; seq AUTOINCREMENT serialized the rows but not the hash linkage. - append_audit wraps head-read + insert in its own BEGIN IMMEDIATE/COMMIT when the connection isn't already in a transaction (folds in otherwise). - connect() now sets busy_timeout=5000 on every connection (not just the migration pass) so a peer mid-write makes us wait, not fail-fast — and a fail-and-retry appender can't re-read a stale head. - test_audit_chain_concurrency.py: 8 threads x 20 append_audit() on one shard must yield a linear chain (one genesis, no forked parents); + the in-transaction fold-in / rollback behaviour. Verified the concurrency test fails without the fix (8 genesis rows).
This commit is contained in:
parent
3ea27aa471
commit
af870bbf23
2 changed files with 168 additions and 25 deletions
|
|
@ -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
|
||||
|
||||
|
||||
|
|
|
|||
121
tests/test_audit_chain_concurrency.py
Normal file
121
tests/test_audit_chain_concurrency.py
Normal file
|
|
@ -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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue