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 the cell, the changed count, the
unreachable list and the whole %Report{}, so a broadcast, a durable scan
record, or a follow-up enqueue is a telemetry handler rather than a fork of
this worker. 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.
Telemetry
Emits [:reactive_dag, :scan, :stop] with changed and passes, and
[:reactive_dag, :scan, :exception] on failure. 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.