Implementing circuit breakers for flaky municipal endpoints

A county GeoServer WFS starts flapping at 09:00 on refresh day: one request succeeds, the next three time out, a fourth returns a 500 with an HTML error page, then two more succeed. Your ingestion worker treats every failure as transient, retries each one four times with backoff, and within a minute it is firing several hundred requests a second at a server that is already falling over. The endpoint gets slower, the failure rate climbs, and eventually the county’s WAF stops answering your IP entirely — a self-inflicted outage layered on top of a minor server hiccup. This guide answers one narrow question: how do you stop retries from amplifying a flaky endpoint’s problems into a full outage or an IP ban? The answer is a per-host circuit breaker that sits inside your error handling and retry logic and short-circuits a failing host before the retries do damage, part of the wider automated feed ingestion and GIS data parsing pipeline that polls dozens of jurisdictions on a schedule.

Diagnosis: reading the retry-storm trace jump to heading

A circuit breaker is only worth building once you can see the storm in your logs. The signature is unmistakable: a burst of retries against a single host, each attempt landing closer together instead of farther apart, and a failure mode that migrates from soft (timeouts, 503s) to hard (connection refused, TLS handshake failure) as the server or its gateway gives up. A trimmed structured-log trace for one host during a flap looks like this.

09:00:04 host=gis.exampleco.gov attempt=1 status=503 latency=8.9s
09:00:06 host=gis.exampleco.gov attempt=2 status=timeout latency=10.0s
09:00:07 host=gis.exampleco.gov attempt=3 status=500 latency=2.1s
09:00:08 host=gis.exampleco.gov attempt=4 status=timeout latency=10.0s
09:00:08 host=gis.exampleco.gov attempt=1 status=503 latency=6.2s   # next URL, same host
09:00:09 host=gis.exampleco.gov attempt=2 status=ConnectionRefused latency=0.02s
09:00:11 host=gis.exampleco.gov attempt=3 status=ConnectionRefused latency=0.01s

Two things matter here. First, the failure is per host, not per URL — every layer under gis.exampleco.gov degrades together because they share one origin server and one gateway. That is why the breaker key must be the host, not the request path. Second, once ConnectionRefused appears with near-zero latency, the server is no longer doing work; the WAF or load balancer is rejecting you, and every additional request pushes you closer to a rate-based block. Naive backoff does not help because you run many URLs concurrently against the same host: each URL backs off on its own schedule, but collectively they still hammer the origin.

The fix is a small state machine, held per host, that trips open after the failure rate crosses a threshold, refuses calls for a cooldown, then allows a small number of half-open probes to test recovery before it closes again. The diagram below shows the three states and the transitions between them.

Step-by-step implementation jump to heading

Each step isolates one concern. Compose them in the final step into a polling loop you can drop into the ingestion worker.

Step 1 — Model the states and thresholds jump to heading

Represent the three states as an enum and gather every tunable into one config object so a jurisdiction with a fragile portal can be given a gentler threshold without touching code. Judge health on a rolling window of recent calls rather than a raw consecutive-failure count, and refuse to trip on too few samples so a cold start does not open the breaker on the first timeout.

from dataclasses import dataclass
from enum import Enum


class BreakerState(str, Enum):
    CLOSED = "closed"      # healthy: calls pass through
    OPEN = "open"          # tripped: calls are refused
    HALF_OPEN = "half_open"  # probing: a few calls test recovery


@dataclass
class BreakerConfig:
    failure_threshold: float = 0.5   # trip when >= 50% of the window fails
    window_size: int = 20            # rolling sample of recent calls
    minimum_calls: int = 8           # don't judge health on too few samples
    cooldown_seconds: float = 30.0   # how long OPEN refuses calls
    half_open_max_calls: int = 3     # probes admitted while HALF_OPEN

Step 2 — Implement the CircuitBreaker class jump to heading

The breaker records each call’s outcome into a bounded deque, computes the failure rate over the window, and transitions between states. allow() is the gate every request must pass: while open it returns False until the cooldown elapses, at which point it promotes the host to half-open and admits a capped number of probes.

import time
from collections import deque
from threading import Lock


class CircuitOpenError(Exception):
    """Raised in place of a network call while the breaker is open."""

    def __init__(self, host: str, retry_after: float):
        super().__init__(f"circuit open for {host}; retry in {retry_after:.1f}s")
        self.host = host
        self.retry_after = retry_after


