arborist/tests/test_store_migration_concurrency.py

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