`orchestrator/step-runner`

Purpose

runStep(...) is the orchestrator's per-step engine. It bundles every cross-cutting concern that fires around an action handler invocation — in_flight transition, saga.step.requested emission, request-context minting, idempotency wrap, optional timeout, result persistence, saga.step.completed / saga.step.failed emission, and step-duration histogram recording — into a single async call. The orchestrator's main loop is therefore short enough to fit comfortably inside the ~250-line soft limit while every step-level concern lives here.

The runner intentionally returns a structured outcome rather than throwing on action-handler failure. The caller wants to capture the error and trigger compensation; an exception path would force the orchestrator into try/catch noise around every step invocation.

Features

Single async entry point — wraps every cross-cutting concern.
Idempotency wrap via SagaIdempotencyWrapper; honours the participant's idempotency: "manual" opt-out.
Optional per-action timeoutMs — fires SagaError(SAGA.STEP_TIMEOUT) when the handler exceeds the budget.
Step-duration histogram emitted on both success and failure paths.
Persistence updates funnel through SagaPersistence's state-machine-guarded transitions.

Usage: Basic

The runner is internal — most callers consume it indirectly via createSagaOrchestrator. A consumer that wants to drive one step manually (e.g. a CLI that replays a stuck step) can invoke it directly:

import { runStep } from "@suaveplan/api-saga/orchestrator/step-runner";

const outcome = await runStep({
    instance,
    stepRow,
    registration,
    principal: null,
    persistence,
    emitter,
    idempotency,
    instruments,
    requestContextFactory,
    nowMs: () => Date.now(),
});
if (outcome.outcome === "completed") {
    console.log(outcome.output);
}

Usage: Advanced

A test that wants to assert the step-runner's behaviour against a fake participant + in-memory persistence + capturing event bus can inject the runner directly without standing up a full orchestrator:

import { runStep } from "@suaveplan/api-saga";

const outcome = await runStep({
    instance,
    stepRow,
    registration: {
        plugin: "p",
        action: "a",
        handler: async () => ({ ok: true }),
        idempotency: "manual",
    },
    principal: null,
    persistence,
    emitter,
    idempotency,
    instruments,
    requestContextFactory: () => testCtx,
    nowMs: () => 1000,
});

API Reference

runStep(args): Promise<StepRunOutcome> — single entry point.
args:
- instance: SagaInstanceRow — the saga we're stepping.
- stepRow: SagaStepRow — the persisted step.
- registration: SagaParticipantRegistration — bound handlers.
- principal: Principal | null — the saga initiator.
- persistence, emitter, idempotency, instruments, requestContextFactory, nowMs — injected dependencies.
StepRunOutcome — { outcome: "completed", output } | { outcome: "failed", code, message, cause }.

Implementation Notes

The optional timeout wrap uses raw setTimeout + a settled guard rather than Promise.race because Promise.race against a slow handler leaks the handler — the handler keeps running in the background even after the race resolves. The settled pattern clears the timer once the handler's promise resolves, AND ignores late handler resolution once the timeout has fired. The handler itself still completes (we can't cancel external work) but the orchestrator no longer waits on it.

The runner emits a step_duration_ms histogram entry on BOTH the success and failure paths so the operator can dashboard "failed step duration" — without the failed-path entry the histogram would implicitly only describe successful steps.

3.8 KiB Raw Permalink Blame History

orchestrator/step-runner