Chapter 04 · ~30 minutes

Circuit Breaker, Retry Budget, and Hedging

Three patterns that, layered correctly, turn an upstream that flaps into a gateway that doesn't. We build them small and from scratch — the production primitives are an evening's work, not a framework.

Three patterns, one goal

The goal is to absorb upstream misbehavior without amplifying it. Each pattern handles a different failure shape:

  • Circuit breaker — when the upstream is clearly broken (high error rate over a window), stop calling it for a cooldown period. Saves the client from waiting, saves the upstream from being hammered while it recovers.
  • Retry budget — transient failures should be retried, but only as a small fraction of overall traffic. Naive "retry 3 times" turns one bad deploy into a 3x traffic spike, which guarantees the bad deploy stays bad.
  • Hedging — for high-tail-latency requests, fire a second copy after the p95 mark and use whichever returns first. Trades a few percent extra upstream load for a meaningfully better tail.
Why we don't reach for tower-retry / tower-balance here

The tower ecosystem has retry and load-balance layers, and they're well-engineered. We're rolling our own because: (a) writing it out makes the semantics legible — every "retry" library hides decisions that bite later; (b) hedging for streaming responses is custom enough that off-the-shelf layers don't quite fit; (c) ~150 lines of focused code is easier to reason about than 3 generic libraries composed.

Circuit breaker

Classic three states:

  • Closed — calls go through. Track error count over a rolling window.
  • Open — calls fail fast with 503. After a cooldown, transition to half-open.
  • Half-open — let exactly one probe through. If it succeeds, go back to closed. If it fails, back to open with a fresh cooldown.

Create infergw/src/breaker.rs:

use std::{
    sync::Mutex,
    time::{Duration, Instant},
};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum State { Closed, Open, HalfOpen }

pub struct CircuitBreaker {
    inner: Mutex<Inner>,
    window: Duration,
    failure_threshold: u32,
    cooldown: Duration,
}

struct Inner {
    state: State,
    failures: Vec<Instant>,
    opened_at: Option<Instant>,
    half_open_probe_in_flight: bool,
}

impl CircuitBreaker {
    pub fn new(window: Duration, failure_threshold: u32, cooldown: Duration) -> Self {
        Self {
            window,
            failure_threshold,
            cooldown,
            inner: Mutex::new(Inner {
                state: State::Closed,
                failures: Vec::with_capacity(64),
                opened_at: None,
                half_open_probe_in_flight: false,
            }),
        }
    }

    /// Returns Ok(()) if the call is allowed; Err(state) if short-circuited.
    pub fn try_acquire(&self) -> Result<Permit, State> {
        let mut g = self.inner.lock().expect("breaker poisoned");
        let now = Instant::now();
        match g.state {
            State::Closed => Ok(Permit { closed_path: true }),
            State::Open => {
                if g.opened_at.map(|t| now.duration_since(t) >= self.cooldown).unwrap_or(false) {
                    g.state = State::HalfOpen;
                    g.half_open_probe_in_flight = true;
                    Ok(Permit { closed_path: false })
                } else {
                    Err(State::Open)
                }
            }
            State::HalfOpen => {
                if g.half_open_probe_in_flight {
                    Err(State::HalfOpen)
                } else {
                    g.half_open_probe_in_flight = true;
                    Ok(Permit { closed_path: false })
                }
            }
        }
    }

    pub fn record_success(&self) {
        let mut g = self.inner.lock().expect("breaker poisoned");
        match g.state {
            State::HalfOpen => {
                g.state = State::Closed;
                g.failures.clear();
                g.opened_at = None;
                g.half_open_probe_in_flight = false;
                tracing::info!("circuit breaker: closed (probe ok)");
            }
            _ => {}
        }
    }

