Написать HTTP‑клиент к другому сервису — обычная задача в бэкенде. В первой версии всё просто: отправить запрос, прочитать ответ, вернуть результат. Затем приходится решать, сколько ждать медленный сервис, какие ошибки повторять и сколько запросов разрешать одновременно.

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

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

Реализуем клиент на Python с aiohttp. Внутри retry_cycle() он делает HTTP‑запросы и при определённых ошибках повторяет их. Снаружи оставим ограничение конкурентности и общий таймаут:

async with semaphore:
    deadline = loop.time() + 1.0
    async with asyncio.timeout_at(deadline):
        return await retry_cycle()

Выглядит разумно: сначала получаем разрешение на работу, затем выполняем её с ограничением по времени. Только вызывающий код ждёт всё это время, включая очередь перед семафором. Попадает ли эта очередь в нашу секунду?

Чтобы проверить, я собрал локальный стенд: клиент обращается к настоящему aiohttp‑серверу, а задержки и занятые ресурсы задаются заранее. В первом сценарии другая задача держит единственный слот семафора 1100 мс от начала вызова. Сам сервер отвечает за 20 мс, повторы не нужны.

Результат: все 90 операций завершились успешно, но позже секунды. Медиана — 1124 мс. Таймаут при этом ни разу не сработал.

Если пройти по коду, причина видна. Пока операция ждёт на async with semaphore, до вычисления deadline она ещё не дошла. Слот освободился — только теперь начинаем отсчитывать секунду. Ответ приходит примерно за 24 мс, и внутренний таймер вполне доволен. Вызывающий код к этому моменту ждёт уже больше секунды.

Код и данные эксперимента лежат в andy‑takker/http‑deadline. Здесь и дальше числа относятся к этому стенду: на другой машине миллисекунды будут отличаться. Проверяем мы то, какие ожидания вообще входят в бюджет операции.

Чью секунду мы считаем?

Значит, сначала надо договориться, откуда считать время. Для вызывающего кода операция началась в момент вызова — назову его t0. Дальше ему всё равно, чем занят клиент: ждёт слот, отправляет запрос или делает паузу перед повтором. Всё это расходует один бюджет в 1000 мс. Deadline в таком случае — конкретный момент t0 + 1000 мс, в который пора прекращать ожидание.

Поэтому вычисляем deadline до очереди и переносим ожидание семафора внутрь timeout_at:

deadline = loop.time() + 1.0
async with asyncio.timeout_at(deadline):
    async with semaphore:
        return await retry_cycle()

Повторяем тот же сценарий с занятым семафором. Теперь все 90 операций заканчиваются по deadline, медиана — 1001 мс. Ни одного HTTP‑запроса на сервере: asyncio.timeout_at отменяет ожидание внутри Semaphore.acquire(). В минимальном примере наружу выходит TimeoutError; в результатах стенда этот исход записан как deadline_exceeded.

По одной операции, ближайшей к медиане своей серии; N = 90 на вариант.

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

Здесь подошёл бы и asyncio.timeout(1.0) вокруг семафора и цикла попыток. Он сам вычисляет срок как loop.time() + delay; timeout_at принимает уже вычисленный момент. Последний удобен, когда одна граница передаётся дальше по локальной цепочке вызовов. Механизм отмены у них общий.

Почему таймаута на попытку недостаточно?

С очередью разобрались. Но в первом примере сервер отвечал быстро, поэтому до повторов дело вообще не дошло. Теперь освободим семафор и задержим ответ сервера: проверим, что происходит внутри retry_cycle().

В aiohttp зададим ClientTimeout(total=0.4) — 400 мс на одну попытку. На уровне HTTP‑вызова это выглядит так; session — уже открытая сессия, url — адрес локального сервера:

import aiohttp

timeout = aiohttp.ClientTimeout(
    total=0.4, 
    connect=0.3, 
    sock_connect=0.3, 
    sock_read=None,
)

async with session.get(url, timeout=timeout) as response:
    body = await response.read()

Этот фрагмент показывает одну попытку. Ниже добавим к ней цикл повторов, общий deadline и измерение времени, а в конце соберём из этих функций клиент каталога. В ClientTimeout время задаётся в секундах, в YAML стенда — в миллисекундах.

