Перейти к содержанию

Как строить HTTP- и gRPC-клиенты для продакшена

Сервис заказов ходит на склад и в платёжный сервис, и любой из них иногда отвечает медленно или не отвечает вовсе. Первое, что хочется сделать, это повторить запрос.

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

Настроим его HTTP- и gRPC-клиенты, сломаем их зависимости и посчитаем, сколько запросов на самом деле до них дошло.

Создаём клиентов при запуске приложения

clientwright добавляет к HTTPX ограничения времени и политику повторов, а build('httpx', ...) по-прежнему возвращает обычный httpx.AsyncClient:

from clientwright import (
    AdapterDeps, ClientConfig, RetryConfig, TimeoutConfig, build,
)


def http_client(name, base_url, *, metrics=None):
    return build(
        'httpx',
        ClientConfig(
            service_name=name,
            base_url=base_url,
            timeout=TimeoutConfig(total=2, connect=0.3, read=0.5),
            retry=RetryConfig(),
            circuit_breaker=None,
            on_unsupported='strict',
        ),
        AdapterDeps(metrics=metrics),
    )

service_name подписывает метрики и логи клиента. RetryConfig() уже означает три попытки вместе с первой, паузы 100 мс с удвоением, разброс ±20 %, учёт Retry-After и бюджет повторов 0,1. Circuit breaker включён по умолчанию (пять неудач, 60 секунд); circuit_breaker=None выключает его, пока мы не настроим его ниже. on_unsupported='strict' превращает настройку, которую адаптер не умеет применить, в ошибку при создании клиента.

Создаём обоих клиентов в области приложения и закрываем после завершения активных обработчиков:

from contextlib import asynccontextmanager


@asynccontextmanager
async def outbound_clients(inventory_url, payments_url):
    async with (
        http_client('inventory', inventory_url) as inventory,
        http_client('payments', payments_url) as payments,
    ):
        yield inventory, payments

Если открывать их на каждый запрос, пропадут переиспользование соединений, бюджет повторов и состояние breaker.

Ограничиваем общее время запроса остатков

Склад отвечает на GET /stock/{sku} телом {"sku": "sku-42", "available": 7}:

async def fetch_stock(client, sku):
    response = await client.get(f'/stock/{sku}')
    response.raise_for_status()
    return response.json()['available']

Клиент даёт 0,3 секунды на подключение, 0,5 секунды между порциями ответа и две секунды на весь вызов вместе с попытками и паузами. Таймаут чтения HTTPX ограничивает паузу, а не весь ответ: в практикуме с дедлайном сервер, отдающий по байту каждые 80 мс, укладывается в таймаут чтения 300 мс, а TimeoutConfig(total=0.35, read=0.3) обрывает его с DeadlineExceededError.

Чтобы склад и оплата делили время входящего запроса, передайте клиентам AdapterDeps(deadline_source=AmbientDeadlineSource()) и установите BudgetContext через use_budget(), как в статье о дедлайнах. Две новые секунды на каждого клиента не равны двухсекундному бюджету на заказ.

Повторяем платёж, только если сервер умеет распознавать дубли

Сервис оплаты сохраняет платёж, а потом отвечает 503. Повторять безопасно, только если API оплаты на тот же Idempotency-Key с тем же телом возвращает сохранённый результат, а тот же ключ с другим телом отклоняет. На заказ приходится одно списание, поэтому ключ строим из ID заказа:

from clientwright.adapters.httpx import IDEMPOTENT_EXTENSION


async def charge(client, order_id, amount_minor):
    response = await client.post(
        '/payments',
        json={'order_id': order_id, 'amount_minor': amount_minor},
        headers={'Idempotency-Key': f'charge-{order_id}'},
        extensions={IDEMPOTENT_EXTENSION: True},
    )
    response.raise_for_status()
    return response.json()['payment_id']

IDEMPOTENT_EXTENSION разрешает clientwright повторять POST, а заголовок позволяет серверу узнать повтор. Практикум с повторами проверяет три варианта на локальном сервисе:

После первого списания сервер вернул 503 HTTP-запросов на вызов Списаний
Обычный POST 1 1; клиент получает 503
POST с флагом идемпотентности, но без ключа на сервере 2 2
POST со стабильным ключом и клиентским флагом 2 1; клиент получает ID сохранённого платежа

