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 @staticmethod 

74 def _validate_timeout_param(timeout: float) -> None: 

75 """Validate timeout parameter.""" 

76 if ( 

77 not isinstance(timeout, (int, float)) 

78 or not math.isfinite(timeout) 

79 or timeout < 0 

80 ): 

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

82 

83 def with_bulkhead( 

84 self, 

85 max_concurrent: int = 10, 

86 max_queue: int = 50, 

87 timeout: float = 30.0, 

88 ) -> ResilienceOrchestrator[T]: 

89 """Add a bulkhead concurrency limiter. 

90 

91 Args: 

92 max_concurrent: Max concurrent executions. 

93 max_queue: Max queued callers. 

94 timeout: Permit acquisition timeout. 

95 

96 Returns: 

97 self for chaining. 

98 

99 """ 

100 self._validate_timeout_param(timeout) 

101 self._bulkhead = Bulkhead( 

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

103 max_concurrent=max_concurrent, 

104 max_queue=max_queue, 

105 timeout=timeout, 

106 ) 

107 return self 

108 

109 def with_circuit_breaker( 

110 self, 

111 breaker: CircuitBreaker | AdaptiveCircuitBreaker, 

112 ) -> ResilienceOrchestrator[T]: 

113 """Add a circuit breaker. 

114 

115 Args: 

116 breaker: Standard or adaptive circuit breaker. 

117 

118 Returns: 

119 self for chaining. 

120 

121 """ 

122 if isinstance(breaker, AdaptiveCircuitBreaker): 

123 self._adaptive_breaker = breaker 

124 self._breaker = None 

125 else: 

126 self._breaker = breaker 

127 return self 

128 

129 def with_retry( 

130 self, 

131 config: RetryConfig | AdaptiveRetry, 

132 ) -> ResilienceOrchestrator[T]: 

133 """Add retry logic. 

134 

135 Args: 

136 config: Standard retry config or adaptive retry. 

137 

138 Returns: 

139 self for chaining. 

140 

141 """ 

142 if isinstance(config, AdaptiveRetry): 

143 self._adaptive_retry = config 

144 self._retry_config = config.to_retry_config() 

145 else: 

146 self._retry_config = config 

147 return self 

148 

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

150 """Add a timeout. 

151 

152 Args: 

153 seconds: Maximum execution time. 

154 

155 Returns: 

156 self for chaining. 

157 

158 """ 

159 self._validate_timeout_param(seconds) 

160 self._timeout = seconds 

161 return self 

162 

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

164 """Add a fallback value for failures. 

165 

166 Args: 

167 value: Value to return on failure. 

168 

169 Returns: 

170 self for chaining. 

171 

172 """ 

173 self._fallback_value = value 

174 return self 

175 

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

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

178 return await bh._acquire_permit() 

179 

180 async def _attempt_bulkhead_acquire( 

181 self, 

182 bh: Bulkhead, 

183 ) -> Result[None, Exception]: 

184 """Attempt to queue and acquire a bulkhead permit.""" 

185 bh._queued += 1 

186 try: 

187 acquire_result = await self._acquire_bulkhead(bh) 

188 if isinstance(acquire_result, Err): 

189 return Err(acquire_result.unwrap_err()) 

190 return Ok(None) 

191 finally: 

192 bh._queued -= 1 

193 

194 async def _execute_bulkhead_inner( 

195 self, 

196 bh: Bulkhead, 

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

198 *args: P.args, 

199 **kwargs: P.kwargs, 

200 ) -> Result[T, Exception]: 

201 """Execute the inner function while holding the bulkhead permit.""" 

202 bh._active += 1 

203 try: 

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

205 except Exception as exc: 

206 return self._apply_fallback(Err(exc)) 

207 finally: 

208 bh._active -= 1 

209 bh._semaphore.release() 

210 

211 async def _execute_with_bulkhead( 

212 self, 

213 bh: Bulkhead, 

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

215 *args: P.args, 

216 **kwargs: P.kwargs, 

217 ) -> Result[T, Exception]: 

218 """Execute through the bulkhead layer.""" 

219 if bh.queued >= bh._max_queue: 

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

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

222 ) 

223 return self._apply_fallback(result) 

224 

225 acquire_result = await self._attempt_bulkhead_acquire(bh) 

226 if isinstance(acquire_result, Err): 

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

228 

229 return await self._execute_bulkhead_inner(bh, fn, *args, **kwargs) 

230 

231 async def execute( 

232 self, 

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

234 *args: P.args, 

235 **kwargs: P.kwargs, 

236 ) -> Result[T, Exception]: 

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

238 

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

240 

241 Args: 

242 fn: Async callable to execute. 

243 *args: Positional arguments. 

244 **kwargs: Keyword arguments. 

245 

246 Returns: 

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

248 

