Coverage for src/taipanstack/resilience/circuit_breaker.py: 100%

1""" 

2Circuit Breaker pattern implementation. 

3 

4Provides protection against cascading failures by temporarily 

5blocking calls to a failing service. Compatible with any 

6Python framework (sync and async). 

7""" 

8 

9import functools 

10import inspect 

11import logging 

12import math 

13import threading 

14import time 

15from collections.abc import Awaitable, Callable 

16from dataclasses import dataclass, field 

17from enum import Enum 

18from typing import ParamSpec, Protocol, TypeGuard, TypeVar, cast, overload 

19 

20from taipanstack.core.result import Err 

21 

22P = ParamSpec("P") 

23R = TypeVar("R") 

24 

25 

26class CircuitBreakerDecorator(Protocol): 

27 """Protocol for the circuit breaker decorator.""" 

28 

29 @overload 

30 def __call__(self, func: Callable[P, R]) -> Callable[P, R]: ... 

31 

32 @overload 

33 def __call__( 

34 self, 

35 func: Callable[P, Awaitable[R]], 

36 ) -> Callable[P, Awaitable[R]]: ... 

37 

38 

39logger = logging.getLogger("taipanstack.resilience.circuit_breaker") 

40 

41try: 

42 import structlog as _structlog 

43 

44 _structlog_logger = _structlog.get_logger("taipanstack.resilience.circuit_breaker") 

45 _HAS_STRUCTLOG = True 

46except ImportError: 

47 _structlog_logger = None 

48 _HAS_STRUCTLOG = False 

49 

50 

51class CircuitState(Enum): 

52 """States of the circuit breaker.""" 

53 

54 CLOSED = "closed" # Normal operation, requests flow through 

55 OPEN = "open" # Circuit is tripped, requests are blocked 

56 HALF_OPEN = "half_open" # Testing if service has recovered 

57 

58 

59class CircuitBreakerError(Exception): 

60 """Raised when circuit breaker is open.""" 

61 

62 def __init__(self, message: str, state: CircuitState) -> None: 

63 """Initialize CircuitBreakerError. 

64 

65 Args: 

66 message: Error description. 

67 state: Current circuit state. 

68 

69 """ 

70 self.state = state 

71 super().__init__(message) 

72 

73 

74@dataclass(frozen=True) 

75class CircuitBreakerConfig: 

76 """Configuration for circuit breaker behavior. 

77 

78 Attributes: 

79 failure_threshold: Number of failures before opening circuit. 

80 success_threshold: Successes needed in half-open to close. 

81 timeout: Seconds before trying half-open after open. 

82 excluded_exceptions: Exceptions that don't count as failures. 

83 failure_exceptions: Exceptions that count as failures. 

84 

85 """ 

86 

87 failure_threshold: int = 5 

88 success_threshold: int = 2 

89 timeout: float = 30.0 

90 excluded_exceptions: tuple[type[Exception], ...] = () 

91 failure_exceptions: tuple[type[Exception], ...] = (Exception,) 

92 

93 def _check_finite(self, value: float, name: str) -> None: 

94 if not math.isfinite(value): 

95 raise ValueError(f"{name} must be finite") 

96 

97 def __post_init__(self) -> None: 

98 """Validate configuration values.""" 

99 self._check_finite(self.failure_threshold, "failure_threshold") 

100 self._check_finite(self.success_threshold, "success_threshold") 

101 self._check_finite(self.timeout, "timeout") 

102 

103 

104@dataclass 

105class CircuitBreakerState: 

106 """Internal state tracking for circuit breaker.""" 

107 

108 state: CircuitState = CircuitState.CLOSED 

109 failure_count: int = 0 

110 success_count: int = 0 

111 half_open_attempts: int = 0 

112 last_failure_time: float = 0.0 

113 lock: threading.Lock = field(default_factory=threading.Lock) 

114 

115 

116class CircuitBreaker: 

117 """Circuit breaker implementation. 

118 

119 Monitors function calls and opens the circuit when too many 

120 failures occur, preventing further calls until the service 

121 recovers. Supports both sync and async functions. 

122 

123 Example: 

124 >>> breaker = CircuitBreaker(failure_threshold=3) 

125 >>> @breaker 

126 ... def call_external_api(): 

127 ... return requests.get("https://api.example.com", timeout=10) 

128 

129 """ 

