Concurrent CLI starts race on the task_readiness view #18

#!/usr/bin/env python3
"""Reproduce concurrent PlanDB CLI startup against one SQLite database.

Run after `cargo build`: python3 target/race/reproduce.py
Temporary databases stay under target/race; results go to the checkout root.
"""

import argparse
import json
import os
from pathlib import Path
import subprocess
import tempfile


ROOT = Path(__file__).resolve().parents[2]
BIN = ROOT / "target" / "debug" / "plandb"
ERRORS = ("no such table: task_readiness", "view task_readiness already exists")


def run_setup(command, cwd, env):
    result = subprocess.run(
        [str(BIN), *command], cwd=cwd, env=env, capture_output=True, text=True
    )
    if result.returncode:
        raise RuntimeError(
            f"setup {' '.join(command)} failed ({result.returncode}): {result.stderr}"
        )


def line_number(needle):
    lines = (ROOT / "src" / "db" / "schema.rs").read_text().splitlines()
    matches = [number for number, line in enumerate(lines, 1) if line.strip() == needle]
    if len(matches) != 1:
        raise RuntimeError(f"expected one occurrence of {needle!r}; found {matches}")
    return matches[0]


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--waves", type=int, default=25, help="waves per attempt")
    parser.add_argument("--attempts", type=int, default=3)
    parser.add_argument("--build-seconds", type=float, default=0.0)
    args = parser.parse_args()
    if args.waves < 1 or args.attempts < 1:
        parser.error("--waves and --attempts must be positive")
    if not BIN.is_file():
        parser.error(f"missing built binary: {BIN}")

    commit = subprocess.check_output(
        ["git", "rev-parse", "HEAD"], cwd=ROOT, text=True
    ).strip()
    lines = {
        "drop_view": line_number("DROP VIEW IF EXISTS task_readiness;"),
        "create_view": line_number("CREATE VIEW task_readiness AS"),
        "init_db_call": line_number("conn.execute_batch(CREATE_TASK_READINESS_VIEW)?;"),
    }
    runs = []
    attempt_lines = []
    examples = []
    with tempfile.TemporaryDirectory(prefix="db-", dir=Path(__file__).parent) as work:
        work = Path(work)
        env = os.environ.copy()
        env["PLANDB_DB"] = str(work / ".plandb.db")
        run_setup(["init", "Readiness race"], work, env)
        for task in (
            ["add", "Design schema", "--as", "schema", "--description", "Define tables"],
            ["add", "Build API", "--as", "api", "--dep", "t-schema", "--description", "Build routes"],
            ["add", "Write tests", "--as", "tests", "--dep", "t-schema", "--description", "Cover routes"],
            ["add", "Deploy", "--as", "deploy", "--dep", "t-api", "--dep", "t-tests", "--description", "Release service"],
        ):
            run_setup(task, work, env)
        run_setup(["list"], work, env)
        run_setup(["status"], work, env)

        for concurrency in (8, 16):
            total = no_such_table = view_exists = other_failures = 0
            for attempt in range(1, args.attempts + 1):
                counts = [0, 0, 0, 0]
                for wave in range(1, args.waves + 1):
                    processes = []
                    for number in range(1, concurrency + 1):
                        command = "list" if (number + wave) % 2 else "status"
                        process = subprocess.Popen(
                            [str(BIN), command],
                            cwd=work,
                            env=env,
                            stdout=subprocess.DEVNULL,
                            stderr=subprocess.PIPE,
                            text=True,
                        )
                        processes.append((number, command, process))
                    for number, command, process in processes:
                        _, stderr = process.communicate()
                        counts[0] += 1
                        if ERRORS[0] in stderr:
                            counts[1] += 1
                        if ERRORS[1] in stderr:
                            counts[2] += 1
                        if process.returncode and not any(error in stderr for error in ERRORS):
                            counts[3] += 1
                        if len(examples) < 3:
                            for line in stderr.splitlines():
                                if any(error in line for error in ERRORS):
                                    examples.append(
                                        (concurrency, attempt, wave, number, command, line)
                                    )
                                    break
                total += counts[0]
                no_such_table += counts[1]
                view_exists += counts[2]
                other_failures += counts[3]
                attempt_lines.append(
                    f"N={concurrency} attempt={attempt}: processes={counts[0]}, "
                    f"no such table={counts[1]}, view exists={counts[2]}, "
                    f"other failures={counts[3]}"
                )
            runs.append(
                {
                    "concurrency": concurrency,
                    "total_processes": total,
                    "errors_no_such_table": no_such_table,
                    "errors_view_exists": view_exists,
                }
            )

    result = {
        "commit": commit,
        "build_ok": True,
        "build_seconds": args.build_seconds,
        "runs": runs,
        "reproduced": any(
            run["errors_no_such_table"] or run["errors_view_exists"] for run in runs
        ),
        "line_numbers": lines,
        "repro_script_path": str(Path(__file__).resolve()),
    }
    (ROOT / "RACE-RESULT.json").write_text(json.dumps(result, indent=2) + "\n")

    log = [
        f"commit: {commit}",
        f"cargo build --locked: succeeded in {args.build_seconds:.3f} seconds",
        f"database: isolated temporary file under {Path(__file__).parent}",
        "setup: plandb init, four tasks, list and status smoke checks",
        f"line numbers: drop={lines['drop_view']}, create={lines['create_view']}, init_db={lines['init_db_call']}",
        "",
        *attempt_lines,
        "",
    ]
    for run in runs:
        log.append(
            f"N={run['concurrency']} total: processes={run['total_processes']}, "
            f"no such table={run['errors_no_such_table']}, "
            f"view exists={run['errors_view_exists']}"
        )
    log.extend(["", "Example error lines:"])
    if examples:
        for concurrency, attempt, wave, number, command, line in examples:
            log.extend(
                [
                    f"N={concurrency} attempt={attempt} wave={wave} process={number} ({command}):",
                    line,
                ]
            )
    else:
        log.append("No matching error lines captured.")
    if not result["reproduced"]:
        log.append("Race did not reproduce after all attempts at N=8 and N=16.")
    if other_failures:
        log.append(f"N=16 other nonzero exits: {other_failures}")
    (ROOT / "RACE-RESULT.log").write_text("\n".join(log) + "\n")
    print(
        f"reproduced={str(result['reproduced']).lower()} "
        f"N=8(no_such_table={runs[0]['errors_no_such_table']},view_exists={runs[0]['errors_view_exists']}) "
        f"N=16(no_such_table={runs[1]['errors_no_such_table']},view_exists={runs[1]['errors_view_exists']})"
    )


if __name__ == "__main__":
    main()