PGMQ Queues

pgmq turns PostgreSQL into a durable message queue: no Redis, RabbitMQ or background worker, just tables and SQL functions in the pgmq schema. Producers send JSON messages, consumers read them with a visibility timeout, and messages stay in the queue until they are explicitly deleted or archived.

Install

pgcli installs the extension package from the Pigsty repository and runs CREATE EXTENSION IF NOT EXISTS pgmq in the instance’s default database:

pg extension install pgmq -i pg01

To use it from another database in the same instance:

pg psql -i pg01 -- -d appdb
CREATE EXTENSION IF NOT EXISTS pgmq;

Check the installed version:

pg exec -i pg01 "SELECT extversion FROM pg_extension WHERE extname = 'pgmq'"

Quick start

-- Create a queue; running it again on an existing queue is a no-op
SELECT pgmq.create('jobs');

-- Send one message; returns its message id
SELECT pgmq.send('jobs', '{"task": "resize", "image_id": 42}');

-- Read up to 10 messages, invisible to other consumers for 60 seconds
SELECT msg_id, read_ct, message
FROM pgmq.read('jobs', 60, 10);

-- Acknowledge one message
SELECT pgmq.delete('jobs', 1);

Run the same statements through pgcli:

pg exec -i pg01 "SELECT pgmq.send('jobs', '{\"task\": \"resize\"}')"
pg psql -i pg01

Queues on disk

Each queue is a table in the pgmq schema:

A live message row contains msg_id, read_ct (number of deliveries), enqueued_at, vt (visible again at) and message jsonb. Recent versions add last_read_at and headers.

Because messages are rows, pgmq.send participates in transactions: it commits or rolls back with the surrounding statement. Queue traffic is also ordinary table traffic — high send/delete rates bloat the table like any other.

Sending messages

SELECT pgmq.send('jobs', '{"task": "resize"}');

-- Visible only after 30 seconds
SELECT pgmq.send('jobs', '{"task": "cleanup"}', 30);

-- Batch send: one statement, one atomic insert
SELECT pgmq.send_batch('jobs', ARRAY[
  '{"task": "a"}',
  '{"task": "b"}',
  '{"task": "c"}'
]::jsonb[]);

send returns the new msg_id; send_batch returns one id per message. Use send_batch instead of looping over send — it is one round trip and the whole batch is atomic.

Payloads must be valid JSON and are stored as jsonb: keys are reordered and duplicates removed. Keep large blobs out of the queue.

Newer versions accept a headers jsonb argument for metadata such as a trace id or a FIFO group key.

Reading and acknowledging

SELECT msg_id, read_ct, enqueued_at, vt, message
FROM pgmq.read('jobs', 60, 10);

read(queue, vt, qty) returns up to qty visible messages and makes them invisible for vt seconds. Reading does not remove anything, so each message must be acknowledged:

SELECT pgmq.delete('jobs', 42);                -- one message
SELECT pgmq.delete('jobs', ARRAY[42, 43, 44]); -- batch

pgmq.pop('jobs') reads and deletes in one call. Use it only when losing an in-flight message on a consumer crash is acceptable.

Avoid a tight polling loop in long-running consumers:

SELECT msg_id, message
FROM pgmq.read_with_poll('jobs', 60, 10, 5, 100);

This waits up to max_poll_seconds (5) for messages, checking every poll_interval_ms (100) milliseconds, and returns as soon as a message arrives.

Visibility timeout and retries

The visibility timeout is PGMQ’s delivery model:

1. read sets vt = now() + timeout and increments read_ct.
2. The message stays invisible to other consumers until vt.
3. delete or archive removes it. If the consumer crashes or does not acknowledge in time, the message becomes visible again after vt and is delivered to another consumer.

Delivery is therefore at-least-once across failures. Choose vt longer than the worst-case processing time, and make consumers idempotent. For long jobs, extend the lease without re-reading:

SELECT pgmq.set_vt('jobs', 42, 300);            -- one message
SELECT pgmq.set_vt('jobs', ARRAY[42, 43], 300); -- batch

read_ct is a ready-made retry counter.

Archive and dead letters

delete is permanent. archive moves a message to pgmq.a_<queue> for retention, audit or replay:

SELECT pgmq.archive('jobs', 42);
SELECT pgmq.archive('jobs', ARRAY[42, 43]);
SELECT * FROM pgmq.a_jobs;

A simple dead-letter pattern is to archive messages that failed too many times:

-- Move messages that have been delivered 5 or more times to the archive
SELECT pgmq.archive('jobs', ARRAY(
  SELECT msg_id FROM pgmq.q_jobs WHERE read_ct >= 5
));

Without a rule like this, a poison message is redelivered forever.

Monitoring

SELECT * FROM pgmq.list_queues();
SELECT * FROM pgmq.metrics('jobs');
SELECT * FROM pgmq.metrics_all();

metrics reports queue_length, queue_visible_length, the age of the newest and oldest messages, and total_messages. Older extension versions return a smaller set of columns.

The queue table can be queried directly:

SELECT count(*) FROM pgmq.q_jobs;

SELECT msg_id, read_ct, enqueued_at, vt, message
FROM pgmq.q_jobs
ORDER BY msg_id DESC
LIMIT 5;

Alert when queue_length keeps growing or oldest_msg_age_sec climbs: either consumers are down or they cannot keep up.

Maintenance

SELECT * FROM pgmq.list_queues();

SELECT pgmq.create_unlogged('scratch');                    -- no WAL, lost on crash
-- Partitioned queues require pg_partman: pg extension install pg_partman -i pg01
SELECT pgmq.create_partitioned('events', '10000', '100000');
SELECT pgmq.purge_queue('jobs');                           -- delete all messages, keep queue
SELECT pgmq.drop_queue('jobs');                            -- queue and archive tables

• create_unlogged trades crash safety for write speed. Use it only for data that can be regenerated.
• create_partitioned requires the pg_partman extension. It keeps a very high-throughput queue from bloating a single table; old partitions are dropped automatically.
• purge_queue empties a queue but keeps it; drop_queue removes it entirely.
• Queue names are limited to 47 characters.

With ordinary queues, heavy send/delete traffic produces dead tuples. Make sure autovacuum keeps up, or archive messages instead of deleting them and consider partitioned queues at high volume.

Permissions

PGMQ functions run with the caller’s privileges. An application role needs access to the schema, the functions and the queue tables:

GRANT USAGE ON SCHEMA pgmq TO app;
GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA pgmq TO app;
GRANT SELECT, INSERT, UPDATE, DELETE ON pgmq.q_jobs TO app;
GRANT SELECT, INSERT ON pgmq.a_jobs TO app;   -- only when archiving

Run pgmq.create as the queue owner; creating tables requires privileges the application role usually should not have. New queues need their own grants, or you can pre-configure default privileges.

Function reference

Troubleshooting

extension "pgmq" is not available. The extension package is not in the instance image, or the extension has no build for this database. Install it with pg extension install pgmq -i <instance> and create it in the target database.

Messages are redelivered. Processing took longer than vt, or the consumer exited before delete. Increase vt, call set_vt during long jobs, and make processing idempotent.

Queue length never drops. Consumers are failing or too slow. Inspect read_ct, application logs and metrics().oldest_msg_age_sec. PGMQ retries failed messages forever unless they are archived or deleted.

Performance degrades over time. Default queues are ordinary tables. Check dead tuples and autovacuum, batch sends and reads, and consider create_partitioned for sustained high throughput.

permission denied for schema pgmq. Grant USAGE on the schema and EXECUTE on the functions to the application role, plus table privileges on each queue it uses.