Skip to content

breaker.md

Latest commit

 

History

History
126 lines (96 loc) · 5.5 KB

breaker.md

File metadata and controls

126 lines (96 loc) · 5.5 KB
Raw

rag-breaker — circuit breakers (Step 4.4)

Per-backend circuit breakers: isolate a failing backend so a sick dependency fails fast instead of stalling every request behind a timeout.

Overview

A circuit breaker is a three-state machine:

State Behaviour
closed Normal — calls pass through; consecutive failures are counted. failure_threshold failures in a row trip it open.
open Calls are rejected immediately with CircuitOpenError (the backend is not touched). After reset_timeout_s a single trial call is admitted (half-open).
half_open Up to half_open_max_calls trial calls are admitted; success_threshold successes close the breaker, any failure re-opens it.

In the gateway, one breaker guards each retrieval backend (vector / keyword / graph). An open breaker makes its wrapped backend raise, and HybridRetriever drops that source and fuses the rest — so a single dead backend degrades gracefully instead of failing the query.

Usage

The breaker directly

from rag_breaker import CircuitBreaker, CircuitBreakerConfig, CircuitOpenError

breaker = CircuitBreaker("vector", config=CircuitBreakerConfig(failure_threshold=3))

# Run a call through the breaker — it records success/failure and fast-fails when open.
try:
    refs = await breaker.call(ctx, lambda: backend.retrieve_ids(ctx, vec, 10, [], None))
except CircuitOpenError:
    ...  # breaker is open — skip this backend

call(ctx, fn) raises CircuitOpenError before invoking fn when the breaker is open. An Exception from fn is recorded as a failure and re-raised; a BaseException such as asyncio.CancelledError is not counted (a cancellation is not a backend fault).

Lower-level API (used by call, handy for callers that own their try/except — e.g. the retrieval router): await breaker.allow() → bool, await breaker.record_success(), await breaker.record_failure(ctx).

The registry (per-backend)

from rag_breaker import BreakerRegistry, CircuitBreakerConfig

registry = BreakerRegistry(config=CircuitBreakerConfig(failure_threshold=5))
vector_breaker = registry.get("vector")        # lazily minted, then cached
snapshots = await registry.snapshot()           # list[BreakerSnapshot], ordered by name
await registry.force_close("vector")             # operator one-click recovery

SPI wrappers

from rag_breaker import BreakerVectorRetrievalBackend

guarded = BreakerVectorRetrievalBackend(real_vector_store, registry.get("vector"))
# `guarded` is a VectorRetrievalBackend — drop it straight into HybridRetriever.

Breaker{Vector,Keyword,Graph}RetrievalBackend route the retrieval call (retrieve_ids / expand) through the breaker; hydrate / query delegate straight through; health() reports unhealthy when the breaker is open.

Configuration (rag.yaml)

breakers:
  enabled: true            # default; closed breakers are pure pass-through
  failure_threshold: 5     # consecutive failures that trip a breaker open
  reset_timeout_s: 30.0    # seconds open before a half-open trial is admitted
  half_open_max_calls: 1   # trial calls admitted while half-open
  success_threshold: 1     # trial successes needed to close

CLI smoke test

$ ragctl breaker -f 3                 # closed → 3 fails → open → probe recovers → closed
$ ragctl breaker -f 2 --no-recover    # ...probe still down → re-open

Status API

  • GET /v1/status/breakers → { "breakers": [BreakerSnapshot, ...], "total": n }
  • POST /v1/status/breakers/{name}/force-close → { "breaker": BreakerSnapshot, "forced_closed": true } (404 if unknown / no registry wired)

Breaker state also folds into GET /v1/status/health: each breaker appears as a breaker:{name} component (closed → up, half_open → degraded, open → down).

Internals

  • CircuitBreakerConfig — frozen dataclass; __post_init__ rejects out-of-range values (failure_threshold ≥ 1, reset_timeout_s ≥ 0, half_open_max_calls ≥ 1, success_threshold ≥ 1).
  • State + bookkeeping — guarded by one asyncio.Lock. allow() is the only method that performs the timed open → half_open transition and reserves a trial slot, so concurrent callers can't all rush a recovering backend.
  • BreakerSnapshot (in rag_core.types) — frozen, PII-free: name, state, failure_count, failure_threshold, opened_total (lifetime trips), reset_timeout_s, seconds_until_retry (set only while open).
  • Event — breaker.opened (pre-registered; typed BreakerEvent) fires once per trip into open, carrying counts + the tripping tenant, never query/answer text. Recovery is silent.
  • Hot path — no OTel span per call (the breaker sits inside the retrieval fan-out); just ints, an enum, and the lock.
  • Lock-free reads — is_open_nowait() / snapshot_nowait() let the synchronous health roll-up read state without awaiting the lock.

Extension points

  • A real backend — wrap it in the matching Breaker*RetrievalBackend at wiring time, drawing the breaker from a shared BreakerRegistry; nothing else changes.
  • A new SPI to protect (LLM, embedder, …) — either use breaker.call(ctx, fn) at the call site, or add a thin wrapper class mirroring the retrieval wrappers.
  • Trip policy — the default trips on consecutive failures; a window- or rate-based policy can be added behind the same CircuitBreaker surface without touching callers.