# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Check that your dead man's switch works.

A scheduler (its own process) runs a job every few seconds; the job's final
step sends a heartbeat over HTTP to a checker outside the scheduler, and the
scheduler also runs its own SLA check on the job. After a while the scheduler
is stopped (SIGSTOP: it hangs, as a stuck scheduler does). The checker alerts
when no heartbeat arrives within the deadline; the scheduler's own SLA alert
runs inside the scheduler, so it cannot fire. Times both from the stop.
"""
import argparse, json, os, signal, subprocess, sys, threading, time
from http.server import BaseHTTPRequestHandler, HTTPServer

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--every", type=float, default=2.0, help="Job interval (s).")
p.add_argument("--deadline", type=float, default=5.0, help="Heartbeat deadline (s).")
p.add_argument("--stop-after", type=float, default=10.0, help="Seconds before the scheduler is stopped.")
p.add_argument("--watch", type=float, default=20.0, help="Seconds to watch after the stop.")
p.add_argument("--port", type=int, default=58301, help="Checker port.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()

SCHEDULER = r"""
import sys, time, urllib.request
every, deadline, port = float(sys.argv[1]), float(sys.argv[2]), int(sys.argv[3])
last_ok = time.time()
while True:
    time.sleep(every)
    time.sleep(0.1)                                              # the job
    last_ok = time.time()
    urllib.request.urlopen(f"http://127.0.0.1:{port}/beat").read()   # final step: heartbeat
    if time.time() - last_ok > deadline:                         # the scheduler's own SLA check
        print("SLA ALERT", flush=True)
"""
beats = []
class H(BaseHTTPRequestHandler):
    def do_GET(self): beats.append(time.time()); self.send_response(200); self.end_headers(); self.wfile.write(b"ok")
    def log_message(self, *a): pass
srv = HTTPServer(("127.0.0.1", args.port), H); threading.Thread(target=srv.serve_forever, daemon=True).start()
sched = subprocess.Popen([sys.executable, "-u", "-c", SCHEDULER, str(args.every), str(args.deadline), str(args.port)], stdout=subprocess.PIPE, text=True)
sla = []; threading.Thread(target=lambda: [sla.append(time.time()) for line in sched.stdout if "SLA ALERT" in line], daemon=True).start()
t0 = time.time(); alert = None; stopped = None
while time.time() - t0 < args.stop_after + args.watch:
    now = time.time()
    if stopped is None and now - t0 >= args.stop_after: os.kill(sched.pid, signal.SIGSTOP); stopped = now
    last = beats[-1] if beats else t0
    if stopped and alert is None and now - last > args.deadline: alert = now
    time.sleep(0.05)
os.kill(sched.pid, signal.SIGKILL); sched.wait(); srv.shutdown()
rows = [{"monitor": "heartbeat checker (outside the scheduler)", "fired": "yes" if alert else "no", "seconds_after_stop": alert - stopped if alert else None},
        {"monitor": "scheduler's own SLA alert", "fired": "yes" if any(t > stopped for t in sla) else "no", "seconds_after_stop": None}]
summary = {"beats_before_stop": sum(1 for b in beats if b < stopped), "beats_after_stop": sum(1 for b in beats if b >= stopped), "deadline_s": args.deadline}
print(f"{summary['beats_before_stop']} heartbeats before the stop, {summary['beats_after_stop']} after")
for r in rows: print(f"{r['monitor']:<44} " + (f"alerted {r['seconds_after_stop']:.1f} s after the scheduler stopped" if r["seconds_after_stop"] else f"fired: {r['fired']} in {args.watch:g} s"))

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