Building Workflows
Every step option, the rules build() enforces, and the node status lifecycle.
Every step option, the rules build() enforces, and the node status lifecycle.
Workflows is the narrative walkthrough. This page is
the reference: every knob on a step, the combinations build() rejects, and the
full node status lifecycle.
A workflow is always built imperatively — there is no decorator or registry that
declares one for you. Start with queue.workflows.define(name), optionally
bump the version, then add steps:
const handle = queue.workflows
.define("etl", 2) // name + version (default 1)
.step("extract", "extractTask", { args: ["s3://bucket/in"] })
.step("transform", "transformTask", { after: "extract" })
.step("load", "loadTask", { after: "transform", maxRetries: 5 })
.submit();.submit() validates the DAG, pre-enqueues the static steps, and returns a
WorkflowHandle. Two variants:
// Build without submitting — for reuse as a sub-workflow.
const child = queue.workflows.define("child").step("a", "taskA").build();
// Submit-time options: a default queue for steps that set none, plus run params.
const builder = queue.workflows.define("etl").step("extract", "extractTask");
const run = queue.workflows.submit(builder, { queueDefault: "io", params: { batchId } });Submitting enqueues jobs; it does not execute them. Start a worker
(queue.runWorker()) or handle.wait() will block until it times out.
Every .step(name, task, options) accepts:
| Option | Type | Meaning |
|---|---|---|
after | string | string[] | Predecessor step name(s) this step waits on. Omit for a root. |
args | unknown[] | Positional arguments passed to the step's task handler. |
queue | string | Queue to run this step on (overrides the submit-time queueDefault). |
maxRetries | number | Retry budget for this step's job. |
timeoutMs | number | Per-attempt timeout, in milliseconds. |
priority | number | Queue priority for this step's job. |
condition | "on_success" | "on_failure" | "always" | When the step runs once its predecessors settle. Default on_success — see Conditions. |
compensate | string | Rollback task run with this step's result if the run later fails — see Sagas. |
cache | boolean | { ttlMs?: number } | Reuse this step's result across runs while its task, args, and upstream results are unchanged — see Incremental runs. |
The specialised step kinds take their own option shapes, each covering the
common queue / maxRetries / timeoutMs / priority set:
| Method | Options | Covered in |
|---|---|---|
fanOut(name, opts) | after, task, itemsFrom? | Fan-out / Fan-in |
fanIn(name, opts) | after, task | Fan-out / Fan-in |
gate(name, opts) | after, timeoutMs?, onTimeout?, message? | Approval gates |
subWorkflow(name, opts) | after, workflow | Sub-workflows |
chain / group / chord | CanvasStep[] + after? | Canvas |
compensate and cache live on WorkflowStepOptions only. The fan-out,
fan-in, gate, and sub-workflow option types omit them, so TypeScript rejects
the combination at compile time rather than at submit.
Adding a duplicate step name throws WorkflowError immediately. The rest is
checked by build() (and therefore by .submit()), also as WorkflowError:
after naming a step that was never added.fanIn whose after is an ordinary
step would never be triggered, so the run would hang.Steps may reference predecessors added later — the DAG is resolved and topologically ordered at build time, not as you chain.
Most steps are pre-enqueued at submit and sequenced by the core scheduler.
Steps the core cannot schedule up front are deferred: the worker-side
tracker enqueues them once their predecessors settle. A step is deferred when it
sets condition or cache, or when it is a fan-out, fan-in, gate, or
sub-workflow — and deferral propagates transitively to everything downstream of
one.
This is why workflows need a running worker even for parts of the DAG that look purely static.
| Status | Terminal | Meaning |
|---|---|---|
pending | No | Waiting on predecessors. |
ready | No | Predecessors settled; the job exists but no worker has claimed it. |
running | No | A worker claimed the job and is executing the handler. |
completed | Yes | The task succeeded. |
failed | Yes | Retries were exhausted, a gate was rejected, or a child run failed. |
skipped | Yes | The step's condition was not met, or it cascaded from a skipped/failed predecessor. |
waiting_approval | No | A gate is parked, awaiting approveGate/rejectGate or its timeout. |
cache_hit | Yes | A prior run's result was reused within the cache TTL; the task did not re-execute. |
compensating | No | The step's rollback job is in flight (saga only). |
compensated | Yes | The step's rollback completed successfully. |
compensation_failed | Yes | The rollback itself failed — this node needs operator attention. |
Read statuses from the handle or the manager:
handle.nodes(); // WorkflowNode[] for this run
queue.workflows.nodes(runId); // the same, by run id
queue.workflows.analyze(runId)?.node("transform")?.status;The run-level state (running, completed, completed_with_failures,
failed, cancelled, paused, and the three saga states) is a separate set —
see the Workflows API reference.