#!/usr/bin/env python3 """Isolated reconciliation/rollback teaching model. Python 3.9+, stdlib only. Run: python3 data-migration-reconciliation.py Fail the initial acceptance gate: add --inject-mismatch (exit status 1). Inputs are the synthetic rows below. All SQLite databases are in memory. No files, network, credentials, production databases, or infrastructure are used. This models one writer, an ordered change journal and reversible transformations; it does not implement real CDC, concurrency fencing, crash recovery, or RTO/RPO. """ import argparse from decimal import Decimal import sqlite3 SOURCE_ROWS = [(101, "12.50", "A"), (102, "7.50", "C"), (103, "20.00", "A")] EXPECTED_INITIAL = {101: (1250, "active"), 102: (750, "closed"), 103: (2000, "active")} EXPECTED_RECOVERY = {101: (1400, "active"), 102: (750, "closed"), 104: (900, "active")} STATUS_FORWARD = {"A": "active", "C": "closed"} STATUS_REVERSE = {"active": "A", "closed": "C"} def canonical_source(db): result = {} for key, dollars, status in db.execute("SELECT id, dollars, status FROM orders"): cents = Decimal(dollars) * 100 if not cents.is_finite() or cents != cents.to_integral_value(): raise ValueError(f"Non-reversible amount for id={key}") result[key] = (int(cents), STATUS_FORWARD[status]) return result def target_rows(db): return {key: (cents, status) for key, cents, status in db.execute( "SELECT id, cents, status FROM orders" )} def reconcile(expected, actual): return [f"id={key}: expected={expected.get(key)}, actual={actual.get(key)}" for key in sorted(expected.keys() | actual.keys()) if expected.get(key) != actual.get(key)] def require(condition, message): if not condition: raise RuntimeError(message) def replay(source, target): # Apply a journal entry and its checkpoint in one local transaction. Replaying # the same sequence again is harmless in this single-writer model. entries = target.execute("SELECT seq, op, id, cents, status FROM changes ORDER BY seq").fetchall() with source: for seq, op, key, cents, status in entries: if source.execute("SELECT 1 FROM applied WHERE seq = ?", (seq,)).fetchone(): continue if op == "delete": source.execute("DELETE FROM orders WHERE id = ?", (key,)) else: dollars = format(Decimal(cents) / 100, ".2f") source.execute( "INSERT INTO orders VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE " "SET dollars=excluded.dollars, status=excluded.status", (key, dollars, STATUS_REVERSE[status]), ) source.execute("INSERT INTO applied VALUES (?)", (seq,)) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--inject-mismatch", action="store_true") args = parser.parse_args() source = sqlite3.connect(":memory:") target = sqlite3.connect(":memory:") try: source.executescript(""" CREATE TABLE orders (id INTEGER PRIMARY KEY, dollars TEXT NOT NULL, status TEXT NOT NULL CHECK (status IN ('A','C'))); CREATE TABLE applied (seq INTEGER PRIMARY KEY); """) with source: source.executemany("INSERT INTO orders VALUES (?, ?, ?)", SOURCE_ROWS) target.executescript(""" CREATE TABLE orders (id INTEGER PRIMARY KEY, cents INTEGER NOT NULL CHECK (typeof(cents) = 'integer'), status TEXT NOT NULL CHECK (status IN ('active','closed'))); """) transformed = canonical_source(source) require(transformed == EXPECTED_INITIAL, "Transformation contract failed") with target: target.executemany("INSERT INTO orders VALUES (?, ?, ?)", [(key, *values) for key, values in transformed.items()]) if args.inject_mismatch: target.execute("UPDATE orders SET cents = 751 WHERE id = 102") mismatches = reconcile(EXPECTED_INITIAL, target_rows(target)) if mismatches: print("FAIL initial reconciliation: " + "; ".join(mismatches)) print("STOP: cutover blocked; escalate to the data acceptance owner.") return 1 print("PASS initial reconciliation: rows=3, total_cents=4000") # Install the journal after the initial load. These triggers record target # inserts, updates and deletions atomically with each local transaction. target.executescript(""" CREATE TABLE changes (seq INTEGER PRIMARY KEY AUTOINCREMENT, op TEXT, id INTEGER, cents INTEGER, status TEXT); CREATE TRIGGER journal_insert AFTER INSERT ON orders BEGIN INSERT INTO changes(op,id,cents,status) VALUES ('insert',NEW.id,NEW.cents,NEW.status); END; CREATE TRIGGER journal_update AFTER UPDATE ON orders BEGIN INSERT INTO changes(op,id,cents,status) VALUES ('update',NEW.id,NEW.cents,NEW.status); END; CREATE TRIGGER journal_delete AFTER DELETE ON orders BEGIN INSERT INTO changes(op,id,cents,status) VALUES ('delete',OLD.id,NULL,NULL); END; """) # Source writers are inactive; only this target connection accepts writes. with target: target.execute("UPDATE orders SET cents=1400 WHERE id=101") target.execute("INSERT INTO orders VALUES (104,900,'active')") target.execute("DELETE FROM orders WHERE id=103") # Local write fence: forbid DML on the target before fixing the boundary. target.execute("PRAGMA query_only = ON") try: target.execute("INSERT INTO orders VALUES (105,1,'active')") except sqlite3.OperationalError as error: require("readonly" in str(error).lower(), f"Unexpected fence error: {error}") else: raise RuntimeError("Target write fence failed") boundary = target.execute("SELECT MAX(seq) FROM changes").fetchone()[0] require(boundary == 3, "Unexpected replay boundary") require(target_rows(target) == EXPECTED_RECOVERY, "Post-cutover fixture failed") stale = reconcile(EXPECTED_RECOVERY, canonical_source(source)) require(len(stale) == 3, "Stale source should differ at three keys") print("PASS stale-source check: differences=3 (updated=101, deleted=103, new=104)") replay(source, target) require(not reconcile(EXPECTED_RECOVERY, canonical_source(source)), "Recovery reconciliation failed") replay(source, target) require(canonical_source(source) == EXPECTED_RECOVERY, "Replay was not idempotent") require(source.execute("SELECT COUNT(*) FROM applied").fetchone()[0] == 3, "Checkpoint count failed") print("PASS fenced rollback reconciliation: journal_boundary=3, rows=3, total_cents=3050") print("PASS repeated replay: applied_events=3, differences=0") print("SCOPE: in-memory toy model; no production RTO/RPO or platform recovery claim.") return 0 finally: source.close() target.close() if __name__ == "__main__": raise SystemExit(main())