Implementing a Replica Health Circuit Breaker in Application Code

Problem statement: Your router keeps sending reads to a replica that is lagging or timing out because health lives in a dashboard, not in the routing decision — you need a per-replica breaker that checks lag before checkout, trips after N consecutive breaches, half-open probes on recovery, and stays correct under concurrency.

Symptom Identification

These are the signals that a routing layer needs a breaker rather than a static replica list:

  • Users report stale data for tens of seconds after a known replica incident, and the routing logs show reads still landing on the degraded node the whole time.
  • A replica that is refusing connections still receives a steady stream of checkout attempts, each of which blocks for the full connection_timeout before the router tries the next node — turning one dead replica into a latency cliff for every request.
  • Two application instances “recover” a replica at the same instant after an outage, both slam it with full traffic, and it immediately falls behind again — a classic missing half-open gate.
  • Adding lag awareness to routing made p99 read latency worse, because the router now issues a pg_last_wal_replay_lsn() query on the hot path — the lag check itself became the bottleneck.

If any of these match, the fix is a breaker with a cached lag signal and lock-guarded transitions.

Root Cause Analysis

Three design gaps produce those symptoms.

1. Health is not consulted at routing time. Lag telemetry exists in Prometheus but the router reads a static replica list. There is no code path that turns “replica B is 40 s behind” into “do not route to replica B right now.” The breaker closes that gap by making health a gate the request must pass through before checkout.

2. The lag signal is measured synchronously. The naive fix — query the replica’s replay position on every read — adds a round trip to the hot path and, perversely, loads the exact node you are trying to relieve. The signal must be sampled by a background poller and cached, so the routing decision is a memory read.

3. Transitions are not concurrency-safe. In a multi-threaded or async server, dozens of requests evaluate the same breaker simultaneously. Without a lock around the state transition, the OPEN → HALF_OPEN → CLOSED sequence races: multiple requests each think they hold the single probe slot, and the “one careful probe” guarantee evaporates. This is the same failure the parent guide on circuit breaker patterns for degraded replicas describes as false recovery, surfacing here as a thread-safety bug.

Architecture

The breaker sits between the router and the connection pool. The read path is pure memory access; the network cost of measuring lag is paid off-path by a poller.

Breaker on the read path with a background lag poller A read request enters the router, which calls breaker.allow. If allowed, it checks out a connection from the replica pool and records success or failure back into the breaker. A separate background poller queries each replica's lag on an interval and writes it into a cached lag value that the breaker reads without any network I/O on the request path. Read request router.route() Breaker allow() • lock reads cached lag allowed Replica pool checkout → query Replica hot standby record_success / record_failure Background poller lag query every 1–2 s writes cached lag → breaker

Step-by-Step Resolution

Step 1: Model the breaker state with a lock

Each replica owns one breaker. All state lives behind a single lock so transitions are atomic.

python
import threading, time, random
from enum import Enum

class State(Enum):
    CLOSED = 0
    OPEN = 1
    HALF_OPEN = 2

class ReplicaBreaker:
    def __init__(self, name, trip_lag_ms=500, reenter_lag_ms=200,
                 trip_threshold=3, cooldown_base=2.0, cooldown_max=60.0):
        self.name = name
        self.trip_lag_ms = trip_lag_ms
        self.reenter_lag_ms = reenter_lag_ms
        self.trip_threshold = trip_threshold
        self.cooldown_base = cooldown_base
        self.cooldown_max = cooldown_max
        self._lock = threading.Lock()
        self._state = State.CLOSED
        self._breaches = 0
        self._probe_fails = 0
        self._opened_at = 0.0
        self._probe_in_flight = False
        self.cached_lag_ms = 0.0   # written by the background poller

Inline verification: ReplicaBreaker("a")._state is State.CLOSED — a fresh breaker starts closed and serving.

Step 2: Refresh the lag signal off the request path

