Bound concurrent work and preserve result order

Pagination within one category is sequential: the next opaque cursor arrives in the previous reply. Different categories can be independent. Concurrency belongs at that independent-work boundary.

A fixed async worker pool gives us a clear limit. The queue contains at most 32 jobs, and at most four consumers run. Each consumer stores its result at the original job index, so a fast request does not steal another category’s result position.

The supplied consume coroutine checks cancellation, takes a queued job, awaits the worker and stores the returned value. Your block owns the tasks around it. Create min(limit,job_count) consumers and await the group.

On any failure, including external CancelledError, cancel every sibling and await them with return_exceptions=True before re-raising. This cleanup catch is intentionally broad; it does not convert failures into success.

import asyncio
async def main():
    results = [None]*3
    async def work(index,value):
        await asyncio.sleep(0)
        results[index] = value*value
    await asyncio.gather(*(work(i,n) for i,n in enumerate([3,1,2])))
    print(results)
asyncio.run(main())

There are two different cancellation boundaries. Async workers that cooperate with cancellation can unwind promptly. A synchronous HTTP call running through asyncio.to_thread cannot be forcibly stopped by canceling its wrapper. The lab combines a shared stop flag before new requests with per-operation socket timeouts. An in-flight operation can still finish; cancellation does not roll it back.

The capstone uses independent category chains, never speculative requests for unknown cursors. Its failure path signals other collectors to stop before another request. The top-level asyncio.run also waits for its default executor during shutdown. None of these facts is a hard total-runtime deadline for arbitrary hostile servers.

Tests measure peak active workers and verify cleanup after a forced failure and an external cancellation. They also check stable output with duplicate input values and legitimate None results.

import asyncio
async def main():
    closed=[]
    async def worker():
        try:await asyncio.sleep(10)
        finally:closed.append("released")
    task=asyncio.create_task(worker())
    await asyncio.sleep(0)
    task.cancel()
    await asyncio.gather(task,return_exceptions=True)
    print(closed)
asyncio.run(main())

Complete task ownership and test peak concurrency, input-order output, empty jobs, worker failure and external cancellation. Keep the distinction between draining async tasks and force-stopping a thread.

Next the report composes all these helpers against a real loopback server, using category concurrency and sequential pagination inside each category.