Сервер в сценарии slow_headers начинает отдавать заголовки только через 800 мс. Дождаться их клиент не успеет, получит таймаут и повторит запрос. Число попыток и паузы задаются отдельно от aiohttp — в experiments/slow_headers.yaml:

operation:
  budget_ms: 1000
  max_attempts: 3
  backoff_ms: [100, 200]
  jitter: false
  retryable_statuses: [503]
  retryable_failures: [attempt_timeout]

scenario:
  type: slow_headers
  response_delay_ms: 800

Это фрагмент конфигурации; полный сценарий запускается через make demo SCENARIO=slow_headers. Для трёх попыток и двух пауз суммарное ожидание составит:

400 + 100 + 400 + 200 + 400 = 1500 мс

Это уже полторы секунды. Настройка total=0.4 работает, просто при каждой новой попытке отсчёт начинается заново.

По одной операции, ближайшей к медиане своей серии; N = 90 на вариант.

Измеренная медиана у варианта с таймаутом на попытку — 1506 мс, разброс медиан между тремя повторами 1506–1507. Каждая попытка завершилась по своему таймауту, а вся операция заняла полторы секунды.

Теперь ограничим весь цикл общим бюджетом в секунду. Он остановится во второй паузе, перед третьей попыткой: медиана — 1001 мс, на сервер успели уйти два запроса. Третьего нет ни в клиентском, ни в серверном журнале. Пауза между попытками тоже находится внутри timeout_at, поэтому ждать её до конца не приходится.

При свободном семафоре оба варианта общего бюджета дали медиану 1001 мс и две попытки: цикл и паузы находятся внутри таймаута.

Что находится внутри retry_cycle?

До сих пор цикл повторов был спрятан за одним вызовом. Раскроем его: иначе непонятно, кто ловит таймаут попытки и почему общий deadline не запускает очередной повтор.

Для начала выделим одну HTTP‑попытку. Она получает статус, дочитывает тело и закрывает response. Здесь session и timeout передаются явно:

async def fetch_once(
    session: aiohttp.ClientSession,
    url: str,
    timeout: aiohttp.ClientTimeout,
) -> tuple[int, bytes]:
    async with session.get(url, timeout=timeout, allow_redirects=False) as response:
        status = response.status
        body = await response.read()
    return status, body

Теперь сам цикл. Это сокращённый пример логики стенда: без журнала событий, ограничения размера тела и проверки маркера в ответе. Полная реализация — в ExecuteOperation. Для примера считаем успешным любой ответ 2xx с полностью прочитанным телом.

import asyncio
from collections.abc import Awaitable, Callable


class OperationDeadlineExceeded(TimeoutError):
    pass


class RetriesExhausted(RuntimeError):
    pass


class UnexpectedStatus(RuntimeError):
    pass


async def retry_cycle(
    fetch: Callable[[], Awaitable[tuple[int, bytes]]],
    deadline: float,
) -> bytes:
    loop = asyncio.get_running_loop()

    for pause_s in (0.0, 0.1, 0.2):  # первая попытка сразу, затем две с паузами
        if pause_s:
            await asyncio.sleep(pause_s)
        if loop.time() >= deadline:
            raise OperationDeadlineExceeded("Бюджет операции исчерпан")

        try:
            status, body = await fetch()
        except aiohttp.ConnectionTimeoutError:
            raise  # на стенде ошибка соединения не входит в причины повтора
        except TimeoutError:
            continue  # таймаут одной попытки

        if 200 <= status < 300:
            if loop.time() >= deadline:
                raise OperationDeadlineExceeded("Ответ пришёл после deadline")
            return body
        if status != 503:
            raise UnexpectedStatus(f"Неповторяемый HTTP-статус: {status}")

    raise RetriesExhausted("Три попытки не дали успешного ответа")

Цикл запускается внутри общего asyncio.timeout_at(deadline), включая sleep между попытками. Сам по себе параметр deadline не прервёт зависший fetch: это делает внешний контекст. Явные проверки нужны, чтобы не начать новую попытку или не вернуть опоздавший результат, если отмена ещё не доставлена.

