Fan Out Over a List

A step can return a list that is known only at run time. Use ForEach to start a second step once for each item of that list. The fan-out node collects the outputs in item order. The next step gets them as one slice.

Declare a Fan-Out Node

ForEach is a method on the step that processes one item:

var (
    discover = cyclone.NewStep("discover", discoverItems) // Request -> []Item
    process  = cyclone.NewStep("process", processItem)    // Item -> Result
    report   = cyclone.NewStep("report", writeReport)     // []Result -> Receipt
)

var Ingest = cyclone.DefineResult("ingest", func(b *cyclone.Builder, in Request) (*cyclone.Future[Receipt], error) {
    items := discover.Do(b, cyclone.Lit(in))                         // *Future[[]Item]
    results := process.ForEach(b, items, cyclone.WithMaxParallel(8)) // *Future[[]Result]
    return report.Do(b, results), nil
})

The compiler checks the types. process is a Step[Item, Result]. items must be an Input[[]Item]. The result is a *Future[[]Result]. A list of the wrong element type does not compile.

items must be the future of an upstream node, or From of it. Lit and FromAll also give a list input. They are builder errors. A []byte list is a builder error too. encoding/json encodes []byte as a string, not as a list. For a list that is known at submit, call Do once for each element in the builder.

Workers register the item step like any other step. The fan-out node has the kind of its item step. ValidateWorkflow and the submit checks require that kind.

The future of Step.ForEach is an ordinary future of the build. A DefineResult closure can return it. Then run.Result returns the item outputs in item order:

var ProcessAll = cyclone.DefineResult("process_all", func(b *cyclone.Builder, in Request) (*cyclone.Future[[]Result], error) {
    return process.ForEach(b, discover.Do(b, cyclone.Lit(in))), nil
})

run, err := ProcessAll.Submit(ctx, client, request)
if err != nil {
    return err
}
results, err := run.Result(ctx) // []Result, in item order

Action.ForEach starts an action once for each item. Its fan-out node collects no output. It returns a *Node. The node gives order only:

sent := notify.ForEach(b, recipients)
archive.Do(b, cyclone.Lit(in), cyclone.After(sent))

What Happens at Run Time

1. The fan-out node is pending until its upstream node completes.
2. The engine expands the fan-out node in the transaction that completes the upstream node. It inserts one item task for each item. An item task is an ordinary step row. Its node ID is the fan-out node ID with the item index: process[0], process[1], and so on.
3. Workers claim the item tasks and start them, as for other steps. ctx.Node() returns the item task ID. Use it in idempotency keys.
4. The engine completes the fan-out node in the transaction that completes its last item task. The output is a JSON array of the item outputs, in item order. Then the nodes after the fan-out node become available.

The fan-out node stays pending while its item tasks are in progress. No worker starts the fan-out node itself.

An empty list completes the fan-out node at once with an empty output. A nil slice is an empty list.

Durability

• The expansion commits together with the completion of the upstream node, or not at all. A crash cannot leave a part of an expansion.
• The ID of an item task comes from the fan-out node ID and the item index. The unique constraint on (run_id, node) rejects a second expansion of the same node.
• The nodes after the fan-out node become available only in the transaction that completes the last item task. They cannot start before all items are complete. In a paused run, they become available at Resume.
• SubmitOnce makes one run for each key. A second call with the same key returns that run. It does not expand the fan-out node again. See Submit a Workflow Once.

Limit Concurrency with WithMaxParallel

WithMaxParallel(n) keeps at most n item tasks of the fan-out node available or running at the same time. The limit applies across all workers of all clients. It does not apply in each process alone. When an item task finishes, the engine releases the next pending item task under the run row lock.

• An item task that waits for a retry backoff keeps its place in the limit.
• A paused item task keeps its place in the limit. See Pause.
• With no WithMaxParallel, all item tasks are available at the same time. The number of workers then limits the parallelism.
• n must be 1 or more. A smaller n is a builder error.
• The limit does not change the claim query. It adds no cost to a step that is not an item task.

Item tasks use the queue and priority of the item step. Give the item step its own WithQueue to keep a large fan-out apart from other work.

Options

An item task uses the options of its step. The options are WithMaxAttempts, WithBackoff, WithTimeout, WithQueue, WithPriority, and WithRetention. The item task also uses the kind version from WithVersion.

ForEach also accepts ID, After, and OnDepFailure for the fan-out node. WithMaxWait applies to wait nodes only, and ForEach rejects it.

results := process.ForEach(b, items,
    cyclone.ID("process-items"),
    cyclone.WithMaxParallel(4),
    cyclone.OnDepFailure(cyclone.CancelBranch),
)

