@spec claim(String.t()) :: [key()]

Atomically claim (delete-returning) all dirty keys for cell.

@spec claim_with_priors(String.t()) :: [{key(), map() | nil}]

claim/1, but returning {key, prior} pairs — the snapshot each key was marked with (nil for a source-fed key, which has no row behind it).

The drain uses this so a parent can derive its claim from what the row WAS, which is the only thing that survives a delete.

@spec dirty_cells() :: [String.t()]

Every cell with dirty keys waiting — what the next drain would work on.

A READ: unlike claim/1 it consumes nothing, so it is safe to call for reporting (ReactiveDag.Insights.pending/1) while a drain is running.

@spec empty?() :: boolean()

True when nothing is dirty.

@spec mark_dirty(String.t(), [key() | {key(), map() | nil}], String.t() | nil) :: :ok

Mark keys of cell dirty, coalesced (idempotent per (cell, key)).

keys is a list of key strings, or of {key, prior} pairs where prior is the row AS IT WAS when marked — a map the parent can derive its claim from without reading the live row.

That snapshot is what makes a claim survive its subject. A deleted row cannot say which unit it belonged to, and a row that MOVED between units cannot say where it came from; the snapshot answers both, so a claim stays precise where it would otherwise degrade to a whole-cell recompute.

Coalescing keeps the FIRST snapshot (ON CONFLICT DO NOTHING), which is deliberate: if a row is written twice before a drain, the oldest prior state is the one that names the unit it started in.

@spec next_cell(%{required(String.t()) => non_neg_integer()}, [String.t()]) ::
  String.t() | nil

The dirty cell with the smallest depth, or nil if the frontier is empty.

except skips cells a caller has already tried this run. The drain uses it for a cell whose recompute FAILED: the failure rolled back, so its keys are still dirty and next_cell would hand back the same cell forever. Excluding it lets the rest of the cascade drain and leaves the retry to the next run.

@spec savepoint((-> result)) :: result | {:error, term()} when result: term()

Run fun in a SAVEPOINT: if it returns {:error, reason}, everything it did is undone and {:error, reason} comes back — without disturbing the transaction around it.

This is what lets one fallible unit fail inside a drain that is otherwise committing. A poll that could not reach its upstream rolls back its own claim and any rows it managed to write, and the drain carries on with every other cell — which is the containment Source.poll_all/2 gives a sweep, expressed where the work actually happens.

It must RETURN its failure, not raise

An exception inside a nested transaction aborts the OUTER one: Postgres marks the connection failed and every later statement errors until the whole thing rolls back. A savepoint only isolates a failure that arrives as a value.

That is why ReactiveDag.Source's poll/1 returns {:error, reason} "contained, not raised" — the contract was already the right shape for this. A scanner that raises anyway is a scanner that takes the drain down with it, and the rescue in safe_poll/2 is what stops that.

Returns fun's value unchanged when it does not error, and {:error, reason} when it does. A repo without transaction/2 runs fun directly: no savepoint, so a failure is not isolated — the same degradation transaction/1 makes.

@spec transaction((-> result)) :: result when result: term()

Run fun in a transaction, so a claim it makes is undone if it raises.

This is what makes a claim survive a failed recompute. claim/1 is a DELETE … RETURNING: the keys are consumed before the work happens, so without this a recompute that raises — a deadlock, a timeout, an upstream 503 — leaves those keys gone from the frontier and silently stale. Rolling back puts nothing back; it means they were never taken.

timeout: :infinity, because the work inside is a recompute and a recompute is the host's: an op that reads a PDF with a model legitimately runs for minutes, and Ecto's 15s default would abort it. The connection is held for the duration, which is affordable only because drains are SERIALIZED — one at a time per graph (with_lock/2) — so this is one connection, not one per concurrent drain.

Readers are unaffected: the DELETE takes row locks on one cell's keys, and Postgres readers never block on row locks. A mark_dirty on a DIFFERENT cell proceeds; only marking the same key of the same cell waits for the commit.

A repo without transaction/2

Runs fun directly. A host may configure a minimal repo — the library only ever needed query!/2 — and requiring a new capability would break it on upgrade for a guarantee it may not want. Such a host keeps the old behaviour: a failed recompute loses its claim.

@spec with_lock(
  (-> result),
  keyword()
) :: {:ok, result} | :busy
when result: term()

Run fun holding a cluster-wide lock on this graph's frontier, or skip.

The per-cell claim is atomic, so two concurrent drains never process a key twice — but they CAN pick the same cell and split its keys, recomputing it twice for a disjoint slice each. ReactiveDag.Drain has always said "run one drain at a time per graph"; this is how a host actually gets that when it runs more than one node.

A Postgres advisory lock, because the requirement is exactly what they are for: cluster-wide, held on a connection, and released automatically if that connection dies — a node crashing mid-drain does not leave the graph locked, which a lock table would.

Returns {:ok, result}, or :busy when another node holds it. Busy is not an error: the other drain is doing this drain's work, and the frontier is a set rather than a queue — anything this one would have claimed is still there for whoever holds the lock. A caller that treats :busy as a failure will retry work that is already happening.

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