    pub fn record_failure(&self) {
        let mut g = self.inner.lock().expect("breaker poisoned");
        let now = Instant::now();
        // Prune failures outside the rolling window.
        let cutoff = now - self.window;
        g.failures.retain(|t| *t >= cutoff);
        g.failures.push(now);

        match g.state {
            State::Closed if g.failures.len() >= self.failure_threshold as usize => {
                g.state = State::Open;
                g.opened_at = Some(now);
                tracing::warn!(
                    failures = g.failures.len(),
                    cooldown_ms = self.cooldown.as_millis() as u64,
                    "circuit breaker: open"
                );
            }
            State::HalfOpen => {
                g.state = State::Open;
                g.opened_at = Some(now);
                g.half_open_probe_in_flight = false;
                tracing::warn!("circuit breaker: re-opened (probe failed)");
            }
            _ => {}
        }
    }

    pub fn state(&self) -> State {
        self.inner.lock().expect("breaker poisoned").state
    }
}

/// RAII-ish marker. Drop doesn't auto-record — caller must explicitly call
/// record_success or record_failure based on the outcome of the request.
pub struct Permit { pub closed_path: bool }
Don't auto-record on Drop

A tempting shortcut is to record failure when the Permit drops without a success call. Don't do it. A client disconnect mid-stream is not an upstream failure, and you'd open the breaker on the wrong signal. Explicit recording forces you to define what counts as a failure in your context.

Retry budget

The pattern: maintain a token bucket of allowed retries. A retry costs one token. Tokens refill slowly (say, 10% of the success rate). Under steady state with low error rate, retries are essentially free. Under high error rate, the bucket empties and retries stop — preventing amplification.

Create infergw/src/retry.rs:

use std::sync::Mutex;
use std::time::{Duration, Instant};

pub struct RetryBudget {
    inner: Mutex<Inner>,
    max_tokens: f64,
    refill_per_sec: f64,
}

struct Inner {
    tokens: f64,
    last_refill: Instant,
}

impl RetryBudget {
    pub fn new(max_tokens: f64, refill_per_sec: f64) -> Self {
        Self {
            max_tokens,
            refill_per_sec,
            inner: Mutex::new(Inner { tokens: max_tokens, last_refill: Instant::now() }),
        }
    }

    pub fn try_consume(&self) -> bool {
        let mut g = self.inner.lock().expect("retry budget poisoned");
        let now = Instant::now();
        let elapsed = now.duration_since(g.last_refill).as_secs_f64();
        g.tokens = (g.tokens + elapsed * self.refill_per_sec).min(self.max_tokens);
        g.last_refill = now;

        if g.tokens >= 1.0 {
            g.tokens -= 1.0;
            true
        } else {
            false
        }
    }

    pub fn remaining(&self) -> f64 {
        self.inner.lock().expect("retry budget poisoned").tokens
    }
}

/// Exponential backoff with full jitter, capped.
pub fn backoff(attempt: u32, base: Duration, max: Duration) -> Duration {
    use rand::Rng;
    let exp = base.saturating_mul(1u32 << attempt.min(10));
    let cap = exp.min(max);
    let jitter_ms = rand::thread_rng().gen_range(0..=cap.as_millis() as u64);
    Duration::from_millis(jitter_ms)
}

Add rand = "0.8" to the dependencies.

The rule of thumb is: only retry idempotent failures (connection refused, 5xx, timeouts on connect), never retry an already-streaming response (you'd duplicate tokens to the client), and never retry 4xx. Streaming complicates this — once a single byte of the response body has been written to the client, retry is off the table. We codify that explicitly in the proxy.

Hedging

The idea is simple: if a request hasn't responded with its first chunk by some threshold (we'll use a configurable p95 estimate), fire a second copy in parallel. Whichever returns the first chunk first wins; cancel the other.

For streaming responses, the subtlety is "first byte" — once the first chunk arrives from either upstream call, commit to that one. Cancel the other immediately so we don't pay for the entire second response.

Create infergw/src/hedge.rs:

use bytes::Bytes;
use futures::{Stream, StreamExt};
use std::{pin::Pin, time::Duration};
use tokio::sync::oneshot;

