Asyncio Tasks Silently Garbage Collected: Debugging the Disappearing Task

Overview

A background coroutine works in testing, then in production it sometimes stops halfway. No exception, no log line, no cancellation message. Add a print statement and the problem changes or disappears. That is the classic signature of a task being garbage collected while it is still pending.

This post explains why it happens and how to avoid it.

1. The Root Cause: The Event Loop Holds Weak References

When you call asyncio.create_task(coro()), the event loop only keeps a weak reference to the task. If your code does not hold a strong reference to the returned Task, the garbage collector may destroy it before it finishes.

The Python documentation for asyncio.create_task warns about this and recommends saving a reference to the result. Whether the task is collected depends on reference cycles and GC timing, which is why the failure is intermittent and hard to reproduce.

import asyncio


async def send_notification(user_id: int) -> None:
    await asyncio.sleep(1)
    print(f"sent to {user_id}")


async def handler(user_id: int) -> None:
    # Risky: nothing keeps the task alive.
    asyncio.create_task(send_notification(user_id))

2. Fix One: Keep a Strong Reference

The pattern recommended in the docs is a module-level set, with a done callback that removes the task:

background_tasks: set[asyncio.Task] = set()


def spawn(coro) -> asyncio.Task:
    task = asyncio.create_task(coro)
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)
    return task

The set holds the strong reference while the task runs, and the callback prevents the set from growing forever.

3. Exceptions Are Also Easy to Lose

A task that raises and is never awaited will not crash your program. Python logs "Task exception was never retrieved" only when the task object is finalized, and that message can be easy to miss or arrive late.

Make exceptions visible by logging them in the done callback:

import logging

logger = logging.getLogger(__name__)


def _on_done(task: asyncio.Task) -> None:
    background_tasks.discard(task)
    if task.cancelled():
        return
    exc = task.exception()
    if exc is not None:
        logger.error("Background task failed", exc_info=exc)


def spawn(coro) -> asyncio.Task:
    task = asyncio.create_task(coro)
    background_tasks.add(task)
    task.add_done_callback(_on_done)
    return task

Calling task.exception() also marks the exception as retrieved, which suppresses the later warning.

4. Prefer TaskGroup When the Work Has a Scope

If the tasks belong to a unit of work, avoid fire-and-forget and use asyncio.TaskGroup, available in Python 3.11 and later. It holds references to its tasks and waits for all of them when the block exits.

async def process(items: list[int]) -> None:
    async with asyncio.TaskGroup() as tg:
        for item in items:
            tg.create_task(handle(item))
    # All tasks finished here, or the group raised.

If a task fails, the group cancels the others and raises an ExceptionGroup, which you handle with except*. That gives you structured lifetimes and visible errors by default. For older Python versions, asyncio.gather with explicit awaiting is the usual substitute.

5. Handle Shutdown

A task set also helps on shutdown. Before exiting, cancel the remaining tasks and await them so cleanup code runs:

async def shutdown() -> None:
    tasks = list(background_tasks)
    for task in tasks:
        task.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)

Decide whether pending work should finish or be dropped, and document it. For work that must not be lost, a fire-and-forget task in process memory is the wrong tool; use a durable queue.

6. Debugging Tips

• Run with PYTHONASYNCIODEBUG=1 or asyncio.run(main(), debug=True) to get extra warnings, including never-retrieved exceptions with creation tracebacks.
• Enable -X dev mode during development.
• Name tasks with the name argument, so logs identify them.
• Add a log line at the start and end of background coroutines to see which ones never finish.

Conclusion

A task that nobody references can be collected mid-flight. Keep strong references, surface exceptions in a done callback, and prefer TaskGroup when the work has a clear scope. For work that must survive, use a real queue.

If you are chasing intermittent failures in an async Python service, I can help trace the cause and make background work more observable. Contact me.

Additional Resources

• Python docs: asyncio Tasks and TaskGroup
• Python docs: Developing with asyncio (debug mode)
• Python docs: Exception groups and except*