With this ID the item tasks are process-items[0], process-items[1], and so on. No other node of the workflow can have an ID of that form.

Failures

All items must succeed:

• When the engine discards an item task, it discards the fan-out node in the same transaction. The error of the fan-out node names the item task. For example, the text is item task process[3] failed: invalid item. Its error kind is the error kind of the item task.
• The engine cancels the item tasks that did not finish. A handler in progress sees its context cancelled. The engine ignores a late completion.
• Then OnDepFailure applies, as for any discarded node. By default the run fails. With OnDepFailure(CancelBranch) on every node after the fan-out node, the failure cancels only that branch. See Per-Branch Failure Policy.
• RunStatus.FailedNode is the fan-out node. RunStatus.Error names the item task.
• When the run fails, run.Result and run.Output return an error that matches ErrRunNotCompleted. The error names the fan-out node and the item task.
• The WithErrorHandler hook fires for the failed item task. The fan-out node does not report the same failure again.

OnDepFailure on the fan-out node itself applies when its upstream node fails. Then the engine cancels the fan-out node before it expands.

This version has no partial success. To continue when some items fail, put the failure in the item output. For example, return a result with an error field. Return no error from the handler.

Limits

Limit	Value	Result
Items for each expansion	MaxFanOutItems (10,000)	The engine discards the fan-out node with error kind permanent. The hook error matches ErrTooManyItems.
Collected output	The OutputBytes payload limit of the client that submitted the run (default 1 MiB). A disabled limit applies no bound	The engine discards the fan-out node with error kind permanent. The hook error wraps a *PayloadTooLargeError.
Upstream output	Must be a JSON array or null	The engine discards the fan-out node with error kind permanent.

For these failures the engine emits EventDiscarded for the fan-out node. The WithErrorHandler hook receives the error value. The error of the discarded node records its message.

For more items, let the upstream step return batches, for example [][]Item. Let each item task process one batch. Each item input is one element of the upstream output. The OutputBytes limit of the upstream step already limits that input. See Payload Limits.

Retry

• RetryAll reopens the fan-out node, its failed item task, and the item tasks that the failure cancelled. Completed item tasks keep their outputs. They do not start again.
• Retry with the fan-out node ID reopens the fan-out node and its failed item tasks together. Retry with the ID of one failed item task does the same.
• A fan-out node that failed for a limit expands or collects again after a retry. The same list gives the same failure. Retry returns no error. The run fails again in the same transaction.

Cancel, Deadlines, Timeouts, and Leases

• Cancel cancels the fan-out node and all its item tasks. A handler in progress sees its context cancelled.
• When a WithRunDeadline elapses, the engine discards the available and running item tasks with error kind deadline. The engine cancels the fan-out node and the pending item tasks. The run fails. RetryAll starts the remaining items.
• WithTimeout on the item step applies to each item task. A timed-out item task retries like any step.
• Each item task has its own lease and heartbeat. The recovery sweep reclaims an item task whose worker crashed. WithMaxReclaims applies to each item task.

Pause

run.Pause holds the item tasks of the run:

• The available item tasks become paused. A running item task continues.
• The engine releases no pending item task until Resume.
• An expansion during the pause gives the paused state to the item tasks that the limit permits.
• A paused item task keeps its place in the WithMaxParallel limit. Resume makes it available again.
• A discarded item task fails the fan-out node during the pause, as with no pause. The engine also cancels the paused item tasks.

See Pause and Resume Runs.

Inspect a Fan-Out

• client.ListSteps and run.Steps put the item tasks after their fan-out node, in item order. StepStatus.FanOutNode and StepStatus.ItemIndex identify an item task.
• cyclone steps <runID> shows the progress of a fan-out node in the ITEMS column. The column shows completed item tasks over all item tasks. See CLI.
• cyclone.OutputNode reads the collected output of a completed fan-out node.

Item tasks are steps of the run. The retention sweeper and PurgeTerminalRuns delete them with their run. An item task keeps the WithRetention of its step.

Testing

cyclonetest starts a fan-out like any workflow. AdvanceTime makes an item task that waits for a retry due. RequireStepState accepts an item task ID:

env.RequireStepState(runID, "process[3]", cyclone.StepCompleted)

When to Use Something Else

• For a list that is known at submit, call Do once for each element in the builder.
• When one item needs more than one step, use a child workflow. ForEach starts one step for each item. See Run a Child Workflow.
• For standalone jobs that are not part of a run, use EnqueueMany. See Enqueue Many Jobs.

Deployment

Apply the fan-out migration before you deploy a binary of this version. A binary that does not use ForEach also needs the migration. The engine reads the new columns in every run. Upgrade every worker before you submit a workflow that uses ForEach. An older worker does not know a fan-out node. See Migrations.