SerializationAutomation DSCSA Track & Trace

Implementing Circuit Breakers for VRS API Calls

A single trading partner’s verification endpoint starts timing out at 2 a.m. — a database failover, an expiring certificate, a regional network blip — and every dispenser query that happens to resolve to that manufacturer now waits the full connection timeout before failing. If your worker pool has fifty requests in flight and each one blocks for ten seconds, you have effectively taken down your own verification service to punish yourself for someone else’s outage. That is the exact failure this page solves: a self-contained async circuit breaker in front of every outbound Verification Router Service (VRS) call, so that one degraded partner endpoint short-circuits fast, falls back to a cached status, and never cascades into your own SLA breach. It is a code-first guide within Verification Router Service Architecture, part of the broader DSCSA Compliance Architecture & Standards Mapping framework, and it assumes the router already resolves a verification request to the correct partner endpoint before this pattern wraps the outbound hop.

Circuit breaker state machine for a VRS partner endpoint The breaker starts CLOSED, letting calls pass through. When the failure rate over a rolling window meets or exceeds the configured threshold, it trips to OPEN and short-circuits every call to the cached fallback. After the recovery timeout elapses, it moves to HALF-OPEN and admits exactly one trial call: success returns it to CLOSED, failure sends it back to OPEN. failure rate ≥ threshold recovery_timeout elapses trial call succeeds trial call fails CLOSED calls pass through OPEN short-circuit to cache HALF-OPEN one trial call allowed

The three states map directly onto operational decisions a compliance officer and an on-call engineer both need to reason about. CLOSED is the default: every verification query is forwarded to the partner and its outcome is recorded. OPEN is the protective state: once the observed failure rate crosses the configured threshold, the breaker stops issuing calls to that partner entirely for a fixed recovery window and answers instead from the cache described below, converting an unbounded string of timeouts into a bounded, predictable latency hit. HALF_OPEN is the probe: after the recovery window elapses, exactly one request is allowed through as a canary — if it succeeds, the partner is presumed healthy again and the breaker closes; if it fails, the breaker reopens and the recovery window restarts. This is deliberately more conservative than a naive retry loop, because a naive loop keeps hammering a partner that is already struggling, while the breaker gives it room to recover before testing it again.

Prerequisites

  • Python 3.10+ — the snippets use X | Y union types and the dataclass/Enum combination for typed state.
  • httpx ≥ 0.24 with httpx.AsyncClient — the async HTTP client the breaker wraps; any awaitable client call works the same way.
  • pytest and pytest-asyncio for the verification scaffold.
  • A cache layer keyed by SGTIN — an in-process dict for the examples here, or the persistent store built in Building a VRS Response Cache in Python, which this breaker’s fallback path reads from.
  • DSCSA data prerequisites — a VRS query shape that carries GTIN (01) and serial number (21) per verification lookup (some partner integrations add lot (10) and expiration (17) for a stricter match), and a stable per-partner identifier (GLN) so failures are tracked independently for each trading partner rather than pooled together.

Before writing code, decide what “failing” means for a VRS call. A 4xx response about a malformed request is a data-quality defect, not a partner outage, and should not count toward the breaker’s failure rate. A 5xx, a connection reset, or a timeout is exactly what the breaker exists to detect. Getting this distinction wrong either trips the breaker on your own bad requests or, worse, never trips it on a genuinely degraded partner because timeouts were swallowed upstream. This distinction is also why the breaker described here is deliberately narrow in scope: it protects one outbound hop to one partner, not the whole verification pipeline. A request that fails validation before it ever reaches the network — a serial that is not 1-20 characters, or a GTIN that fails its GS1 modulo-10 check digit — belongs in the schema-validation gate upstream of this code, not in the breaker’s failure count, because retrying or falling back on a structurally invalid request only wastes a cache lookup on a query that was never going to resolve correctly.

Step-by-Step Solution

Step 1 — Define the breaker’s states and configuration

The breaker cycles through three states: CLOSED (normal traffic), OPEN (short-circuiting every call), and HALF_OPEN (a single trial call to probe recovery). Configuration is a rolling failure-rate threshold rather than a bare consecutive-failure count, because a rate over a window tolerates the occasional blip without ignoring a partner that is failing half the time. A fixed-size deque is the simplest correct structure for the rolling window: it automatically evicts the oldest outcome as new ones arrive, so the failure rate always reflects the most recent window_size calls rather than an ever-growing history that would make an old outage linger in the statistics long after the partner recovered.

import time
from collections import deque
from dataclasses import dataclass
from enum import Enum


