ergon_pgmq_runner (ergon v0.5.0)

View Source

Executes 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

args()

-type args() :: #{queue := binary(), handler := pgmq_handler(), handler_timeout := timeout()}.

pg_null()

-type pg_null() :: null.

pgmq_handler()

-type pgmq_handler() :: fun((pgmq_message()) -> ok | {error, binary()}).

pgmq_message()

-type pgmq_message() ::
          #{id := non_neg_integer(),
            read_ct := non_neg_integer(),
            message := json:decode_value(),
            headers := json:decode_value() | pg_null()}.

Functions

handle_call(Request, From, State)

handle_cast/2

init/1

-spec init(args()) -> {ok, args()}.

start_link(Args)

-spec start_link(args()) -> {ok, pid()} | {error, term()}.