ReactiveDag.Frontier (reactive_dag v0.17.0-rc.2)

Copy Markdown View Source

The dirty frontier, owned by the library and backed by the reactive_dag_dirty table (created by ReactiveDag.Migration). The host is an Ash/AshPostgres app, so we go through its repo with raw SQL — values always parameterized; the table name (the one identifier SQL cannot parameterize) comes from config and is validated against an identifier grammar at read time, so a typo fails loudly instead of as a syntax error deep in a query. Claim-as-delete is a raw DELETE … RETURNING that Ash actions don't express cleanly.

The host supplies its repo (its AshPostgres repo module) via config, and may override the table name (default reactive_dag_dirty) so a host adopting the library keeps its existing table without a rename:

config :reactive_dag, repo: MyApp.Repo, dirty_table: "my_dirty"

Coalesced by (cell, key); depth-ordered next_cell; claim atomic per cell (DELETE … RETURNING — a key is consumed exactly once). The next_cell-then-claim PAIR is not serialized: concurrent drains can pick the same cell — see the concurrency note on ReactiveDag.Drain. This is the shared substrate both hosts previously hand-rolled (cascade's Cascade.Engine.Frontier, the portal's model_dirty access) — now provided.

Summary

Functions

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

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).

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

True when nothing is dirty.

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

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

Types

key()

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

Functions

claim(cell)

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

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

claim_with_priors(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.

dirty_cells()

@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.

empty?()

@spec empty?() :: boolean()

True when nothing is dirty.

mark_dirty(cell, keys, reason)

@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.

next_cell(depths)

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

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