A background thread polls each replica and writes the result into cached_lag_ms. The routing path never issues this query.

python
def start_lag_poller(breaker, get_lag_ms, interval=1.5):
    def loop():
        while True:
            try:
                breaker.cached_lag_ms = get_lag_ms(breaker.name)
            except Exception:
                breaker.cached_lag_ms = float("inf")   # unreachable == max lag
            time.sleep(interval)
    t = threading.Thread(target=loop, daemon=True)
    t.start()
    return t

# get_lag_ms runs, e.g.:
#   SELECT EXTRACT(EPOCH FROM (now() - pg_last_xact_replay_timestamp()))*1000;

Treating an unreachable poll as infinite lag means a node that stops answering trips the breaker through the same lag path, no separate connectivity branch needed.

Inline verification: kill a replica’s network and confirm breaker.cached_lag_ms becomes inf within one interval, then the breaker opens on the next allow().

Step 3: Gate routing with allow()

allow() is the whole decision, and it runs entirely against cached state under the lock.

python
    def allow(self):
        now = time.monotonic()
        with self._lock:
            if self._state == State.OPEN:
                if now - self._opened_at < self._cooldown():
                    return False                     # short-circuit
                self._state = State.HALF_OPEN        # cooldown elapsed
                self._probe_in_flight = False
            if self._state == State.HALF_OPEN:
                if self._probe_in_flight:
                    return False                     # someone else holds the probe
                self._probe_in_flight = True         # claim the single probe slot
            # CLOSED, or the one admitted HALF_OPEN probe:
            if self.cached_lag_ms > self.trip_lag_ms:
                self._trip(now)
                return False
            return True

    def _cooldown(self):
        return min(self.cooldown_base * (2 ** self._probe_fails),
                   self.cooldown_max) + random.uniform(0, 0.5)

Inline verification: set cached_lag_ms = 900 on a closed breaker and call allow() three times — the third call returns False and the state is now OPEN.

Step 4: Feed outcomes back in

The router reports what happened so the breaker can trip on errors and close on a good probe.

python
    def record_success(self):
        with self._lock:
            self._breaches = 0
            if self._state == State.HALF_OPEN and \
               self.cached_lag_ms <= self.reenter_lag_ms:
                self._state = State.CLOSED           # hysteresis on re-entry
                self._probe_fails = 0
            self._probe_in_flight = False

    def record_failure(self):
        with self._lock:
            self._trip(time.monotonic())

    def _trip(self, now):
        self._breaches += 1
        if self._state == State.HALF_OPEN or self._breaches >= self.trip_threshold:
            if self._state != State.OPEN:
                self._probe_fails += 1 if self._state == State.HALF_OPEN else 0
            self._state = State.OPEN
            self._opened_at = now
            self._probe_in_flight = False

Inline verification: after allow() returns True in HALF_OPEN, calling record_failure() returns the breaker to OPEN and increments the cooldown via _probe_fails.

Step 5: Wire it into the router with metrics

The router iterates candidate replicas, skips any whose breaker denies, and records the outcome. Metrics make every transition visible.

python
from prometheus_client import Gauge, Counter

BREAKER_STATE = Gauge("replica_breaker_state", "0=closed 1=open 2=half", ["replica"])
BREAKER_TRIPS = Counter("replica_breaker_trips_total", "trips", ["replica", "reason"])

def route_read(query, replicas, primary):
    for r in replicas:
        if not r.breaker.allow():
            continue
        try:
            result = r.pool.execute(query)
            r.breaker.record_success()
            return result
        except (ConnectionError, TimeoutError):
            r.breaker.record_failure()
            BREAKER_TRIPS.labels(r.name, "error").inc()
    return primary.pool.execute(query)   # degraded fallback — see graceful degradation guide

Inline verification: scrape /metrics and confirm replica_breaker_state{replica="a"} flips to 1 within one request after replica A crosses the lag threshold.

Configuration Snippet

