1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174

# Production-Grade Circuit Breaker Implementation
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
from threading import Lock
from typing import Callable, TypeVar, Optional
import time
 
class CircuitState(Enum):
    CLOSED = "closed"        # Normal operation
    OPEN = "open"            # Failing fast
    HALF_OPEN = "half_open"  # Testing recovery
 
@dataclass
class CircuitBreakerConfig:
    """Circuit breaker configuration."""
    failure_threshold: int = 5           # Failures before opening
    success_threshold: int = 3           # Successes to close from half-open
    timeout_seconds: float = 30.0        # Time in open before half-open
    half_open_max_calls: int = 3         # Max concurrent half-open calls
    failure_rate_threshold: float = 0.5  # Failure rate to trigger (0-1)
    min_calls_for_rate: int = 10         # Min calls before rate matters
 
T = TypeVar('T')
 
class CircuitBreaker:
    """
    Thread-safe circuit breaker with multiple tripping strategies.
    
    Usage:
        breaker = CircuitBreaker("payment-service")
        
        try:
            result = breaker.call(lambda: payment_client.charge(amount))
        except CircuitOpenError:
            return fallback_response()
    """
    
    def __init__(self, name: str, config: CircuitBreakerConfig = None):
        self.name = name
        self.config = config or CircuitBreakerConfig()
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._success_count = 0
        self._total_calls = 0
        self._last_failure_time: Optional[datetime] = None
        self._half_open_calls = 0
        self._lock = Lock()
    
    @property
    def state(self) -> CircuitState:
        with self._lock:
            self._check_state_transition()
            return self._state
    
    def _check_state_transition(self) -> None:
        """Check if circuit should transition states."""
        if self._state == CircuitState.OPEN:
            if self._last_failure_time:
                elapsed = datetime.now() - self._last_failure_time
                if elapsed.total_seconds() >= self.config.timeout_seconds:
                    self._transition_to(CircuitState.HALF_OPEN)
    
    def _transition_to(self, new_state: CircuitState) -> None:
        """Transition to new state with logging."""
        old_state = self._state
        self._state = new_state
        
        if new_state == CircuitState.CLOSED:
            self._failure_count = 0
            self._success_count = 0
            self._total_calls = 0
        elif new_state == CircuitState.HALF_OPEN:
            self._half_open_calls = 0
            self._success_count = 0
        
        print(f"[CircuitBreaker:{self.name}] {old_state.value} -> {new_state.value}")
    
    def _should_trip(self) -> bool:
        """Determine if circuit should trip to OPEN."""
        # Consecutive failure threshold
        if self._failure_count >= self.config.failure_threshold:
            return True
        
        # Failure rate threshold (only after minimum calls)
        if self._total_calls >= self.config.min_calls_for_rate:
            failure_rate = self._failure_count / self._total_calls
            if failure_rate >= self.config.failure_rate_threshold:
                return True
        
        return False
    
    def call(self, func: Callable[[], T], fallback: Callable[[], T] = None) -> T:
        """
        Execute function through circuit breaker.
        
        Args:
            func: The function to execute
            fallback: Optional fallback if circuit is open
            
        Raises:
            CircuitOpenError: If circuit is open and no fallback provided
        """
        with self._lock:
            self._check_state_transition()
            
            if self._state == CircuitState.OPEN:
                if fallback:
                    return fallback()
                raise CircuitOpenError(f"Circuit {self.name} is OPEN")
            
            if self._state == CircuitState.HALF_OPEN:
                if self._half_open_calls >= self.config.half_open_max_calls:
                    if fallback:
                        return fallback()
                    raise CircuitOpenError(f"Circuit {self.name} half-open limit reached")
                self._half_open_calls += 1
        
        try:
            result = func()
            self._record_success()
            return result
        except Exception as e:
            self._record_failure()
            raise
    
    def _record_success(self) -> None:
        """Record successful call."""
        with self._lock:
            self._success_count += 1
            self._total_calls += 1
            
            if self._state == CircuitState.HALF_OPEN:
                if self._success_count >= self.config.success_threshold:
                    self._transition_to(CircuitState.CLOSED)
    
    def _record_failure(self) -> None:
        """Record failed call."""
        with self._lock:
            self._failure_count += 1
            self._total_calls += 1
            self._last_failure_time = datetime.now()
            
            if self._state == CircuitState.HALF_OPEN:
                self._transition_to(CircuitState.OPEN)
            elif self._state == CircuitState.CLOSED:
                if self._should_trip():
                    self._transition_to(CircuitState.OPEN)
 
