4. Circuit Breakers

Description

With retry you handle transient failures, but what happens when a service has been down for 10 minutes? Your retry will keep retrying — and each retry with 3 attempts means 3 failed calls. For 100 concurrent users, that's 300 useless calls per minute, all processing exceptions, all waiting on timeouts. The circuit breaker detects that the service is down and stops trying, failing immediately until the service recovers. In this capsule you'll implement a thread-safe circuit breaker from scratch, wrap it as a provider, and test each state transition.

The motivation: the cost of blind retry

Scenario: OpenAI has a 30-minute outage
Without circuit breaker:
  - Each request: 3 retries × 5s each = 15s of waiting
  - 100 users/minute × 15s of processing = 1500s of CPU/minute
  - 100 users × 3 retries = 300 requests to OpenAI/minute (all failing)
  - OpenAI sees 9000 failed requests in 30 minutes
  - Your server: 1500s CPU/minute processing timeouts
  - Users: 15s of waiting + error 500

With circuit breaker:
  - First minute: 5 failures → circuit OPENS
  - Next 29 minutes: fails immediately (<1ms), without calling OpenAI
  - Users: fallback response in <1ms
  - Your server: 0 CPU processing timeouts
  - OpenAI: 5 requests (the ones that tripped the circuit)

The three states of the circuit breaker

                    ┌─────────────────────────────────────┐
                    │           CLOSED                    │
                    │  (Normal state)                     │
                    │  Requests pass normally             │
                    │  Failures are counted               │
                    └──────────────┬──────────────────────┘
                                   │
                    failures >= threshold
                                   │
                                   ▼
                    ┌─────────────────────────────────────┐
                    │             OPEN                    │
                    │  (Service outage detected)          │
                    │  Requests FAIL IMMEDIATELY          │
                    │  Without calling the API            │
                    └──────────────┬──────────────────────┘
                                   │
                    recovery_timeout seconds
                                   │
                                   ▼
                    ┌─────────────────────────────────────┐
                    │           HALF-OPEN                 │
                    │  (Testing recovery)                 │
                    │  ONE test request passes            │
                    │  If OK → CLOSED (recovered)         │
                    │  If it fails → OPEN (still down)    │
                    └─────────────────────────────────────┘

Key parameters:
  failure_threshold: how many consecutive failures to open
  recovery_timeout: how many seconds to wait before probing
  success_threshold: how many successes in half-open to close

Implementation from scratch

# src/infrastructure/circuit_breaker.py
import threading
from datetime import datetime, timedelta
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional, Callable
import structlog

log = structlog.get_logger()

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitOpenError(Exception):
    """Raised when the circuit is open and a call is attempted."""
    def __init__(self, circuit_name: str, reset_at: datetime):
        self.circuit_name = circuit_name
        self.reset_at = reset_at
        seconds_until_reset = (reset_at - datetime.now()).total_seconds()
        super().__init__(
            f"Circuit '{circuit_name}' is OPEN. "
            f"Will try to reset in {seconds_until_reset:.0f}s."
        )

@dataclass
class CircuitBreakerStats:
    """Circuit breaker statistics for monitoring."""
    total_calls: int = 0
    successful_calls: int = 0
    failed_calls: int = 0
    rejected_calls: int = 0  # Rejected when the circuit was open
    circuit_opened_count: int = 0
    last_opened_at: Optional[datetime] = None
    last_closed_at: Optional[datetime] = None

