rest-pipeline-js 1.3.0: параллельность, middleware, пауза и экспорт состояния

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

Начну с нового.

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

Параллельные шаги

Самое большое добавление в этом релизе. Раньше pipeline умел только последовательное выполнение — шаг за шагом. Это удобно, когда каждый следующий шаг зависит от результата предыдущего. Но часто несколько запросов вообще не связаны между собой и ждать завершения одного ради запуска другого — чистые потери времени.

Теперь в конфиге можно определить группу параллельных шагов через поле parallel:

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [
      { key: "auth", request: async () => getToken() },

      {
        key: "load-data",
        parallel: [
          { key: "loadUsers",    request: async () => fetchUsers() },
          { key: "loadProducts", request: async () => fetchProducts() },
          { key: "loadSettings", request: async () => fetchSettings() },
        ],
      },

      { key: "render", request: async ({ allResults }) => render(allResults) },
    ],
  },
});

auth отрабатывает первым, потом три запроса летят одновременно, и только когда все три завершились — запускается render. Всё это работает через один run(), никаких Promise.all руками.

Каждый шаг внутри группы — полноценный: у него свой ключ, свои before/after хуки, своя запись в stageResults, его можно перезапустить через rerunStep. Если хотя бы один из параллельных шагов упал с ошибкой — pipeline останавливается, success: false.

Global middleware

Раньше если нужно было логировать каждый шаг или отправлять метрики — приходилось или дублировать логику в каждый before/after, или вешать кучу on() подписок снаружи. Теперь есть middleware на уровне всего pipeline:

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [ /* ... */ ],
    middleware: {
      beforeEach: async ({ stage, index, sharedData }) => {
        sharedData.stepStartedAt = Date.now();
        console.log(`[${index}] Старт: ${stage.key}`);
      },
      afterEach: async ({ stage, result, sharedData }) => {
        const ms = Date.now() - sharedData.stepStartedAt;
        analytics.track("step_done", { key: stage.key, ms, data: result.data });
      },
      onError: async ({ stage, error }) => {
        await Sentry.captureException(error, { tags: { stage: stage.key } });
      },
    },
  },
});

beforeEach вызывается перед каждым шагом (раньше вызова before-хука самого шага), afterEach — после успешного завершения, onError — при любой ошибке. Middleware не заменяет per-stage errorHandler, они работают вместе.

Одна точка — всё логирование, трекинг, мониторинг. Больше не надо ничего разбрасывать по шагам.

pause() / resume()

Иногда нужно остановиться между шагами и подождать — пока пользователь что-то подтвердит, пока придёт WebSocket-сообщение, пока оператор нажмёт кнопку. Раньше для этого был onStepPause — callback, который прокидывался в run(). Работало, но неудобно: нельзя было поставить паузу из обработчика события.

Теперь у оркестратора есть явный API:

// Ставим паузу как только step1 завершился
orchestrator.on("step:step1:success", () => orchestrator.pause());

const runPromise = orchestrator.run();

// Ждём подтверждения от пользователя
await showConfirmDialog();

// Продолжаем
orchestrator.resume();
await runPromise;

pause() работает «мягко» — текущий шаг доводится до конца со всеми событиями, после чего pipeline ждёт. resume() запускает следующий шаг. Если во время паузы вызвать abort() — pipeline разблокируется и корректно завершится.

Проверить состояние можно через isPaused().

exportState() / importState()

Длинные pipeline — это больно, если страница перезагружается на середине. Или если нужно вернуться к результатам предыдущего запуска без повторного выполнения всех шагов.

// После завершения — сохраняем снимок
const snapshot = orchestrator.exportState();
sessionStorage.setItem("pipeline", JSON.stringify(snapshot));

// При следующей загрузке страницы
const saved = JSON.parse(sessionStorage.getItem("pipeline"));
const orchestrator = new PipelineOrchestrator({ config });
orchestrator.importState(saved);

// Теперь можно посмотреть что было, или запустить только нужный шаг
const prevUser = orchestrator.getProgress();
await orchestrator.rerunStep("processData"); // перезапустить только один шаг

exportState() возвращает обычный JSON-сериализуемый объект: stageResults и logs. Временны́е метки в логах хранятся как ISO-строки и автоматически конвертируются обратно в Date при importState.

Retry с backoff и фильтрацией по статусу

Конфиг retry существовал с первой версии, но внутри RequestExecutor просто повторял запрос сразу и без разбора. Теперь всё работает по-настоящему:

• delayMs — базовая задержка между попытками
• backoffMultiplier — экспоненциальный рост задержки (delayMs * multiplier^attempt)
• jitter — небольшой случайный разброс, чтобы не создавать thundering herd
• retriableStatus — список HTTP-статусов, при которых повтор имеет смысл

