Pipeline Orchestrator

PipelineOrchestrator

Main class for building and managing a pipeline of sequential (and parallel) stages.

Constructor

new PipelineOrchestrator({
  config,       // PipelineConfig — stages and optional middleware
  httpConfig?,  // HttpConfig — HTTP client settings
  sharedData?,  // Record<string, any> — shared pool across all stages
  options?,     // { autoReset?: boolean }
})

Methods

Method	Description
run(onStepPause?, externalSignal?)	Execute all stages. Returns { stageResults, success }
rerunStep(stepKey, options?)	Re-execute a single stage (respects condition, before, after, middleware)
abort()	Abort pipeline execution (cancels the current HTTP request via AbortSignal)
isAborted()	Check if pipeline was aborted
pause()	Pause after the current stage completes
resume()	Resume a paused pipeline
isPaused()	Check if pipeline is paused
exportState()	Serialize stageResults and logs to a plain object
importState(state)	Restore stageResults and logs from a snapshot
getStageResults()	Synchronous snapshot of all stage results
getRunId()	ID of the current/last run() or rerunStep() — see Correlating a run
destroy()	Run cleanup callbacks from all installed plugins
subscribeProgress(listener)	Subscribe to progress updates
subscribeStageResults(listener)	Subscribe to stageResults changes
subscribeStepProgress(stepKey, listener)	Subscribe to a specific stage's progress
on(eventName, handler)	Subscribe to any event (step:<key>:start|success|error|skipped|progress, log)
onStepStart/Finish/Error(handler)	Subscribe to stage lifecycle events
getProgress()	Get current progress snapshot
getLogs()	Get all pipeline logs (capped at options.maxLogs entries, if set — see below)
clearStageResults()	Reset results and progress

Stage parameters (PipelineStageConfig)

Every hook below also receives signal: AbortSignal in its params object — the same signal used by orchestrator.abort(). Pass it down to fetch/axios/etc. inside request/before/after so cancellation actually stops in-flight work, not just the pipeline's bookkeeping.

Parameter	Description
key	Unique stage identifier
request({ prev, allResults, sharedData, signal })	Main stage function — return value becomes the stage result
condition({ prev, allResults, sharedData, signal })	If returns false, stage is skipped with status "skipped"
before({ prev, allResults, sharedData, signal })	Pre-processing hook — returned value replaces prev passed to request
after({ result, allResults, sharedData, signal })	Post-processing hook — returned value replaces the stage result
errorHandler({ error, key, sharedData, signal })	Per-stage error handler — see Error recovery below
retryCount	Override retry count for this stage
timeoutMs	Override timeout for this stage
pauseBefore	Delay in ms before executing request
pauseAfter	Delay in ms after executing request

Error recovery (errorHandler + recoverStep)

By default, whatever errorHandler returns is wrapped into an ApiError and the stage stays "error" — it can transform/enrich the error but not turn the failure into a success. Return recoverStep(data) to recover the stage instead: it's committed exactly like a successful stage (status "success", afterEach middleware, metrics, persistence) and the pipeline continues normally.

import { recoverStep } from "rest-pipeline-js";

{
  key: "fetchPrice",
  request: async () => fetchPriceFromApi(),
  errorHandler: ({ error }) => {
    if (isNetworkError(error)) return recoverStep(0); // fall back to a default and continue
    return error; // anything else: keep failing as before
  },
}

Stage execution flow

condition? → false → [status: skipped] → next stage
           ↓ true
middleware.beforeEach
           ↓
pauseBefore
           ↓
before() hook
           ↓
request()
           ↓
after() hook
           ↓
pauseAfter
           ↓
middleware.afterEach
           ↓
[status: success] → next stage

On error at any point:
  └─► stage.errorHandler (if set)
        ├─► returns recoverStep(data) → [status: success] → next stage
        └─► otherwise → middleware.onError → [status: error] → stop

Full example