class BreakerState(str, Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"


@dataclass
class BreakerConfig:
    failure_rate_threshold: float = 0.5   # trip above 50% failures in the window
    min_calls_in_window: int = 10         # never trip on noise from a cold start
    window_size: int = 30                 # rolling count of recent call outcomes
    recovery_timeout: float = 30.0        # seconds OPEN before probing HALF_OPEN
    half_open_max_calls: int = 1          # concurrent trial calls admitted
    call_timeout: float = 3.0             # seconds before a VRS call counts failed


class CircuitOpenError(Exception):
    """Raised when a call is rejected because the breaker is OPEN."""

DSCSA/GS1 note: min_calls_in_window matters because a verification router that just onboarded a new trading partner GLN has no failure history yet; requiring a minimum sample before evaluating the rate stops a single cold-start timeout from tripping the breaker for a partner that is otherwise healthy.

Step 2 — Implement the async CircuitBreaker state machine

The breaker itself has no dependency on httpx or any specific transport — it wraps any awaitable. call() admits or rejects the request based on current state, enforces a per-call timeout, and records the outcome to drive the next transition. An asyncio.Lock guards state reads/writes so concurrent coroutines never race each other into inconsistent state.

import asyncio


class CircuitBreaker:
    """A self-contained async circuit breaker for one trading partner's VRS endpoint."""

    def __init__(self, name: str, config: BreakerConfig | None = None) -> None:
        self.name = name
        self.config = config or BreakerConfig()
        self._state = BreakerState.CLOSED
        self._outcomes: deque[bool] = deque(maxlen=self.config.window_size)
        self._opened_at: float | None = None
        self._half_open_calls_in_flight = 0
        self._lock = asyncio.Lock()

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

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

    def _to_open(self) -> None:
        self._state = BreakerState.OPEN
        self._opened_at = time.monotonic()
        self._outcomes.clear()

    def _to_half_open(self) -> None:
        self._state = BreakerState.HALF_OPEN
        self._half_open_calls_in_flight = 0

    def _to_closed(self) -> None:
        self._state = BreakerState.CLOSED
        self._opened_at = None
        self._outcomes.clear()

    async def _admit(self) -> bool:
        async with self._lock:
            if self._state == BreakerState.OPEN:
                assert self._opened_at is not None
                if time.monotonic() - self._opened_at >= self.config.recovery_timeout:
                    self._to_half_open()
                else:
                    return False
            if self._state == BreakerState.HALF_OPEN:
                if self._half_open_calls_in_flight >= self.config.half_open_max_calls:
                    return False
                self._half_open_calls_in_flight += 1
            return True

    async def _record(self, success: bool) -> None:
        async with self._lock:
            if self._state == BreakerState.HALF_OPEN:
                self._to_closed() if success else self._to_open()
                return
            self._outcomes.append(success)
            if (self._state == BreakerState.CLOSED
                    and len(self._outcomes) >= self.config.min_calls_in_window
                    and self._failure_rate() >= self.config.failure_rate_threshold):
                self._to_open()

    async def call(self, coro_fn, *args, **kwargs):
        if not await self._admit():
            raise CircuitOpenError(f"circuit '{self.name}' is open")
        try:
            result = await asyncio.wait_for(
                coro_fn(*args, **kwargs), timeout=self.config.call_timeout
            )
        except Exception:
            await self._record(success=False)
            raise
        await self._record(success=True)
        return result

DSCSA/GS1 note: wrapping the call in asyncio.wait_for means a hung connection to a partner’s endpoint counts as a breaker failure exactly like an HTTP 5xx — a verification request that never returns is functionally identical to one that fails outright, and the breaker’s SLA protection depends on treating them the same way.

Step 3 — Wrap the VRS httpx call and fall back to cached status

With the breaker in place, the actual VRS query is a plain async function. The GTIN (01) and serial number (21) form the SGTIN pair the request is keyed on; when the breaker is open or the call itself fails, the wrapper falls back to the last cached status for that same pair rather than propagating the failure to the dispenser. Note that _query_vrs never sees the breaker at all — it is an ordinary coroutine, which is the point: the breaker composes around any awaitable without the wrapped function needing to know it exists, so the same pattern applies unchanged if you later swap httpx for another transport.

import httpx


async def _query_vrs(
    client: httpx.AsyncClient, endpoint: str, gtin: str, serial: str
) -> dict:
    response = await client.get(
        endpoint,
        params={"gtin": gtin, "serialNumber": serial},  # AI (01) and (21)
    )
    response.raise_for_status()
    return response.json()


async def verify_with_breaker(
    breaker: CircuitBreaker,
    client: httpx.AsyncClient,
    cache: dict[tuple[str, str], dict],
    endpoint: str,
    gtin: str,
    serial: str,
) -> dict:
    key = (gtin, serial)
    try:
        result = await breaker.call(_query_vrs, client, endpoint, gtin, serial)
    except (CircuitOpenError, httpx.HTTPError, asyncio.TimeoutError):
        cached = cache.get(key)
        if cached is None:
            raise
        return {**cached, "source": "cache", "stale": True}
    cache[key] = result
    return {**result, "source": "live", "stale": False}

DSCSA/GS1 note: the returned stale flag is not cosmetic — it is what lets a downstream suspect-product check distinguish a live “verified” ObjectEvent disposition from a cached one that may be minutes old, which matters if the same serial is later flagged in a suspect product investigation.

Step 4 — One breaker per trading partner, never one breaker for all

A single shared breaker instance across every partner GLN means one manufacturer’s outage trips the breaker for every other manufacturer’s healthy traffic too — the opposite of isolation. Keep a small registry keyed by partner identifier so each endpoint’s failure history is independent. This registry pattern also gives operations a natural place to expose per-partner health: iterating _breakers and reading each instance’s state property is enough to build a dashboard panel that shows which trading partners are currently degraded without touching the request path at all.

_breakers: dict[str, CircuitBreaker] = {}


def get_breaker(partner_gln: str, config: BreakerConfig | None = None) -> CircuitBreaker:
    """Return the CircuitBreaker for a partner GLN, creating one on first use."""
    if partner_gln not in _breakers:
        _breakers[partner_gln] = CircuitBreaker(name=partner_gln, config=config)
    return _breakers[partner_gln]

DSCSA/GS1 note: partner isolation mirrors the same principle the router uses to resolve a verification query to the correct manufacturer by GTIN in the first place — the breaker is scoped to the same boundary the routing logic already establishes, so a per-partner GLN key is the natural choice rather than inventing a new one.

Step 5 — Log every state transition for the audit trail

A breaker that trips silently is invisible to the compliance team until an SLA report already shows the damage. Emit a structured log entry on every transition so operations can correlate a partner outage with the exact window the breaker protected against it.

import logging
from datetime import datetime, timezone

logger = logging.getLogger("vrs.circuit_breaker")


def log_transition(partner: str, from_state: BreakerState, to_state: BreakerState) -> None:
    logger.warning(
        "circuit_breaker_transition",
        extra={
            "partner": partner,
            "from_state": from_state.value,
            "to_state": to_state.value,
            "at": datetime.now(timezone.utc).isoformat(),
        },
    )

DSCSA/GS1 note: call log_transition from _to_open, _to_half_open, and _to_closed before assigning self._state. Retaining these transition records alongside the rest of the verification audit trail is what lets an inspector confirm that a degraded partner window was handled by design — cached fallback, not silent data loss — rather than by an unhandled exception.

Verification

Confirm the full lifecycle — trip, short-circuit, recover — with a short-window configuration so the test runs in milliseconds instead of waiting on the production recovery_timeout.

import pytest


async def _always_fails():
    raise RuntimeError("simulated VRS outage")


async def _always_succeeds():
    return {"status": "verified"}


@pytest.mark.asyncio
async def test_breaker_opens_then_recovers_through_half_open():
    config = BreakerConfig(
        failure_rate_threshold=0.5,
        min_calls_in_window=4,
        window_size=4,
        recovery_timeout=0.2,
        half_open_max_calls=1,
        call_timeout=1.0,
    )
    breaker = CircuitBreaker("partner-test", config)

    for _ in range(4):
        with pytest.raises(RuntimeError):
            await breaker.call(_always_fails)
    assert breaker.state is BreakerState.OPEN

    with pytest.raises(CircuitOpenError):
        await breaker.call(_always_succeeds)

    await asyncio.sleep(0.25)  # let recovery_timeout elapse

    result = await breaker.call(_always_succeeds)
    assert result == {"status": "verified"}
    assert breaker.state is BreakerState.CLOSED

Beyond this unit test, run the wrapped verify_with_breaker against a mock httpx.AsyncClient transport that returns 500s for a burst of calls, confirm the breaker opens and every subsequent call resolves from cache with stale: True, then let the mock start returning 200s and confirm the half-open trial closes the breaker again. Assert on breaker.state at each stage rather than only on the returned payload — a test that only checks the response shape can pass even if the breaker tripped for the wrong reason, or never tripped at all and simply got lucky with a fast-failing mock. For integration coverage against a real or sandboxed partner endpoint, pair this with the scenarios in VRS Integration Testing with Python, which exercises the same breaker against endpoints that intentionally simulate latency, malformed responses, and connection drops rather than a hand-rolled mock.

Gotchas & Edge Cases

Most of the failure modes below only show up under real concurrency and real partner flakiness — they rarely appear in a quick manual smoke test, which is exactly why they cause incidents in production instead of in review.

  • Thundering herd on half-open. Without the half_open_max_calls guard and the lock around _admit(), every queued request storms the recovering partner simultaneously the instant the timeout elapses, and one slow response re-trips the breaker before it ever proves the endpoint is healthy.
  • Timeouts must count as failures. If a hung connection is caught somewhere upstream and silently retried without ever reaching _record(success=False), the breaker never sees the failure rate that should trip it, and your own workers keep queuing behind a partner that is effectively down.
  • Cache staleness must be surfaced, not hidden. Returning a cached status without the stale flag lets a downstream process treat a five-minute-old “verified” result as equivalent to a live one — acceptable for absorbing a brief outage, but it must be visible in the response contract.
  • One breaker per partner GLN, not one global breaker. Pooling all trading partners behind a single breaker instance means one manufacturer’s certificate expiry blocks verification traffic to every other manufacturer, which defeats the entire purpose of isolating the failure.
  • Breaker state is in-memory and process-local. A rolling deployment or a multi-worker process pool means each worker tracks its own failure window and can independently trip or reset; if you run more than one worker per partner, pair this pattern with shared metrics (or a distributed store) so operations see one coherent picture rather than N noisy ones.

Continue reading

Related VRS Response Caching