Process a Stream with a Bounded Queue

Build a small pipeline. A producer reads items from an async source into a bounded queue. A fixed number of workers take items from the queue, call a handler on each, and collect the results. The queue keeps the producer from running far ahead.

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

For example, take the items 0, 1, 2, a handler that returns item * 10, workers=1 and maxsize=1. The result is [0, 10, 20].

What to write

workers and maxsize are both at least 1. The source is finite, and it never yields an instance of the starter's Sentinel class.

• Create asyncio.Queue(maxsize=maxsize) and start exactly workers worker tasks. Each worker handles one item at a time.
• Append each result as soon as its handler returns, so the list is in the order the handlers finished.
• Read the source one item at a time with async for, and await queue.put(item) for each one. Reading it all into a list first fails the tests. They check that the source stays at most maxsize + workers + 1 items ahead of the finished work.
• After the last item, put SENTINEL in the queue once per worker. A worker that gets a sentinel returns.
• Call queue.task_done() exactly once for every item a worker gets, sentinels included, in a finally.
• Run the producer and the workers in one asyncio.TaskGroup. The producer can be the group's body or its own task. Return only after every worker has finished.
• If a handler raises, the group cancels the other tasks and raises an ExceptionGroup. Let it propagate.
• Annotate the queue as asyncio.Queue[T | Sentinel], and check isinstance(item, Sentinel) before you pass an item to handle.

Edge cases

• The source can be empty. The workers still need their sentinels to stop.
• There can be more workers than items. Some workers get only a sentinel.

What the driver reports

Keep solve as it is. Its handler takes a set number of await asyncio.sleep(0) steps for each item, so items can finish out of order. Some tests make it raise ValueError for one item. The report has these keys:

• results: the list you returned, or [] when the run fails.
• handled: the items handled successfully, sorted. An item handled twice appears twice.
• max_lead: the most items the source was ever ahead of the finished work.
• lead_ok: True when max_lead stayed within maxsize + workers + 1.
• outcome: "ok", or "ExceptionGroup" when the run failed.
• group_size and group_types: the number of errors in that group and their type names. The tests where a handler fails expect one ValueError.
• pending: tasks other than the driver still running at the end. It must be 0.