PostgreSQL:�simple tasks queue
fprochazka.cz
Hi, I’m Filip Procházka
Agenda
Use-cases for task queues?
Use-cases for task queues?
Problem definition
Task queues with message brokers
Message brokers
Total Cost of Ownership (TCO)
Is the technology adding enough benefits to be worth the additional cost?
Message brokers
Problem: distributed commit (or two-phase-commit)
Simple & reasonably safe in small scale
⚠️ not bulletproof � data inconsistency may occur, but the risk is usually tolerable for simpler apps
Outbox Pattern
⚠️ not bulletproof � exchanges the "catastrophic data inconsistency" problem for the "at-least-once delivery" problem
Having only one database eliminates the need for the distributed commit�(reducing complexity and TCO)
Problem: locking a resource
Filesystem lock
Redlock
TTL lock with a database
CREATE TABLE shedlock(
name TEXT NOT NULL, lock_until TIMESTAMP NOT NULL,
locked_at TIMESTAMP NOT NULL, locked_by TEXT NOT NULL,
PRIMARY KEY (name) );
INSERT INTO shedlock (name, lock_until, locked_at, locked_by) VALUES (...)
ON CONFLICT (name) DO UPDATE
SET lock_until = :lockUntil, locked_at = :now, locked_by = :lockedBy
WHERE shedlock.lock_until <= :now;
-- once you’re done, you should explicitly unlock
Advisory lock with a database
SET LOCK_TIMEOUT = '15s'; -- default timeout is 0 => indefinitely
SELECT pg_advisory_xact_lock(10);
-- obtains a shared transaction-level advisory lock, waiting if necessary
SELECT pg_try_advisory_xact_lock(10);
-- either obtain the lock immediately and return true
-- or return false without waiting if the lock cannot be acquired immediately
Transactional row locking with a database
SET LOCK_TIMEOUT = '15s'; -- default timeout is 0 => indefinitely
SELECT * FROM table FOR UPDATE;
-- wait for a set time and error if row cannot be locked
SELECT * FROM table FOR UPDATE NOWAIT;
-- if row cannot be locked, do not wait and throw error
SELECT * FROM table FOR UPDATE SKIP LOCKED;
-- any selected rows that cannot be immediately locked are skipped
-- behaves intuitively and cooperates with LIMIT
If you don’t have complex requirements or huge volumes of data, the problem can be solved using just PostgreSQL
(you are not Google, so simpler approaches are probably good enough)
Using PostgreSQL as task-queue
Use-case: notifications (email, sms, push, …)
Use-case: notifications
-- schema
CREATE TYPE notification_status AS
ENUM ('pending', 'sent');
CREATE TABLE notification_email (
id UUID NOT NULL,
created_at TIMESTAMP NOT NULL,
status notification_status NOT NULL,
sent_at TIMESTAMP,
PRIMARY KEY (id)
);
Use-case: notifications
-- schedule
INSERT INTO notification_email (id, created_at, status)
VALUES (uuid_generate_v7(), NOW(), 'pending');
-- consume & process
SELECT *
FROM notification_email
WHERE status = 'pending'
LIMIT 100
FOR NO KEY UPDATE SKIP LOCKED;
-- you should explicitly open & close a transaction
-- the lock is released on COMMIT
FOR UPDATE vs FOR NO KEY UPDATE
CREATE TABLE notification_email_something (
id UUID NOT NULL,
notification_id UUID NOT NULL,
PRIMARY KEY (id),
FOREIGN KEY (notification_id) REFERENCES notification_email (id)
);
SELECT * FROM notification_email FOR UPDATE;
-- When postgre validates FK of the notification_email_something�-- it needs to acquire FOR KEY SHARE on the notification_email row,
-- which is blocked by the strong lock acquired by the FOR UPDATE
SELECT * FROM notification_email FOR NO KEY UPDATE;
-- this acquires weaker lock, which does not block FOR KEY SHARE
Easy: naive FIFO
SELECT *
FROM notification_email
WHERE status = 'pending'
ORDER BY created_at ASC
LIMIT 100
FOR NO KEY UPDATE SKIP LOCKED;
Easy: delayed sending
ALTER TABLE notification_email
ADD COLUMN send_after TIMESTAMP;
SELECT *
FROM notification_email
WHERE status = 'pending' AND send_after >= NOW()
ORDER BY created_at ASC
LIMIT 100
FOR NO KEY UPDATE SKIP LOCKED;
Easy: priorities
ALTER TABLE notification_email
ADD COLUMN priority INT DEFAULT 10; -- 0 low, 10 normal, 100 high
SELECT *
FROM notification_email
WHERE status = 'pending' AND send_after >= NOW()
ORDER BY priority DESC, created_at ASC
LIMIT 100
FOR NO KEY UPDATE SKIP LOCKED;
Easy: max attempts
ALTER TABLE notification_email
ADD COLUMN attempts INT DEFAULT 0;
SELECT *
FROM notification_email
WHERE status = 'pending' AND send_after >= NOW() AND attempts < 50
ORDER BY priority DESC, created_at ASC
LIMIT 100
FOR NO KEY UPDATE SKIP LOCKED;
Use-case: notifications
Optimizing task queue tables
Monitoring locks
CREATE EXTENSION pgrowlocks;
-- available also by default on AWS RDS
SELECT *
FROM notification_email AS a,
pgrowlocks('notification_email') AS p
WHERE p.locked_row = a.ctid;
-- simple way to inspect table row locks
SELECT * FROM pg_locks;
-- all locks in the system, but non-trivial to join onto table
Risk: deadlocks
Risk: DDL
Messaging in PostgreSQL - LISTEN/NOTIFY
LISTEN my_channel;
-- start listening within this connection
-- pick up messages using pg_get_notify()
NOTIFY my_channel;
-- pushes to active listeners
-- when nobody is listening, its voided
-- there are no retries or delays
Messaging in PostgreSQL - pgmq extension
SELECT pgmq.create('my_queue');
-- creates the queue
SELECT * FROM pgmq.send(queue_name => 'my_queue', msg => '{"foo": "bar1"}');
-- returns ids
SELECT * FROM pgmq.read(queue_name => 'my_queue', vt => 30, qty => 1);
-- read a msg from a queue and "lock" it for 30 seconds
SELECT * FROM pgmq.pop('my_queue');
-- read and immediately delete a msg
Use-case: exports
Use-case: dynamic crons
Use-case: dynamic crons
CREATE TABLE task (id UUID NOT NULL, name TEXT NOT NULL, PRIMARY KEY (id));
CREATE TABLE task_run_scheduled (
id UUID NOT NULL,
task_id UUID NOT NULL,
run_after TIMESTAMP NOT NULL,
status TEXT NOT NULL, -- enum
PRIMARY KEY (id),
FOREIGN KEY (task_id) REFERENCES task (id)
);
-- insert into task_run with next time
-- query every few seconds for tasks to run
-- before commit, insert row with next run
Questions?
fprochazka.cz