type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>;

/// Fire a primary request immediately. If `hedge_after` elapses with no first
/// chunk yet, fire a backup. Return the stream from whichever yields a first
/// chunk first; the loser is cancelled by dropping its future.
pub async fn hedged<F>(
    spawn_request: F,
    hedge_after: Duration,
) -> Result<ByteStream, std::io::Error>
where
    F: Fn() -> ByteStream + Send + Sync + 'static,
{
    let primary = spawn_request();
    let (winner_tx, winner_rx) = oneshot::channel::<(Bytes, ByteStream)>();

    let winner_tx = std::sync::Arc::new(tokio::sync::Mutex::new(Some(winner_tx)));
    let tx_a = winner_tx.clone();
    let tx_b = winner_tx.clone();

    tokio::spawn(async move {
        race(primary, tx_a).await;
    });

    tokio::spawn(async move {
        tokio::time::sleep(hedge_after).await;
        // Only fire the hedge if no one has won yet.
        if winner_tx.lock().await.is_some() {
            tracing::info!("hedge: firing backup request");
            let backup = spawn_request();
            race(backup, tx_b).await;
        }
    });

    match winner_rx.await {
        Ok((first_chunk, rest)) => {
            let prepend = futures::stream::iter(std::iter::once(Ok(first_chunk)));
            Ok(Box::pin(prepend.chain(rest)) as ByteStream)
        }
        Err(_) => Err(std::io::Error::other("both hedged requests failed")),
    }
}

async fn race(mut s: ByteStream, tx: std::sync::Arc<tokio::sync::Mutex<Option<oneshot::Sender<(Bytes, ByteStream)>>>>) {
    if let Some(Ok(first)) = s.next().await {
        let mut guard = tx.lock().await;
        if let Some(sender) = guard.take() {
            let _ = sender.send((first, s));
        }
        // If guard was None, someone else already won — drop our stream, cancelling.
    }
}
Why hedging is so effective

In a fleet of upstreams, individual workers occasionally pause for GC, model paging, KV-cache eviction, or just an unlucky load spike. Per-worker tail latency is bursty. Firing a hedge after p95 means: if the primary was about to be slow, the hedge has a high chance of hitting a fresh worker. Studies inside Google ("Tail at Scale") show 30-50% p99 reductions for ~5% extra load.

When hedging is bad

Hedging is bad when the upstream is already overloaded — you're adding load while it's struggling. The circuit breaker prevents the worst version (if upstream is dying, breaker opens and no hedge fires), but you should still gate hedging behind a "only when upstream is healthy" check. We do that below.

Wire it all into the proxy

Update AppState to carry the breaker and budget:

use crate::{breaker::CircuitBreaker, coalesce::Coalescer, retry::RetryBudget, upstream::Upstream};
use std::sync::Arc;
// ... existing imports

#[derive(Clone)]
pub struct AppState {
    pub ready: Arc<std::sync::atomic::AtomicBool>,
    pub upstream: Upstream,
    pub coalescer: Coalescer,
    pub breaker: Arc<CircuitBreaker>,
    pub retry_budget: Arc<RetryBudget>,
    pub hedge_after_ms: u64,
}

impl AppState {
    pub fn new(upstream: Upstream, hedge_after_ms: u64) -> Self {
        use std::time::Duration;
        Self {
            ready: Arc::new(std::sync::atomic::AtomicBool::new(true)),
            upstream,
            coalescer: Coalescer::default(),
            breaker: Arc::new(CircuitBreaker::new(
                Duration::from_secs(30),
                10,
                Duration::from_secs(10),
            )),
            retry_budget: Arc::new(RetryBudget::new(10.0, 0.5)),
            hedge_after_ms,
        }
    }
}

Add a config field for hedging:

// in config.rs
pub hedge_after_ms: u64,
// in from_env():
hedge_after_ms: env::var("INFERGW_HEDGE_AFTER_MS")
    .ok().and_then(|s| s.parse().ok()).unwrap_or(800),

