asyncio Queue.join Hangs With an Empty Queue: Track Work Items and Worker Failures

If an asyncio queue is empty but await queue.join() still waits, inspect the consumer that removed the last item. Removing an item and acknowledging its processing are separate operations. A worker may have raised an exception between them. Also check whether the queue contains coroutine objects that nobody actually awaits: putting an object in a queue does not execute it.

This guide uses Python 3.11 or later. Its examples make no network requests and write no files. They demonstrate accounting and failure reporting, not durable delivery or automatic retries.

Reproduce the empty queue with unfinished work

The first example deliberately leaves one item unacknowledged. An event makes the ordering explicit; a timeout bounds the diagnostic wait instead of leaving the process stuck.

import asyncio

async def main():
    queue = asyncio.Queue()
    removed = asyncio.Event()
    queue.put_nowait("job-1")

    async def broken_worker():
        item = await queue.get()
        removed.set()
        raise RuntimeError(f"failed: {item}")

    worker = asyncio.create_task(broken_worker())
    await removed.wait()
    assert queue.empty()
    try:
        await asyncio.wait_for(queue.join(), timeout=0.1)
    except TimeoutError:
        print("Queue is empty, but join is still waiting")
    else:
        raise AssertionError("Expected missing acknowledgement")

    try:
        await worker
    except RuntimeError as exc:
        print(exc)
    # This isolated demonstration explicitly abandons the failed item.
    # Production code needs a recorded failure/retry policy first.
    queue.task_done()
    await queue.join()

asyncio.run(main())

Expected output is Queue is empty, but join is still waiting, then failed: job-1. Increasing the timeout does not repair the missing acknowledgement. The final acknowledgement is explicit cleanup for the abandoned demonstration item, not evidence that its processing succeeded.

For diagnosis, record accepted, retrieved, succeeded, failed and acknowledged item counts alongside worker exceptions. Avoid repairing a live queue by repeatedly calling task_done() until an exception appears. That destroys the relationship between real processing and the accounting you need to investigate.

A queue holds data; it does not schedule it

A reader reported this naming confusion in CPython issue 124433 in September 2024. The documentation issue was closed in January 2025. It is historical evidence of a reader problem, not an unresolved runtime defect or measured current search volume.

Here, “task” means a unit of work. It need not be an asyncio.Task. Prefer putting plain job data in the queue, then having a worker call and await the processing function. If you instead enqueue process(job), you have created a coroutine object; some consumer must still await it. If you enqueue an already scheduled Task, define separately who observes its result and who acknowledges the work item. Neither choice gives queue accounting automatic knowledge of processing success.

Report worker failure while waiting for completion

For a finite batch, use structured task ownership so a failed worker can interrupt the wait. This example deliberately treats each processing attempt as terminal: success and failure are recorded separately, and every retrieved item receives one acknowledgement when its attempt ends. A failed item is not silently retried.

import asyncio

async def run_batch(jobs):
    queue = asyncio.Queue()
    results, failures = [], []
    for job in jobs:
        queue.put_nowait(job)

    async def process(job):
        await asyncio.sleep(0)
        if job == "bad":
            raise ValueError("invalid job")
        return job.upper()

    async def worker():
        while True:
            job = await queue.get()
            try:
                try:
                    results.append(await process(job))
                except Exception as exc:
                    failures.append((job, type(exc).__name__))
                    raise
            finally:
                queue.task_done()

    try:
        async with asyncio.TaskGroup() as group:
            workers = [group.create_task(worker()) for _ in range(2)]
            await queue.join()
            for task in workers:
                task.cancel()
    except* ValueError:
        print("Batch failed:", failures)
        raise
    return results

async def main():
    assert sorted(await run_batch(["a", "b"])) == ["A", "B"]
    try:
        await run_batch(["bad", "a", "b", "c"])
    except ExceptionGroup:
        print("Caller observed failure; do not report batch success")
    else:
        raise AssertionError("Worker failure was lost")

asyncio.run(main())

The successful batch returns two results. The failing batch prints its failure record and the caller's failure message. Other jobs may have completed or may remain unprocessed when sibling workers are cancelled; do not assume a fixed result count or retry the entire batch blindly. The abandoned queue is local to that failed invocation. A real service must retain or reconcile unprocessed job IDs before discarding it.

Decide what cancellation and acknowledgement mean

The finally belongs after a successful get(). Cancellation while waiting for an item must not acknowledge work that was never retrieved. Cancellation during processing does run this cleanup, but it does not imply success. In a durable system, record an interrupted attempt and decide who owns retry before adopting this pattern. This in-memory example intentionally provides no crash recovery.

A TaskGroup propagates ordinary worker failures and cancels siblings; a worker that swallows its exception and continues has chosen a different policy. An externally cancelled worker also needs explicit supervision. Do not use a completion counter as your only health signal. If every worker can exit while items remain queued, a completion wait still has nobody left to satisfy it.

Verify the fix against the actual failure

Run the empty-queue reproduction, the successful batch and the failed batch. Check that failure reaches the caller, not just the log. Add a processing-cancellation case for your chosen retry policy, and verify that no item is both retried and reported as successful without an idempotency rule.

For Python 3.13 and later, queue shutdown is a separate lifecycle feature. Immediate shutdown can discard queued work and allow a join to finish without processing it. It is an abort mechanism, not a repair for missing success records. Do not add it merely to make the hang disappear.

If your underlying worker runs blocking code through to_thread, cancellation has another boundary: see why a timed-out thread can keep running. For synchronous consumers, why Queue.empty is not a reservation covers a different race. Neither replaces work-item accounting or worker-error observation.