The Oban job that polls one scanner and drains what it changed.
Every host that scans grew this worker independently, and each time it was the same five lines of engine logic wrapped in that host's own observability: poll, normalise the return shape, mark the frontier, propagate to parents, drain. Providing it means a host schedules scans without re-deriving the loop — and gets the poll/drain split right by construction rather than by reading the guide.
Scheduling it
ReactiveDag.Source.crontab/2 reads the cadence each leaf declared and emits
entries for Oban's cron plugin:
config :my_app, Oban,
queues: [scans: 1],
plugins: [
{Oban.Plugins.Cron,
crontab: ReactiveDag.Source.crontab(MyApp.Dag.plan(), ReactiveDag.ScanWorker)}
]A single-concurrency :scans queue is the usual choice: two concurrent polls
of the same upstream are wasted requests, and the drain is cheaper batched.
Running one on demand
%{"cell" => "agenda_docs"} |> ReactiveDag.ScanWorker.new() |> Oban.insert()or with a wider bound than the leaf's standing default:
%{"cell" => "agenda_docs", "opts" => %{"recent" => false}}
|> ReactiveDag.ScanWorker.new()
|> Oban.insert()What it does not own
The plan, because a job argument cannot carry one — see :plan_mfa below.
Domain observability — auditing crawls, recording run ids, enqueuing follow-up work. Not because those belong outside a library, but because this module is a convenience over two public calls and has no opinion about them.
Most of that needs one thing: the loop finished, here is what happened.
[:reactive_dag, :scan, :stop] carries a ReactiveDag.ScanRun under run
— the poll and the drain it triggered as ONE value, which is what a
broadcast, a durable scan record or a follow-up enqueue actually wants. The
same facts are also present as flat keys (cell, args, unreachable,
detail, report) for handlers written before the struct existed.
ScanRun.total/2 is the one that needed a value rather than a payload: a
run's cost lives in BOTH phases — the crawl's own spend in detail, its
downstream recomputes' in the report's steps — and adding them was left to
every caller. :start covers the same work at
the other end, for anything a person watches while the poll runs. Anything inside the poll itself — wrapping each HTTP request,
mirroring listing pages — belongs in your poll/1, which the library never
looks inside.
Where that is not enough, call ReactiveDag.Source.refresh/3 and
ReactiveDag.Drain.run/2 directly: that is all this module does. It exists
to save you writing the loop, not to stop you writing a different one.
Watching a sweep
A sweep is one job that can run for minutes, so :start and :stop bracket
the whole run and say nothing about what is happening inside it.
[:reactive_dag, :scan, :source_stop] fires as each source finishes,
carrying that source's own result — which source went, how long it took, and
what it found. That is the progress signal and the per-source record both.
:stop then carries results (%{module => result}) as well as the
aggregate, so a host that only wants the end state has it in one payload.
Three outcomes
| the scanner returns | the job | why |
|---|---|---|
{:ok, result} | :ok | it looked |
{:error, :not_scannable} | {:cancel, reason} | it cannot look, and retrying will not change that |
{:error, {:not_scannable, why}} | {:cancel, reason} | …and it can say why |
{:error, reason} | {:error, reason} | it failed; a retry might work |
A source with no credential configured, or an integration not enabled for
this tenant, is not a fault: retrying cannot conjure a missing credential,
and burning every attempt to land in discarded reads as "something is
broken" when the honest answer is "this was never going to work".
The judgement belongs to the SCANNER, because only it knows the difference between an upstream that is down and one that was never configured.
:stop still fires for an unscannable source, carrying not_scannable: in
its metadata — it is a completed scan that found nothing, and a host
recording scan results wants the row. An outage is not a quiet success, and
neither is a missing credential.
Telemetry
| event | measurements | metadata |
|---|---|---|
[:reactive_dag, :scan, :start] | system_time | cell, args |
[:reactive_dag, :scan, :stop] | duration_us, changed, passes | cell, args, unreachable, detail, report, run |
[:reactive_dag, :scan, :exception] | duration_us | cell, args, reason |
[:reactive_dag, :scan, :source_stop] | duration_us | source, result |
[:reactive_dag, :scan, :progress] | done, total | cell, label, source |
:progress comes from a SCANNER, via ReactiveDag.Source.progress/3, and is
the only signal from inside one poll: a crawl of 700 documents is otherwise a
single :source_stop that fires once it is already over. Nothing emits it
unless a scanner chooses to.
:source_stop fires once per source inside a sweep, as it finishes. It comes
from Source.poll_all/2 rather than this worker, so a host calling that
directly gets the same signal.
A poll can run for minutes, so :start is what lets a page show a crawl as
in-flight rather than appearing only once it is over.
args is the job's own arguments, verbatim. A scan is often one leg of a
RUN whose id the enqueuer chose — crontab/3 takes args: for exactly this
— and only the job carries it. A handler that sees cell alone knows which
cell finished and not which run it belonged to, so it cannot write the row,
address the broadcast or group the trace, and the work has to fork this
worker instead of attaching to it.
The drain inside emits its own events, so a host attaching to
[:reactive_dag, :drain, :stop] sees the recompute trace without attaching
here at all.