Structured concurrency в asyncio: TaskGroup, отмена задач и graceful shutdown без «осиротевших» корутин

Частая ошибка в asyncio-коде: раскидать asyncio.create_task() по разным углам приложения и надеяться, что при shutdown всё само закроется.

Обычно не закрывается.

Задача живёт отдельно от места, где её запустили. Потеряли ссылку, забыли await, подавили отмену — и на выходе получаем фоновые корутины с открытыми соединениями и Task was destroyed but it is pending!.

Structured concurrency предлагает правило: если конкурентная работа создана внутри scope, она должна завершиться внутри него же.

В Python 3.11 для этого есть asyncio.TaskGroup:

async with asyncio.TaskGroup() as tg:
    tg.create_task(worker("a"))
    tg.create_task(worker("b"))
    tg.create_task(worker("c"))

Вышли из блока — все дочерние задачи завершены. Если одна упала, остальные будут отменены, а наружу прилетит ExceptionGroup.

Главное отличие от хаотичного create_task(): у задач появляется владелец и понятная граница жизни.

Пример graceful shutdown:

import asyncio
import signal

async def worker(name: str):
    try:
        while True:
            print(f"{name}: tick")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print(f"{name}: cleanup")
        # close connections, flush buffers, release locks
        raise  # не проглатываем отмену

async def main():
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()

    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, stop.set)

    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(worker("w1"))
            tg.create_task(worker("w2"))

            await stop.wait()
            raise asyncio.CancelledError

    except asyncio.CancelledError:
        print("shutdown complete")

asyncio.run(main())

Что происходит:

1. Воркеры запускаются внутри TaskGroup.
2. Приложение ждёт сигнал завершения.
3. При shutdown отменяется родительский scope.
4. TaskGroup отменяет дочерние задачи.
5. Каждый worker получает CancelledError, делает cleanup и пробрасывает исключение дальше.
6. asyncio.run() завершает loop без потерянных фоновых задач.

CancelledError — это не просто ошибка, а сигнал управления жизненным циклом задачи. Если поймать его и не сделать raise, можно сломать нормальную отмену.

Плохо:

try:
    await something()
except asyncio.CancelledError:
    log.info("cancelled")
    # забыли raise

Лучше:

try:
    await something()
except asyncio.CancelledError:
    log.info("cancelled")
    raise

Для cleanup используйте try/finally, async context managers или ловите CancelledError, но не подавляйте его без очень веской причины.

TaskGroup также даёт fail-fast поведение. Если одна задача упала, остальные связанные задачи часто не должны продолжать работу на старом состоянии. Например, consumer потерял соединение с брокером — группа отменяется, а родитель решает: перезапустить её или завершить сервис.

Типовая схема:

async def run_service():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(consume_events())
        tg.create_task(flush_metrics())
        tg.create_task(healthcheck_server())

Практические правила:

• Не создавайте «вечные» задачи через create_task() без владельца.
• Используйте TaskGroup для связанных фоновых задач.
• На shutdown отменяйте родительский scope, а не каждую задачу вручную.
• Не проглатывайте CancelledError.
• Освобождайте ресурсы в finally и async context managers.
• Помните про ExceptionGroup: несколько ошибок могут прийти вместе.
• Если задача должна пережить текущий scope, это отдельный lifecycle, а не случайный create_task() в середине функции.