Resilience — Timeouts, Fallbacks & Circuit Breakers¶

requests read timeout=1:  10 bytes in 2.7 s    (no error)
httpx timeout=1:          10 bytes in 2.7 s    (no error)
asyncio.timeout(1.5):     TimeoutError after 1.5 s

async with asyncio.timeout(5):                     # whole request, slow body included
    response = await client.get(url)

# app/deadline.py
import time
from collections.abc import Iterator
from contextlib import contextmanager
from contextvars import ContextVar

_deadline: ContextVar[float | None] = ContextVar("deadline", default=None)


class DeadlineExceeded(TimeoutError):
    pass


@contextmanager
def deadline(seconds: float) -> Iterator[None]:
    """Set a deadline for everything inside; a nested deadline can only be shorter."""
    new = time.monotonic() + seconds
    current = _deadline.get()
    token = _deadline.set(new if current is None else min(current, new))
    try:
        yield
    finally:
        _deadline.reset(token)


def remaining(default: float) -> float:
    """Timeout for the next call: what is left of the deadline, at most `default`."""
    at = _deadline.get()
    if at is None:
        return default
    left = at - time.monotonic()
    if left <= 0:
        raise DeadlineExceeded("deadline exceeded before the call")
    return min(default, left)

from app.deadline import deadline, remaining

with deadline(2.0):                                  # set once, e.g. in middleware
    print(round(remaining(10), 1))                   # 2.0  - capped by the deadline
    with deadline(0.5):                              # a stricter step inside
        print(round(remaining(10), 1))               # 0.5
    print(round(remaining(0.3), 1))                  # 0.3  - capped by the call's own limit
print(remaining(7))                                  # 7    - no deadline set

response = await client.get(url, timeout=remaining(5.0))                 # never longer than what is left
headers = {"X-Deadline-Ms": str(int(remaining(5.0) * 1000))}           # tell the next service, too

# app/fallbacks.py
import asyncio
import logging
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass

from app.breaker import CircuitBreaker, CircuitOpenError

logger = logging.getLogger(__name__)

Fetcher = Callable[[], Awaitable[dict[str, float]]]
DEFAULT_RATES = {"USD": 1.0}
FALLBACK_ON = (TimeoutError, ConnectionError, CircuitOpenError)


@dataclass(frozen=True)
class Rates:
    values: dict[str, float]
    source: str                  # primary, secondary, cache, default
    degraded: bool               # tell the caller (and the UI, and metrics)


