Reconciliation count and duplicate-rate metric

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Reconciliation: a count check and a duplicate-rate metric.

Records per hour at the source and at the sink, with an alert above 0.1%
difference. A bug is introduced on purpose: an `except: pass` around the
sink write that swallows 1% of records. The error dashboard stays empty.
"""
import argparse, json, random

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--hours", type=int, default=24, help="Hours of traffic.")
p.add_argument("--per-hour", type=int, default=20_000, help="Records per hour.")
p.add_argument("--bug-from", type=int, default=12, help="Hour from which the bug swallows records.")
p.add_argument("--swallow", type=float, default=0.01, help="Share of records the bug swallows.")
p.add_argument("--dup-rate", type=float, default=0.002, help="Share of records redelivered (rejected as duplicates).")
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)

rows, errors = [], 0
for h in range(args.hours):
    source = sink = rejected = 0
    for _ in range(args.per_hour):
        source += 1
        if random.random() < args.dup_rate: rejected += 1          # a redelivery the sink's key rejects
        try:
            if h >= args.bug_from and random.random() < args.swallow: raise IOError("write failed")
            sink += 1
        except Exception:
            pass                                                     # the bug: no log, no metric, no error
    gap = (source - sink) / source
    rows.append({"hour": h, "source": source, "sink": sink, "gap_pct": gap * 100, "rejected_dups": rejected,
                 "alert": "yes" if gap > 0.001 else "no"})
alerts = sum(r["alert"] == "yes" for r in rows)
print(f"errors reported: {errors}")
print(f"hours with a count gap above 0.1%: {alerts} of {args.hours} (the bug ran from hour {args.bug_from})")
print(f"mean gap while the bug ran: {sum(r['gap_pct'] for r in rows[args.bug_from:]) / (args.hours - args.bug_from):.2f}%")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"hours": rows, "summary": [{"alerts": alerts, "errors": errors,
                   "gap_during_pct": sum(r['gap_pct'] for r in rows[args.bug_from:]) / (args.hours - args.bug_from)}]}}, f)