rest-pipeline-js 2.0.0 - распределённый rate limiter, circuit breaker, трассировка запросов и идемпотентность

В прошлый раз закрывал пробелы в DX: pipe(), createPipeline(), валидатор конфига, метрики, плагины, stream-шаги. С тех пор пакетом стали пользоваться не только в браузерных SPA, но и на бэкенде — а там всплыли вопросы, которых в клиентском коде просто не существует: что если инстансов сервиса не один, а десять? Что если запрос ушёл, ответ потерялся, и непонятно — повторять его или нет? Что если пакет вообще не грузится в окружении без бандлера?

2.0.0 — ответ на эти вопросы. Плюс один реальный баг, который обнаружился только когда попытались честно прогнать пакет через голый node, без Vite и webpack за спиной.

Версия мажорная не для галочки: часть изменений ломает обратную совместимость. Собрал их отдельным блоком в конце.

Новые возможности

Распределённый rate limiter

Rate limiter в пакете всегда был in-memory и жил в пределах одного процесса. Это нормально для браузерного SPA — вкладка одна. Но на бэкенде за балансировщиком обычно крутится не один инстанс, а несколько, и каждый считает лимит независимо. maxRequestsPerInterval: 100 при трёх инстансах на деле означает 300 запросов к бэкенду — лимит есть, а толку от него как от резинового замка.

Теперь rateLimit принимает store — реализацию интерфейса RateLimiterStore с двумя методами: incrementWindow(key, intervalMs) для скользящего окна запросов и acquireConcurrencySlot(key, maxConcurrent, leaseMs) для ограничения одновременных запросов. Backend может быть чем угодно — Redis, Memcached, что угодно с атомарным инкрементом. Без store поведение не меняется ни на бит: старый in-memory алгоритм работает как раньше.

Два нюанса, о которых честно сказано в JSDoc, а не спрятано под капот:

• incrementWindow — это fixed-window счётчик, у него тот же эффект «двойного бёрста» на границе окна, что и у любого fixed-window лимитера (в отличие от sliding-log). Это осознанный компромисс ради простоты интерфейса.
• acquireConcurrencySlot не может быть математически точным без центрального сервиса блокировок — это approximate ограничение, как у большинства распределённых семафоров на практике. leaseMs защищает от deadlock'а, если процесс, занявший слот, упал и не освободил его.

Без явного key каждый инстанс RateLimiter получает свой случайный ключ — то есть store сам по себе шеринг не включает, нужен общий key на всех инстансах.

Распределённый circuit breaker

Та же история, что и с rate limiter'ом, только больнее. Circuit breaker после failureThreshold подряд идущих ошибок должен переставать долбить упавший бэкенд. Но если состояние (closed/open/half-open) хранится в памяти каждого из N инстансов отдельно — каждому из них нужно накопить свой failureThreshold ошибок, прежде чем он откроется. На деле это значит, что упавший бэкенд получит не failureThreshold лишних запросов, а N × failureThreshold — ровно то, от чего circuit breaker должен защищать.