Повторный charge() для того же заказа возвращает тот же ID платежа. Практикум убирает дубли в памяти и последовательно; рабочему API нужны база данных и атомарная обработка одновременных дублей.

Учитываем паузу сервера и ограничиваем лишние запросы

С политикой по умолчанию первый повтор ждёт от 80 до 120 мс, второй от 160 до 240 мс. Retry-After: 1 заменяет это одной секундой. Если запрошенная пауза не помещается в оставшееся время, как Retry-After: 60 при общем лимите 0,3 секунды в практикуме, клиент сразу возвращает 503 после единственного запроса.

Бюджет повторов хранится внутри клиента для каждого origin, то есть сочетания схемы, хоста и порта. Новый origin получает десять токенов, больше десяти в запасе не бывает, а каждый вызов добавляет 0,1. Практикум с бюджетом делает 50 последовательных вызовов к серверу, который всегда отвечает 503; breaker выключен:

Настройки повторов Запросов на сервере
Три попытки, budget_ratio=None 150
Три попытки, budget_ratio=0.1 64

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

Сбой оплаты не должен блокировать склад

У отдельных клиентов и breaker отдельные. Общему клиенту, который ходит в несколько origin по полным URL, нужен breaker с ключом по origin, и по умолчанию так и сделано:

from clientwright import CircuitBreakerConfig


def shared_http_client():
    return build('httpx', ClientConfig(
        service_name='shop-backends',
        timeout=TimeoutConfig(total=2),
        retry=RetryConfig(),
        circuit_breaker=CircuitBreakerConfig(
            fail_threshold=3,
            recovery_timeout=10,
            half_open_max_calls=1,
        ),
    ))

В практикуме с breaker три неудачных вызова оплаты дают девять попыток, а четвёртый вызов получает CircuitOpenError без запроса; запрос остатков к другому origin при этом возвращает 200. Breaker считает подряд идущие неудачные вызовы со статусом 500 и выше; любой другой исход, включая 400 и 429, сбрасывает счёт. С тестовыми часами практикум проверяет и восстановление: из open через half-open в closed. Проверка breaker охватывает весь цикл повторов целиком:

ИДЕЯ В СХЕМЕОдин HTTP-вызов, несколько возможных попыток
---
config:
  theme: default
  look: classic
  flowchart:
    useMaxWidth: false
    wrappingWidth: 150
    padding: 12
    nodeSpacing: 24
    rankSpacing: 32
---
flowchart TD
    accTitle: Один HTTP-вызов, несколько возможных попыток
    accDescr: Breaker проверяется один раз на вызов и учитывает итоговый результат. Все попытки делят дедлайн вызова, и каждому повтору нужен токен бюджета.
    A["Дедлайн вызова"]
    B["Проверка circuit breaker"]
    C["Попытка запроса"]
    D["Оценка результата"]
    E["Повтор допустим, попытки есть, время есть, токен есть?"]
    F["Пауза с разбросом"]
    G["Учесть итог и вернуть результат"]
    A --> B --> C --> D --> E
    E -->|"повтор"| F --> C
    E -->|"без повтора"| G

Breaker проверяется один раз на вызов и учитывает итоговый результат. Все попытки делят дедлайн вызова, и каждому повтору нужен токен бюджета.

Если за одним origin шлюза стоят два независимо отказывающих сервиса, нужен ключ точнее, например CircuitKey.ORIGIN_ROUTE вместе с ROUTE_EXTENSION, или отдельные клиенты.

Явно перечисляем gRPC-методы, которые можно повторять

У gRPC-версии склада есть GetStock (чтение), Reserve (запись) и Export (поток). gRPC-практикум связывает их в InventoryStub на сырых байтах; свой stub генерируйте из protobuf. grpc-client-kit разрешает повторы только для GetStock:

import grpc
from grpc_client_kit import (
    RetryConfig as GrpcRetry,
    TimeoutConfig as GrpcTimeout,
    build_interceptors,
)


def inventory_chain():
    return build_interceptors(
        timeout=GrpcTimeout(default=1),
        retry=GrpcRetry(
            max_attempts=3,
            initial_backoff=0.05,
            retryable_codes={grpc.StatusCode.UNAVAILABLE},
            idempotent_methods={'/shop.Inventory/GetStock'},
        ),
    )