import { PipelineOrchestrator } from 'rest-pipeline-js'

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [
      {
        key: 'fetchUser',
        request: async ({ sharedData }) => {
          const res = await fetch(`/api/users/${sharedData.userId}`)
          return res.json()
        },
      },
      {
        key: 'processData',
        condition: ({ prev }) => prev !== null,
        before: ({ prev }) => ({ ...prev, processed: true }),
        request: async ({ prev }) => prev,
        after: ({ result }) => ({ ...result, finishedAt: Date.now() }),
      },
    ],
    middleware: {
      beforeEach: ({ stage }) => console.log('Starting:', stage.key),
      afterEach: ({ stage, result }) => console.log('Done:', stage.key, result.data),
      onError: ({ stage, error }) => console.error('Error in', stage.key, error),
    },
  },
  httpConfig: {
    baseURL: 'https://api.example.com',
    retry: { attempts: 2, delayMs: 1000, backoffMultiplier: 2 },
    cache: { enabled: true, ttlMs: 60000 },
  },
  sharedData: { userId: 42 },
  options: { autoReset: true },
})

orchestrator.subscribeProgress((progress) => {
  console.log('Stage:', progress.currentStage, 'Statuses:', progress.stageStatuses)
})

orchestrator.on('step:fetchUser:success', (payload) => {
  console.log('fetchUser done:', payload.data)
})

const result = await orchestrator.run()
console.log('Pipeline finished:', result.success)
console.log('Stage results:', result.stageResults)

Parallel stages

Group stages for concurrent execution using parallel:

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

      // Parallel group — all run concurrently
      {
        key: 'load-data',
        parallel: [
          { key: 'loadUsers', request: async () => fetchUsers() },
          { key: 'loadProducts', request: async () => fetchProducts() },
          { key: 'loadSettings', request: async () => fetchSettings() },
        ],
      },

      // Sequential stage after the group
      { key: 'render', request: async ({ allResults }) => render(allResults) },
    ],
  },
})

• All stages in a parallel group run simultaneously via Promise.all — unless concurrency is set (see below).
• If any stage in the group fails, the pipeline stops and marks success: false.
• Each parallel stage has its own key and result in stageResults.
• rerunStep(key) works for stages inside parallel groups too.

Limiting concurrency

For fan-out over many items (e.g. paginated fetches), set concurrency on the group to cap how many stages run at once instead of starting all of them immediately:

{
  key: "fetch-all-pages",
  parallel: pageNumbers.map((n) => ({
    key: `page-${n}`,
    request: async () => fetchPage(n),
  })),
  concurrency: 5, // at most 5 requests in flight at a time
}

Results land in stageResults under their own key regardless of concurrency, in the same shape as an unlimited group. With the pipe() builder: .parallel(stages, { concurrency: 5 }).

Global middleware

Apply hooks to every stage without modifying individual stage configs:

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [/* ... */],
    middleware: {
      beforeEach: async ({ stage, index, sharedData }) => {
        console.log(`[${index}] Starting: ${stage.key}`)
        sharedData.startedAt = Date.now()
      },
      afterEach: async ({ stage, index, result, sharedData }) => {
        const ms = Date.now() - sharedData.startedAt
        console.log(`[${index}] Done: ${stage.key} in ${ms}ms`, result.data)
      },
      onError: async ({ stage, error, sharedData }) => {
        await reportError({ stage: stage.key, error, context: sharedData })
      },
    },
  },
})

Middleware runs in addition to (not instead of) per-stage errorHandler.

Pause / Resume

Pause the pipeline after a stage and resume later:

const orchestrator = new PipelineOrchestrator({ config })

// Pause after step1 completes
orchestrator.on('step:step1:success', () => orchestrator.pause())

const runPromise = orchestrator.run()

// At some point later (e.g. after user confirmation):
await showConfirmDialog()
orchestrator.resume()

await runPromise

• pause() — pipeline waits after the current stage finishes (including events).
• resume() — continues from the next stage.
• abort() while paused unblocks the pipeline and terminates it.

Export / Import state

Save and restore the pipeline state across page reloads or sessions:

const orchestrator = new PipelineOrchestrator({ config })
await orchestrator.run()

// Save state
const snapshot = orchestrator.exportState()
localStorage.setItem('pipelineState', JSON.stringify(snapshot))

// Later — restore and inspect without re-running
const saved = JSON.parse(localStorage.getItem('pipelineState'))
const orchestrator2 = new PipelineOrchestrator({ config })
orchestrator2.importState(saved)

console.log(orchestrator2.getProgress()) // restored progress
console.log(orchestrator2.getLogs()) // restored logs (timestamps as Date objects)