class CircuitOpenError(Exception):
    """Raised when circuit breaker is open."""
    pass
 
# Example usage
payment_breaker = CircuitBreaker(
    "payment-service",
    CircuitBreakerConfig(
        failure_threshold=5,
        timeout_seconds=30,
        success_threshold=3
    )
)
 
def process_payment(amount: float) -> dict:
    """Process payment with circuit breaker protection."""
    def call_payment_service():
        # Actual payment service call
        return {"status": "success", "amount": amount}
    
    def fallback():
        # Queue for later processing
        return {"status": "queued", "message": "Payment queued for retry"}
    
    return payment_breaker.call(call_payment_service, fallback)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

# Comprehensive Retry Implementation
import random
import time
from dataclasses import dataclass
from typing import Callable, TypeVar, Set, Type
from functools import wraps
 
T = TypeVar('T')
 
@dataclass
class RetryConfig:
    """Configuration for retry behavior."""
    max_attempts: int = 3
    base_delay_ms: float = 100
    max_delay_ms: float = 30000
    exponential_base: float = 2
    jitter_ms: float = 100
    retryable_exceptions: Set[Type[Exception]] = None
    
    def __post_init__(self):
        if self.retryable_exceptions is None:
            self.retryable_exceptions = {
                ConnectionError,
                TimeoutError,
                IOError,
            }
 
def calculate_delay(attempt: int, config: RetryConfig) -> float:
    """Calculate delay with exponential backoff and jitter."""
    exponential_delay = config.base_delay_ms * (config.exponential_base ** attempt)
    capped_delay = min(exponential_delay, config.max_delay_ms)
    jitter = random.uniform(0, config.jitter_ms)
    return capped_delay + jitter
 
def is_retryable(exception: Exception, config: RetryConfig) -> bool:
    """Determine if exception is retryable."""
    return any(
        isinstance(exception, exc_type) 
        for exc_type in config.retryable_exceptions
    )
 
