Generic retry helper with configurable backoff

The core is a loop: call the function, and if it succeeds return its value.
If it raises, decide whether this error is worth another try and whether any
attempts are left. If so, wait and try again. If not, re-raise the last error.
The design questions are about keeping that loop small while letting callers
choose the policy.

Start simple. A first version takes fn and max_attempts, catches
Exception, and loops. Say out loud that max_attempts counts the first call
too (3 attempts = 1 call + 2 retries), because off-by-one here is the most
common bug.

Then separate policy from mechanism. The things that vary are:

• how long to wait before retry n (fixed, linear, exponential, capped),
• whether to add jitter (so many clients do not retry in lockstep),
• which errors are retryable (a timeout is; a validation error is not).

Put these in a small RetryPolicy object with one method, delay(n), and a
predicate should_retry(exc). The runner only asks the policy questions. This
keeps the loop the same for sync and async callers. The async version differs
only in await fn() and await asyncio.sleep(...), so blocking sleep never
stalls the event loop.

Inject the clock. Pass sleep in as a parameter. Tests then record the
delays instead of really sleeping, so they run instantly and can assert the
exact schedule (for example 1, 2, 4, 8, 8 with a cap of 8).

Invariants:

• fn is called at most max_attempts times.
• A non-retryable error is raised at once, without sleeping.
• After the final attempt the runner never sleeps; it raises the last error
  with its original traceback (a bare raise).
• delay(n) is never above max_delay and never negative.

import asyncio
import random
import time
from dataclasses import dataclass, field
from typing import Awaitable, Callable, Optional, Tuple, Type, TypeVar

T = TypeVar("T")


@dataclass
class RetryPolicy:
    max_attempts: int = 3                 # includes the first call
    base_delay: float = 1.0               # seconds before the first retry
    strategy: str = "exponential"         # "fixed" | "linear" | "exponential"
    multiplier: float = 2.0               # growth factor for exponential
    max_delay: float = float("inf")       # cap on any single wait
    jitter: bool = False                  # full jitter: uniform(0, delay)
    retry_on: Tuple[Type[BaseException], ...] = (Exception,)
    rng: random.Random = field(default_factory=random.Random)

    def __post_init__(self) -> None:
        if self.max_attempts < 1:
            raise ValueError("max_attempts must be >= 1")
        if self.base_delay < 0:
            raise ValueError("base_delay must be >= 0")
        if self.strategy not in ("fixed", "linear", "exponential"):
            raise ValueError(f"unknown strategy {self.strategy!r}")

    def delay(self, retry_number: int) -> float:
        """Wait before retry `retry_number` (1 = first retry)."""
        if self.strategy == "fixed":
            d = self.base_delay
        elif self.strategy == "linear":
            d = self.base_delay * retry_number
        else:
            d = self.base_delay * self.multiplier ** (retry_number - 1)
        d = min(d, self.max_delay)
        return self.rng.uniform(0, d) if self.jitter else d

    def should_retry(self, exc: BaseException) -> bool:
        return isinstance(exc, self.retry_on)


def retry(
    fn: Callable[[], T],
    policy: Optional[RetryPolicy] = None,
    sleep: Callable[[float], None] = time.sleep,
) -> T:
    policy = policy or RetryPolicy()
    for attempt in range(1, policy.max_attempts + 1):
        try:
            return fn()
        except BaseException as exc:
            if not policy.should_retry(exc) or attempt == policy.max_attempts:
                raise
            sleep(policy.delay(attempt))
    raise AssertionError("unreachable")


async def retry_async(
    fn: Callable[[], Awaitable[T]],
    policy: Optional[RetryPolicy] = None,
    sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
) -> T:
    policy = policy or RetryPolicy()
    for attempt in range(1, policy.max_attempts + 1):
        try:
            return await fn()
        except BaseException as exc:
            if not policy.should_retry(exc) or attempt == policy.max_attempts:
                raise
            await sleep(policy.delay(attempt))
    raise AssertionError("unreachable")


