Fan-Out & Fan-In

Split a step's result into parallel child jobs, then collect all results into a downstream step.

Basic fan-out and fan-in

Use .fanOut() to expand a predecessor's array result into one child job per item, and .fanIn() to collect the children's results into a combiner task.

queue.task("listFiles", () => ["a.csv", "b.csv", "c.csv"]); // returns the items array
queue.task("processFile", (file: string) => file.length);   // runs once per item
queue.task("summarize", (sizes: number[]) => sizes.reduce((a, b) => a + b, 0));

const handle = queue.workflows
  .define("batch")
  .step("list", "listFiles")
  .fanOut("process", { after: "list", task: "processFile", itemsFrom: "list" })
  .fanIn("collect", { after: "process", task: "summarize" })
  .submit();

queue.runWorker(); // advances workflow nodes by default

const run = await handle.wait();
console.log(run.state); // "completed"

Coming from BullMQ? FlowProducer builds a static parent/child tree — children is a fixed array at definition time. flexiq's fan-out is dynamic: the child count comes from list's return value at run time, so a batch of 3 files and a batch of 3,000 use the same workflow definition.

itemsFrom names the predecessor whose result is the item array. It defaults to the sole predecessor when omitted. Each item is passed to the child task as its single positional argument.

Child nodes are named process[0], process[1], process[2] and appear in status queries. The parent node carries fanOutCount (the number of children spawned).

const nodes = handle.nodes();
// nodes.find(n => n.nodeName === "process")?.fanOutCount === 3
// nodes.find(n => n.nodeName === "process[0]")?.status  === "completed"

How it works

1. list completes — the tracker reads its return value (must be an array)
2. The tracker calls expandFanOut, creating N child nodes and N jobs — each receives one item from the array; children are ready immediately (no depends_on)
3. Children execute in parallel; their statuses surface as process[0], process[1], …
4. As each child settles, the tracker calls checkFanOutCompletion
5. When all children complete, the tracker collects results in index order and calls createDeferredJob for the fan-in step
6. The collect fan-in node runs summarize with the results array as its single argument

The WorkflowTracker is driven entirely by the worker outcome stream and reconstructs the run plan from storage on each event, so submission and execution may run in different processes — no submit-time coordinator is needed.

Empty fan-out

If the predecessor returns an empty array, the fan-out parent is marked completed immediately with fanOutCount 0 and the fan-in runs with []:

queue.task("listFiles", () => []);     // nothing to process
queue.task("summarize", (sizes: number[]) => sizes.reduce((a, b) => a + b, 0));

const handle = queue.workflows
  .define("batch")
  .step("list", "listFiles")
  .fanOut("process", { after: "list", task: "processFile" })
  .fanIn("collect", { after: "process", task: "summarize" })
  .submit();

// summarize receives [] → result is 0

Downstream steps

Steps after the fan-in work normally — declare them with after pointing to the fan-in node:

queue.task("notify", (total: number) => sendSlack(`Processed ${total} bytes`));

const handle = queue.workflows
  .define("full-pipeline")
  .step("list", "listFiles")
  .fanOut("process", { after: "list", task: "processFile", itemsFrom: "list" })
  .fanIn("collect", { after: "process", task: "summarize" })
  .step("notify", "notify", { after: "collect" }) // runs after summarize
  .submit();

Per-step options

.fanOut() accepts the same per-step options as .step():

Children and the combiner each inherit the fan-out step's queue, maxRetries, timeoutMs, and priority unless overridden.

Failure handling

Fan-out is fail-fast: if any child dead-letters:

• Remaining pending children are cancelled
• The fan-out parent is marked failed
• The fan-in and downstream steps are skipped
• The workflow run transitions to failed

const run = await handle.wait();
if (run.state === "failed") {
  const failed = handle.nodes().filter(n => n.status === "failed");
  console.error("failed nodes:", failed.map(n => n.nodeName));
}