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 

11from collections.abc import Awaitable, Callable, Sequence 

12from dataclasses import dataclass 

13 

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

15from taipanstack.resilience.circuit_breaker import CircuitBreaker, CircuitState 

16from taipanstack.resilience.watchdogs._base import BaseWatcher 

17 

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

19 

20 

21@dataclass 

22class HealthTarget: 

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

24 

25 Attributes: 

26 name: Human-readable name for logging. 

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

28 healthy, ``False`` otherwise. 

29 circuit_breaker: Optional circuit breaker to open on failure. 

30 

31 """ 

32 

33 name: str 

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

35 circuit_breaker: CircuitBreaker | None = None 

36 

37 

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

39 """Run a single health check. 

40 

41 Args: 

42 target: The target to check. 

43 

44 Returns: 

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

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

47 

48 """ 

49 try: 

50 healthy = await target.check() 

51 return Ok(healthy) 

52 except Exception as exc: 

53 return Err(exc) 

54 

55 

56async def check_all( 

57 targets: Sequence[HealthTarget], 

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

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

60 

61 Args: 

62 targets: Targets to check. 

63 

64 Returns: 

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

66 

67 """ 

68 # Run all checks concurrently 

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

70 

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

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

73 if isinstance(result, Ok): 

74 results[target.name] = result.ok_value 

75 else: 

76 logger.warning( 

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

78 target.name, 

79 result.err_value, 

80 ) 

81 results[target.name] = False 

82 return Ok(results) 

83 

84 

85class HealthPinger(BaseWatcher): 

86 """Background watcher that pings external dependencies. 

87 

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

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

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

91 preventively. 

92 

93 Args: 

94 targets: Dependencies to monitor. 

95 interval: Seconds between ping cycles. 

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

97 

98 Example: 

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

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

101 >>> pinger = HealthPinger( 

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

103 ... ) 

104 >>> await pinger.start() 

105 

106 """ 

107 

108 def __init__( 

109 self, 

110 *, 

111 targets: Sequence[HealthTarget], 

112 interval: float = 10.0, 

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

114 ) -> None: 

115 """Initialize the health pinger. 

116 

117 Args: 

118 targets: Dependencies to monitor. 

119 interval: Seconds between ping cycles. 

120 on_health_change: Optional callback on status change. 

121 

122 """ 

123 super().__init__(interval=interval) 

124 self._targets = list(targets) 

125 self._on_health_change = on_health_change 

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

127 

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

129 """Process a single health target.""" 

130 result = await check_target(target) 

131 

132 if isinstance(result, Ok): 

133 is_healthy = result.ok_value 

134 else: 

135 logger.warning( 

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

137 target.name, 

138 result.err_value, 

139 ) 

140 is_healthy = False 

141 

142 self._update_target_status(target, is_healthy) 

143 

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

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

146 if ( 

147 not is_healthy 

148 and target.circuit_breaker is not None 

149 and target.circuit_breaker.state != CircuitState.OPEN 

150 ): 

151 _force_open_breaker(target.circuit_breaker, target.name) 

152 

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

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

155 if self._on_health_change is not None: 

156 self._on_health_change(target.name, is_healthy) 

157 

158 if is_healthy: 

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

160 else: 

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

162 

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

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

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

166 

167 self._check_and_open_breaker(target, is_healthy) 

168 

169 if previous == is_healthy: 

170 return 

171 

172 self._last_status[target.name] = is_healthy 

173 

174 self._notify_health_change(target, is_healthy) 

175 

176 async def _run(self) -> None: 

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

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

179 

180 

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

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

183 

184 Simulates enough failures to trip the breaker by recording 

185 a synthetic exception. 

186 

187 Args: 

188 breaker: The circuit breaker to trip. 

189 target_name: Target name for logging context. 

190 

191 """ 

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

193 

194 # Record failures until the breaker opens, but cap it to avoid infinite loops 

195 # if the circuit breaker's state is corrupted or mutated. 

196 max_attempts = breaker.config.failure_threshold + 5 

197 attempts = 0 

198 

199 while breaker.state != CircuitState.OPEN and attempts < max_attempts: 

200 breaker._record_failure(synthetic) 

201 attempts += 1 

202 

203 if breaker.state != CircuitState.OPEN: 

204 # Force open if it didn't open normally 

205 old_state = breaker.state 

206 with breaker._state.lock: 

207 breaker._state.state = CircuitState.OPEN 

208 breaker._notify_state_change(old_state, CircuitState.OPEN) 

209 

210 logger.warning( 

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

212 breaker.name, 

213 target_name, 

214 )