Coverage for src/taipanstack/resilience/adaptive/bulkhead.py: 100%
112 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-10-13 20:52 +0000
1"""
2Bulkhead pattern — concurrency isolation via semaphore.
4Limits the number of concurrent executions to prevent a single
5failing dependency from consuming all available resources.
6"""
8from __future__ import annotations
10import asyncio
11import contextlib
12import logging
13import math
14from collections.abc import Awaitable, Callable, Coroutine
15from typing import ParamSpec, TypeVar
17from taipanstack.core.result import Err, Ok, Result
19logger = logging.getLogger("taipanstack.resilience.adaptive.bulkhead")
21T = TypeVar("T")
22P = ParamSpec("P")
25class BulkheadFullError(Exception):
26 """Raised when the bulkhead queue is at capacity."""
28 def __init__(self, name: str, max_concurrent: int, max_queue: int) -> None:
29 """Initialize BulkheadFullError.
31 Args:
32 name: Bulkhead name.
33 max_concurrent: Concurrency limit.
34 max_queue: Queue limit.
36 """
37 self.bulkhead_name = name
38 self.max_concurrent = max_concurrent
39 self.max_queue = max_queue
40 super().__init__(
41 f"Bulkhead '{name}' is full "
42 f"(max_concurrent={max_concurrent}, max_queue={max_queue})",
43 )
46class Bulkhead:
47 """Concurrency limiter using ``asyncio.Semaphore``.
49 Limits the number of concurrent executions of a callable.
50 Excess callers are queued up to ``max_queue``; beyond that
51 a ``BulkheadFullError`` is returned.
53 Args:
54 name: Identifier for logging.
55 max_concurrent: Maximum concurrent executions.
56 max_queue: Maximum queued callers beyond concurrent limit.
57 timeout: Seconds to wait for a permit before timing out.
59 Example:
60 >>> bulk = Bulkhead("db", max_concurrent=5, max_queue=10)
61 >>> result = await bulk.execute(fetch_data, user_id)
63 """
65 def __init__(
66 self,
67 name: str = "default",
68 *,
69 max_concurrent: int = 10,
70 max_queue: int = 50,
71 timeout: float = 30.0,
72 ) -> None:
73 """Initialize the bulkhead.
75 Args:
76 name: Bulkhead name.
77 max_concurrent: Concurrency limit.
78 max_queue: Queue limit.
79 timeout: Permit acquisition timeout.
81 """
82 self.name = name
83 self._max_concurrent = self._validate_max_concurrent(max_concurrent)
84 self._max_queue = self._validate_max_queue(max_queue)
85 self._timeout = self._validate_timeout(timeout)
86 self._semaphore = asyncio.Semaphore(self._max_concurrent)
87 self._queued = 0
88 self._active = 0
90 @staticmethod
91 def _validate_max_concurrent(value: int) -> int:
92 if not isinstance(value, int) or isinstance(value, bool) or value < 1:
93 raise ValueError("max_concurrent must be an integer >= 1")
94 return value
96 @staticmethod
97 def _validate_max_queue(value: int) -> int:
98 if not isinstance(value, int) or isinstance(value, bool) or value < 0:
99 raise ValueError("max_queue must be an integer >= 0")
100 return value
102 @staticmethod
103 def _validate_timeout(value: float) -> float:
104 if not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0:
105 raise ValueError("timeout must be a finite non-negative number")
106 return float(value)
108 @property
109 def available_permits(self) -> int:
110 """Number of available concurrency permits."""
111 return self._max_concurrent - self._active
113 @property
114 def queued(self) -> int:
115 """Number of callers currently waiting in the queue."""
116 return self._queued
118 @property
119 def active(self) -> int:
120 """Number of currently executing tasks."""
121 return self._active
123 async def _cleanup_acquire_task(
124 self,
125 acquire_task: asyncio.Task[bool],
126 ) -> None:
127 """Helper to cancel an acquisition task and release semaphore if acquired."""
128 acquire_task.cancel()
129 with contextlib.suppress(asyncio.CancelledError):
130 await acquire_task
131 self._semaphore.release()
133 def _get_acquire_coro(self) -> Result[Coroutine[object, object, bool], Exception]:
134 """Attempt to get the semaphore acquire coroutine."""
135 try:
136 return Ok(self._semaphore.acquire())
137 except (RuntimeError, OSError, MemoryError) as e:
138 return Err(RuntimeError(f"Resource exhaustion: {e!s}"))
139 except Exception as e:
140 return Err(RuntimeError(f"Resource exhaustion: {e!s}"))
142 def _close_coro_if_needed(self, coro: Coroutine[object, object, bool]) -> None:
143 if asyncio.iscoroutine(coro):
144 coro.close()
146 async def _wait_for_coro(
147 self, coro: Coroutine[object, object, bool]
148 ) -> Result[None, Exception]:
149 """Wait for the acquire task to complete, handling errors and cleanup."""
150 try:
151 # We explicitly check for exception during the coroutine creation or wait
152 acquire_task: asyncio.Task[bool] = asyncio.create_task(coro)
153 return await _wait_for_permit_task(
154 acquire_task,
155 self._timeout,
156 self.name,
157 self._cleanup_acquire_task,
158 )
159 except Exception as e:
160 self._close_coro_if_needed(coro)
161 return Err(RuntimeError(f"Resource exhaustion: {e!s}"))
163 async def _acquire_permit(self) -> Result[None, Exception]:
164 """Wait for and acquire a concurrency permit."""
165 coro_result = self._get_acquire_coro()
166 if isinstance(coro_result, Err):
167 return coro_result
169 return await self._wait_for_coro(coro_result.unwrap())
171 async def _attempt_acquire(self) -> Result[None, Exception]:
172 """Attempt to queue and acquire a permit."""
173 self._queued += 1
174 try:
175 permit_result = await self._acquire_permit()
176 if isinstance(permit_result, Err):
177 return permit_result
178 return Ok(None)
179 finally:
180 self._queued -= 1
182 async def _execute_with_permit(
183 self,
184 fn: Callable[P, Awaitable[T]],
185 *args: P.args,
186 **kwargs: P.kwargs,
187 ) -> Result[T, Exception]:
188 """Execute the function within the acquired permit."""
189 self._active += 1
190 try:
191 result = await fn(*args, **kwargs)
192 return Ok(result)
193 except Exception as exc:
194 logger.warning(
195 "Bulkhead '%s' execution failed: %s",
196 self.name,
197 exc,
198 )
199 return Err(exc)
200 finally:
201 self._active -= 1
202 self._semaphore.release()
204 async def execute(
205 self,
206 fn: Callable[P, Awaitable[T]],
207 *args: P.args,
208 **kwargs: P.kwargs,
209 ) -> Result[T, Exception]:
210 """Execute a callable within bulkhead limits.
212 Args:
213 fn: Async callable to execute.
214 *args: Positional arguments for fn.
215 **kwargs: Keyword arguments for fn.
217 Returns:
218 ``Ok(result)`` on success, ``Err`` on failure.
220 """
221 # Check queue capacity
222 if self._queued >= self._max_queue:
223 return Err(
224 BulkheadFullError(
225 self.name,
226 self._max_concurrent,
227 self._max_queue,
228 ),
229 )
231 acquire_result = await self._attempt_acquire()
232 if isinstance(acquire_result, Err):
233 return acquire_result
235 return await self._execute_with_permit(fn, *args, **kwargs)
238async def _wait_for_permit_task(
239 acquire_task: asyncio.Task[bool],
240 timeout: float,
241 name: str,
242 cleanup_func: Callable[[asyncio.Task[bool]], Awaitable[None]],
243) -> Result[None, Exception]:
244 """Helper to wait for a permit task with timeout and cancellation."""
245 try:
246 await asyncio.wait_for(
247 asyncio.shield(acquire_task),
248 timeout=timeout,
249 )
250 return Ok(None)
251 except TimeoutError:
252 await cleanup_func(acquire_task)
253 return Err(
254 TimeoutError(
255 f"Bulkhead '{name}' timed out after {timeout}s waiting for permit",
256 ),
257 )
258 except asyncio.CancelledError:
259 await cleanup_func(acquire_task)
260 raise