exportState() returns { stageResults, logs } — a plain JSON-serializable object. Timestamps in logs are stored as ISO strings and restored as Date objects on importState.

Capping log growth (maxLogs)

logs grows by one entry per step event and is never trimmed automatically — fine for a single run(), but an orchestrator instance reused across many runs without autoReset (e.g. a long-lived SPA singleton) accumulates logs indefinitely. Set options.maxLogs to keep only the N most recent entries (oldest evicted first):

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [/* ... */],
    options: { maxLogs: 500 },
  },
})

Without maxLogs, behavior is unchanged from previous versions.

Pipeline metrics

Observe pipeline execution without modifying stage logic:

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [/* ... */],
    metrics: {
      onPipelineStart: ({ timestamp, runId }) => {
        console.log(`[${runId}] Pipeline started at`, new Date(timestamp).toISOString())
      },
      onPipelineEnd: ({ durationMs, success, stageResults, runId }) => {
        analytics.track('pipeline_complete', { durationMs, success, runId })
      },
      onStepDuration: ({ stepKey, durationMs, status, runId }) => {
        console.log(`[${runId}] [${stepKey}] ${status} in ${durationMs}ms`)
      },
    },
  },
})

Callback	Receives	Description
onPipelineStart	{ timestamp, runId }	Fires at the beginning of run()
onPipelineEnd	{ durationMs, success, stageResults, runId }	Fires when run() completes
onStepDuration	{ stepKey, durationMs, status, runId }	Fires after every executed step

Correlating a run (runId)

Every run() call generates a fresh runId (a UUID, or a timestamp-based fallback in environments without crypto.randomUUID), shared by all metrics callbacks, log entries (getLogs()), and step events (PipelineStepEvent.runId) produced during that run — including all attempts of pipelineRetry. rerunStep() generates its own separate runId. Use orchestrator.getRunId() to read the current/last one, or read runId off any event/log/metrics callback to correlate everything that happened during one execution in your logging/tracing backend:

orchestrator.on('log', (entry) => sendToLogBackend({ ...entry, runId: orchestrator.getRunId() }))

createPipeline() + pipe() builder

createPipeline() — short factory

import { createPipeline } from 'rest-pipeline-js'

const orchestrator = createPipeline(
  [
    { key: 'fetchUser', request: async () => fetchUser() },
    { key: 'process', request: async ({ prev }) => process(prev) },
  ],
  {
    httpConfig: { baseURL: 'https://api.example.com' },
    sharedData: { userId: 42 },
    pipelineOptions: { continueOnError: false },
    metrics: {
      onStepDuration: ({ stepKey, durationMs }) => console.log(stepKey, durationMs),
    },
  },
)

pipe() — fluent builder

import { pipe } from 'rest-pipeline-js'

const orchestrator = pipe()
  .step({ key: 'auth', request: async () => getToken() })
  .step({ key: 'fetchUser', request: async ({ prev }) => fetchUser(prev) })
  .parallel([
    { key: 'loadPosts', request: async () => fetchPosts() },
    { key: 'loadNotifs', request: async () => fetchNotifications() },
  ])
  .stream({
    key: 'liveUpdates',
    stream: async function* () {
      yield* subscribe('/events')
    },
    onChunk: (chunk) => updateUI(chunk),
  })
  .build({ httpConfig: { baseURL: 'https://api.example.com' } })

Builder method	Description
.step(stage)	Add a sequential stage
.parallel(stages, options?)	Add a parallel group (key/concurrency optional, see Limiting concurrency)
.subPipeline(item)	Embed a sub-pipeline as a stage
.stream(stage)	Add a stream stage (AsyncIterable)
.build(options?)	Create and return a PipelineOrchestrator
.toConfig(options?)	Return PipelineConfig without creating an orchestrator

Typed chaining (TypeScript)