У fetch здесь узкий контракт: TimeoutError приходит от таймера HTTP‑попытки. Если добавить внутрь разбор данных или собственные таймеры, их ошибки нужно классифицировать отдельно. ConnectionTimeoutError тоже наследуется от TimeoutError, поэтому его обработчик стоит первым. Внешний deadline доставляет в цикл CancelledError; этот код его не ловит, и повтор не начинается. После выхода из внешнего контекста его собственная отмена превратится в TimeoutError.

В примерах выше retry_cycle() без аргументов обозначал подготовленный вызов. Связать функции можно локальной обёрткой; ниже она же пригодится для метрик:

async def fetch() -> tuple[int, bytes]:
    return await fetch_once(session, url, timeout)

Почему семафор занят даже во время паузы?

Семафор окружает весь retry_cycle: слот получает целая операция обращения к каталогу и удерживает его во время пауз. Мы ограничиваем число допущенных операций, но часть слотов может быть занята клиентами, которые сейчас спят.

Если нужна граница именно на число одновременных HTTP‑попыток, семафор можно поставить вокруг fetch_once, освобождая его перед паузой. Тогда во время backoff слот сможет занять другая операция, а повтору придётся снова встать в очередь. Общий deadline всё равно вычисляется один раз до первой очереди и охватывает весь цикл.

На стенде оставляю первый вариант: так сравнивается положение таймаута при неизменной границе семафора. Для своего клиента этот выбор стоит сделать явно. И в обоих случаях семафор сам по себе не ограничивает число задач, которые ждут допуска: управление входящей нагрузкой остаётся отдельной задачей.

А если сервер отвечает ошибкой быстрее таймаута?

В предыдущем опыте каждая попытка ждала до своего таймаута. Но сервер может быстро ответить 503, и тогда за секунду клиент успеет сделать все разрешённые повторы. Проверим и этот случай: сначала сервер ошибается один раз, затем — при каждом запросе. Каждый ответ занимает 50 мс, на вариант приходится 90 операций.

Настройки клиента и повторов оставляем прежними. Меняется поведение сервера в двух файлах сценариев:

# experiments/transient_error.yaml
scenario:
  type: transient_error
  steps:
    - {status: 503, delay_ms: 50}
    - {status: 200, delay_ms: 50}
---
# experiments/persistent_error.yaml
scenario:
  type: persistent_error
  status: 503
  response_delay_ms: 50

Запуск сценариев: make demo SCENARIO=transient_error или make demo SCENARIO=persistent_error.

Диапазоны в таблице — округлённые медианы разных политик; исходы у них совпали. При временной ошибке две попытки с паузой (50 + 100 + 50 мс) укладываются в бюджет. При постоянной ошибке повторы утроили число обращений и не дали ни одного успешного результата, потратив почти в девять раз больше времени. Общий deadline этого не предотвратил: во всех измеренных операциях три попытки завершились раньше секунды, и он не сработал ни разу.

Число попыток ограничивается отдельно, в retry‑политике. Для ограничения дополнительного трафика от повторов используют retry budget, а для прекращения обращений к деградирующей зависимости — circuit breaker; про то, как повторы усиливают нагрузку под настоящим трафиком, лучше читать разбор от Яндекса, этот стенд такого не моделирует.

Для сравнения оставил и здоровую зависимость с ответом за 20 мс: все четыре варианта дали медиану 21 мс после округления и 90 из 90 успешных операций в срок. Это контрольный сценарий; измерять накладные расходы самого timeout_at он не позволяет.

Какие ожидания уже внутри HTTP‑вызова?

До сих пор мы смотрели на обвязку вокруг HTTP‑запроса: семафор, цикл повторов и паузы. Теперь спустимся внутрь одной попытки. Там тоже есть ожидания, причём отправка запроса может начаться далеко не сразу.

Клиенту сначала нужно получить соединение из пула — это происходит внутри session.get(). А после получения заголовков ещё нужно дочитать тело через await response.read(). На стенде оба действия входят в попытку и находятся под общим deadline:

async with asyncio.timeout_at(deadline):
    async with session.get(url) as response:
        body = await response.read()

У aiohttp есть собственные ограничения на эти этапы. По документации, total охватывает всю попытку, включая пул и тело; connect — получение соединения, включая очередь в пуле; sock_connect — само TCP‑подключение; sock_read — ожидание очередной порции данных.

