Нужно обеспечить безопасную отмену и таймаут для цепочки из нескольких зависимых асинхронных задач, запущенных через asyncio.gather, где одна из корутин может зависнуть на внешнем вызове без собственного таймаута, и при этом не допустить утечки ресурсов у отменённых и ещё выполняющихся задач. Как спроектировать такое решение? Обсудите компромиссы, риски и критерии выбора — единственного правильного решения может не быть.

Короткий ответ

Нужно обернуть всю группу в общий таймаут (asyncio.wait_for вокруг gather) и дать каждой отдельной корутине собственный внутренний таймаут и try/finally для освобождения ресурсов при CancelledError, потому что внешняя отмена gather сама по себе отменяет входящие задачи, но не гарантирует, что их код корректно освободит ресурсы, если это не предусмотрено явно внутри каждой корутины.

Как это работает подробнее

asyncio.wait_for(asyncio.gather(...), timeout=N) при превышении N отменяет все задачи, переданные в gather; если конкретная корутина выполняет внешний вызов без собственного таймаута и без await внутри самого вызова, отмена может не сработать вовремя, и ресурс останется удерживаемым дольше ожидаемого, даже после формального завершения wait_for с исключением таймаута.

• Компромисс. Выставление собственных таймаутов на каждый внешний вызов повышает надёжность и предсказуемость освобождения ресурсов, но требует больше кода и более тщательного проектирования каждой корутины; опора только на один общий внешний таймаут проще в реализации, но создаёт риск «зависших» ресурсов при внешних вызовах без встроенной поддержки отмены — выбор зависит от критичности ресурсов и от того, насколько библиотеки внешних вызовов поддерживают корректную реакцию на CancelledError.

Пример кода

example-289.pyКопировать

import asyncio
async def step_with_resource(name, delay, external_has_no_timeout=False):
    conn = await acquire_connection()  # условный ресурс
    try:
        if external_has_no_timeout:
            await asyncio.sleep(delay)  # имитация внешнего вызова без собственного таймаута
        else:
            await asyncio.wait_for(asyncio.sleep(delay), timeout=2)  # собственный таймаут шага
        return name
    finally:
        await conn.close()  # гарантированно освобождаем ресурс даже при CancelledError
async def main():
    try:
        results = await asyncio.wait_for(
            asyncio.gather(step_with_resource("A", 1), step_with_resource("B", 10)),
            timeout=3,
        )
    except asyncio.TimeoutError:
        print("цепочка не завершилась за 3с, задачи отменены, ресурсы освобождены в finally")
asyncio.run(main())

Что использовать на практике

• Практика. Каждую отдельную корутину снабжают собственным разумным таймаутом на её конкретный внешний вызов, и оборачивают открытие ресурсов в async with или try/finally, чтобы закрытие происходило и при нормальном завершении, и при отмене. Для зависимых задач явно продумывают, что делать с уже выполненными промежуточными шагами при отмене на середине цепочки — откатывать их эффект или оставлять как есть, в зависимости от идемпотентности операций.

Подводные камни

• Риски. Полагаются только на внешний wait_for/gather таймаут, не добавляя таймаутов и try/finally на уровне отдельных вызовов; не учитывают, что отмена одной задачи в gather без return_exceptions=True немедленно поднимает исключение вызывающему коду, в то время как остальные задачи всё ещё продолжают выполняться в фоне и тоже должны быть корректно отменены.