#!/usr/bin/env python3 """Run local SQLite crash/replay experiments. No network, AWS, or credentials. Use an output directory that does not yet exist: python3 -I retry-boundaries.py --output evidence/run-1 """ import argparse import datetime import hashlib import json import os from pathlib import Path import platform import select import sqlite3 import subprocess import sys import time from urllib.parse import quote EVENT_ID = "demo-event-001" PAYLOAD = {"counter_delta": 1} CRASH_EXIT = 86 BUSY_EXIT = 74 CHILD_TIMEOUT_SECONDS = 10 SCHEMA = """ CREATE TABLE processed ( event_id TEXT PRIMARY KEY, payload_sha256 TEXT NOT NULL ); CREATE TABLE effects ( sequence INTEGER PRIMARY KEY, event_id TEXT NOT NULL, delta INTEGER NOT NULL CHECK (delta = 1) ); """ # Fixed before execution. These are hypotheses, not generated observations. CASES = [ {"id": "A-no-dedup", "mode": "naive", "crash": "none", "expected": [[0, 1, 0], [0, 2, 0]]}, {"id": "B-marker-first", "mode": "marker-first", "crash": "after-marker", "expected": [[1, 0, 0], [1, 0, 0]]}, {"id": "C-effect-first", "mode": "effect-first", "crash": "after-effect", "expected": [[0, 1, 0], [1, 2, 0]]}, {"id": "D-atomic-before-commit", "mode": "atomic", "crash": "before-commit", "expected": [[0, 0, 0], [1, 1, 0]]}, {"id": "E-atomic-after-commit", "mode": "atomic", "crash": "after-commit", "expected": [[1, 1, 0], [1, 1, 0]]}, {"id": "F-overlapping-workers", "mode": "atomic", "crash": "none", "expected": [[0, 0, 0], [1, 1, 0], [1, 1, 0]]}, {"id": "G-side-effect-outside-db", "mode": "external", "crash": "before-commit", "expected": [[0, 0, 1], [1, 1, 2]]}, ] def now(): return datetime.datetime.now(datetime.timezone.utc).isoformat() def encoded(value): return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") def sha256(data): return hashlib.sha256(data).hexdigest() def write_json(path, data): with path.open("w", encoding="utf-8") as handle: json.dump(data, handle, ensure_ascii=False, indent=2) handle.write("\n") handle.flush() os.fsync(handle.fileno()) def worker_event(event, **fields): print(json.dumps({"observedAt": now(), "event": event, **fields}, ensure_ascii=False), flush=True) def crash_at(requested, current): if requested == current: worker_event("forced-process-exit", point=current, exitCode=CRASH_EXIT) os._exit(CRASH_EXIT) def worker(args): db = sqlite3.connect(args.database, timeout=0, isolation_level=None) db.execute("PRAGMA synchronous = FULL") digest = sha256(encoded(PAYLOAD)) try: worker_event("opened-database", mode=args.mode) if args.mode == "naive": db.execute("INSERT INTO effects(event_id,delta) VALUES (?,1)", (EVENT_ID,)) worker_event("effect-autocommitted") elif args.mode in {"marker-first", "effect-first"}: if db.execute("SELECT 1 FROM processed WHERE event_id=?", (EVENT_ID,)).fetchone(): worker_event("duplicate-skipped") return 0 if args.mode == "marker-first": db.execute("INSERT INTO processed VALUES (?,?)", (EVENT_ID, digest)) worker_event("marker-autocommitted") crash_at(args.crash, "after-marker") db.execute("INSERT INTO effects(event_id,delta) VALUES (?,1)", (EVENT_ID,)) worker_event("effect-autocommitted") else: db.execute("INSERT INTO effects(event_id,delta) VALUES (?,1)", (EVENT_ID,)) worker_event("effect-autocommitted") crash_at(args.crash, "after-effect") db.execute("INSERT INTO processed VALUES (?,?)", (EVENT_ID, digest)) worker_event("marker-autocommitted") else: worker_event("begin-immediate-attempt") db.execute("BEGIN IMMEDIATE") worker_event("transaction-open") try: db.execute("INSERT INTO processed VALUES (?,?)", (EVENT_ID, digest)) except sqlite3.IntegrityError: db.execute("ROLLBACK") worker_event("duplicate-skipped") return 0 db.execute("INSERT INTO effects(event_id,delta) VALUES (?,1)", (EVENT_ID,)) worker_event("marker-and-effect-written-uncommitted") if args.mode == "external": with Path(args.external).open("ab") as handle: handle.write(encoded({"event_id": EVENT_ID, "action": "append-notification"}) + b"\n") handle.flush() os.fsync(handle.fileno()) worker_event("external-file-appended-and-fsynced") if args.hold: worker_event("waiting-for-parent-commit-command") if sys.stdin.readline().strip() != "COMMIT": raise RuntimeError("parent did not send the explicit commit command") crash_at(args.crash, "before-commit") db.execute("COMMIT") worker_event("transaction-committed") crash_at(args.crash, "after-commit") worker_event("completion-returned") return 0 except sqlite3.OperationalError as error: worker_event("sqlite-operational-error", sqliteCode=getattr(error, "sqlite_errorcode", None), sqliteName=getattr(error, "sqlite_errorname", None), message=str(error)) return BUSY_EXIT if getattr(error, "sqlite_errorcode", None) == sqlite3.SQLITE_BUSY else 75 except Exception as error: worker_event("unexpected-worker-failure", errorType=type(error).__name__, message=str(error)) return 76 finally: db.close() class Recorder: def __init__(self, path): self.path = path self.sequence = 0 def record(self, event, **fields): self.sequence += 1 item = {"sequence": self.sequence, "recordedAt": now(), "event": event, **fields} with self.path.open("ab") as handle: handle.write(encoded(item) + b"\n") handle.flush() os.fsync(handle.fileno()) return item def setup_database(path): with sqlite3.connect(path) as db: db.execute("PRAGMA journal_mode = DELETE") db.execute("PRAGMA synchronous = FULL") db.executescript(SCHEMA) def snapshot(case_dir): database = case_dir / "observations.sqlite" uri = "file:" + quote(str(database.resolve())) + "?mode=ro" with sqlite3.connect(uri, uri=True, timeout=0) as db: markers = db.execute("SELECT event_id,payload_sha256 FROM processed ORDER BY event_id").fetchall() effects = db.execute("SELECT sequence,event_id,delta FROM effects ORDER BY sequence").fetchall() integrity = db.execute("PRAGMA integrity_check").fetchall() external = case_dir / "external-notifications.jsonl" external_rows = [json.loads(line) for line in external.read_text().splitlines()] if external.exists() else [] return { "processedRows": markers, "effectRows": effects, "effectTotal": sum(row[2] for row in effects), "externalRows": external_rows, "integrityCheck": integrity, "counts": [len(markers), sum(row[2] for row in effects), len(external_rows)], } def spawn(case_dir, case, label, recorder, crash="none", hold=False): args = [sys.executable, "-I", str(Path(__file__).resolve()), "--worker", "--database", str((case_dir / "observations.sqlite").resolve()), "--external", str((case_dir / "external-notifications.jsonl").resolve()), "--mode", case["mode"], "--crash", crash] if hold: args.append("--hold") recorder.record("worker-start", case=case["id"], worker=label, mode=case["mode"], crash=crash, hold=hold) process = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0) return process def save_child_events(raw, case_id, label, recorder): events = [] for line in raw.decode("utf-8").splitlines(): item = json.loads(line) recorder.record("worker-observation", case=case_id, worker=label, observation=item) events.append(item) return events def finish(process, case_id, label, recorder, expected_exit): try: output, errors = process.communicate(timeout=CHILD_TIMEOUT_SECONDS) except subprocess.TimeoutExpired: process.kill() # Only this runner's own child handle. output, errors = process.communicate(timeout=CHILD_TIMEOUT_SECONDS) save_child_events(output, case_id, label, recorder) recorder.record("worker-timeout", case=case_id, worker=label) raise RuntimeError("child timed out; run stopped with observations preserved") events = save_child_events(output, case_id, label, recorder) recorder.record("worker-exit", case=case_id, worker=label, exitCode=process.returncode, stderr=errors.decode("utf-8")) if process.returncode != expected_exit or errors: raise RuntimeError(f"{case_id}/{label}: unexpected exit code or stderr") return events def wait_for_event(process, name, case_id, label, recorder): deadline = time.monotonic() + CHILD_TIMEOUT_SECONDS while time.monotonic() < deadline: ready, _, _ = select.select([process.stdout], [], [], max(0, deadline - time.monotonic())) if not ready: break raw = process.stdout.readline() if not raw: raise RuntimeError("held worker ended before the barrier") events = save_child_events(raw, case_id, label, recorder) if events[0]["event"] == name: return raise RuntimeError("held worker did not reach barrier within the bounded deadline") def observe(case_dir, case, stage, recorder, observations): actual = snapshot(case_dir) item = {"stage": stage, "observedAt": now(), **actual} observations.append(item) recorder.record("database-readback", case=case["id"], observation=item) def run_case(out, case, recorder): case_dir = out / case["id"] case_dir.mkdir() setup_database(case_dir / "observations.sqlite") observations = [] if case["id"] == "F-overlapping-workers": held = spawn(case_dir, case, "writer-1-held", recorder, hold=True) try: wait_for_event(held, "waiting-for-parent-commit-command", case["id"], "writer-1-held", recorder) competing = spawn(case_dir, case, "writer-2-overlap", recorder) events = finish(competing, case["id"], "writer-2-overlap", recorder, BUSY_EXIT) if not any(event.get("sqliteCode") == sqlite3.SQLITE_BUSY for event in events): raise RuntimeError("overlap did not observe SQLITE_BUSY") observe(case_dir, case, "during-uncommitted-writer-and-confirmed-lock-conflict", recorder, observations) held.stdin.write(b"COMMIT\n") held.stdin.flush() finish(held, case["id"], "writer-1-held", recorder, 0) finally: if held.poll() is None: held.kill() held.communicate(timeout=CHILD_TIMEOUT_SECONDS) observe(case_dir, case, "after-first-writer-commit", recorder, observations) # This replay is part of the declared scenario, not an automatic busy retry. replay = spawn(case_dir, case, "writer-2-explicit-replay-after-commit", recorder) finish(replay, case["id"], "writer-2-explicit-replay-after-commit", recorder, 0) observe(case_dir, case, "after-explicit-replay", recorder, observations) else: first = spawn(case_dir, case, "first-attempt", recorder, crash=case["crash"]) finish(first, case["id"], "first-attempt", recorder, 0 if case["crash"] == "none" else CRASH_EXIT) observe(case_dir, case, "after-first-attempt", recorder, observations) replay = spawn(case_dir, case, "explicit-replay", recorder) finish(replay, case["id"], "explicit-replay", recorder, 0) observe(case_dir, case, "after-explicit-replay", recorder, observations) comparisons = [] for actual, expected in zip(observations, case["expected"], strict=True): comparisons.append({"stage": actual["stage"], "expectedCounts": expected, "observedCounts": actual["counts"], "matches": actual["counts"] == expected and actual["integrityCheck"] == [("ok",)]}) result = {"case": case["id"], "observations": observations, "comparisons": comparisons, "status": "MATCHED_PREDECLARED_EXPECTATIONS" if all(c["matches"] for c in comparisons) else "MISMATCH"} write_json(case_dir / "result.json", result) recorder.record("case-completed", case=case["id"], status=result["status"]) if result["status"] == "MISMATCH": raise RuntimeError(f"{case['id']}: mismatch; stop without changing hypotheses") return result def run(out): out.mkdir(parents=True, exist_ok=False) recorder = Recorder(out / "observations.jsonl") manifest = {"experiment": "sqlite-retry-boundaries-v1", "createdAt": now(), "input": {"event_id": EVENT_ID, "payload": PAYLOAD}, "inputSHA256": sha256(encoded({"event_id": EVENT_ID, "payload": PAYLOAD})), "scriptSHA256": sha256(Path(__file__).read_bytes()), "environment": {"python": platform.python_version(), "sqlite": sqlite3.sqlite_version, "os": platform.system(), "osRelease": platform.release()}, "countOrder": ["processed_rows", "effect_total", "external_file_rows"], "cases": CASES, "limits": ["Local SQLite and process termination only; no AWS or network execution.", "Synthetic input; no production data, payment, learning outcomes, or performance claims.", "No power-loss, disk-failure, distributed-transaction, or all-scheduling guarantee."]} write_json(out / "manifest.json", manifest) recorder.record("run-started", manifestSHA256=sha256((out / "manifest.json").read_bytes())) results = [] failure = None try: for case in CASES: results.append(run_case(out, case, recorder)) except Exception as error: failure = {"type": type(error).__name__, "message": str(error)} recorder.record("run-stopped", failure=failure) complete = failure is None and len(results) == len(CASES) summary = {"experiment": manifest["experiment"], "completedAt": now(), "status": "COMPLETE" if complete else "STOPPED", "casesCompleted": len(results), "casesPlanned": len(CASES), "failure": failure, "completedCases": [r["case"] for r in results], "unfinishedCases": [c["id"] for c in CASES if c["id"] not in {r["case"] for r in results}], "results": results} write_json(out / "results.json", summary) recorder.record("run-finished", status=summary["status"], casesCompleted=len(results)) inventory = {str(p.relative_to(out)): sha256(p.read_bytes()) for p in sorted(out.rglob("*")) if p.is_file()} write_json(out / "artifact-hashes.json", inventory) print(json.dumps({"status": summary["status"], "casesCompleted": len(results), "casesPlanned": len(CASES), "failure": failure}, ensure_ascii=False)) return 0 if complete else 2 def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", type=Path) parser.add_argument("--worker", action="store_true", help=argparse.SUPPRESS) parser.add_argument("--database", help=argparse.SUPPRESS) parser.add_argument("--external", help=argparse.SUPPRESS) parser.add_argument("--mode", choices=["naive", "marker-first", "effect-first", "atomic", "external"], help=argparse.SUPPRESS) parser.add_argument("--crash", default="none", choices=["none", "after-marker", "after-effect", "before-commit", "after-commit"], help=argparse.SUPPRESS) parser.add_argument("--hold", action="store_true", help=argparse.SUPPRESS) args = parser.parse_args() if args.worker: if not args.database or not args.external or not args.mode: parser.error("internal worker arguments missing") return worker(args) if args.output is None: parser.error("--output must name a new directory") if args.output.exists(): parser.error("output already exists; refusing to overwrite or resume") return run(args.output) if __name__ == "__main__": sys.exit(main())