Поэтому здесь важно не приписать внешнему deadline то, что уже умеет библиотека. В следующих двух опытах я намеренно выставляю таймауты aiohttp так, чтобы они позволяли ждать дольше общего бюджета. Это даст увидеть именно внешнюю границу.

Очередь за соединением

Начнём с пула из одного соединения. Для всех сравниваемых вариантов выставим total и connect по 2000 мс, а общий бюджет оставим прежним — 1000 мс.

Размер пула задаётся при создании сессии, таймаут передаётся в каждый запрос:

connector = aiohttp.TCPConnector(limit=1, limit_per_host=1)
timeout = aiohttp.ClientTimeout(
    total=2.0, 
    connect=2.0, 
    sock_connect=0.3, 
    sock_read=None,
)

async with aiohttp.ClientSession(connector=connector) as session:
    async with session.get(url, timeout=timeout) as response:
        body = await response.read()

Фрагмент показывает создание сессии и одну попытку. Чтобы получить очередь, стенд сначала занимает соединение отдельным запросом через эту же сессию; новая сессия для измеряемой попытки не создаётся. Готовый сценарий — make demo SCENARIO=pool_wait.

Соединение заранее занято вспомогательным запросом через ту же сессию: сервер отдал ему заголовки, а тело дописывает только по сигналу через 1100 мс. Просто «не читать маленький ответ» не сработало бы: aiohttp вернул бы соединение в пул, как только сервер дослал бы тело. Нужен именно незавершённый ответ.

По одной операции, ближайшей к медиане своей серии; N = 90 на вариант.

Без общего deadline клиент ждёт соединение 1100 мс и возвращает ответ на 1124 мс: двух секунд на попытку ему хватает. Общий бюджет прерывает ожидание на 1001 мс. В журнале клиента при этом есть одна попытка, а на сервере — ноль запросов: попытка уже началась, но до отправки по сети так и не дошла.

Оба варианта общего deadline здесь работают одинаково: пул ждёт внутри HTTP‑попытки, которую охватывают оба контекста. При ревью клиента важно проверять положение каждой очереди относительно таймаута.

Заодно отдельный тест подтверждает: короткий connect (100 мс) сам обрывает очередь пула с ConnectionTimeoutError. aiohttp «считает» пул в свой таймаут; я просто дал ему 2000 мс, чтобы посмотреть на внешнюю границу.

В тесте с коротким connect используется такая настройка:

timeout = aiohttp.ClientTimeout(
    total=2.0, 
    connect=0.1, 
    sock_connect=0.1, 
    sock_read=None,
)

Чтение тела

Допустим, соединение получили и сервер ответил. Здесь легко закончить измерение слишком рано: объект response уже есть, но вызывающему коду обычно нужно ещё и содержимое ответа.

Чтобы разделить эти моменты, сервер отдаёт заголовки сразу, а тело — 16 порциями с паузами 75 мс между ними. Получается около 1125 мс на передачу тела. У попытки total=1600 мс и sock_read=200 мс, общий бюджет — 1000 мс. Каждая порция приходит раньше, чем истечёт sock_read, хотя всё тело за секунду получить не получится.

В вызов session.get() передаётся такой таймаут:

timeout = aiohttp.ClientTimeout(
    total=1.6, 
    connect=0.3, 
    sock_connect=0.3, 
    sock_read=0.2,
)

Порции ответа настраиваются на серверной стороне, в experiments/slow_body.yaml:

scenario:
  type: slow_body
  status: 200
  chunks: 16
  chunk_interval_ms: 75

Полный сценарий: make demo SCENARIO=slow_body. Общий deadline по‑прежнему охватывает и session.get(), и response.read().

По одной операции, ближайшей к медиане своей серии; N = 90 на вариант.

На показанных операциях заголовки пришли за 1–2 мс. Клиент без общего deadline дочитал тело на 1146 мс и вернул успех. С общим бюджетом чтение прервалось на 1001 мс: заголовки получены, но полного тела нет. До проверки содержимого код уже не дошёл.

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

Бюджет истёк. Что именно остановилось?

