Circuit Breaker: Handle Service Failures

Problem

In distributed systems, services depend on each other. When one service fails or becomes slow, calling services keep retrying, consuming resources and causing cascading failures across the entire system. This creates a domino effect where a single failure brings down multiple services.

Example Scenario:

• Service A calls Service B
• Service B becomes unavailable
• Service A keeps retrying, exhausting its thread pool
• Service A becomes unresponsive
• Service C, which depends on Service A, also fails
• System collapses

Solution

The Circuit Breaker pattern prevents cascading failures by monitoring service calls and stopping requests to failing services. It works like an electrical circuit breaker—when too many failures occur, it "trips" and stops sending traffic, allowing the failing service time to recover.

How It Works

The circuit breaker has three states:

1. CLOSED (Normal): Requests pass through normally. Failures are counted.
2. OPEN (Failing): Failure threshold exceeded. Requests fail immediately without calling the service.
3. HALF_OPEN (Recovery): After a timeout, allows limited requests to test if service recovered.

State Transitions

CLOSED → OPEN (failure threshold exceeded)
OPEN → HALF_OPEN (timeout elapsed)
HALF_OPEN → CLOSED (test requests succeed)
HALF_OPEN → OPEN (test requests fail)

Benefits

• Prevents cascading failures: Stops propagation of failures
• Fast failure: Fails immediately instead of waiting for timeouts
• Resource protection: Prevents thread pool exhaustion
• Self-healing: Automatically tests recovery
• Graceful degradation: Can return cached data or defaults

Code

Python Implementation

import time
from enum import Enum
from typing import Callable, Any, Optional
from datetime import datetime, timedelta
import threading

class CircuitState(Enum):
    CLOSED = "CLOSED"
    OPEN = "OPEN"
    HALF_OPEN = "HALF_OPEN"

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

class CircuitBreaker:
    """
    Circuit Breaker pattern implementation for handling service failures.

    Prevents cascading failures by stopping requests to failing services
    and allowing them time to recover.
    """

    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        expected_exception: type = Exception,
        name: str = "CircuitBreaker"
    ):
        """
        Initialize Circuit Breaker.

        Args:
            failure_threshold: Number of failures before opening circuit
            recovery_timeout: Seconds to wait before attempting recovery
            expected_exception: Exception type to catch
            name: Identifier for logging
        """
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.expected_exception = expected_exception
        self.name = name

        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[datetime] = None
        self.state = CircuitState.CLOSED
        self._lock = threading.RLock()

    def call(self, func: Callable, *args, **kwargs) -> Any:
        """
        Execute function through circuit breaker.

        Args:
            func: Function to execute
            *args: Positional arguments for function
            **kwargs: Keyword arguments for function

        Returns:
            Function result

        Raises:
            CircuitBreakerException: If circuit is open
            Exception: Original exception from function
        """
        with self._lock:
            if self.state == CircuitState.OPEN:
                if self._should_attempt_reset():
                    self.state = CircuitState.HALF_OPEN
                    print(f"[{self.name}] State: OPEN → HALF_OPEN (testing recovery)")
                else:
                    raise CircuitBreakerException(
                        f"Circuit breaker '{self.name}' is OPEN. "
                        f"Service unavailable."
                    )

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except self.expected_exception as e:
            self._on_failure()
            raise

    def _on_success(self) -> None:
        """Handle successful call"""
        with self._lock:
            self.failure_count = 0

            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                if self.success_count >= 2:  # 2 successes to close
                    self.state = CircuitState.CLOSED
                    self.success_count = 0
                    print(f"[{self.name}] State: HALF_OPEN → CLOSED (recovered)")

    def _on_failure(self) -> None:
        """Handle failed call"""
        with self._lock:
            self.failure_count += 1
            self.last_failure_time = datetime.now()
            self.success_count = 0

            print(f"[{self.name}] Failure #{self.failure_count}")

            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
                print(f"[{self.name}] State: CLOSED → OPEN (threshold exceeded)")
            elif self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.OPEN
                print(f"[{self.name}] State: HALF_OPEN → OPEN (recovery failed)")

    def _should_attempt_reset(self) -> bool:
        """Check if enough time has passed to attempt recovery"""
        if self.last_failure_time is None:
            return True

        elapsed = (datetime.now() - self.last_failure_time).total_seconds()
        return elapsed >= self.recovery_timeout

    def get_state(self) -> str:
        """Get current circuit breaker state"""
        with self._lock:
            return self.state.value

    def reset(self) -> None:
        """Manually reset circuit breaker"""
        with self._lock:
            self.failure_count = 0
            self.success_count = 0
            self.last_failure_time = None
            self.state = CircuitState.CLOSED
            print(f"[{self.name}] Manually reset to CLOSED")


