Coverage for src/taipanstack/resilience/adaptive/adaptive_breaker.py: 100%
123 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-10-13 20:52 +0000
1"""
2Adaptive Circuit Breaker — auto-tunes failure threshold via rolling window.
4Unlike standard Circuit Breakers that use static absolute failure counts,
5the AdaptiveCircuitBreaker opens its circuit ONLY when the error rate
6exceeds a target percentage in a rolling window of recent calls AND a
7minimum throughput of requests has been met.
8"""
10from __future__ import annotations
12import logging
13import math
14import threading
15import time
16from collections import deque
17from dataclasses import dataclass
18from typing import TypeVar
20from taipanstack.resilience.circuit_breaker import CircuitState
22logger = logging.getLogger("taipanstack.resilience.adaptive.breaker")
24T = TypeVar("T")
27def _build_adaptive_metrics(
28 window: deque[bool], state: CircuitState
29) -> AdaptiveMetrics:
30 total = len(window)
31 errors = sum(1 for ok in window if not ok)
32 error_rate = errors / total if total > 0 else 0.0
33 return AdaptiveMetrics(
34 success_rate=1.0 - error_rate,
35 error_rate=error_rate,
36 total_calls=total,
37 error_count=errors,
38 state=state,
39 )
42@dataclass(frozen=True)
43class AdaptiveMetrics:
44 """Snapshot of adaptive circuit breaker metrics.
46 Attributes:
47 success_rate: Current success rate (0.0 - 1.0).
48 error_rate: Current error rate (0.0 - 1.0).
49 total_calls: Total calls in the window.
50 error_count: Errors in the window.
51 state: Current circuit state.
53 """
55 success_rate: float
56 error_rate: float
57 total_calls: int
58 error_count: int
59 state: CircuitState
62class AdaptiveCircuitBreaker:
63 """Circuit breaker that opens based on an error rate percentage.
65 Maintains a rolling window of call outcomes. The circuit trips to OPEN if:
66 1. The `window_size` history has at least `min_throughput` events.
67 2. The error rate (errors / total) > `target_error_rate`.
69 Once OPEN, it waits `recovery_timeout` seconds before transitioning
70 to HALF_OPEN. In HALF_OPEN, if a request succeeds, it CLOSES and
71 clears the window. If it fails, it returns to OPEN.
73 Args:
74 name: Identifier for logging.
75 window_size: Number of recent calls to track.
76 min_throughput: Minimum requests before considering error rate.
77 target_error_rate: Desired error rate boundary (0.0 - 1.0).
78 recovery_timeout: Seconds before attempting half-open recovery.
80 """
82 def __init__(
83 self,
84 name: str = "default",
85 *,
86 window_size: int = 100,
87 min_throughput: int = 10,
88 target_error_rate: float = 0.5,
89 recovery_timeout: float = 30.0,
90 ) -> None:
91 """Initialize the adaptive circuit breaker."""
92 self.name = name
93 self._min_throughput = self._validate_min_throughput(min_throughput)
94 self._target_error_rate = target_error_rate
95 self._recovery_timeout = self._validate_recovery_timeout(recovery_timeout)
97 # Rolling window: True = success, False = failure
98 self._window: deque[bool] = deque(maxlen=window_size)
99 self._state = CircuitState.CLOSED
100 self._last_opened_at: float = 0.0
101 self._lock = threading.Lock()
103 @staticmethod
104 def _validate_min_throughput(min_throughput: int) -> int:
105 if min_throughput < 1:
106 raise ValueError("min_throughput must be at least 1")
107 return min_throughput
109 @staticmethod
110 def _validate_recovery_timeout(recovery_timeout: float) -> float:
111 if not isinstance(recovery_timeout, (int, float)):
112 raise TypeError("recovery_timeout must be a finite non-negative number")
113 if not math.isfinite(recovery_timeout):
114 raise ValueError("recovery_timeout must be a finite non-negative number")
115 if recovery_timeout < 0:
116 raise ValueError("recovery_timeout must be a finite non-negative number")
117 return float(recovery_timeout)
119 def _calculate_elapsed_time(self, now: float) -> float:
120 """Calculate elapsed time since the circuit opened."""
121 last_opened = self._last_opened_at
122 if not isinstance(last_opened, (int, float)) or not math.isfinite(last_opened):
123 return self._recovery_timeout
125 elapsed = now - last_opened
126 return self._recovery_timeout if elapsed < 0 else elapsed
128 def _check_half_open_transition(self) -> None:
129 """Evaluate if an OPEN circuit should transition to HALF_OPEN."""
130 now = time.monotonic()
131 elapsed = self._calculate_elapsed_time(now)
133 if math.isfinite(now) and elapsed >= self._recovery_timeout:
134 self._state = CircuitState.HALF_OPEN
135 logger.info("Adaptive breaker '%s' entering HALF_OPEN state", self.name)
137 @property
138 def state(self) -> CircuitState:
139 """Current circuit state. May evaluate timeouts and switch to HALF_OPEN."""
140 with self._lock:
141 if self._state is CircuitState.OPEN:
142 self._check_half_open_transition()
143 return self._state
145 def _calculate_error_rate(self, total: int) -> float:
146 """Calculate the current error rate in the window."""
147 if total <= 0:
148 return 0.0
149 errors = sum(1 for ok in self._window if not ok)
150 return errors / total
152 def _get_safe_target_rate(self) -> float:
153 """Safely retrieve the target error rate."""
154 target_rate = self._target_error_rate
155 if not isinstance(target_rate, (int, float)) or not math.isfinite(target_rate):
156 return -1.0 # Fail closed
157 return float(target_rate)
159 def _transition_to_open(self, error_rate: float, target_rate: float) -> None:
160 """Transition the circuit to the OPEN state."""
161 self._state = CircuitState.OPEN
162 self._last_opened_at = time.monotonic()
164 log_target = target_rate if target_rate >= 0 else 0.0
165 logger.warning(
166 "Adaptive breaker '%s' OPENED. Error rate %.2f > %.2f",
167 self.name,
168 error_rate,
169 log_target,
170 )
172 def _get_safe_min_throughput(self) -> int:
173 """Safely retrieve the minimum throughput."""
174 min_throughput = self._min_throughput
175 if not isinstance(min_throughput, (int, float)) or not math.isfinite(
176 min_throughput
177 ):
178 return 1
179 return int(min_throughput)
181 def _evaluate_trip(self) -> None:
182 """Evaluate if the circuit should trip open.
184 MUST BE CALLED UNDER LOCK.
185 """
186 if self._state is not CircuitState.CLOSED:
187 return
189 total = len(self._window)
190 min_throughput = self._get_safe_min_throughput()
192 if total < min_throughput:
193 return
195 error_rate = self._calculate_error_rate(total)
196 target_rate = self._get_safe_target_rate()
198 if error_rate > target_rate:
199 self._transition_to_open(error_rate, target_rate)
201 def record_success(self) -> None:
202 """Record a successful call."""
203 with self._lock:
204 if self._state is CircuitState.HALF_OPEN:
205 # Full recovery on success
206 self._state = CircuitState.CLOSED
207 self._window.clear()
208 logger.info(
209 "Adaptive breaker '%s' CLOSED after successful half-open recovery.",
210 self.name,
211 )
213 self._window.append(True)
214 self._evaluate_trip()
216 def record_failure(self, _exc: Exception) -> None:
217 """Record a failed call.
219 Args:
220 _exc: The exception that occurred.
222 """
223 with self._lock:
224 if self._state is CircuitState.HALF_OPEN:
225 # Return to open immediately on failure
226 self._state = CircuitState.OPEN
227 self._last_opened_at = time.monotonic()
228 logger.warning(
229 "Adaptive breaker '%s' RETURNED to OPEN after half-open failure.",
230 self.name,
231 )
233 self._window.append(False)
234 self._evaluate_trip()
236 def should_allow(self) -> bool:
237 """Check if a call should be attempted.
239 Returns:
240 ``True`` if the circuit permits a call.
242 """
243 return self.state in (CircuitState.CLOSED, CircuitState.HALF_OPEN)
245 def reset(self) -> None:
246 """Reset the breaker and window."""
247 with self._lock:
248 self._window.clear()
249 self._state = CircuitState.CLOSED
250 self._last_opened_at = 0.0
252 @property
253 def metrics(self) -> AdaptiveMetrics:
254 """Snapshot of current adaptive metrics."""
255 with self._lock:
256 return _build_adaptive_metrics(self._window, self._state)