Coverage for src/taipanstack/resilience/watchdogs/health_pinger.py: 100%

1""" 

2Health pinger — proactively checks dependency health. 

3 

4Runs async health checks against registered targets. If a target 

5becomes unhealthy the associated ``CircuitBreaker`` is opened 

6preventively. 

7""" 

8 

9import asyncio 

10import logging 

11import math 

12from collections.abc import Awaitable, Callable, Sequence 

13from dataclasses import dataclass 

14 

15from taipanstack.core.result import Err, Ok, Result 

16from taipanstack.resilience.circuit_breaker import CircuitBreaker, CircuitState 

17from taipanstack.resilience.watchdogs._base import BaseWatcher 

18 

19logger = logging.getLogger("taipanstack.resilience.watchdogs.health") 

20 

21 

22@dataclass 

23class HealthTarget: 

24 """A dependency to be monitored by :class:`HealthPinger`. 

25 

26 Attributes: 

27 name: Human-readable name for logging. 

28 check: Async callable returning ``True`` if the target is 

29 healthy, ``False`` otherwise. 

30 circuit_breaker: Optional circuit breaker to open on failure. 

31 

32 """ 

33 

34 name: str 

35 check: Callable[[], Awaitable[bool]] 

36 circuit_breaker: CircuitBreaker | None = None 

37 

38 

39async def check_target(target: HealthTarget) -> Result[bool, Exception]: 

40 """Run a single health check. 

41 

42 Args: 

43 target: The target to check. 

44 

45 Returns: 

46 ``Ok(True)`` if healthy, ``Ok(False)`` if unhealthy, 

47 ``Err`` if the check itself raises. 

48 

49 """ 

50 try: 

51 healthy = await target.check() 

52 return Ok(healthy) 

53 except Exception as exc: 

54 return Err(exc) 

55 

56 

57async def check_all( 

58 targets: Sequence[HealthTarget], 

59) -> Result[dict[str, bool], Exception]: 

60 """Run health checks for all targets concurrently. 

61 

62 Args: 

63 targets: Targets to check. 

64 

65 Returns: 

66 ``Ok(dict)`` mapping target names to health status. 

67 

68 """ 

69 # Run all checks concurrently 

70 check_results = await asyncio.gather(*(check_target(t) for t in targets)) 

71 

72 results: dict[str, bool] = {} 

73 for target, result in zip(targets, check_results, strict=True): 

74 if isinstance(result, Ok): 

75 results[target.name] = result.ok_value 

76 else: 

77 logger.warning( 

78 "Health check for '%s' failed during aggregation: %s", 

79 target.name, 

80 result.err_value, 

81 ) 

82 results[target.name] = False 

83 return Ok(results) 

84 

85 

86class HealthPinger(BaseWatcher): 

87 """Background watcher that pings external dependencies. 

88 

89 For each registered ``HealthTarget``, calls its ``check`` 

90 coroutine on every cycle. If a target is unhealthy and has 

91 an associated ``CircuitBreaker``, the breaker is opened 

92 preventively. 

93 

94 Args: 

95 targets: Dependencies to monitor. 

96 interval: Seconds between ping cycles. 

97 on_health_change: Optional callback ``(name, is_healthy)``. 

98 

99 Example: 

100 >>> async def db_ping() -> bool: 

101 ... return await pool.fetchval("SELECT 1") == 1 

102 >>> pinger = HealthPinger( 

103 ... targets=[HealthTarget("db", db_ping, breaker)], 

104 ... ) 

105 >>> await pinger.start() 

106 

107 """ 

108 

109 def __init__( 

110 self, 

111 *, 

112 targets: Sequence[HealthTarget], 

113 interval: float = 10.0, 

114 on_health_change: Callable[[str, bool], None] | None = None, 

115 ) -> None: 

116 """Initialize the health pinger. 

117 

118 Args: 

119 targets: Dependencies to monitor. 

120 interval: Seconds between ping cycles. 

121 on_health_change: Optional callback on status change. 

122 

123 """ 

