A DLQ that does not become an outage amplifier

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""A DLQ that does not become a second data loss.

10,000 records, 0.5% malformed; after record 5,000 the downstream times out
on every call for 30 seconds. A naive policy (three attempts, then the
dead-letter queue) against a classified one (transient errors retry with
backoff and never dead-letter; malformed records dead-letter at once; pause
and alert when more than 2% of the last 1,000 failed).
"""
import argparse, json, random

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--records", type=int, default=10_000, help="Records.")
p.add_argument("--malformed", type=float, default=0.005, help="Share of malformed records.")
p.add_argument("--rate", type=float, default=100, help="Records per second offered.")
p.add_argument("--outage", type=float, default=30, help="Seconds the downstream times out, starting at record 5,000.")
p.add_argument("--fail-ms", type=float, default=1.0, help="Milliseconds a failing call takes to raise its timeout error.")
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)
bad = set(random.sample(range(args.records), int(args.records * args.malformed)))
start = 5_000 / args.rate                                         # the outage starts when record 5,000 is due


def run(policy):
    t, dlq, applied, recent, paused_s = 0.0, [], [], [], 0.0
    for r in range(args.records):
        t = max(t, r / args.rate)
        while True:
            down = start <= t < start + args.outage
            if r in bad:
                if policy == "classified": dlq.append(r); break        # malformed: no retry can fix it
                for attempt in range(3): t += 0.05
                dlq.append(r); break
            if not down: applied.append(r); recent = (recent + [True])[-1000:]; break
            if policy == "naive":
                t += 3 * args.fail_ms / 1000                             # three quick failures, then give up
                dlq.append(r); recent = (recent + [False])[-1000:]; break
            # classified: a transient error; back off and retry the same record (order kept)
            recent = (recent + [False])[-1000:]
            if recent.count(False) > 20: paused_s += 1.0               # > 2% of the last 1,000: pause, alert
            t += 1.0
    out_of_order = sum(1 for a, b in zip(applied, applied[1:]) if b < a)
    late = len([x for x in dlq if x not in bad])
    return {"policy": policy, "dlq": len(dlq), "good_records_in_dlq": late, "applied": len(applied),
            "applied_out_of_order": len(dlq) - len(bad) if policy == "naive" else out_of_order}


rows = [run("naive"), run("classified")]
for r in rows:
    print(f"{r['policy']:<10} DLQ {r['dlq']:>5}  (good records in it: {r['good_records_in_dlq']:>5})   applied {r['applied']:>6}")

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