130 

131 @staticmethod 

132 def _is_valid_metric(value: object, min_val: float = 0) -> TypeGuard[int | float]: 

133 """Check if metric is a valid finite number >= min_val.""" 

134 return ( 

135 isinstance(value, (int, float)) 

136 and math.isfinite(value) 

137 and value >= min_val 

138 ) 

139 

140 @staticmethod 

141 def _get_safe_threshold(value: object, min_val: float, default: float) -> float: 

142 """Return the threshold if valid, otherwise return the default.""" 

143 if CircuitBreaker._is_valid_metric(value, min_val): 

144 return float(value) 

145 return default 

146 

147 @staticmethod 

148 def _check_finite_val(value: float, min_val: float, err_msg: str) -> None: 

149 if not math.isfinite(value) or value < min_val: 

150 raise ValueError(err_msg) 

151 

152 @staticmethod 

153 def _validate_thresholds( 

154 timeout: float, 

155 failure_threshold: int, 

156 success_threshold: int, 

157 ) -> None: 

158 CircuitBreaker._check_finite_val( 

159 timeout, 

160 0, 

161 "timeout must be a finite non-negative number", 

162 ) 

163 CircuitBreaker._check_finite_val( 

164 failure_threshold, 

165 1, 

166 "failure_threshold must be a finite number >= 1", 

167 ) 

168 CircuitBreaker._check_finite_val( 

169 success_threshold, 

170 1, 

171 "success_threshold must be a finite number >= 1", 

172 ) 

173 

174 def __init__( 

175 self, 

176 *, 

177 failure_threshold: int = 5, 

178 success_threshold: int = 2, 

179 timeout: float = 30.0, 

180 excluded_exceptions: tuple[type[Exception], ...] = (), 

181 failure_exceptions: tuple[type[Exception], ...] = (Exception,), 

182 name: str = "default", 

183 on_state_change: Callable[[CircuitState, CircuitState], None] | None = None, 

184 ) -> None: 

185 """Initialize CircuitBreaker. 

186 

187 Args: 

188 failure_threshold: Failures before opening circuit. 

189 success_threshold: Successes to close from half-open. 

190 timeout: Seconds before attempting half-open. 

191 excluded_exceptions: Exceptions that don't trip circuit. 

192 failure_exceptions: Exceptions that count as failures. 

193 name: Name for logging/identification. 

194 on_state_change: Optional callback invoked on state transitions 

195 with (old_state, new_state). Useful for custom monitoring. 

196 

197 """ 

198 CircuitBreaker._validate_thresholds( 

199 timeout, 

200 failure_threshold, 

201 success_threshold, 

202 ) 

203 

204 self.config = CircuitBreakerConfig( 

205 failure_threshold=failure_threshold, 

206 success_threshold=success_threshold, 

207 timeout=timeout, 

208 excluded_exceptions=excluded_exceptions, 

209 failure_exceptions=failure_exceptions, 

210 ) 

211 self.name = name 

212 self._state = CircuitBreakerState() 

213 self._on_state_change = on_state_change 

214 

215 @property 

216 def state(self) -> CircuitState: 

217 """Get current circuit state.""" 

218 return self._state.state 

219 

220 @property 

221 def failure_count(self) -> int: 

222 """Get current failure count.""" 

223 return self._state.failure_count 

224 

225 def _log_callback_failure( 

226 self, 

227 old_state: CircuitState, 

228 new_state: CircuitState, 

229 e: Exception, 

230 ) -> None: 

231 if _HAS_STRUCTLOG and _structlog_logger is not None: 

232 _structlog_logger.error( 

233 "circuit_state_change_callback_failed", 

234 circuit=self.name, 

235 old_state=old_state.value, 

236 new_state=new_state.value, 

237 error=str(e), 

238 ) 

239 else: 

240 logger.error( 

241 "Circuit %s state change callback failed: %s", 

242 self.name, 

243 str(e), 

244 ) 

245 

