Checkpoint interval optimum

# /// script
# requires-python = ">=3.10"
# dependencies = ["numpy"]
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""The checkpoint interval optimum.

A 30-day job with Poisson failures (MTBF 1 day) and a 30 s checkpoint; on a
failure it loses the work since the last checkpoint. Wasted time across
intervals from 2 minutes to 6 hours, against C/I + I/(2 MTBF) and its minimum
I* = sqrt(2 C MTBF). Then the same intervals when the MTBF is really 4 days.
"""
import argparse, json, math
import numpy as np

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--days", type=float, default=30, help="Simulated days of useful work.")
p.add_argument("--mtbf-h", type=float, default=24, help="Mean time between failures, hours.")
p.add_argument("--cost-s", type=float, default=30, help="Seconds per checkpoint.")
p.add_argument("--trials", type=int, default=20, help="Simulated jobs per interval.")
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()
rng = np.random.default_rng(args.seed)
C = args.cost_s; work = args.days * 86400
intervals = np.unique(np.geomspace(120, 6 * 3600, 18).astype(int))

def overhead(I, mtbf):
    total = 0.0
    for _ in range(args.trials):
        done, clock = 0.0, 0.0
        next_fail = rng.exponential(mtbf)
        while done < work:
            seg = min(I, work - done)
            if clock + seg + C <= next_fail:
                clock += seg + C; done += seg
            else:
                clock = next_fail; next_fail = clock + rng.exponential(mtbf)       # the segment is lost
        total += clock
    return (total / args.trials - work) / (total / args.trials)

rows = []
for real in (args.mtbf_h, args.mtbf_h * 4):
    for I in intervals:
        rows.append({"mtbf_h": real, "interval_s": int(I), "waste_pct": overhead(I, real * 3600) * 100,
                     "formula_pct": (C / I + I / (2 * real * 3600)) * 100})
best = math.sqrt(2 * C * args.mtbf_h * 3600)
at_best = min((r for r in rows if r["mtbf_h"] == args.mtbf_h), key=lambda r: abs(r["interval_s"] - best))
same_I_4x = min((r for r in rows if r["mtbf_h"] == args.mtbf_h * 4), key=lambda r: abs(r["interval_s"] - best))
best4 = math.sqrt(2 * C * args.mtbf_h * 4 * 3600)
opt_4x = min((r for r in rows if r["mtbf_h"] == args.mtbf_h * 4), key=lambda r: r["waste_pct"])
print(f"I* = sqrt(2 C MTBF) = {best:.0f} s ({best / 60:.0f} min); simulated waste near it {at_best['waste_pct']:.2f}% "
      f"(formula {at_best['formula_pct']:.2f}%)")
print(f"if the MTBF is really 4x longer: waste at the same interval {same_I_4x['waste_pct']:.2f}%, best possible "
      f"{opt_4x['waste_pct']:.2f}% at {opt_4x['interval_s']:.0f} s (I* would be {best4:.0f} s)")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"intervals": rows, "summary": [{"best_s": best, "waste_at_best": at_best["waste_pct"], "formula_at_best": at_best["formula_pct"],
                   "waste_wrong_mtbf": same_I_4x["waste_pct"], "waste_opt_4x": opt_4x["waste_pct"], "best4_s": best4}]}}, f)