def with_retry(config: RetryConfig = None):
    """
    Decorator for retrying functions with exponential backoff.
    
    Usage:
        @with_retry(RetryConfig(max_attempts=5))
        def call_external_service():
            ...
    """
    if config is None:
        config = RetryConfig()
    
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> T:
            last_exception = None
            
            for attempt in range(config.max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    
                    if not is_retryable(e, config):
                        raise  # Non-retryable, fail immediately
                    
                    if attempt < config.max_attempts - 1:
                        delay_ms = calculate_delay(attempt, config)
                        print(f"Retry {attempt + 1}/{config.max_attempts} "
                              f"after {delay_ms:.0f}ms: {e}")
                        time.sleep(delay_ms / 1000)
                    else:
                        print(f"All {config.max_attempts} attempts failed")
            
            raise last_exception
        
        return wrapper
    return decorator
 
# Retry budget pattern - limit total retry time
class RetryBudget:
    """
    Limits retries globally to prevent retry storms.
    
    Instead of each request getting N retries, the system gets
    a budget of retries per time window.
    """
    
    def __init__(self, 
                 budget_per_second: float = 0.1,  # 10% of traffic can be retries
                 min_retries_per_second: int = 10):  # But always allow some
        self.budget_ratio = budget_per_second
        self.min_retries = min_retries_per_second
        self._request_count = 0
        self._retry_count = 0
        self._window_start = time.time()
        self._window_seconds = 1.0
    
    def _reset_window_if_needed(self):
        now = time.time()
        if now - self._window_start >= self._window_seconds:
            self._request_count = 0
            self._retry_count = 0
            self._window_start = now
    
    def record_request(self):
        self._reset_window_if_needed()
        self._request_count += 1
    
    def can_retry(self) -> bool:
        """Check if retry budget allows another retry."""
        self._reset_window_if_needed()
        
        # Always allow minimum retries
        if self._retry_count < self.min_retries:
            return True
        
        # Check if under budget
        if self._request_count == 0:
            return True
        
        retry_ratio = self._retry_count / self._request_count
        return retry_ratio < self.budget_ratio
    
    def record_retry(self) -> bool:
        """Record a retry, return True if allowed."""
        if self.can_retry():
            self._retry_count += 1
            return True
        return False
 
# Example: Using retry budget
retry_budget = RetryBudget(budget_per_second=0.1)
 
def make_request_with_budget(url: str) -> dict:
    retry_budget.record_request()
    
    for attempt in range(3):
        try:
            return http_get(url)  # hypothetical
        except TimeoutError:
            if attempt < 2 and retry_budget.record_retry():
                time.sleep(calculate_delay(attempt, RetryConfig()))
                continue
            raise
    raise RuntimeError("Should not reach here")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

# Thread Pool Bulkhead Implementation
from concurrent.futures import ThreadPoolExecutor, TimeoutError
from dataclasses import dataclass
from typing import Callable, TypeVar, Dict
import threading
 
T = TypeVar('T')
 
@dataclass
class BulkheadConfig:
    """Configuration for a bulkhead."""
    max_concurrent: int = 10
    max_waiting: int = 5
    timeout_seconds: float = 30.0
 
class Bulkhead:
    """
    Thread pool bulkhead for isolating concurrent operations.
    
    Usage:
        payment_bulkhead = Bulkhead("payments", BulkheadConfig(max_concurrent=10))
        result = payment_bulkhead.execute(lambda: payment_service.charge(amount))
    """
    
    def __init__(self, name: str, config: BulkheadConfig = None):
        self.name = name
        self.config = config or BulkheadConfig()
        self._executor = ThreadPoolExecutor(
            max_workers=self.config.max_concurrent,
            thread_name_prefix=f"bulkhead-{name}"
        )
        self._semaphore = threading.Semaphore(
            self.config.max_concurrent + self.config.max_waiting
        )
        self._active = 0
        self._waiting = 0
        self._lock = threading.Lock()
    
    def execute(self, func: Callable[[], T]) -> T:
        """
        Execute function within bulkhead constraints.
        
        Raises:
            BulkheadFullError: If bulkhead is at capacity
            TimeoutError: If execution times out
        """
        # Check if we can acquire a slot
        acquired = self._semaphore.acquire(blocking=False)
        if not acquired:
            raise BulkheadFullError(
                f"Bulkhead {self.name} is full: "
                f"{self._active} active, {self._waiting} waiting"
            )
        
        try:
            with self._lock:
                self._waiting += 1
            
            future = self._executor.submit(func)
            
            with self._lock:
                self._waiting -= 1
                self._active += 1
            
            try:
                return future.result(timeout=self.config.timeout_seconds)
            except TimeoutError:
                future.cancel()
                raise
            finally:
                with self._lock:
                    self._active -= 1
        finally:
            self._semaphore.release()
    
    def metrics(self) -> dict:
        """Get current bulkhead metrics."""
        with self._lock:
            return {
                "name": self.name,
                "max_concurrent": self.config.max_concurrent,
                "max_waiting": self.config.max_waiting,
                "active": self._active,
                "waiting": self._waiting,
                "available": self.config.max_concurrent - self._active,
            }
 
class BulkheadFullError(Exception):
    """Raised when bulkhead cannot accept more work."""
    pass
 
class BulkheadRegistry:
    """
    Central registry for managing multiple bulkheads.
    Provides a single point for monitoring and configuration.
    """
    
    def __init__(self):
        self._bulkheads: Dict[str, Bulkhead] = {}
        self._lock = threading.Lock()
    
    def get_or_create(self, name: str, config: BulkheadConfig = None) -> Bulkhead:
        """Get existing bulkhead or create new one."""
        with self._lock:
            if name not in self._bulkheads:
                self._bulkheads[name] = Bulkhead(name, config)
            return self._bulkheads[name]
    
    def all_metrics(self) -> list:
        """Get metrics for all bulkheads."""
        with self._lock:
            return [b.metrics() for b in self._bulkheads.values()]
 
# Global registry
bulkheads = BulkheadRegistry()
 
# Usage example - different bulkheads for different dependencies
payment_bulkhead = bulkheads.get_or_create(
    "payments", 
    BulkheadConfig(max_concurrent=10, timeout_seconds=30)
)
 
inventory_bulkhead = bulkheads.get_or_create(
    "inventory",
    BulkheadConfig(max_concurrent=50, timeout_seconds=5)
)
 
def checkout(cart) -> dict:
    """Checkout with bulkhead isolation."""
    # If payments is slow, it won't affect inventory checks
    inventory_result = inventory_bulkhead.execute(
        lambda: check_inventory(cart.items)
    )
    
    if not inventory_result.available:
        return {"error": "Items not available"}
    
    payment_result = payment_bulkhead.execute(
        lambda: process_payment(cart.total)
    )
    
    return {"order_id": payment_result.order_id}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

// Graceful Degradation Framework
interface DegradationLevel {
  level: 'normal' | 'degraded' | 'minimal' | 'emergency';
  enabledFeatures: string[];
  disabledFeatures: string[];
}
 
interface DependencyHealth {
  name: string;
  healthy: boolean;
  latencyP99Ms: number;
  errorRate: number;
}
 
class DegradationController {
  private currentLevel: DegradationLevel['level'] = 'normal';
  private dependencies: Map<string, DependencyHealth> = new Map();
  
  // Define what features are available at each level
  private readonly levels: Record<string, DegradationLevel> = {
    normal: {
      level: 'normal',
      enabledFeatures: ['ml_search', 'personalization', 'real_time_inventory', 
                        'dynamic_pricing', 'reviews', 'recommendations'],
      disabledFeatures: [],
    },
    degraded: {
      level: 'degraded',
      enabledFeatures: ['basic_search', 'cached_inventory', 'static_pricing', 
                        'reviews_readonly', 'popular_items'],
      disabledFeatures: ['ml_search', 'personalization', 'real_time_inventory',
                         'dynamic_pricing', 'recommendations'],
    },
    minimal: {
      level: 'minimal',
      enabledFeatures: ['basic_search', 'cached_inventory', 'static_pricing'],
      disabledFeatures: ['reviews', 'recommendations', 'personalization',
                         'real_time_inventory', 'dynamic_pricing'],
    },
    emergency: {
      level: 'emergency',
      enabledFeatures: ['static_catalog', 'cached_content'],
      disabledFeatures: ['search', 'checkout', 'user_accounts'],
    },
  };
 
  updateDependencyHealth(health: DependencyHealth): void {
    this.dependencies.set(health.name, health);
    this.evaluateDegradationLevel();
  }
 
  private evaluateDegradationLevel(): void {
    const healths = Array.from(this.dependencies.values());
    
    // Calculate overall system health score
    const unhealthyCount = healths.filter(h => !h.healthy).length;
    const avgErrorRate = healths.reduce((sum, h) => sum + h.errorRate, 0) / healths.length;
    const avgLatency = healths.reduce((sum, h) => sum + h.latencyP99Ms, 0) / healths.length;
 
    // Determine appropriate degradation level
    let newLevel: DegradationLevel['level'];
    
    if (unhealthyCount === 0 && avgErrorRate < 0.01 && avgLatency < 100) {
      newLevel = 'normal';
    } else if (unhealthyCount <= 2 && avgErrorRate < 0.05) {
      newLevel = 'degraded';
    } else if (unhealthyCount <= 4 && avgErrorRate < 0.20) {
      newLevel = 'minimal';
    } else {
      newLevel = 'emergency';
    }
 
    if (newLevel !== this.currentLevel) {
      console.log(`Degradation level changing: ${this.currentLevel} -> ${newLevel}`);
      this.currentLevel = newLevel;
      this.emitLevelChange(newLevel);
    }
  }
 
  isFeatureEnabled(featureName: string): boolean {
    const level = this.levels[this.currentLevel];
    return level.enabledFeatures.includes(featureName);
  }
 
  getCurrentLevel(): DegradationLevel {
    return this.levels[this.currentLevel];
  }
 
  private emitLevelChange(level: string): void {
    // Emit metric, send alert, update feature flags
    metrics.emit('degradation_level', { level });
  }
}
 
// Usage in application code
const degradation = new DegradationController();
 
async function getProductRecommendations(userId: string): Promise<Product[]> {
  if (degradation.isFeatureEnabled('recommendations')) {
    try {
      return await mlRecommendationService.getPersonalized(userId);
    } catch (error) {
      // Fall through to fallback
    }
  }
  
  if (degradation.isFeatureEnabled('popular_items')) {
    return await getPopularItems();
  }
  
  return []; // No recommendations available
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

# Deadline Propagation Pattern
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional
import time
 
@dataclass
class RequestContext:
    """Context propagated through service calls."""
    request_id: str
    deadline: datetime  # Absolute time by which request must complete
    
    @property
    def remaining_ms(self) -> float:
        """Milliseconds remaining before deadline."""
        remaining = (self.deadline - datetime.now()).total_seconds() * 1000
        return max(0, remaining)
    
    @property
    def is_expired(self) -> bool:
        """Check if deadline has passed."""
        return datetime.now() >= self.deadline
    
    def with_reduced_deadline(self, buffer_ms: float = 100) -> 'RequestContext':
        """
        Create child context with slightly earlier deadline.
        Buffer ensures we have time to handle timeout gracefully.
        """
        new_deadline = self.deadline - timedelta(milliseconds=buffer_ms)
        return RequestContext(
            request_id=self.request_id,
            deadline=new_deadline
        )
 
class DeadlineExceededError(Exception):
    """Raised when operation deadline is exceeded."""
    pass
 
def check_deadline(ctx: RequestContext) -> None:
    """Check deadline and raise if exceeded."""
    if ctx.is_expired:
        raise DeadlineExceededError(
            f"Deadline exceeded for request {ctx.request_id}"
        )
 
# Example: Deadline-aware service call
async def process_order(ctx: RequestContext, order: dict) -> dict:
    """Process order with deadline propagation."""
    
    # Check deadline before starting
    check_deadline(ctx)
    
    # Validate inventory with remaining time
    child_ctx = ctx.with_reduced_deadline(100)  # Reserve 100ms
    inventory_result = await inventory_service.check(
        child_ctx,
        order['items'],
        timeout_ms=child_ctx.remaining_ms
    )
    
    check_deadline(ctx)
    
    # Process payment with remaining time
    child_ctx = ctx.with_reduced_deadline(100)
    payment_result = await payment_service.charge(
        child_ctx,
        order['total'],
        timeout_ms=child_ctx.remaining_ms
    )
    
    return {
        "order_id": generate_order_id(),
        "status": "completed"
    }
 
# HTTP header for deadline propagation
# Request: X-Deadline: 2024-01-15T10:30:00.000Z
# Each service subtracts buffer and passes reduced deadline