Без списка методов RetryConfig по умолчанию повторяет любой unary-метод при UNAVAILABLE и RESOURCE_EXHAUSTED. Но UNAVAILABLE не доказывает, что обработчик ничего не сделал: Reserve в практикуме меняет остаток и потом возвращает этот статус, и политика по умолчанию фиксирует три резервирования. Со списком резервирование одно, а клиент получает ошибку; GetStock при этом повторяется и возвращает b'7'. RESOURCE_EXHAUSTED мы тоже не повторяем: серверу, у которого кончились ресурсы, лишний трафик не нужен. Секундный таймаут относится ко всему RPC: если дать вызову 300 мс, медленная первая попытка оставит второй только остаток, и та завершится с DEADLINE_EXCEEDED.

Переиспользуем каналы только при совместимых настройках

Клиент аудита ходит по тому же адресу, но должен делать ровно одну попытку:

from contextlib import asynccontextmanager
from grpc_client_kit import ChannelPool, GrpcClient, GrpcClientConfig


@asynccontextmanager
async def grpc_clients(target):
    config = GrpcClientConfig(
        target=target,
        insecure=True,  # Local lab; configure credentials for a TLS deployment.
        options=[('grpc.enable_retries', 0)],
    )
    async with ChannelPool() as pool:
        orders = GrpcClient(
            InventoryStub, config, pool,
            interceptor_factory=lambda target: inventory_chain(),
        )
        audit = GrpcClient(
            InventoryStub, config, pool,
            interceptor_factory=lambda target: build_interceptors(
                timeout=GrpcTimeout(default=1), retry=None,
            ),
        )
        yield orders, audit

interceptor_factory создаёт и кеширует цепочку интерцепторов для каждого адреса, и идентичность канала остаётся стабильной. Пул различает каналы по настройкам безопасности, учётным данным, параметрам, сжатию и цепочке интерцепторов, а не по одному адресу; практикум с пулом проверяет переиспользование и разделение. Каналами владеет ChannelPool и закрывает их при выходе; выход из контекста GrpcClient каналы не трогает. grpc.enable_retries=0 оставляет повторы интерцептору, чтобы встроенные повторы gRPC их не умножали. Для клиента аудита retry=None убирает интерцептор повторов; пустой idempotent_methods не означал бы «ничего не повторять».

От HTTP-клиента gRPC-клиент отличается тремя вещами:

  • В grpc-client-kit 0.4.0 нет бюджета повторов.
  • Breaker хранится отдельно для каждого метода и стоит внутри цикла повторов, считая попытки: при пороге два breaker в практикуме блокирует третью попытку уже в первом неудачном вызове.
  • Перезапуск Export после элементов 1, 2 может дать 1, 2, 1, 2, 3. Практикум проверяет и этот явно включённый режим, и обычное завершение с ошибкой; не включайте повторы потоков, пока протокол не умеет продолжать с нужного места или убирать дубли. Как интерцептор видит итог потокового вызова, разобрано в отдельной статье.

Считаем попытки, а не только успешные вызовы

Для склада, который дважды ответил 503, а потом успешно, эта функция возвращает (7, 1, 3): количество товара, число вызовов и число попыток.

from clientwright.core.testing import RecordingMetrics


async def observe_stock(base_url):
    metrics = RecordingMetrics()
    async with http_client('inventory', base_url, metrics=metrics) as client:
        available = await fetch_stock(client, 'sku-42')
    return available, len(metrics.calls), len(metrics.attempts)

RecordingMetrics взят из библиотеки как тестовый сборщик; в рабочем сервисе свою реализацию передают через AdapterDeps(metrics=...). Один успешный вызов может скрывать три запроса. Длительность вызова заканчивается на заголовках ответа, а чтение тела записывается отдельной метрикой.

Настраиваем политики под операции, которые действительно вызываем

Начните с общего ограничения времени и списка операций, которые безопасно повторять, а потом подберите бюджет повторов и breaker на зависимости, которая намеренно отказывает. clientwright делает это для HTTP, сохраняя родные клиенты, а grpc-client-kit делает то же для gRPC-каналов и интерцепторов.

Примеры и практикумы

Фрагменты кода взяты из практикумов и проверены на Python 3.13, clientwright==0.5.0, httpx==0.28.1, grpc-client-kit==0.4.0 и grpcio==1.84.0 с локальными HTTP- и gRPC-серверами. Они проверяют поведение клиентов, а не производительность, TLS, прокси или хранение платежей.