Coverage for src/taipanstack/resilience/adaptive/orchestrator.py: 100%

1""" 

2Resilience Orchestrator — compose multiple patterns into a pipeline. 

3 

4Provides a fluent builder to combine bulkhead, circuit breaker, 

5retry, timeout, and fallback into a single execution pipeline. 

6 

7Execution order: bulkhead → circuit breaker → retry → timeout → fn → fallback. 

8""" 

9 

10from __future__ import annotations 

11 

12import asyncio 

13import logging 

14import math 

15from collections.abc import Awaitable, Callable 

16from typing import Generic, ParamSpec, TypeVar, cast 

17 

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

19from taipanstack.resilience.adaptive.adaptive_breaker import AdaptiveCircuitBreaker 

20from taipanstack.resilience.adaptive.adaptive_retry import AdaptiveRetry 

21from taipanstack.resilience.adaptive.bulkhead import Bulkhead, BulkheadFullError 

22from taipanstack.resilience.circuit_breaker import ( 

23 CircuitBreaker, 

24 CircuitBreakerError, 

25) 

26from taipanstack.resilience.retry import RetryConfig, calculate_delay 

27 

28logger = logging.getLogger("taipanstack.resilience.adaptive.orchestrator") 

29 

30T = TypeVar("T") 

31E = TypeVar("E", bound=Exception) 

32P = ParamSpec("P") 

33 

34 

35class ResilienceOrchestrator(Generic[T]): 

36 """Compose resilience patterns into a single pipeline. 

37 

38 Provides a fluent builder API to add patterns in order. 

39 Execution proceeds through each configured layer. 

40 

41 Args: 

42 name: Pipeline name for logging. 

43 

44 Example: 

45 >>> orch = ( 

46 ... ResilienceOrchestrator("api") 

47 ... .with_bulkhead(max_concurrent=5) 

48 ... .with_circuit_breaker(breaker) 

49 ... .with_retry(RetryConfig(max_attempts=3)) 

50 ... .with_timeout(10.0) 

51 ... .with_fallback({"status": "cached"}) 

52 ... ) 

53 >>> result = await orch.execute(call_api, endpoint) 

54 

55 """ 

56 

57 def __init__(self, name: str = "default") -> None: 

58 """Initialize the orchestrator. 

59 

60 Args: 

61 name: Pipeline name. 

62 

63 """ 

64 self.name = name 

65 self._bulkhead: Bulkhead | None = None 

66 self._breaker: CircuitBreaker | None = None 

67 self._adaptive_breaker: AdaptiveCircuitBreaker | None = None 

68 self._retry_config: RetryConfig | None = None 

69 self._adaptive_retry: AdaptiveRetry | None = None 

70 self._timeout: float | None = None 

71 self._fallback_value: T | object = _SENTINEL 

72 

73 def with_bulkhead( 

74 self, 

75 max_concurrent: int = 10, 

76 max_queue: int = 50, 

77 timeout: float = 30.0, 

78 ) -> ResilienceOrchestrator[T]: 

79 """Add a bulkhead concurrency limiter. 

80 

81 Args: 

82 max_concurrent: Max concurrent executions. 

83 max_queue: Max queued callers. 

84 timeout: Permit acquisition timeout. 

85 

86 Returns: 

87 self for chaining. 

88 

89 """ 

90 if not math.isfinite(timeout) or timeout < 0: 

91 raise ValueError("timeout must be a finite non-negative number") 

92 self._bulkhead = Bulkhead( 

93 f"{self.name}-bulkhead", 

94 max_concurrent=max_concurrent, 

95 max_queue=max_queue, 

96 timeout=timeout, 

97 ) 

98 return self 

99 

100 def with_circuit_breaker( 

101 self, 

102 breaker: CircuitBreaker | AdaptiveCircuitBreaker, 

103 ) -> ResilienceOrchestrator[T]: 

104 """Add a circuit breaker. 

105 

106 Args: 

107 breaker: Standard or adaptive circuit breaker. 

108 

109 Returns: 

110 self for chaining. 

111 

112 """ 

113 if isinstance(breaker, AdaptiveCircuitBreaker): 

114 self._adaptive_breaker = breaker 

115 self._breaker = None 