In TypeScript, pipe().step(...) tracks the type of prev across the chain: each .step()'s prev is typed as the previous step's return value (undefined for the very first step, matching the orchestrator's actual runtime behavior). .parallel() / .subPipeline() / .stream() don't change it — exactly like at runtime, where prev for the next step still comes from the last regular .step(), not from a parallel group's results:

const orchestrator = pipe()
  .step({ key: 'auth', request: async (): Promise<string> => getToken() })
  .step({ key: 'fetchUser', request: async ({ prev }) => fetchUser(prev) }) // prev: string — inferred, autocompletes
  .step({ key: 'oops', request: async ({ prev }) => prev.totallyNotAMethod() }) // ✗ compile error: wrong type for prev
  .build()

This works whether or not you keep reassigning the chain (builder.step(...) without capturing the return value still mutates the same instance, just like before) — the typing is purely additive and doesn't change runtime behavior.

Schema validation (validateInput / validateOutput)

Each stage can validate (and, since the return value replaces the data, optionally coerce) its input and output with validateInput/validateOutput. Neither depends on a particular schema library — pass any function (data) => T; it should throw on invalid data:

import { createRestClient, PipelineOrchestrator } from 'rest-pipeline-js'
import { z } from 'zod' // not a dependency of this package — bring your own

const userSchema = z.object({ id: z.number(), name: z.string() })

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [
      {
        key: 'fetchUser',
        request: async ({ sharedData }) => client.get(`/users/${sharedData.userId}`),
        // Validates the response shape before it's stored as this stage's
        // result and handed to the next stage as `prev` — catches a backend
        // contract change at the pipeline boundary instead of downstream.
        validateOutput: (data) => userSchema.parse((data as { data: unknown }).data),
      },
      {
        key: 'greet',
        validateInput: (data) => userSchema.parse(data),
        request: async ({ prev }) => `Hello, ${prev.name}!`,
      },
    ],
  },
})

• validateInput runs after the before hook (sees its result), right before request — so it can validate/coerce the value request is about to receive as prev.
• validateOutput runs after the after hook (sees its result), right before the step is committed as successful — so it validates/coerces the actual value that becomes this stage's data and the next stage's prev.
• Both receive (data, { allResults, sharedData, signal }), matching the other stage hooks.
• A thrown error goes through the exact same path as any other stage error — errorHandler can inspect it and return recoverStep(fallbackValue) to recover, same as a failed request.
• Both also apply to stages inside a ParallelStageGroup (they share the same execution path as top-level stages).

See examples/zod-validation.ts for a full withZodSchema() adapter helper.

validatePipelineConfig()

Catch configuration errors before runtime:

import { validatePipelineConfig } from 'rest-pipeline-js'

const { valid, errors } = validatePipelineConfig({
  stages: [
    { key: 'step1', request: async () => data },
    { key: 'step1', request: async () => other }, // duplicate!
    { key: '', request: async () => other }, // empty key!
  ],
})

if (!valid) console.error(errors)
// ["[root] duplicate stage key: "step1"", "[root] stage key must be a non-empty string"]

Validates: duplicate keys, empty/invalid keys, empty stages array, invalid field types (request, condition, retryCount, timeoutMs), and recursively validates nested subPipeline configs.

Plugin system

Package reusable orchestrator behavior into plugins:

const loggingPlugin = {
  name: 'logging',
  install(orchestrator) {
    const off = orchestrator.on('log', (event) => {
      if (event.type === 'step:success') console.log('✓', event.stepKey)
      if (event.type === 'step:error') console.error('✗', event.stepKey, event.error)
    })
    return () => off() // cleanup on orchestrator.destroy()
  },
}

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [/* ... */],
    options: { plugins: [loggingPlugin, analyticsPlugin] },
  },
})

// Call when the orchestrator is no longer needed:
orchestrator.destroy()

• install(orchestrator) — receives the orchestrator instance; may subscribe to events, set up middleware, etc.
• If install returns a function, it is registered as a cleanup callback and invoked by destroy().

Persist adapter

Automatically save and restore pipeline state across page reloads:

const localStorageAdapter = {
  save: (state) => localStorage.setItem('pipeline', JSON.stringify(state)),
  load: () => {
    const raw = localStorage.getItem('pipeline')
    return raw ? JSON.parse(raw) : null
  },
}

const orchestrator = new PipelineOrchestrator({
  config: {
    stages: [/* ... */],
    options: { persistAdapter: localStorageAdapter },
  },
})

// run() loads saved state at start; saves after each completed step
await orchestrator.run()

The adapter interface:

type PipelineStateAdapter = {
  save(state: PipelineExportedState): void | Promise<void>
  load(): PipelineExportedState | null | Promise<PipelineExportedState | null>
}

Both methods may be async (useful for IndexedDB or remote storage).