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.

Steps 2–4 run in ONE transaction, so a recompute that raises leaves the frontier exactly as it found it. A claim is a delete: without that, a transient failure — a deadlock, a timeout, an upstream 503 — consumes the work item and those keys go silently stale. The drain still fails loudly; the next one retries rather than never knowing.

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

So run ONE drain at a time per graph. On a single node that is a single worker; across a CLUSTER it is ReactiveDag.Frontier.with_lock/2, a Postgres advisory lock that ReactiveDag.ScanWorker's sweep already takes:

case Frontier.with_lock(fn -> Drain.run(plan, opts) end) do
  {:ok, {:ok, report}} -> report
  :busy -> :already_draining
end

:busy is not an error. The frontier is a set rather than a queue, so anything this drain would have claimed is still there for whoever holds the lock — a caller that retries on :busy retries work already in progress.

Failing that, 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).
• :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.

Telemetry

The drain emits :telemetry events, so a host observes it without threading a callback through every call site — a dashboard, a metrics backend and a log can all attach independently, and none of them changes how the drain is invoked.

:step carries the changed KEYS, not just their count, because that is what makes a consumer incremental: a dashboard that knows which cells moved reads only those instead of re-reading the graph.

:telemetry.attach("my-drain-log", [:reactive_dag, :drain, :stop], fn _e, m, meta, _ ->
  Logger.info("drained #{length(meta.cells_touched)} cells in #{m.duration_us}us")
end, nil)

:exception fires for a RunawayError too, carrying the partial report — a monitor should see the runaway, not just the crash.

This replaces an earlier :on_step option. A closure threaded through run/2 could only serve whoever owned that call site — a dashboard, a metrics backend and a log could not all have one, and adding a second consumer meant editing every place the drain was invoked. Telemetry has no such limit, and a caller who wants a plain closure can attach one in three lines.