116 else: 

117 self._breaker = breaker 

118 return self 

119 

120 def with_retry( 

121 self, 

122 config: RetryConfig | AdaptiveRetry, 

123 ) -> ResilienceOrchestrator[T]: 

124 """Add retry logic. 

125 

126 Args: 

127 config: Standard retry config or adaptive retry. 

128 

129 Returns: 

130 self for chaining. 

131 

132 """ 

133 if isinstance(config, AdaptiveRetry): 

134 self._adaptive_retry = config 

135 self._retry_config = config.to_retry_config() 

136 else: 

137 self._retry_config = config 

138 return self 

139 

140 def with_timeout(self, seconds: float) -> ResilienceOrchestrator[T]: 

141 """Add a timeout. 

142 

143 Args: 

144 seconds: Maximum execution time. 

145 

146 Returns: 

147 self for chaining. 

148 

149 """ 

150 if not math.isfinite(seconds) or seconds < 0: 

151 raise ValueError("timeout must be a finite non-negative number") 

152 self._timeout = seconds 

153 return self 

154 

155 def with_fallback(self, value: T) -> ResilienceOrchestrator[T]: 

156 """Add a fallback value for failures. 

157 

158 Args: 

159 value: Value to return on failure. 

160 

161 Returns: 

162 self for chaining. 

163 

164 """ 

165 self._fallback_value = value 

166 return self 

167 

168 async def _acquire_bulkhead(self, bh: Bulkhead) -> Result[None, Exception]: 

169 """Attempt to acquire a bulkhead permit, handling timeouts and errors.""" 

170 try: 

171 await asyncio.wait_for( 

172 bh._semaphore.acquire(), 

173 timeout=bh._timeout, 

174 ) 

175 return Ok(None) 

176 except TimeoutError: 

177 return Err( 

178 TimeoutError(f"Bulkhead '{bh.name}' timed out after {bh._timeout}s") 

179 ) 

180 except (RuntimeError, OSError, MemoryError) as e: 

181 return Err(RuntimeError(f"Resource exhaustion: {e!s}")) 

182 

183 async def _execute_with_bulkhead( 

184 self, 

185 bh: Bulkhead, 

186 fn: Callable[P, Awaitable[T]], 

187 *args: P.args, 

188 **kwargs: P.kwargs, 

189 ) -> Result[T, Exception]: 

190 """Execute through the bulkhead layer.""" 

191 if bh.queued >= bh._max_queue: 

192 result: Result[T, Exception] = Err( 

193 BulkheadFullError(bh.name, bh._max_concurrent, bh._max_queue), 

194 ) 

195 return self._apply_fallback(result) 

196 

197 bh._queued += 1 

198 try: 

199 acquire_result = await self._acquire_bulkhead(bh) 

200 if isinstance(acquire_result, Err): 

201 return self._apply_fallback(cast(Result[T, Exception], acquire_result)) 

202 finally: 

203 bh._queued -= 1 

204 

205 bh._active += 1 

206 try: 

207 try: 

208 return await self._execute_inner(fn, *args, **kwargs) 

209 except Exception as exc: 

210 return self._apply_fallback(Err(exc)) 

211 finally: 

212 bh._active -= 1 

213 bh._semaphore.release() 

214 

215 async def execute( 

216 self, 

217 fn: Callable[P, Awaitable[T]], 

218 *args: P.args, 

219 **kwargs: P.kwargs, 

220 ) -> Result[T, Exception]: 

221 """Execute the function through the resilience pipeline. 

222 

223 Order: bulkhead → circuit breaker → retry → timeout → fn → fallback. 

224 

225 Args: 

226 fn: Async callable to execute. 

227 *args: Positional arguments. 

228 **kwargs: Keyword arguments. 

229 

230 Returns: 

231 ``Ok(result)`` on success, ``Err`` on failure. 

232 

233 """ 

234 # Layer 1: Bulkhead — use semaphore directly to avoid double-wrapping 

235 if self._bulkhead is not None: 

236 return await self._execute_with_bulkhead( 

237 self._bulkhead, 

238 fn, 

239 *args, 

240 **kwargs, 

241 ) 

242 

243 try: 

