Problem 86 · Concurrency & Python

Retry Decorator with Exponential Backoff and Jitter

MediumClassic DSA & systems

01Problem

Write a decorator that retries a function on specified exceptions using capped exponential backoff with full jitter, re-raising after max attempts.

02Key insight

Full jitter (sleep = random(0, min(cap, base*2^n))) decorrelates clients and avoids thundering herds after an outage. Only retry idempotent operations and retryable errors; inject sleep/random for testability.

03Solution

solution.pypython30 lines

import functools
import random
import time
from typing import Callable, ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")


def retry(max_attempts: int = 5, base_delay: float = 0.1, max_delay: float = 10.0,
          retry_on: tuple[type[BaseException], ...] = (Exception,),
          sleep: Callable[[float], None] = time.sleep,
          rng: Callable[[], float] = random.random) -> Callable[[Callable[P, R]], Callable[P, R]]:
    if max_attempts < 1:
        raise ValueError("max_attempts must be >= 1")

    def decorator(fn: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(fn)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            for attempt in range(max_attempts):
                try:
                    return fn(*args, **kwargs)
                except retry_on:
                    if attempt == max_attempts - 1:
                        raise
                    cap = min(max_delay, base_delay * (2 ** attempt))
                    sleep(rng() * cap)  # full jitter
            raise AssertionError("unreachable")
        return wrapper
    return decorator

Try it first.

TimeO(max_attempts)

SpaceO(1)

tests.py · 3 passing

tests.pypython28 lines

sleeps = []
state = {"n": 0}
@retry(max_attempts=4, base_delay=0.1, retry_on=(ConnectionError,), sleep=sleeps.append, rng=lambda: 1.0)
def flaky():
    state["n"] += 1
    if state["n"] < 3:
        raise ConnectionError("down")
    return "ok"
assert flaky() == "ok" and sleeps == [0.1, 0.2]
sleeps.clear()
@retry(max_attempts=3, base_delay=1, max_delay=1.5, sleep=sleeps.append, rng=lambda: 1.0)
def always_fail():
    raise TimeoutError()
try:
    always_fail(); assert False
except TimeoutError:
    assert sleeps == [1, 1.5]
@retry(retry_on=(ConnectionError,), sleep=sleeps.append)
def bad_input():
    raise ValueError()
try:
    bad_input(); assert False
except ValueError:
    assert len(sleeps) == 2  # non-retryable: no extra sleeps
try:
    retry(max_attempts=0); assert False
except ValueError:
    pass

04Staff-level follow-up

Retries amplify load during an outage - what else do you need? A retry budget/circuit breaker (e.g. retries capped at 10% of requests), deadline propagation so you never retry past the caller's timeout, and idempotency keys for non-idempotent writes.