Дело не в нагрузке: почему пул httpx теряет соединения при отменах

TL;DR. В async-сервисе на httpx пул иногда теряет слоты. Появляется PoolTimeout, хотя свободные соединения есть. Метрика busy растёт при ровном RPS. Причина в отмене запроса в момент, когда закрывается уже прочитанный ответ. httpcore оставляет соединение в ACTIVE и не возвращает его в пул. Ниже — как это увидеть, воспроизвести и исправить обёрткой транспорта.

Почему мы этим занялись

У нас сервис-агрегатор. Один входящий запрос превращается в 20–30 исходящих в разные сервисы, и почти вся эта связь идёт через httpx. Заменить его быстро не выйдет, а развивается он медленно: стабильного релиза не было почти два года. Поэтому решили не ждать исправления, а разобраться, как защититься на своей стороне.

При высоком RPS иногда появлялся PoolTimeout. Апстримы отвечали нормально, нагрузка не менялась, а занятость пула росла. Эти факты не складывались в общую картину: при ровной нагрузке и здоровых апстримах пул не должен упираться в лимит.

Некоторое время мы перезапускали поды. Это возвращало пул в норму, но ничего не объясняло и только откладывало проблему. Нужно было понять, что именно остаётся занятым. Оказалось, что соединение не возвращается в пул, если запрос отменили в момент закрытия ответа.

Откуда берутся отмены

Отмена запроса здесь — обычное дело. Вызывающая сторона не обязана ждать ответ до конца. Вот несколько примеров.

• Пользователь ушёл со страницы, и соединение закрылось.

• Мобильный клиент не готов ждать дольше N секунд.

• Сработал серверный таймаут RESPONSE_TIMEOUT.

• Одна ветка параллельных запросов упала, и остальные уже не нужны.

• Пользователь повторил запрос или переключил вкладку.

Но вопрос не в том, откуда пришла отмена, а в том, переживает ли её клиентская библиотека.

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

В пуле соединение либо свободно, либо занято. Свободное можно взять под новый запрос, занятое — нет. Метрика busy считает занятыми все соединения, кроме свободных. Туда попадают и закрытые, которые ещё не убрали из пула. Поэтому busy растёт и при ровном RPS: метрика не отличает работающие соединения от закрытых.

Главная причина потери слота — отмена в момент закрытия ответа. Ответ уже прочитан, обмен завершён: h11 в состоянии our=DONE, their=DONE. Это ключевой признак: обе стороны отработали без ошибок, соединение исправно. httpx начинает закрывать поток, и ровно здесь приходит отмена. Шаг, который возвращает соединение в пул, не выполняется. Повторно его никто не запускает: ответ уже помечен закрытым. Соединение остаётся занятым до конца жизни пода. Вот этот момент в коде:

# httpx/_models.py
async def aclose(self) -> None:
    if not self.is_closed:
        self.is_closed = True            # флаг ставится ДО await
        await self.stream.aclose()       # ← CancelledError

# httpcore/_async/http11.py
async def _response_closed(self) -> None:
    async with self._state_lock:         # ← единственный await
        if our_state is DONE and their_state is DONE:
            self._state = IDLE           # сюда уже не доходим

Вторая причина — отмена до установления соединения. Пул занимает слот под запрос ещё до подключения. Если запрос отменить, в пуле остаётся заготовка соединения. Штатная чистка пула её не трогает. Слот освобождается только после таймаута подключения, когда заготовка уходит по возрасту.

Как увидеть

Публичного API нет. Состояние можно прочитать из приватных атрибутов.

pool = client._transport._pool

for conn in pool.connections:
    state = "CONNECTING" if conn._connection is None else conn._connection._state.name
    print(state, conn.info(), conn.is_available())

Если транспорт обёрнут инструментированием, пул лежит во вложенном _transport. До него достаточно добраться один раз и запомнить ссылку.

Состояния нужно снять дважды с небольшим интервалом. Если соединение стоит в ACTIVE, с обеих сторон DONE, а счётчик запросов не изменился, это отмена на закрытии. В логе такое соединение выглядит так:

state=ACTIVE  our=DONE  their=DONE  Request Count: 727  available=False

Воспроизведение

Искусственно расширим окно закрытия, чтобы не ловить микросекунду.

import asyncio, httpx
from httpcore._async import http11 as h11

async def handler(reader, writer):
    while True:
        await reader.readuntil(b"\r\n\r\n")
        writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
        await writer.drain()

