#!/usr/bin/env python3 """Original LockedIn Labs lab. Python 3.11+, standard library, invented records. Two connections to a disposable SQLite database; no network or platform writes. --check-before-lock deliberately moves only the case read outside the transaction. This is a local transaction exercise, not a banking adapter or authorization service. """ import argparse from copy import deepcopy from pathlib import Path import json import sqlite3 import sys from tempfile import TemporaryDirectory if sys.version_info < (3, 11): raise SystemExit("This lab requires Python 3.11 or later. Check your runtime with python3 --version.") WORKSPACE = "SYNTHETIC-WORKSPACE-1" REQUEST = { "workspace": WORKSPACE, "operation_id": "SYNTHETIC-NOTE-1", "case_id": "SYNTHETIC-CASE-104", "case_revision": 1, "source_revision": 1, "statement": "temporary_credit_recorded", } MEANING = {"provisional_credit": "temporary_credit_recorded", "final_credit": "final_credit_recorded"} def approved(request): """Trusted fixture only: a real request must not be able to approve itself.""" return {"request": deepcopy(request), "decision": "accepted", "role": "case_operations"} def connect(path): connection = sqlite3.connect(path, timeout=0, isolation_level=None) connection.row_factory = sqlite3.Row return connection def initialize(connection): connection.executescript(""" CREATE TABLE cases ( workspace TEXT NOT NULL, case_id TEXT NOT NULL, case_revision INTEGER NOT NULL, source_revision INTEGER NOT NULL, credit_state TEXT NOT NULL, PRIMARY KEY(workspace, case_id) ); CREATE TABLE notes ( note_id INTEGER PRIMARY KEY, workspace TEXT NOT NULL, operation_id TEXT NOT NULL, request_json TEXT NOT NULL, case_id TEXT NOT NULL, case_revision INTEGER NOT NULL, source_revision INTEGER NOT NULL, statement TEXT NOT NULL, UNIQUE(workspace, operation_id) ); """) connection.execute("INSERT INTO cases VALUES (?, ?, 1, 1, 'provisional_credit')", (WORKSPACE, REQUEST["case_id"])) def source(connection, request): row = connection.execute("SELECT * FROM cases WHERE workspace=? AND case_id=?", (request["workspace"], request["case_id"])).fetchone() return dict(row) if row else None def notes(connection): return [dict(row) for row in connection.execute("SELECT * FROM notes ORDER BY note_id")] def change_source(connection): connection.execute( "UPDATE cases SET case_revision=2, source_revision=2, credit_state='final_credit' WHERE workspace=? AND case_id=?", (WORKSPACE, REQUEST["case_id"])) class InjectedFailure(Exception): pass def write_note(connection, request, review, *, trusted_workspace=WORKSPACE, check_before_lock=False, before_transaction=None, after_insert=None, timeline=None, actor="A"): trace = [] def record(event): trace.append(event) if timeline is not None: timeline.append({"connection": actor, "event": event}) def finish(outcome, receipt=None): owners = { "saved": "case_operations", "replayed": "case_operations", "stale_review": "case_operations", "review_not_bound": "case_operations", "unsupported_meaning": "case_operations", "operation_conflict": "delivery_operations", "database_busy": "delivery_operations", "rolled_back": "delivery_operations", "scope_denied": "access_operations", "case_missing": "case_operations", } return {"outcome": outcome, "new_write": outcome == "saved", "receipt": receipt, "receipt_kind": "historical_committed_note" if receipt else None, "receiving_owner": owners[outcome], "trace": trace} # This trusted scope is supplied by the exercise, not inferred from generated text. if request["workspace"] != trusted_workspace: record("trusted_scope_rejected") return finish("scope_denied") payload = json.dumps(request, sort_keys=True, separators=(",", ":")) captured = source(connection, request) if check_before_lock else None if check_before_lock: record("case_read_before_transaction") if before_transaction: before_transaction() try: connection.execute("BEGIN IMMEDIATE") record("write_transaction_acquired") existing = connection.execute("SELECT * FROM notes WHERE workspace=? AND operation_id=?", (trusted_workspace, request["operation_id"])).fetchone() record("operation_receipt_checked") if existing: connection.execute("ROLLBACK") record("no_new_write_transaction_closed") return finish("replayed", dict(existing)) if existing["request_json"] == payload else finish("operation_conflict") current = captured if check_before_lock else source(connection, request) record("captured_case_used" if check_before_lock else "case_read_inside_transaction") reason = None if not current: reason = "case_missing" elif review != {"request": request, "decision": "accepted", "role": "case_operations"}: reason = "review_not_bound" elif any(request[key] != current[key] for key in ("case_revision", "source_revision")): reason = "stale_review" elif MEANING.get(current["credit_state"]) != request["statement"]: reason = "unsupported_meaning" if reason: record(reason) connection.execute("ROLLBACK") record("no_new_write_transaction_closed") return finish(reason) record("review_versions_and_meaning_match") cursor = connection.execute( "INSERT INTO notes (workspace, operation_id, request_json, case_id, case_revision, source_revision, statement) VALUES (?, ?, ?, ?, ?, ?, ?)", (trusted_workspace, request["operation_id"], payload, request["case_id"], request["case_revision"], request["source_revision"], request["statement"])) record("note_inserted_uncommitted") if after_insert: after_insert() receipt = dict(connection.execute("SELECT * FROM notes WHERE note_id=?", (cursor.lastrowid,)).fetchone()) connection.execute("COMMIT") record("note_committed") return finish("saved", receipt) except InjectedFailure: connection.rollback() record("injected_failure_rolled_back") return finish("rolled_back") except sqlite3.OperationalError as error: connection.rollback() if getattr(error, "sqlite_errorcode", None) != sqlite3.SQLITE_BUSY: raise record("sqlite_busy_no_commit") return finish("database_busy") except Exception: connection.rollback() raise def attempt(connection, request, review, **kwargs): before = {"source": source(connection, request), "notes": notes(connection)} result = write_note(connection, request, review, **kwargs) return {"connection": kwargs.get("actor", "A"), "request": deepcopy(request), "review": deepcopy(review), "before": before, "result": result, "after": {"source": source(connection, request), "notes": notes(connection)}} def run_case(path, name, broken): a, b = connect(path), connect(path) try: initialize(a) request, review = deepcopy(REQUEST), approved(REQUEST) timeline, attempts = [], [] options = {"check_before_lock": broken, "timeline": timeline} def changed(): change_source(b) timeline.append({"connection": "B", "event": "source_revision_2_committed"}) if name == "source-changed-before-write": attempts.append(attempt(a, request, review, before_transaction=changed, **options)) elif name == "changed-reviewed-content": request["statement"] = "final_credit_recorded" attempts.append(attempt(a, request, review, **options)) elif name in ("identical-retry-after-source-change", "changed-operation"): attempts.append(attempt(a, request, review, **options)) changed() # A separate connection reads the stored receipt, not process memory. if name == "changed-operation": request.update(case_revision=2, source_revision=2, statement="final_credit_recorded") review = approved(request) attempts.append(attempt(b, request, review, actor="B", **options)) elif name == "rollback-and-retry": def fail(): raise InjectedFailure("Failure after INSERT, before COMMIT") attempts.append(attempt(a, request, review, after_insert=fail, **options)) attempts.append(attempt(b, request, review, actor="B", **options)) elif name == "overlapping-deliveries": waiting = [] def overlap(): waiting.append(attempt(b, request, review, actor="B", **options)) attempts.append(attempt(a, request, review, after_insert=overlap, **options)) attempts.extend(waiting) attempts.append(attempt(b, request, review, actor="B", **options)) else: attempts.append(attempt(a, request, review, **options)) return {"name": name, "timeline": timeline, "attempts": attempts, "final_notes": notes(b)} finally: a.close() b.close() def run(broken=False): names = ["supported-note", "source-changed-before-write", "changed-reviewed-content", "identical-retry-after-source-change", "changed-operation", "rollback-and-retry", "overlapping-deliveries"] with TemporaryDirectory(prefix="fde-financial-handoff-") as folder: return {"synthetic": True, "execution": "real scratch SQLite; two connections; controlled transaction overlap", "mode": "check-before-lock" if broken else "guarded", "cases": [run_case(Path(folder) / f"{index}.db", name, broken) for index, name in enumerate(names)]} if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--json", action="store_true", help="Print complete inputs, rows and actual transaction traces") parser.add_argument("--check-before-lock", action="store_true", help="Deliberately use a case read from before the write transaction") args = parser.parse_args() report = run(args.check_before_lock) if args.json: print(json.dumps(report, indent=2)) else: print(report["execution"] + " · " + report["mode"]) for case in report["cases"]: print(case["name"] + ": " + " → ".join(item["result"]["outcome"] for item in case["attempts"]) + f"; committed notes={len(case['final_notes'])}") print("Use --json to inspect each attempt and the chronological timeline. All records are invented.")