class CircuitBreaker:
    def __init__(self, host: str, config: BreakerConfig | None = None):
        self.host = host
        self.cfg = config or BreakerConfig()
        self._state = BreakerState.CLOSED
        self._results: deque[bool] = deque(maxlen=self.cfg.window_size)
        self._opened_at = 0.0
        self._half_open_calls = 0
        self._lock = Lock()  # cheap, non-blocking guard for multi-worker use

    @property
    def state(self) -> BreakerState:
        return self._state

    def _failure_rate(self) -> float:
        if len(self._results) < self.cfg.minimum_calls:
            return 0.0
        failures = sum(1 for ok in self._results if not ok)
        return failures / len(self._results)

    def retry_after(self) -> float:
        elapsed = time.monotonic() - self._opened_at
        return max(0.0, self.cfg.cooldown_seconds - elapsed)

    def allow(self) -> bool:
        """Return True if a call may proceed; promote OPEN -> HALF_OPEN on cooldown."""
        with self._lock:
            if self._state == BreakerState.OPEN:
                if self.retry_after() <= 0:
                    self._state = BreakerState.HALF_OPEN
                    self._half_open_calls = 0
                else:
                    return False
            if self._state == BreakerState.HALF_OPEN:
                if self._half_open_calls >= self.cfg.half_open_max_calls:
                    return False
                self._half_open_calls += 1
            return True

    def _trip(self) -> None:
        self._state = BreakerState.OPEN
        self._opened_at = time.monotonic()
        self._half_open_calls = 0

    def record_success(self) -> None:
        with self._lock:
            self._results.append(True)
            if self._state == BreakerState.HALF_OPEN:
                # a clean probe closes the breaker and forgets old failures
                self._state = BreakerState.CLOSED
                self._results.clear()

    def record_failure(self) -> None:
        with self._lock:
            self._results.append(False)
            if self._state == BreakerState.HALF_OPEN:
                self._trip()  # any probe failure re-opens immediately
            elif self._failure_rate() >= self.cfg.failure_threshold:
                self._trip()

Step 3 — Register one breaker per host jump to heading

A single global breaker is wrong: a struggling county server should never silence a healthy neighboring city. Keep a registry that lazily creates one breaker per host so each jurisdiction trips independently, and expose a snapshot for telemetry.

class BreakerRegistry:
    """Lazily creates and holds one CircuitBreaker per host."""

    def __init__(self, config: BreakerConfig | None = None):
        self._breakers: dict[str, CircuitBreaker] = {}
        self._config = config or BreakerConfig()
        self._lock = Lock()

    def get(self, host: str) -> CircuitBreaker:
        with self._lock:
            breaker = self._breakers.get(host)
            if breaker is None:
                breaker = CircuitBreaker(host, self._config)
                self._breakers[host] = breaker
            return breaker

    def snapshot(self) -> dict[str, str]:
        """Host -> current state, for dashboards and alerting."""
        return {host: b.state.value for host, b in self._breakers.items()}

Step 4 — Wrap the async call and fold in backoff jump to heading

The breaker and exponential backoff are complementary, not competing: backoff spaces out attempts against a momentarily busy host, while the breaker stops attempts altogether once a host is persistently unhealthy. Check allow() before every attempt and raise CircuitOpenError without touching the network when the breaker is open — that single guard is what ends the retry storm. Count only server faults and timeouts as failures; a 404 or 401 says nothing about endpoint health.

import asyncio
import random
from urllib.parse import urlparse

import httpx

RETRYABLE_STATUS = {429, 500, 502, 503, 504}
registry = BreakerRegistry()


async def fetch_with_breaker(
    client: httpx.AsyncClient, url: str, max_attempts: int = 4
) -> httpx.Response:
    host = urlparse(url).netloc
    breaker = registry.get(host)

    for attempt in range(1, max_attempts + 1):
        if not breaker.allow():
            # short-circuit: refuse without a network round trip
            raise CircuitOpenError(host, breaker.retry_after())
        try:
            resp = await client.get(url, timeout=10.0)
        except (httpx.TimeoutException, httpx.TransportError):
            breaker.record_failure()
            if attempt == max_attempts:
                raise
        else:
            if resp.status_code in RETRYABLE_STATUS:
                breaker.record_failure()
                if attempt == max_attempts:
                    resp.raise_for_status()
            else:
                breaker.record_success()  # includes 2xx, 3xx, and 4xx client errors
                return resp
        # full-jitter backoff only between genuine attempts
        await asyncio.sleep(min(2 ** attempt, 30) * random.random())

    raise RuntimeError(f"exhausted retries for {url}")