async def main():
    server = await asyncio.start_server(handler, "127.0.0.1", 0)
    port = server.sockets[0].getsockname()[1]

    entered = asyncio.Event()
    orig = h11.AsyncHTTP11Connection._response_closed

    async def slow(self):
        entered.set()
        await asyncio.sleep(0.05)
        await orig(self)

    h11.AsyncHTTP11Connection._response_closed = slow

    async with httpx.AsyncClient() as client:
        task = asyncio.create_task(client.get(f"http://127.0.0.1:{port}/"))
        await entered.wait()
        task.cancel()
        try:
            await task
        except BaseException:
            pass

        conn = client._transport._pool.connections[0]
        print(conn._connection._state.name, conn._h11_state.our_state,
              conn._h11_state.their_state, conn.is_available())

asyncio.run(main())

Вывод: ACTIVE DONE DONE False. Слот потерян, хотя ответ уже прочитан полностью.

Решение: обёртка транспорта

Патчить httpcore не обязательно. Можно обернуть транспорт клиента и не давать отмене прервать закрытие. Закрытие выполняется отдельной задачей. Вызывающий при этом получает отмену как обычно.

import asyncio
import httpx


def spawn(coro, tasks: set[asyncio.Task]) -> asyncio.Task:
    # Закрытие уходит в отдельную задачу, чтобы отмена вызывающего его не прервала.
    # Ссылку храним в наборе: иначе сборщик может убрать задачу до завершения.
    task = asyncio.get_running_loop().create_task(coro)
    tasks.add(task)
    task.add_done_callback(tasks.discard)
    return task


class ShieldedResponseStream(httpx.AsyncByteStream):
    def __init__(self, stream: httpx.AsyncByteStream, tasks: set[asyncio.Task]) -> None:
        self._stream = stream
        self._tasks = tasks

    async def aclose(self) -> None:
        # shield не даёт отменить задачу закрытия: соединение вернётся в пул,
        # а вызывающий получит свою отмену как обычно.
        await asyncio.shield(spawn(self._stream.aclose(), self._tasks))

    async def __aiter__(self):
        # Чтение тела не меняем, обёртка нужна только ради закрытия.
        async for chunk in self._stream:
            yield chunk


class ShieldedTransport(httpx.AsyncBaseTransport):
    def __init__(self, transport: httpx.AsyncBaseTransport, tasks: set[asyncio.Task]) -> None:
        self._transport = transport
        self._tasks = tasks

    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        response = await self._transport.handle_async_request(request)
        # Подменяем поток ответа на обёртку, которая защищает закрытие.
        if isinstance(response.stream, httpx.AsyncByteStream):
            response.stream = ShieldedResponseStream(response.stream, self._tasks)
        return response

    async def aclose(self) -> None:
        # Сначала дожидаемся задач закрытия, потом закрываем транспорт:
        # иначе при остановке часть закрытий потеряется.
        while self._tasks:
            await asyncio.gather(*list(self._tasks), return_exceptions=True)
        await self._transport.aclose()

Установка — одна строка после создания клиента.

client = httpx.AsyncClient(...)
client._transport = ShieldedTransport(client._transport, set())

Почему это работает.

• asyncio.shield не отменяет задачу закрытия. task.cancel() отменяет только ожидание вызывающего, а соединение возвращается в пул.

• Ссылки на задачи хранятся в наборе. Иначе сборщик может убрать задачу до завершения.

• При close() мы дожидаемся задач, чтобы не осталось висящих закрытий.

• anyio.CancelScope(shield=True) здесь не помогает. task.cancel() — это нативная asyncio-отмена, и anyio-shield её не перехватывает.

Ограничения.

• Нельзя дозакрывать вслепую. Если соединение уже вернулось в пул и его взял другой запрос, повторный _response_closed закроет здоровый сокет. Поэтому обёртка защищает закрытие от отмены, а не дозакрывает соединение по факту состояния.

• Для h2 нужна отдельная реализация. Там сначала удаляется стрим из _events, и только потом берётся лок. Повторный _response_closed для удалённого стрима упадёт.

До этого у нас работал фоновый обход пулов. Он снимал CLOSED, брошенные CONNECTING и залипшие ACTIVE по возрасту и счётчику запросов. Это компенсация. Она возвращает слоты, но не устраняет причину. После обёртки такой обход больше не нужен.

Итог

Отмены неизбежны, поэтому закрытие ответа нужно защищать. Обёртка транспорта делает это без форка библиотеки, и профилактические перезапуски больше не нужны.

Дефект и близкие к нему случаи описаны в issue httpx/httpcore (encode/httpcore#1116, encode/httpcore#658, encode/httpx#3782), там же — примеры кода, которые его воспроизводят. Ждать исправления придётся долго: стабильного релиза httpx не было почти два года (0.28.1 вышел в декабре 2024), последний релиз httpcore 1.0.9 — апрель 2025. Пока фикса нет, можно защититься самостоятельно: пример обёртки из этой статьи снимает основную причину, и PoolTimeout из-за отмен больше не появляется.