ergon_pgmq (ergon v0.5.0)
View SourceThin wrappers over pgmq, plus the queue and notification management a host needs.
pgmq is Ergon's other queue, and it is worth being clear about which is which.
ergon.jobs is the bi-temporal job table: retries, workflow dependencies,
history, one row per job. pgmq is a durable message transport: a host creates
its own queues on it and streams through them at volume, and delivery is
at-least-once by virtue of visibility timeouts rather than state transitions.
Nothing here touches ergon.jobs.
ergon_pgmq_consumer is the read side; ergon_reconciler and ergon_health
(Phase 6) are the operational side. Every query lives in priv/queries/pgmq/
and runs through ergon_sql, so nothing here builds SQL, except the
*_sql/1,2 forms at the bottom, which exist precisely to hand SQL to a host's
own migration.
Delivery contract
read/3 hides each message behind a visibility timeout. A message not archived
before that timeout expires becomes visible again and is redelivered. So
archive means acknowledge, and a message must be archived only after its
handler succeeded. Failing to archive is how a failure is reported; there is no
negative acknowledgement to send.
read_ct counts how many times a message has been delivered. Nothing here
dead-letters a message that keeps failing, since pgmq has no such concept, so a
handler that wants to give up on a poison message must check read_ct itself.
Sending inside your own transaction
send/2,3 and send_topic/2,3 are ordinary queries on Ergon's pool, so calling
one inside ergon_repo:transaction/1 enlists it in that transaction: pgo
binds the transaction's connection in the process dictionary and every nested
query rides it. That makes a message and the rows that justify it commit or roll
back together:
ergon_repo:transaction(fun() ->
{ok, _} = ergon_repo:query("UPDATE accounts SET balance = balance - $1 WHERE id = $2",
[Amount, From]),
{ok, _} = ergon_pgmq:send(~"payments", #{~"from" => From, ~"amount" => Amount})
end).This is the transactional outbox pattern with no outbox: there is no window in
which the balance moved but the message was lost, and none in which the message
went out for a transfer that rolled back. The usual machinery (an outbox table,
a poller, a reconciler for drift) exists to close a gap that does not open when
the queue lives in the same database as the data. The same holds for
ergon:enqueue/1.
Summary
Functions
Acknowledge MsgIds on Queue.
Bind a routing pattern to a queue.
Create a pgmq queue and its archive table.
The statement create_queue/1 runs, as literal SQL.
The channel Queue is notified on by default: pgmq_<queue>.
Stop notifying for Queue. Its consumers fall back to polling.
The statement disable_notify/1 runs, as literal SQL.
Drop a pgmq queue, its archive table, and any notification registration.
The statement drop_queue/1 runs, as literal SQL.
Register Queue for wake-up notifications on its default channel.
Like enable_notify/1 with an explicit channel. Re-registering moves it.
The statement enable_notify/1 runs, as literal SQL.
The statement enable_notify/2 runs, as literal SQL.
The header that places a message in a FIFO group.
Every pattern bound to Queue.
Health snapshot of one queue: total length, visible length, and the age of the oldest message.
Read up to Limit messages from Queue, hiding each behind a visibility
timeout of VtSeconds.
Like read/4 with driver options, which is how a long-polling consumer pins the
call to its own connection pool.
Force-expire every in-flight visibility lease on Queue, making the held
messages immediately re-readable. Returns how many were released.
Send Message to Queue. Returns its message id.
Like send/2 with headers.
Publish by routing key, fanning out to every queue whose bound pattern matches. Answers how many queues received it.
Like send_topic/2 with headers, e.g. a FIFO group.
Remove a binding. Answers whether one was actually removed.
Types
-type db_error() :: empty_result | would_create_cycle | {job_not_found, ergon_job:job_id()} | {pgo_error, map()} | term().
-type pg_null() :: null.
-type pg_timestamp() :: {calendar:date(), {0..23, 0..59, number()}} | infinity | '-infinity'.
-type pgmq_message() :: #{id := non_neg_integer(), read_ct := non_neg_integer(), message := json:decode_value(), headers := json:decode_value() | pg_null()}.
-type pgmq_metrics() :: #{queue_length := non_neg_integer(), queue_visible_length := non_neg_integer(), oldest_msg_age_sec := number() | pg_null()}.
-type pgmq_read_strategy() :: plain | grouped | grouped_head | grouped_rr | {long_poll, MaxSeconds :: pos_integer(), IntervalMs :: pos_integer()} | {long_poll, plain | grouped | grouped_head | grouped_rr, pos_integer(), pos_integer()}.
-type pgmq_topic_binding() :: #{pattern := binary(), queue_name := binary(), bound_at := pg_timestamp(), compiled_regex := binary()}.
Functions
-spec archive(binary(), [non_neg_integer()]) -> {ok, [non_neg_integer()]} | {error, db_error()}.
Acknowledge MsgIds on Queue.
pgmq.archive moves them from pgmq.q_<queue> to the pgmq.a_<queue> audit
table, so a processed message leaves a durable trail rather than vanishing.
Returns the ids actually archived; an id already archived is silently absent.
An empty list short-circuits without a round-trip. That is not just an optimisation: it is the normal case for a batch in which every handler failed, and sending an empty array would cost a query per cycle to archive nothing.
-spec archive(binary(), [non_neg_integer()], query_options()) -> {ok, [non_neg_integer()]} | {error, db_error()}.
Bind a routing pattern to a queue.
* matches exactly one dot-separated segment and # matches zero or more, so
logs.# catches both logs.error and logs.api.error while logs.* catches
only the first. Idempotent: rebinding the same pattern to the same queue is a
no-op.
Create a pgmq queue and its archive table.
The statement create_queue/1 runs, as literal SQL.
The channel Queue is notified on by default: pgmq_<queue>.
Stop notifying for Queue. Its consumers fall back to polling.
The statement disable_notify/1 runs, as literal SQL.
Drop a pgmq queue, its archive table, and any notification registration.
The statement drop_queue/1 runs, as literal SQL.
Register Queue for wake-up notifications on its default channel.
This installs nothing. Ergon's migrations already schedule a single
ergon.notify_pending_pgmq() tick that reads the registry every second and
notifies each registered queue holding a visible message.
One tick rather than one cron job per queue, because every notifying transaction takes the global notification-queue lock at commit. That is the cost that makes trigger-per-insert designs plateau, measured at ~2.9K writes/sec with no resource saturation, against ~60K once notifications were batched into fewer transactions. A per-queue tick would put the notifying-transaction count back in proportion to the queue count; this keeps it at one per second regardless.
Wake latency floor is therefore the tick's 1 s cadence, pg_cron's finest
granularity. Consumers that need lower latency should lower poll_interval
instead; the poll is the durable path in either case.
Like enable_notify/1 with an explicit channel. Re-registering moves it.
The statement enable_notify/1 runs, as literal SQL.
The statement enable_notify/2 runs, as literal SQL.
The header that places a message in a FIFO group.
ergon_pgmq:send(Queue, Payload, ergon_pgmq:group_header(~"customer-42"))-spec list_topic_bindings(binary()) -> {ok, [pgmq_topic_binding()]} | {error, db_error()}.
Every pattern bound to Queue.
compiled_regex is what pgmq actually matches routing keys against, which is
the thing to look at when a binding is not catching what its author expected.
-spec metrics(binary()) -> {ok, pgmq_metrics()} | {error, db_error()}.
Health snapshot of one queue: total length, visible length, and the age of the oldest message.
The difference between the two lengths is messages hidden behind a visibility
lease: in flight, or stranded by a consumer that died. oldest_msg_age_sec is
null on an empty queue.
queue_visible_length is computed against transaction-frozen now(), so a
message sent inside the same transaction reads as invisible. Assert on
queue_length in transactional tests.
-spec metrics(binary(), query_options()) -> {ok, pgmq_metrics()} | {error, db_error()}.
-spec read(binary(), pos_integer(), pos_integer()) -> {ok, [pgmq_message()]} | {error, db_error()}.
Read up to Limit messages from Queue, hiding each behind a visibility
timeout of VtSeconds.
message and headers are jsonb, and the driver's default json configuration
returns those as raw binaries. Both are decoded here, so a handler receives
terms, not JSON text.
-spec read(binary(), pos_integer(), pos_integer(), pgmq_read_strategy() | query_options()) -> {ok, [pgmq_message()]} | {error, db_error()}.
Like read/3 with an explicit strategy.
plain takes whatever is visible, in no particular order. The three grouped
strategies deliver in strict order within each group named by the
x-pgmq-group header, while letting different groups proceed in parallel:
grouped_headtakes at most one message per group, so no group can ever be worked out of order.FOR UPDATE SKIP LOCKEDunderneath, so distinct consumers take distinct groups.grouped_rrround-robins across groups, so a busy group cannot starve others.groupedbatches from the earliest group, favouring throughput over fairness.
Ordering is enforced by the visibility timeout rather than by locking: the next message in a group stays hidden until the one ahead is archived or its lease expires. A message sent without the header joins a default group.
The {long_poll, MaxSeconds, IntervalMs} strategy modifier blocks server-side
until a message arrives. See read/5.
-spec read(binary(), pos_integer(), pos_integer(), pgmq_read_strategy(), query_options()) -> {ok, [pgmq_message()]} | {error, db_error()}.
Like read/4 with driver options, which is how a long-polling consumer pins the
call to its own connection pool.
That pinning is not optional for {long_poll, _, _}: the call blocks
server-side for up to MaxSeconds, holding its connection the whole time, so
issuing it against the shared pool would take a connection out of circulation
for every other query on the node.
-spec release_leases(binary()) -> {ok, non_neg_integer()} | {error, db_error()}.
Force-expire every in-flight visibility lease on Queue, making the held
messages immediately re-readable. Returns how many were released.
The recovery tool for messages stranded by consumers that died mid-processing:
rather than waiting out each visibility timeout individually, free them all at
once. ergon_reconciler is the intended caller.
-spec release_leases(binary(), query_options()) -> {ok, non_neg_integer()} | {error, db_error()}.
-spec send(binary(), json:encode_value()) -> {ok, non_neg_integer()} | {error, db_error()}.
Send Message to Queue. Returns its message id.
-spec send(binary(), json:encode_value(), json:encode_value() | pg_null()) -> {ok, non_neg_integer()} | {error, db_error()}.
Like send/2 with headers.
Headers are how FIFO grouping is expressed: group_header/1 builds the
x-pgmq-group header the grouped read strategies order by.
-spec send(binary(), json:encode_value(), json:encode_value() | pg_null(), query_options()) -> {ok, non_neg_integer()} | {error, db_error()}.
-spec send_topic(binary(), json:encode_value()) -> {ok, non_neg_integer()} | {error, db_error()}.
Publish by routing key, fanning out to every queue whose bound pattern matches. Answers how many queues received it.
Zero is not an error. A routing key nothing is bound to is silently dropped,
which is the behaviour a topic exchange is supposed to have but is also an easy
way to lose messages to a typo. Check the count, or list_topic_bindings/1,
if delivery matters.
The fan-out is one transaction: every delivery succeeds or none does.
-spec send_topic(binary(), json:encode_value(), json:encode_value() | pg_null()) -> {ok, non_neg_integer()} | {error, db_error()}.
Like send_topic/2 with headers, e.g. a FIFO group.
-spec send_topic(binary(), json:encode_value(), json:encode_value() | pg_null(), query_options()) -> {ok, non_neg_integer()} | {error, db_error()}.
Remove a binding. Answers whether one was actually removed.