Step 5 — Wire it into the polling loop and emit telemetry jump to heading

Finally, drive the breaker from the concurrent poller. A bounded Semaphore caps in-flight requests — the same concurrency discipline covered in async batch processing — and the CircuitOpenError branch is where you see the payoff: skipped hosts never reach the network. Log every state transition so a host that stays open becomes visible immediately.

import logging

logger = logging.getLogger("feed_ingestion.breaker")


async def poll_endpoints(urls: list[str], concurrency: int = 20) -> dict[str, str]:
    sem = asyncio.Semaphore(concurrency)
    results: dict[str, str] = {}

    limits = httpx.Limits(
        max_connections=concurrency, max_keepalive_connections=concurrency
    )
    async with httpx.AsyncClient(limits=limits) as client:

        async def worker(url: str) -> None:
            async with sem:
                try:
                    resp = await fetch_with_breaker(client, url)
                    results[url] = f"ok:{resp.status_code}"
                except CircuitOpenError as exc:
                    # refused with no network cost — this is the whole point
                    results[url] = f"skipped:{exc.retry_after:.0f}s"
                    logger.warning("breaker open host=%s", exc.host)
                except httpx.HTTPError as exc:
                    results[url] = f"error:{type(exc).__name__}"

        await asyncio.gather(*(worker(u) for u in urls))

    logger.info("breaker states=%s", registry.snapshot())
    return results

Verification & testing jump to heading

Prove the breaker trips, refuses, and recovers deterministically rather than trusting it in production.

• It opens on a sustained flap. Feed a breaker minimum_calls results at a failure rate above the threshold and assert breaker.state == BreakerState.OPEN. Below the threshold it must stay CLOSED.
• Open means zero network calls. Wrap a mock client that counts invocations, trip the breaker, then call fetch_with_breaker and assert the mock was never awaited and CircuitOpenError was raised. A single stray request means allow() is being bypassed.
• Cooldown promotes to half-open. Trip the breaker, monkeypatch time.monotonic (or set cooldown_seconds=0) so the cooldown elapses, and assert the next allow() returns True and the state is HALF_OPEN.
• A clean probe closes; a failed probe re-opens. In HALF_OPEN, one record_success() must return the breaker to CLOSED with a cleared window; one record_failure() must jump straight back to OPEN without waiting for the full window.
• Half-open admits only the cap. Call allow() half_open_max_calls + 1 times while half-open and assert the last call is refused, so a recovering host is probed gently instead of flooded.

Failure recovery jump to heading

The breaker changes how failures propagate, so the surrounding recovery machinery has to account for its new behavior.

• Dead-letter the refused URLs, do not discard them. A CircuitOpenError is a deferral, not a permanent failure. Route those URLs to a retry queue keyed on retry_after and replay them after the cooldown — the dead-letter and checkpoint patterns in error handling and retry logic apply unchanged.
• Never trip on client errors or rate limits you can honor. A 429 with a Retry-After header is a scheduling signal for your municipal API rate limit management layer to obey, not proof the host is broken — let the limiter pace those and reserve the breaker for genuine 5xx and transport failures.
• Coordinate half-open across workers. In-process breakers reset on restart and each worker keeps its own copy, so a fleet can send one probe per worker and re-flood a recovering host. Back the state with a shared store (Redis SETNX on a half_open:{host} key) so exactly one probe is admitted cluster-wide.
• Alert on stuck-open hosts. If a host stays OPEN across several cooldown cycles, the outage is real and the feed is stale — page a human and freeze downstream change detection rather than silently serving yesterday’s zoning data.

Frequently asked questions jump to heading

Related jump to heading

• Parent topic: Error Handling & Retry Logic
• Section overview: Automated Feed Ingestion & GIS Data Parsing
• Municipal API Rate Limit Management — pace 429s instead of tripping on them
• Async Batch Processing — the bounded concurrency the breaker plugs into
• aiohttp vs httpx for municipal API polling — choosing the client you wrap