Implementing Exponential Backoff for Secretary of State APIs #

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import random
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from enum import Enum
from typing import Optional

import httpx

# Structured JSON logging — every line is a parseable audit/observability event.
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("ingestion.backoff")


class Resolution(str, Enum):
    SUCCESS = "success"        # portal returned a final 2xx
    ESCALATED = "escalated"    # backoff would breach the grace window — divert to priority channel
    FALLOUT = "fallout"        # retries exhausted — routed to dead-letter queue


@dataclass(frozen=True)
class BackoffPolicy:
    base_delay: float = 1.0    # first-retry floor, seconds
    cap: float = 60.0          # per-wait ceiling; never sleep longer than this on one attempt
    multiplier: float = 3.0    # decorrelated-jitter growth factor (prev_wait * multiplier)
    max_attempts: int = 5      # bounds the loop so a degraded portal cannot burn the budget


@dataclass(frozen=True)
class AttemptRecord:
    attempt: int
    timestamp_utc: str
    http_status: Optional[int]
    wait_applied: float
    retry_after_honored: bool
    prev_hash: str             # chains to the previous attempt for tamper evidence
    this_hash: str

    @staticmethod
    def chain(prev_hash: str, attempt: int, status: Optional[int],
              wait: float, honored: bool) -> "AttemptRecord":
        ts = datetime.now(timezone.utc).isoformat()
        body = json.dumps(
            {"prev": prev_hash, "attempt": attempt, "status": status,
             "wait": round(wait, 3), "honored": honored, "ts": ts},
            sort_keys=True,
        )
        digest = hashlib.sha256(body.encode("utf-8")).hexdigest()
        return AttemptRecord(attempt, ts, status, round(wait, 3), honored, prev_hash, digest)


def parse_retry_after(value: Optional[str]) -> Optional[float]:
    """A portal's Retry-After is authoritative; honor it over any computed schedule.

    The header is legally either delta-seconds OR an HTTP-date (RFC 7231 §7.1.3).
    Parse both; an unparseable header falls back to None so the caller computes its own wait.
    """
    if not value:
        return None
    if value.strip().isdigit():
        return float(value.strip())
    try:
        when = parsedate_to_datetime(value)
        if when.tzinfo is None:
            when = when.replace(tzinfo=timezone.utc)
        return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())
    except (TypeError, ValueError):
        return None


class BackoffEngine:
    """Deadline-bounded exponential backoff with decorrelated jitter and an audit chain."""

    # Statuses the engine will wait-and-retry; everything else is the caller's concern.
    RETRYABLE = frozenset({429, 502, 503, 504})

    def __init__(self, client: httpx.AsyncClient, policy: BackoffPolicy | None = None) -> None:
        self._client = client
        self._policy = policy or BackoffPolicy()
        self._trail: list[AttemptRecord] = []

    def _next_wait(self, prev_wait: float, retry_after: Optional[float]) -> tuple[float, bool]:
        # A portal-supplied Retry-After always overrides our computed schedule.
        if retry_after is not None:
            return min(retry_after, self._policy.cap), True
        # Decorrelated jitter: uniform across [base, prev_wait * multiplier], capped.
        upper = max(self._policy.base_delay, prev_wait * self._policy.multiplier)
        wait = min(self._policy.cap, random.uniform(self._policy.base_delay, upper))
        return wait, False

    async def run(self, request: httpx.Request, grace_seconds: float) -> Resolution:
        """Execute `request` with backoff, never spending more than `grace_seconds` total."""
        prev_hash = "GENESIS"
        prev_wait = self._policy.base_delay
        spent = 0.0

        for attempt in range(1, self._policy.max_attempts + 1):
            response = await self._client.send(request)
            if response.status_code not in self.RETRYABLE:
                response.raise_for_status()  # 4xx terminal errors surface to the caller
                self._append(prev_hash, attempt, response.status_code, 0.0, False)
                return Resolution.SUCCESS

            retry_after = parse_retry_after(response.headers.get("Retry-After"))
            wait, honored = self._next_wait(prev_wait, retry_after)

            # COMPLIANCE GUARD: if waiting would consume the cure window, do not sleep —
            # escalate the obligation to the priority channel instead of risking a penalty.
            if spent + wait > grace_seconds:
                self._append(prev_hash, attempt, response.status_code, wait, honored)
                logger.warning(json.dumps({
                    "event": "deadline_escalation", "attempt": attempt,
                    "would_spend": round(spent + wait, 2), "grace": grace_seconds,
                }))
                return Resolution.ESCALATED

            record = self._append(prev_hash, attempt, response.status_code, wait, honored)
            prev_hash, prev_wait, spent = record.this_hash, wait, spent + wait
            await asyncio.sleep(wait)

        # Retries exhausted within budget: never silently drop — preserve for forensic replay.
        logger.error(json.dumps({"event": "backoff_exhausted",
                                 "attempts": self._policy.max_attempts}))
        return Resolution.FALLOUT

    def _append(self, prev_hash: str, attempt: int, status: Optional[int],
                wait: float, honored: bool) -> AttemptRecord:
        record = AttemptRecord.chain(prev_hash, attempt, status, wait, honored)
        self._trail.append(record)
        logger.info(json.dumps({
            "event": "backoff_attempt", "attempt": attempt, "status": status,
            "wait": record.wait_applied, "retry_after_honored": honored,
            "hash": record.this_hash[:12],
        }))
        return record