В предыдущем опыте я написал «чтение прервалось». За этим стоит механизм отмены asyncio: timeout_at запрашивает отмену задачи, а та получает CancelledError и выходит из контекстов, освобождая ресурсы. Между срабатыванием таймера и возвратом управления есть работа, которую ещё нужно выполнить. Каталог в это время может продолжать обработку запроса: прекращение ожидания на нашей стороне ещё не означает остановку серверного обработчика.

Поэтому одной отметки «таймаут» в журнале мало. Нужно различать четыре момента:

1. отмена запрошена (таймер сработал или вызвали task.cancel());

2. код операции увидел CancelledError;

3. вызывающий код получил управление;

4. серверный обработчик завершил работу.

Чтобы проверить именно реакцию клиента на отмену, в следующих сценариях я вызываю task.cancel() снаружи. Такая отмена может прийти и от вызывающего кода, которому ответ больше не нужен.

В первом сценарии отменяю задачу через 100 мс после начала обработки запроса сервером, который готовил бы ответ 300 мс. Во втором — через 50 мс после начала паузы между попытками; сама пауза длится 600 мс. Так можно проверить и освобождение HTTP‑ресурсов, и то, что цикл повторов не продолжит работу после отмены.

В конфигурации указано событие, от которого стенд отсчитывает задержку до task.cancel():

# experiments/cancel_in_http.yaml
scenario:
  type: cancel_in_http
  response_delay_ms: 300
  trigger_event: server_request_started
  delay_after_event_ms: 100
  handler_cancellation: false
---
# experiments/cancel_in_backoff.yaml
operation:
  backoff_ms: [600, 600]
scenario:
  type: cancel_in_backoff
  trigger_event: backoff_started
  delay_after_event_ms: 50
  handler_cancellation: false

Запуск: make demo SCENARIO=cancel_in_http или make demo SCENARIO=cancel_in_backoff. Задержка отсчитывается после указанного события, а не после создания задачи.

Журнал одной операции первого сценария (cancel_in_http-r1-B-m0016, ближайшая к медиане, время от t0):

     0.0 ms  client  attempt_started a1
     2.0 ms  server  server_request_started
   103.1 ms  client  external_cancel_requested
   103.4 ms  client  attempt_cancelled a1
   103.4 ms  client  cancellation_observed
   103.4 ms  client  operation_finished
   303.3 ms  server  server_headers_prepared status=200
   303.4 ms  server  server_handler_finished

Что получилось на 630 отменённых операциях (360 в HTTP, 270 в паузе; для варианта с одной попыткой паузы нет, он в сценарии не участвует):

• между запросом отмены и возвратом управления вызывающему коду — медиана 0,3 мс, максимум 4 мс;

• ни одной второй попытки: adapter_attempts = 1 у всех, backoff_finished в журнале отсутствует, на сервере по одному запросу на операцию;

• CancelledError вышел наружу как CancelledError. При выходе закрывается response, освобождается слот семафора, исключение пробрасывается вызывающему коду.

Серверный обработчик продолжил работу и подготовил ответ после отмены клиента. Он запущен с handler_cancellation=False. Приведённый журнал показывает завершение обработчика; фактическую запись ответа в соединение эти события не измеряют. Здесь обработчик только спит, но у настоящего API за эти 300 мс мог произойти побочный эффект. Клиентский deadline его не откатывает.

Проверил и освобождение ресурсов: после отменённых операций семафор и пул продолжают пропускать следующие запросы. В первом опыте задача, занимавшая семафор, освободила его на 1101 мс; отменённый клиент слот за собой не оставил.

Теперь можно вернуться к той лишней миллисекунде из первого графика. На стенде timeout_at назначает отмену на t0 + 1000 мс. Для её доставки event loop должен получить управление. По 360 операциям с бюджетом на операцию в четырёх сценариях, где он срабатывал: медиана превышения над t0 + 1000 — 1,3 мс, максимум 5,3 мс. Deadline — кооперативная отмена, не hard real‑time: пока event loop не получил управление, никакой таймер не сработает. Блокирующий синхронный код в той же задаче отодвинул бы отмену на всё время блокировки; стенд этого не проверял.

Где записывать честные метрики для Prometheus?

