ergon_worker_sup (ergon v0.5.0)

View Source

The dynamic supervisor queue workers run under.

Workers are added at runtime with start_worker/2, exposed through ergon:start_worker/2, rather than being listed statically: which queues to drain, with what concurrency, and with which handler is the host application's decision, not the library's.

Each call adds one ergon_queue_sup (a pool and its poller) supervised independently, so one queue collapsing does not disturb the others.

transient restart: a queue tree that exits abnormally comes back, but one shut down deliberately stays down.

Summary

Types

handler()
job()
job_state()
pg_null()
pg_timestamp()
queue()

Functions

init/1
start_link()
start_worker/2

Start a supervised worker draining Queue, running Handler on each job.

stop_worker(Pid)

Stop a worker started by start_worker/2, along with its 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([]) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.

start_link()

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

start_worker/2

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

Start a supervised worker draining Queue, running Handler on each job.

Returns the pid of the queue's supervisor, which is what stop_worker/1 takes. Calling this twice for the same queue is allowed and useful: the two workers race to drain and FOR UPDATE SKIP LOCKED keeps them from colliding. Their executor pools would collide on a registered name, though, so prefer raising concurrency for in-process parallelism and reserve a second worker for a genuinely separate configuration.

stop_worker(Pid)

-spec stop_worker(pid()) -> ok | {error, not_found}.

Stop a worker started by start_worker/2, along with its executor pool.