Debugging and introspecting async code

Async bugs are rarely wrong logic. They are a task nobody awaited, a synchronous call blocking the loop, or an exception that vanished — and none of those look like a normal stack trace.

Turn on debug mode first

asyncio.run(main(), debug=True)
# or: PYTHONASYNCIODEBUG=1

That one flag gives you three things you cannot get otherwise: a warning when a callback occupies the loop too long, a warning for coroutines that were never awaited, and the traceback of where a task was created rather than only where it failed.

That last one is the important one. Without it, a failure inside a task points at the event loop internals and tells you nothing about which call site started it.

The blocked event loop

The single most common production symptom, and the least obvious: p99 latency rises on every endpoint at once, including ones that do nothing. One synchronous call is stalling every other task.

async def handler(user_id):
    # Any of these stops the whole loop.
    time.sleep(1)                  # not asyncio.sleep
    requests.get(url)              # not httpx.AsyncClient
    hashlib.pbkdf2_hmac(...)       # CPU-bound

Debug mode names the offender:

Executing <Task ... handler() at app.py:41>
took 1.003 seconds

The fixes, in order of preference: use the async client, move CPU work to asyncio.to_thread or a process pool, or accept it and run it outside the request path.

digest = await asyncio.to_thread(
    hashlib.pbkdf2_hmac, "sha256", pw, salt, 600_000
)

The task that vanished

asyncio.create_task returns a task. If nothing keeps a reference, the garbage collector can take it mid-flight, and the exception surfaces — if at all — as a log line at an unrelated moment.

# Wrong: the task may be collected before it finishes.
asyncio.create_task(send_email(user))

# Right: something owns it for its whole life.
async with asyncio.TaskGroup() as tg:
    tg.create_task(send_email(user))

TaskGroup (3.11+) is the fix that also handles cancellation: if one child raises, the rest are cancelled and the failures arrive together as an ExceptionGroup.

try:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(a())
        tg.create_task(b())
except* TimeoutError as eg:
    log.warning("timeouts: %d", len(eg.exceptions))
except* ValueError as eg:
    log.error("bad data: %s", eg.exceptions)

except* matters because more than one task can fail. A plain except would show you the first and discard the rest.

Seeing what is actually running

When a service hangs, dump the loop rather than guessing:

for task in asyncio.all_tasks():
    print(task.get_name(), task.get_coro())
    task.print_stack(limit=3)

That output is only readable if tasks have names, so name them at creation:

tg.create_task(sync_orders(), name=f"sync:{tenant}")

An unnamed fleet dumps as Task-1 through Task-400. Wire the dump to a signal handler and you can inspect a stuck process in production without a debugger.

Gotcha: stepping over an await in a debugger hands control to other tasks, so the next line you land on may belong to a different coroutine. Use “run to cursor” rather than repeated step-over when following one flow.