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 savepoint per cell, so a cell that fails 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.

A recompute reports failure two ways, and they mean different things:

• it RAISES — the drain rolls that cell back and re-raises. Something is wrong with the graph or the host, and stopping is right.
• it returns {:error, reason} — the drain rolls that cell back, records it, and CARRIES ON with every other cell. The failed cell is excluded from selection for the rest of the run (its keys are still dirty, so it would otherwise be re-selected forever) and retried by the next drain.

The second is what lets a fallible unit live in the graph. A poll that could not reach its upstream is one cell staying dirty while the rest of the cascade runs — the containment ReactiveDag.Source.poll_all/2 gives a sweep, expressed where the work happens. It must be a RETURNED value: an exception inside a nested transaction aborts the outer one, so only a value can be isolated.

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.