Mid-run fan-out: one running step turns a list it could not know at build time into one job per item, then waits for them.
A static fan-out (Baton.Flow.FanOutSpec over $input/$context) knows N
before anything runs, so Baton.Flow.Compiler emits N jobs in the insert
transaction. When the collection is an upstream result — a claim list an
LLM just produced — N is only knowable once that step has finished. This
module is how the step that knows it creates the rest of the graph.
The expander is its own fan-in
The step is inserted once, as an expander, holding the logical step name and the node's declared deps. It runs twice:
- Expand. Insert one child job per item, plus their
workflow_nodesrows (fan_out_of= this step,item_index= position). Then append the children to this step's owndepsand return{:snooze, _}. - Finish. The snoozed expander is woken by
Baton.Reschedulewhen the last child settles — the ordinary promotion path, because the children are now genuinely its dependencies.Baton.Checkhas already confirmed every child is terminal, sofinish/1only has to record the manifest and decide the outcome.
Rewriting its own row rather than its readers' is what makes this safe.
A reader's dependencies never change under it, so there is no window in
which a reader observes the expander as complete while the children it must
wait for are missing from its dep list. And because the expander sits in
scheduled for the whole expansion, Baton.Completion counts it as pending
— a workflow can never be announced finished with children still running.
Nothing downstream needs to know any of this happened. A reader declares a
dep on the logical step exactly as it would for a static fan-out, and
the flow runtime resolves $steps.<logical> to the ordered child results.
Idempotency
"Has this already expanded?" is answered by the children themselves — the
existence of any workflow_nodes row with fan_out_of set to this step —
never by scratch state that could be written separately from the rows it
describes. The whole expansion (child jobs, child rows, the dep rewrite)
commits in one transaction, so that question has only two answers and both
are correct:
- Crash before commit — no children, so a retry expands from scratch.
- Crash after commit — children exist, so a retry skips straight to waiting for them.
The transaction additionally takes FOR UPDATE on the expander's node row
before checking, so two runners of the same step (Oban rescuing a job whose
original is still alive) serialize rather than double-expanding.
Outcome
finish/1 mirrors what a reader would have seen had the fan-out been
static, so that ignore_discarded means the same thing either way:
- every child completed — the expander completes.
- any child did not — the expander stores its manifest and discards.
A reader that set
ignore_discarded: trueproceeds and reads the partial list, exactly as it would with a directly-depended-on child that exhausted its retries; a reader that did not, cancels.
Because the expander must survive its children to make that decision itself,
the expansion transaction also sets ignore_cancelled/ignore_discarded on
the expander's own row. That loosening applies only from the moment the
children are attached — the node's declared deps were already satisfied to
get here, so nothing the author asked to be strict about is weakened.
Summary
Types
One expansion to create: its step name, the item it stands for, and its job.
Functions
Create children as expansions of this step, in one transaction.
Whether this step has already expanded — i.e. whether any node row in the
workflow names it as its fan_out_of.
Record the expansion and report its outcome.
Whether a step result is an expansion manifest written by finish/1.
The key marking a step result as a fan-out manifest rather than ordinary data.
Types
@type child() :: %{name: String.t(), item: term(), changeset: Ecto.Changeset.t()}
One expansion to create: its step name, the item it stands for, and its job.
@type outcome() :: {:expanded, non_neg_integer()} | :already_expanded | :empty
What expand/3 did. :empty means the collection held no items.
Functions
@spec expand(Oban.Job.t(), [child()], keyword()) :: {:ok, outcome()} | {:error, term()}
Create children as expansions of this step, in one transaction.
Each child is inserted available — its dependencies are the expander's
own, which were satisfied before the expander ran — carrying the node's
declared deps, flags, and (under the :sequential gate) an ordering edge to
its predecessor. The expander's own row picks the children up as
dependencies, which is what parks it until they settle.
Returns {:ok, outcome}; the caller should snooze on {:expanded, _} and
fall through to finish/1 on :empty.
Options
:gate—:parallel(default) or:sequential, asBaton.Flow.FanOut.
@spec expanded?(Oban.Job.t()) :: boolean()
Whether this step has already expanded — i.e. whether any node row in the
workflow names it as its fan_out_of.
This is the only "have I run before?" signal the expander has, and it is deliberately derived from the rows themselves rather than from a separate marker: a marker can be written when the rows were not.
@spec finish(Oban.Job.t()) :: {:ok, map()} | {:discard, term()} | {:error, term()}
Record the expansion and report its outcome.
Called once every child is terminal (Baton.Check guarantees this before
perform_workflow/1 runs again). Returns {:ok, manifest} when every child
completed, and {:discard, {:fan_out_children_failed, names}} otherwise —
having stored the manifest first, so a reader that tolerates a dead
expansion can still resolve the partial list.
Whether a step result is an expansion manifest written by finish/1.
@spec marker() :: String.t()
The key marking a step result as a fan-out manifest rather than ordinary data.