4. Circuit Breakers

Descripción

Con retry manejas failures transitorios, pero ¿qué pasa cuando un servicio lleva 10 minutos caído? Tu retry seguirá reintentando — y cada retry con 3 intentos significa 3 llamadas fallidas. Para 100 usuarios concurrentes, eso son 300 llamadas inútiles por minuto, todas procesando excepciones, todas esperando timeouts. El circuit breaker detecta que el servicio está caído y deja de intentar, fallando inmediatamente hasta que el servicio se recupere. En esta cápsula vas a implementar un circuit breaker thread-safe desde cero, wrapearlo como provider, y testear cada transición de estado.

La motivación: el costo del retry ciego

Escenario: OpenAI tiene un outage de 30 minutos
Sin circuit breaker:
  - Cada request: 3 retries × 5s cada uno = 15s de espera
  - 100 usuarios/minuto × 15s de procesamiento = 1500s de CPU/minuto
  - 100 usuarios × 3 retries = 300 requests a OpenAI/minuto (todos fallando)
  - OpenAI ve 9000 requests fallidos en 30 minutos
  - Tu servidor: 1500s CPU/minuto procesando timeouts
  - Usuarios: 15s de espera + error 500

Con circuit breaker:
  - Primer minuto: 5 failures → circuit ABRE
  - Siguientes 29 minutos: falla inmediatamente (<1ms), sin llamar a OpenAI
  - Usuarios: respuesta de fallback en <1ms
  - Tu servidor: 0 CPU procesando timeouts
  - OpenAI: 5 requests (los que activaron el circuit)

Los tres estados del circuit breaker

                    ┌─────────────────────────────────────┐
                    │           CLOSED                    │
                    │  (Estado normal)                    │
                    │  Requests pasan normalmente          │
                    │  Se cuentan failures                 │
                    └──────────────┬──────────────────────┘
                                   │
                    failures >= threshold
                                   │
                                   ▼
                    ┌─────────────────────────────────────┐
                    │             OPEN                    │
                    │  (Servicio caído detectado)         │
                    │  Requests FALLAN INMEDIATAMENTE     │
                    │  Sin llamar a la API                │
                    └──────────────┬──────────────────────┘
                                   │
                    recovery_timeout segundos
                                   │
                                   ▼
                    ┌─────────────────────────────────────┐
                    │           HALF-OPEN                 │
                    │  (Probando recuperación)            │
                    │  UN request de prueba pasa          │
                    │  Si OK → CLOSED (recuperado)        │
                    │  Si falla → OPEN (sigue caído)      │
                    └─────────────────────────────────────┘

Parámetros clave:
  failure_threshold: cuántos failures consecutivos para abrir
  recovery_timeout: cuántos segundos esperar antes de probar
  success_threshold: cuántos successes en half-open para cerrar

Implementación desde cero

# 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):
    """Se lanza cuando el circuit está abierto y se intenta hacer una llamada."""
    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:
    """Estadísticas del circuit breaker para monitoring."""
    total_calls: int = 0
    successful_calls: int = 0
    failed_calls: int = 0
    rejected_calls: int = 0  # Rechazados cuando el circuit estaba abierto
    circuit_opened_count: int = 0
    last_opened_at: Optional[datetime] = None
    last_closed_at: Optional[datetime] = None

class CircuitBreaker:
    """
    Circuit breaker para proteger llamadas a servicios externos.
    
    Thread-safe: usa threading.Lock para operaciones de estado.
    
    Parámetros:
    - name: identificador para logs y métricas
    - failure_threshold: failures consecutivos para abrir el circuit
    - recovery_timeout: segundos en estado OPEN antes de half-open
    - success_threshold: successes en half-open para cerrar el circuit
    - expected_exceptions: qué excepciones cuentan como 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:
        """Evalúa el estado actual, incluyendo la transición OPEN → HALF_OPEN."""
        if self._state == CircuitState.OPEN:
            if (self._last_failure_time and
                datetime.now() - self._last_failure_time >= timedelta(seconds=self.recovery_timeout)):
                # El tiempo de recovery pasó → probar de nuevo
                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):
        """
        Ejecuta la función con protección del circuit breaker.
        
        Si el circuit está OPEN, lanza CircuitOpenError inmediatamente.
        Si está CLOSED o HALF_OPEN, ejecuta la función.
        """
        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:
                # Un failure en half-open vuelve a abrir inmediatamente
                self._open_circuit()
            elif self._failure_count >= self.failure_threshold:
                self._open_circuit()
    
    def _open_circuit(self):
        """Abre el circuit. Asume que se llama dentro del 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):
        """Cierra el circuit. Asume que se llama dentro del 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):
        """Para testing: fuerza el estado a OPEN."""
        with self._lock:
            self._open_circuit()
    
    def force_close(self):
        """Para testing: fuerza el estado a CLOSED."""
        with self._lock:
            self._close_circuit()
    
    def get_metrics(self) -> dict:
        """Devuelve métricas para 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 del 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 que añade circuit breaker protection a cualquier LLMProvider.
    
    Cuando el inner provider falla repetidamente, el circuit se abre
    y las llamadas subsiguientes fallan inmediatamente sin llamar a la 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:
            # Convertir CircuitOpenError a LLMProviderError con categoría OUTAGE
            raise LLMProviderError(
                message=f"Circuit open — servicio temporalmente no disponible",
                original_error=e,
                category=ErrorCategory.OUTAGE,
                should_retry=False  # No retry cuando el circuit está abierto
            )
    
    @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()

