process_stream: a bounded pipeline that cannot hang

Build the bounded pipeline: one source, workers consumers, a queue of maxsize, and no way to hang.

async def process_stream[T, R](
    items: AsyncIterator[T],
    handle: Callable[[T], Awaitable[R]],
    *,
    workers: int,
    maxsize: int,
) -> list[R]

• Results in completion order — not input order.
• The source must never get more than maxsize + workers + 1 items ahead of completed work. That is not a target, it is an invariant of the design: at most maxsize items sit in the queue, at most workers are in flight, plus the one the producer has pulled from the source and not yet enqueued.
• Clean shutdown when the source is exhausted. The starter provides a Sentinel class and a SENTINEL singleton — push one per worker.
• A handler that raises cancels the rest, and no task may be left pending.

The two ways this hangs

A missing task_done(). The queue’s unfinished counter is incremented by put and decremented by task_done() — never by get(). A handler that raises and skips its task_done() leaves the counter above zero forever. The finally is not optional.

A dead consumer and a full queue. If a worker dies while the producer is blocked in await queue.put(...), the producer waits for space that will never appear, and the surviving workers wait on get() for sentinels the producer will never push. Deadlock, no exception, no log line.

The fix for the second is structural and is the whole reason to use a TaskGroup here: put the producer in the body of the async with, not in a task. When a consumer fails, the group cancels the body — so the blocked put is cancelled instead of hanging — then cancels the other consumers, waits for them to unwind, and raises.

What the report proves

• results — completion order, which for these inputs is genuinely not input order.
• handled — every item handled exactly once (sorted, so it is a set-equality check rather than an ordering one).
• max_lead / lead_ok — the furthest the source ever got ahead. An implementation that drains the whole AsyncIterator into a list first will have a lead equal to the item count and fails here. This is what “backpressure is real” means as an assertion.
• outcome / group_size / group_types — a failing handler produces an ExceptionGroup with exactly the real failure in it.
• pending — 0 on every path, including the failure path.

Where the type system earns its keep

The sentinel forces asyncio.Queue[T | Sentinel], and every consumer must now narrow with isinstance before it can touch an item. That union is a tax the shutdown protocol charges the element type, and it propagates into everything the item is passed to.

This is the concrete argument for Python 3.13’s Queue.shutdown() and QueueShutDown: the queue stays Queue[T], the consumer loop becomes a try/except QueueShutDown, and the element type stops carrying a lifecycle concern it has nothing to do with. This problem targets 3.12, so you write the sentinel version — and you feel exactly what the newer API is buying.