Update the leader spawner in proxy.rs to consult the breaker and (optionally) hedge. Replace spawn_leader:

use crate::breaker::State as BreakerState;

fn spawn_leader(state: AppState, key: String, body: Value, inflight: Arc<InFlight>) {
    tokio::spawn(async move {
        // 1) Check the breaker before making any call.
        let permit = match state.breaker.try_acquire() {
            Ok(p) => p,
            Err(s) => {
                tracing::warn!(?s, "breaker short-circuited request");
                let _ = inflight.tx.send(Frame::Error(format!("circuit {s:?}")));
                state.coalescer.remove(&key);
                return;
            }
        };

        // 2) Decide whether to hedge: only when breaker is fully closed,
        //    and only for streaming requests (which we're already in).
        let allow_hedge = permit.closed_path
            && state.breaker.state() == BreakerState::Closed
            && state.hedge_after_ms > 0;

        let outcome = if allow_hedge {
            do_hedged_upstream(&state, &body).await
        } else {
            do_single_upstream(&state, &body).await
        };

        match outcome {
            Ok(mut stream) => {
                state.breaker.record_success();
                while let Some(chunk) = stream.next().await {
                    match chunk {
                        Ok(b) => {
                            inflight.replay.lock().await.push(Frame::Chunk(b.clone()));
                            let _ = inflight.tx.send(Frame::Chunk(b));
                        }
                        Err(e) => {
                            let _ = inflight.tx.send(Frame::Error(format!("chunk: {e}")));
                            break;
                        }
                    }
                }
                inflight.replay.lock().await.push(Frame::Done);
                let _ = inflight.tx.send(Frame::Done);
            }
            Err(e) => {
                state.breaker.record_failure();
                let _ = inflight.tx.send(Frame::Error(format!("upstream: {e}")));
            }
        }
        state.coalescer.remove(&key);
    });
}

async fn do_single_upstream(
    state: &AppState,
    body: &Value,
) -> Result<Pin<Box<dyn futures::Stream<Item = Result<Bytes, std::io::Error>> + Send>>, String> {
    let url = state.upstream.chat_completions_url();
    let resp = state.upstream.client.post(&url).json(body).send().await
        .map_err(|e| format!("send: {e}"))?;
    if !resp.status().is_success() {
        return Err(format!("upstream status {}", resp.status()));
    }
    let s = resp.bytes_stream().map(|c| c.map_err(std::io::Error::other));
    Ok(Box::pin(s))
}

async fn do_hedged_upstream(
    state: &AppState,
    body: &Value,
) -> Result<Pin<Box<dyn futures::Stream<Item = Result<Bytes, std::io::Error>> + Send>>, String> {
    use std::time::Duration;
    let state2 = state.clone();
    let body2 = body.clone();
    let spawn_one = move || -> Pin<Box<dyn futures::Stream<Item = Result<Bytes, std::io::Error>> + Send>> {
        let state = state2.clone();
        let body = body2.clone();
        Box::pin(async_stream::stream! {
            match do_single_upstream(&state, &body).await {
                Ok(mut s) => {
                    while let Some(item) = s.next().await { yield item; }
                }
                Err(e) => { yield Err(std::io::Error::other(e)); }
            }
        })
    };
    crate::hedge::hedged(spawn_one, Duration::from_millis(state.hedge_after_ms))
        .await
        .map_err(|e| e.to_string())
}

Add std::pin::Pin to the imports and register the new modules in main.rs:

mod breaker;
mod hedge;
mod retry;

And in main() pass the config through:

let state = routes::AppState::new(upstream, cfg.hedge_after_ms);

Make the mock misbehave

Let's prove the breaker trips and recovers. Add a controllable failure mode to the mock — edit mock-upstream/src/main.rs to fail when an env var asks:

