@spec approve(String.t(), [key()] | :all, keyword()) :: [key()]

Approve a gated cell's held changes, so the next drain claims them.

keys names which — or :all for every held change of that cell. Returns the keys it released, so a caller can report what a click actually did rather than assuming.

Approving something already claimable is a no-op rather than an error: a double click, or two reviewers, must not fail.

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

The changes a gated cell is holding — {key, version_id}, for review.

The version is the record of the change: a reviewer resolves it to see what moved. It outlives this queue row, which is why the queue references it rather than copying it.

A READ: it consumes nothing, so a UI can poll it while a drain runs.

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

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

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

claim/1, but returning {key, version_id} pairs — the VERSION recording the change that dirtied each key (nil for a source-fed key, which has no Ash row behind it, or a recalculation, which is not a row change at all).

A queue row says which entity changed; the version says what the change DID. The consumer resolves the version to a diff and derives its own affected units from both sides — which is the only thing that names the unit a row LEFT as well as the one it landed in.

@spec dirty_cells(keyword()) :: [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?(keyword()) :: boolean()

True when nothing is dirty.

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

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

keys is a list of key strings, or of {key, diff} pairs where diff is the change's two sides — %{attr => %{"from" => old, "to" => new}} — which a consumer derives its claim from without reading the live row.

The diff is what makes a claim survive its subject. A deleted row cannot say which unit it belonged to; a row that MOVED between units says only where it went. The diff answers both, so a claim stays precise where it would otherwise degrade to a whole-cell recompute.

Coalescing MERGES the diffs: the earliest from and the latest to, per attribute. If a row moves meals → travel → lodging before a drain, the claim must name meals (where the last settled state had it) and lodging (where it is now) — never travel, an intermediate no settled state ever saw.

DO NOTHING would keep meals → travel and strand lodging; overwriting would keep travel → lodging and strand meals. Both lose a unit that needs repricing, which is why this merges rather than picks.

The merge is done in SQL, in the ON CONFLICT clause, because marks arrive from arbitrary concurrent writes. Reading the stored diff into Elixir to merge it there would be a read-modify-write with no lock around it — and the marking path deliberately holds none, since it runs inside a host's own write transaction.

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

The dirty cell with the smallest depth of the cells in depths, or nil.

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.

Tenant, and the cell this plan does not know

Reads only this tenant's rows (tenant:, default "*"). One frontier serves every plan in the application, and that is what makes the tenant load-bearing rather than bookkeeping: without it, "a cell this plan does not know" and "a cell nobody owns" are the same observation, and they need OPPOSITE handling.

• Foreign — a row belonging to another tenant. Never returned, never touched. It has an owner; the frontier is a set, so it is still there for the drain that can recompute it.
• Orphaned — a row in THIS tenant whose cell_id the plan does not declare: a renamed or removed cell, a source writing an old leaf id. Still returned, so the drain can claim it, log it and drop it (ReactiveDag.Drain does exactly that). Claiming rather than skipping is what stops the row being re-selected on every pass forever.

So the tenant filter is in SQL and the plan filter is not: an unknown cell of ours is work to clear, and another tenant's cell is not ours to look at.

@spec reject(String.t(), [key()] | :all, keyword()) :: [key()]

Reject a gated cell's held changes — DISCARD the marks without recomputing.

The rows are already written; this says the graph should not propagate from them. So a rejected change leaves the derived table as it stands and the consumers as they were, which is the honest meaning of "no" given the gate holds propagation rather than the write.

Returns the keys it discarded.

@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 tenant(keyword()) :: String.t()

The tenant a call is about: opts[:tenant], else "*" (untenanted).

"*" rather than nil because the coalescing unique index backs mark_dirty's ON CONFLICT and Postgres treats NULLs as DISTINCT in a unique index — a nullable column would stop untenanted marks coalescing and grow a queue row per mark. "*" is also already this library's spelling for "the whole thing" in claim sets, so the vocabulary is not new.

Normalised in ONE place so a caller passing nil, omitting the option, or passing an atom all mean the same row.

@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