#!/usr/bin/env python3 """Pure, offline retry-budget planner. It makes no HTTP requests or sleeps. The caller must recheck before dispatch. One shared retry owner must enforce target-wide pauses, source limits, concurrency, and atomic token consumption. """ import argparse import json import math from datetime import datetime, timezone from email.utils import parsedate_to_datetime class InputError(ValueError): pass def _number(value, name, positive=False): if isinstance(value, bool) or not isinstance(value, (int, float)): raise InputError(name + " must be a number") try: value = float(value) except (OverflowError, ValueError): raise InputError(name + " is out of range") if not math.isfinite(value) or value < 0 or (positive and value == 0): raise InputError(name + " is out of range") return value def _integer(value, name, minimum): if isinstance(value, bool) or not isinstance(value, int) or value < minimum: raise InputError(name + " must be an integer >= " + str(minimum)) return value def _aware(value, name): if not isinstance(value, datetime) or value.tzinfo is None or value.utcoffset() is None: raise InputError(name + " must be an aware datetime") return value def _retry_after(value, observed_at): if value is None: return 0.0, None if not isinstance(value, str) or not value.strip(): return None, "unknown_retry_after" text = value.strip() if text.isascii() and text.isdecimal(): try: seconds = float(text) except (OverflowError, ValueError): return None, "unknown_retry_after" return (seconds, None) if math.isfinite(seconds) else (None, "unknown_retry_after") try: when = parsedate_to_datetime(text) if when.tzinfo is None or when.utcoffset() is None: return None, "unknown_retry_after" observed_at = _aware(observed_at, "observed_at") return max(0.0, (when - observed_at).total_seconds()), None except (TypeError, ValueError, IndexError, OverflowError, InputError): return None, "unknown_retry_after" def _upper(base, cap, attempts_used): if base == 0 or cap == 0: return 0.0 exponent = attempts_used - 1 if base >= cap: return cap limit = (math.log(cap) - math.log(base)) / math.log(2) if exponent >= limit: return cap try: return min(cap, math.ldexp(base, exponent)) except OverflowError: return cap def _stop(reason, tokens, server_min=0.0): return {"decision": "STOP", "reason": reason, "delay_seconds": server_min, "next_attempt": None, "observed_remaining_retry_tokens": tokens, "retry_token_cost_if_dispatched": 0} def plan_next_attempt(*, method, outcome, attempts_used, max_attempts, elapsed_seconds, deadline_seconds, per_attempt_timeout_seconds, remaining_retry_tokens, retry_allowed, access_allowed, jitter_draw, base_backoff_seconds, backoff_cap_seconds, retry_after=None, observed_at=None): """Plan one retry; token projections do not reserve or consume a token.""" if not isinstance(method, str): raise InputError("method must be a string") if not isinstance(outcome, (int, str)) or isinstance(outcome, bool): raise InputError("outcome must be an HTTP status or transport_timeout") attempts_used = _integer(attempts_used, "attempts_used", 1) max_attempts = _integer(max_attempts, "max_attempts", 1) tokens = _integer(remaining_retry_tokens, "remaining_retry_tokens", 0) elapsed = _number(elapsed_seconds, "elapsed_seconds") deadline = _number(deadline_seconds, "deadline_seconds") timeout = _number(per_attempt_timeout_seconds, "per_attempt_timeout_seconds", True) draw = _number(jitter_draw, "jitter_draw") base = _number(base_backoff_seconds, "base_backoff_seconds") cap = _number(backoff_cap_seconds, "backoff_cap_seconds") if draw > 1: raise InputError("jitter_draw must be in 0..1") if not isinstance(retry_allowed, bool) or not isinstance(access_allowed, bool): raise InputError("retry_allowed and access_allowed must be booleans") server_min, error = _retry_after(retry_after, observed_at) if error: return _stop(error, tokens) if method not in ("GET", "HEAD"): return _stop("method_not_retryable", tokens, server_min) status = str(outcome).lower() if status == "429": return _stop("rate_limited_coordinate_target_wide_pause", tokens, server_min) if status in ("403", "407", "challenge", "invalid_content") or status == "200": return _stop("non_retryable_" + status, tokens, server_min) if status not in ("502", "503", "504", "transport_timeout"): return _stop("unrecognized_outcome", tokens, server_min) if not retry_allowed: return _stop("operator_retry_not_allowed", tokens, server_min) if not access_allowed: return _stop("access_not_allowed", tokens, server_min) if attempts_used >= max_attempts: return _stop("max_attempts_reached", tokens, server_min) if tokens == 0: return _stop("shared_retry_tokens_exhausted", tokens, server_min) jitter = draw * _upper(base, cap, attempts_used) delay = max(jitter, server_min) remaining = deadline - elapsed if delay + timeout >= remaining: return _stop("deadline_insufficient_for_next_attempt", tokens, server_min) return {"decision": "WAIT_THEN_RECHECK", "reason": "eligible_retry", "delay_seconds": delay, "next_attempt": attempts_used + 1, "remaining_deadline_slack_seconds": remaining - delay - timeout, "observed_remaining_retry_tokens": tokens, "retry_token_cost_if_dispatched": 1, "projected_remaining_retry_tokens": tokens - 1} def _case(**overrides): values = dict(method="GET", outcome=503, attempts_used=1, max_attempts=3, elapsed_seconds=1, deadline_seconds=20, per_attempt_timeout_seconds=3, remaining_retry_tokens=2, retry_allowed=True, access_allowed=True, jitter_draw=0.5, base_backoff_seconds=2, backoff_cap_seconds=4, retry_after=None, observed_at=None) values.update(overrides) return plan_next_attempt(**values) def self_test(): assert _case()["decision"] == "WAIT_THEN_RECHECK" assert _case(elapsed_seconds=16)["reason"] == "deadline_insufficient_for_next_attempt" assert _case(attempts_used=3)["reason"] == "max_attempts_reached" assert _case(remaining_retry_tokens=0)["reason"] == "shared_retry_tokens_exhausted" for outcome in (502, 503, 504, "transport_timeout"): assert _case(outcome=outcome)["decision"] == "WAIT_THEN_RECHECK" assert _case(outcome=429, retry_after="3")["decision"] == "STOP" for value in (200, 403, 407, "challenge", "invalid_content", "unknown"): assert _case(outcome=value)["decision"] == "STOP" assert _case(method="POST")["reason"] == "method_not_retryable" now = datetime(2026, 1, 1, tzinfo=timezone.utc) assert _case(retry_after="3")["delay_seconds"] >= 3 assert _case(retry_after="Thu, 01 Jan 2026 00:00:05 GMT", observed_at=now)["delay_seconds"] >= 5 assert _case(retry_after="Thu, 01 Jan 2025 00:00:00 GMT", observed_at=now)["delay_seconds"] == 1 assert _case(retry_after="-1")["reason"] == "unknown_retry_after" assert _case(retry_after="nonsense")["reason"] == "unknown_retry_after" assert _case(base_backoff_seconds=100, backoff_cap_seconds=2, jitter_draw=1)["delay_seconds"] == 2 assert _case(base_backoff_seconds=1e-308, backoff_cap_seconds=1e308, attempts_used=1400, max_attempts=2000)["delay_seconds"] <= 1e308 assert _case(base_backoff_seconds=1e-308, backoff_cap_seconds=1e308, attempts_used=10 ** 100, max_attempts=10 ** 100 + 1)["delay_seconds"] <= 1e308 for bad in (float("nan"), True, 10 ** 10000): try: _case(elapsed_seconds=bad) except InputError: pass else: raise AssertionError("bad input accepted") for key, bad in (("jitter_draw", True), ("max_attempts", True), ("method", 1)): try: _case(**{key: bad}) except InputError: pass else: raise AssertionError("bad input accepted") return {"self_test": "passed", "synthetic": True} def main(): parser = argparse.ArgumentParser(description="Offline retry-budget planner") parser.add_argument("--self-test", action="store_true") parser.add_argument("--demo", action="store_true", help="print deterministic synthetic plan") args = parser.parse_args() if args.self_test: print(json.dumps(self_test(), sort_keys=True)) return now = datetime(2026, 1, 1, tzinfo=timezone.utc) print(json.dumps(_case(retry_after="Thu, 01 Jan 2026 00:00:05 GMT", observed_at=now), sort_keys=True)) if __name__ == "__main__": main()