Вернёмся к обработчику страницы товара. После ошибок каталога мы открываем дашборд и хотим понять, где прошла выделенная секунда. В опыте с семафором HTTP‑обмен занимал около 24 мс, а вызывающий код ждал 1124 мс. Если поставить таймер только вокруг session.get(), на графике зависимость будет быстрой, хотя бюджет вызова уже нарушен. Если закончить измерение на заголовках, потеряем ещё и время чтения тела.

Поэтому основной таймер ставлю на входе в операцию, до семафора, а запись делаю в finally после выхода из всех контекстов. В измерение попадут очередь, попытки, паузы, чтение тела и очистка ресурсов. Так мы измеряем ожидание вызывающего кода до границы возврата из клиента. Сам вызов записи метрики в эту длительность не входит.

Добавим измерение к уже знакомым fetch_once и retry_cycle. measured_call будет внешней границей клиента: именно здесь начинаются общий бюджет и таймер. Для примера нужен установленный prometheus_client. Это предлагаемая инструментализация клиента; исторические результаты стенда получены из журналов, Prometheus в них не участвовал. Метрика создаётся один раз на уровне модуля, а не при каждом запросе. Использую обычный Histogram с явными границами корзин; его API описан в документации Python‑клиента.

from enum import StrEnum

from prometheus_client import Histogram


class OperationOutcome(StrEnum):
    SUCCESS = "success"
    DEADLINE = "deadline"
    CANCELLED = "cancelled"
    ERROR = "error"


OPERATION_SECONDS: Histogram = Histogram(
    "upstream_operation_duration_seconds",
    "Полное ожидание операции, включая очередь, повторы и очистку",
    labelnames=("dependency", "outcome"),
    buckets=(0.05, 0.1, 0.25, 0.5, 0.75, 1.0, 1.1, 1.5, 2.0, 5.0),
)

for known_outcome in OperationOutcome:
    OPERATION_SECONDS.labels(dependency="catalog", outcome=known_outcome)


async def measured_call(
    semaphore: asyncio.Semaphore,
    budget_s: float,
    fetch: Callable[[], Awaitable[tuple[int, bytes]]],
) -> bytes:
    loop = asyncio.get_running_loop()
    started = loop.time()  # до первой очереди
    deadline = started + budget_s
    scope = asyncio.timeout_at(deadline)
    outcome = OperationOutcome.ERROR
    try:
        async with scope:
            if loop.time() >= deadline:
                raise OperationDeadlineExceeded("Бюджет уже исчерпан")
            async with semaphore:
                result: bytes = await retry_cycle(fetch, deadline)
        outcome = OperationOutcome.SUCCESS
        return result
    except OperationDeadlineExceeded:
        outcome = OperationOutcome.DEADLINE
        raise
    except TimeoutError:
        if scope.expired():
            outcome = OperationOutcome.DEADLINE
        raise
    except asyncio.CancelledError:
        outcome = OperationOutcome.CANCELLED
        raise
    finally:
        elapsed = loop.time() - started
        OPERATION_SECONDS.labels(
            dependency="catalog",
            outcome=outcome,
        ).observe(elapsed)

Вызов с уже созданными semaphore, session, url и timeout выглядит так:

body = await measured_call(semaphore, budget_s=1.0, fetch=fetch)

В finally ровно одно наблюдение на завершившуюся операцию. Три неудачные попытки дадут одну операцию с исходом error; ожидание семафора до deadline — одну операцию с исходом deadline, даже если HTTP‑запроса не было. Внешнюю отмену тоже записываем и пробрасываем. Когда она гоняется с deadline, outcome отражает вышедшее исключение; подробности гонки, включая scope.expired(), лучше сохранять в trace или журнале, не угадывать единственную первопричину.

У Histogram уже есть серия count. При таких labels она даёт число завершённых операций по исходам, поэтому отдельный счётчик с теми же измерениями не нужен. Операции, которые ещё висят, туда не попадают: для них полезен gauge upstreamoperations_in_flight, увеличиваемый перед очередью и уменьшаемый в том же внешнем finally.

Как понять, куда ушло время?

Общая длительность покажет проблему, но для разбора понадобятся дополнительные измерения. Я бы добавлял их по мере необходимости:

