Zizq.Worker (Zizq v0.6.0)

Copy Markdown View Source

Takes jobs from a queue and runs them.

Add one to your supervision tree alongside a client:

children = [
  {Zizq, name: MyApp.Zizq, url: "http://localhost:7890"},
  {Zizq.Worker,
   client: MyApp.Zizq,
   queues: ["emails"],
   concurrency: 25,
   handler: &MyApp.handle_job/1}
]

The handler

A one-argument function receiving a Zizq.Job. What it returns decides what happens to the job:

ReturnsOutcome
:ok or {:ok, term}acknowledged as complete
{:error, reason}failed; retried per the backoff policy
{:cancel, reason}killed now, whatever retries remain
{:snooze, milliseconds}retried after that long, ignoring backoff
{:snooze, %DateTime{}}retried at that time, ignoring backoff
raises, exits, or is killedfailed, with the exception and stacktrace

Snooze durations are milliseconds, as every duration in this client is, so {:snooze, :timer.minutes(5)} says plainly what {:snooze, 300_000} means. Note that if you are coming from Oban, Oban's equivalent is in seconds; take care if porting jobs from Oban to Zizq.

Anything else is acknowledged as complete and logged as a warning naming what was returned. Failing the job instead would be worse — the handler most likely did its work and ended on the wrong value, e.g. due to a Logger call, so failing would re-run a side effect that already happened. Return :ok explicitly to keep the log quiet.

Isolation

Each job runs in its own supervised task. A handler that raises, exits, or is killed outright cannot take the worker down — the crash is reported to the server as a failure and the next job starts. A slow handler cannot block the stream from being read either.

Shutdown

On shutdown the worker stops taking new work, waits for running jobs to finish, flushes outstanding acknowledgements, and only then closes the connection — in that order, so the server keeps the in-flight jobs until the acknowledgements arrive. Anything still running when :drain_timeout expires is abandoned and redelivered later.

Summary

Functions

Start a new worker.

Functions

start_link(opts)

@spec start_link(keyword()) :: Supervisor.on_start()

Start a new worker.

Options

  • :client (atom/0) - Required. Name of a running Zizq client.

  • :handler - Required. What to run for each job: a function taking a Zizq.Job, or a Zizq.Router to dispatch by type. A router is compiled to a function once, when the worker starts.

  • :name (atom/0) - Name for this worker. Defaults to the module.

  • :queues (list of String.t/0) - Queues to take from. Empty means every queue. The default value is [].

  • :concurrency (pos_integer/0) - How many jobs may run at once. The default value is 10.

  • :prefetch (pos_integer/0) - Maximum unacknowledged jobs the server will send. Defaults to twice :concurrency, so a replacement job is already waiting when one finishes rather than costing a round trip.

  • :worker_id (String.t/0) - Identifies this worker in the server's logs. Assigned by the server if omitted.

  • :drain_timeout (pos_integer/0) - Milliseconds to allow on shutdown for running jobs to finish and acknowledgements to be flushed.

    This is the whole budget, not a per-step one: stopping the worker will not take materially longer than this, so it can be set from whatever deadline a deployment or orchestrator imposes. Jobs still running when it expires are abandoned locally, and redelivered by the server upon worker disconnection.

    The default value is 30000.