244 return await self._execute_inner(fn, *args, **kwargs) 

245 except Exception as exc: 

246 return self._apply_fallback(Err(exc)) 

247 

248 def _evaluate_adaptive_breaker(self) -> Err[Exception] | None: 

249 if ( 

250 self._adaptive_breaker is not None 

251 and not self._adaptive_breaker.should_allow() 

252 ): 

253 return Err( 

254 CircuitBreakerError( 

255 f"Circuit '{self._adaptive_breaker.name}' is open", 

256 state=self._adaptive_breaker.state, 

257 ), 

258 ) 

259 return None 

260 

261 def _evaluate_standard_breaker(self) -> Err[Exception] | None: 

262 if self._breaker is not None and not self._breaker._should_attempt(): 

263 return Err( 

264 CircuitBreakerError( 

265 f"Circuit '{self._breaker.name}' is open", 

266 state=self._breaker.state, 

267 ), 

268 ) 

269 return None 

270 

271 def _evaluate_circuit_breaker(self) -> Err[Exception] | None: 

272 """Check if execution is allowed by the circuit breaker.""" 

273 if self._adaptive_breaker is not None: 

274 return self._evaluate_adaptive_breaker() 

275 return self._evaluate_standard_breaker() 

276 

277 def _record_success_outcome(self, attempt: int) -> None: 

278 """Record a successful execution outcome.""" 

279 if self._adaptive_breaker is not None: 

280 self._adaptive_breaker.record_success() 

281 elif self._breaker is not None: 

282 self._breaker._record_success() 

283 

284 if self._adaptive_retry is not None: 

285 self._adaptive_retry.record_outcome(attempt, True, 0.0) 

286 

287 def _record_failure_outcome(self, error: Exception, attempt: int) -> None: 

288 """Record a failed execution outcome.""" 

289 if self._adaptive_breaker is not None: 

290 self._adaptive_breaker.record_failure(error) 

291 elif self._breaker is not None: 

292 self._breaker._record_failure(error) 

293 

294 if self._adaptive_retry is not None: 

295 self._adaptive_retry.record_outcome(attempt, False, 0.0) 

296 

297 def _calculate_retry_delay(self, attempt: int) -> float: 

298 """Calculate the retry delay for the given attempt.""" 

299 if self._adaptive_retry is not None: 

300 return self._adaptive_retry.get_delay(attempt) 

301 if self._retry_config is not None: 

302 return calculate_delay(attempt, self._retry_config) 

303 return 0.0 

304 

305 async def _execute_inner( 

306 self, 

307 fn: Callable[P, Awaitable[T]], 

308 *args: P.args, 

309 **kwargs: P.kwargs, 

310 ) -> Result[T, Exception]: 

311 """Execute through breaker → retry → timeout → fn layers.""" 

312 max_attempts = ( 

313 self._retry_config.max_attempts if self._retry_config is not None else 1 

314 ) 

315 return await self._execute_with_retries(max_attempts, fn, *args, **kwargs) 

316 

317 async def _handle_retry_failure( 

318 self, 

319 error: Exception, 

320 attempt: int, 

321 max_attempts: int, 

322 ) -> bool: 

323 self._record_failure_outcome(error, attempt) 

324 if self._retry_config is not None and attempt < max_attempts: 

325 delay = self._calculate_retry_delay(attempt) 

326 await asyncio.sleep(min(delay, 3600.0)) 

327 return True 

328 return False 

329 

330 def _handle_circuit_breaker_open( 

331 self, 

332 cb_err: Err[Exception], 

333 attempt: int, 

334 ) -> Result[T, Exception] | Exception: 

335 """Handle circuit breaker evaluation logic.""" 

336 if attempt == 1: 

337 return self._apply_fallback(cb_err) 

338 return cb_err.err_value 

339 

340 async def _execute_single_attempt( 

341 self, 

342 attempt: int, 

343 fn: Callable[P, Awaitable[T]], 

344 *args: P.args, 

345 **kwargs: P.kwargs, 

346 ) -> Result[T, Exception]: 

347 """Execute a single attempt including timeout.""" 

348 result = await self._execute_with_timeout(fn, *args, **kwargs) 

349 if isinstance(result, Ok): 

