The reactive propagation loop — the heart of the substrate, shared by both hosts.
Given a compiled Plan and a host config (recompute strategy + key_rule),
drain the frontier to empty:
- Pick the dirty cell with the smallest depth (
Frontier.next_cell) — no cell recomputes while an input is still dirty (topological order, no external scheduler). - Atomically claim its dirty keys (
Frontier.claim— delete-returning). - Recompute via the host's
RecomputeStrategy→ the keys that changed. - Propagate: mark the changed keys on the cell's parents, applying the
host's
KeyRule(Graph.dirty_parents). - Repeat until empty.
A leaf carries no recompute — a source writes its tuples and marks parents dirty directly, so a leaf shouldn't appear in the frontier; if one does, its claimed keys are treated as changed and just propagated.
Returns {:ok, %ReactiveDag.Drain.Report{}} — the processing trace: one
step per cell recompute (cell, pass, claimed, changed, triggered_by,
duration_us), in execution order, plus run totals. The drain knows all of
this as it works; the report is that knowledge kept instead of discarded.
Persistence is the host's (an Oban job's meta, a run table) — the library
reports, the host records.
Concurrency
The per-cell claim is atomic (a DELETE … RETURNING): a dirty KEY is
consumed exactly once. But the pick-then-claim PAIR is not serialized — two
concurrent drains over the same graph can select the same cell and both
recompute it (each claiming a disjoint slice of its keys). Run ONE drain at a
time per graph (both hosts do: a single worker), or make recomputes
idempotent so a doubled recompute is merely wasted work.
run/2 opts:
:recompute— aReactiveDag.RecomputeStrategymodule (required unless the graph is leaves-only).:key_rule— aReactiveDag.KeyRulemodule (default: identity mapping).:on_step— optional(cell, step) -> anyfor STREAMING (live progress UI) — called after each cell recomputes with the same fields the report step carries (minus:cell/:pass, which the report adds). The full trace arrives in the report either way; useon_steponly when you need it before the drain finishes.:max_passes— runaway guard (default 100000): exceeding it raisesReactiveDag.Drain.RunawayError, whose:reportfield carries the partial trace — the step list showing which cells keep re-dirtying each other is exactly the diagnostic for the cycle the guard suspects.
Summary
Functions
@spec run( ReactiveDag.Plan.t(), keyword() ) :: {:ok, ReactiveDag.Drain.Report.t()}