The engine, viewed from outside: what the graph LOOKS like, what state each cell is in, and what the last drains actually did.
Everything here is a READ. Nothing new is computed — the plan already carries
the structure and depths, each cell's own resource already carries its rows,
and ReactiveDag.Drain.Report is already a complete causal trace. This module
is those three assembled into the shape a human (or a dashboard, a mix task,
an alerting check) actually asks for.
Per-cell state is read through ReactiveDag.Node.Rows, which reads the node's
resource under the DAG's own cell keys. A node with no :status column
reports its keys with a nil status — the count is still the truth about how
many units the cell holds.
Deliberately UI-free: no Phoenix, no rendering, no assumptions about a web
layer. reactive_dag_dashboard renders this; a host with no web layer at all
can still call summary/1 from a mix task or a health check.
Retaining runs
record/1 keeps the last N runs in a rolling in-memory window, so a host gets
a processing log without writing storage of its own. It takes what the engine
hands back, in either of the two shapes the engine produces:
# a scan — the poll and the drain it triggered
:telemetry.attach("scans", [:reactive_dag, :scan, :stop], fn _e, _m, meta, _ ->
ReactiveDag.Insights.record(meta.run)
end, nil)
# a drain someone triggered directly, with no poll in front of it
{:ok, report} = ReactiveDag.Drain.run(plan, opts)
ReactiveDag.Insights.record(report)The buffer holds %ReactiveDag.ScanRun{} either way. A bare %Report{} is a
drain with no poll around it, so it is wrapped in a run at the boundary rather
than stored as a second shape — recent/1 returns ONE shape, and a consumer
reads run.report for the drain and run.duration_us, run.changed,
run.unreachable, run.detail for the poll without testing which kind of
entry it got.
Why the whole run, and not just the drain
The buffer used to hold %Report{} alone, so a host with a %ScanRun{}
unwrapped it and threw the envelope away. That discarded everything the POLL
did: its wall time (usually most of a scan's — the drain is the cheap half),
the keys it found changed, what it cost, and its unreachable list. A
two-minute scan of an upstream it could not reach logged as 0 cells · 0 changed · 6.1ms, identical to a scan that looked at everything and found
nothing. That is precisely the honest-gap failure ReactiveDag.Source warns
about, reintroduced by the observer.
recent/1's entries therefore carry the run, and polled? says whether there
was a poll at all, so "this drain reported no changes" and "this scan could
not look" stay different sentences.
This is an opt-in observer, not a durable log: the buffer is per-BEAM-node, in-memory, and lost on restart. A host that needs history stores the run where its runs already live.
Summary
Functions
One cell's observable state: its declaration (depth, inputs, shape) plus what it currently holds (status histogram, key count, and a small sample of failing keys for the drawer).
The graph's edges as {from, to} pairs — an input edge points from the input
cell TO the cell that reads it, i.e. the direction change flows.
Forget every retained run.
The most recent run, or nil if none has been recorded.
The graph's shape: cells grouped by depth, in execution order.
Cell ids with dirty keys waiting — what the NEXT drain would work on.
The most recent runs, newest first (default all retained).
Keep a run in the rolling in-memory window (default 20,
configurable with config :reactive_dag, insights_keep: n).
Every cell's state, in execution order — the dashboard's landing view.
Types
@type cell_status() :: %{ id: String.t(), depth: non_neg_integer(), inputs: [String.t()], leaf?: boolean(), op: atom() | nil, statuses: %{required(String.t() | nil) => non_neg_integer()}, key_count: non_neg_integer(), rows: :stored | :elsewhere | :unreadable, failing_sample: [String.t()] }
A cell's observable state.
@type entry() :: %{run: ReactiveDag.ScanRun.t(), at: DateTime.t(), polled?: boolean()}
One retained run, as recent/1 returns it.
run— the%ReactiveDag.ScanRun{}. Always present, always this struct, whichever shape was recorded.at— whenrecord/1was called (wall clock, for ordering a log).polled?— was there a poll?truefor a scan,falsefor a bare drain recorded on its own. The one field that tells the two apart, so nothing has to infer it from a nilcellor an emptychanged.
The drain is run.report — nil when a scan never drained (see
ReactiveDag.ScanRun.drained?/1), so a consumer reading it must handle nil.
Functions
@spec cell_status(ReactiveDag.Plan.t(), String.t()) :: cell_status() | nil
One cell's observable state: its declaration (depth, inputs, shape) plus what it currently holds (status histogram, key count, and a small sample of failing keys for the drawer).
Reads the node's own resource, and pushes the reduction into the datastore: a
histogram is a DISTINCT plus one COUNT per status, and the failing sample
is a filtered LIMIT. No row is loaded to be counted, which matters most for
a node whose payload is a blob — decoding it to discard it was the bulk of the
old cost.
Still several small queries per cell, so summary/1 over a large graph is
many round trips. That is the remaining cost, and it is a different one.
@spec edges(ReactiveDag.Plan.t()) :: [{String.t(), String.t()}]
The graph's edges as {from, to} pairs — an input edge points from the input
cell TO the cell that reads it, i.e. the direction change flows.
@spec forget_runs() :: :ok
Forget every retained run.
@spec last_run() :: entry() | nil
The most recent run, or nil if none has been recorded.
@spec levels(ReactiveDag.Plan.t()) :: [{non_neg_integer(), [ReactiveDag.Cell.t()]}]
The graph's shape: cells grouped by depth, in execution order.
Depth is the longest path from a leaf, so every cell in a level can only depend on shallower ones — which is exactly the order the drain runs them in, and therefore the order worth drawing them in.
@spec pending(ReactiveDag.Plan.t()) :: [String.t()]
Cell ids with dirty keys waiting — what the NEXT drain would work on.
Reads the frontier rather than the nodes themselves: a cell is pending because something dirtied it, whether or not its rows have changed yet.
@spec recent(pos_integer() | :all) :: [entry()]
The most recent runs, newest first (default all retained).
Each entry is a entry/0: the %ReactiveDag.ScanRun{}, when it was
recorded, and whether a poll produced it.
for %{run: run, at: at, polled?: polled?} <- Insights.recent(20) do
%{
at: at,
kind: if(polled?, do: :scan, else: :drain),
# the whole run's wall time: a scan's poll AND its drain
duration_us: run.duration_us,
# the POLL. `cell` is what was scanned; the rest is what it found, what
# it could not reach, and what it cost. All at their empty defaults on
# a bare drain, where there was no poll to report them.
scanned: run.cell,
changed: length(run.changed),
unreachable: run.unreachable,
detail: run.detail,
# the DRAIN — `report` is nil when a scan never drained, so guard it
passes: run.report && run.report.passes,
steps: (run.report && run.report.steps) || [],
# …and either phase's cost, or both summed
tokens_in: ReactiveDag.ScanRun.total(run, :tokens_in)
}
endrun.duration_us is the WHOLE run — for a scan, the poll plus its drain. The
drain's own share is run.report.duration_us, and the gap between the two is
the poll, which is usually the larger number by orders of magnitude. Showing
the drain's figure as the run's is what made a two-minute scan render as
6.1ms.
run.unreachable is [{upstream, reason}] — non-empty means the poll could
NOT see everything it meant to (ReactiveDag.ScanRun.complete?/1), and a
consumer that renders it the same as an empty one is reporting a gap as a
clean run.
@spec record(ReactiveDag.ScanRun.t()) :: ReactiveDag.ScanRun.t()
@spec record(ReactiveDag.Drain.Report.t()) :: ReactiveDag.Drain.Report.t()
Keep a run in the rolling in-memory window (default 20,
configurable with config :reactive_dag, insights_keep: n).
Takes either shape the engine produces:
- a
%ReactiveDag.ScanRun{}— a scan, kept whole. The poll's duration, changed keys, costdetailandunreachablelist are the point: they are most of what a scan did, and unwrapping to the report throws them away. - a bare
%ReactiveDag.Drain.Report{}— a drain someone triggered directly, with no poll in front of it. Wrapped in a%ScanRun{}here sorecent/1returns one shape;polled?on the entry isfalse.
Opt-in: neither the scan nor the drain persists anything on its own. Returns its argument unchanged, so it drops into a pipeline either way:
plan |> ReactiveDag.Drain.run(opts) |> then(fn {:ok, r} -> Insights.record(r) end)@spec summary(ReactiveDag.Plan.t()) :: [cell_status()]
Every cell's state, in execution order — the dashboard's landing view.
One status query per cell. Fine for the graphs this library is built for
(tens of cells); for a very large graph, page it with cell_status/2 instead.