Generic retry helper with configurable backoff

Calls to remote services fail now and then for reasons that go away on their own: a dropped connection, a busy server, a timeout. Write a generic retry helper that runs a function, returns its result on success, and retries it on failure according to a configurable policy.

What to implement

def retry(fn: Callable[[], T], policy: RetryPolicy) -> T: ...
async def retry_async(fn: Callable[[], Awaitable[T]], policy: RetryPolicy) -> T: ...

RetryPolicy must support:

• max_attempts — total number of calls, including the first one (at least 1).
• A wait before each retry, chosen by strategy:
  • fixed: the same wait every time;
  • linear: base * n before retry n;
  • exponential: base * multiplier^(n-1) before retry n.
• max_delay — an upper limit on any single wait.
• Optional jitter, so many clients do not retry at the same moment.
• Which errors are retryable. A non-retryable error is raised immediately.

Rules

• Return the first successful result.
• Do not wait after the final attempt.
• If every attempt fails, raise the error from the last attempt.
• The async version must not block the event loop while waiting (use asyncio.sleep).
• Make the sleep function injectable so tests do not really wait.

Example 1

fn raises ConnectionError on calls 1 and 2 and returns "ok" on call 3. Policy: max_attempts=4, exponential, base 0.5 s, multiplier 2.

Result: "ok". Waits recorded: [0.5, 1.0]. fn was called 3 times.

Example 2

fn always raises TimeoutError. Policy: max_attempts=5, exponential, base 1 s, multiplier 3, max_delay=10.

Waits recorded: [1, 3, 9, 10], then the fifth TimeoutError is raised.

Follow-ups you may be asked

Add an overall deadline, retry based on the returned value, honor a server's Retry-After hint, and expose the helper as a decorator.