pgque-postgres-queue

# PgQue – Zero-Bloat Postgres Queue > Skill by [ara.so](https://ara.so) — Daily 2026 Skills collection. PgQue is a pure PL/pgSQL reimplementation of the battle-tested PgQ (Skype/Postgres) architecture. It uses **snapshot-based batching** and **TRUNCATE-based table rotation** instead of row-level locking, delivering zero dead-tuple bloat, predictable performance under sustained load, and native fan-out — all from a single SQL file on any Postgres 14+ instance including managed providers (RDS, Aurora, Cloud SQL, Supabase, Neon). ## Key Concepts - **Tick**: A periodic snapshot that closes a batch of events. Nothing is delivered until a tick fires. - **Batch**: A group of events captured in one tick, consumed atomically by a subscriber. - **Subscriber/Consumer**: A named cursor on the event log. Multiple consumers get independent copies of every batch (fan-out). - **Zero bloat**: Events are stored in rotating tables and cleared via `TRUNCATE`, never `DELETE`. No dead tuples. - **Latency trade-off**: End-to-end delivery is ~1–2 s (one tick interval + poll). Per-call function latency is microseconds. ## Installation ### Requirements - Postgres 14+ - `pg_cron` (recommended) **or** an external scheduler calling `pgque.ticker()` every second ### Install from SQL file ```bash # Clone the repo git clone https://github.com/NikolayS/pgque.git cd pgque # Install in a single transaction PAGER=cat psql --no-psqlrc --single-transaction -d mydb -f sql/pgque.sql ``` Or inside a `psql` session: ```sql begin; \i sql/pgque.sql commit; ``` ### Start the ticker (pg_cron) ```sql -- Creates pg_cron jobs for ticker (every 1s) and maintenance (every 30s) select pgque.start(); ``` ### Start the ticker (without pg_cron) Run these externally on a schedule: ```bash # Every 1 second psql -d mydb -c "select pgque.ticker()" # Every 30 seconds psql -d mydb -c "select pgque.maint()" ``` > **Warning**: Without a running ticker, consumers see nothing. Enqueue works, but no batches are created. ### Uninstall ```sql \i sql/pgque_uninstall.sql ``` ## Roles & Grants | Role | Use | |---|---| | `pgque_reader` | Dashboards, metrics, read-only | | `pgque_writer` | Producers and consumers (most apps) | | `pgque_admin` | Operators, migrations | ```sql -- Grant producer/consumer access to app user CREATE USER app_worker WITH PASSWORD '...'; GRANT pgque_writer TO app_worker; -- Grant read-only metrics access CREATE USER metrics_reader WITH PASSWORD '...'; GRANT pgque_reader TO metrics_reader; ``` ## Core API (Modern Style) ### Create a Queue ```sql SELECT pgque.create_queue('orders'); ``` ### Subscribe a Consumer ```sql -- Register a named consumer on the queue SELECT pgque.subscribe('orders', 'order-processor'); ``` ### Send Events (Enqueue) ```sql -- Send a single event (type, data) SELECT pgque.send('orders', 'new_order', '{"order_id": 42, "amount": 99.99}'); -- Send a batch of events SELECT pgque.send_batch('orders', ARRAY[ ROW('new_order', '{"order_id": 43}')::pgque.event_data, ROW('new_order', '{"order_id": 44}')::pgque.event_data ]); ``` ### Receive and Acknowledge Events ```sql -- Receive next batch for a consumer (returns batch_id + events) SELECT * FROM pgque.receive('orders', 'order-processor'); -- Acknowledge successful processing (batch_id from receive) SELECT pgque.ack('orders', 'order-processor', :batch_id); -- Negative-acknowledge (retry / dead-letter) SELECT pgque.nack('orders', 'order-processor', :batch_id); ``` ### Unsubscribe ```sql SELECT pgque.unsubscribe('orders', 'order-processor'); ``` ## Low-Level PgQ API These map directly to the original PgQ primitives and are also available via `pgque_writer`: ```sql -- Enqueue a single event SELECT pgque.insert_event('orders', 'new_order', '{"order_id": 42}'); -- Register a consumer SELECT pgque.register_consumer('orders', 'order-processor'); -- Get the next available batch ID SELECT pgque.next_batch('orders', 'order-processor'); -- Returns: batch_id (bigint), or NULL if nothing ready -- Fetch all events in a batch SELECT * FROM pgque.get_batch_events(:batch_id); -- Returns: ev_id, ev_time, ev_txid, ev_retry, ev_type, ev_data, ev_extra1..4 -- Mark batch as successfully processed SELECT pgque.finish_batch(:batch_id); -- Schedule an event for retry (with delay in seconds) SELECT pgque.event_retry(:batch_id, :ev_id, 60); -- retry in 60s -- Unregister consumer SELECT pgque.unregister_consumer('orders', 'order-processor'); ``` ## Complete Working Example ### Producer (Python with psycopg2) ```python import psycopg2 import json import os conn = psycopg2.connect(os.environ["DATABASE_URL"]) conn.autocommit = False def enqueue_order(order_id: int, amount: float): with conn.cursor() as cur: cur.execute( "SELECT pgque.send(%s, %s, %s)", ("orders", "new_order", json.dumps({"order_id": order_id, "amount": amount})) ) conn.commit() enqueue_order(42, 99.99) ``` ### Consumer (Python with psycopg2) ```python import psycopg2 import psycopg2.extras import json import os import time conn = psycopg2.connect(os.environ["DATABASE_URL"]) conn.autocommit = False QUEUE = "orders" CONSUMER = "order-processor" def setup(): with conn.cursor() as cur: cur.execute("SELECT pgque.subscribe(%s, %s)", (QUEUE, CONSUMER)) conn.commit() def process_batch(): with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: cur.execute("SELECT * FROM pgque.receive(%s, %s)", (QUEUE, CONSUMER)) rows = cur.fetchall() if not rows: conn.rollback() return False batch_id = rows[0]["batch_id"] for row in rows: event = json.loads(row["ev_data"]) print(f"Processing order {event['order_id']}") # ... your processing logic ... with conn.cursor() as cur: cur.execute("SELECT pgque.ack(%s, %s, %s)", (QUEUE, CONSUMER, batch_id)) conn.commit() return True setup() while True: if not process_batch(): time.sleep(1) # wait for next tick ``` ### Fan-Out Example (Multiple Independent Consumers) ```sql -- One queue, multiple consumers each get ALL events independently SELECT pgque.create_queue('user-events'); SELECT pgque.subscribe('user-events', 'analytics-service'); SELECT pgque.subscribe('user-events', 'notification-service'); SELECT pgque.subscribe('user-events', 'audit-log'); -- Producer sends once SELECT pgque.send('user-events', 'user_signup', '{"user_id": 1}'); -- Each consumer independently receives the same event SELECT * FROM pgque.receive('user-events', 'analytics-service'); SELECT * FROM pgque.receive('user-events', 'notification-service'); SELECT * FROM pgque.receive('user-events', 'audit-log'); ``` ### Retry / Dead Letter Pattern ```sql -- Using low-level API with retry logic DO $$ DECLARE v_batch_id bigint; v_ev record; BEGIN -- Get next batch SELECT pgque.next_batch('orders', 'order-processor') INTO v_batch_id; IF v_batch_id IS NULL THEN RAISE NOTICE 'No batch available'; RETURN; END IF; -- Process each event FOR v_ev IN SELECT * FROM pgque.get_batch_events(v_batch_id) LOOP BEGIN -- Attempt processing RAISE NOTICE 'Processing event % type %', v_ev.ev_id, v_ev.ev_type; -- On transient failure, retry after 30 seconds -- SELECT pgque.event_retry(v_batch_id, v_ev.ev_id, 30); EXCEPTION WHEN OTHERS THEN -- Schedule retry PERFORM pgque.event_retry(v_batch_id, v_ev.ev_id, 60); RAISE NOTICE 'Event % queued for retry', v_ev.ev_id; END; END LOOP; -- Finish batch (events not retried are acked) PERFORM pgque.finish_batch(v_batch_id); END; $$; ``` ## Monitoring & Introspection ```sql -- Queue info (depth, consumer count, last tick) SELECT * FROM pgque.get_queue_info(); SELECT * FROM pgque.get_queue_info('orders'); -- Consumer lag and position SELECT * FROM pgque.get_consumer_info(); SELECT * FROM pgque.get_consumer_info('orders'); SELECT * FROM pgque.get_consumer_info('orders', 'order-processor'); -- Batch details SELECT * FROM pgque.get_batch_info(:batch_id); -- Version SELECT pgque.version(); ``` ## Configuration & Tuning ### Tick Frequency ```sql -- Default: ticker called every 1 second via pg_cron -- To change tick interval, update the pg_cron job: SELECT cron.alter_job( job_id := (SELECT jobid FROM cron.job WHERE command LIKE '%pgque.ticker%'), schedule := '* * * * *' -- every minute (coarser) ); ``` ### Force Immediate Tick (Testing/Demos) ```sql -- Force a tick right now without waiting for pg_cron SELECT pgque.force_tick('orders'); -- or SELECT pgque.ticker(); ``` ### pg_cron Log Hygiene ```sql -- Disable run logging (ticker runs every second = 3600 rows/hour) ALTER SYSTEM SET cron.log_run = off; SELECT pg_reload_conf(); -- Or periodically purge: SELECT cron.schedule('pgque-cron-purge', '0 * * * *', $$DELETE FROM cron.job_run_details WHERE end_time < now() - interval '1 hour'$$ ); ``` ### pg_cron in Different Database If pg_cron is in `postgres` DB but PgQue is in `mydb`: ```sql -- Run from the pg_cron database (postgres) SELECT cron.schedule_in_database( 'pgque-ticker', '* * * * *', 'SELECT pgque.ticker()', 'mydb' ); SELECT cron.schedule_in_database( 'pgque-maint', '* * * * *', 'SELECT pgque.maint()', 'mydb' ); ``` ## Common Patterns ### Transactional Enqueue (Send with Business Logic) ```sql -- Event is only enqueued if the whole transaction commits BEGIN; INSERT INTO orders (id, amount) VALUES (42, 99.99); SELECT pgque.send('orders', 'new_order', '{"order_id": 42}'); COMMIT; ``` ### Queue Depth Check Before Scaling ```sql SELECT queue_name, ev_per_sec, consumer_count, pending_events FROM pgque.get_queue_info() WHERE pending_events > 1000; ``` ### List All Consumers with Lag ```sql SELECT queue_name, consumer_name, pending_events AS lag, last_seen FROM pgque.get_consumer_info() ORDER BY lag DESC; ``` ## Troubleshooting ### Consumers receive nothing **Cause**: Ticker is not running. ```sql -- Check if ticker has fired recently SELECT * FROM pgque.get_queue_info('orders'); -- Look at last_tick timestamp -- Manually fire a tick SELECT pgque.ticker(); -- Check pg_cron jobs exist SELECT * FROM cron.job WHERE command LIKE '%pgque%'; ``` ### Events not appearing after send ```sql