# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/across-what-three-state-circuit-breaker-and-single-probe-half-open-gates-must-an-instance-isolate-failing-upstreams-to-prevent-cascading-thread-exhaustion.md # Remnant Warrant: https://corpusfetch.com/c/across-what-three-state-circuit-breaker-and-single-probe-half-open-gates-must-an-instance-isolate-failing-upstreams-to-prevent-cascading-thread-exhaustion.warrant.json # Markdown SHA-256: 1239bdaa632ce81438a9c8a47ff35a729eda934f244f7af57df5592844663568 # Code AST SHA-256: a8af79f8d1aa7ed97ba08ad7c0751a68fcc92b49369b6c10cf09061f5ebabeaa # Verified Classes: CircuitState, CircuitOpenError, RollingCircuitBreaker # Defensive Guards: 4 fail-closed invariants (zero external deps) import threading import time from collections import deque from dataclasses import dataclass from enum import Enum from typing import Callable, TypeVar T = TypeVar("T") class CircuitState(str, Enum): CLOSED = "CLOSED" OPEN = "OPEN" HALF_OPEN = "HALF_OPEN" class CircuitOpenError(RuntimeError): '''Raised immediately (zero network I/O) when the circuit breaker is OPEN or probe-saturated.''' def __init__(self, dependency_name: str, retry_after_sec: float) -> None: self.dependency_name = dependency_name self.retry_after_sec = round(max(0.1, retry_after_sec), 2) super().__init__( f"Circuit breaker for {dependency_name!r} is OPEN; " f"failing fast (retry_after={self.retry_after_sec}s)" ) def is_default_infrastructure_fault(exc: BaseException) -> bool: ''' Classifies whether an exception indicates an upstream infrastructure/availability outage (True) vs. a client-side 4xx / validation error (False). ''' if isinstance(exc, (TimeoutError, ConnectionError, OSError)): return True # Support HTTP client exceptions carrying a `.status_code` or `.status` attribute status = getattr(exc, "status_code", getattr(exc, "status", None)) if isinstance(status, int): return status >= 500 or status == 429 return False class RollingCircuitBreaker: ''' Three-state (`CLOSED -> OPEN -> HALF_OPEN`) circuit breaker with: 1. Sliding monotonic time window for failure counting (`window_sec`). 2. Strict single-probe admission gating in `HALF_OPEN` (`max_half_open_probes`). 3. Selective fault classification (ignores 4xx client errors). ''' def __init__( self, dependency_name: str, *, failure_threshold: int = 5, window_sec: float = 30.0, cooldown_sec: float = 15.0, required_half_open_successes: int = 2, max_half_open_probes: int = 1, fault_classifier: Callable[[BaseException], bool] = is_default_infrastructure_fault, ) -> None: if failure_threshold < 1 or window_sec <= 0 or cooldown_sec <= 0: raise ValueError("Invalid circuit breaker thresholds") self.dependency_name = dependency_name self.failure_threshold = failure_threshold self.window_sec = window_sec self.cooldown_sec = cooldown_sec self.required_half_open_successes = max(1, required_half_open_successes) self.max_half_open_probes = max(1, max_half_open_probes) self.fault_classifier = fault_classifier self._lock = threading.Lock() self._state: CircuitState = CircuitState.CLOSED self._failure_timestamps: deque[float] = deque() self._opened_at_monotonic: float = 0.0 self._active_probes: int = 0 self._consecutive_probe_successes: int = 0 @property def state(self) -> CircuitState: with self._lock: return self._state def _acquire_permission_locked(self, now: float) -> bool: '''Returns True if this call is a designated HALF_OPEN probe, False if normal CLOSED call.''' if self._state == CircuitState.OPEN: elapsed = now - self._opened_at_monotonic if elapsed < self.cooldown_sec: raise CircuitOpenError(self.dependency_name, self.cooldown_sec - elapsed) # Cooldown expired: transition OPEN -> HALF_OPEN self._state = CircuitState.HALF_OPEN self._active_probes = 0 self._consecutive_probe_successes = 0 if self._state == CircuitState.HALF_OPEN: # Gate: Only allow `max_half_open_probes` concurrent requests during recovery! if self._active_probes >= self.max_half_open_probes: raise CircuitOpenError(self.dependency_name, 1.0) self._active_probes += 1 return True return False def _record_success_locked(self, is_probe: bool) -> None: if is_probe or self._state == CircuitState.HALF_OPEN: self._active_probes = max(0, self._active_probes - 1) self._consecutive_probe_successes += 1 if self._consecutive_probe_successes >= self.required_half_open_successes: # Recovery confirmed: graduate HALF_OPEN -> CLOSED self._state = CircuitState.CLOSED self._failure_timestamps.clear() self._consecutive_probe_successes = 0 def _record_infra_failure_locked(self, now: float, is_probe: bool) -> None: if is_probe or self._state == CircuitState.HALF_OPEN: # Any infrastructure failure during HALF_OPEN immediately re-opens the circuit self._active_probes = max(0, self._active_probes - 1) self._state = CircuitState.OPEN self._opened_at_monotonic = now self._consecutive_probe_successes = 0 return # Evict failures older than the sliding monotonic window cutoff = now - self.window_sec while self._failure_timestamps and self._failure_timestamps[0] < cutoff: self._failure_timestamps.popleft() self._failure_timestamps.append(now) if len(self._failure_timestamps) >= self.failure_threshold: self._state = CircuitState.OPEN self._opened_at_monotonic = now def call(self, fn: Callable[..., T], *args: object, **kwargs: object) -> T: ''' Executes `fn(*args, **kwargs)` guarded by the three-state circuit breaker. ''' now = time.monotonic() with self._lock: is_probe = self._acquire_permission_locked(now) try: result = fn(*args, **kwargs) except BaseException as exc: with self._lock: if self.fault_classifier(exc): self._record_infra_failure_locked(time.monotonic(), is_probe) else: # 4xx client/domain errors prove the upstream is reachable and responsive self._record_success_locked(is_probe) raise else: with self._lock: self._record_success_locked(is_probe) return result