ergon_queue_sup (ergon v0.5.0)
View SourceEverything one call to ergon:start_worker/2 creates: an executor pool and the
poller that feeds it.
ergon_queue_sup (rest_for_one)
|- wpool concurrency x ergon_job_runner
\- ergon_worker the pollerrest_for_one and this order together encode the dependency. The poller casts
into the pool, so the pool must exist first, and if it dies the poller has to
come back with it. Otherwise the poller would go on casting into a name that no
longer resolves, checking out jobs that nothing would run. The reverse is not
true: a poller crash leaves the pool's in-flight jobs to finish normally.
The in-flight counter
Created here rather than in either child, because both need the same reference
and it has to outlive a restart of either. ergon_worker reads it to size each
checkout; ergon_job_runner decrements it when a job finishes.
One consequence worth knowing: if the pool restarts, jobs its runners were
executing die with it and their slots are never returned, so the counter drifts
upward and the queue's effective concurrency drops. rest_for_one handles this
by restarting the poller too, but the counter itself is only reset when this
supervisor restarts. That is the intended blast radius: a pool that crashes
repeatedly should take its queue down rather than silently throttle it.
Summary
Functions
The registered name of a queue's executor pool.
Types
-type job() :: #{id := ergon_job:job_id(), queue := binary(), worker := binary(), payload := json:decode_value(), state := job_state(), fingerprint := binary(), attempt := ergon_job:attempt(), max_attempts := pos_integer(), last_error := binary() | pg_null(), scheduled_at := pg_timestamp(), inserted_at := pg_timestamp()}.
-type job_state() :: available | executing | completed | failed | discarded.
-type pg_null() :: null.
-type pg_timestamp() :: {calendar:date(), {0..23, 0..59, number()}} | infinity | '-infinity'.
-type queue() :: #{name := binary(), poll_interval := pos_integer(), batch_size := pos_integer(), concurrency := pos_integer(), handler_timeout := timeout()}.
Functions
-spec init({queue(), handler()}) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
The registered name of a queue's executor pool.
Derived from the queue name, so it is stable across restarts and predictable
from the outside for wpool:stats/1. binary_to_atom/2 is safe here because
queue names come from the host's own configuration, not from the database.