ergon_worker (ergon v0.5.0)
View SourceThe polling loop for one queue.
On each tick a worker checks out a batch of jobs and casts them to its queue's
executor pool, then schedules the next tick. It does not run handlers itself;
ergon_job_runner does, so a slow handler never delays the next poll.
The periodic poll is the reliable path. On init the worker also joins
ergon_worker_registry under its queue name, so ergon_job_notifier can send it
a wake within a tick of a job becoming runnable and it drains immediately
rather than waiting out poll_interval. With the notifier disabled, pg_cron
absent, or a wake lost, the poll still drains everything, only later.
Backpressure
A checked-out job is already executing in the database with an attempt
consumed. Fetching more than the pool can run therefore does not just queue work,
it parks live jobs in a mailbox where a node restart strands them until the
reconciler notices. So each poll fetches at most the pool's free capacity, read
from a shared counters reference, and fetches nothing at all when the pool is
saturated.
That gate is why dispatch can be a cast. wpool:call/4 would supply
backpressure of its own, but it is synchronous per request whatever the strategy,
which would serialise the batch and block this process against wake while a
handler runs.
The tick is scheduled on an absolute deadline
Draining takes time, so scheduling the next tick poll_interval from now
yields a real period of interval + drain_time that drifts under load. The next
deadline is instead advanced by exactly one interval and scheduled with
{abs, true}. When a drain overruns its interval the deadlines collapse to
"immediately" rather than accumulating a backlog of timers.
Summary
Types
-type args() :: #{queue := queue(), pool := atom(), in_flight := counters:counters_ref()}.
-type queue() :: #{name := binary(), poll_interval := pos_integer(), batch_size := pos_integer(), concurrency := pos_integer(), handler_timeout := timeout()}.