246 def _log_structlog_warning( 

247 self, 

248 old_state: CircuitState, 

249 new_state: CircuitState, 

250 ) -> None: 

251 if _HAS_STRUCTLOG and _structlog_logger is not None: 

252 _structlog_logger.warning( 

253 "circuit_state_changed", 

254 circuit=self.name, 

255 old_state=old_state.value, 

256 new_state=new_state.value, 

257 failure_count=self._state.failure_count, 

258 ) 

259 

260 def _notify_state_change( 

261 self, 

262 old_state: CircuitState, 

263 new_state: CircuitState, 

264 ) -> None: 

265 """Notify callback of state transition if registered. 

266 

267 Emit a structured log via structlog when no callback is provided 

268 and structlog is available. 

269 """ 

270 if self._on_state_change is not None: 

271 try: 

272 self._on_state_change(old_state, new_state) 

273 except Exception as e: 

274 self._log_callback_failure(old_state, new_state, e) 

275 return 

276 

277 self._log_structlog_warning(old_state, new_state) 

278 

279 def _get_safe_timeout(self) -> float: 

280 try: 

281 return float(self.config.timeout) 

282 except (TypeError, ValueError): 

283 return 30.0 

284 

285 def _calculate_elapsed_time(self, now: float) -> float | None: 

286 """Calculate time elapsed since last failure.""" 

287 if not isinstance(self._state.last_failure_time, (int, float)): 

288 return None # type: ignore[unreachable] 

289 

290 safe_timeout = self._get_safe_timeout() 

291 

292 if not math.isfinite(self._state.last_failure_time): 

293 return safe_timeout 

294 

295 elapsed = now - float(self._state.last_failure_time) 

296 

297 # Safe check against NaN and Inf time corruption 

298 # If elapsed < 0, a backward clock jump occurred. We should 

299 # allow a transition to prevent permanent lockout. 

300 if elapsed < 0: 

301 return safe_timeout 

302 return elapsed 

303 

304 def _transition_to_half_open( 

305 self, 

306 elapsed: float, 

307 ) -> tuple[bool, tuple[CircuitState, CircuitState] | None]: 

308 """Transition the circuit to half-open state.""" 

309 # Before transitioning, verify if we can make an attempt 

310 # This happens in a lock, so it's thread-safe. However, once 

311 # the state changes to HALF_OPEN, subsequent threads in the 

312 # same lock block will hit the HALF_OPEN case. 

313 self._state.state = CircuitState.HALF_OPEN 

314 self._state.success_count = 0 

315 # Initialize half_open_attempts to 1 because this first call 

316 # that transitions the state is also an attempt. 

317 self._state.half_open_attempts = 1 

318 logger.info( 

319 "Circuit %s entering half-open state (was open for %.1fs, failures=%d)", 

320 self.name, 

321 elapsed, 

322 self._state.failure_count, 

323 ) 

324 return True, (CircuitState.OPEN, CircuitState.HALF_OPEN) 

325 

326 def _evaluate_open_timeout( 

327 self, 

328 elapsed: float, 

329 ) -> tuple[bool, tuple[CircuitState, CircuitState] | None]: 

330 timeout = self._get_safe_threshold(self.config.timeout, 0, 30.0) 

331 if elapsed >= timeout: 

332 return self._transition_to_half_open(elapsed) 

333 return False, None 

334 

335 def _get_valid_elapsed(self) -> float | None: 

336 try: 

337 now = time.monotonic() 

338 except Exception: 

339 return None 

340 if not math.isfinite(now): 

341 return None 

342 return self._calculate_elapsed_time(now) 

343 

344 def _handle_open_state( 

345 self, 

346 ) -> tuple[bool, tuple[CircuitState, CircuitState] | None]: 

347 """Handle logic for OPEN state in _should_attempt.""" 

348 elapsed = self._get_valid_elapsed() 

349 if elapsed is None: 

350 return False, None 

351 return self._evaluate_open_timeout(elapsed) 

352 

353 def _handle_attempt_half_open(self) -> bool: 

354 if not self._is_valid_metric(self._state.half_open_attempts): 

355 return False 

356 

357 success_threshold = self._get_safe_threshold( 

358 self.config.success_threshold, 

359 1, 

360 2, 

361 ) 