Tests del circuit breaker

# 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):
        """Cuando el circuit está abierto, debe rechazar sin llamar a la función."""
        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  # No se llamó a la función
        assert cb.stats.rejected_calls == 1
    
    def test_transitions_to_half_open_after_timeout(self):
        """Después del recovery_timeout, debe pasar a HALF_OPEN."""
        cb = CircuitBreaker("test", failure_threshold=2, recovery_timeout=0)
        cb.force_open()
        
        # Simular que pasó el tiempo (recovery_timeout=0 permite esto)
        state = cb.state  # Trigea la evaluación del timeout
        
        assert state == CircuitState.HALF_OPEN
    
    def test_closes_after_success_in_half_open(self):
        """Un success en HALF_OPEN debe cerrar el circuit."""
        cb = CircuitBreaker("test", failure_threshold=2, recovery_timeout=0)
        cb.force_open()
        
        # Forzar transición a 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):
        """Un failure en HALF_OPEN vuelve a 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):
        """Las métricas deben reflejar el estado correctamente."""
        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

Ejercicios

Ejercicio 1: El umbral correcto

¿Qué failure_threshold usarías para cada escenario?

1. Una API crítica que maneja pagos (falso positivo muy costoso)
2. Un LLM de análisis no crítico
3. Un endpoint de health check

Ejercicio 2: Composición con retry

¿Cuál debe estar más "afuera" en la composición: retry o circuit breaker?

# Opción A:
provider = CircuitBreakerProvider(RetryProvider(base_provider))
# Opción B:
provider = RetryProvider(CircuitBreakerProvider(base_provider))

Ejercicio 3: Diseñar el recovery

Tu circuit breaker tiene recovery_timeout=60 y success_threshold=1. Después de un outage de OpenAI de 20 minutos, el servicio se restaura. Describe paso a paso qué ocurre:

1. ¿Cuándo pasa a HALF_OPEN?
2. ¿Qué request es el primero en pasar?
3. ¿Qué pasa si ese request falla?
4. ¿Qué pasa si ese request funciona?

Ejercicio 4: Test de métricas

Escribe un test que verifique que después de: 2 llamadas exitosas + 5 fallidas (con threshold=5) + 3 rechazadas, las métricas del circuit breaker sean correctas:

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 → abre el 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

"Mi circuit breaker se abre con un solo error"

Verifica tu failure_threshold. Si está en 1, cualquier error lo abre. Para la mayoría de servicios LLM, un threshold de 5-7 es razonable. Los errores aislados son normales — solo los patterns de failures repetidos indican un outage real.

"El circuit nunca pasa a HALF_OPEN"

Revisa dos cosas: (1) que recovery_timeout no sea demasiado alto (ej: 3600 = 1 hora), y (2) que la evaluación del estado ocurra. El CircuitBreaker evalúa el timeout cuando accedes a state o cuando haces una llamada. Si nadie llama, no se evalúa. En producción esto no es problema porque siempre hay requests entrando.

"Tengo deadlocks con el threading.Lock"

El CircuitBreaker usa un solo threading.Lock. Si tu código hace algo como circuit.call(lambda: circuit.get_metrics()) — es decir, llama al circuit breaker dentro de sí mismo — puedes tener un deadlock. Nunca anides llamadas al circuit breaker. Si necesitas métricas dentro de una llamada, usa un threading.RLock (reentrant lock) en vez de Lock.

"¿Debo tener un circuit breaker por modelo o uno global?"

Uno por modelo/provider. Si gpt-4o falla, no quieres que el circuit bloquee también las llamadas a gpt-4o-mini. En dependencies.py creas instancias separadas: _primary_circuit_breaker y _fallback_circuit_breaker.

Resumen

• El problema: retry ciego durante outages desperdicia recursos y degrada la UX
• Circuit breaker states: CLOSED (normal) → OPEN (falla rápido) → HALF-OPEN (prueba recuperación)
• Parámetros clave: failure_threshold (cuándo abrir), recovery_timeout (cuándo probar)
• Composición: circuit breaker afuera, retry adentro
• Fail fast: cuando el circuit está OPEN, falla en <1ms sin llamar a la API
• Thread-safe: el CircuitBreaker usa threading.Lock para ser seguro en entornos concurrentes

Recursos adicionales

1. Circuit Breaker Pattern (Martin Fowler) — El artículo canónico
2. pybreaker — Implementación alternativa lista para usar
3. Release It! (Michael Nygard) — El origen del patrón
4. Microsoft Azure — Circuit Breaker Pattern — Variantes y trade-offs