ergon_queue_sup (ergon v0.5.0)

View Source

Everything 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 poller

rest_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

handler()

-type handler() :: fun((job()) -> ok | {error, binary()}).

job()

-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()}.

job_state()

-type job_state() :: available | executing | completed | failed | discarded.

pg_null()

-type pg_null() :: null.

pg_timestamp()

-type pg_timestamp() :: {calendar:date(), {0..23, 0..59, number()}} | infinity | '-infinity'.

queue()

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

Functions

init/1

-spec init({queue(), handler()}) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.

pool_name(QueueName)

-spec pool_name(binary()) -> atom().

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.

start_link(Queue, Handler)

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