362 

363 if self._state.half_open_attempts < success_threshold: 

364 self._state.half_open_attempts += 1 

365 return True 

366 return False 

367 

368 def _evaluate_state_for_attempt( 

369 self, 

370 ) -> tuple[bool, tuple[CircuitState, CircuitState] | None]: 

371 state = self._state.state 

372 if state == CircuitState.CLOSED: 

373 return True, None 

374 if state == CircuitState.OPEN: 

375 return self._handle_open_state() 

376 if state == CircuitState.HALF_OPEN: 

377 return self._handle_attempt_half_open(), None 

378 return False, None # type: ignore[unreachable] 

379 

380 def _should_attempt(self) -> bool: 

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

382 try: 

383 with self._state.lock: 

384 should_attempt, state_change = self._evaluate_state_for_attempt() 

385 except Exception: 

386 # If lock acquisition fails, fail-safe by preventing the call. 

387 return False 

388 

389 if state_change: 

390 self._notify_state_change(*state_change) 

391 

392 return should_attempt 

393 

394 def _handle_success_half_open(self) -> tuple[CircuitState, CircuitState] | None: 

395 if not self._is_valid_metric(self._state.success_count): 

396 # Type corruption detected, reset and increment 

397 self._state.success_count = 1 

398 else: 

399 self._state.success_count += 1 

400 

401 success_threshold = self._get_safe_threshold( 

402 self.config.success_threshold, 

403 1, 

404 2, 

405 ) 

406 

407 if self._state.success_count >= success_threshold: 

408 self._state.state = CircuitState.CLOSED 

409 self._state.failure_count = 0 

410 self._state.half_open_attempts = 0 

411 logger.info( 

412 "Circuit %s closed after recovery (%d consecutive successes)", 

413 self.name, 

414 self._state.success_count, 

415 ) 

416 return (CircuitState.HALF_OPEN, CircuitState.CLOSED) 

417 return None 

418 

419 def _get_success_state_change(self) -> tuple[CircuitState, CircuitState] | None: 

420 state = self._state.state 

421 if state == CircuitState.HALF_OPEN: 

422 return self._handle_success_half_open() 

423 if state == CircuitState.CLOSED: 

424 # Reset failure count on success 

425 self._state.failure_count = 0 

426 return None 

427 return None 

428 

429 def _record_success(self) -> None: 

430 """Record a successful call.""" 

431 try: 

432 with self._state.lock: 

433 state_change = self._get_success_state_change() 

434 except Exception: 

435 return 

436 

437 if state_change: 

438 self._notify_state_change(*state_change) 

439 

440 def _handle_failure_half_open(self) -> tuple[CircuitState, CircuitState] | None: 

441 """Handle failure when in HALF_OPEN state.""" 

442 self._state.state = CircuitState.OPEN 

443 self._state.half_open_attempts = 0 

444 logger.warning( 

445 "Circuit %s reopened after failure in half-open", 

446 self.name, 

447 ) 

448 return (CircuitState.HALF_OPEN, CircuitState.OPEN) 

449 

450 def _handle_failure_closed(self) -> tuple[CircuitState, CircuitState] | None: 

451 """Handle failure when in CLOSED state.""" 

452 # Check against corrupted NaN/Inf failure_count 

453 if not self._is_valid_metric(self._state.failure_count): 

454 self._state.state = CircuitState.OPEN 

455 logger.warning( 

456 "Circuit %s opened due to state/type corruption in failure_count", 

457 self.name, 

458 ) 

459 return (CircuitState.CLOSED, CircuitState.OPEN) 

460 

461 failure_threshold = self._get_safe_threshold( 

462 self.config.failure_threshold, 

463 1, 

464 5, 

465 ) 

466 

467 if self._state.failure_count >= failure_threshold: 

468 self._state.state = CircuitState.OPEN 

469 logger.warning( 

470 "Circuit %s opened after %d failures (threshold=%d)", 

471 self.name, 

472 self._state.failure_count, 

473 failure_threshold, 

474 ) 

475 return (CircuitState.CLOSED, CircuitState.OPEN) 

476 

477 return None 

478 

