# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Dead-letter queue against a rate threshold.

100,000 records; 0.01% invalid for the first 50,000, then a schema change makes
40% invalid. (a) a pure dead-letter queue: bad records are set aside and
processing continues; (b) a windowed threshold: halt when more than 2% of the
last 1,000 records are invalid. Reports records dead-lettered, good records
processed, how long until the stream stops, and how much a human would see.
"""
import argparse, json, random
from collections import deque

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--records", type=int, default=100_000, help="Records.")
p.add_argument("--change-at", type=int, default=50_000, help="Record at which the schema change lands.")
p.add_argument("--bad-after", type=float, default=0.4, help="Invalid share after the change.")
p.add_argument("--threshold", type=float, default=0.02, help="Halt threshold (share of the window).")
p.add_argument("--window", type=int, default=1000, help="Window size.")
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)

stream = [random.random() < (0.0001 if i < args.change_at else args.bad_after) for i in range(args.records)]   # True = invalid
def run(policy):
    dlq = good = 0; window = deque(maxlen=args.window); halted = None
    for i, bad in enumerate(stream):
        window.append(bad)
        if policy == "threshold" and len(window) == args.window and sum(window) / args.window > args.threshold:
            halted = i; break
        if bad: dlq += 1
        else: good += 1
    held = args.records - (halted if halted is not None else args.records)
    return {"policy": "pure dead-letter queue" if policy == "dlq" else f"halt above {args.threshold:.0%} of last {args.window:,}",
            "dead_lettered": dlq, "good_processed": good, "halted_at": halted, "records_after_change_before_stop": (halted - args.change_at) if halted else None,
            "held_back": held, "bad_after_change_in_dlq": sum(stream[args.change_at:halted if halted else args.records]) }
rows = [run("dlq"), run("threshold")]
for r in rows: print(f"{r['policy']:<30} dead-lettered {r['dead_lettered']:>6,}  good processed {r['good_processed']:>6,}  " +
                      (f"halted {r['records_after_change_before_stop']:,} records after the change ({r['held_back']:,} held back)" if r["halted_at"] else "never halted"))

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"policies": rows}}, f)