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

The cells a source feeds, derived from the graph.

A node declares poll MyCrawler; everything reading that node is an ordinary edge, so a source's outputs are its children. One declaration, read in one place, and nothing that can go stale.

It used to be the reverse: the module declared leaf_cells/1, each fed leaf declared scan, and verify_scan!/3 caught the two disagreeing. Every implementation in this repo and its host apps ignored the graph argument and returned a literal list, restating what the leaves had already said.

Returns [] for a source no node polls — not an error: a plan built from a subset of resources legitimately excludes some.

@spec controls(graph()) :: %{required(String.t()) => map()}

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.

@spec crontab(graph(), module(), keyword()) :: [{String.t(), module(), keyword()}]

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.

One entry per SCANNER, not per leaf

A source feeding several leaves — one crawl of one site whose rows land in two cells — is ONE unit of scheduled work. scan_jobs/1 lists it per-cell, which is right for a control panel (a human triggers a leaf, and each leaf declares its own bound), and wrong for a scheduler: two leaves declaring the same module would emit two entries and crawl the same site twice an hour, each poll marking both leaves and the second achieving nothing.

So entries are deduplicated by scanner. The "cell" carried is the first of its leaves alphabetically — any of them reaches the same scanner, and refresh/3 marks every leaf the poll reports regardless of which one was named.

Two leaves declaring DIFFERENT cadences for one scanner get one entry each, on the assumption that a host writing two different cadences meant them; the duplicate crawl is then declared rather than accidental.

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.

One sweep, not N jobs

The default emits ONE entry per distinct cadence — a sweep. The job polls every source of that cadence sequentially, in graph order, then drains once.

That is the shape worth defaulting to, because N independent cron entries cannot order themselves: they fire concurrently whatever order the list is in, so a source that needs another to have run first has no way to say so. Inside one job it does — depends_on on a source is a sequencing edge, and scanners/2 sorts by depth.

# a source that must be polled after :observations
reactive do
  id :verdicts
  leaf? true
  poll MyApp.Sources.Verdicts, every: "0 * * * *"
  depends_on [:observations]
end

per_cell: true opts back into one entry per source, for a host that wants independent jobs — separate queues, separate retry policies — and accepts that they are unordered. order: then applies to the emitted list, which is a reading convenience rather than a happens-before. 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.

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

Ordering

Sources are polled SEQUENTIALLY, in cell order. A host whose sweep order carries meaning — observers before the nodes that read them — declares it:

Source.poll_all(plan, order: [:observations, :verdicts])

Anything unlisted follows, in cell order. Unlike crontab/3's order:, this one is a real happens-before: these polls run one after another in this process, so a later source genuinely sees what an earlier one wrote.

Telemetry

[:reactive_dag, :scan, :source_stop] fires as each source finishes, with %{source: module, result: result} — including the failures, since a source that could not run is a thing a host wants recorded.

Sources are polled one at a time, so this doubles as PROGRESS: a sweep over eight sources can run for minutes, and without it the only observable moments are the beginning and the end of the whole run.

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

@spec standing_args(graph()) :: %{required(module()) => keyword()}

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.

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

Verify a poll Mod declaration on cell_id: the module must be a loadable ReactiveDag.Source.

That is the whole check now. It used to also assert the module's own leaf_cells/1 claimed this cell — a check that existed only because the pairing was written twice, and whose error message asked you to work out which copy was stale. The cells a source feeds are plan.parents[id], so there is one declaration and nothing to disagree with.