Fan-Out & Fan-In

Split a step's result into parallel child jobs, then collect all results into a downstream step. This is the closest analog to Celery's chord (a group of parallel tasks followed by a callback), but the fan-out size is derived at runtime from the predecessor's return value instead of being fixed when the group is built.

Fan-out with "each"

The predecessor's return value must be iterable. Each element becomes a separate child job:

@queue.task()
def fetch() -> list[int]:
    return [10, 20, 30]

@queue.task()
def process(item: int) -> int:
    return item * 2

@queue.task()
def aggregate(results: list[int]) -> int:
    return sum(results)  # receives [20, 40, 60]

wf = Workflow(name="map_reduce")
wf.step("fetch", fetch)
wf.step("process", process, after="fetch", fan_out="each")
wf.step("aggregate", aggregate, after="process", fan_in="all")

Child nodes are named process[0], process[1], process[2] and appear in status queries.

How it works

1. fetch completes — the tracker reads its return value
2. The tracker splits the returned list into individual items, one per child
3. It creates N child nodes and N jobs, all ready to run immediately (no depends_on wait)
4. Each child runs independently in parallel
5. When every child has completed, the tracker marks the fan-out parent done
6. The tracker collects all child results in index order
7. The fan-in job is created with ((results_list,), {}) as its payload

Empty fan-out

If the predecessor returns an empty list, the fan-out parent is marked COMPLETED immediately with zero children, and the fan-in receives an empty list:

@queue.task()
def fetch() -> list:
    return []  # nothing to process

# aggregate receives []

Fan-out with downstream steps

Steps after the fan-in work normally:

wf = Workflow(name="full_pipeline")
wf.step("fetch", fetch)
wf.step("process", process, after="fetch", fan_out="each")
wf.step("aggregate", aggregate, after="process", fan_in="all")
wf.step("report", send_report, after="aggregate")  # runs after aggregate

Failure handling

By default (on_failure="fail_fast"), if any fan-out child fails:

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

Combine with conditions for more control:

wf.step("handle_error", alert, after="process", condition="on_failure")