#!/usr/bin/env python3 """ The Report Engine — a nine-task DAG with three gates that builds the morning report by itself, and refuses to publish a wrong one. Part of "The Engine Room" on simranjaiswal.in. python3 engine.py # reference run, seed 42, writes results.json python3 engine.py --seed 7 --crm429 0.35 --sheets429 0.25 --schema 0.15 --late 0.30 --rowcount 0.12 Everything here is deterministic. Thirty mornings are simulated from the printed rules below with a tiny 32-bit PRNG (mulberry32) that the in-browser engine on the case page implements bit-for-bit, so the page's reference run reproduces this file's results.json exactly. No real pipeline data is involved. Stdlib only. """ import argparse, json, math, os # ----------------------------------------------------------------------------- deterministic randomness class RNG: """mulberry32 — identical to the JavaScript port on the case page.""" M = 0xFFFFFFFF def __init__(self, seed): self.a = seed & self.M def random(self): self.a = (self.a + 0x6D2B79F5) & self.M t = self.a t = ((t ^ (t >> 15)) * (1 | t)) & self.M t = ((t + (((t ^ (t >> 7)) * (61 | t)) & self.M)) & self.M) ^ t return ((t ^ (t >> 14)) & self.M) / 4294967296 def normal(self, mu=0.0, sd=1.0): u = self.random() or 1e-12; v = self.random() return mu + sd * math.sqrt(-2 * math.log(u)) * math.cos(6.283185307179586 * v) # ----------------------------------------------------------------------------- the DAG MORNINGS = 30 RUN_AT = 6 * 3600 # 06:00, seconds since midnight # name, base duration in seconds, what it waits for. The three gates run once per extract. TASKS = [ ("extract_mysql", 40, []), ("extract_crm", 55, []), ("extract_zoho", 35, []), ("gate_schema", 3, ["extract_*"]), ("gate_freshness", 4, ["gate_schema"]), ("gate_rowcount", 3, ["gate_freshness"]), ("transform_orders", 25, ["gate_rowcount:mysql", "gate_rowcount:crm"]), ("transform_collections", 20, ["gate_rowcount:zoho", "gate_rowcount:mysql"]), ("publish_sheets", 30, ["transform_orders", "transform_collections"]), ("notify", 2, ["publish_sheets"]), ] BASE = {t[0]: t[1] for t in TASKS} EXTRACTS = ["mysql", "crm", "zoho"] ROWS = {"mysql": 18400, "crm": 6200, "zoho": 3900} # typical daily row counts per source BACKOFF = [2, 4, 8, 16] # seconds before retry 1, 2, 3, 4 POLL_S = 300 # the freshness gate polls every 5 min MAX_POLLS = 9 # ... for up to 45 min ROWCOUNT_TOL = 0.20 # day-on-day move that blocks publish ANOMALY = 0.65 # a row-count anomaly is one extract at 65% of normal MANUAL_MIN = 108 # the manual era: 9 h/week over 5 mornings MANUAL_ERR = 1 / 12 # ... with a 1-in-12 chance of a paste error that ships REVIEW_MIN = 8 # after: 40 min/week of review over 5 mornings BLOCK_FIX_MIN = 25 # after: a blocked morning costs a human fix STALE_READ_MIN = 5 # after: a stale morning costs reading the alert def hms(t): t = RUN_AT + t return "%02d:%02d:%02d" % (t // 3600, (t // 60) % 60, t % 60) # ----------------------------------------------------------------------------- one morning def simulate_morning(rng, day, p, last_good): """ Draw order (the browser port mirrors it exactly): 1. task jitter, one normal(0, 0.10) per task in TASKS order; 2. row counts, one normal(0, 0.03) per extract in EXTRACTS order; 3. incidents in the order crm429, sheets429, schema, late, rowcount — one uniform each, plus the extra draws written next to each one. Then the DAG is scheduled: the three extracts start together at 06:00, each extract's three gates run in sequence behind it, the transforms start when their gates are done, publish waits for both. """ dur = {} for name, base, _ in TASKS: z = rng.normal(0, 0.10) dur[name] = max(1, int(round(base * (1 + z)))) rows = {} for e in EXTRACTS: rows[e] = int(round(ROWS[e] * (1 + rng.normal(0, 0.03)))) inc = {"crm429": 0, "sheets429": 0, "schema": None, "late": None, "rowcount": None} if rng.random() < p["crm429"]: inc["crm429"] = 1 + int(rng.random() * 3) # succeeds on retry k if rng.random() < p["sheets429"]: inc["sheets429"] = 1 + int(rng.random() * 3) if rng.random() < p["schema"]: inc["schema"] = EXTRACTS[int(rng.random() * 3)] # a renamed column if rng.random() < p["late"]: e = EXTRACTS[int(rng.random() * 3)] lands = rng.random() < 0.7 polls = 1 + int(rng.random() * MAX_POLLS) if lands else MAX_POLLS inc["late"] = {"extract": e, "lands": lands, "polls": polls} if rng.random() < p["rowcount"]: e = EXTRACTS[int(rng.random() * 3)] inc["rowcount"] = e rows[e] = int(rows[e] * ANOMALY) seg, ev = [], [] def add(task, start, d, kind, note=""): seg.append({"task": task, "start": start, "end": start + d, "kind": kind, "note": note}); return start + d def log(t, text): ev.append((t, len(ev), text)) gate_end = {} failed = [] retries = 0 stale = False late_seen = None # the late-source incident only counts if its gate actually ran for e in EXTRACTS: name = "extract_" + e t = 0 log(t, name + " started") if e == "crm" and inc["crm429"]: for i in range(inc["crm429"]): t = add(name, t, BASE[name] // 5, "fail", "HTTP 429") log(t, "%s HTTP 429 · retry %d in %d s" % (name, i + 1, BACKOFF[i])) t = add(name, t, BACKOFF[i], "wait", "backoff") retries += 1 t = add(name, t, dur[name], "run", "%d rows" % rows[e]) log(t, "%s done · %d rows" % (name, rows[e])) # gate 1: schema if inc["schema"] == e: t = add("gate_schema", t, dur["gate_schema"], "fail", e) log(t, "gate_schema FAILED on %s · a column was renamed upstream · run stopped · alert sent · yesterday's sheet stays" % e) failed.append({"gate": "schema", "extract": e}) continue t = add("gate_schema", t, dur["gate_schema"], "pass", e) log(t, "gate_schema ok · %s" % e) # gate 2: freshness if inc["late"] and inc["late"]["extract"] == e: L = inc["late"] late_seen = L log(t, "gate_freshness · %s has no rows after midnight · polling every 5 min, up to 45" % e) t = add("gate_freshness", t, L["polls"] * POLL_S, "wait", "polling") if L["lands"]: log(t, "gate_freshness · %s landed after %d min · re-extracting" % (e, L["polls"] * 5)) t = add(name, t, dur[name], "run", "re-extract") t = add("gate_freshness", t, dur["gate_freshness"], "pass", e) log(t, "gate_freshness ok · %s" % e) else: stale = True t = add("gate_freshness", t, dur["gate_freshness"], "stale", e) log(t, "gate_freshness · %s still stale after 45 min · continuing with a STALE banner · alert sent" % e) else: t = add("gate_freshness", t, dur["gate_freshness"], "pass", e) log(t, "gate_freshness ok · %s" % e) # gate 3: row count vs the last count that passed last = last_good[e] pct = int((rows[e] - last) / last * 100) if abs(rows[e] - last) > ROWCOUNT_TOL * last: t = add("gate_rowcount", t, dur["gate_rowcount"], "fail", e) log(t, "gate_rowcount FAILED on %s · %d rows vs %d last published (%+d%%) · publish blocked · alert sent" % (e, rows[e], last, pct)) failed.append({"gate": "rowcount", "extract": e}) continue t = add("gate_rowcount", t, dur["gate_rowcount"], "pass", e) last_good[e] = rows[e] log(t, "gate_rowcount ok · %s · %d rows (%+d%% vs last)" % (e, rows[e], pct)) gate_end[e] = t blocked = len(failed) > 0 publish_s = None if not blocked: t1 = max(gate_end["mysql"], gate_end["crm"]) log(t1, "transform_orders started") t1 = add("transform_orders", t1, dur["transform_orders"], "run") log(t1, "transform_orders done") t2 = max(gate_end["zoho"], gate_end["mysql"]) log(t2, "transform_collections started") t2 = add("transform_collections", t2, dur["transform_collections"], "run") log(t2, "transform_collections done") t = max(t1, t2) log(t, "publish_sheets started" + (" · STALE banner on every tab" if stale else "")) if inc["sheets429"]: for i in range(inc["sheets429"]): t = add("publish_sheets", t, BASE["publish_sheets"] // 5, "fail", "HTTP 429") log(t, "publish_sheets HTTP 429 · retry %d in %d s" % (i + 1, BACKOFF[i])) t = add("publish_sheets", t, BACKOFF[i], "wait", "backoff") retries += 1 t = add("publish_sheets", t, dur["publish_sheets"], "run", "batch write") publish_s = t log(t, "publish_sheets done · one batch write per tab · sheet is live") tn = max(s["end"] for s in seg) log(tn, "notify started") tn = add("notify", tn, dur["notify"], "run") status = "BLOCKED" if blocked else "STALE" if stale else "RETRIED" if retries else "CLEAN" tail = {"BLOCKED": "no publish · owner paged", "STALE": "published with a STALE banner · owner told", "RETRIED": "published · %d retries absorbed" % retries, "CLEAN": "published"}[status] log(tn, "notify · %s · %s" % (status, tail)) incidents = [] if inc["crm429"]: incidents.append("CRM 429 x%d" % inc["crm429"]) if inc["sheets429"] and not blocked: incidents.append("Sheets 429 x%d" % inc["sheets429"]) if inc["schema"]: incidents.append("schema drift " + inc["schema"]) if late_seen: incidents.append("late " + late_seen["extract"] + (" (landed)" if late_seen["lands"] else " (stale)")) for f in failed: if f["gate"] == "rowcount": incidents.append("row-count anomaly " + f["extract"]) analyst_min = REVIEW_MIN + (BLOCK_FIX_MIN if blocked else 0) + (STALE_READ_MIN if stale else 0) ev.sort(key=lambda x: (x[0], x[1])) return {"day": day, "status": status, "publish_s": publish_s, "total_s": tn, "retries": retries, "incidents": incidents, "gate_failures": failed, "late": late_seen, "stale": stale, "analyst_min": analyst_min, "segments": seg, "log": ["%s %s" % (hms(t), text) for t, _, text in ev]} # ----------------------------------------------------------------------------- the month def run(seed=42, crm429=0.15, sheets429=0.10, schema=0.07, late=0.12, rowcount=0.05): p = {"crm429": crm429, "sheets429": sheets429, "schema": schema, "late": late, "rowcount": rowcount} rng = RNG(seed) last_good = dict(ROWS) mornings = [simulate_morning(rng, d, p, last_good) for d in range(MORNINGS)] # the manual era, on its own stream so the incident sliders do not move it brng = RNG(seed + 1000) manual_errors = [brng.random() < MANUAL_ERR for _ in range(MORNINGS)] for m, err in zip(mornings, manual_errors): m["manual_error"] = err st = {s: sum(1 for m in mornings if m["status"] == s) for s in ["CLEAN", "RETRIED", "STALE", "BLOCKED"]} catches = {"schema": 0, "late_landed": 0, "late_stale": 0, "rowcount": 0} for m in mornings: for f in m["gate_failures"]: catches[f["gate"]] += 1 if m["late"]: catches["late_landed" if m["late"]["lands"] else "late_stale"] += 1 published = [m["publish_s"] for m in mornings if m["publish_s"] is not None] published_sorted = sorted(published) n = len(published) mean_s = sum(published) // n if n else 0 p95_s = published_sorted[int(math.ceil(0.95 * n)) - 1] if n else 0 task_stats = [] for name, base, deps in TASKS: runs = [s["end"] - s["start"] for m in mornings for s in m["segments"] if s["task"] == name and s["kind"] in ("run", "pass", "stale")] task_stats.append({"task": name, "base_s": base, "waits_for": deps, "runs": len(runs), "mean_s": (sum(runs) // len(runs)) if runs else 0}) analyst_after = sum(m["analyst_min"] for m in mornings) manual_before = MANUAL_MIN * MORNINGS worst = 0 for i, m in enumerate(mornings): if m["total_s"] > mornings[worst]["total_s"]: worst = i totals = { "mornings": MORNINGS, "clean": st["CLEAN"], "retried": st["RETRIED"], "stale": st["STALE"], "blocked": st["BLOCKED"], "published": n, "wrong_numbers_prevented": catches["schema"] + catches["rowcount"], "gate_catches": catches, "retries_total": sum(m["retries"] for m in mornings), "mean_publish_s": mean_s, "p95_publish_s": p95_s, "analyst_min_after": analyst_after, "manual_min_before": manual_before, "minutes_returned": manual_before - analyst_after, "errors_shipped_manual": sum(1 for e in manual_errors if e), "errors_shipped_engine": 0, } return { "engine": "report-engine", "seed": seed, "run_at": "06:00", "params": {"seed": seed, "p_crm429": crm429, "p_sheets429": sheets429, "p_schema": schema, "p_late": late, "p_rowcount": rowcount}, "constants": {"mornings": MORNINGS, "backoff_s": BACKOFF, "poll_s": POLL_S, "max_polls": MAX_POLLS, "rowcount_tol": ROWCOUNT_TOL, "anomaly_factor": ANOMALY, "manual_min": MANUAL_MIN, "manual_err": "1/12", "review_min": REVIEW_MIN, "block_fix_min": BLOCK_FIX_MIN, "stale_read_min": STALE_READ_MIN, "typical_rows": ROWS}, "tasks": task_stats, "totals": totals, "worst_morning": worst, "mornings": [{k: m[k] for k in ("day", "status", "publish_s", "total_s", "retries", "incidents", "analyst_min", "manual_error")} for m in mornings], "worst": {"day": worst, "segments": mornings[worst]["segments"], "log": mornings[worst]["log"]}, } if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--seed", type=int, default=42) ap.add_argument("--crm429", type=float, default=0.15); ap.add_argument("--sheets429", type=float, default=0.10) ap.add_argument("--schema", type=float, default=0.07); ap.add_argument("--late", type=float, default=0.12) ap.add_argument("--rowcount", type=float, default=0.05) ap.add_argument("--out", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "results.json")) a = ap.parse_args() res = run(a.seed, a.crm429, a.sheets429, a.schema, a.late, a.rowcount) with open(a.out, "w") as f: json.dump(res, f, indent=1) T = res["totals"] print("30 mornings at 06:00 · seed %d" % a.seed) print(" clean %d · retried %d · stale %d · blocked %d (published %d)" % (T["clean"], T["retried"], T["stale"], T["blocked"], T["published"])) print(" gate catches: schema %d · late source landed %d · stale %d · row-count %d → wrong numbers prevented %d" % (T["gate_catches"]["schema"], T["gate_catches"]["late_landed"], T["gate_catches"]["late_stale"], T["gate_catches"]["rowcount"], T["wrong_numbers_prevented"])) print(" retries absorbed %d · minutes to publish mean %.1f, p95 %.1f" % (T["retries_total"], T["mean_publish_s"] / 60, T["p95_publish_s"] / 60)) print(" analyst minutes: %d by hand → %d with the engine · %.1f hours returned" % (T["manual_min_before"], T["analyst_min_after"], T["minutes_returned"] / 60)) print(" errors shipped: %d by hand → %d with the engine" % (T["errors_shipped_manual"], T["errors_shipped_engine"])) for t in res["tasks"]: print(" %-22s base %3d s · mean %3d s over %d runs" % (t["task"], t["base_s"], t["mean_s"], t["runs"])) print(" worst morning: day %d · %s · %s" % (res["worst_morning"], res["mornings"][res["worst_morning"]]["status"], ", ".join(res["mornings"][res["worst_morning"]]["incidents"]) or "no incidents")) print("wrote", a.out)