350 self._record_success_outcome(attempt) 

351 return result 

352 

353 def _check_circuit_breaker_for_attempt( 

354 self, 

355 attempt: int, 

356 ) -> Result[T, Exception] | Exception | None: 

357 """Evaluate circuit breaker for the current attempt.""" 

358 cb_err = self._evaluate_circuit_breaker() 

359 if cb_err is not None: 

360 return self._handle_circuit_breaker_open(cb_err, attempt) 

361 return None 

362 

363 async def _process_retry_attempt( 

364 self, 

365 attempt: int, 

366 max_attempts: int, 

367 fn: Callable[P, Awaitable[T]], 

368 *args: P.args, 

369 **kwargs: P.kwargs, 

370 ) -> Result[T, Exception] | tuple[bool, Exception]: 

371 """Process a single retry attempt including circuit breaker check.""" 

372 cb_res = self._check_circuit_breaker_for_attempt(attempt) 

373 if cb_res is not None: 

374 if isinstance(cb_res, Exception): 

375 return False, cb_res 

376 return cb_res 

377 

378 result = await self._execute_single_attempt(attempt, fn, *args, **kwargs) 

379 if isinstance(result, Ok): 

380 return result 

381 

382 last_error = result.err_value 

383 should_continue = await self._handle_retry_failure( 

384 last_error, attempt, max_attempts 

385 ) 

386 return should_continue, last_error 

387 

388 async def _execute_with_retries( 

389 self, 

390 max_attempts: int, 

391 fn: Callable[P, Awaitable[T]], 

392 *args: P.args, 

393 **kwargs: P.kwargs, 

394 ) -> Result[T, Exception]: 

395 last_error: Exception = RuntimeError("Execution failed") 

396 for attempt in range(1, max_attempts + 1): 

397 outcome = await self._process_retry_attempt( 

398 attempt, max_attempts, fn, *args, **kwargs 

399 ) 

400 if not isinstance(outcome, tuple): 

401 return outcome 

402 

403 should_continue, last_error = outcome 

404 if not should_continue: 

405 break 

406 

407 return self._apply_fallback(Err(last_error)) 

408 

409 async def _run_fn_with_timeout( 

410 self, 

411 fn: Callable[P, Awaitable[T]], 

412 *args: P.args, 

413 **kwargs: P.kwargs, 

414 ) -> T | Result[T, Exception]: 

415 if self._timeout is not None: 

416 return await asyncio.wait_for( 

417 fn(*args, **kwargs), 

418 timeout=self._timeout, 

419 ) 

420 return await fn(*args, **kwargs) 

421 

422 async def _execute_with_timeout( 

423 self, 

424 fn: Callable[P, Awaitable[T]], 

425 *args: P.args, 

426 **kwargs: P.kwargs, 

427 ) -> Result[T, Exception]: 

428 """Execute fn with optional timeout. 

429 

430 Args: 

431 fn: Async callable. 

432 *args: Positional arguments. 

433 **kwargs: Keyword arguments. 

434 

435 Returns: 

436 ``Ok(result)`` or ``Err``. 

437 

438 """ 

439 try: 

440 result = await self._run_fn_with_timeout(fn, *args, **kwargs) 

441 

442 if isinstance(result, (Ok, Err)): 

443 return cast(Result[T, Exception], result) 

444 return Ok(result) 

445 except TimeoutError: 

446 return Err( 

447 TimeoutError( 

448 f"Pipeline '{self.name}' timed out after {self._timeout}s" 

449 ), 

450 ) 

451 except Exception as exc: 

452 return Err(exc) 

453 

454 def _apply_fallback( 

455 self, 

456 result: Result[T, Exception], 

457 ) -> Result[T, Exception]: 

458 """Apply fallback if configured and result is Err. 

459 

460 Args: 

461 result: The result to potentially replace. 

462 

463 Returns: 

464 Original result or ``Ok(fallback_value)``. 

465 

466 """ 

467 if isinstance(result, Err) and self._fallback_value is not _SENTINEL: 

468 return Ok(cast(T, self._fallback_value)) 

469 return result 

470 

471 

472# Sentinel for distinguishing "no fallback" from "fallback=None" 

473_SENTINEL = object()