Zizq.Worker.Acker (Zizq v0.6.0)

Copy Markdown View Source

Acknowledges completed work, batching successes into single requests.

Normally started for you by Zizq.Worker. Use it directly only if you are driving Zizq.Stream.Take yourself.

{:ok, acker} = Zizq.Worker.Acker.start_link(client: MyApp.Zizq)

Zizq.Worker.Acker.success(acker, job)
Zizq.Worker.Acker.failure(acker, job, message: "SMTP timeout")

Both are asynchronous: a handler should not wait on a network round trip it has no use for, and acknowledgement is not something the handler can meaningfully recover from.

Batching never delays an acknowledgement

There is no timer. A success is buffered and a flush is queued to this process immediately, which lands behind everything already in the mailbox — so a burst of completions collapses into one request while a lone completion goes out on its own, without waiting for company.

Batching then grows with load rather than with a clock: while a request is in flight the process is busy, so acknowledgements arriving meanwhile accumulate and leave together in the next one.

A flush interval would be actively harmful here. Prefetch is released by acknowledgement, so holding acknowledgements back to fill a batch throttles the throughput the batching is meant to serve.

Why successes batch and failures do not

A success carries nothing but an id, so a hundred of them fit in one request. Since the server's prefetch is released by acknowledgement, batching them directly raises how fast work can be pulled.

A failure carries a message, an error type and a stack trace that belong to one job, and the server responds with that job's new state. There is nothing to combine and failures are lower volume than successes, so failures go one at a time.

Losing acknowledgements is safe, but wasteful

In the case of a crashed worker an unacknowledged job is redelivered after the server detects the dead worker connection, so a dropped acknowledgement costs duplicated work rather than lost work. That is why transient failures are retried with backoff and permanent ones are logged and dropped: retrying a 4xx could only produce the same answer, and blocking on it would stall every acknowledgement behind it.

Summary

Functions

Returns a specification to start this module under a supervisor.

Record a job as failed. Sent on its own.

Send everything buffered and wait for it.

How many successes are waiting to be sent.

Start an acker.

Record a job as completed. Buffered, then sent with others.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

failure(acker, job, opts)

@spec failure(GenServer.server(), Zizq.Job.t() | String.t(), keyword()) :: :ok

Record a job as failed. Sent on its own.

Takes the same options as Zizq.report_failure/3, including :kill and :retry_at.

flush(acker, timeout \\ 5000)

@spec flush(GenServer.server(), timeout()) :: :ok

Send everything buffered and wait for it.

Called during shutdown, while the take stream is still open so the server will still accept the acknowledgements. Returns :ok even if some could not be sent — they will be redelivered, and refusing to shut down over it would be worse.

pending(acker)

@spec pending(GenServer.server()) :: non_neg_integer()

How many successes are waiting to be sent.

start_link(opts)

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

Start an acker.

Options

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

  • :max_batch (pos_integer/0) - Most ids to put in a single request. A cap on request size, not a trigger: a larger buffer is split across requests sent one after another, never held back waiting to fill.

    Rarely reached. A job stays in flight until it is acknowledged, so the buffer cannot exceed the worker's prefetch — this only applies at a prefetch above the cap, or while a backlog is being retried through an outage. It is here to bound request size in those cases rather than to shape ordinary batching.

    The default value is 1000.

  • :name (term/0) - Optional GenServer name.

success(acker, job)

@spec success(GenServer.server(), Zizq.Job.t() | String.t()) :: :ok

Record a job as completed. Buffered, then sent with others.