ReactiveDag.Source behaviour (reactive_dag v0.17.0-rc.2)

Copy Markdown View Source

A scanner — the fourth seam, alongside ReactiveDag.RecomputeStrategy, ReactiveDag.KeyRule, and ReactiveDag.CoordinationWriter.

A source reads external state (a fleet API, a cloud estate, a repo, an LLM) and writes a leaf cell in a poll phase that is deliberately OUTSIDE the drain:

  1. poll — run each source's poll/1: fetch → write its leaf tuples → return the leaf keys that CHANGED (so the caller can mark parents dirty). Sources are independent; a failure is contained to its own leaf.
  2. drain — the engine recomputes everything downstream from the dirty frontier (ReactiveDag.Drain). No source runs here.

This split is a design invariant, not an accident: the drain is pure set/graph computation over tuples already present — deterministic, re-runnable, and it never fails on a network outage. Effectful, non-deterministic, fallible I/O (that's every scanner) stays in phase 1.

The scanner↔leaf binding has ONE home, chosen by cardinality:

  • 1:1 (the common case) — inline on the leaf. source :fleet_scan / driver MyApp.Sources.FleetScan in the leaf's reactive block co-locates the leaf and its scanner in one declaration (they travel together). ReactiveDag.Source.drivers/2 reads these off the graph (each leaf cell's meta.driver).
  • fan-out (rare) — on the driver. A scanner that writes cells no single leaf owns (e.g. many guarantee sub-cells) has no inline driver and names its cells via its own leaf_cells/1; the host passes it as an extra driver.

Either way verify!/2 confirms every named leaf is a real cell in the built plan (an inline driver's leaf is the node it's declared on).

The contract

Two apps (a data pipeline and a compliance model) independently grew the same three-callback shape — id / leaf_cells / poll → changed-keys — which is why it lives here rather than in either app. A single-leaf source (the common case) may export leaf_cell/0 instead of leaf_cells/1; cells_of/2 resolves whichever is present.

defmodule MyApp.Sources.FleetScan do
  @behaviour ReactiveDag.Source

  @impl true
  def id, do: :fleet_scan

  @impl true
  def leaf_cells(_graph), do: ["machines"]

  @impl true
  def poll(_opts) do
    # fetch the fleet, write the "machines" leaf's tuples, return changed keys
    {:ok, %{changed: ["host-1", "host-7"]}}
  rescue
    e -> {:error, Exception.message(e)}
  end
end

Fan-out and multi-leaf sources

leaf_cells/1 takes the lowered graph and returns a list of cell ids, so a source that feeds many leaves (e.g. one per discovered kind) computes them from the graph, and a source that direct-writes several cells lists them all. A single-leaf source returns [one_id]. This list is what verify/2 checks.

Summary

Types

The lowered graph — a ReactiveDag.Plan (or any map with a :cells map keyed by id).

Callbacks

Stable id of this source (matches the source :id binding where declared).

Single-leaf fallback for leaf_cells/1: the ONE cell id this source feeds. For the common one-scanner-one-leaf driver, this is the whole binding — no graph-dependent computation to write.

The authoritative set of cell ids this source feeds, given the lowered graph — the binding verify/2 validates. Single-leaf sources return [leaf]; fan-out sources compute their per-instance leaves from the graph; multi-leaf sources list every cell they write.

Optional lineage for display: where this source's data comes from, as a map like %{label: "Fleet · Huntress", url: "https://…", store: "Tigris"} (any subset). Not implemented = origin unknown.

Poll the external source: fetch → write the leaf tuples → return the leaf keys that changed. {:error, reason} when it couldn't run at all (no credential, API down) — contained, not raised, so one bad source doesn't abort a refresh. arg is source-specific (a since-timestamp, a manifest path, opts).

Functions

The cells source feeds in graph — the resolver behind verify!/2. Uses the module's leaf_cells/1 when exported, else the single-leaf leaf_cell/0 fallback (as [to_string(leaf_cell())]). Raises ArgumentError when the module exports neither.

The scanner drivers feeding a lowered graph: the inline ones declared with driver MyApp.Sources.FleetScan on a leaf's reactive block (read from each leaf cell's meta.driver), unioned with any extra fan-out drivers a host passes (drivers that name their cells via leaf_cells/1 because no single leaf owns them). This is the full scanner set — feed it to verify!/2, poll it in phase 1.

Poll every scanner the PLAN declares (via scan Mod on its leaves), in the poll phase before a drain.

The distinct scanner modules a plan's leaves declare via scan.

Verify every source's declared leaves resolve to real cells in graph — the authoritative scanner↔leaf check. Each driver's leaves are resolved via cells_of/2 (leaf_cells/1, or the single-leaf leaf_cell/0 fallback); this confirms every one is a real cell in the built plan. Needs the lowered graph (a host may expand generator leaves from live data), so it runs at assembly/boot time, not compile time.

The same dangling-leaf check as verify!/2, but over already-resolved {source, [cell_id]} pairs instead of resolving each module via cells_of/2. Use this when a host resolves fed cells itself (its own conventions beyond leaf_cells/1 / leaf_cell/0). Returns :ok, or raises ArgumentError naming every {source, dangling_leaf}.