Bootstrap one breaker per replica and start its poller:

python
replicas = []
for name, dsn in [("a", DSN_A), ("b", DSN_B), ("c", DSN_C)]:
    br = ReplicaBreaker(name, trip_lag_ms=500, reenter_lag_ms=200,
                        trip_threshold=3, cooldown_base=2.0, cooldown_max=60.0)
    start_lag_poller(br, get_lag_ms=poll_replay_lag, interval=1.5)
    replicas.append(Replica(name=name, pool=make_pool(dsn), breaker=br))

For asyncio services, swap threading.Lock for asyncio.Lock, make allow() a coroutine, and run the poller as an asyncio.create_task loop — the state logic is identical. This mirrors the context-propagation discipline the ORM middleware for automatic query routing guide applies to routing hints.

Verification and Rollback

Confirm the breaker behaves under a controlled lag injection:

bash
# Inject apply lag on replica A, e.g. pause WAL replay
psql -h replica_a -c "SELECT pg_wal_replay_pause();"
# Watch the breaker open, then resume and watch it half-open -> close
psql -h replica_a -c "SELECT pg_wal_replay_resume();"
python
# Assert the full cycle in an integration test
assert br.allow() is True
br.cached_lag_ms = 5000
for _ in range(3):
    br.allow()
assert br._state is State.OPEN            # tripped
time.sleep(br._cooldown())
br.cached_lag_ms = 100
assert br.allow() is True                 # half-open probe admitted
br.record_success()
assert br._state is State.CLOSED          # closed on good probe

To roll back, set trip_lag_ms to float("inf") in config — allow() then never trips on lag and the breaker is inert while you keep the metrics and error-based tripping. This is safer than ripping the code out, because it leaves the observability in place.

Edge Cases and Gotchas

The poller thread dies silently

If the background poller raises an unhandled exception outside the try, the thread exits and cached_lag_ms freezes at its last value — the breaker then makes decisions on permanently stale data. Wrap the whole loop body, export a poller_last_run_timestamp gauge, and alert if it stops advancing. A frozen poller is more dangerous than a missing one because it looks healthy.

Holding the lock across a database call

It is tempting to move the lag query inside allow() under the lock “for simplicity.” Doing so serializes every request behind one network round trip and can deadlock if the query hangs while other threads wait on the lock. Keep the lock critical section to pure in-memory work; all I/O happens in the poller or after allow() returns.

Per-process breakers disagree

Each application process runs its own breakers, so instance 1 may have replica B open while instance 5 still has it closed — they sampled lag at slightly different times. This is acceptable and self-correcting, but do not rely on application breakers for fleet-wide ejection; that belongs at the proxy layer described in automatic replica ejection and reintroduction on lag spikes.

FAQ

Should the lag check run on the request path or in the background?

Run it in the background. Querying pg_last_wal_replay_lsn or a heartbeat table on every read adds a full round trip to the hot path and, worse, hits the replica you are trying to protect. A background poller refreshes a cached lag value every one to two seconds; the routing path reads that cached value with no I/O. The breaker trips on the cached sample, which is at most one poll interval stale.

How do I make the breaker safe across threads and async tasks?

Guard all state transitions with a per-breaker lock so two threads cannot both admit the half-open probe. Keep the critical section tiny — read the clock, mutate counters, release — and never hold the lock across a database call. For asyncio, use asyncio.Lock and an async background poll task; for threads, use threading.Lock. The half-open probe slot should be a single-permit gate so exactly one caller probes at a time.

What should happen when every replica breaker is open?

The router should fall through to a degraded path rather than raising an unhandled error. When allow() returns false for all replicas, route the read to the primary under a rate limit, or serve a cached response, depending on the endpoint’s tolerance. Track this as a distinct degraded-mode metric so an all-open fleet pages you, separate from a single replica tripping.

← Back to Circuit Breaker Patterns for Degraded Replicas