Implementing circuit breakers for PBM API timeouts

The precise implementation decision on this page is not whether to wrap an adjudicator endpoint in a circuit breaker — the Fallback Routing Logic Design router already assumes one exists — but how to calibrate its two load-bearing constants: the failure threshold that opens the circuit and the recovery window that lets it probe again, and whether to run a single breaker or one breaker per pharmacy network segment. Set the threshold too low and normal P99 latency at a busy switch trips the breaker, dumping healthy claims onto the secondary node and inflating cost-of-goods reconciliation breaks. Set it too high and a genuinely dead endpoint keeps starving the connection pool while thread after thread blocks on a socket that will never answer. Both failures are adjudication-correctness problems, not just availability problems: a prematurely opened circuit routes a claim that would have adjudicated cleanly on the primary to a node whose copay accumulator may be stale, and a breaker that never opens turns a vendor timeout into a wall of 92 (Host Unavailable) rejects at the pharmacy counter. This page specifies how to pick those numbers against real latency percentiles, gives a deployable Python breaker, and shows how to prove the calibration with tests.

Calibration Decision Matrix #

The breaker guards each vendor endpoint independently, and retail, mail-order, and specialty switches have wildly different latency shapes — so the threshold and recovery window are calibrated per segment against that segment’s observed P95/P99, never with a single global constant. The table below is the calibration starting point; the numbers are illustrative baselines to be re-fit from your own switch telemetry, but the relationships between them (timeout above P99, threshold above expected concurrent P99 stragglers, recovery window ≥ vendor mean-time-to-recover) hold across segments.

Two constants drive everything downstream. The transport timeout must sit just above the segment P99, so a legitimately slow-but-alive vendor is not counted as a failure while a hung socket is abandoned before it exhausts the pool. The failure threshold is a count within the recovery window, not an all-time count — it must exceed the number of concurrent P99 stragglers you expect at peak, or ordinary tail latency opens the circuit. When the circuit is OPEN, every routed claim returns an NCPDP 92 (System Unavailable / Host Unavailable) envelope from the fallback path — versus 99 (Host Processing Error) for a transport error that is not a timeout — so mis-tuning is directly observable as a spike in one reject code. Because the OPEN state short-circuits before any payload deserialization, an open breaker also protects 302-C2 Cardholder ID and 310-CA Cardholder Name from ever being parsed out of an in-flight response that is about to be discarded.

Step-by-Step Implementation #

The breaker below is deployable as middleware around a requests or httpx client. It is GIL-aware (state mutation under a single lock), memory-bounded (a fixed collections.deque instead of an unbounded failure list), and PHI-safe (the fallback envelope projects only routing identifiers and never logs claim bytes). Follow the steps in order.

1. Bound the failure memory. Use a fixed-length deque(maxlen=...) so a sustained outage cannot grow the failure record without limit. Combined with __slots__, each breaker instance stays flat regardless of how long the vendor is down.
2. Time with a monotonic clock. Window math uses time.monotonic(), never time.time() — a wall-clock adjustment (NTP step, DST) must never retroactively re-open or heal a circuit.
3. Enforce the timeout at the transport layer. The primary_call is expected to carry the per-segment timeout from the matrix above; the breaker classifies a Timeout as a 92 failure and any other transport error as 99.
4. Mutate state under one lock, project PHI-free on the way out. Only 301-C1, 101-A1, and 407-D7 cross into the fallback envelope; 302-C2/310-CA are never copied.

import time
import threading
import logging
from collections import deque
from typing import Any, Callable
import requests

# GUARDRAIL: this logger never receives raw claim bytes. 302-C2 Cardholder ID and
# 310-CA Cardholder Name are tokenized upstream and must not reach any handler here.
logger = logging.getLogger("pbm.circuit_breaker")
logger.setLevel(logging.INFO)

# NCPDP reject codes emitted on the fallback path (511-FB Reject Code):
REJ_HOST_UNAVAILABLE = "92"   # System Unavailable / Host Unavailable  (timeout / OPEN)
REJ_HOST_PROCESSING = "99"    # Host Processing Error                  (non-timeout transport)


