ReactiveDag.Tuple (reactive_dag v0.16.0)

The shared COORDINATION tuple — the reactive layer's projection of a cell into a thin (cell_id, key, status, freshness) row. A cell IS its set of these rows; a parent reads a child's set by (cell_id, key).

This is NOT payload. The authoritative value for a key lives in the host's own typed resource (cascade's BudgetVsActual, the portal's Grant/Attestation), joined back by key. The coordination row carries only the verdict (status) and freshness — enough for the substrate to schedule and for a downstream cell to know which keys exist, without copying their payload.

Spine vs. extension

The library owns the SPINE — the columns both hosts share:

cell_id, key            (composite PK)
status                  (string; the host defines the vocabulary)
observed_at, updated_at, stale_after   (freshness)

Each host's physical table ALSO carries its own extension columns, which the library neither reads nor writes:

the portal adds strength (the evidence modality — its derive output);
cascade adds source_ref / last_seen_at / tombstoned_at (its retain-if-vanished + fingerprint policy).

So the library provides a spine CONTRACT + shared operators over the configured table — not the table itself (the host owns that, extension columns and all), exactly as ReactiveDag.Frontier owns the dirty ops but the host owns the *_dirty table. put/3 writes only spine columns (leaving any extension columns to their DB defaults / a host wrapper that sets them in the same upsert). Reads project spine columns.

config :reactive_dag, repo: MyApp.Repo, tuple_table: "my_tuple"

The join contract (what makes stratification work)

key is the universal join handle. A BEAM producing-node writes its spine row

its typed payload under a key; a SQL proving-node reads other cells' spine rows by (cell_id, key). Both wrote rows into ONE tuple_table, so a SQL cell can consume a BEAM cell's output by joining on key. That is the two-layer (produce → prove) graph, made concrete. It commits hosts to one constraint: a node's key strings must be stable and join-compatible across the producer/consumer seam.

Summary

Types

key()

key_scope()

A KEY-SCOPE selector — a host-declared narrowing of a spine read to a subset of a cell's keys, WITHOUT the host hand-writing SQL. The library turns each shape into a safe PARAMETERIZED predicate (no string interpolation of host values)

Functions

all_keys(cell_id, opts \\ [])

All keys of a cell (any status), optionally narrowed by :key_scope (a key_scope/0) — e.g. only the keys whose i-th segment names one service.

counts()

Count of tuples per cell, as %{cell_id => count}.

delete(cell_id, keys)

Delete the tuples for (cell_id, keys). No-op on an empty key list.

keys_by_status(cell_id, statuses, opts \\ [])

Keys of a cell whose status is in statuses, ordered by key. Options

max_observed_at(cell_ids)

Most-recent observed_at across the given cells, or nil if none have rows.

present_keys(cell_id, opts \\ [])

Keys of a cell whose status is "present", optionally narrowed by :key_scope (a key_scope/0). Equivalent to keys_by_status(cell, ["present"], key_scope: …) but unordered.

put(cell_id, key, opts \\ [])

Upsert the SPINE of a tuple for (cell_id, key): presence/verdict + freshness. Only spine columns are touched — extension columns (strength, source_ref, …) keep their DB defaults on insert and are left untouched on update, so a host that needs them writes its own upsert (calling this for the spine, or setting them in one combined statement host-side).

put_changed(cell_id, key, opts \\ [])

put/3, but returning whether the row's VERDICT actually changed: true for a new (cell_id, key) or a status flip, false for a re-put of the same status (freshness columns still update either way). This is the boolean CHANGED signal the ReactiveDag.CoordinationWriter contract lets a writer report, which ops use to propagate only real changes. Read-compare-then-upsert (same shape as the payload loop's change detection) — per-cell writes are serialized by the drain, which is what makes the two steps safe.

reconcile(cell_id, want_keys, opts)

Reconcile a cell's tuple set against a host-computed DESIRED key set — the one algorithm the portal's leaf drivers AND cascade's leaf refresh both hand-rolled

reconcile_set(cell_id, want_keys, opts)

