Python queues: bound waiting work and own its completion

First principles

Concurrency and backlog are different quantities. A semaphore can restrict how many tasks execute a dependency call, but thousands of tasks may still wait to acquire it. Each waiting task retains its coroutine state and often its request payload. A service with ten active operations can therefore consume enormous memory despite an apparently conservative concurrency setting. To bound waiting work, put a capacity at the point where ownership transfers from the producer to the service.

An asyncio queue supplies that boundary within one process. A producer attempting put waits when the queue is full. A worker takes an item, becoming responsible for that item until it reports completion. Queue capacity describes items still waiting inside the queue, not those already removed by workers. With capacity three and two workers, as many as five items can be outstanding inside this processing arrangement. Upstream producers may retain additional work while blocked, so the whole admission path still needs a deliberate limit.

Queue completion is accounting, not persistence. Every successful get increments the responsibility of its consumer to eventually call task_done exactly once. join waits until the unfinished count reaches zero. It cannot tell whether an email was delivered, a database transaction committed, or an exception was swallowed. The business outcome must be represented separately. This separation is useful because a failed item can still be fully handled when its failure is recorded and the retry policy is explicit.

Runnable example

Run this file with Python 3.11 or later. It uses fixed synthetic values and no network. TaskGroup ensures an unexpected worker failure reaches the caller rather than leaving a detached failed task and a permanently waiting producer.

import asyncio

async def worker(queue, results):
    while True:
        item = await queue.get()
        try:
            if item is None:
                return
            await asyncio.sleep(0)
            results.append(item * item)
        finally:
            queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=3)
    results = []
    async with asyncio.TaskGroup() as group:
        for _ in range(2):
            group.create_task(worker(queue, results))
        for item in range(8):
            await queue.put(item)
            assert queue.qsize() <= 3
        for _ in range(2):
            await queue.put(None)
        await queue.join()
    assert sorted(results) == [0, 1, 4, 9, 16, 25, 36, 49]
    print("completed", len(results))

asyncio.run(main())

The sentinel is valid here because real jobs are integers and None is reserved for shutdown. Each worker receives one sentinel and exits after acknowledging it. FIFO placement means sentinels follow accepted jobs, although parallel workers need not finish jobs in input order. Sorting results verifies the completed set without accidentally making an ordering promise. A real job model should make shutdown messages unambiguous instead of choosing a value that could occur in user data.

Contextual failure case

A report endpoint launches one task per uploaded row and then uses a semaphore around SQL execution. Uploads initially seem fast because task creation does not wait for database capacity. During a burst, memory climbs, callers disconnect, and stale tasks continue holding large records. Replacing the list of tasks with a bounded queue changes where waiting occurs, but the endpoint still needs a maximum upload size and an admission deadline. Otherwise the producer merely becomes the new unbounded buffer.

A second failure occurs when a worker throws before acknowledging an item. Without structured task ownership, join can wait forever while no healthy worker remains. The finally block prevents accounting leakage, and TaskGroup exposes unexpected failures. Neither mechanism makes the job successful. If work must survive process termination, persist it in a durable queue or transactionally recorded work table. An in-memory queue is deliberately lost when the process exits.

Verification

The example prints completed 8. Introduce an exception for item four and verify that the program terminates with an exception group instead of silently reporting eight successful jobs. To test backpressure deterministically, fill a separate queue of capacity one, start a second put, yield once, and assert that producer task is unfinished. Remove and acknowledge the first item, then await the producer and acknowledge the second. Avoid timing-sensitive claims based on a few milliseconds of sleep.

Inspect a small synthetic job record with payload tools when deciding which fields enter the queue. Retain only the fields workers require. This is both a memory budget decision and a data minimization decision; storing the entire request object for convenience also keeps headers, credentials, and unrelated personal data alive.

Exercise

Replace square calculation with a handler that raises a known validation error for negative inputs. Collect successful results and rejected job IDs separately, while allowing unexpected errors to escape. Submit two valid values and one invalid value, and prove that queue completion accounts for all three but business success accounts for only two. Explain why retrying a permanently invalid value is a different policy from retrying a transient dependency failure.

Safety and caveats

Asyncio queues are for one event loop, not a thread-safe bridge. Blocking handlers stall that loop even when queue capacity is small. Bound payload bytes as well as item count because three enormous jobs can exceed memory limits. Cancellation after get requires an explicit disposition for the item; acknowledging it during unwinding does not preserve it for retry. During graceful shutdown, decide whether to drain, persist, or reject remaining jobs before the host's termination deadline arrives.