# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Inject loss, then catch it.

A toy CDC pipeline (source table, change list, sink table) that silently drops
a small share of changes; a per-day count + checksum comparison that catches
it; and a simulated crash that shows why the cursor must be committed with
the state, not before or after it.
"""
import argparse, hashlib, json, random
from collections import defaultdict

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--days", type=int, default=30, help="Days of orders.")
p.add_argument("--orders-per-day", type=int, default=2_000, help="Orders created per day.")
p.add_argument("--drop", type=float, default=0.001, help="Share of changes silently dropped before they are applied.")
p.add_argument("--batch", type=int, default=500, help="Changes per batch for the crash experiment.")
p.add_argument("--seed", type=int, default=0, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
random.seed(args.seed)

# Source and its change list: inserts, then status updates and a few deletes
source, changes, lsn = {}, [], 0
for day in range(args.days):
    for i in range(args.orders_per_day):
        oid = day * args.orders_per_day + i
        row = {"day": day, "status": "placed", "total": round(random.uniform(5, 500), 2)}
        source[oid] = row; lsn += 1; changes.append((lsn, "I", oid, dict(row)))
    for oid in random.sample(sorted(k for k in source if source[k]["day"] == day), args.orders_per_day // 4):
        source[oid]["status"] = "shipped"; lsn += 1; changes.append((lsn, "U", oid, dict(source[oid])))
    for oid in random.sample(sorted(k for k in source if source[k]["day"] == day), args.orders_per_day // 100):
        del source[oid]; lsn += 1; changes.append((lsn, "D", oid, None))

def apply(sink, change):
    _, op, oid, row = change
    if op == "D": sink.pop(oid, None)
    else: sink[oid] = row                              # an upsert keyed by order id

# 1. Drop a small share of changes: nothing fails, the run finishes
sink, dropped = {}, 0
for c in changes:
    if random.random() < args.drop: dropped += 1; continue
    apply(sink, c)
print(f"{len(changes):,} changes, {dropped} silently dropped ({dropped / len(changes):.2%}); pipeline finished with no error")

# 2. Reconcile: count + checksum per day on both sides
def summary(table):
    out = defaultdict(lambda: [0, 0])
    for oid, r in table.items():
        out[r["day"]][0] += 1
        out[r["day"]][1] ^= int(hashlib.md5(f"{oid}|{r['status']}|{r['total']}".encode()).hexdigest()[:15], 16)
    return out
src_s, snk_s = summary(source), summary(sink)
bad_days = [d for d in range(args.days) if src_s[d] != snk_s[d]]
row_diffs = sum(1 for k in set(source) | set(sink) if source.get(k) != sink.get(k))
print(f"per-day count + checksum: {len(bad_days)} of {args.days} days differ ({row_diffs} rows wrong in the sink)")

# 3. A crash in the middle of a batch, under three commit orders
def crash_run(order, sink_kind):
    """Apply the change list in batches; the process crashes once, half-way through batch 3, and restarts
    from the last committed cursor. order: when the cursor is committed relative to the sink state."""
    sink = {} if sink_kind == "upsert" else []
    def put(c):
        if sink_kind == "upsert": apply(sink, c)
        else: sink.append(c[0])                         # append-only: one row per change applied
    cursor, crashed = 0, False
    while cursor < len(changes):
        start = cursor
        batch = changes[start: start + args.batch]
        crash_here = not crashed and start == 3 * args.batch
        if order == "cursor first": cursor = start + len(batch)          # cursor committed, then the state
        staged = []
        for c in batch[: len(batch) // 2 if crash_here else len(batch)]:
            if order == "together": staged.append(c)                    # state and cursor in one transaction
            else: put(c)
        if crash_here:                                                  # the process dies here
            crashed = True
            continue                                                    # restart from the committed cursor
        for c in staged: put(c)
        if order != "cursor first": cursor = start + len(batch)        # cursor committed after (or with) the state
    if sink_kind == "upsert":
        wrong = sum(1 for k in source if sink.get(k) != source[k]) + sum(1 for k in sink if k not in source)
        return {"lost_or_wrong": wrong, "duplicates": 0}
    return {"lost_or_wrong": len(changes) - len(set(sink)), "duplicates": len(sink) - len(set(sink))}

crash = []
for order in ("cursor first", "cursor last", "together"):
    for kind in ("upsert", "append"):
        r = crash_run(order, kind); crash.append({"cursor": order, "sink": kind, **r})
        print(f"crash, {order:<13} {kind:>6} sink: {r['lost_or_wrong']:>4} lost or wrong, {r['duplicates']:>4} duplicates")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"crash": crash, "summary": [{
            "changes": len(changes), "dropped": dropped, "bad_days": len(bad_days), "days": args.days, "row_diffs": row_diffs}]}}, f)