circuitBreaker.store — интерфейс CircuitBreakerStore с get/set и опциональным incrementCounter. Без incrementCounter брейкер откатывается на get + вычислить + set, что может терять часть инкрементов под высокой конкурентной нагрузкой между инстансами — но остаётся fail-safe, просто не идеально точным. С incrementCounter (атомарный инкремент на стороне backend'а, например INCR в Redis) гонки не возникает.

Из-за этого все публичные методы CircuitBreaker — getState, canExecute, onSuccess, onFailure — стали асинхронными. Без store они по-прежнему резолвятся мгновенно, никакой реальной асинхронной работы не происходит — просто единый интерфейс независимо от backend'а.

Распределённый кэш ответов

Третий из этой же серии: cache.store — интерфейс CacheStore (get/set/delete/clear, опционально getStale/deleteWhere), которым можно заменить встроенный TtlCache. Кэш, живущий только в памяти процесса, — то же самое узкое место, что и с лимитером: у каждого инстанса свой холодный кэш, hit rate падает пропорционально числу инстансов. С Redis-бэкендом кэш общий, и не важно, на какой инстанс попал запрос.

invalidateCache() и clearCache() теперь async — иначе с асинхронным backend'ом (сетевой поход в Redis) их нельзя было бы честно реализовать.

Трассировка запросов

Два независимых, но дополняющих друг друга механизма.

tracing.generateTraceparent добавляет к каждому запросу заголовок traceparent по спецификации W3C Trace Context — если он ещё не проставлен явно. Любая система мониторинга, которая понимает трейс-контекст (а таких большинство), сможет связать этот запрос с остальной распределённой трассой.

tracing.provider — хук, оборачивающий каждый запрос в реальный спан вашей системы трассировки: startSpan() перед запросом, end()/setStatus()/recordException() после. Форма интерфейса (TracingProvider/TracingSpan) намеренно повторяет подмножество API Span из OpenTelemetry — но зависимости от @opentelemetry/api в пакете нет. Реальный OTel SDK подключается тонким адаптером в одну строчку; то же самое верно для Sentry, Datadog и любой другой системы со схожим API.

Отдельно — RestRequestConfig.traceId: явный 32-символьный hex trace-id, чтобы связать несколько запросов в одну трассу вместо случайного нового на каждый вызов. Удобное совпадение: UUID без дефисов — это ровно 32 hex-символа, а значит orchestrator.getRunId().replace(/-/g, "") можно напрямую передавать как traceId — и все HTTP-запросы одного запуска пайплайна лягут в одну трассу.

Идемпотентность мутирующих запросов

RestRequestConfig.idempotencyKey отправляет заголовок Idempotency-Key (имя настраивается через HttpConfig.idempotencyHeaderName). Смысл — та же проблема, что решают идемпотентные ключи в Stripe, PayPal и куче внутренних API: если ответ на мутирующий запрос (POST/PUT/PATCH/DELETE) потерялся — по таймауту, обрыву соединения — вызывающая сторона не знает, применилась ли операция. Повторный запрос с тем же ключом бэкенд может безопасно распознать как дубликат вместо повторного применения. Библиотека только отправляет заголовок — дедупликация всегда на стороне backend'а.

Отдельного внимания заслуживает то, где именно реализован retry в пакете: не в createRestClient() — там его никогда не было, — а в RequestExecutor, том самом классе, который используется оркестратором для шагов без request-функции. Именно поэтому HttpConfig.autoIdempotencyKey реализован в RequestExecutor: он генерирует ключ один раз перед началом retry-цикла и переиспользует его на каждой попытке. Если этого не сделать — а понять это можно, только зная внутреннее устройство, — то при ручном повторении запроса вне RequestExecutor каждая попытка получит свой ключ, и вся идея идемпотентности сломается молча.

Ограничение роста логов

Мелкая, но неприятная деталь: orchestrator.getLogs() копил все события с начала жизни оркестратора и никогда сам себя не подрезал. Для одноразового run() это незаметно, но для долгоживущего инстанса — например, singleton в SPA, который гоняют через run() сотни раз без autoReset — это тихий memory leak.

options.maxLogs ограничивает лог N последними записями (FIFO). Без опции поведение не меняется — старые логи, как и раньше, копятся бесконечно, если явно не задано иное.

Безопасность по умолчанию

Раньше sanitizeHeaders был выключен по умолчанию — значит, Authorization/Cookie по умолчанию улетали в onRequestStart/onRequestEnd в открытом виде. А эти коллбэки почти всегда прокидываются во внешние системы наблюдаемости. Теперь маскирование включено по умолчанию; чтобы увидеть заголовки как есть (например, локально при отладке), нужно явно передать sanitizeHeaders: false.

Примеры использования

Распределённый rate limiter

redisRateLimiterStore реализует два метода интерфейса: incrementWindow атомарно увеличивает счётчик в Redis и выставляет TTL только при первом инкременте (count === 1), а acquireConcurrencySlot (заглушка здесь — полная реализация с Lua-скриптом в examples/) отвечает за ограничение одновременных запросов. Дальше стор просто передаётся в rateLimit.store, а key — это имя бакета: если несколько инстансов сервиса используют один и тот же key, они делят один лимит в Redis вместо того, чтобы каждый считал свой.

import { createRestClient, type RateLimiterStore } from "rest-pipeline-js";

const redisRateLimiterStore: RateLimiterStore = {
  async incrementWindow(key, intervalMs) {
    const count = await redis.incr(key);
    if (count === 1) await redis.pexpire(key, intervalMs);
    return count;
  },
  async acquireConcurrencySlot(key, maxConcurrent, leaseMs) {
    // атомарная проверка "меньше лимита → инкремент" обычно нужен Lua-скрипт;
    // полный пример — examples/redis-rate-limiter-store.ts
  },
};

const client = createRestClient({
  baseURL: "https://api.example.com",
  rateLimit: {
    maxRequestsPerInterval: 100,
    intervalMs: 60_000,
    store: redisRateLimiterStore,
    key: "api-example-com", // общий бакет на все инстансы
  },
});

Распределённый circuit breaker

Здесь get/set читают и пишут состояние брейкера (closed/open/half-open плюс счётчики) как JSON-блоб в Redis. incrementCounter — опциональный, но важный метод: он атомарно инкрементирует счётчик ошибок или успехов прямо в Redis (INCR), избегая гонки между инстансами, которая была бы при обычном чтении-изменении-записи. key здесь играет ту же роль, что и в rate limiter'е — общий бакет состояния на все инстансы. Последняя строка показывает, что getCircuitBreakerState() теперь возвращает промис — это касается вызова независимо от того, настроен store или нет.

import { createRestClient, type CircuitBreakerStore } from "rest-pipeline-js";

const redisCircuitBreakerStore: CircuitBreakerStore = {
  async get(key) {
    const raw = await redis.get(key);
    return raw ? JSON.parse(raw) : null;
  },
  async set(key, state, ttlMs) {
    await redis.set(key, JSON.stringify(state), "PX", ttlMs);
  },
  async incrementCounter(key, field, ttlMs) {
    const n = await redis.incr(`${key}:${field}`);
    if (n === 1) await redis.pexpire(`${key}:${field}`, ttlMs);
    return n;
  },
};

const client = createRestClient({
  baseURL: "https://api.example.com",
  circuitBreaker: {
    failureThreshold: 5,
    openMs: 30_000,
    store: redisCircuitBreakerStore,
    key: "api-example-com",
  },
});

await client.getCircuitBreakerState(); // теперь Promise<"closed" | "open" | "half-open">

Распределённый кэш ответов

Самый простой из трёх сторов: get/set/delete/clear один в один ложатся на стандартные команды Redis (GET, SET ... PX, DEL, FLUSHDB). Значение сериализуется в JSON перед записью и парсится обратно при чтении — сам пакет ничего не знает про формат хранения, это целиком забота стора. client.invalidateCache(...) в последней строке — уже знакомый метод точечной инвалидации по URL, просто теперь он async, потому что поход в Redis тоже асинхронный.

import { createRestClient, type CacheStore, type ApiResponse } from "rest-pipeline-js";

const redisStore: CacheStore<ApiResponse<unknown>> = {
  async get(key) {
    const raw = await redis.get(key);
    return raw ? JSON.parse(raw) : undefined;
  },
  async set(key, value, ttlMs) {
    await redis.set(key, JSON.stringify(value), "PX", ttlMs);
  },
  async delete(key) { await redis.del(key); },
  async clear() { await redis.flushdb(); },
};

const client = createRestClient({
  baseURL: "https://api.example.com",
  cache: { enabled: true, ttlMs: 60_000, store: redisStore },
});

await client.invalidateCache("/users/1"); // тоже теперь async

Трассировка запросов

otelProvider — это и есть тот самый «тонкий адаптер»: startSpan просто делегирует вызов настоящему трейсеру из @opentelemetry/api, потому что форма TracingProvider совпадает с нужным подмножеством API Span. generateTraceparent: true включает автоматическую простановку W3C-заголовка на каждый запрос. А в блоке под комментарием показано, как связать несколько запросов в одну трассу вручную: runId пайплайна (это UUID) избавляется от дефисов и передаётся как traceId в оба вызова — оба запроса окажутся в одной трассе, а не в двух разных.

import { createRestClient, type TracingProvider } from "rest-pipeline-js";
import { trace } from "@opentelemetry/api";

const tracer = trace.getTracer("my-app");
const otelProvider: TracingProvider = {
  startSpan: (name, attributes) => tracer.startSpan(name, { attributes }),
};

const client = createRestClient({
  baseURL: "https://api.example.com",
  tracing: { generateTraceparent: true, provider: otelProvider },
});

// Один traceId на все запросы одного запуска пайплайна:
const traceId = orchestrator.getRunId().replace(/-/g, "");
await client.get("/users/1", { traceId });
await client.get("/orders", { traceId }); // та же трасса

Идемпотентность мутирующих запросов

Пример показывает два способа получить один и тот же результат. В первом ключ генерируется вручную один раз и передаётся в idempotencyKey — подходит, когда retry реализован своим кодом снаружи библиотеки. Во втором создаётся RequestExecutor (тот самый класс, который реально умеет ретраить) с autoIdempotencyKey: true — он сам сгенерирует ключ перед первой попыткой и молча передаст его на каждый повтор внутри своего retry-цикла, так что все попытки одного execute()-вызова будут нести одинаковый заголовок Idempotency-Key.

// Вручную — один ключ на логическую операцию, переиспользуемый при ручных ретраях
const idempotencyKey = crypto.randomUUID();
await client.post("/orders", cart, { idempotencyKey });

// Автоматически — RequestExecutor сам генерирует ключ один раз перед retry-циклом
import { RequestExecutor } from "rest-pipeline-js";

const executor = new RequestExecutor({
  baseURL: "https://api.example.com",
  autoIdempotencyKey: true,
  retry: { attempts: 2, delayMs: 300, backoffMultiplier: 2 },
});

// Каждая попытка внутри отправит один и тот же Idempotency-Key
await executor.execute("/orders", { method: "POST", data: { items: ["sku-1"] } });

Ограничение роста логов

Самый короткий пример из всех: одна опция maxLogs в options. Как только внутренний массив логов оркестратора превышает 500 записей, самые старые вытесняются новыми — обычный FIFO. Ничего больше настраивать не нужно, и без этой опции поведение остаётся прежним — лог продолжает расти неограниченно, как и раньше.

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [/* ... */],
    options: { maxLogs: 500 }, // старые записи вытесняются новыми (FIFO)
  },
});

