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. A host that audits its crawls, records a run id, or
enqueues follow-up work wraps this rather than extending it: call
ReactiveDag.Source.refresh/3 and ReactiveDag.Drain.run/2 directly inside
your own worker, which is all this module does. It exists to save you writing
it, not to stop you.
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.