Workflows

The workflow builder, run handle, and queries.

queue.workflows
  .define(name: string)
  // ... steps ...
  .submit(): WorkflowHandle;

Builder steps

Method	Description
`step(name, task, opts?)`	A task node. `opts`: `after`, `args`, `queue`, `maxRetries`, `timeoutMs`, `priority`, `condition`, `compensate`, `cache`.
`fanOut(name, { after, task, itemsFrom })`	Expand into one child per item.
`fanIn(name, { after, task })`	Collect children's results into an array.
`gate(name, { after, timeoutMs?, onTimeout?, message? })`	Pause for approval.
`subWorkflow(name, { after, workflow })`	Run a child workflow as a node.
`chain(steps[], { after? })`	Canvas: wire `steps` sequentially.
`group(steps[], { after? })`	Canvas: run `steps` in parallel.
`chord(steps[], callback, { after? })`	Canvas: a parallel group joined by `callback`.
`submit()`	Pre-enqueue the DAG → `WorkflowHandle`.
`build()`	Build (don't submit) — for use as a sub-workflow.

Canvas steps are { name, task, ...stepOptions } (after is managed by the helper).

condition: "on_success" (default) · "on_failure" · "always".

The chained submit() takes no arguments. To set submit-time queueDefault/params, stop before the final .submit() and pass the builder to queue.workflows.submit(builder, { queueDefault?, params? }) instead:

const workflow = queue.workflows.define("etl").step("extract", "extract");
const run = queue.workflows.submit(workflow, { queueDefault: "io", params: { runId } });

`WorkflowHandle`

Member	Description
`runId`	The run's id.
`status()`	The current `WorkflowRun` snapshot, or `undefined`.
`wait()`	Resolves with the `WorkflowRun` when terminal.
`nodes()`	Per-step status (fan-out parents carry `fanOutCount`).

Queries

queue.workflows.list({ state: "running" });
queue.workflows.run(runId); // a single run snapshot
queue.workflows.nodes(runId); // per-step status
queue.workflows.dag(runId); // the DAG graph as a JSON string (nodes + edges); JSON.parse it, or use analyze()
queue.workflows.children(runId); // spawned sub-workflow runs
queue.workflows.approveGate(runId, "review");
queue.workflows.rejectGate(runId, "review", reason);
queue.workflows.resolveGate(runId, "review", approved);
queue.workflows.clearCache(); // drop all cached step results

Analysis

analyze(runId) returns a WorkflowAnalysis over the run's DAG + node statuses (or undefined if the run is unknown), for graph introspection:

const a = queue.workflows.analyze(runId);
a?.node("load"); // one node's status record, or undefined
a?.roots(); // entry nodes
a?.leaves(); // exit nodes
a?.ancestors("load"); // transitive upstream deps
a?.descendants("extract"); // transitive downstream
a?.topologicalOrder(); // a valid run order (throws on a cycle)
a?.topologicalLevels(); // nodes grouped by dependency depth
a?.criticalPath(); // longest dependency chain (root → leaf)
a?.stats(); // { total, byStatus, completed, failed, running, pending }

See the Workflows guides for fan-out, conditions, gates, sub-workflows, and saga compensation. Requires the addon built with the workflows cargo feature.

Builder steps

Method	Description
`step(name, task, opts?)`	A task node. `opts`: `after`, `args`, `queue`, `maxRetries`, `timeoutMs`, `priority`, `condition`, `compensate`, `cache`.
`fanOut(name, { after, task, itemsFrom })`	Expand into one child per item.
`fanIn(name, { after, task })`	Collect children's results into an array.
`gate(name, { after, timeoutMs?, onTimeout?, message? })`	Pause for approval.
`subWorkflow(name, { after, workflow })`	Run a child workflow as a node.
`chain(steps[], { after? })`	Canvas: wire `steps` sequentially.
`group(steps[], { after? })`	Canvas: run `steps` in parallel.
`chord(steps[], callback, { after? })`	Canvas: a parallel group joined by `callback`.
`submit()`	Pre-enqueue the DAG → `WorkflowHandle`.
`build()`	Build (don't submit) — for use as a sub-workflow.

Canvas steps are { name, task, ...stepOptions } (after is managed by the helper).

condition: "on_success" (default) · "on_failure" · "always".

const workflow = queue.workflows.define("etl").step("extract", "extract");
const run = queue.workflows.submit(workflow, { queueDefault: "io", params: { runId } });

Member

Description

runId

The run's id.

status()

The current WorkflowRun snapshot, or undefined.

wait()

Resolves with the WorkflowRun when terminal.

nodes()

Per-step status (fan-out parents carry fanOutCount).

Queries

queue.workflows.list({ state: "running" });
queue.workflows.run(runId); // a single run snapshot
queue.workflows.nodes(runId); // per-step status
queue.workflows.dag(runId); // the DAG graph as a JSON string (nodes + edges); JSON.parse it, or use analyze()
queue.workflows.children(runId); // spawned sub-workflow runs
queue.workflows.approveGate(runId, "review");
queue.workflows.rejectGate(runId, "review", reason);
queue.workflows.resolveGate(runId, "review", approved);
queue.workflows.clearCache(); // drop all cached step results

Analysis

analyze(runId) returns a WorkflowAnalysis over the run's DAG + node statuses (or undefined if the run is unknown), for graph introspection:

const a = queue.workflows.analyze(runId);
a?.node("load"); // one node's status record, or undefined
a?.roots(); // entry nodes
a?.leaves(); // exit nodes
a?.ancestors("load"); // transitive upstream deps
a?.descendants("extract"); // transitive downstream
a?.topologicalOrder(); // a valid run order (throws on a cycle)
a?.topologicalLevels(); // nodes grouped by dependency depth
a?.criticalPath(); // longest dependency chain (root → leaf)
a?.stats(); // { total, byStatus, completed, failed, running, pending }

See the Workflows guides for fan-out, conditions, gates, sub-workflows, and saga compensation. Requires the addon built with the workflows cargo feature.