asyncio TaskGroups: Structured Concurrency That Cancels Cleanly Under Load

asyncio.gather(..., return_exceptions=True) plus manual cancels is how orphans and half-closed HTTP sessions accumulate under load. The unfair advantage is asyncio.TaskGroup (3.11+): one failure cancels siblings, exceptions surface as ExceptionGroup, and the scope cannot exit until every child is done.

⚡ TL;DR: Fan-out with async with asyncio.TaskGroup() as tg; never fire-and-forget create_task without ownership; handle ExceptionGroup explicitly; bound concurrency with semaphores inside the group. Pair with uvloop With asyncio and Graceful Shutdown for Node for cross-runtime drain habits.

Replace gather spaghetti

# ❌ Orphans when the first await raises mid-gather without shielding
import asyncio
import aiohttp

async def fetch_all_bad(urls: list[str]) -> list[object]:
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(session.get(u)) for u in urls]
        # if something else raises before await gather, tasks keep running
        return await asyncio.gather(*tasks)

# ✅ TaskGroup owns lifetime
async def fetch_all(urls: list[str], limit: int = 32) -> list[bytes]:
    sem = asyncio.Semaphore(limit)
    results: list[bytes] = [b""] * len(urls)

    async def one(i: int, url: str) -> None:
        async with sem:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
                resp.raise_for_status()
                results[i] = await resp.read()

    async with aiohttp.ClientSession() as session:
        try:
            async with asyncio.TaskGroup() as tg:
                for i, url in enumerate(urls):
                    tg.create_task(one(i, url))
        except* aiohttp.ClientResponseError as eg:
            # ✅ ExceptionGroup via except* — inspect eg.exceptions
            raise RuntimeError(f"{len(eg.exceptions)} upstream failures") from eg
    return results

Cancellation that actually runs finally

When a sibling fails, TaskGroup cancels the rest. Your coroutines must checkpoint on CancelledError and close sockets in finally.

async def write_with_lease(conn, payload: bytes) -> None:
    lease = await conn.acquire()
    try:
        await conn.execute(payload)
    finally:
        # ✅ runs on cancel — return connection to pool
        await conn.release(lease)

Timeouts belong outside the group

async def fanout_with_sla(urls: list[str]) -> list[bytes]:
    async with asyncio.timeout(2.0):  # ✅ whole fan-out budget
        return await fetch_all(urls)

❌ Nesting a 30s timeout per child when the ALB budget is 2s — you invent 500s that look like “random” pool issues (see pool timeouts post).

Closing checklist

✅ Dos
– ✅ Use TaskGroup for any multi-child IO scope
– ✅ Handle ExceptionGroup with except*
– ✅ Put pool/session cleanup in finally
– ✅ Bound concurrency with Semaphore inside the group
– ✅ Apply one outer asyncio.timeout matching the caller SLA

❌ Don’ts
– ❌ Don’t create_task without an owner scope
– ❌ Don’t swallow CancelledError
– ❌ Don’t use gather(return_exceptions=True) to hide cancel bugs
– ❌ Don’t hold lock/lease across an unbounded fan-out

Related reading

• uvloop With asyncio: When the Drop-In Speedup Stops Helping
• Python Pool Timeouts
• Graceful Shutdown for Node on K8s/ECS
• Lambda Recursive Loop Detection

Last updated on September 11, 2026