Структурная конкурентность: TaskGroup, except*, asyncio.timeout и почему голый create_task — утечка

Утечка голого create_task

asyncio.create_task возвращает Task и уходит — и у этой сироты три режима отказа, которые стоит назвать точно. Исчезновение: цикл событий держит на работающие задачи лишь слабую ссылку; если ваш код выбросил возвращённый Task, сборщик мусора может убить его посреди исполнения — документация прямым текстом велит сохранять ссылку. Тихий провал: исключение внутри fire-and-forget-задачи сохраняется на объекте Task и ждёт, что кто-то её await-нет; никто не await-ит, и оно всплывает только при GC задачи строкой Task exception was never retrieved — через минуты или часы, без привязки к запросу, trace ID или алерту. Пережить приложение: на выключении asyncio.run отменяет всё, что ещё работает, — полузаписанные данные, сокеты посреди хендшейка; пропавшие аудит-строки из вступления — ровно это в продакшен-одежде. В литературе по структурной конкурентности у паттерна есть имя — проблема go-statement: любая конструкция, позволяющая работе сбежать из скоупа вызывающего, делает время жизни ресурсов, распространение ошибок и shutdown вопросами без ответа. Структурный ответ: у каждой задачи есть родительский скоуп, который её await-ит, слышит её исключение и отменяет её, когда скоуп умирает.

TaskGroup: скоуп, владеющий своими детьми

Если три режима отказа выше звучат абстрактно, задайте себе конкретный вопрос: когда запись платежа умирает посреди INSERT во время деплоя, кто отвечает за её повтор? С голым create_task ответ — никто. TaskGroup делает ответ структурным.

asyncio.TaskGroup из Python 3.11 — асинхронный контекстный менеджер, владеющий каждой задачей, порождённой через него. В контракте три пункта. Ни один ребёнок не переживает скоуп: выход из блока async with await-ит всех детей — забыть невозможно. Отказ коллективен: когда любой ребёнок кидает (что угодно, кроме asyncio.CancelledError), группа отменяет всех оставшихся детей, дожидается их фактического завершения и затем перебрасывает. Ошибки агрегируются: поскольку во время одной волны отмены могут упасть несколько детей, группа кидает ExceptionGroup со всеми детскими исключениями — не только первым, — и обрабатывается это синтаксисом except*, который сопоставляет и извлекает подгруппы по типу. Заметьте, что except* меняет семантически: для одной ExceptionGroup могут отработать несколько клауз except*, каждая получит подгруппу совпавших исключений, а несовпавшее пропагирует дальше. Сравните с gather из прошлого урока: при отказе gather отдаёт вам первое исключение, пока выжившие соседи продолжают работать без присмотра — с побочными эффектами, открытыми соединениями и без того, кто их await-нет. Отмена-при-отказе TaskGroup — разница между ошибкой и ошибкой плюс утечкой.

import asyncio

async def enrich_order(order_id: int) -> dict:
    try:
        async with asyncio.TaskGroup() as tg:
            user  = tg.create_task(fetch_user(order_id))
            items = tg.create_task(fetch_items(order_id))
            fraud = tg.create_task(fraud_score(order_id))
        # здесь: все три завершились успешно — результаты безопасно читать
        return {"user": user.result(), "items": items.result(),
                "fraud": fraud.result()}
    except* FraudServiceError as eg:        # подгруппа совпавших ошибок
        for exc in eg.exceptions:
            log.warning("fraud check failed: %s", exc)
        return await enrich_without_fraud(order_id)
    # любой другой ExceptionGroup передаётся вызывающему нетронутым

Таймауты и отмена: один механизм под капотом

asyncio.timeout (3.11+) заменил старый wait_for композируемым контекстным менеджером, и понять его — значит понять саму отмену. Отмена в asyncio — не рубильник: task.cancel() организует возбуждение CancelledError внутри задачи в её следующей точке await — задача, которая не await-ит, неотменяема (она заодно блокирует цикл, так что у неё проблемы посерьёзнее). Дальше исключение раскручивает стек корутины как любое другое, по пути исполняя finally-блоки и выходы async with. asyncio.timeout(5) строится поверх: он ставит таймер; если тело ещё работает, когда таймер сработал, таймер отменяет тело, а контекстный менеджер — на собственной границе — конвертирует этот CancelledError в TimeoutError. Дизайн «конверсия на краю» и делает вложенные таймауты композируемыми: внутренний таймаут конвертирует только отмены, которые вызвал сам; внешняя отмена проходит сквозь нетронутой. Доверие к этой машинерии держится на двух дисциплинах. Никогда не глотать CancelledError: голый except Exception безопасен (CancelledError наследуется от BaseException ровно затем, чтобы ускользать из этой сети), но except BaseException: pass или except asyncio.CancelledError: continue ломает таймауты, отмену TaskGroup и shutdown сервера одним махом — если перехватили ради очистки, перебросьте. Очистка должна быть устойчива к отмене: код в finally исполняется, пока задачу отменяют, и если очистка сама await-ит (закрытие async-соединения, откат транзакции), она может получить вторую отмену. Для очистки, которая обязана завершиться, в 3.11+ есть паттерн экранирования одного критичного await.

async def transfer(src, dst, amount):
    async with asyncio.timeout(5):                 # cancels body at T+5s
        await debit(src, amount)
        # коммит не должен быть разорван таймаутом/отменой на полпути:
        await asyncio.shield(credit_and_commit(dst, amount))
    # таймаут сработал? CancelledError стал TimeoutError ЗДЕСЬ, на границе

async def handler():
    try:
        await transfer("a", "b", 100)
    except TimeoutError:
        ...  # the transfer either fully committed (shield) or fully debit-rolled-back

Миграция fire-and-forget

У аудит-бага из вступления есть структурное переписывание с явными компромиссами. Встроить — tg.create_task(audit.write(event)) внутри TaskGroup запроса — и запись под присмотром, с ретраями и на виду, ценой присоединения её латентности к запросу (часто это нормально: INSERT — единицы миллисекунд). Если латентность правда нельзя тратить, структурный fire-and-forget — это долгоживущая воркер-задача, принадлежащая скоупу времени жизни приложения и питающаяся из очереди: хендлеры делают queue.put_nowait(event) (микросекунды), воркер разгребает её внутри собственного TaskGroup, а shutdown закрывает скоуп, который опустошает очередь до смерти цикла. В обоих случаях у каждой задачи есть владелец, у каждого исключения — адрес, а shutdown — свойство структуры, а не молитва. Что ревью не переживает: create_task с комментарием, обещающим, что кто-нибудь когда-нибудь это await-нет. Теперь, когда видите голый create_task в PR, задаёте три вопроса: кто держит ссылку, кто видит исключение, кто гарантирует слив при shutdown — и не аппрувите, пока все три не закрыты.