Эти дополнительные метрики в сниппете выше не реализованы. Для пула aiohttp предоставляет сигналы очереди соединений. Нельзя учитывать только пары «встал в очередь — получил соединение»: при отмене второго события может не быть. Начатое ожидание нужно закрыть на пути ошибки или отмены и записать один раз.

Попытку клиентского адаптера также не стоит называть запросом, полученным сервером. В опыте с занятым пулом попытка была одна, а запросов на сервере — ноль. Для числа реальных обработанных запросов нужна серверная метрика. Клиентская и серверная стороны дополняют друг друга — это соответствует рекомендациям Prometheus по инструментализации.

В labels оставляйте ограниченный набор значений: имя зависимости, имя операции, фиксированный исход. Полный URL, user ID, request ID и текст исключения отправляйте в логи или traces. В примере зависимость одна — catalog; это условное имя клиента. Не подставляйте туда произвольный адрес из запроса.

Как читать результат?

Рядом с latency нужно видеть доли исходов. После включения общего deadline длительность может уменьшиться просто потому, что клиент раньше сдаётся. По одному графику p95 это легко принять за ускорение. Квантили смотрите отдельно по исходам или по всей совокупности с явно указанным фильтром; p95 разных реплик нельзя усреднять, сначала агрегируются корзины гистограмм.

Для фиксированного бюджета в одну секунду долю успешных операций, завершённых не позже секунды, можно посчитать так:

sum(rate(upstream_operation_duration_seconds_bucket{
  dependency="catalog", outcome="success", le="1.0"
}[5m]))
/
sum(rate(upstream_operation_duration_seconds_count{
  dependency="catalog"
}[5m]))

В знаменателе все завершённые операции, включая ошибки и отмены. Пустой интервал без операций даст отсутствие осмысленной доли, а не 100% успеха. Корзина 1.0 выбрана под порог этого примера; при другом SLO поменяйте порог и границы корзин. Если бюджеты различаются от вызова к вызову, фиксированная корзина уже не отвечает на вопрос «успел ли каждый в свой бюджет»: этот признак надо вычислять при завершении операции и учитывать отдельно. Про агрегацию и выбор корзин подробнее написано в документации Prometheus.

Собираем клиент каталога

Соберём знакомые функции в get_catalog: fetch_once дочитывает ответ, retry_cycle управляет попытками, а measured_call ограничивает и измеряет всю операцию. Остаётся определить, какую ошибку получит обработчик страницы.

В этом примере данные каталога обязательны. После исчерпания попыток или времени клиент поднимает CatalogUnavailable; обработчик страницы преобразует её в ответ о временной недоступности, например HTTP 503. Повторять весь вызов клиента из обработчика уже не нужно: это выдало бы ему новый бюджет. Если продукт допускает устаревшие данные из кеша, решение о таком ответе тоже принимает вызывающий код, который знает требования страницы.

class CatalogUnavailable(RuntimeError):
    pass


async def get_catalog(
    session: aiohttp.ClientSession,
    semaphore: asyncio.Semaphore,
    url: str,
    budget_s: float,
) -> bytes:
    timeout = aiohttp.ClientTimeout(
        total=0.4, 
        connect=0.3, 
        sock_connect=0.3, 
        sock_read=None,
    )

    async def fetch() -> tuple[int, bytes]:
        return await fetch_once(session, url, timeout)

    try:
        return await measured_call(semaphore, budget_s, fetch)
    except (TimeoutError, RetriesExhausted, UnexpectedStatus, aiohttp.ClientError) as exc:
        raise CatalogUnavailable("Не удалось получить данные каталога") from exc

В метрику исход попадает до преобразования ошибки: истечение общего deadline остаётся deadline, исчерпание повторов — error. Причина сохраняется в цепочке исключений. CancelledError этот обработчик не ловит: если вызывающая задача отменена, клиент освобождает ресурсы и передаёт отмену дальше.

Для краткости здесь все неповторяемые HTTP‑статусы тоже превращаются в недоступность каталога. В настоящем контракте, например, «товар не найден» может быть самостоятельным результатом, а не ошибкой зависимости. Пример возвращает полностью прочитанные bytes; если клиент должен возвращать разобранный объект, разбор и проверку ответа нужно перенести внутрь измеряемой операции. Синхронную работу, блокирующую event loop, timeout_at при этом прервать не сможет.

