The canonical home for the fan-out safety rule, plus the one helper
every Task.async_stream site shares.
The rule (measured, do not re-derive)
Task.async_stream links its children. With trap_exit off — every
process in this repo; grep -rn "trap_exit" lib/ finds no Process.flag/2
call — a child that raises or exits kills the calling process before the
stream can emit anything for that element. With trap_exit on it emits
{:exit, reason} normally.
Two consequences, and they point in opposite directions:
- An
{:exit, _}clause in the consumer is dead code. Neither anEnum.reduceclause nor anEnum.maphead can run for a dead child, because the caller is already gone. Adding one is not a fix; it reads as one, which is worse. Consumers may therefore unpack{:ok, result}exhaustively, provided (2) holds. - The fix belongs in the CHILD. Wrap the per-item work in
catch kind, reasonso it never dies at all, and degrade to whatever per-item failure value the caller already handles.rescuealone is not sufficient — an exit is not an exception, sorescuenever sees it, andGenServer.calltimeouts (Playwright, the browser manager) andTaskdeaths surface as exits.
Measured 2026-08-13 (n = 8 probes, Elixir 1.17.3/OTP 27) and independently by
a consumer's rescale service. Scope
is the mechanism, not any particular call site. It does not license
removing an existing catch — that is what keeps the child alive.
Sites
Every Task.async_stream in this repo either fans out work that is total
by construction, or wraps its per-item work in a catch:
| Site | How it is kept safe |
|---|---|
ALLM.Pipeline.Dsl.Runtime.run_concurrent/7 | always-on catch via guarded_item/6 — the concurrent path is unconditionally wrapped, which is link safety (the sequential path calls run_item/6 directly and is NOT wrapped) |
The rule spans repos, and each repo's table covers its own tree. A
consumer's own fan-out sites are tabled and machine-guarded in that
consumer's repo by its host twin of this repo's guard — a
FrameworkBoundaryGuardsTest-shaped "fan-out site census". A new consumer
repo that fans out owes itself the same pair: a Sites table beside its code
and a census test pinning it.
(One ported committee pipeline carried three rows here until
Phase 4.4 ported it onto use ALLM.Pipeline — its detail, transform and load
fan-outs are now the framework's single site above, a declared behaviour
change recorded in that pipeline's own moduledoc.)
This table's MEMBERSHIP is machine-guarded by
test/allm/pipeline/fan_out_test.exs, which scans this repo's lib/ and
fails by name when the site set changes. What it cannot check is the
right-hand column — how each site is kept safe — so a new fan-out still has
to add its own row by hand.
Summary
Types
The value a per-item body returns, before it is wrapped into an Item. A
3-tuple {:ok, value, %StepLog{}} nominates a producing step log; every other
shape leaves Item.step_log nil.
How each item's steps are parented — see reduce/5.
A per-item failure that escaped every named error path. Persisted through the
caller's own error shape; inspected to a string before it reaches jsonb.
Functions
Run a per-item body closure under the always-on link-safe catch.
The lineage parent for one item of a fan-out.
Fold a body over a list of items, sequentially, with per-item failure isolation.
Log an uncaught per-item failure with its stacktrace and return a tagged tuple.
Wrap a per-item body's item_result into an %Item{}.
Types
@type acc() :: term()
The fold accumulator threaded through reduce/5.
@type item_result() :: ALLM.Pipeline.Dsl.Item.result() | {:ok, term(), ALLM.Pipeline.StepLog.t()}
The value a per-item body returns, before it is wrapped into an Item. A
3-tuple {:ok, value, %StepLog{}} nominates a producing step log; every other
shape leaves Item.step_log nil.
@type parent_mode() :: :source_stage | :per_item
How each item's steps are parented — see reduce/5.
A per-item failure that escaped every named error path. Persisted through the
caller's own error shape; inspected to a string before it reaches jsonb.
Functions
Run a per-item body closure under the always-on link-safe catch.
catch kind, reason, never rescue: an exit is not an exception, and
Task.async_stream links its children, so an uncaught exit kills the caller.
on_uncaught builds the caller's own fallback pair from the tagged failure,
because the two consumers — reduce/5 here and
Dsl.Runtime.guarded_item/7 — degrade to different shapes. This is the single
home of the catch: writing it a second time is how the two paths silently
diverge on one kind (package CLAUDE.md §7).
@spec item_parent(parent_mode(), term(), Ecto.UUID.t() | nil) :: Ecto.UUID.t() | nil
The lineage parent for one item of a fan-out.
Under :source_stage, source_parent — every item's steps hang off the
source stage's step log. Under :per_item, the item's OWN producing step log,
which requires the item to be an %Item{} wrapper (ok_items/1 keeps it),
and raises otherwise.
@spec reduce( ALLM.Pipeline.Context.t(), Enumerable.t(), acc(), (ALLM.Pipeline.Context.t(), term(), acc() -> {item_result(), acc()}), keyword() ) :: {[ALLM.Pipeline.Dsl.Item.t()], acc()}
Fold a body over a list of items, sequentially, with per-item failure isolation.
This is the FUNCTION form of a body:-mode fan_out: an ordinary stage
body calls it instead of declaring the mode. It owns exactly the properties
§8.3 of the DSL review found genuinely need centralising — the always-on
per-item catch (failure isolation, not link safety — reduce/5 is
sequential; see the bullet below), the %Item{} wrapping with its
producing-step-log capture,
and per-item lineage — and pushes section:/delay:/over: back to the
caller as ordinary code.
Runs fun.(ctx, item, acc) for each item, threading acc sequentially
(a fold), and returns the list of %Item{} in input order plus the final
accumulator.
- Each
funcall runs under an always-oncatch kind, reasonwrapper whose job here is per-item failure isolation: an uncaughtraise/exit/throwbecomes%Item{result: {:error, {:uncaught, kind, reason}}}and leavesaccunchanged for that item, so the fold degrades one item and keeps folding rather than aborting the run. This isreduce/5's own reason — a sequential fold has no link hazard (the moduledoc'sProjectRefreshPipelinesibling makes the same point) — and it also keepsreduce/5behaviourally equivalent to the concurrent fan-out's always-on catch. It iscatch, neverrescue: an exit is not an exception, and Playwright/GenServerteardowns arrive as exits. (Link safety proper applies only to theTask.async_streampath inDsl.Runtime.run_concurrent/7.) fun'sitem_resultis wrapped into%Item{}: a 3-tuple{:ok, value, %StepLog{}}carries the producing step log onto%Item{step_log: …}; every other shape leaves itnil.reduce/5is sequential only — it folds a changing accumulator, and a concurrent fold has undefined order (Dsl.Runtime'sassert_no_concurrent_fold!/2exists to reject that very shape). A concurrency-capable non-folding sibling is deliberately not provided (Phase 4.5 Alternatives).
Options
parent:—:source_stage(default) hangs every item's steps offContext.input_step_id(ctx);:per_itemre-parents each item's context to its OWN producing step log, which requires the items to be%Item{}wrappers (filter a previous fan-out withALLM.Pipeline.Dsl.Item.ok_items/1), and raises otherwise.
section: and delay: are not options here — the caller inlines them
(Executor.log_section/3 and Process.sleep/1). That split is the whole
point: it keeps this function from re-accepting the costume Phase 4.5 removed.
@spec tag_uncaught(String.t(), atom(), term(), Exception.stacktrace()) :: uncaught()
Log an uncaught per-item failure with its stacktrace and return a tagged tuple.
Call from inside a catch kind, reason -> clause, where __STACKTRACE__ is
available. These handlers exist for failures nobody anticipated, so the
stacktrace is the whole value — the fact of the failure is already implied by
the degraded result.
label should identify the item (a URL, an index, a record id), not the
module — the message reads "<label> aborted with <kind> <reason>".
@spec to_item(term(), item_result()) :: ALLM.Pipeline.Dsl.Item.t()
Wrap a per-item body's item_result into an %Item{}.
A 3-tuple {:ok, value, %StepLog{}} moves its log onto step_log and keeps
{:ok, value} as the result; every other shape leaves step_log nil (the
two-element {:ok, value} form nominates no new parent — Phase 4 D2).