"""Consumer scope, payload conflicts, and lost deduplication history. Run: python outbox-crash-replay.py Python standard library only. All database files live in one temporary directory. The owned child process intentionally exits with code 71 after consumer commit. No network, cloud account, external queue, or real payment is used. """ from contextlib import closing, contextmanager import json import os from pathlib import Path import platform import sqlite3 import subprocess import sys import tempfile def connect(path): connection = sqlite3.connect(path, isolation_level=None) connection.execute("PRAGMA foreign_keys = ON") return connection @contextmanager def transaction(connection): connection.execute("BEGIN IMMEDIATE") try: yield except BaseException: connection.execute("ROLLBACK") raise else: connection.execute("COMMIT") def initialize(producer, consumer): with closing(connect(producer)) as connection: connection.executescript(""" CREATE TABLE orders ( order_id TEXT PRIMARY KEY NOT NULL, amount INTEGER NOT NULL CHECK (amount > 0) ); CREATE TABLE outbox ( event_id TEXT PRIMARY KEY NOT NULL, order_id TEXT NOT NULL REFERENCES orders(order_id), payload TEXT NOT NULL, sent INTEGER NOT NULL DEFAULT 0 CHECK (sent IN (0, 1)) ); """) with closing(connect(consumer)) as connection: connection.executescript(""" CREATE TABLE inbox ( consumer_id TEXT NOT NULL, event_id TEXT NOT NULL, payload TEXT NOT NULL, PRIMARY KEY (consumer_id, event_id) ); CREATE TABLE effects ( effect_id INTEGER PRIMARY KEY, consumer_id TEXT NOT NULL, event_id TEXT NOT NULL, amount INTEGER NOT NULL ); """) def canonical(payload): # This fixed demo schema contains only strings and integers. This is not a # cross-language canonical JSON standard or a general schema validator. return json.dumps(payload, sort_keys=True, separators=(",", ":")) def create_order(producer, suffix, fail=False): event = { "event_id": "order-created:" + suffix, "payload": {"order_id": "order:" + suffix, "amount": 100}, } with closing(connect(producer)) as connection, transaction(connection): connection.execute("INSERT INTO orders VALUES (?, ?)", (event["payload"]["order_id"], 100)) connection.execute( "INSERT INTO outbox(event_id, order_id, payload) VALUES (?, ?, ?)", (event["event_id"], event["payload"]["order_id"], canonical(event["payload"])), ) if fail: raise RuntimeError("before producer commit") return event def consume(consumer, subscriber, event, fail=False): payload = canonical(event["payload"]) with closing(connect(consumer)) as connection, transaction(connection): previous = connection.execute( "SELECT payload FROM inbox WHERE consumer_id=? AND event_id=?", (subscriber, event["event_id"]), ).fetchone() if previous is not None: if previous[0] != payload: raise ValueError("same event ID with different payload") return False connection.execute("INSERT INTO inbox VALUES (?, ?, ?)", (subscriber, event["event_id"], payload)) # Deliberately no unique key on effects: inbox provides this lesson's # deduplication. Deleting inbox history will expose the failure. connection.execute( "INSERT INTO effects(consumer_id, event_id, amount) VALUES (?, ?, ?)", (subscriber, event["event_id"], event["payload"]["amount"]), ) if fail: raise RuntimeError("before consumer commit") return True def scalar(path, sql, parameters=()): with closing(connect(path)) as connection: return connection.execute(sql, parameters).fetchone()[0] def effect_count(consumer, subscriber, event_id): return scalar(consumer, "SELECT COUNT(*) FROM effects WHERE consumer_id=? AND event_id=?", (subscriber, event_id)) def child_delivery(producer, consumer): with closing(connect(producer)) as connection: row = connection.execute( "SELECT event_id, payload FROM outbox WHERE sent=0 ORDER BY event_id LIMIT 1" ).fetchone() event = {"event_id": row[0], "payload": json.loads(row[1])} assert consume(consumer, "audit", event) # The consumer transaction has committed and closed. The producer has not # yet marked the row sent. No cleanup or exception handler runs after this. os._exit(71) def run(): print(f"Python {platform.python_version()} / SQLite {sqlite3.sqlite_version}") with tempfile.TemporaryDirectory(prefix="qinghuan-outbox-") as temp: root = Path(temp) producer, consumer = root / "producer.sqlite", root / "consumer.sqlite" initialize(producer, consumer) try: create_order(producer, "rollback", fail=True) except RuntimeError: pass else: raise AssertionError("producer failure was not triggered") assert scalar(producer, "SELECT COUNT(*) FROM orders") == 0 assert scalar(producer, "SELECT COUNT(*) FROM outbox") == 0 print("producer rollback: orders=0 outbox=0") event = create_order(producer, "001") child = subprocess.run( [sys.executable, str(Path(__file__).resolve()), "--owned-delivery-child", str(producer), str(consumer)], check=False, capture_output=True, text=True, timeout=15, ) assert child.returncode == 71, (child.returncode, child.stderr) assert scalar(producer, "SELECT COUNT(*) FROM outbox WHERE sent=0") == 1 assert effect_count(consumer, "audit", event["event_id"]) == 1 print("child exit=71: pending=1 audit_effects=1") assert consume(consumer, "audit", event) is False with closing(connect(producer)) as connection, transaction(connection): connection.execute("UPDATE outbox SET sent=1 WHERE event_id=?", (event["event_id"],)) assert effect_count(consumer, "audit", event["event_id"]) == 1 print("replay with history: skipped; audit_effects=1") assert consume(consumer, "search", event) is True assert consume(consumer, "search", event) is False assert effect_count(consumer, "search", event["event_id"]) == 1 print("independent subscriber: audit_effects=1 search_effects=1") conflicting = {"event_id": event["event_id"], "payload": {**event["payload"], "amount": 999}} try: consume(consumer, "audit", conflicting) except ValueError: pass else: raise AssertionError("conflicting payload was accepted") assert effect_count(consumer, "audit", event["event_id"]) == 1 assert scalar(consumer, "SELECT SUM(amount) FROM effects WHERE consumer_id='audit'") == 100 print("same ID, changed payload: rejected; audit_total=100") second = create_order(producer, "002") try: consume(consumer, "audit", second, fail=True) except RuntimeError: pass else: raise AssertionError("consumer failure was not triggered") assert scalar(consumer, "SELECT COUNT(*) FROM inbox WHERE event_id=?", (second["event_id"],)) == 0 assert effect_count(consumer, "audit", second["event_id"]) == 0 assert consume(consumer, "audit", second) is True print("consumer rollback: inbox=0 effects=0; retry applied once") with closing(connect(consumer)) as connection, transaction(connection): connection.execute("DELETE FROM inbox WHERE consumer_id=? AND event_id=?", ("audit", event["event_id"])) assert consume(consumer, "audit", event) is True assert effect_count(consumer, "audit", event["event_id"]) == 2 assert effect_count(consumer, "search", event["event_id"]) == 1 print("delete audit history, replay old event: audit_effects=2 search_effects=1") print("PASS: rollback, crash replay, subscriber scope, payload conflict, retention") print("temporary databases removed: True") if __name__ == "__main__": if not __debug__: raise SystemExit("Assertions are required. Run without -O or -OO.") if len(sys.argv) == 4 and sys.argv[1] == "--owned-delivery-child": child_delivery(Path(sys.argv[2]), Path(sys.argv[3])) elif len(sys.argv) == 1: run() else: raise SystemExit("Run without arguments: python outbox-crash-replay.py")