Capstone: a fully typed, cancellable, observable async worker pool

This object exists, badly, in most production Python codebases. It gets written incrementally: gather instead of a TaskGroup, no backpressure, cancellation that swallows CancelledError, and a shutdown path that has never been tested because the process is always killed instead.

class WorkerPool[T, R]:
    def __init__(self, handle: Callable[[T], Awaitable[R]], *,
                 workers: int, queue_size: int, item_timeout: float,
                 retryable: tuple[type[Exception], ...] = (),
                 max_retries: int = 0,
                 metrics: Callable[[str, T], None] | None = None) -> None: ...

    def request_stop(self) -> None: ...
    async def run(self, source: AsyncIterator[T]) -> AsyncIterator[R]: ...

The driver, the handler and the source are given. You write the pool.

The eight properties

1. Exactly once. Every item produces exactly one result or exactly one recorded failure. Never both, never neither.

2. Backpressure. The source must not run ahead. asyncio.Queue(maxsize=...) does this by itself: await inbox.put(item) blocks the producer, the producer stops pulling from the async for, and the async generator suspends. That is the entire mechanism, and it is why asyncio.gather(*[handle(i) async for i in source]) is the anti-pattern — it drains an unbounded source into memory before doing any work. The bound checked here is produced - started <= queue_size + workers + 1: at the moment an item is yielded, at most queue_size are waiting, at most workers are in flight, and one is the item itself.

3. A per-item deadline. asyncio.wait_for(task, item_timeout). A timeout produces a retry or a recorded failure — never a hang, because a hung worker is an outage that no alert fires for.

4. Retries with a budget. Exceptions listed in retryable (plus TimeoutError) are retried up to max_retries. Everything else fails immediately: retrying a ValueError from malformed input is just doing the same wrong thing four more times.

5. Failures aggregate. At the end, if anything failed permanently, raise a single ExceptionGroup carrying exactly those exceptions. Not the first one — an operator needs to see that eleven items failed, not one.

6. request_stop() drains. It stops the source, lets in-flight items finish, and returns normally. It must not raise CancelledError at the caller — a graceful stop is not an error, and the caller’s except clauses should not have to know the difference.

7. Context is per item. A ContextVar set while handling item 5 must not be visible while handling item 6, even in the same worker. This is free if each item runs in its own Task: creating a Task copies the current context. It is not free if you await self._handle(item) inline in the worker loop, where every item shares the worker’s context and request-scoped state leaks between unrelated units of work. That is a genuine production bug class — the trace id on your log line belongs to the previous request.

8. No orphans. asyncio.all_tasks() after run() returns must contain only the caller. asyncio.TaskGroup gives you this: it cancels its siblings on the first exception and does not return until every child is finished. A list of create_task results does not.

The observability seam

metrics(event, item) is called with "ok", "retry" or "failed". It is someone else’s code, and one of the test cases has it raise on every call. The pool must be unaffected. The general rule: instrumentation is allowed to be lossy, never allowed to be fatal.

Version tolerance is part of the grade

The clean-shutdown problem — telling workers consumers that no more work is coming — has two answers. On 3.13+, asyncio.Queue.shutdown() wakes every waiter with QueueShutDown. On 3.12 you push one sentinel per worker. A library that targets both writes the sentinel version, or feature-detects; a library that targets only its author’s interpreter breaks on a downstream user’s.

The typing

WorkerPool[T, R] with no Any in any public signature. handle is Callable[[T], Awaitable[R]]; run is AsyncIterator[R]. The end-of-stream sentinel needs care: None is a legitimate R, so a private sentinel class and Queue[R | _Done] is the honest encoding — isinstance(item, _Done) narrows, and nothing is ambiguous.

One small friction worth knowing about: asyncio.create_task is typed to accept a Coroutine, not an Awaitable, so it will not take the result of calling a Callable[[T], Awaitable[R]]. asyncio.ensure_future accepts both and still wraps a coroutine in a Task — same context copy, same cancellation semantics, and it type-checks.