249 """ 

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

251 if self._bulkhead is not None: 

252 return await self._execute_with_bulkhead( 

253 self._bulkhead, 

254 fn, 

255 *args, 

256 **kwargs, 

257 ) 

258 

259 try: 

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

261 except Exception as exc: 

262 return self._apply_fallback(Err(exc)) 

263 

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

265 if ( 

266 self._adaptive_breaker is not None 

267 and not self._adaptive_breaker.should_allow() 

268 ): 

269 return Err( 

270 CircuitBreakerError( 

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

272 state=self._adaptive_breaker.state, 

273 ), 

274 ) 

275 return None 

276 

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

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

279 return Err( 

280 CircuitBreakerError( 

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

282 state=self._breaker.state, 

283 ), 

284 ) 

285 return None 

286 

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

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

289 if self._adaptive_breaker is not None: 

290 return self._evaluate_adaptive_breaker() 

291 return self._evaluate_standard_breaker() 

292 

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

294 """Record a successful execution outcome.""" 

295 if self._adaptive_breaker is not None: 

296 self._adaptive_breaker.record_success() 

297 elif self._breaker is not None: 

298 self._breaker._record_success() 

299 

300 if self._adaptive_retry is not None: 

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

302 

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

304 """Record a failed execution outcome.""" 

305 if self._adaptive_breaker is not None: 

306 self._adaptive_breaker.record_failure(error) 

307 elif self._breaker is not None: 

308 self._breaker._record_failure(error) 

309 

310 if self._adaptive_retry is not None: 

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

312 

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

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

315 if self._adaptive_retry is not None: 

316 return self._adaptive_retry.get_delay(attempt) 

317 if self._retry_config is not None: 

318 return calculate_delay(attempt, self._retry_config) 

319 return 0.0 

320 

321 async def _execute_inner( 

322 self, 

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

324 *args: P.args, 

325 **kwargs: P.kwargs, 

326 ) -> Result[T, Exception]: 

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

328 max_attempts = ( 

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

330 ) 

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

332 

333 async def _handle_retry_failure( 

334 self, 

335 error: Exception, 

336 attempt: int, 

337 max_attempts: int, 

338 ) -> bool: 

339 self._record_failure_outcome(error, attempt) 

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

341 delay = self._calculate_retry_delay(attempt) 

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

343 return True 

344 return False 

345 

346 def _handle_circuit_breaker_open( 

347 self, 

348 cb_err: Err[Exception], 

349 attempt: int, 

350 ) -> Result[T, Exception]: 

351 """Handle circuit breaker evaluation logic.""" 

352 if attempt == 1: 

353 return self._apply_fallback(cb_err) 

354 return cb_err 

355 

356 async def _execute_single_attempt( 

357 self, 

358 attempt: int, 

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

360 *args: P.args, 

361 **kwargs: P.kwargs, 

362 ) -> Result[T, Exception]: 

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

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

365 if isinstance(result, Ok): 

366 self._record_success_outcome(attempt) 

367 return result 

368 

369 def _check_circuit_breaker_for_attempt( 

370 self, 

371 attempt: int, 

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

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

374 cb_err = self._evaluate_circuit_breaker() 

375 if cb_err is not None: 

376 return self._handle_circuit_breaker_open(cb_err, attempt) 

377 return None 

378 

379 async def _process_retry_attempt( 

380 self, 

381 attempt: int, 

382 max_attempts: int, 

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

384 *args: P.args, 

385 **kwargs: P.kwargs, 

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

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

388 cb_res = self._check_circuit_breaker_for_attempt(attempt) 

389 if cb_res is not None: 

390 if isinstance(cb_res, Err): 

391 return False, cb_res.err_value 

392 return cb_res 

393 

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

395 if isinstance(result, Ok): 

396 return result 

397 

398 last_error = result.err_value 

399 should_continue = await self._handle_retry_failure( 

400 last_error, attempt, max_attempts 

401 ) 

402 return should_continue, last_error 

403 

404 async def _execute_with_retries( 

405 self, 

406 max_attempts: int, 

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

408 *args: P.args, 

409 **kwargs: P.kwargs, 

410 ) -> Result[T, Exception]: 

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

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

413 outcome = await self._process_retry_attempt( 

414 attempt, max_attempts, fn, *args, **kwargs 

415 ) 

416 if not isinstance(outcome, tuple): 

417 return outcome 

418 

419 should_continue, last_error = outcome 

420 if not should_continue: 

421 break 

422 

423 return self._apply_fallback(Err(last_error)) 

424 

425 async def _run_fn_with_timeout( 

426 self, 

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

428 *args: P.args, 

429 **kwargs: P.kwargs, 

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

431 if self._timeout is not None: 

432 return await asyncio.wait_for( 

433 fn(*args, **kwargs), 

434 timeout=self._timeout, 

435 ) 

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

437 

438 def _process_timeout_success_result( 

439 self, 

440 result: T | Result[T, Exception], 

441 ) -> Result[T, Exception]: 

442 """Process successful results from timeout execution.""" 

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

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

445 return Ok(result) 

446 

447 async def _execute_with_timeout( 

448 self, 

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

450 *args: P.args, 

451 **kwargs: P.kwargs, 

452 ) -> Result[T, Exception]: 

453 """Execute fn with optional timeout. 

454 

455 Args: 

456 fn: Async callable. 

457 *args: Positional arguments. 

458 **kwargs: Keyword arguments. 

459 

460 Returns: 

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

462 

463 """ 

464 try: 

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

466 return self._process_timeout_success_result(result) 

467 except TimeoutError: 

468 return Err( 

469 TimeoutError( 

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

471 ), 

472 ) 

473 except Exception as exc: 

474 return Err(exc) 

475 

476 def _apply_fallback( 

477 self, 

478 result: Result[T, Exception], 

479 ) -> Result[T, Exception]: 

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

481 

482 Args: 

483 result: The result to potentially replace. 

484 

485 Returns: 

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

487 

488 """ 

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

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

491 return result 

492 

493 

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

495_SENTINEL = object()