Building a Circuit Breaker

Let's build a circuit breaker from scratch. Understanding the implementation helps you configure and debug circuit breakers effectively.

Core Components

A circuit breaker needs these components:

Basic Implementation

Here's a complete, working circuit breaker in Python:

import time
import threading
from enum import Enum
from collections import deque
from typing import Callable, TypeVar, Optional

T = TypeVar('T')

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

class CircuitBreakerError(Exception):
    """Raised when circuit is open"""
    pass

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: float = 0.5,     # 50% failure rate
        window_size: int = 10,               # Last 10 requests
        reset_timeout: float = 30.0,         # 30 seconds
        half_open_max_calls: int = 3         # Test requests
    ):
        self.failure_threshold = failure_threshold
        self.window_size = window_size
        self.reset_timeout = reset_timeout
        self.half_open_max_calls = half_open_max_calls

        # State
        self._state = CircuitState.CLOSED
        self._results = deque(maxlen=window_size)  # True=success, False=failure
        self._opened_at: Optional[float] = None
        self._half_open_calls = 0

        # Thread safety
        self._lock = threading.RLock()

    @property
    def state(self) -> CircuitState:
        with self._lock:
            self._check_state_transition()
            return self._state

    def _check_state_transition(self):
        """Check if we should transition from OPEN to HALF_OPEN"""
        if self._state == CircuitState.OPEN:
            if self._opened_at and (time.time() - self._opened_at >= self.reset_timeout):
                self._transition_to_half_open()

    def _transition_to_open(self):
        self._state = CircuitState.OPEN
        self._opened_at = time.time()
        print(f"Circuit OPENED at {self._opened_at}")

    def _transition_to_half_open(self):
        self._state = CircuitState.HALF_OPEN
        self._half_open_calls = 0
        print("Circuit HALF-OPEN, testing...")

    def _transition_to_closed(self):
        self._state = CircuitState.CLOSED
        self._results.clear()
        self._opened_at = None
        self._half_open_calls = 0
        print("Circuit CLOSED, service recovered")

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

    def call(self, func: Callable[[], T]) -> T:
        """Execute function through circuit breaker"""
        with self._lock:
            self._check_state_transition()

            if self._state == CircuitState.OPEN:
                raise CircuitBreakerError("Circuit is OPEN")

            if self._state == CircuitState.HALF_OPEN:
                if self._half_open_calls >= self.half_open_max_calls:
                    raise CircuitBreakerError("Circuit is HALF-OPEN, max test calls reached")
                self._half_open_calls += 1

        # Execute outside lock to avoid blocking
        try:
            result = func()
            self._record_success()
            return result
        except Exception as e:
            self._record_failure()
            raise

    def _record_success(self):
        with self._lock:
            if self._state == CircuitState.HALF_OPEN:
                # Success in half-open - check if we can close
                if self._half_open_calls >= self.half_open_max_calls:
                    self._transition_to_closed()
            else:
                self._results.append(True)

    def _record_failure(self):
        with self._lock:
            if self._state == CircuitState.HALF_OPEN:
                # Failure in half-open - reopen
                self._transition_to_open()
            else:
                self._results.append(False)
                if len(self._results) >= self.window_size:
                    if self._failure_rate() >= self.failure_threshold:
                        self._transition_to_open()

Using the Circuit Breaker

import requests

# Create a circuit breaker for the payment service
payment_circuit = CircuitBreaker(
    failure_threshold=0.5,
    window_size=10,
    reset_timeout=30.0
)

def call_payment_service(order_id: str, amount: float):
    """Call payment service through circuit breaker"""
    def make_payment():
        response = requests.post(
            "https://payment-service/charge",
            json={"order_id": order_id, "amount": amount},
            timeout=5.0
        )
        response.raise_for_status()
        return response.json()

    try:
        return payment_circuit.call(make_payment)
    except CircuitBreakerError:
        # Circuit is open - return fallback
        return {"status": "pending", "message": "Payment queued"}
    except requests.RequestException as e:
        # Request failed - circuit breaker recorded it
        raise PaymentError(f"Payment failed: {e}")

# Usage
result = call_payment_service("order-123", 99.99)

Implementation Deep Dive

Thread Safety

Circuit breakers must be thread-safe because multiple requests access them concurrently:

# Bad: Race condition
def _record_failure(self):
    self._failures += 1  # Thread A reads 5
    # Thread B also reads 5
    # Both write 6, but should be 7!

