A scanner — the third seam, alongside ReactiveDag.RecomputeStrategy and
ReactiveDag.KeyRule.
A source reads external state (a fleet API, a cloud estate, a repo, an LLM) and writes a leaf cell's rows in a poll phase deliberately OUTSIDE the drain:
- poll — run each source's
poll/1: fetch → write its leaf's rows → return the leaf keys that CHANGED (so the caller can mark parents dirty). Sources are independent; a failure is contained to its own leaf. - 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 rows already written — 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.FleetScanin the leaf'sreactiveblock co-locates the leaf and its scanner in one declaration (they travel together).ReactiveDag.Source.drivers/2reads these off the graph (each leaf cell'smeta.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
driverand names its cells via its ownleaf_cells/1; the host passes it as anextradriver.
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 rows, return changed keys
{:ok, %{changed: ["host-1", "host-7"]}}
rescue
e -> {:error, Exception.message(e)}
end
endFan-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.
What a host needs to render a scan control for each cell that has one:
%{cell_id => %{source:, args:, every:, origin:}}.
The Oban-style crontab entries a plan's leaves declare, as
{cron, worker, args: %{"cell" => id}}.
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.
Poll the scanner feeding ONE cell — the "re-run this scanner" affordance.
Poll a cell's scanner AND mark what changed — the whole poll half of the two-phase loop, in one call.
Every scannable unit of work in the plan: %{cell:, source:, args:, every:}.
The distinct scanner modules a plan's leaves declare via scan.
The standing args: each scanner's leaf declared, as %{module => keyword}.
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
The lowered graph — a ReactiveDag.Plan (or any map with a :cells map keyed by id).
Callbacks
@callback id() :: atom()
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: 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.
@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.
@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
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.
What a host needs to render a scan control for each cell that has one:
%{cell_id => %{source:, args:, every:, origin:}}.
The library describes; the host renders. A cell with no scanner is absent, and
a scanner declaring no args:/every: reports them empty — so a leaf cheap
enough to run whole gets a plain "refresh" and no misleading range picker,
without the dashboard having to know which scanners are expensive.
origin: is the source's own origin/0 when it implements it, so a control
can say where it is about to fetch from.
The Oban-style crontab entries a plan's leaves declare, as
{cron, worker, args: %{"cell" => id}}.
The library never schedules anything. A leaf declaring every: states how
often a routine poll should run; this collects those declarations into data
the host hands to its own scheduler:
plugins: [
{Oban.Plugins.Cron, crontab: ReactiveDag.Source.crontab(plan, MyApp.ScanWorker)}
]Emitting data rather than inserting jobs keeps the library out of the host's supervision tree and out of its deploy story — and lets a host filter, rewrite or ignore the entries, which it could not do if they were already scheduled.
The worker receives %{"cell" => "agenda_docs"} and is expected to poll that
one leaf. A leaf declaring no every: contributes nothing, which is the
correct outcome for a scanner cheap enough to run on any cadence the host
likes.
Adding your own arguments
A crontab entry is built once at config time, so anything it carries is fixed
for every firing — a per-run id has to be minted when the job fires, inside
the worker. What a host CAN do here is add its own standing arguments, with
args::
Source.crontab(plan, MyApp.ScanWorker, args: %{"queue" => "crawls"})merged UNDER the "cell" this computes, so a typo cannot silently retarget
the job at a different leaf. A host that wants per-firing values wraps the
worker rather than the crontab: mint the id in perform/1, then call
refresh/3 with reason: "scan:#{run_id}" — :reason is a free string and
rides through to the frontier rows, so the trace says which run dirtied a cell.
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.
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.
@spec poll_cell(graph(), String.t(), keyword()) :: {:ok, map()} | {:error, term()} | {:error, :no_scanner}
Poll the scanner feeding ONE cell — the "re-run this scanner" affordance.
poll_all/2 is the routine sweep. This is what a host wires a button to: a
dashboard has a cell in hand, not a source module, and a human asking to
refresh is asking about this leaf, not about every scanner in the graph.
# routine, on the declared cadence
Source.poll_all(plan)
# a human pressed "refresh", accepting the cheap default
Source.poll_cell(plan, "agenda_docs")
# ...or asked for the deep pass
Source.poll_cell(plan, "agenda_docs", recent: false)The leaf's declared args: apply exactly as they do in poll_all/2, with the
caller's opts winning — so a button that passes nothing gets the cheap pass,
and one that passes recent: false gets the expensive one.
Returns {:ok, result}, {:error, reason} if the poll failed, or
{:error, :no_scanner} when the cell declares none — which a host should
render as "no refresh available" rather than as a failure.
Note a source feeding several leaves is polled whole: poll/1 takes options,
not a cell, so asking for one leaf runs whatever that scanner does. The
scanner narrows itself through args: if that matters.
@spec refresh(graph(), String.t(), keyword()) :: {:ok, %{changed: [String.t()], marked: map(), unreachable: list()}} | {:error, term()}
Poll a cell's scanner AND mark what changed — the whole poll half of the two-phase loop, in one call.
poll_cell/3 returns what the scanner said and stops, leaving every host to
hand-write the same three steps: normalise the return shape, mark the
frontier, propagate to parents. That loop is documented in the guide and was
provided nowhere, which is why every host grew a worker to hold it.
{:ok, result} = Source.refresh(plan, "agenda_docs")
#=> %{changed: ["a", "b"], marked: %{"agenda_docs" => ["a", "b"]},
# unreachable: []}Then drain — separately, and deliberately so. The poll/drain split is a design invariant (external I/O must not sit inside a depth-ordered recompute), and a host polling several sources usually wants ONE drain after all of them rather than one each.
Return shapes
A single-leaf source returns a flat key list; a fan-out source returns
%{leaf_id => keys}. Both are in the Source contract, so both are
normalised here rather than in each host.
:reason labels the frontier rows (default "scan"), which is what makes a
drain's trace say why a cell was dirty.
Nothing is marked for an unreachable upstream, because the scanner reported no keys for it — an outage propagates nothing, which is the honest gap holding by construction rather than by rule.
@spec scan_jobs(graph()) :: [ %{ cell: String.t(), source: module(), args: keyword(), every: String.t() | nil } ]
Every scannable unit of work in the plan: %{cell:, source:, args:, every:}.
One entry per cell that declares a scan, whether or not it declares a
cadence. This is the list a host schedules from, renders controls from, and
triggers ad-hoc runs from — crontab/2 is a projection of the subset that
declared every:.
Source.scan_jobs(plan)
#=> [%{cell: "agenda_docs", source: MyApp.Crawler,
# args: [recent: true], every: "0 * * * *"}]Keyed by CELL rather than by source, because a source feeding several leaves is several units of work — each with its own declared bound.
The distinct scanner modules a plan's leaves declare via scan.
The standing args: each scanner's leaf declared, as %{module => keyword}.
poll_all/2 merges these under the caller's opts. Exposed because a host
driving one scanner directly wants the same default rather than a second copy
of it.
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}.
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.
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.