Исправленные ошибки

Пакет не грузился нативным Node ESM. Самая серьёзная находка релиза. У dist/esm/*.js не было package.json с {"type":"module"}, поэтому Node сначала пытался парсить их как CommonJS, ловил предупреждение MODULE_TYPELESS_PACKAGE_JSON и перепарсивал заново. Хуже: относительные импорты в сборке (export * from "./rest-client") были без расширения — bundler-резолвинг (Vite, webpack) это прощает, а нативный ESM-резолвер Node — нет, падает с ERR_MODULE_NOT_FOUND. То есть import "rest-pipeline-js" под голым node, без бандлера — включая как раз те edge/serverless окружения, под которые сделан HttpAdapter, — не работал вообще. Работали только консьюмеры через бандлер. Починили добавлением .js расширений во все относительные импорты и генерацией dist/esm/package.json / dist/cjs/package.json при сборке; теперь это проверяется отдельным скриптом в CI на каждый пуш.

useRestClientReact держал протухшее замыкание колбэков. Хук мемоизировал REST-клиент по JSON.stringify(config) — что незаметно роняло из сравнения все функции (auth, metrics, onError, interceptors, adapter), потому что JSON.stringify просто выбрасывает функции из объекта. Новый инлайновый колбэк на следующем рендере никогда не подхватывался — клиент продолжал звать замыкание, захваченное на первом рендере. Переключили на мемоизацию по ссылке (стандартная семантика useMemo) — пробела больше нет, ценой того, что теперь конфиг нужно мемоизировать самому, если не хочется нового клиента на каждый рендер.

Кэш клиентов слеп к тому, какой именно store/isFailure/provider передан. Внутренний кэш getRestClient() строит ключ через JSON.stringify над конфигом — а функции в JSON.stringify пропадают молча. Это значит, что два конфига, отличающиеся только какой стор или предикат в них передан (например, разные Redis-инстансы для cache.store), могли схлопнуться в один закэшированный клиент. Нашли это, пока реализовывали как раз RateLimiterStore/CircuitBreakerStore — добавили в ключ boolean-флаги присутствия для всех функциональных полей, по той же схеме, что уже использовалась для metrics/auth/adapter.

React-хук был на одну функцию беднее Vue-хука. usePipelineRunVue возвращал clearStageResults, usePipelineRunReact — нет. Просто рассинхронизировались в процессе развития; починили паритет.

Демо-баг, найденный уже во время написания демо-страницы для новых фич. Строили демо-страницу для autoIdempotencyKey — сценарий «нажать кнопку дважды с одними и теми же настройками». Второй клик вёл себя не так: retry вообще не проигрывался, а результат появлялся будто из ниоткуда. Причина — тот же класс бага, что и с store/isFailure выше, только с другой стороны: RequestExecutor строит клиента через закэшированный getRestClient(), и второй вызов с конфигом, который выглядит идентично первому (кроме identity замыкания adapter), получал клиента из кэша — со старым адаптером первого запуска, чей счётчик попыток уже досчитал до конца. Починили явным clearRestClientCache() перед каждым прогоном демо-сценария — и заодно перепроверили это отдельным Node-скриптом прямо на собранном dist, а не только на глаз в браузере.

Vue-тесты хуков молчаливо тестировали не то, что нужно. usePipelineRunVue и другие композаблы вызывались в тестах напрямую, вне какого-либо компонента — из-за этого onUnmounted (нужен для отписки от stageResults) ловил предупреждение Vue о том, что нет активного instance компонента, а сам путь очистки при размонтировании ни разу не проверялся по-настоящему. Переписали тесты через withSetup — маленький хелпер, монтирующий композабл внутри setup() настоящего компонента, — предупреждение исчезло, а cleanup-путь наконец реально протестирован.

Обратная совместимость: что ломается

Версия мажорная не просто по счётчику. Четыре изменения требуют правок в существующем коде:

• sanitizeHeaders теперь true по умолчанию (было false) — если нужен старый вывод в открытом виде, задайте sanitizeHeaders: false явно.
• client.clearCache() и client.invalidateCache() стали async — добавьте await на местах вызова.
• client.getCircuitBreakerState() и все публичные методы CircuitBreaker стали async — то же самое, добавьте await.
• useRestClientReact(config) теперь пересоздаёт клиент по смене ссылки на config, а не по JSON.stringify — если раньше полагались на дефолтное поведение с инлайновым объектом конфига, замемоизируйте его сами (useMemo, useState, модульная константа).

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

Релиз 2.0.0:

npm i rest-pipeline-js@2.0.0

Репозиторий: github.com/macrulezru/pipeline-js
Документация: всё в README, плюс копируемые примеры в examples/
npm: npmjs.com/package/rest-pipeline-js