#!/usr/bin/env python3 """LockedIn Labs: synthetic local receiver transaction, Python 3.11+. Run `test` or `demo`. No packages, network, model, real dispatch or credentials. The database row IS the exercise's effect; it does not represent an external send. """ import argparse import hashlib import json import os from pathlib import Path import re import sqlite3 import subprocess import sys import tempfile import unittest PAYLOAD = {"work_order": "WO-1", "note": "Inspect the isolation valve."} GRANTS = {"north": {"WO-1"}, "south": {"WO-9"}} class Receiver: def __init__(self, path): # Explicit SQL transaction control works on Python 3.11 and later. self.db = sqlite3.connect(path, isolation_level=None, timeout=5) self.db.execute("PRAGMA synchronous=FULL") self.db.execute("CREATE TABLE IF NOT EXISTS effects (id INTEGER PRIMARY KEY, tenant TEXT NOT NULL, body TEXT NOT NULL)") self.db.execute("CREATE TABLE IF NOT EXISTS receipts (tenant TEXT NOT NULL, request_key TEXT NOT NULL, digest TEXT NOT NULL, effect_id INTEGER NOT NULL REFERENCES effects(id), PRIMARY KEY(tenant,request_key))") self.db.execute("PRAGMA foreign_keys=ON") def close(self): self.db.close() def receive(self, tenant, key, payload, fault=None): """tenant stands for authenticated host context; this lab has no login.""" if not isinstance(key, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", key): raise ValueError("invalid_key") if not isinstance(payload, dict) or set(payload) != {"work_order", "note"}: raise ValueError("invalid_payload") if not all(isinstance(v, str) and 0 < len(v) <= 512 for v in payload.values()): raise ValueError("invalid_payload") # Revalidate current scope even when returning an earlier receipt. if payload["work_order"] not in GRANTS.get(tenant, set()): raise PermissionError("work_order_unavailable") if fault not in (None, "before-commit", "after-commit"): raise ValueError("invalid_fault") body = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False) digest = hashlib.sha256(body.encode("utf-8")).hexdigest() self.db.execute("BEGIN IMMEDIATE") try: existing = self.db.execute("SELECT digest,effect_id FROM receipts WHERE tenant=? AND request_key=?", (tenant, key)).fetchone() if existing: if existing[0] != digest: raise ValueError("key_payload_conflict") effect_id = existing[1] else: effect_id = self.db.execute("INSERT INTO effects(tenant,body) VALUES (?,?)", (tenant, body)).lastrowid self.db.execute("INSERT INTO receipts VALUES (?,?,?,?)", (tenant, key, digest, effect_id)) if fault == "before-commit": os._exit(24) # Only used in this script's owned child process. self.db.execute("COMMIT") except BaseException: if self.db.in_transaction: self.db.execute("ROLLBACK") raise if fault == "after-commit": os._exit(23) # Durable effect + receipt, but no response to caller. return {"tenant": tenant, "key": key, "effect_id": effect_id, "digest": digest} def inspect(self): return {"effects": self.db.execute("SELECT COUNT(*) FROM effects").fetchone()[0], "receipts": self.db.execute("SELECT COUNT(*) FROM receipts").fetchone()[0], "integrity": self.db.execute("PRAGMA integrity_check").fetchone()[0]} def child(path, mode): return subprocess.run([sys.executable, str(Path(__file__).resolve()), "_child", str(path), mode], capture_output=True, text=True, timeout=10) def broken_receive(path, crash=False): """Teaching mutant: effect and receipt commit separately. Never a repair.""" r = Receiver(path) body = json.dumps(PAYLOAD, sort_keys=True, separators=(",", ":")) existing = r.db.execute("SELECT effect_id FROM receipts WHERE tenant='north' AND request_key='R1'").fetchone() if not existing: effect_id = r.db.execute("INSERT INTO effects(tenant,body) VALUES ('north',?)", (body,)).lastrowid if crash: os._exit(25) digest = hashlib.sha256(body.encode()).hexdigest() r.db.execute("INSERT INTO receipts VALUES ('north','R1',?,?)", (digest, effect_id)) r.close() def demo(): report = {"basis": "actual local child exits and SQLite rows; synthetic records only", "scenarios": {}} with tempfile.TemporaryDirectory(prefix="fde-receiver-") as directory: for mode, expected_exit in (("before-commit", 24), ("after-commit", 23), ("broken", 25)): path = Path(directory) / (mode + ".sqlite") r = Receiver(path) r.close() interrupted = child(path, mode) if interrupted.returncode != expected_exit: raise RuntimeError("child_did_not_reach_named_boundary") r = Receiver(path) before = r.inspect() if mode == "broken": r.close() broken_receive(path) r = Receiver(path) receipt = None else: receipt = r.receive("north", "R1", PAYLOAD) report["scenarios"][mode] = {"child_exit": interrupted.returncode, "response_received": bool(interrupted.stdout), "before_retry": before, "after_retry": r.inspect(), "retry_receipt": receipt} r.close() return report class Tests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix="fde-receiver-test-") self.path = Path(self.tmp.name) / "receiver.sqlite" self.r = Receiver(self.path) def tearDown(self): self.r.close() self.tmp.cleanup() def test_success_is_useful_and_replay_returns_same_receipt(self): first = self.r.receive("north", "R1", PAYLOAD) self.assertEqual(first, self.r.receive("north", "R1", dict(reversed(list(PAYLOAD.items()))))) self.assertEqual(self.r.inspect(), {"effects": 1, "receipts": 1, "integrity": "ok"}) def test_changed_payload_conflicts_without_second_effect(self): self.r.receive("north", "R1", PAYLOAD) with self.assertRaisesRegex(ValueError, "key_payload_conflict"): self.r.receive("north", "R1", {**PAYLOAD, "note": "A different action"}) self.assertEqual(self.r.inspect()["effects"], 1) def test_foreign_work_order_cannot_write(self): with self.assertRaisesRegex(PermissionError, "work_order_unavailable"): self.r.receive("south", "R1", PAYLOAD) self.assertEqual(self.r.inspect()["effects"], 0) def test_same_key_is_scoped_to_each_tenant(self): a = self.r.receive("north", "R1", PAYLOAD) b = self.r.receive("south", "R1", {**PAYLOAD, "work_order": "WO-9"}) self.assertNotEqual(a["effect_id"], b["effect_id"]) self.assertEqual(self.r.inspect()["receipts"], 2) def test_replay_rechecks_current_permission(self): self.r.receive("north", "R1", PAYLOAD) GRANTS["north"].remove("WO-1") try: with self.assertRaises(PermissionError): self.r.receive("north", "R1", PAYLOAD) finally: GRANTS["north"].add("WO-1") self.assertEqual(self.r.inspect()["effects"], 1) def test_invalid_arguments_never_write(self): for key, payload in (("", PAYLOAD), ("R/1", PAYLOAD), ("R1", {**PAYLOAD, "tenant": "south"}), ("R1", {**PAYLOAD, "note": "x" * 513})): with self.assertRaises(ValueError): self.r.receive("north", key, payload) self.assertEqual(self.r.inspect()["effects"], 0) def test_process_exit_before_commit_rolls_back_both_rows(self): self.assertEqual(child(self.path, "before-commit").returncode, 24) self.assertEqual(self.r.inspect(), {"effects": 0, "receipts": 0, "integrity": "ok"}) self.r.receive("north", "R1", PAYLOAD) self.assertEqual(self.r.inspect()["effects"], 1) def test_lost_response_after_commit_replays_existing_effect(self): result = child(self.path, "after-commit") self.assertEqual((result.returncode, result.stdout), (23, "")) self.assertEqual(self.r.inspect()["effects"], 1) receipt = self.r.receive("north", "R1", PAYLOAD) self.assertEqual(receipt["effect_id"], 1) self.assertEqual(self.r.inspect()["effects"], 1) def test_distinct_processes_racing_same_key_share_one_effect(self): args = [sys.executable, str(Path(__file__).resolve()), "_child", str(self.path), "normal"] children = [subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) for _ in range(2)] try: results = [p.communicate(timeout=10) for p in children] self.assertEqual([p.returncode for p in children], [0, 0]) self.assertEqual(json.loads(results[0][0]), json.loads(results[1][0])) self.assertEqual(self.r.inspect()["effects"], 1) finally: for p in children: if p.poll() is None: p.kill() p.wait(timeout=5) def test_broken_separate_commits_duplicate_on_retry(self): self.assertEqual(child(self.path, "broken").returncode, 25) self.assertEqual(self.r.inspect(), {"effects": 1, "receipts": 0, "integrity": "ok"}) broken_receive(self.path) self.assertEqual(self.r.inspect(), {"effects": 2, "receipts": 1, "integrity": "ok"}) def test_new_key_means_new_effect_not_business_deduplication(self): self.r.receive("north", "R1", PAYLOAD) self.r.receive("north", "R2", PAYLOAD) self.assertEqual(self.r.inspect()["effects"], 2) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=["test", "demo", "_child"]) parser.add_argument("path", nargs="?") parser.add_argument("mode", nargs="?", choices=["normal", "before-commit", "after-commit", "broken"]) args = parser.parse_args() if args.command == "test": result = unittest.TextTestRunner(verbosity=2).run(unittest.defaultTestLoader.loadTestsFromTestCase(Tests)) return 0 if result.wasSuccessful() else 1 if args.command == "demo": print(json.dumps(demo(), indent=2)) return 0 if not args.path or not args.mode: parser.error("_child requires a local database path and mode") if args.mode == "broken": broken_receive(args.path, crash=True) else: r = Receiver(args.path) try: print(json.dumps(r.receive("north", "R1", PAYLOAD, None if args.mode == "normal" else args.mode))) finally: r.close() return 0 if __name__ == "__main__": sys.exit(main())