Closes the payload loop for a resource-backed node: writes a combinator's output
row into the node's OWN resource (cell.meta.resource), keyed by the cell key.
This is what makes "the resource IS the node, and its rows ARE its payload" true
in the code, not just the docs. A reduce/join whose into returns a row and
that omits an explicit upsert: has its row written HERE — an Ash upsert into the
node's resource, with change-detection — so the common case needs no host write
callback, and writing into a different resource becomes the explicit deviation
(a custom upsert:), not the default.
The key attribute
The cell key (a string) maps to one resource attribute — the payload key. It
defaults to :key; a resource whose primary key is named otherwise declares
payload_key :flow_key in its reactive block. The row is written with that
attribute set to the cell key; a :key field on the row itself is dropped (it's
the cell key, not a payload column).
Change detection
upsert/5 reads the existing row first and compares the writable attributes; it
returns :changed only when a create or a real value change happened, so the
drain's parent-dirty only fires for genuine changes (a no-op recompute
stays a no-op). Requires an upsert action on the resource — by default the
action named :upsert with upsert?: true; override with payload_action.
Lapsing a human's mark
The upsert also holds the prior record long enough for lapse/5 to ask its
OWN question of it: did the fields a human's mark was about move? A lapse
declaration then nulls that column, or destroys the child rows, as a SEPARATE
write ordered after the create.
Separate deliberately. Folding it into the create's attrs would need the
payload action to accept the human column — and it would then null it on
every pass, destroying the default this whole loop gives for free: a column
the payload action does not accept is never touched, so a mark survives
recomputes with no declaration at all.
Summary
Functions
Clear the human marks a recompute invalidated: null the declared attributes, destroy the declared child rows.
RETIRE a vanished unit's row: the unit's input rows are gone, so the derived row must go too — a derived table whose whole point is that you query it cannot keep rows for units that no longer exist (a stale row is indistinguishable from a live one).
Upsert row into resource under cell_key (written to key_attr), via
action. row's :key field is dropped before writing.
Upsert an IDENTITY-KEYED row (a composite-primary-key node): the row carries
its identity fields, the upsert conflicts on the primary key (Ash's default
for upsert? true), and no key column exists — the cell key is the
identity's serialization, derived elsewhere. Returns :changed | :unchanged
with the same read-compare change detection as upsert/6, and the same
:fingerprint / :fingerprint_attribute / :lapse options.
Functions
Clear the human marks a recompute invalidated: null the declared attributes, destroy the declared child rows.
prior is the record as it was BEFORE this pass (nil on a create). attrs is
what was just written. Each lapse spec fires only when the fields IT watches
moved between the two — its own comparison, deliberately not the propagate
verdict, because a recompute is :changed the moment any column moves and a
mark about the totals must survive a spelling fix.
nil prior never lapses: no prior record means no mark, and there is nothing
to compare against.
Never raises
This runs inside the drain's per-cell savepoint, where a raise would abort the OUTER transaction and roll back a recompute that was perfectly good. So a failure to clear is logged and contained. That is safe only because everything checkable without writing — a missing attribute, an absent lapse action, a child with no destroy — already raised at ASSEMBLY, which is off the hot path. What can still fail here is a genuine write failure, and losing the recompute as well would not make the mark any more correct.
RETIRE a vanished unit's row: the unit's input rows are gone, so the derived row must go too — a derived table whose whole point is that you query it cannot keep rows for units that no longer exist (a stale row is indistinguishable from a live one).
keys are cell keys; for an identity-keyed node each is the identity's "|"
serialization, split back into its fields. Rows already absent are skipped, so
this is idempotent. Returns the keys whose row was actually destroyed.
Requires a destroy action (default :destroy); a resource without one raises
with the fix, since silently keeping the row would defeat the reconcile.
@spec upsert(module(), atom(), String.t(), map(), atom(), keyword()) :: :created | :changed | :unchanged
Upsert row into resource under cell_key (written to key_attr), via
action. row's :key field is dropped before writing.
Returns :created (no row existed), :changed (a row existed and moved) or
:unchanged. The first two both mean "propagate"; they are distinguished
because a caller reporting what a scan did wants them apart, and this is the
only place that knows — one row of the check it already performs.
Options
:fingerprint— compare ONE value instead of every attribute: a field list (hashed) or(row -> value). The value is written to:fingerprint_attribute(default:fingerprint), so the next call has something to compare against.This is what a SOURCE-FED LEAF needs. Comparing every attribute is right for a derived node, where every attribute is part of the result — but a leaf's row carries fields that move on every observation without the observation having changed: a
last_seen_atby definition, anetaga server may re-issue for identical bytes. Comparing those reports a change on every poll and re-runs the whole cascade.:fingerprint_attribute— where the value is stored (default:fingerprint).:compare— compare ONLY these fields, storing nothing. For a DERIVED row that carries fields which are part of the record but not part of the result:doc_id(provenance),ordinal(position in the source document), amatch_keya join builds. A re-parse that reorders rows moves everyordinal, and comparing them reports a change nothing made — which then re-runs every fold downstream.Prefer it to
:fingerprintfor a derived node: a fingerprint needs a column to hold the digest, and a digest of fields already on the row earns nothing when the comparison can just read them.:fingerprintremains the answer for a leaf, where the fields that move are the ones you must NOT compare and the honest witness is a hash of the ones you must.:lapse— the human marks this write clears when the content moves (the assembledlapseentities, fromcell.meta[:lapse]). Applied AFTER the create, as its own write, and only where the watched fields actually moved. Seelapse/5.
@spec upsert_identity(module(), [atom()], map(), atom(), keyword()) :: :created | :changed | :unchanged
Upsert an IDENTITY-KEYED row (a composite-primary-key node): the row carries
its identity fields, the upsert conflicts on the primary key (Ash's default
for upsert? true), and no key column exists — the cell key is the
identity's serialization, derived elsewhere. Returns :changed | :unchanged
with the same read-compare change detection as upsert/6, and the same
:fingerprint / :fingerprint_attribute / :lapse options.