Coverage for src/taipanstack/resilience/adaptive/adaptive_breaker.py: 100%

1""" 

2Adaptive Circuit Breaker — auto-tunes failure threshold via rolling window. 

3 

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""" 

9 

10from __future__ import annotations 

11 

12import logging 

13import math 

14import threading 

15import time 

16from collections import deque 

17from dataclasses import dataclass 

18from typing import TypeVar 

19 

20from taipanstack.core.result import Ok, Result 

21from taipanstack.resilience.circuit_breaker import CircuitState 

22 

23logger = logging.getLogger("taipanstack.resilience.adaptive.breaker") 

24 

25T = TypeVar("T") 

26 

27 

28@dataclass(frozen=True) 

29class AdaptiveMetrics: 

30 """Snapshot of adaptive circuit breaker metrics. 

31 

32 Attributes: 

33 success_rate: Current success rate (0.0 - 1.0). 

34 error_rate: Current error rate (0.0 - 1.0). 

35 total_calls: Total calls in the window. 

36 error_count: Errors in the window. 

37 state: Current circuit state. 

38 

39 """ 

40 

41 success_rate: float 

42 error_rate: float 

43 total_calls: int 

44 error_count: int 

45 state: CircuitState 

46 

47 

48class AdaptiveCircuitBreaker: 

49 """Circuit breaker that opens based on an error rate percentage. 

50 

51 Maintains a rolling window of call outcomes. The circuit trips to OPEN if: 

52 1. The `window_size` history has at least `min_throughput` events. 

53 2. The error rate (errors / total) > `target_error_rate`. 

54 

55 Once OPEN, it waits `recovery_timeout` seconds before transitioning 

56 to HALF_OPEN. In HALF_OPEN, if a request succeeds, it CLOSES and 

57 clears the window. If it fails, it returns to OPEN. 

58 

59 Args: 

60 name: Identifier for logging. 

61 window_size: Number of recent calls to track. 

62 min_throughput: Minimum requests before considering error rate. 

63 target_error_rate: Desired error rate boundary (0.0 - 1.0). 

64 recovery_timeout: Seconds before attempting half-open recovery. 

65 

66 """ 

67 

68 def __init__( 

69 self, 

70 name: str = "default", 

71 *, 

72 window_size: int = 100, 

73 min_throughput: int = 10, 

74 target_error_rate: float = 0.5, 

75 recovery_timeout: float = 30.0, 

76 ) -> None: 

77 """Initialize the adaptive circuit breaker.""" 

78 self.name = name 

79 if min_throughput < 1: 

80 raise ValueError("min_throughput must be at least 1") 

81 self._min_throughput = min_throughput 

82 self._target_error_rate = target_error_rate 

83 if not math.isfinite(recovery_timeout) or recovery_timeout < 0: 

84 raise ValueError("recovery_timeout must be a finite non-negative number") 

85 self._recovery_timeout = recovery_timeout 

86 

87 # Rolling window: True = success, False = failure 

88 self._window: deque[bool] = deque(maxlen=window_size) 

89 self._state = CircuitState.CLOSED 

90 self._last_opened_at: float = 0.0 

91 self._lock = threading.Lock() 

92 

93 def _calculate_elapsed_time(self, now: float) -> float: 

94 """Calculate elapsed time since the circuit opened.""" 

95 last_opened = self._last_opened_at 

96 if not isinstance(last_opened, (int, float)) or not math.isfinite(last_opened): 

97 return self._recovery_timeout 

98 

99 elapsed = now - last_opened 

100 return self._recovery_timeout if elapsed < 0 else elapsed 

101 

102 def _check_half_open_transition(self) -> None: 

103 """Evaluate if an OPEN circuit should transition to HALF_OPEN.""" 

104 now = time.monotonic() 

105 elapsed = self._calculate_elapsed_time(now) 

106 

107 if math.isfinite(now) and elapsed >= self._recovery_timeout: 

108 self._state = CircuitState.HALF_OPEN 

109 logger.info("Adaptive breaker '%s' entering HALF_OPEN state", self.name) 

110 

111 @property 

112 def state(self) -> CircuitState: 

113 """Current circuit state. May evaluate timeouts and switch to HALF_OPEN.""" 

114 with self._lock: 

115 if self._state == CircuitState.OPEN: 

116 self._check_half_open_transition() 

