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:

1. 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).
2. Atomically claim its dirty keys (Frontier.claim — delete-returning).
3. Recompute via the host's RecomputeStrategy → the keys that changed.
4. Propagate: mark the changed keys on the cell's parents, applying the host's KeyRule (Graph.dirty_parents).
5. 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 — a ReactiveDag.RecomputeStrategy module (required unless the graph is leaves-only).
• :key_rule — a ReactiveDag.KeyRule module (default: identity mapping).
• :on_step — optional (cell, step) -> any for 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; use on_step only when you need it before the drain finishes.
• :max_passes — runaway guard (default 100000): exceeding it raises ReactiveDag.Drain.RunawayError, whose :report field 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.