ergon_queue (ergon v0.5.0)

Runtime configuration for a single named queue: how often a worker polls it and how many jobs it takes per poll. Build with new/1 plus the setters so call sites are unaffected when a new tuning knob is added.

The default poll_interval is 5 s. With ergon_job_notifier running and pg_cron installed, a runnable job wakes its workers over LISTEN/NOTIFY within the ~1 s notifier tick, so the poll is only the fallback that covers the boot gap and reconnect windows. Where pg_cron is absent no NOTIFY ever fires and the poll is the only wake path, which is why the default stays moderate rather than long. Raise it with with_poll_interval/2 on pg_cron-backed deployments, or lower it if the notifier is disabled and latency has to come from polling alone.

Summary

Types

queue()

Functions

new(Name)

A queue named Name: polled every 5 s, one job per checkout, one handler at a time, and no handler deadline.

with_batch_size(Queue, BatchSize)

Set the maximum number of jobs checked out per poll.

with_concurrency(Queue, Concurrency)

Set how many of this queue's handlers may run at once. Sizes the executor pool.

with_handler_timeout/2

Set a per-job deadline, in milliseconds, after which a running handler is killed.

with_poll_interval(Queue, Milliseconds)

Set how long the worker waits between polls, in milliseconds.

Types

queue()

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

Functions

new(Name)

-spec new(binary()) -> queue().

A queue named Name: polled every 5 s, one job per checkout, one handler at a time, and no handler deadline.

The defaults make a worker behave as a single sequential drain, so raising concurrency is an explicit opt-in rather than something a host discovers by surprise.

with_batch_size(Queue, BatchSize)

-spec with_batch_size(queue(), pos_integer()) -> queue().

Set the maximum number of jobs checked out per poll.

This is a round-trip optimisation, not a concurrency setting: a poll fetches at most this many, and never more than the executor pool has free capacity for.

with_concurrency(Queue, Concurrency)

-spec with_concurrency(queue(), pos_integer()) -> queue().

Set how many of this queue's handlers may run at once. Sizes the executor pool.

Raising this is the in-process way to add throughput. Starting several workers on the same queue, or on several nodes, is the other, and the two compose: checkout uses FOR UPDATE SKIP LOCKED, so no two executors anywhere can take the same job.

with_handler_timeout/2

-spec with_handler_timeout(queue(), timeout()) -> queue().

Set a per-job deadline, in milliseconds, after which a running handler is killed.

The job is then recorded as errored rather than abandoned, so it consumes an attempt and the database's jittered backoff reschedules it. Without a deadline (the infinity default) a handler that hangs occupies an executor slot until the node restarts, and its job stays executing for the reconciler to find.

with_poll_interval(Queue, Milliseconds)

-spec with_poll_interval(queue(), pos_integer()) -> queue().

Set how long the worker waits between polls, in milliseconds.