Утечки соединений в асинхронном FastAPI-сервисе: разбор инцидента

Симптомы: пул исчерпан, но запросов немного

Типичная картина: сервис на FastAPI с SQLAlchemy async и PostgreSQL работает стабильно при деплое, но через 4–8 часов начинает отвечать таймаутами. В логах — TimeoutError: QueuePool limit of N overflow M reached. Метрики показывают, что активных запросов к API в этот момент меньше половины лимита пула. Соединения заняты, но никто ими не пользуется. Это утечка.

Механизм: как соединение «теряется»

В синхронном коде утечка соединения — редкость: сессия привязана к потоку, а поток завершает работу предсказуемо. В асинхронном — всё сложнее. Корутина может быть отменена в любой точке await, и если соединение было взято из пула до момента отмены, блок finally или контекстный менеджер могут не отработать так, как вы ожидаете.

Три главных сценария утечки

1. Сессия без контекстного менеджера. Вызов session = AsyncSession(engine) напрямую, без async with. Если в обработчике возникает исключение до await session.close(), соединение остаётся занятым до сборки мусора — а в CPython с asyncio момент сборки непредсказуем.
2. Background task с собственным обращением к базе. FastAPI позволяет добавить фоновую задачу через BackgroundTasks. Если она использует ту же сессию, что и обработчик, сессия может быть закрыта dependency-инъекцией раньше, чем задача завершится. Если задача создаёт свою сессию без async with — см. пункт 1.
3. Отмена запроса клиентом (disconnect). Клиент закрывает соединение, ASGI-сервер (Uvicorn) отменяет корутину через CancelledError. SQLAlchemy async session при этом может не вернуть соединение в пул, если транзакция была начата и rollback не вызван явно.

Утечка одного соединения в час при пуле в 20 штук — это 20 часов до полной остановки сервиса. Достаточно, чтобы пройти все тесты и сломаться только в проде.

Как найти утечку

Прежде чем чинить — убедитесь, что это именно утечка, а не медленные запросы, держащие соединение. Два инструмента:

Мониторинг пула

SQLAlchemy предоставляет события пула. Подпишитесь на checkout и checkin и логируйте разницу. Если checkout растёт, а checkin отстаёт — соединения не возвращаются.

from sqlalchemy import event
from sqlalchemy.pool import Pool
import logging

logger = logging.getLogger("pool_monitor")

# Счётчик для наблюдения за балансом checkout/checkin
_checked_out = 0

@event.listens_for(Pool, "checkout")
def on_checkout(dbapi_conn, connection_record, connection_proxy):
    global _checked_out
    _checked_out += 1
    logger.info("Pool checkout, active: %d", _checked_out)

@event.listens_for(Pool, "checkin")
def on_checkin(dbapi_conn, connection_record):
    global _checked_out
    _checked_out -= 1
    logger.info("Pool checkin, active: %d", _checked_out)

Код выше работает с SQLAlchemy 2.x. Глобальный счётчик — для отладки; в проде отправляйте значение в Prometheus или аналог.

Проверка со стороны PostgreSQL

Запрос к pg_stat_activity покажет, сколько соединений реально открыто и в каком состоянии:

SELECT state, count(*)
FROM pg_stat_activity
WHERE datname = 'your_db'
GROUP BY state;

Если видите десятки строк со state = 'idle' — соединения открыты, но ничего не делают. Это и есть утёкшие сессии, которые пул считает занятыми.

Как закрыть утечку

Правило: сессия только через async with

Единственный надёжный паттерн для FastAPI-зависимости:

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

# async_session_factory создаётся при старте приложения

async def get_session():
    async with async_session_factory() as session:
        async with session.begin():
            yield session
    # Здесь соединение гарантированно возвращено в пул,
    # даже если обработчик упал с исключением

Двойной контекстный менеджер — не перестраховка. Внешний (async with ... as session) гарантирует закрытие сессии. Внутренний (session.begin()) гарантирует откат транзакции при ошибке. Без внутреннего — соединение возвращается в пул с незавершённой транзакцией, и PostgreSQL продолжает держать блокировки.

Background tasks: изоляция сессии

Фоновая задача должна создавать собственную сессию, а не получать её из dependency injection обработчика. Зависимость обработчика живёт до конца response, а фоновая задача выполняется после.

Отмена запроса: обработка CancelledError

Если ваш код выполняет длинные операции с базой, оберните критический участок так, чтобы CancelledError не прервал его без отката:

import asyncio

async def long_db_operation(session: AsyncSession):
    try:
        await session.execute(...)  # длинная операция
        await session.commit()
    except asyncio.CancelledError:
        await session.rollback()
        raise  # важно: не глотайте CancelledError

Граничные случаи и где подход ломается

Сноска: QueuePool — стандартная стратегия пула соединений в SQLAlchemy. Она держит фиксированное число соединений (pool_size) и допускает временное превышение (max_overflow). Когда все слоты заняты, новый запрос ждёт освобождения в течение pool_timeout секунд, после чего падает с TimeoutError.

Что делать дальше

После устранения утечки настройте два рубежа защиты. Первый — параметр pool_pre_ping=True в create_async_engine: он проверяет живость соединения перед выдачей из пула и отсекает «мёртвые» соединения, закрытые PostgreSQL по idle_in_transaction_session_timeout. Второй — алерт на метрику checked_out_connections: если она растёт монотонно в течение часа, значит, утечка вернулась. Нагрузочное тестирование длительностью от получаса после каждого релиза — единственный способ поймать проблему до прода.