Baton.Expansion (Baton v0.27.4)

Copy Markdown View Source

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:

  1. Expand. Insert one child job per item, plus their workflow_nodes rows (fan_out_of = this step, item_index = position). Then append the children to this step's own deps and return {:snooze, _}.
  2. Finish. The snoozed expander is woken by Baton.Reschedule when the last child settles — the ordinary promotion path, because the children are now genuinely its dependencies. Baton.Check has already confirmed every child is terminal, so finish/1 only 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: true proceeds 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.

What expand/3 did. :empty means the collection held no items.

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

child()

@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.

outcome()

@type outcome() :: {:expanded, non_neg_integer()} | :already_expanded | :empty

What expand/3 did. :empty means the collection held no items.

Functions

expand(job, children, opts \\ [])

@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

expanded?(job)

@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.

finish(job)

@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.

manifest?(arg1)

@spec manifest?(term()) :: boolean()

Whether a step result is an expansion manifest written by finish/1.

marker()

@spec marker() :: String.t()

The key marking a step result as a fan-out manifest rather than ordinary data.