ergon_pgmq_runner (ergon v0.5.0)
View SourceExecutes one pgmq message and reports the result back to its consumer.
worker_pool starts concurrency of these per queue. Unlike ergon_job_runner,
this one does not persist anything: acknowledgement is archiving, archiving is a
batch operation, and the consumer is the only process that knows when a batch is
complete. So the runner's whole job is to run the handler and answer.
The handler runs in a child process under a deadline, for the same reasons as on
the job side: a handler is host code that may return garbage, raise, exit, or
hang, and all four have to cost one message rather than an executor.
spawn_request/2 with its own reply and monitor tags keeps those two replies
distinct from the wpool protocol traffic sharing this mailbox.
A timeout here means the visibility timeout has effectively run out (that is the default deadline), so the message is already deliverable again and possibly already in another consumer's hands. Reporting it as failed is not merely convenient, it is the only correct answer: archiving afterwards would remove a message someone else is still working on.
Summary
Types
-type args() :: #{queue := binary(), handler := pgmq_handler(), handler_timeout := timeout()}.
-type pg_null() :: null.
-type pgmq_handler() :: fun((pgmq_message()) -> ok | {error, binary()}).
-type pgmq_message() :: #{id := non_neg_integer(), read_ct := non_neg_integer(), message := json:decode_value(), headers := json:decode_value() | pg_null()}.