The reactive propagation loop — the heart of the substrate, shared by both hosts.

Given a compiled Plan, 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 it (ReactiveDag.Node.Recompute, dispatching on what the node DECLARED) → the keys that changed.
4. Propagate: mark the changed keys on the cell's parents, applying the node's declared key rule (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.

One engine

There is no strategy to supply. A node's reactive block declares what it computes and how its changes propagate, and the drain reads that — so the same loop serves a per-key LLM pipeline and a set-based SQL model without either one bringing its own dispatch.

Earlier versions took recompute:/key_rule: modules. Both hosts ended up passing the library's own, because what actually varied between them was DATA — a module named in compute, a combinator, a key rule — not control flow. A pluggable engine that everyone plugs the same thing into is just an indirection.

run/2 opts:

• :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.

• :force — a cell id, a list of them, or :all: these cells' claimed keys propagate WHETHER OR NOT the recompute reported them changed. For a RE-RUN, where the point is that the graph catches up rather than that the work is redone.

  It does not make an op work harder. One that memoises on something the library cannot see — an md5-keyed cache, a content digest — still skips, and should: the expensive call is rarely what a re-run is after. What :force overrides is the conclusion drawn from changed == [], which otherwise stops the cascade at that cell. So a host that recomputed a cell out-of-band, or cleared a payload by hand, can drain and have everything downstream reflect it:

  Drain.run(plan, force: "transcript_record")

  Only the named cells are forced. Their parents propagate on their own verdicts, so a genuinely unchanged consumer still stops the cascade — forcing transitively would recompute the whole downstream graph on every re-run and make change detection pointless past the first hop.

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.