use std::sync::atomic::{AtomicUsize, Ordering};
static REQS: AtomicUsize = AtomicUsize::new(0);

async fn chat(/* same args */) -> impl axum::response::IntoResponse {
    let n = REQS.fetch_add(1, Ordering::Relaxed);
    let fail_every: usize = std::env::var("FAIL_EVERY").ok()
        .and_then(|s| s.parse().ok()).unwrap_or(0);
    if fail_every > 0 && n % fail_every == 0 {
        return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, "synthetic failure").into_response();
    }
    // ... existing SSE response code, but call .into_response() at the end
}

(You'll need to refactor the existing handler to return Response instead of Sse<...>; the body of the function is unchanged. The shape is: wrap the existing Sse::new(stream).keep_alive(...) in .into_response().)

Run the mock with deterministic failures:

FAIL_EVERY=1 cargo run -p mock-upstream

Now every request fails. Hit the gateway 12 times in a loop:

for i in $(seq 1 12); do
  curl -sN -o /dev/null -w "%{http_code}\n" \
    -H 'content-type: application/json' \
    -d '{"model":"x","messages":[{"role":"user","content":"hi"}],"stream":true}' \
    http://127.0.0.1:8080/v1/chat/completions
done

You'll see the first ~10 succeed at the HTTP layer (200 with an error frame in the stream) — those failures accumulate. After 10 failures in the 30-second window, the breaker opens. Subsequent requests get an immediate error frame from the gateway without touching the upstream. Watch the gateway logs:

WARN infergw::breaker: circuit breaker: open failures=10 cooldown_ms=10000
WARN infergw::proxy: breaker short-circuited request s=Open

Now stop the mock failures: Ctrl-C the mock and restart it with cargo run -p mock-upstream (no FAIL_EVERY). After the 10-second cooldown, the next request becomes the half-open probe, succeeds, and the breaker closes.

INFO infergw::breaker: circuit breaker: closed (probe ok)

Tradeoffs to internalize

  • Breaker threshold — too low and you flap on minor blips; too high and the upstream gets pummeled before you back off. 10 failures in 30s is a starting point. Tune from real prod metrics.
  • Retry budget — capped retries prevent amplification, but they also mean a small fraction of requests will see un-retried failures during incidents. That's the correct tradeoff: better that 1% of requests fail than that 100% fail because you DDoS'd your own backend with retries.
  • Hedging cost — at hedge_after=p95, you fire a hedge on ~5% of requests. Token cost matters when the upstream is a paid API: count carefully, gate by tier, or skip hedging for low-tier users.
  • Order matters — apply breaker first (cheap, refuses fast), then retry budget, then hedging. Don't hedge through an open breaker.
Stretch: per-upstream breakers

In a real fleet, you'd have multiple upstream addresses (one per vLLM replica). Keep a HashMap<String, CircuitBreaker> keyed by upstream address. When you pick an upstream for a request, consult its individual breaker. A single sick replica trips its own breaker without taking the rest of the fleet down — and load-balancing naturally routes around it. This is one of the things tower::balance does, well, when you commit to using it.

Stretch: tie hedging to a real latency histogram

Right now hedge_after_ms is configured statically. Chapter 05 wires real latency histograms; once those exist, you can read p95 dynamically and set hedge_after from observed behavior, not a guess. Be careful: feedback loops where bad latency triggers more hedging which causes more latency are a thing. Cap the fraction of requests that can hedge (e.g. 10%) regardless of what the histogram says.

Checkpoint

  • Circuit breaker opens after 10 failures in 30s, transitions to half-open after 10s cooldown, closes on probe success.
  • Hedging fires a backup request after the configured threshold (default 800ms).
  • Retry budget exists and is wired through AppState (full retry-on-connect-error implementation is left as an exercise — the primitives are all there).
  • The mock can be made to fail deterministically with FAIL_EVERY.

Reliability layer done. Time to see what's actually happening.

→ Chapter 05: Prometheus metrics and OpenTelemetry traces