Готовый файл со всеми функциями — examples/catalog_client.py. Из корня репозитория его можно запустить против своего тестового HTTP‑сервера:

uv sync --locked --all-groups --all-extras
.venv/bin/python examples/catalog_client.py http://localhost:8080/product/42

Адрес условный; в README есть команда для простого локального сервера. Клиент выведет размер полученного тела или сообщение о недоступности каталога. Сессия и семафор создаются в main; в веб‑сервисе их нужно создавать при запуске приложения, переиспользовать между вызовами и закрывать сессию при остановке. Число слотов 20 в примере — иллюстрация настройки, а не универсальная рекомендация.

Секунда тоже выбрана для эксперимента. В обработчике страницы бюджет каталога нужно согласовать с оставшимся временем на весь ответ. Если у вызывающего кода уже есть deadline, вместо выдачи новой секунды передавайте его дальше: для локальных вызовов с общей шкалой loop.time() используйте min(свой_deadline, внешний_deadline) и проверяйте, не истёк ли срок до начала работы. Наш короткий пример принимает отдельный бюджет клиента; общий deadline обработчика в нём не реализован. Передача срока по сети требует отдельного контракта.

Проверяем тот же клиент на занятой очереди

Вернёмся к проблеме, с которой начали: HTTP‑сервер отвечает быстро, а клиент долго ждёт допуска. Для проверки не нужен медленный каталог. Достаточно заранее занять слот и убедиться, что measured_call завершит ожидание, ни разу не вызвав fetch.

import pytest


@pytest.mark.asyncio
async def test_deadline_covers_admission() -> None:
    semaphore = asyncio.Semaphore(1)
    await semaphore.acquire()
    calls = 0

    async def fetch() -> tuple[int, bytes]:
        nonlocal calls
        calls += 1
        return 200, b"catalog data"

    try:
        with pytest.raises(TimeoutError):
            await measured_call(semaphore, budget_s=0.02, fetch=fetch)
        assert calls == 0
    finally:
        semaphore.release()  # освобождает владелец слота, а не ожидавшая задача

    result = await measured_call(semaphore, budget_s=1.0, fetch=fetch)
    assert result == b"catalog data"
    assert calls == 1

Первая операция не попадает в fetch, а после освобождения слота следующая проходит. В готовом примере тесты get_catalog проверяют и преобразование ошибки в CatalogUnavailable. В своём проекте проведите такой тест через публичный метод клиента, подменив только отправку HTTP‑запроса. Для очереди соединений понадобится настоящий локальный сервер и заранее занятый пул; на стенде это проверяет test_c_and_d_cover_the_pool_queue.

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

Не привязывайте тест к точному попаданию в 1000 мс: у event loop есть задержка доставки отмены и время на очистку. Здесь важны границы ожидания, отсутствие лишнего запроса и состояние ресурсов после завершения. Для запросов с побочными эффектами до включения повторов нужен договор об их безопасности, например поддерживаемый сервером ключ идемпотентности. Клиентский таймаут не сообщает, успел ли сервер выполнить предыдущую попытку.

Как повторить эксперименты?

Чтобы увидеть журнал сценария с занятым семафором, в подготовленном окружении репозитория выполните:

make demo SCENARIO=semaphore_wait

Установка и требования к окружению описаны в README. Команда запускает по одной операции каждого варианта; для повторения измерений там же есть benchmark-smoke и валидация результатов.

Что получилось у обработчика страницы?

В начале у клиента уже были таймаут, повторы и семафор. Но обработчик страницы мог прождать больше выделенной секунды и получить формально успешный ответ, который пришёл слишком поздно. После переноса границы очередь, HTTP‑попытки, паузы и чтение тела расходуют один бюджет. Если он исчерпан, клиент прекращает ожидание и возвращает управление обработчику; тот отвечает о недоступности каталога. Метрика показывает, сколько времени ушло на всю операцию, в том числе на неудачную.

Сам каталог от этого быстрее не стал. В опыте с занятой очередью мы получили ошибку без единого HTTP‑запроса. Для нашего обработчика это ожидаемый исход: он больше не начинает обращение, на которое уже не осталось времени. Работу, которую сервер успел выполнить, deadline при этом не откатывает.

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