ALLM.Pipeline.FanOut (allm_pipeline v0.1.0)

Copy Markdown View Source

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:

  1. An {:exit, _} clause in the consumer is dead code. Neither an Enum.reduce clause nor an Enum.map head 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.
  2. The fix belongs in the CHILD. Wrap the per-item work in catch kind, reason so it never dies at all, and degrade to whatever per-item failure value the caller already handles. rescue alone is not sufficient — an exit is not an exception, so rescue never sees it, and GenServer.call timeouts (Playwright, the browser manager) and Task deaths 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:

SiteHow it is kept safe
ALLM.Pipeline.Dsl.Runtime.run_concurrent/7always-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 fold accumulator threaded through reduce/5.

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

acc()

@type acc() :: term()

The fold accumulator threaded through reduce/5.

item_result()

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

parent_mode()

@type parent_mode() :: :source_stage | :per_item

How each item's steps are parented — see reduce/5.

uncaught()

@type uncaught() :: {:uncaught, kind :: atom(), reason :: term()}

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

guard(label, fun, on_uncaught)

@spec guard(String.t(), (-> result), (uncaught() -> result)) :: result
when result: var

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

item_parent(arg1, item, source_parent)

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

reduce(ctx, items, acc, fun, opts \\ [])

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 fun call runs under an always-on catch kind, reason wrapper whose job here is per-item failure isolation: an uncaught raise/exit/throw becomes %Item{result: {:error, {:uncaught, kind, reason}}} and leaves acc unchanged for that item, so the fold degrades one item and keeps folding rather than aborting the run. This is reduce/5's own reason — a sequential fold has no link hazard (the moduledoc's ProjectRefreshPipeline sibling makes the same point) — and it also keeps reduce/5 behaviourally equivalent to the concurrent fan-out's always-on catch. It is catch, never rescue: an exit is not an exception, and Playwright/GenServer teardowns arrive as exits. (Link safety proper applies only to the Task.async_stream path in Dsl.Runtime.run_concurrent/7.)
  • fun's item_result is wrapped into %Item{}: a 3-tuple {:ok, value, %StepLog{}} carries the producing step log onto %Item{step_log: …}; every other shape leaves it nil.
  • reduce/5 is sequential only — it folds a changing accumulator, and a concurrent fold has undefined order (Dsl.Runtime's assert_no_concurrent_fold!/2 exists 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 off Context.input_step_id(ctx); :per_item re-parents each item's context to its OWN producing step log, which requires the items to be %Item{} wrappers (filter a previous fan-out with ALLM.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.

tag_uncaught(label, kind, reason, stacktrace)

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

to_item(item, result)

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