If you've been using asyncio for a while, you've probably leaned heavily on asyncio.gather(). It feels intuitive: you throw a bunch of coroutines into a list, await the result, and you're done. I used to do the same thing until I spent a miserable weekend debugging a production leak where "ghost tasks" were eating up memory and hammering a database long after the original request had timed out.

The Myth that gather() handles task lifecycles

The common misconception is that asyncio.gather() provides a structured way to manage a group of tasks. People assume that if one task in the group fails, the rest are cleaned up. They aren't. gather() is essentially a "fire and hope" mechanism. If one task raises an exception, gather() will immediately propagate that exception to you, but the other tasks keep running in the background. They become orphans.

import asyncio

async def flaky_service():
    await asyncio.sleep(1)
    print("Service A failed!")
    raise RuntimeError("Connection lost")

async def slow_service():
    try:
        await asyncio.sleep(10)
        print("Service B finished")
    except asyncio.CancelledError:
        print("Service B was cancelled")

async def main():
    try:
        # This is the dangerous part
        await asyncio.gather(flaky_service(), slow_service())
    except RuntimeError:
        print("Caught the error from Service A")

asyncio.run(main())
# Notice that "Service B was cancelled" is NEVER printed. 
# Service B is still running in the background for 10 seconds 
# even though the main function has already moved on.

In a real application, this is a nightmare. You end up with "zombie" tasks that hold onto sockets or file handles, and you have no easy way to track them down because the gather() call has already returned.

Structured Concurrency with TaskGroups

This is why Python 3.11 introduced asyncio.TaskGroup. It implements "structured concurrency," a concept borrowed from languages like Go and libraries like Trio. The core idea is simple: if you open a group of tasks, you cannot leave that block until every task in that group has finished—either by completing successfully or by being cancelled.

When you use a TaskGroup, if any task within the group raises an exception, all other remaining tasks in the group are automatically cancelled. No orphans, no zombies.

import asyncio

async def flaky_service():
    await asyncio.sleep(1)
    print("Service A failed!")
    raise RuntimeError("Connection lost")

async def slow_service():
    try:
        await asyncio.sleep(10)
        print("Service B finished")
    except asyncio.CancelledError:
        print("Service B was cancelled!") # This WILL now trigger

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(flaky_service())
            tg.create_task(slow_service())
    except ExceptionGroup as eg:
        print(f"Caught grouped exceptions: {eg}")

asyncio.run(main())

Notice two major changes here. First, we use async with. The block acts as a boundary; the code will not proceed past the end of that indentation until the tasks are resolved. Second, we catch an ExceptionGroup. Because multiple tasks could potentially fail simultaneously, Python doesn't just throw one exception; it wraps all failures into a group.

I strongly recommend moving toward TaskGroup for almost everything. Use gather() only if you specifically need the results returned as a sorted list and you are absolutely certain that your tasks are independent and "leak-proof." For everything else, the safety guardrails of structured concurrency are worth the slightly more verbose syntax.

📋 Practical Task

Build a Fail-Fast Multi-Endpoint Health Monitor

You need to build a health check system that monitors three different microservices. If any critical service fails, the monitor should immediately stop checking the other services to save resources and report the failure.

Requirements:

• Create three coroutines: check_db(), check_cache(), and check_api().
• check_db() should simulate a failure by raising a ConnectionError after 0.5 seconds.
• check_cache() and check_api() should simulate longer checks (1.5 and 2.0 seconds respectively) using asyncio.sleep().
• Use an asyncio.TaskGroup to launch all three checks concurrently.
• Wrap the TaskGroup in a try/except block that catches ExceptionGroup.
• Ensure that when check_db() fails, the other two tasks are cancelled immediately (you can verify this by adding a try/except asyncio.CancelledError block inside the cache and api functions to print a "Cancelled" message).