class CircuitBreaker:
    """
    Circuit breaker to protect calls to external services.
    
    Thread-safe: uses threading.Lock for state operations.
    
    Parameters:
    - name: identifier for logs and metrics
    - failure_threshold: consecutive failures to open the circuit
    - recovery_timeout: seconds in OPEN state before half-open
    - success_threshold: successes in half-open to close the circuit
    - expected_exceptions: which exceptions count as failures
    """
    
    def __init__(
        self,
        name: str,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        success_threshold: int = 1,
        expected_exceptions: tuple = (Exception,)
    ):
        self.name = name
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        self.expected_exceptions = expected_exceptions
        
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._success_count_in_half_open = 0
        self._last_failure_time: Optional[datetime] = None
        self._lock = threading.Lock()
        self.stats = CircuitBreakerStats()
    
    @property
    def state(self) -> CircuitState:
        with self._lock:
            return self._get_state()
    
    def _get_state(self) -> CircuitState:
        """Evaluate the current state, including the OPEN → HALF_OPEN transition."""
        if self._state == CircuitState.OPEN:
            if (self._last_failure_time and
                datetime.now() - self._last_failure_time >= timedelta(seconds=self.recovery_timeout)):
                # The recovery time passed → try again
                self._state = CircuitState.HALF_OPEN
                self._success_count_in_half_open = 0
                log.info(
                    "circuit_breaker_half_open",
                    circuit=self.name,
                    recovery_timeout=self.recovery_timeout
                )
        return self._state
    
    def call(self, func: Callable, *args, **kwargs):
        """
        Execute the function with circuit breaker protection.
        
        If the circuit is OPEN, raise CircuitOpenError immediately.
        If it's CLOSED or HALF_OPEN, execute the function.
        """
        with self._lock:
            current_state = self._get_state()
            
            if current_state == CircuitState.OPEN:
                self.stats.rejected_calls += 1
                reset_at = self._last_failure_time + timedelta(seconds=self.recovery_timeout)
                log.warning(
                    "circuit_breaker_rejected_call",
                    circuit=self.name,
                    state=current_state.value
                )
                raise CircuitOpenError(self.name, reset_at)
            
            self.stats.total_calls += 1
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except self.expected_exceptions as e:
            self._on_failure()
            raise
    
    def _on_success(self):
        with self._lock:
            if self._state == CircuitState.HALF_OPEN:
                self._success_count_in_half_open += 1
                if self._success_count_in_half_open >= self.success_threshold:
                    self._close_circuit()
            elif self._state == CircuitState.CLOSED:
                # Reset failure count on success
                self._failure_count = 0
            self.stats.successful_calls += 1
    
    def _on_failure(self):
        with self._lock:
            self._failure_count += 1
            self._last_failure_time = datetime.now()
            self.stats.failed_calls += 1
            
            if self._state == CircuitState.HALF_OPEN:
                # A failure in half-open reopens immediately
                self._open_circuit()
            elif self._failure_count >= self.failure_threshold:
                self._open_circuit()
    
    def _open_circuit(self):
        """Open the circuit. Assumes it's called within the lock."""
        previous_state = self._state
        self._state = CircuitState.OPEN
        self.stats.circuit_opened_count += 1
        self.stats.last_opened_at = datetime.now()
        
        log.warning(
            "circuit_breaker_opened",
            circuit=self.name,
            failure_count=self._failure_count,
            failure_threshold=self.failure_threshold,
            previous_state=previous_state.value,
            recovery_timeout=self.recovery_timeout
        )
    
    def _close_circuit(self):
        """Close the circuit. Assumes it's called within the lock."""
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._success_count_in_half_open = 0
        self.stats.last_closed_at = datetime.now()
        
        log.info(
            "circuit_breaker_closed",
            circuit=self.name,
            circuit_was_open_for_seconds=(
                (datetime.now() - self.stats.last_opened_at).total_seconds()
                if self.stats.last_opened_at else None
            )
        )
    
    def force_open(self):
        """For testing: forces the state to OPEN."""
        with self._lock:
            self._open_circuit()
    
    def force_close(self):
        """For testing: forces the state to CLOSED."""
        with self._lock:
            self._close_circuit()
    
    def get_metrics(self) -> dict:
        """Returns metrics for monitoring."""
        with self._lock:
            return {
                "circuit": self.name,
                "state": self._get_state().value,
                "failure_count": self._failure_count,
                "failure_threshold": self.failure_threshold,
                "stats": {
                    "total_calls": self.stats.total_calls,
                    "successful_calls": self.stats.successful_calls,
                    "failed_calls": self.stats.failed_calls,
                    "rejected_calls": self.stats.rejected_calls,
                    "circuit_opened_count": self.stats.circuit_opened_count,
                }
            }

CircuitBreakerProvider: wrapping the LLMProvider

# src/infrastructure/circuit_breaker_provider.py
import structlog
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
from src.infrastructure.circuit_breaker import CircuitBreaker, CircuitOpenError
from src.infrastructure.error_classifier import ErrorCategory

log = structlog.get_logger()

