Streams messages off a pgmq queue, one batch at a time.

Replaces the Broadway pipeline. What it keeps is the delivery contract; what it drops is the demand protocol, which existed to move backpressure between stages that no longer exist.

The cycle

Read a batch, hand every message to the executor pool, wait for all of them, then archive the ones that succeeded in a single call. Repeat.

That shape is the backpressure, which is why there is no equivalent of ergon_worker's in-flight counter here: exactly one batch is ever outstanding, and nothing is read until the last one is accounted for. Do not add a counter by analogy with the job side. It would be measuring something that cannot exceed one batch.

A full batch means there is probably more waiting, so the next cycle starts immediately rather than sleeping out the poll interval. A short batch means the queue is drained, and the consumer goes back to waiting.

Archive means acknowledge

A message is archived only after its handler returned ok. A failure is reported by not archiving: the visibility timeout expires and pgmq redelivers. There is no negative acknowledgement, and nothing here deletes a message outright.

That is what makes delivery survive a BEAM crash mid-batch. Whatever was in flight was never archived, so it comes back. The cost is that redelivery is at-least-once, so handlers must tolerate seeing a message twice.

Two consequences of batching worth knowing:

• One slow message delays its siblings' archive, since the cycle waits for the whole batch. Bounded by handler_timeout, which defaults to the visibility timeout.
• A message whose handler always fails is redelivered forever. pgmq has no dead-letter queue; read_ct counts deliveries and is the only signal a handler has to give up on one.

Waking up

The poll is always underneath. On top of it, optionally, either a LISTEN subscription through the node's shared ergon_listener, or a server-side long poll that blocks until a message arrives. See ergon_pgmq_queue for which to pick.