class CircuitBreaker:
    __slots__ = (
        "failure_threshold", "recovery_window",
        "failures", "last_failure_time", "state", "_lock",
    )

    def __init__(self, failure_threshold: int = 5, recovery_window: int = 15):
        # Calibrate per segment from the decision matrix, not with global constants.
        self.failure_threshold = failure_threshold
        self.recovery_window = recovery_window
        # Fixed-size deque: sustained outages cannot grow memory without bound.
        self.failures: deque[float] = deque(maxlen=failure_threshold * 3)
        self.last_failure_time = 0.0
        self.state = "CLOSED"
        self._lock = threading.Lock()

    def _window_breached(self) -> bool:
        """True once >= threshold failures fall inside the recovery window."""
        cutoff = time.monotonic() - self.recovery_window  # monotonic: NTP-step safe
        return sum(1 for t in self.failures if t > cutoff) >= self.failure_threshold

    def _record_failure(self) -> None:
        with self._lock:
            now = time.monotonic()
            self.failures.append(now)
            self.last_failure_time = now
            # A failed HALF_OPEN probe re-opens immediately; a CLOSED circuit opens
            # only once the windowed count crosses the calibrated threshold.
            if self.state == "HALF_OPEN":
                self.state = "OPEN"
                logger.info("circuit_reopened probe_failed")
            elif self.state == "CLOSED" and self._window_breached():
                self.state = "OPEN"
                logger.info("circuit_opened threshold=%d", self.failure_threshold)

    def execute(self, primary_call: Callable[[], dict[str, Any]],
                payload: dict[str, Any]) -> dict[str, Any]:
        with self._lock:
            if self.state == "OPEN":
                if time.monotonic() - self.last_failure_time >= self.recovery_window:
                    self.state = "HALF_OPEN"   # allow exactly one probe
                else:
                    return self._fallback(payload, REJ_HOST_UNAVAILABLE)

        try:
            response = primary_call()          # carries the per-segment transport timeout
            if self.state == "HALF_OPEN":
                self._reset()
            return response
        except requests.exceptions.Timeout:
            self._record_failure()
            return self._fallback(payload, REJ_HOST_UNAVAILABLE)
        except requests.exceptions.RequestException as exc:
            logger.warning("transport_error type=%s", type(exc).__name__)  # no payload
            self._record_failure()
            return self._fallback(payload, REJ_HOST_PROCESSING)

    def _reset(self) -> None:
        with self._lock:
            self.failures.clear()
            self.state = "CLOSED"
            self.last_failure_time = 0.0

    def _fallback(self, payload: dict[str, Any], reject_code: str) -> dict[str, Any]:
        # PHI-free projection: only routing identifiers cross into the envelope.
        return {
            "AdjudicationStatus": "FALLBACK_ROUTED",
            "RejectCode": reject_code,               # 511-FB
            "GroupID": payload.get("GroupID"),       # 301-C1 Group ID
            "BIN": payload.get("BIN"),               # 101-A1 BIN Number
            "ProductServiceID": payload.get("NDC"),  # 407-D7 Product/Service ID
            "RxRef": payload.get("RxRef"),           # 402-D2 Prescription/Service Reference #
        }

The state machine the calibration drives is unchanged from segment to segment — only the constants differ:

Figure: Circuit breaker state machine transitioning CLOSED to OPEN on threshold breach, OPEN to HALF_OPEN after the recovery window, then CLOSED on probe success or OPEN on probe failure.

Verification & Testing Pattern #

Calibration is only trustworthy if the transitions are asserted deterministically. Inject a fake clock so tests never sleep, then prove the three claims that matter: the circuit opens at exactly the threshold, an OPEN circuit returns a 92 fallback envelope while preserving routing identifiers and carrying no PHI, and a single HALF_OPEN probe success heals it. The fixtures use a known 301-C1/101-A1/407-D7 triple so the envelope projection can be asserted field by field.

import pytest
from breaker import CircuitBreaker  # module under test
import requests