class CircuitBreakerProvider:
    """
    Wrapper that adds circuit breaker protection to any LLMProvider.
    
    When the inner provider fails repeatedly, the circuit opens
    and subsequent calls fail immediately without calling the API.
    """
    
    def __init__(
        self,
        inner: LLMProvider,
        circuit_breaker: CircuitBreaker = None,
        failure_threshold: int = 5,
        recovery_timeout: int = 60
    ):
        self._inner = inner
        self._circuit = circuit_breaker or CircuitBreaker(
            name=f"circuit_{type(inner).__name__}",
            failure_threshold=failure_threshold,
            recovery_timeout=recovery_timeout
        )
    
    def complete(self, messages: list[dict], **kwargs) -> str:
        try:
            return self._circuit.call(self._inner.complete, messages, **kwargs)
        except CircuitOpenError as e:
            # Convert CircuitOpenError to LLMProviderError with OUTAGE category
            raise LLMProviderError(
                message=f"Circuit open — service temporarily unavailable",
                original_error=e,
                category=ErrorCategory.OUTAGE,
                should_retry=False  # No retry when the circuit is open
            )
    
    @property
    def is_open(self) -> bool:
        from src.infrastructure.circuit_breaker import CircuitState
        return self._circuit.state == CircuitState.OPEN
    
    def get_metrics(self) -> dict:
        return self._circuit.get_metrics()

Circuit breaker tests

# tests/unit/test_circuit_breaker.py
import pytest
import time
from src.infrastructure.circuit_breaker import (
    CircuitBreaker, CircuitState, CircuitOpenError
)
from src.infrastructure.llm_provider import LLMProviderError
from src.infrastructure.error_classifier import ErrorCategory

class TestCircuitBreaker:
    
    def test_starts_closed(self):
        cb = CircuitBreaker("test", failure_threshold=3, recovery_timeout=60)
        assert cb.state == CircuitState.CLOSED
    
    def test_opens_after_threshold_failures(self):
        cb = CircuitBreaker("test", failure_threshold=3, recovery_timeout=60)
        
        def always_fail():
            raise ValueError("failure")
        
        for _ in range(3):
            with pytest.raises(ValueError):
                cb.call(always_fail)
        
        assert cb.state == CircuitState.OPEN
    
    def test_rejects_calls_when_open(self):
        """When the circuit is open, it must reject without calling the function."""
        cb = CircuitBreaker("test", failure_threshold=3, recovery_timeout=60)
        cb.force_open()
        
        call_count = 0
        def track_calls():
            nonlocal call_count
            call_count += 1
            return "result"
        
        with pytest.raises(CircuitOpenError):
            cb.call(track_calls)
        
        assert call_count == 0  # The function was not called
        assert cb.stats.rejected_calls == 1
    
    def test_transitions_to_half_open_after_timeout(self):
        """After the recovery_timeout, it must move to HALF_OPEN."""
        cb = CircuitBreaker("test", failure_threshold=2, recovery_timeout=0)
        cb.force_open()
        
        # Simulate that time passed (recovery_timeout=0 allows this)
        state = cb.state  # Triggers the timeout evaluation
        
        assert state == CircuitState.HALF_OPEN
    
    def test_closes_after_success_in_half_open(self):
        """A success in HALF_OPEN must close the circuit."""
        cb = CircuitBreaker("test", failure_threshold=2, recovery_timeout=0)
        cb.force_open()
        
        # Force transition to HALF_OPEN
        _ = cb.state
        
        def succeed():
            return "ok"
        
        cb.call(succeed)
        
        assert cb.state == CircuitState.CLOSED
    
    def test_returns_to_open_on_failure_in_half_open(self):
        """A failure in HALF_OPEN returns to OPEN."""
        cb = CircuitBreaker("test", failure_threshold=2, recovery_timeout=0)
        cb.force_open()
        _ = cb.state  # Trigger half-open
        
        def fail_again():
            raise ValueError("still failing")
        
        with pytest.raises(ValueError):
            cb.call(fail_again)
        
        assert cb.state == CircuitState.OPEN
    
    def test_metrics_track_correctly(self):
        """The metrics must reflect the state correctly."""
        cb = CircuitBreaker("test", failure_threshold=3, recovery_timeout=60)
        
        # 1 success
        cb.call(lambda: "ok")
        
        # 2 failures
        for _ in range(2):
            with pytest.raises(ValueError):
                cb.call(lambda: (_ for _ in ()).throw(ValueError("fail")))
        
        metrics = cb.get_metrics()
        assert metrics["stats"]["successful_calls"] == 1
        assert metrics["stats"]["failed_calls"] == 2
        assert metrics["state"] == CircuitState.CLOSED.value  # Still closed