class RatesService:
    def __init__(
        self,
        primary: Fetcher,
        secondary: Fetcher,
        *,
        breaker: CircuitBreaker | None = None,
        timeout: float = 1.0,
        max_stale: float = 3600.0,
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        self._primary = primary
        self._secondary = secondary
        self._breaker = breaker or CircuitBreaker(failure_types=FALLBACK_ON)
        self._timeout = timeout
        self._max_stale = max_stale
        self._clock = clock
        self._last_good: tuple[dict[str, float], float] | None = None

    async def _timed(self, fetch: Fetcher) -> dict[str, float]:
        async with asyncio.timeout(self._timeout):              # each step gets its own budget
            return await fetch()

    async def _call_primary(self) -> dict[str, float]:
        # timeout INSIDE the breaker: a slow primary counts as a failure
        return await self._breaker.call_async(lambda: self._timed(self._primary))

    async def _call_secondary(self) -> dict[str, float]:
        return await self._timed(self._secondary)

    async def get(self) -> Rates:
        for source, fetch in (("primary", self._call_primary), ("secondary", self._call_secondary)):
            try:
                values = await fetch()
            except FALLBACK_ON as exc:
                logger.warning("rates: %s failed: %r", source, exc)
                continue
            self._last_good = (values, self._clock())
            return Rates(values, source, degraded=source != "primary")

        if self._last_good and self._clock() - self._last_good[1] <= self._max_stale:
            return Rates(self._last_good[0], "cache", degraded=True)
        return Rates(DEFAULT_RATES, "default", degraded=True)

WARNING app.fallbacks: rates: primary failed: TimeoutError()
WARNING app.fallbacks: rates: secondary failed: ConnectionError('refused')
Rates(values={'USD': 1.0, 'EUR': 0.91}, source='cache', degraded=True)

# app/breaker.py
import threading
import time
from collections.abc import Awaitable, Callable
from enum import StrEnum


class State(StrEnum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"


class CircuitOpenError(Exception):
    """Fail fast: the dependency is considered down."""


class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        reset_timeout: float = 30.0,
        failure_types: tuple[type[BaseException], ...] = (Exception,),
        clock: Callable[[], float] = time.monotonic,
    ) -> None:
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failure_types = failure_types
        self._clock = clock
        self._lock = threading.Lock()            # guards state only, never held during the call
        self._state = State.CLOSED
        self._failures = 0
        self._opened_at = 0.0
        self._probe_in_flight = False

    @property
    def state(self) -> State:
        with self._lock:
            return self._current_state()

    def _current_state(self) -> State:
        if self._state is State.OPEN and self._clock() - self._opened_at >= self.reset_timeout:
            self._state = State.HALF_OPEN
        return self._state

    def _before_call(self) -> None:
        with self._lock:
            state = self._current_state()
            if state is State.OPEN:
                raise CircuitOpenError("circuit is open")
            if state is State.HALF_OPEN:
                if self._probe_in_flight:
                    raise CircuitOpenError("half-open: a trial call is already running")
                self._probe_in_flight = True     # exactly one trial call

    def _on_success(self) -> None:
        with self._lock:
            self._state = State.CLOSED
            self._failures = 0
            self._probe_in_flight = False

    def _on_neutral(self) -> None:
        with self._lock:
            self._probe_in_flight = False        # not a verdict on the dependency

    def _on_failure(self) -> None:
        with self._lock:
            self._probe_in_flight = False
            self._failures += 1
            if self._state is State.HALF_OPEN or self._failures >= self.failure_threshold:
                self._state = State.OPEN
                self._opened_at = self._clock()

    def call[T](self, fn: Callable[[], T]) -> T:
        self._before_call()
        try:
            result = fn()
        except self.failure_types:
            self._on_failure()
            raise
        except BaseException:
            self._on_neutral()                   # e.g. ValueError, CancelledError
            raise
        self._on_success()
        return result

    async def call_async[T](self, fn: Callable[[], Awaitable[T]]) -> T:
        self._before_call()
        try:
            result = await fn()
        except self.failure_types:
            self._on_failure()
            raise
        except BaseException:
            self._on_neutral()
            raise
        self._on_success()
        return result

import logging

import httpx
import pybreaker

logger = logging.getLogger("breakers")


class LogTransitions(pybreaker.CircuitBreakerListener):
    def state_change(self, cb, old_state, new_state) -> None:
        logger.warning("breaker %s: %s -> %s", cb.name, old_state.name, new_state.name)


def is_client_error(exc: BaseException) -> bool:
    return isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code < 500


rates_breaker = pybreaker.CircuitBreaker(
    name="rates-api",
    fail_max=3,                      # consecutive failures that open the circuit
    reset_timeout=30,                # seconds in "open" before one trial call
    exclude=[is_client_error],       # 4xx: the service is up; counts as success
    listeners=[LogTransitions()],
)


@rates_breaker
def get_rates(client: httpx.Client) -> dict:
    response = client.get("https://rates.test/latest", timeout=2)
    response.raise_for_status()
    return response.json()

real call failed: 503
real call failed: 503
WARNING breaker rates-api: closed -> open
fail fast: Failures threshold reached, circuit breaker opened
fail fast: Timeout not elapsed yet, circuit breaker still open
fail fast: Timeout not elapsed yet, circuit breaker still open

import httpx
from purgatory import AsyncCircuitBreakerFactory
from purgatory.domain.model import OpenedState

breakers = AsyncCircuitBreakerFactory(default_threshold=3, default_ttl=30)
# shared state for several processes: AsyncCircuitBreakerFactory(uow=AsyncRedisUnitOfWork("redis://localhost:6379/0"))


@breakers("rates-api", exclude=[(httpx.HTTPStatusError, lambda exc: exc.response.status_code < 500)])
async def get_rates(client: httpx.AsyncClient) -> dict:
    response = await client.get("https://rates.test/latest", timeout=2)
    response.raise_for_status()
    return response.json()


async def fetch_or_none(client: httpx.AsyncClient) -> dict | None:
    try:
        return await get_rates(client)
    except OpenedState:                          # "Circuit rates-api is open"
        return None

# app/bulkhead.py
import asyncio
from collections.abc import Awaitable, Callable


class BulkheadFull(Exception):
    """No free slot for this dependency: fail fast instead of queueing."""


class Bulkhead:
    """At most `size` concurrent calls; extra callers wait up to `max_wait` seconds, then fail."""

    def __init__(self, name: str, size: int, max_wait: float = 0.0) -> None:
        self.name = name
        self.max_wait = max_wait
        self._slots = asyncio.Semaphore(size)

    async def call[T](self, fn: Callable[[], Awaitable[T]]) -> T:
        try:
            async with asyncio.timeout(self.max_wait):
                await self._slots.acquire()
        except TimeoutError:
            raise BulkheadFull(self.name) from None
        try:
            return await fn()
        finally:
            self._slots.release()


payments = Bulkhead("payments", size=20, max_wait=0.5)          # critical: may wait a little
recommendations = Bulkhead("recommendations", size=5)           # optional: never queue