Coverage for qreward / utils / scheduler / circuit_breaker.py: 98.90%

1"""Circuit breaker implementation for fault tolerance. 

2 

3Provides automatic circuit breaking when consecutive failures reach a 

4threshold, with half-open state probing for recovery. 

5""" 

6 

7import logging (empty)

8import threading (empty)

9import time (empty)

10from enum import Enum (empty)

11from typing import Callable, Optional (empty)

12 

13logger = logging.getLogger(__name__) (empty)

14 

15 

16class CircuitState(Enum): (empty)

17 """Circuit breaker state enumeration. 

18 

19 Attributes: 

20 CLOSED: Normal state, requests are allowed. 

21 OPEN: Tripped state, requests are blocked. 

22 HALF_OPEN: Probing state, limited requests allowed to test recovery. 

23 """ 

24 

25 CLOSED = "closed" (empty)

26 OPEN = "open" (empty)

27 HALF_OPEN = "half_open" (empty)

28 

29 

30class CircuitBreaker: (empty)

31 """Thread-safe circuit breaker implementation. 

32 

33 When consecutive failures reach ``failure_threshold`` the breaker trips 

34 to OPEN state, blocking all requests. After ``recovery_timeout`` seconds 

35 it transitions to HALF_OPEN, allowing up to ``half_open_max_calls`` 

36 probe requests. A successful probe resets the breaker to CLOSED; a 

37 failed probe re-opens it. 

38 

39 Args: 

40 failure_threshold: Number of consecutive failures before opening. 

41 recovery_timeout: Seconds to wait before transitioning to HALF_OPEN. 

42 half_open_max_calls: Max probe requests allowed in HALF_OPEN state. 

43 time_func: Clock function for testability (default ``time.time``). 

44 """ 

45 

46 def __init__( (empty)

47 self, 

48 failure_threshold: int = 5, 

49 recovery_timeout: float = 30.0, 

50 half_open_max_calls: int = 1, 

51 time_func: Optional[Callable[[], float]] = None, 

52 ) -> None: 

53 self._failure_threshold = failure_threshold (empty)

54 self._recovery_timeout = recovery_timeout (empty)

55 self._half_open_max_calls = half_open_max_calls (empty)

56 self._time_func = time_func or time.time (empty)

57 

58 self._lock = threading.Lock() (empty)

59 self._state = CircuitState.CLOSED (empty)

60 self._failure_count = 0 (empty)

61 self._last_failure_time: float = 0.0 (empty)

62 self._half_open_calls = 0 (empty)

63 

64 # ------------------------------------------------------------------ 

65 # Public API 

66 # ------------------------------------------------------------------ 

67 

68 @property (empty)

69 def state(self) -> CircuitState: (empty)

70 """Return the current circuit state. 

71 

72 Automatically transitions from OPEN → HALF_OPEN when the recovery 

73 timeout has elapsed. 

74 """ 

75 with self._lock: (empty)

76 self._maybe_transition_to_half_open() (empty)

77 return self._state (empty)

78 

79 def allow_request(self) -> bool: (empty)

80 """Check whether a request should be allowed. 

81 

82 Returns: 

83 True if the request is permitted, False otherwise. 

84 """ 

85 with self._lock: (empty)

86 self._maybe_transition_to_half_open() (empty)

87 

88 if self._state == CircuitState.CLOSED: (empty)

89 return True (empty)

90 

91 if self._state == CircuitState.HALF_OPEN: (empty)

92 if self._half_open_calls < self._half_open_max_calls: (empty)

93 self._half_open_calls += 1 (empty)

94 return True (empty)

95 return False (empty)

96 

97 # OPEN 

98 return False (empty)

99 

100 def record_success(self) -> None: (empty)

101 """Record a successful request and update state accordingly.""" 

102 with self._lock: (empty)

103 if self._state == CircuitState.HALF_OPEN: (empty)

104 logger.info( (empty)

105 "CircuitBreaker transitioning HALF_OPEN -> CLOSED " 

106 "(probe succeeded)" 

107 ) 

108 self._reset_unlocked() (empty)

109 elif self._state == CircuitState.CLOSED: 109 ↛ exitline 109 didn't jump to the function exit(empty)

110 self._failure_count = 0 (empty)

111 

112 def record_failure(self) -> None: (empty)

113 """Record a failed request and update state accordingly.""" 

114 with self._lock: (empty)

115 if self._state == CircuitState.HALF_OPEN: (empty)

116 logger.warning( (empty)

117 "CircuitBreaker transitioning HALF_OPEN -> OPEN " 

118 "(probe failed)" 

119 ) 

120 self._state = CircuitState.OPEN (empty)

121 self._last_failure_time = self._time_func() (empty)

122 self._half_open_calls = 0 (empty)

123 return (empty)

124 

125 self._failure_count += 1 (empty)

126 if self._failure_count >= self._failure_threshold: (empty)

127 logger.warning( (empty)

128 "CircuitBreaker transitioning CLOSED -> OPEN " 

129 "(failure_count=%d >= threshold=%d)", 

130 self._failure_count, 

131 self._failure_threshold, 

132 ) 

133 self._state = CircuitState.OPEN (empty)

134 self._last_failure_time = self._time_func() (empty)

135 

136 def reset(self) -> None: (empty)

137 """Reset the breaker to its initial CLOSED state.""" 

138 with self._lock: (empty)

139 self._reset_unlocked() (empty)

140 

141 # ------------------------------------------------------------------ 

142 # Internal helpers (must be called with lock held) 

143 # ------------------------------------------------------------------ 

144 

145 def _reset_unlocked(self) -> None: (empty)

146 self._state = CircuitState.CLOSED (empty)

147 self._failure_count = 0 (empty)

148 self._half_open_calls = 0 (empty)

149 self._last_failure_time = 0.0 (empty)

150 

151 def _maybe_transition_to_half_open(self) -> None: (empty)

152 if self._state != CircuitState.OPEN: (empty)

153 return (empty)

154 elapsed = self._time_func() - self._last_failure_time (empty)

155 if elapsed >= self._recovery_timeout: (empty)

156 logger.info( (empty)

157 "CircuitBreaker transitioning OPEN -> HALF_OPEN " 

158 "(recovery_timeout=%.1fs elapsed)", 

159 self._recovery_timeout, 

160 ) 

161 self._state = CircuitState.HALF_OPEN (empty)

162 self._half_open_calls = 0 (empty)