def flaky(failures: int, value: str = "ok", exc: Type[Exception] = ConnectionError):
    """Helper for tests: fails `failures` times, then returns `value`."""
    calls = {"n": 0}

    def fn() -> str:
        calls["n"] += 1
        if calls["n"] <= failures:
            raise exc(f"attempt {calls['n']} failed")
        return value

    return fn, calls

• retry / retry_async: O(k) calls of fn and O(k) policy work, where
  k ≤ max_attempts. Extra space O(1).
• RetryPolicy.delay: O(1) time and space.
• Total wall-clock wait is the sum of the delays. For exponential with no cap
  it is base * (m^(k-1) - 1) / (m - 1), which is why a cap matters.

• Succeeds on the first call: retry(lambda: 5) → 5, no sleeps.
• Fails twice, then succeeds, max_attempts=3, exponential, base 1 → returns
  "ok", sleeps [1, 2], fn called 3 times.
• Always fails, max_attempts=3 → raises the third ConnectionError
  ("attempt 3 failed"), sleeps [1, 2] (no sleep after the last attempt).
• max_attempts=1 → one call, error raised, no sleeps.
• Non-retryable error (ValueError with retry_on=(ConnectionError,)) →
  raised after one call, no sleeps.
• Exponential, base 1, cap 8, six attempts all failing → sleeps
  [1, 2, 4, 8, 8].
• Fixed base 0.5 → sleeps [0.5, 0.5, ...]; linear base 1 → [1, 2, 3, ...].
• Jitter on, cap 8 → every delay is in [0, capped delay].
• max_attempts=0 → ValueError when the policy is built.
• Async: coroutine fails once then returns 42 → 42, async sleeps [1].

fn, calls = flaky(2)
slept = []
assert retry(fn, RetryPolicy(max_attempts=3), slept.append) == "ok"
assert slept == [1.0, 2.0] and calls["n"] == 3

1. Async callers. Run many retried operations at once with
   asyncio.gather. Use retry_async so waiting never blocks the event
   loop. Cancelling the task must not be retried: asyncio.CancelledError is a
   BaseException (Python 3.8+), so the default retry_on=(Exception,)
   re-raises it at once. Never widen retry_on to BaseException.
2. Overall deadline and per-attempt timeout. Add deadline (total
   seconds). Before sleeping, check now + delay < deadline; if not, raise
   the last error. Wrap each attempt in asyncio.wait_for(fn(), timeout) so
   a hung call counts as a failure.
3. Retry on results, honor server hints. Add retry_if_result(value) for
   APIs that return error codes instead of raising, and read a Retry-After
   value from the error (HTTP 429/503) to override delay(n).
4. Decorator form and hooks. @retryable(policy) wraps any function and
   forwards *args, **kwargs. Add on_retry(attempt, exc, delay) for logging
   and metrics, and keep all attempt errors (ExceptionGroup or a custom
   RetryError with .attempts) when callers want the full history.
5. Protect the downstream service. A retry budget (for example at most
   10% extra traffic) or a circuit breaker stops retries from turning an
   outage into a retry storm. Only retry idempotent operations, or send an
   idempotency key with each call.

Mid-level (L4)

• Working loop with the correct attempt count and re-raises the last error.
• Supports at least fixed and exponential backoff.
• Tests success, eventual success, and all-fail cases.

Senior (L5)

• Separates policy (delay, cap, jitter, which errors) from the runner.
• Async version that uses asyncio.sleep, plus an injected clock for fast tests.
• Raises non-retryable errors at once and never sleeps after the last attempt.
• Explains why jitter and a cap are needed.

Staff (L6+)

• Talks about idempotency, deadlines, Retry-After, retry budgets and
  circuit breakers, and cancellation semantics.
• Considers layered retries (client, gateway, and server all retrying
  multiplies load) and where retries belong.
• Proposes observability: attempt counts, final failure reasons, latency added by retries.