# Good: Thread-safe with lock
def _record_failure(self):
    with self._lock:
        self._failures += 1

Sliding Window with Deque

The deque with maxlen automatically maintains the sliding window:

from collections import deque

# Automatically removes oldest when full
results = deque(maxlen=10)

results.append(True)   # [T]
results.append(True)   # [T, T]
results.append(False)  # [T, T, F]
# ... after 10 items ...
results.append(True)   # Oldest removed, newest added

Time-Based Windows (Alternative)

For high-throughput systems, use a time-based window:

from collections import deque
import time

class TimeBasedWindow:
    def __init__(self, window_seconds: float = 60.0):
        self.window_seconds = window_seconds
        self._events: deque = deque()  # (timestamp, success)

    def record(self, success: bool):
        now = time.time()
        self._events.append((now, success))
        self._cleanup()

    def _cleanup(self):
        """Remove events outside the window"""
        cutoff = time.time() - self.window_seconds
        while self._events and self._events[0][0] < cutoff:
            self._events.popleft()

    def failure_rate(self) -> float:
        self._cleanup()
        if not self._events:
            return 0.0
        failures = sum(1 for _, success in self._events if not success)
        return failures / len(self._events)

Decorator Pattern

Make it easy to apply circuit breakers with a decorator:

from functools import wraps

def circuit_breaker(
    failure_threshold: float = 0.5,
    reset_timeout: float = 30.0,
    fallback=None
):
    """Decorator to wrap a function with a circuit breaker"""
    cb = CircuitBreaker(
        failure_threshold=failure_threshold,
        reset_timeout=reset_timeout
    )

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return cb.call(lambda: func(*args, **kwargs))
            except CircuitBreakerError:
                if fallback:
                    return fallback(*args, **kwargs)
                raise
        return wrapper
    return decorator

# Usage
@circuit_breaker(failure_threshold=0.5, fallback=lambda: {"cached": True})
def get_recommendations(user_id: str):
    response = requests.get(f"https://recs-service/users/{user_id}")
    return response.json()

Production Considerations

1. Separate Circuit Breakers Per Service

2. Metrics and Monitoring

class ObservableCircuitBreaker(CircuitBreaker):
    def __init__(self, name: str, metrics_client, **kwargs):
        super().__init__(**kwargs)
        self.name = name
        self.metrics = metrics_client

    def _transition_to_open(self):
        super()._transition_to_open()
        self.metrics.increment(f"circuit_breaker.{self.name}.opened")
        self.metrics.gauge(f"circuit_breaker.{self.name}.state", 1)  # 1 = open

    def _transition_to_closed(self):
        super()._transition_to_closed()
        self.metrics.increment(f"circuit_breaker.{self.name}.closed")
        self.metrics.gauge(f"circuit_breaker.{self.name}.state", 0)  # 0 = closed

    def call(self, func):
        try:
            result = super().call(func)
            self.metrics.increment(f"circuit_breaker.{self.name}.success")
            return result
        except CircuitBreakerError:
            self.metrics.increment(f"circuit_breaker.{self.name}.rejected")
            raise
        except Exception:
            self.metrics.increment(f"circuit_breaker.{self.name}.failure")
            raise

3. Configurable Reset Strategy

class ExponentialBackoffCircuitBreaker(CircuitBreaker):
    """Circuit breaker with exponential backoff on repeated failures"""

    def __init__(self, base_timeout: float = 30.0, max_timeout: float = 300.0, **kwargs):
        super().__init__(reset_timeout=base_timeout, **kwargs)
        self.base_timeout = base_timeout
        self.max_timeout = max_timeout
        self._consecutive_opens = 0

    def _transition_to_open(self):
        super()._transition_to_open()
        self._consecutive_opens += 1
        # Exponential backoff: 30s, 60s, 120s, 240s, 300s (max)
        self.reset_timeout = min(
            self.base_timeout * (2 ** (self._consecutive_opens - 1)),
            self.max_timeout
        )

    def _transition_to_closed(self):
        super()._transition_to_closed()
        self._consecutive_opens = 0
        self.reset_timeout = self.base_timeout

Popular Libraries

Instead of building your own, consider these battle-tested libraries:

Key Takeaways

Next up: Configuration & Fallbacks - How to tune your circuit breaker and design effective fallbacks.