117 return self._state 

118 

119 def _calculate_error_rate(self, total: int) -> float: 

120 """Calculate the current error rate in the window.""" 

121 if total <= 0: 

122 return 0.0 

123 errors = sum(1 for ok in self._window if not ok) 

124 return errors / total 

125 

126 def _get_safe_target_rate(self) -> float: 

127 """Safely retrieve the target error rate.""" 

128 target_rate = self._target_error_rate 

129 if not isinstance(target_rate, (int, float)) or not math.isfinite(target_rate): 

130 return -1.0 # Fail closed 

131 return float(target_rate) 

132 

133 def _transition_to_open(self, error_rate: float, target_rate: float) -> None: 

134 """Transition the circuit to the OPEN state.""" 

135 self._state = CircuitState.OPEN 

136 self._last_opened_at = time.monotonic() 

137 

138 log_target = target_rate if target_rate >= 0 else 0.0 

139 logger.warning( 

140 "Adaptive breaker '%s' OPENED. Error rate %.2f > %.2f", 

141 self.name, 

142 error_rate, 

143 log_target, 

144 ) 

145 

146 def _evaluate_trip(self) -> None: 

147 """Evaluate if the circuit should trip open. 

148 

149 MUST BE CALLED UNDER LOCK. 

150 """ 

151 if self._state != CircuitState.CLOSED: 

152 return 

153 

154 total = len(self._window) 

155 if total < self._min_throughput: 

156 return 

157 

158 error_rate = self._calculate_error_rate(total) 

159 target_rate = self._get_safe_target_rate() 

160 

161 if error_rate > target_rate: 

162 self._transition_to_open(error_rate, target_rate) 

163 

164 def record_success(self) -> None: 

165 """Record a successful call.""" 

166 with self._lock: 

167 if self._state == CircuitState.HALF_OPEN: 

168 # Full recovery on success 

169 self._state = CircuitState.CLOSED 

170 self._window.clear() 

171 logger.info( 

172 "Adaptive breaker '%s' CLOSED after successful half-open recovery.", 

173 self.name, 

174 ) 

175 

176 self._window.append(True) 

177 self._evaluate_trip() 

178 

179 def record_failure(self, _exc: Exception) -> None: 

180 """Record a failed call. 

181 

182 Args: 

183 _exc: The exception that occurred. 

184 

185 """ 

186 with self._lock: 

187 if self._state == CircuitState.HALF_OPEN: 

188 # Return to open immediately on failure 

189 self._state = CircuitState.OPEN 

190 self._last_opened_at = time.monotonic() 

191 logger.warning( 

192 "Adaptive breaker '%s' RETURNED to OPEN after half-open failure.", 

193 self.name, 

194 ) 

195 

196 self._window.append(False) 

197 self._evaluate_trip() 

198 

199 def evaluate_result(self, result: Result[T, Exception]) -> Result[T, Exception]: 

200 """Evaluate a Result and record success or failure. 

201 

202 Args: 

203 result: A ``Result`` to evaluate. 

204 

205 Returns: 

206 The original Result. 

207 

208 """ 

209 if isinstance(result, Ok): 

210 self.record_success() 

211 else: 

212 self.record_failure(result.err_value) 

213 return result 

214 

215 def should_allow(self) -> bool: 

216 """Check if a call should be attempted. 

217 

218 Returns: 

219 ``True`` if the circuit permits a call. 

220 

221 """ 

222 return self.state in (CircuitState.CLOSED, CircuitState.HALF_OPEN) 

223 

224 def reset(self) -> None: 

225 """Reset the breaker and window.""" 

226 with self._lock: 

227 self._window.clear() 

228 self._state = CircuitState.CLOSED 

229 self._last_opened_at = 0.0 

230 

231 @property 

232 def metrics(self) -> AdaptiveMetrics: 

233 """Snapshot of current adaptive metrics.""" 

234 with self._lock: 

235 total = len(self._window) 

236 errors = sum(1 for ok in self._window if not ok) 

237 error_rate = errors / total if total > 0 else 0.0 

238 success_rate = 1.0 - error_rate 

239 state_val = self._state 

240 

241 return AdaptiveMetrics( 

242 success_rate=success_rate, 

243 error_rate=error_rate, 

244 total_calls=total, 

245 error_count=errors, 

246 state=state_val, 

247 )