Exercises

Exercise 1: The right threshold

What failure_threshold would you use for each scenario?

1. A critical API that handles payments (a false positive is very costly)
2. A non-critical analysis LLM
3. A health check endpoint

Exercise 2: Composition with retry

Which one should be more "on the outside" in the composition: retry or circuit breaker?

# Option A:
provider = CircuitBreakerProvider(RetryProvider(base_provider))
# Option B:
provider = RetryProvider(CircuitBreakerProvider(base_provider))

Exercise 3: Designing the recovery

Your circuit breaker has recovery_timeout=60 and success_threshold=1. After a 20-minute OpenAI outage, the service is restored. Describe step by step what happens:

1. When does it move to HALF_OPEN?
2. Which request is the first to pass?
3. What happens if that request fails?
4. What happens if that request works?

Exercise 4: Metrics test

Write a test that verifies that after: 2 successful calls + 5 failed (with threshold=5) + 3 rejected, the circuit breaker metrics are correct:

def test_metrics_complete_scenario():
    cb = CircuitBreaker("test", failure_threshold=5, recovery_timeout=60)
    
    # 2 successes
    for _ in range(2):
        cb.call(lambda: "ok")
    
    # 5 failures → opens the circuit
    for _ in range(5):
        with pytest.raises(ValueError):
            cb.call(lambda: (_ for _ in ()).throw(ValueError("fail")))
    
    assert cb.state == CircuitState.OPEN
    
    # 3 rejected calls
    for _ in range(3):
        with pytest.raises(CircuitOpenError):
            cb.call(lambda: "should not run")
    
    metrics = cb.get_metrics()
    assert metrics["stats"]["successful_calls"] == 2
    assert metrics["stats"]["failed_calls"] == 5
    assert metrics["stats"]["rejected_calls"] == 3
    assert metrics["stats"]["circuit_opened_count"] == 1
    assert metrics["state"] == "open"

Troubleshooting

"My circuit breaker opens with a single error"

Check your failure_threshold. If it's set to 1, any error opens it. For most LLM services, a threshold of 5-7 is reasonable. Isolated errors are normal — only patterns of repeated failures indicate a real outage.

"The circuit never moves to HALF_OPEN"

Check two things: (1) that recovery_timeout isn't too high (e.g., 3600 = 1 hour), and (2) that the state evaluation happens. The CircuitBreaker evaluates the timeout when you access state or when you make a call. If nobody calls, it isn't evaluated. In production this isn't a problem because there are always requests coming in.

"I get deadlocks with the threading.Lock"

The CircuitBreaker uses a single threading.Lock. If your code does something like circuit.call(lambda: circuit.get_metrics()) — that is, it calls the circuit breaker within itself — you can get a deadlock. Never nest calls to the circuit breaker. If you need metrics within a call, use a threading.RLock (reentrant lock) instead of Lock.

"Should I have one circuit breaker per model or one global?"

One per model/provider. If gpt-4o fails, you don't want the circuit to also block calls to gpt-4o-mini. In dependencies.py you create separate instances: _primary_circuit_breaker and _fallback_circuit_breaker.

Summary

• The problem: blind retry during outages wastes resources and degrades UX
• Circuit breaker states: CLOSED (normal) → OPEN (fails fast) → HALF-OPEN (tests recovery)
• Key parameters: failure_threshold (when to open), recovery_timeout (when to probe)
• Composition: circuit breaker outside, retry inside
• Fail fast: when the circuit is OPEN, it fails in <1ms without calling the API
• Thread-safe: the CircuitBreaker uses threading.Lock to be safe in concurrent environments

Additional resources

1. Circuit Breaker Pattern (Martin Fowler) — The canonical article
2. pybreaker — Alternative ready-to-use implementation
3. Release It! (Michael Nygard) — The origin of the pattern
4. Microsoft Azure — Circuit Breaker Pattern — Variants and trade-offs