The BULK variant of reconcile/3, for set-based recomputes. Where reconcile calls upsert.(key) once per want-key (N statements — the per-key leaf shape), reconcile_set hands the WHOLE want set to one :upsert_all callback, so an interior set-op that produced its result row-set in one pass writes it in ONE bulk statement (spine + the host's extension columns together). Same skeleton, same set math, one write.

rows(cell_id, opts \\ [])

The SPINE ROWS of a cell — %{key, status, observed_at} maps, ordered by key, optionally narrowed by :key_scope. The full-row companion to the key reads above: what an evaluation that needs status alongside key consumes (the attestation machinery's raw-rows and basis-digest input).

status_histogram(cell_id, opts \\ [])

Status histogram for a cell: %{status => count} over its tuples (optionally narrowed by :key_scope). The spine read behind a cell's verdict (failing / pending / green rollups are the host's to compute from this).

Types

key()

@type key() :: String.t()

key_scope()

@type key_scope() ::
  nil
  | {:prefix, String.t()}
  | {:exact_or_prefix, String.t(), String.t()}
  | {:segment, pos_integer(), String.t(), String.t()}

{:prefix, p} → key LIKE p (p is a full LIKE pattern, e.g. "app|%")
{:exact_or_prefix, k, p}→ (key = k OR key LIKE p) (a bare id OR its children)
{:segment, i, sep, v} → split_part(key, sep, i) = v (the i-th key segment)

Key GRAMMAR stays the host's — it names which segment / prefix means what; the library only assembles the predicate. nil (the default) means no scoping.

Functions

all_keys(cell_id, opts \\ [])

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

All keys of a cell (any status), optionally narrowed by :key_scope (a key_scope/0) — e.g. only the keys whose i-th segment names one service.

counts()

@spec counts() :: %{required(String.t()) => non_neg_integer()}

Count of tuples per cell, as %{cell_id => count}.

delete(cell_id, keys)

@spec delete(String.t(), [key()]) :: :ok

Delete the tuples for (cell_id, keys). No-op on an empty key list.

keys_by_status(cell_id, statuses, opts \\ [])

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

Keys of a cell whose status is in statuses, ordered by key. Options:

:limit — cap the result (e.g. a failing-sample)
:key_scope — a key_scope/0 narrowing to a subset of the cell's keys

This is the general spine status-read; present_keys/1 is the ["present"] special case.

max_observed_at(cell_ids)

@spec max_observed_at([String.t()]) :: DateTime.t() | nil

Most-recent observed_at across the given cells, or nil if none have rows.

present_keys(cell_id, opts \\ [])

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

Keys of a cell whose status is "present", optionally narrowed by :key_scope (a key_scope/0). Equivalent to keys_by_status(cell, ["present"], key_scope: …) but unordered.

put(cell_id, key, opts \\ [])

@spec put(String.t(), key(), keyword()) :: :ok

Opts:

:status — verdict string (default "present")
:stale_after — freshness horizon (default nil)
:observed_at — when the source produced this (default now)

put_changed(cell_id, key, opts \\ [])

@spec put_changed(String.t(), key(), keyword()) :: boolean()

reconcile(cell_id, want_keys, opts)

@spec reconcile(String.t(), [key()] | MapSet.t(), keyword()) :: {:ok, [key()]}

Reconcile a cell's tuple set against a host-computed DESIRED key set — the one algorithm the portal's leaf drivers AND cascade's leaf refresh both hand-rolled:

current  = the cell's current keys
want     = `want_keys` (the host computed the desired set + its payload)
upsert   each want key   → host writes the spine + its own extension columns
vanished = current − want
retire   the vanished    → host policy: DELETE (portal) or TOMBSTONE (cascade)
⇒ changed_upserts ++ vanished     (the keys to propagate to parents)

The library owns the SKELETON and the current/vanished set math; the two variation points are seams the host supplies:

:upsert — (key -> boolean) called per want-key; returns true iff this key's verdict ACTUALLY changed (so only real changes propagate). The host writes the row here (spine + strength/source_ref/…), because WHAT a present row contains and WHAT counts as "changed" are host domain logic.
:retire — how vanished keys leave. :delete (the default) uses the spine delete/2; a host with a retain-if-vanished policy passes a (keys -> any) fun (cascade tombstones). Vanished keys always propagate.
:current — the baseline set vanished is computed against, as current − want. Defaults to all_keys(cell). A host whose "live" set is narrower than all rows passes it explicitly — cascade's retain-if-vanished leaf passes its NON-tombstoned keys, so already-tombstoned keys are neither re-retired nor spuriously reported as newly vanished.

Returns {:ok, changed_keys} where changed_keys = changed_upserts ++ vanished.

reconcile_set(cell_id, want_keys, opts)

@spec reconcile_set(String.t(), [key()] | MapSet.t(), keyword()) :: {:ok, [key()]}

Options:

:upsert_all (required) — ([key] -> [changed_key]): write every want row in one statement (the host's bulk VALUES upsert; WHAT a row contains is host domain), returning the subset whose verdict ACTUALLY changed (an IS DISTINCT FROM-guarded upsert's RETURNING). Not called for an empty want set.
:retire — how vanished keys leave, as reconcile/3: :delete (the default) or a (keys -> any) fun (tombstone).
:current — the baseline vanished is computed against, as reconcile/3. Defaults to all_keys(cell_id, key_scope: opts[:key_scope]).
:key_scope — a key_scope/0 narrowing the DEFAULT baseline to the slice this recompute repriced: a dirty-key-scoped set-op must not see keys outside its slice as vanished. Ignored when :current is given.

Returns {:ok, changed_keys} where changed_keys = changed_upserts ++ vanished.

rows(cell_id, opts \\ [])

@spec rows(
  String.t(),
  keyword()
) :: [%{key: key(), status: String.t(), observed_at: term()}]

status_histogram(cell_id, opts \\ [])

@spec status_histogram(
  String.t(),
  keyword()
) :: %{required(String.t()) => non_neg_integer()}