const client = createRestClient({
  baseURL: "https://api.example.com",
  retry: {
    attempts: 3,
    delayMs: 500,
    backoffMultiplier: 2,
    retriableStatus: [429, 500, 502, 503, 504],
  },
});

При 429 (rate limit) повторит через 500мс, потом через 1000мс, потом через 2000мс. При 404 не будет повторять вовсе — смысла нет.

Таймаут тоже переработан: раньше он просто отклонял промис через Promise.race, но реальный HTTP-запрос продолжал выполняться в фоне. Теперь таймаут реализован через AbortController — запрос физически отменяется.

PATCH и кэш на уровне клиента

Два маленьких, но полезных дополнения.

patch() — стандартный HTTP-метод для частичного обновления ресурса. Странно, что его не было раньше:

await client.patch("/users/42", { name: "Alice" });

clearCache() — очистить кэш ответов конкретного экземпляра клиента. Пригодится когда нужно инвалидировать данные без пересоздания клиента:

client.clearCache();

abort(), pause(), resume() в хуках Vue и React

usePipelineRunVue и usePipelineRunReact теперь возвращают больше:

Vue:

const { run, running, result, error, stageResults, abort, pause, resume, rerunStep } =
  usePipelineRunVue(orchestrator);

React:

const [run, { running, result, error, stageResults, abort, pause, resume, rerunStep }] =
  usePipelineRunReact(orchestrator);

Раньше чтобы позвать abort() из компонента — нужно было держать ссылку на оркестратор отдельно. Теперь всё из одного хука.

Исправленные баги

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

condition никогда не проверялся. Поле condition было в типах, статус "skipped" существовал — но внутри run() никакой проверки не было. Шаги выполнялись всегда. Исправлено: если condition возвращает false, шаг пропускается со статусом "skipped" и эмитируется событие step:<key>:skipped.

request() вызывался дважды. Перед выполнением шага оркестратор вызывал stage.request() первый раз — чтобы посмотреть, вернёт ли функция строку (URL). Потом вызывал второй раз — уже для реального выполнения. Если запрос имел побочные эффекты (POST, запись в базу) — он срабатывал дважды. Исправлено: request() вызывается строго один раз, возвращаемое значение всегда является данными.

rerunStep() игнорировал хуки. При повторном запуске шага через rerunStep хуки before, after, проверка condition и вызов middleware не выполнялись. Поведение при повторном запуске отличалось от первоначального. Теперь rerunStep и run используют один и тот же внутренний метод executeStage — логика идентична.

Двойной emit событий в rerunStep. emitStepStart уже внутри себя эмитировал step:<key>:start, но rerunStep дополнительно вызывал emit() вручную — подписчики получали событие дважды. То же самое с step:<key>:success. Исправлено вместе с рефакторингом на общий executeStage.

Таймаут не отменял HTTP-запрос. Promise.race отклонял промис когда истекало время, но сетевой запрос продолжал выполняться в фоне. На мобильных соединениях или при частых таймаутах это создавало утечку запросов. Теперь таймаут реализован через AbortController — axios получает signal и реально отменяет запрос.

CancelToken (deprecated API). Отменяемые запросы через cancellableRequest использовали axios.CancelToken.source(), который помечен как deprecated начиная с axios 0.22. Заменено на AbortController.

cache и rateLimit были заглушкой. Поля CacheConfig и RateLimitConfig существовали в типах и принимались конструктором, но ни кэш, ни ограничение запросов внутри createRestClient не были реализованы. Теперь реализованы по-настоящему: TTL-кэш с LRU-eviction для ответов, семафор и скользящее окно для rate limiting.

autoReset не очищал логи. При autoReset: true перед каждым запуском сбрасывались только stageResults. Логи накапливались между запусками неограниченно. Теперь this.logs = [] тоже сбрасывается.

getProgressRef() возвращал живую ссылку. Метод возвращал прямую ссылку на внутренний объект прогресса, который мог быть мутирован снаружи. getProgress() при этом возвращал копию. Поведение двух методов было несогласованным. getProgressRef() теперь тоже возвращает копию.

Утечка памяти в кэше клиентов. getRestClient() хранил все созданные экземпляры клиентов в Map без ограничения размера и без возможности очистки. При динамическом изменении конфигураций (разные токены, разные baseURL) кэш рос бесконечно. Добавлен LRU-eviction при превышении лимита и экспортирована функция clearRestClientCache().

Сообщения об ошибках были на русском. В toApiError два сообщения ("Запрос был отменен", "Произошла неизвестная ошибка") были захардкожены на русском языке. Для библиотеки с английской документацией и международными потребителями — очевидная проблема. Заменено на английские строки.

Версия 1.3.0 уже на npm:

npm i rest-pipeline-js@1.3.0

Репозиторий и вся документация: github.com/macrulezru/pipeline-js

Страница пакета на npm: npmjs.com/package/rest-pipeline-js