Verify a scan Mod declaration on the leaf cell_id: the module must be a loadable ReactiveDag.Source, and its own leaf_cells/1 must claim this leaf.

Types

graph()

@type graph() :: %{cells: %{optional(String.t()) => struct()}}

The lowered graph — a ReactiveDag.Plan (or any map with a :cells map keyed by id).

Callbacks

id()

@callback id() :: atom()

Stable id of this source (matches the source :id binding where declared).

leaf_cell()

(optional)
@callback leaf_cell() :: String.t() | atom()

Single-leaf fallback for leaf_cells/1: the ONE cell id this source feeds. For the common one-scanner-one-leaf driver, this is the whole binding — no graph-dependent computation to write.

leaf_cells(graph)

(optional)
@callback leaf_cells(graph()) :: [String.t()]

The authoritative set of cell ids this source feeds, given the lowered graph — the binding verify/2 validates. Single-leaf sources return [leaf]; fan-out sources compute their per-instance leaves from the graph; multi-leaf sources list every cell they write.

OPTIONAL: a single-leaf source may instead export leaf_cell/0 (the common case — one scanner, one leaf) and skip this; cells_of/2 resolves whichever the module exports. A module must export at least one of the two.

origin()

(optional)
@callback origin() :: map() | nil

Optional lineage for display: where this source's data comes from, as a map like %{label: "Fleet · Huntress", url: "https://…", store: "Tigris"} (any subset). Not implemented = origin unknown.

poll(arg)

@callback poll(arg :: term()) ::
  {:ok,
   %{:changed => [String.t()], optional(:unreachable) => [{String.t(), term()}]}}
  | {:error, term()}

Poll the external source: fetch → write the leaf tuples → return the leaf keys that changed. {:error, reason} when it couldn't run at all (no credential, API down) — contained, not raised, so one bad source doesn't abort a refresh. arg is source-specific (a since-timestamp, a manifest path, opts).

A multi-upstream source that could observe SOME of its inputs reports the others under the optional unreachable: key ({upstream_label, reason} pairs) — the honest-gap discipline: a scan that couldn't look must never render as a scan that found nothing, so write what you observed, retire nothing you couldn't see, and surface the outage for the host to display.

Functions

cells_of(source, graph)

@spec cells_of(module(), graph()) :: [String.t()]

The cells source feeds in graph — the resolver behind verify!/2. Uses the module's leaf_cells/1 when exported, else the single-leaf leaf_cell/0 fallback (as [to_string(leaf_cell())]). Raises ArgumentError when the module exports neither.

drivers(graph, extra \\ [])

@spec drivers(graph(), [module()]) :: [module()]

The scanner drivers feeding a lowered graph: the inline ones declared with driver MyApp.Sources.FleetScan on a leaf's reactive block (read from each leaf cell's meta.driver), unioned with any extra fan-out drivers a host passes (drivers that name their cells via leaf_cells/1 because no single leaf owns them). This is the full scanner set — feed it to verify!/2, poll it in phase 1.

poll_all(graph, opts \\ [])

@spec poll_all(
  graph(),
  keyword()
) :: {:ok, map()} | {:error, [{module(), term()}]}

Poll every scanner the PLAN declares (via scan Mod on its leaves), in the poll phase before a drain.

Scanners are found from the graph rather than a list the host maintains alongside it — the list is the thing that drifts. A source feeding many leaves appears once, however many leaves declare it.

Returns {:ok, %{module => result}}, or {:error, failures} where failures are {module, reason}: one scanner failing must not silently cancel the others, and must not look like success.

scanners(graph)

@spec scanners(graph()) :: [module()]

The distinct scanner modules a plan's leaves declare via scan.

verify!(sources, graph)

@spec verify!([module()], graph()) :: :ok

Verify every source's declared leaves resolve to real cells in graph — the authoritative scanner↔leaf check. Each driver's leaves are resolved via cells_of/2 (leaf_cells/1, or the single-leaf leaf_cell/0 fallback); this confirms every one is a real cell in the built plan. Needs the lowered graph (a host may expand generator leaves from live data), so it runs at assembly/boot time, not compile time.

Returns :ok, or raises ArgumentError naming every {source, dangling_leaf}.

verify_cells!(source_cells, graph)

@spec verify_cells!([{module(), [String.t()]}], graph()) :: :ok

The same dangling-leaf check as verify!/2, but over already-resolved {source, [cell_id]} pairs instead of resolving each module via cells_of/2. Use this when a host resolves fed cells itself (its own conventions beyond leaf_cells/1 / leaf_cell/0). Returns :ok, or raises ArgumentError naming every {source, dangling_leaf}.

verify_scan!(source, cell_id, graph)

@spec verify_scan!(module(), String.t(), graph()) :: :ok

Verify a scan Mod declaration on the leaf cell_id: the module must be a loadable ReactiveDag.Source, and its own leaf_cells/1 must claim this leaf.

The second half matters more than it looks. A scanner already knows which cells it feeds; scan states the same fact from the other side. Two statements of one fact can disagree, so this is the check that they don't — a scanner refactored to feed "agenda_docs_v2" while a resource still declares scan fails at assembly rather than polling into a cell nobody reads.

Called by ReactiveDag.Node.graph/2 for every cell carrying a scan, so a host declaring scanners in the DSL needs no verify!/2 call of its own.