Reads a cell's own rows, keyed the way the DAG keys them.
A node's rows live in its resource, but the DAG addresses them by cell key
— a "|"-joined identity for a composite-PK node, a single column's value
otherwise. Anything that wants to ask a question about a cell rather than
about a table (ReactiveDag.Insights, ReactiveDag.Verdict, a union
reading its inputs) needs the rows under those keys, not under the resource's
own primary key.
This is the read side of what ReactiveDag.Node.Payload writes, and it
deliberately mirrors that module's key derivation: identity_fields when the
node is identity-keyed, payload_key (defaulting to :key) otherwise.
Why it is not the coordination tuple
These reads used to go to reactive_dag_tuple, which carried a status per
(cell_id, key). That made the tuple a second home for derived results — one
with a fixed two-column schema, updated by a writer the host had to configure,
and queryable only through this library. Reading the resource instead means a
status is an ordinary column with ordinary Ash semantics: policies apply,
loads work, and a host can add a second column without asking us.
Missing columns are not errors
A node need not have a :status column — most don't; a rollup has sums. Rows
from such a node come back with status: nil, and callers that count statuses
simply find nothing to count. A node with no attributes at all keeps its rows
somewhere else entirely and reads as empty. Asking is always safe.
Summary
Types
one of a cell's rows, addressed by cell key
Where a node's rows live and how they are keyed — a cell's meta, or the
same three fields lifted out of it (which is what a union carries for each
of its inputs, since it reads rows it does not own).
Functions
Every row the cell currently holds, as %{key:, status:, record:}.
Clear the stored fingerprint on keys, so the next recompute treats those rows
as needing work.
How many units the cell holds, counted by the datastore.
The keys whose status is in statuses, at most limit of them (sorted, so a
sample is stable between calls rather than reshuffling on every render).
The keys whose rows match filter — the selection behind "reprocess just this
year".
Turn a selected slice into the options a POLL is asked with.
Reconcile a leaf's rows against the key set a scan found — the algorithm every leaf driver otherwise hand-rolls.
The dimensions this node declared a human may select it by, with their options resolved.
%{status => count} over the cell's rows — the histogram Insights shows and
Verdict folds into one answer.
Types
one of a cell's rows, addressed by cell key
@type source() :: %{ optional(:resource) => module() | nil, optional(:payload_key) => atom() | nil, optional(:identity_fields) => [atom()] | nil }
Where a node's rows live and how they are keyed — a cell's meta, or the
same three fields lifted out of it (which is what a union carries for each
of its inputs, since it reads rows it does not own).
Functions
@spec all(ReactiveDag.Cell.t() | source()) :: [row()]
Every row the cell currently holds, as %{key:, status:, record:}.
Returns [] for a node that keeps no rows here — no resource at all, or a
resource with no attributes (the shape a compute/custom-upsert: node has,
where the real writes land somewhere this library never sees). That is
different from "holds nothing", so a caller that must tell the two apart
should check meta[:resource] itself.
Raises whatever the underlying Ash.read!/1 raises. Callers on a display path
(Insights) already run these behind their own rescue; a caller on a compute
path wants the failure.
@spec invalidate(ReactiveDag.Cell.t() | source(), [String.t()] | :all) :: [String.t()]
Clear the stored fingerprint on keys, so the next recompute treats those rows
as needing work.
A fingerprint answers "did the input move?". After a prompt change or a fixed
fold the input has NOT moved and the answer is still valid — it is simply the
wrong question, because what changed was the function. A per_key node would
therefore skip exactly the rows you asked it to redo.
Clearing the stored value makes the comparison fail honestly rather than bypassing it: a null fingerprint means "no valid prior result", which is precisely true once the code that produced it has changed. Nothing needs a force flag threaded through the recompute, and the next run stores a fresh fingerprint as it always would.
A node with no fingerprint column has nothing to clear and recomputes regardless, so this is a no-op there. Returns the keys it actually cleared.
@spec key_count(ReactiveDag.Cell.t() | source()) :: non_neg_integer()
How many units the cell holds, counted by the datastore.
Ash.count! rather than reading the rows: an overview wants the number, and
loading a table to reduce it to one integer decodes every payload column on the
way — which for a node whose rows carry a JSON blob is the whole cost of the
read for none of the value.
Returns 0 for a node that keeps no rows here.
@spec keys_by_status(ReactiveDag.Cell.t() | source(), [String.t() | nil], keyword()) :: [String.t()]
The keys whose status is in statuses, at most limit of them (sorted, so a
sample is stable between calls rather than reshuffling on every render).
@spec keys_where( ReactiveDag.Cell.t() | source(), keyword() ) :: [String.t()]
The keys whose rows match filter — the selection behind "reprocess just this
year".
Rows.keys_where(cell, fiscal_year: "FY25")
#=> ["FY25|gf", "FY25|water"]The filter goes to the datastore; keys are built here, because an
identity-keyed node's key is a "|"-join no datastore knows. So this decodes
what matched rather than the table.
A node declares which columns are meant for this with slice, and
slices/1 reports them — but nothing stops a caller filtering on any column
it knows about. The declaration is what makes a UI possible, not what makes
the filter legal.
@spec poll_opts(ReactiveDag.Cell.t() | source(), map() | keyword()) :: keyword()
Turn a selected slice into the options a POLL is asked with.
Rows.poll_opts(cell, %{"fiscal_year" => "FY25/26"})
#=> [fiscal: "FY25/26"]A slice narrows two different things. keys_where/2 filters rows this node
already HOLDS — "re-derive FY25 from documents I have". This narrows the
FETCH: a source whose upstream is addressable by the same dimension can be
asked for just that part, and a crawler that takes fiscal: walks twelve
months instead of the whole corpus.
The selection is keyed by COLUMN, because that is what a UI has — it rendered
a button per value under the slice's own name. The result is keyed by
poll_as, because that is the scanner's vocabulary. Translating here is the
point: neither side has to learn the other's spelling.
Accepts string or atom column names, since a selection usually arrives from a
form or a job argument. A column this node never declared as a slice is
IGNORED rather than passed through — an unrecognised option would otherwise
reach poll/1 as if the node had offered it, and a scanner that pattern
matches its arguments would crash on a typo the DSL could not vouch for.
@spec reconcile(ReactiveDag.Cell.t(), [String.t()] | MapSet.t(), keyword()) :: {:ok, [String.t()], %{ created: [String.t()], updated: [String.t()], revived: [String.t()], retired: [String.t()] }}
Reconcile a leaf's rows against the key set a scan found — the algorithm every leaf driver otherwise hand-rolls.
Returns {:ok, changed, detail} — changed is the flat list that propagates,
and detail is %{created:, updated:, revived:, retired:} saying WHY each key
is in it.
The detail is free: reconcile/3 computes those four sets to build changed
and used to flatten them away, leaving a host to rebuild the classification a
scan report shows. It cannot be reconstructed afterwards — by the time you
look, the rows are already written.
current = the cell's current keys (read from its resource)
want = `want_keys`, what the scan found
upsert each want key → the host writes the row, returns true iff CHANGED
vanished = current − want → retired
⇒ changed_upserts ++ vanished (the keys to propagate):upsert is called once per want-key, in either of two forms:
(key -> row | nil)— the common case. Return the row you observed and the library writes it, decidingchanged?against the node's declaredfingerprint(or every attribute, if it declares none). Returnnilfor a key you could not observe: nothing is written and the key is not reported, which is how a partial outage stays honest.(key -> boolean | :created | :changed | :unchanged)— full control. Write the row yourself and say what happened. Use this when the write is not an upsert into the node's own resource.true/falsemean:changed/:unchanged. Prefer the atoms when you know which: only you can tell an insert from an update, since the library never saw the row, andtruereports a brand-new key asupdated— which it cannot be, having had no prior row.:observed— was this scan COMPLETE?:all(the default) means absence fromwant_keysis evidence a key is gone, so the vanish diff runs.:partialmeans the scan looked at only part of the upstream — a scopedonly:poll, a windowedrecent:one, a crawl whose index page failed — so absence means "not looked at" and nothing can vanish. See below.:retire— how vanished keys leave. See the table below.:current— the baselinevanishedis computed against. Defaults to the cell's current keys. Pass it when your live set is narrower than all your rows, or when the scan asked a NARROWER QUESTION than "everything" — a date-scoped scan with a whole-table baseline retires everything outside its window.
When a key stops being returned
One decision, three answers:
| you want | you write | the row | propagates? |
|---|---|---|---|
| destroy it | nothing — the default | destroyed | yes |
| keep it | retain_if_vanished true on the node | untouched | no |
| mark it | retain_if_vanished mark: &tombstone/1 | yours to write | yes |
Keep and mark are the same operation with one question between them — do we write something to say it is gone? — and propagation follows from the answer rather than being a separate switch:
- destroying removes a unit downstream was counting, so it is a change;
- keeping changes nothing — the row is still there, unmodified — so reporting it would be a lie, and would report it again on every poll forever, since nothing marks it as handled;
- marking is a change you made, so downstream hears about it. It also
means you own
:current, and inherit one case a fingerprint cannot see: a marked-retired row returning with unmoved bytes.changed?compares fingerprints, so the revival is invisible — the library warns when it sees that shape but cannot fix it (u2i/reactive_dag#82). Report it yourself with the boolean:upsertform.
A leaf declaring fingerprint needs no :upsert at all for the row-returning
form to be worth using — that is the point: poll/1 becomes fetch, build
rows, call reconcile.
Partial observations
Retiring a key is an inference: the upstream no longer lists it, so it is gone. That inference is only valid from a complete observation. A scoped poll, a windowed one, or a crawl whose index page failed all produce a want-set that is real but incomplete, where absence means "not looked at".
Rows.reconcile(cell, observed_keys, observed: :partial, upsert: &fetch/1)Nothing vanishes, nothing is retired, and the keys you did see are written and reported exactly as usual.
This was previously spelled current: [] — "measure vanishing against
nothing" — which works, and says nothing about why. The distinction is worth a
name because the failure is silent and severe in one direction only: getting
:partial wrong under-retires, leaving rows that should have gone. Getting
:all wrong tombstones everything the scan did not happen to look at, which
for an archival consumer is a mass-deletion wave from one upstream 500.
The honest gap
Call this only with a want-set you actually observed. An upstream you could
not reach must write NOTHING — handing an empty want_keys to a scan that
failed retires every key the cell has, and a downstream rollup over an empty
set typically reads as vacuously fine. A scan that couldn't look must never
render as a scan that found nothing.
A total outage and a partial one differ: an outage writes nothing at all, so it marks nothing dirty and the drain correctly does no downstream work. A partial observation writes what it saw — it just must not conclude anything from what it did not.
@spec slices(ReactiveDag.Cell.t() | source()) :: [map()]
The dimensions this node declared a human may select it by, with their options resolved.
Rows.slices(cell)
#=> [%{column: :fiscal_year, label: "fiscal_year",
# values: ["FY24", "FY25"], poll_as: :fiscal}]values is nil when the node named no options — a UI then takes free text,
or offers nothing.
poll_as is what to call this dimension when asking the SCANNER for it, which
defaults to the column. See poll_opts/2.
@spec status_histogram(ReactiveDag.Cell.t() | source()) :: %{ required(String.t() | nil) => non_neg_integer() }
%{status => count} over the cell's rows — the histogram Insights shows and
Verdict folds into one answer.
Rows with no status are counted under nil, so the counts always sum to the
cell's key count and a node without a :status column reports
%{nil => n} rather than lying with %{}.