rest-pipeline-js 1.3.6: DAG-переходы, вложенные пайплайны, SWR-кэш и перехватчики

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

DAG-переходы через next()

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

Теперь у каждого шага появилось поле next() — функция, которая решает, куда идти дальше:

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [
      { key: "start", request: async () => ({ role: "admin" }) },
      { key: "adminTask", request: async () => "admin work" },
      { key: "userTask", request: async () => "user work" },
      {
        key: "router",
        request: async ({ prev }) => prev,
        next: ({ result }) => result.role === "admin" ? "adminTask" : "userTask"
      },
      { key: "finish", request: async () => "done" }
    ]
  }
});

next() получает результат текущего шага, все результаты предыдущих шагов и sharedData. Возвращает ключ следующего шага или null (продолжить по порядку).

Защита от бесконечных циклов тоже есть: если пайплайн сделал больше шагов, чем stages.length × 10, выбрасывается ошибка.

Вложенные пайплайны (subPipeline)

Большие пайплайны хочется декомпозировать. Раньше для этого приходилось создавать отдельные оркестраторы и запускать их вручную. Теперь можно вложить один пайплайн в другой как обычный шаг:

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [
      { key: "setup", request: async () => ({ ready: true }) },
      {
        key: "dataProcessing",
        subPipeline: {
          stages: [
            { key: "fetch", request: async () => fetchData() },
            { key: "validate", request: async ({ prev }) => validate(prev) },
            { key: "enrich", request: async ({ prev }) => enrich(prev) }
          ],
          options: { continueOnError: false }
        }
      },
      { key: "finalize", request: async ({ allResults }) => finalize(allResults.dataProcessing) }
    ]
  }
});

Дочерний пайплайн получает свой собственный оркестратор, но наследует sharedData и сигнал отмены от родителя. Результат вложенного пайплайна целиком сохраняется в stageResults[key]. Если дочерний пайплайн завершился с ошибкой — родительский тоже останавливается (если только не включён continueOnError).

continueOnError: не останавливаться на ошибке

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

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [
      { key: "widget1", request: fetchWidget1, continueOnError: true },
      { key: "widget2", request: fetchWidget2, continueOnError: true },
      { key: "widget3", request: fetchWidget3, continueOnError: true },
      { key: "render", request: ({ allResults }) => renderDashboard(allResults) }
    ],
    options: { continueOnError: false } // глобальный фоллбэк
  }
});

Можно задать на уровне шага (continueOnError: true) или глобально в options. Приоритет — у шага. Ошибочные шаги получают статус "error", но пайплайн идёт дальше.

pipelineRetry: перезапуск всего пайплайна

Если пайплайн упал — его можно перезапустить автоматически. И не обязательно с самого начала:

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [ /* ... */ ],
    options: {
      pipelineRetry: {
        attempts: 3,
        delayMs: 2000,
        retryFrom: "failed-step" // или "start"
      }
    }
  }
});

С retryFrom: "failed-step" успешно выполненные шаги не перезапускаются — пайплайн продолжается с того места, где упал. Это экономит время и ресурсы.

pipelineTimeoutMs: глобальный таймаут

Раньше таймаут можно было задать только на отдельный HTTP-запрос. Но пайплайн целиком тоже может зависнуть — например, если какой-то шаг никогда не резолвится.

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [ /* ... */ ],
    options: { pipelineTimeoutMs: 30000 } // 30 секунд на всё
  }
});

При превышении лимита оркестратор вызывает abort(), отменяя все текущие запросы через AbortController.

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

Перехватчики (Interceptors)

Наконец-то. Request, response и error — поодиночке или массивами:

const client = createRestClient({
  baseURL: "https://api.example.com",
  interceptors: {
    request: [
      (config) => ({ ...config, headers: { ...config.headers, 'X-Client': 'my-app' } }),
      async (config) => {
        const token = await getToken();
        return { ...config, headers: { ...config.headers, Authorization: `Bearer ${token}` } };
      }
    ],
    response: (response) => ({ ...response, data: response.data.results }),
    error: (error) => { console.error("API Error:", error); return error; }
  }
});

Перехватчики применяются последовательно в том порядке, в котором переданы. Request-перехватчики срабатывают до отправки запроса, response — после успешного ответа, error — при любой ошибке (включая ошибки в response-перехватчиках).

Глобальный onError

Простой способ поймать все ошибки в одном месте без возни с перехватчиками:

const client = createRestClient({
  baseURL: "https://api.example.com",
  onError: (error, config) => {
    myErrorTracker.capture(error, { url: config.url });
  }
});

Вызывается перед тем, как ошибка улетит дальше. Может быть асинхронным.

Stale-While-Revalidate

Самое вкусное для UI. Данные сначала показываются из кэша (даже устаревшего), а в фоне обновляются:

const client = createRestClient({
  baseURL: "https://api.example.com",
  cache: {
    enabled: true,
    ttlMs: 60000,        // свежие 60 секунд
    strategy: "stale-while-revalidate",
    staleMs: 30000       // ещё 30 секунд отдаём устаревшие, пока обновляем
  }
});

// Первый вызов: кэшируем
await client.get("/slow-endpoint");

// Через 70 секунд: мгновенно получаем устаревшие данные,
// а в фоне идёт обновление кэша
await client.get("/slow-endpoint"); // нет задержки!

Пользователь никогда не видит скелетон или лоадер, если данные уже были когда-то загружены. Интерфейс остаётся отзывчивым.

Дедупликация запросов (request deduplication)

В React компонентах часто случается так, что несколько экземпляров одного компонента запрашивают одни и те же данные одновременно. Раньше это означало несколько параллельных одинаковых запросов.

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

// Три одинаковых GET-запроса одновременно
Promise.all([
  client.get("/users/me"),
  client.get("/users/me"),
  client.get("/users/me")
]); // → выполнится только один реальный запрос

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

HEAD и OPTIONS методы

Два стандартных HTTP-метода, которых не хватало для полноты картины:

const headers = await client.head("/users/1");      // только заголовки
const options = await client.options("/users/1");   // Allow, CORS и т.д.

Оба работают с теми же конфигами, что и остальные методы — таймауты, ретраи, перехватчики.

Что нового в типах и внутренностях

TtlCache.getStale()

Кэш теперь умеет не только get() и set(), но и getStale(key, staleMs) — возвращает значение даже если TTL истёк, но не вышел за пределы staleMs. Возвращает { value, isStale: true } для устаревших записей.

Используется внутри SWR-стратегии, но может пригодиться и напрямую.

Новые экспорты из types.ts

• RequestInterceptor, ResponseInterceptor<T>, ErrorInterceptor
• SubPipelineStage и обновлённый PipelineItem
• Поля continueOnError, next, pipelineRetry, pipelineTimeoutMs
• CacheConfig с strategy и staleMs

Исправления и обратная совместимость

Все изменения полностью обратно совместимы. Все новые поля — опциональные. Старые конфиги работают как раньше.

Единственный момент, который потребовал внимания: конструктор PipelineOrchestrator принимает params.options для autoReset. Новое поле config.options не конфликтует с ним — они живут отдельно.

Дополнительно:

• Добавлена защита от бесконечных циклов в DAG-переходах
• ParallelStageGroup теперь корректно работает с continueOnError
• Улучшена типизация при работе с вложенными пайплайнами

Релиз 1.3.6 уже на npm:

npm i rest-pipeline-js@1.3.6

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