FIXTURE = {  # canonical claim after ingress normalization; NO 302-C2/310-CA present
    "GroupID": "GRP7", "BIN": "610011",
    "NDC": "00093721410", "RxRef": "RX-9901",
}

def _timeout():
    raise requests.exceptions.Timeout()

def test_opens_at_threshold(monkeypatch):
    clock = [1000.0]
    monkeypatch.setattr("breaker.time.monotonic", lambda: clock[0])
    cb = CircuitBreaker(failure_threshold=3, recovery_window=15)
    for _ in range(3):
        out = cb.execute(_timeout, FIXTURE)
    assert cb.state == "OPEN"
    assert out["RejectCode"] == "92"                 # 511-FB Host Unavailable
    assert out["GroupID"] == "GRP7"                  # 301-C1 preserved
    assert out["ProductServiceID"] == "00093721410"  # 407-D7 preserved
    assert "CardholderID" not in out and "310-CA" not in out  # PHI never crosses

def test_half_open_probe_heals(monkeypatch):
    clock = [1000.0]
    monkeypatch.setattr("breaker.time.monotonic", lambda: clock[0])
    cb = CircuitBreaker(failure_threshold=2, recovery_window=15)
    for _ in range(2):
        cb.execute(_timeout, FIXTURE)
    assert cb.state == "OPEN"
    clock[0] += 16                                    # advance past recovery window
    ok = cb.execute(lambda: {"AdjudicationStatus": "PAID"}, FIXTURE)
    assert ok["AdjudicationStatus"] == "PAID"
    assert cb.state == "CLOSED"                        # probe success healed it

Run this in CI on every change to the breaker or the per-segment constants — a threshold regression should fail test_opens_at_threshold, and a leak of 302-C2 into the envelope should fail the PHI assertion loudly rather than silently in production.

Gotchas & PHI Guardrails #

• Threshold below concurrent P99 stragglers. At peak, several in-flight requests can legitimately sit near P99 at once. If the windowed threshold is lower than that natural burst, the circuit opens on healthy traffic and floods the secondary node — the exact terminal-reject amplification the parent router warns against. Fit the threshold from real peak telemetry, not a guess.
• One breaker for all segments. A shared breaker lets a specialty-switch outage open the retail path. Instantiate one CircuitBreaker per segment (retail / mail-order / specialty) so failures are bulkheaded; this is the isolation the decision matrix assumes.
• Thundering herd on recovery. If every worker’s breaker flips straight from OPEN to CLOSED the instant the window elapses, the recovering vendor is stampeded. The single-probe HALF_OPEN state is the guard — never bypass it, and pair the surrounding retries with exponential backoff and jitter.
• Wall-clock instead of monotonic. An NTP step or DST change against time.time() can retroactively heal or re-open a circuit. Every timing comparison here uses time.monotonic().
• PHI in the exception path. The tempting except: logger.error(payload) is a HIPAA exposure. Handlers here log the error type only; the fallback envelope projects 301-C1, 101-A1, 407-D7, and 402-D2 and never touches 302-C2 or 310-CA.
• Unbounded failure memory. A plain list of failure timestamps grows for the entire duration of a long outage. The fixed deque(maxlen=...) keeps the footprint flat under sustained degradation.
• Stale accumulators on the routed-to node. A 92 fallback is only correct if the secondary node shares eligibility and copay-accumulator state; gate failover on parity from the PBM Portal Sync Architecture, not on node health alone.

Related #

• Fallback Routing Logic Design — the router this breaker plugs into, including the full trigger matrix and failover flow.
• Handling PBM 404 and 503 errors in adjudication scripts — the transport-error taxonomy that feeds the breaker’s failure count.
• Asynchronous batch adjudication workflows — the concurrency model whose peak fan-out sets your realistic failure threshold.
• Security & Compliance Boundaries for Claims Data — the PHI and audit obligations the fallback envelope must satisfy.
• PBM Portal Sync Architecture — the eligibility and accumulator parity that makes a routed-to node safe.

← Back to Fallback Routing Logic Design