ergon_worker (ergon v0.5.0)

View Source

The 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

args()

-type args() :: #{queue := queue(), pool := atom(), in_flight := counters:counters_ref()}.

queue()

-type queue() ::
          #{name := binary(),
            poll_interval := pos_integer(),
            batch_size := pos_integer(),
            concurrency := pos_integer(),
            handler_timeout := timeout()}.

Functions

handle_call(Request, From, State)

handle_cast(Msg, State)

handle_continue/2

handle_info/2

init/1

-spec init(args()) -> {ok, map(), {continue, poll}}.

start_link(Args)

-spec start_link(args()) -> {ok, pid()} | {error, term()}.