# Example: Unreliable Service
class UnreliableService:
    """Simulates a service that fails intermittently"""

    def __init__(self, failure_rate: float = 0.5):
        self.failure_rate = failure_rate
        self.call_count = 0

    def fetch_data(self, user_id: int) -> dict:
        """Fetch user data (fails randomly)"""
        self.call_count += 1

        import random
        if random.random() < self.failure_rate:
            raise Exception(f"Service error on call #{self.call_count}")

        return {"user_id": user_id, "name": f"User {user_id}"}


# Example: Client with Circuit Breaker
class UserServiceClient:
    """Client that uses circuit breaker to call external service"""

    def __init__(self, service: UnreliableService):
        self.service = service
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=3,
            recovery_timeout=5,
            expected_exception=Exception,
            name="UserService"
        )
        self.cache = {}

    def get_user(self, user_id: int) -> dict:
        """Get user data with circuit breaker protection"""
        try:
            data = self.circuit_breaker.call(
                self.service.fetch_data,
                user_id
            )
            self.cache[user_id] = data
            return data
        except CircuitBreakerException:
            # Return cached data or default
            if user_id in self.cache:
                print(f"  → Returning cached data for user {user_id}")
                return self.cache[user_id]
            else:
                print(f"  → No cached data, returning default")
                return {"user_id": user_id, "name": "Unknown"}
        except Exception as e:
            print(f"  → Service error: {e}")
            if user_id in self.cache:
                return self.cache[user_id]
            return {"user_id": user_id, "name": "Unknown"}


# Demonstration
def main():
    print("=" * 60)
    print("CIRCUIT BREAKER PATTERN DEMONSTRATION")
    print("=" * 60)

    service = UnreliableService(failure_rate=0.7)
    client = UserServiceClient(service)

    print("\n--- Phase 1: Normal Operation (CLOSED) ---")
    for i in range(1, 6):
        print(f"\nRequest #{i}:")
        result = client.get_user(i)
        print(f"  Result: {result}")
        print(f"  Circuit State: {client.circuit_breaker.get_state()}")

    print("\n--- Phase 2: Circuit Opens (too many failures) ---")
    for i in range(6, 9):
        print(f"\nRequest #{i}:")
        result = client.get_user(i)
        print(f"  Result: {result}")
        print(f"  Circuit State: {client.circuit_breaker.get_state()}")

    print("\n--- Phase 3: Waiting for Recovery Timeout ---")
    print("Waiting 6 seconds for recovery timeout...")
    time.sleep(6)

    print("\n--- Phase 4: Testing Recovery (HALF_OPEN) ---")
    for i in range(9, 12):
        print(f"\nRequest #{i}:")
        result = client.get_user(i)
        print(f"  Result: {result}")
        print(f"  Circuit State: {client.circuit_breaker.get_state()}")

    print("\n" + "=" * 60)
    print("DEMONSTRATION COMPLETE")
    print("=" * 60)


if __name__ == "__main__":
    main()

Key Features

1. Thread-safe: Uses locks for concurrent access
2. Configurable: Adjustable thresholds and timeouts
3. State management: Clear state transitions
4. Graceful degradation: Returns cached data when circuit open
5. Automatic recovery: Tests service health periodically
6. Logging: Tracks state changes and failures

Real-World Usage

# Database connection
db_breaker = CircuitBreaker(
    failure_threshold=5,
    recovery_timeout=30,
    name="DatabaseConnection"
)

# API call
api_breaker = CircuitBreaker(
    failure_threshold=3,
    recovery_timeout=60,
    name="ExternalAPI"
)

# Use with decorator pattern
def call_with_breaker(breaker, func, *args, **kwargs):
    return breaker.call(func, *args, **kwargs)

This implementation provides robust protection against cascading failures in distributed systems.