124 if ( 

125 not isinstance(interval, (int, float)) 

126 or not math.isfinite(interval) 

127 or interval <= 0 

128 ): 

129 raise ValueError("interval must be a finite positive number") 

130 super().__init__(interval=interval) 

131 self._targets = list(targets) 

132 self._on_health_change = on_health_change 

133 self._last_status: dict[str, bool] = {} 

134 

135 async def _process_target(self, target: HealthTarget) -> None: 

136 """Process a single health target.""" 

137 result = await check_target(target) 

138 

139 if isinstance(result, Ok): 

140 is_healthy = result.ok_value 

141 else: 

142 logger.warning( 

143 "Health check for '%s' raised: %s", 

144 target.name, 

145 result.err_value, 

146 ) 

147 is_healthy = False 

148 

149 self._update_target_status(target, is_healthy) 

150 

151 def _check_and_open_breaker(self, target: HealthTarget, is_healthy: bool) -> None: 

152 """Open circuit breaker preventively on failure (always checked).""" 

153 if ( 

154 not is_healthy 

155 and target.circuit_breaker is not None 

156 and target.circuit_breaker.state is not CircuitState.OPEN 

157 ): 

158 _force_open_breaker(target.circuit_breaker, target.name) 

159 

160 def _notify_health_change(self, target: HealthTarget, is_healthy: bool) -> None: 

161 """Notify on health status change and log it.""" 

162 if self._on_health_change is not None: 

163 self._on_health_change(target.name, is_healthy) 

164 

165 if is_healthy: 

166 logger.info("Target '%s' is now healthy", target.name) 

167 else: 

168 logger.warning("Target '%s' is now unhealthy", target.name) 

169 

170 def _update_target_status(self, target: HealthTarget, is_healthy: bool) -> None: 

171 """Update target status and handle side-effects.""" 

172 previous = self._last_status.get(target.name) 

173 

174 self._check_and_open_breaker(target, is_healthy) 

175 

176 if previous == is_healthy: 

177 return 

178 

179 self._last_status[target.name] = is_healthy 

180 

181 self._notify_health_change(target, is_healthy) 

182 

183 async def _run(self) -> None: 

184 """Execute a single health-check cycle concurrently.""" 

185 await asyncio.gather(*(self._process_target(t) for t in self._targets)) 

186 

187 

188def _simulate_failures_to_open(breaker: CircuitBreaker, synthetic: Exception) -> None: 

189 max_attempts = breaker.config.failure_threshold + 5 

190 attempts = 0 

191 

192 while breaker.state is not CircuitState.OPEN and attempts < max_attempts: 

193 breaker._record_failure(synthetic) 

194 attempts += 1 

195 

196 

197def _force_state_to_open(breaker: CircuitBreaker) -> bool: 

198 if breaker.state is CircuitState.OPEN: 

199 return True 

200 

201 old_state = breaker.state 

202 acquired = breaker._state.lock.acquire(timeout=0.1) 

203 if not acquired: 

204 return False 

205 try: 

206 breaker._state.state = CircuitState.OPEN 

207 finally: 

208 breaker._state.lock.release() 

209 breaker._notify_state_change(old_state, CircuitState.OPEN) 

210 return True 

211 

212 

213def _force_open_breaker(breaker: CircuitBreaker, target_name: str) -> None: 

214 """Force a circuit breaker into OPEN state. 

215 

216 Simulates enough failures to trip the breaker by recording 

217 a synthetic exception. 

218 

219 Args: 

220 breaker: The circuit breaker to trip. 

221 target_name: Target name for logging context. 

222 

223 """ 

224 synthetic = ConnectionError(f"Health ping failed for '{target_name}'") 

225 

226 _simulate_failures_to_open(breaker, synthetic) 

227 if not _force_state_to_open(breaker): 

228 return 

229 

230 logger.warning( 

231 "Circuit breaker '%s' opened preventively for target '%s'", 

232 breaker.name, 

233 target_name, 

234 )