479 def _increment_failure_count(self) -> None: 

480 if not self._is_valid_metric(self._state.failure_count): 

481 # Handle type mutation (e.g. failure_count became string) or Inf/NaN 

482 # Safe degradation: reset to max so it opens immediately 

483 self._state.failure_count = self.config.failure_threshold 

484 else: 

485 self._state.failure_count += 1 

486 

487 def _update_failure_metrics(self) -> None: 

488 self._increment_failure_count() 

489 

490 try: 

491 now = time.monotonic() 

492 except Exception: 

493 now = float("nan") 

494 

495 if math.isfinite(now): 

496 self._state.last_failure_time = now 

497 

498 def _get_failure_state_change(self) -> tuple[CircuitState, CircuitState] | None: 

499 """Get state change for a failure.""" 

500 if self._state.state == CircuitState.HALF_OPEN: 

501 return self._handle_failure_half_open() 

502 if self._state.state == CircuitState.CLOSED: 

503 return self._handle_failure_closed() 

504 return None 

505 

506 def _is_excluded_exception(self, exc: Exception) -> bool: 

507 try: 

508 return isinstance(exc, self.config.excluded_exceptions) 

509 except TypeError: 

510 return False 

511 

512 def _record_failure(self, exc: Exception) -> None: 

513 """Record a failed call.""" 

514 if self._is_excluded_exception(exc): 

515 return 

516 

517 state_change: tuple[CircuitState, CircuitState] | None = None 

518 

519 try: 

520 with self._state.lock: 

521 self._update_failure_metrics() 

522 state_change = self._get_failure_state_change() 

523 except Exception: 

524 return 

525 

526 if state_change: 

527 self._notify_state_change(*state_change) 

528 

529 def reset(self) -> None: 

530 """Reset circuit breaker to closed state.""" 

531 try: 

532 with self._state.lock: 

533 self._state.state = CircuitState.CLOSED 

534 self._state.failure_count = 0 

535 self._state.success_count = 0 

536 self._state.half_open_attempts = 0 

537 logger.info("Circuit %s manually reset", self.name) 

538 except Exception: 

539 return 

540 

541 def _is_failure_exception(self, exc: Exception) -> bool: 

542 try: 

543 return isinstance(exc, self.config.failure_exceptions) 

544 except TypeError: 

545 return True 

546 

547 def _process_result(self, result: R) -> R: 

548 """Process Result outcome and record success/failure. 

549 

550 Args: 

551 result: The result to process. 

552 

553 Returns: 

554 The original result. 

555 

556 """ 

557 if isinstance(result, Err): 

558 err_val = result.unwrap_err() 

559 if self._is_failure_exception(err_val): 

560 self._record_failure(err_val) 

561 return result 

562 # Ignored exception in Result monad 

563 return result 

564 self._record_success() 

565 return result 

566 

567 def _is_valid_half_open_attempts(self) -> bool: 

568 """Validate half-open attempts value against corruption.""" 

569 return ( 

570 isinstance(self._state.half_open_attempts, (int, float)) 

571 and math.isfinite(self._state.half_open_attempts) 

572 and self._state.half_open_attempts >= 0 

573 ) 

574 

575 def _safe_decrement_half_open_attempts(self) -> None: 

576 """Safely decrement half-open attempts.""" 

577 if self._state.state != CircuitState.HALF_OPEN: 

578 return 

579 

580 if not self._is_valid_half_open_attempts(): 

581 # Reset if state is corrupted to prevent crash 

582 self._state.half_open_attempts = 0 

583 return 

584 

585 if self._state.half_open_attempts > 0: 

586 self._state.half_open_attempts -= 1 

587 

588 def _decrement_half_open(self, is_half_open: bool) -> None: 

589 """Decrement half-open attempt count if applicable. 

590 

591 Args: 

592 is_half_open: Whether the circuit was half-open before attempt. 

593 

594 """ 

595 if is_half_open: 

596 try: 

597 with self._state.lock: 

598 self._safe_decrement_half_open_attempts() 

599 except Exception: 

600 return 

601 

602 def __call__( 

603 self, 

604 func: Callable[P, R] | Callable[P, Awaitable[R]], 

605 ) -> Callable[P, R] | Callable[P, Awaitable[R]]: 

