Add Retries With Backoff
The billing service calls a payment provider that times out under load, and every caller has its own retry loop. Build one shared helper with the assistant. The README has the rules. Accept, edit, or reject each suggestion, and keep the tests green.
backoff.py1 of 3
Files
backoff.py
Python · Editableimport functools
import random
import time
class RetryError(Exception):
"""Raised when retrying stops. The last error is chained as __cause__."""
def backoff_delay(retry_n, base_s, cap_s, rng):
"""Seconds to wait before retry number retry_n, counting from 0."""
raise NotImplementedError
def fits_deadline(start, now, delay, deadline_s):
"""True when a sleep of delay seconds starting at now ends by start + deadline_s."""
raise NotImplementedError
def retry(
fn,
attempts=3,
base_s=0.1,
cap_s=5.0,
retry_on=(ConnectionError, TimeoutError),
deadline_s=None,
sleep=time.sleep,
clock=time.monotonic,
rng=random.random,
):
"""Calls fn() and returns its result, retrying errors in retry_on with backoff."""
start = clock()
raise NotImplementedError
def retrying(**options):
"""The decorator form of retry. Takes the same keyword arguments."""
raise NotImplementedError
import pytest
from backoff import RetryError, backoff_delay, fits_deadline, retry, retrying
class Flaky:
"""Raises the given errors in order, then returns what it was called with."""
def __init__(self, *errors):
self.errors = list(errors)
self.calls = 0
def __call__(self, value="ok"):
self.calls += 1
if self.errors:
raise self.errors.pop(0)
return value
class FakeTime:
"""A clock, and a sleep that records each delay and advances the clock."""
def __init__(self):
self.now = 0.0
self.sleeps = []
def clock(self):
return self.now
def sleep(self, seconds):
self.sleeps.append(seconds)
self.now += seconds
def make(**options):
t = FakeTime()
settings = {"base_s": 0.25, "cap_s": 1.0, "sleep": t.sleep, "clock": t.clock, "rng": lambda: 1.0}
settings.update(options)
return t, settings
def timeouts(n):
return [TimeoutError("timed out") for _ in range(n)]
def test_returns_on_first_success():
t, s = make()
fn = Flaky()
assert retry(fn, **s) == "ok"
assert fn.calls == 1
assert t.sleeps == []
def test_retries_until_success():
t, s = make()
fn = Flaky(TimeoutError("timed out"), ConnectionError("connection reset"))
assert retry(fn, **s) == "ok"
assert t.sleeps == [0.25, 0.5]
def test_attempts_counts_every_call():
t, s = make(attempts=3)
fn = Flaky(*timeouts(5))
with pytest.raises(RetryError):
retry(fn, **s)
assert fn.calls == 3
assert t.sleeps == [0.25, 0.5]
def test_other_errors_are_not_retried():
t, s = make()
fn = Flaky(ValueError("card declined"))
with pytest.raises(ValueError):
retry(fn, **s)
assert fn.calls == 1
def test_delay_doubles_up_to_the_cap():
assert [backoff_delay(n, 0.25, 1.0, lambda: 1.0) for n in range(4)] == [0.25, 0.5, 1.0, 1.0]
def test_delay_is_scaled_by_jitter():
assert backoff_delay(3, 0.25, 5.0, lambda: 0.5) == 1.0
def test_stops_before_a_sleep_would_pass_the_deadline():
t, s = make(attempts=10, deadline_s=1.0)
fn = Flaky(*timeouts(10))
with pytest.raises(RetryError):
retry(fn, **s)
assert t.sleeps == [0.25, 0.5]
assert fn.calls == 3
def test_rejects_zero_attempts():
t, s = make(attempts=0)
with pytest.raises(ValueError):
retry(Flaky(), **s)
def test_rejects_a_cap_below_the_base():
t, s = make(base_s=2.0, cap_s=1.0)
with pytest.raises(ValueError):
retry(Flaky(), **s)
def test_decorator_retries_and_keeps_the_name():
t, s = make()
flaky = Flaky(ConnectionError("connection reset"))
@retrying(**s)
def charge(amount):
return flaky(amount)
assert charge(5) == 5
assert flaky.calls == 2
assert charge.__name__ == "charge"
# Retry Helper
The billing service calls a payment provider that times out under load, and every caller has its own retry loop. Add one shared retry(fn, ...) to backoff.py.
## What to Build
- attempts counts every call, the first one included. attempts=3 means one call and up to two retries.
- Only errors in retry_on are retried. Any other error is raised at once.
- Before retry n (0 for the first retry), wait rng() * min(cap_s, base_s * 2 ** n) seconds. Full jitter keeps a fleet of clients from retrying in lockstep.
- With deadline_s set, never start a sleep that would end more than deadline_s seconds after the first call. Raise RetryError instead.
- When retrying stops, raise RetryError with the last error as its cause.
- Reject attempts below 1, and any backoff outside 0 < base_s <= cap_s.
- retrying(**options) is the decorator form of retry, with the same options.
The tests in test_backoff.py inject a fake clock, a sleep that only records, and a fixed rng, so nothing waits and every delay is predictable.
Test Output
No runs yet.