arborist/tests/test_audit_chain_concurrency.py

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