606 """Decorate a sync or async function with circuit breaker protection.""" 

607 if inspect.iscoroutinefunction(func): 

608 func_coro = cast(Callable[P, Awaitable[R]], func) 

609 

610 @functools.wraps(func_coro) 

611 async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

612 if not self._should_attempt(): 

613 raise CircuitBreakerError( 

614 f"Circuit {self.name} is open", 

615 state=self._state.state, 

616 ) 

617 

618 is_half_open = self._state.state == CircuitState.HALF_OPEN 

619 

620 try: 

621 result = await func_coro(*args, **kwargs) 

622 return self._process_result(result) 

623 except Exception as e: 

624 try: 

625 is_failure = isinstance(e, self.config.failure_exceptions) 

626 except TypeError: 

627 is_failure = True 

628 if is_failure: 

629 self._record_failure(e) 

630 raise 

631 raise 

632 finally: 

633 self._decrement_half_open(is_half_open) 

634 

635 return async_wrapper 

636 

637 func_sync = cast(Callable[P, R], func) 

638 

639 @functools.wraps(func_sync) 

640 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

641 if not self._should_attempt(): 

642 raise CircuitBreakerError( 

643 f"Circuit {self.name} is open", 

644 state=self._state.state, 

645 ) 

646 

647 is_half_open = self._state.state == CircuitState.HALF_OPEN 

648 

649 try: 

650 result = func_sync(*args, **kwargs) 

651 return self._process_result(result) 

652 except Exception as e: 

653 try: 

654 is_failure = isinstance(e, self.config.failure_exceptions) 

655 except TypeError: 

656 is_failure = True 

657 if is_failure: 

658 self._record_failure(e) 

659 raise 

660 raise 

661 finally: 

662 self._decrement_half_open(is_half_open) 

663 

664 return wrapper 

665 

666 

667def circuit_breaker( 

668 *, 

669 failure_threshold: int = 5, 

670 success_threshold: int = 2, 

671 timeout: float = 30.0, 

672 excluded_exceptions: tuple[type[Exception], ...] = (), 

673 failure_exceptions: tuple[type[Exception], ...] = (Exception,), 

674 name: str | None = None, 

675 on_state_change: Callable[[CircuitState, CircuitState], None] | None = None, 

676) -> CircuitBreakerDecorator: 

677 """Decorate a sync or async function with circuit breaker pattern. 

678 

679 Args: 

680 failure_threshold: Failures before opening circuit. 

681 success_threshold: Successes to close from half-open. 

682 timeout: Seconds before attempting half-open. 

683 excluded_exceptions: Exceptions that don't trip circuit. 

684 failure_exceptions: Exceptions that count as failures. 

685 name: Optional name for the circuit. 

686 on_state_change: Optional callback invoked on state transitions 

687 with (old_state, new_state). 

688 

689 Returns: 

690 Decorated function with circuit breaker protection. 

691 

692 Example: 

693 >>> @circuit_breaker(failure_threshold=3, timeout=60) 

694 ... def call_api(endpoint: str) -> dict: 

695 ... return requests.get(endpoint, timeout=10).json() 

696 

697 >>> @circuit_breaker( 

698 ... failure_threshold=3, 

699 ... on_state_change=lambda old, new: print(f"{old} -> {new}"), 

700 ... ) 

701 ... def monitored_call() -> str: 

702 ... return service.call() 

703 

704 """ 

705 

706 def decorator( 

707 func: Callable[P, R] | Callable[P, Awaitable[R]], 

708 ) -> Callable[P, R] | Callable[P, Awaitable[R]]: 

709 breaker = CircuitBreaker( 

710 failure_threshold=failure_threshold, 

711 success_threshold=success_threshold, 

712 timeout=timeout, 

713 excluded_exceptions=excluded_exceptions, 

714 failure_exceptions=failure_exceptions, 

715 name=name or cast(str, getattr(func, "__name__", "unknown")), 

716 on_state_change=on_state_change, 

717 ) 

718 return breaker(func) 

719 

720 return cast(CircuitBreakerDecorator, decorator)