postgres for everything

• https://topicpartition.io/blog/postgres-pubsub-queue-benchmarks
• https://news.ycombinator.com/item?id=45747018
• https://github.com/Olshansk/postgres_for_everything
• https://adriano.fyi/posts/2023-09-24-choose-postgres-queue-technology
• https://lobste.rs/s/rk3eft/choose_postgres_queue_technology#c_iix4ph

Postgres competes with purpose-built solutions like:

• Workflow https://github.com/earendil-works/absurd/tree/main
• Elasticsearch (functionality supported by Postgres’ tsvector/tsquery)
• MongoDB (jsonb)
• Redis (CREATE UNLOGGED TABLE)
• AI Vector Databases (pgvector, pgai)
• Snowflake (pg_mooncake, pg_duckdb)
• Kafka - kolejki - konstrukcja SELECT ... FOR UPDATE SKIP LOCKED
  • pub-sub - pg-pubsub
  • queeu - pgmq PG as a Pub/Sub https://topicpartition.io/blog/postgres-pubsub-queue-benchmarks#pg-as-a-pubsub
• Table

CREATE TABLE log_counter (
  id           INT PRIMARY KEY, -- topicpartition table name id
  next_offset  BIGINT NOT NULL  -- next offset to assign
);
 
for i in NUM_PARTITIONS:
  CREATE TABLE topicpartition%d (
    id          BIGSERIAL PRIMARY KEY,
    -- strictly increasing offset (indexed by UNIQUE)
    c_offset    BIGINT UNIQUE NOT NULL,
    payload     BYTEA NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
  );
  INSERT INTO log_counter(id, next_offset) VALUES (%d, 1);
 
CREATE TABLE consumer_offsets (
  group_id     TEXT NOT NULL,     -- consumer group identifier
  -- topic-partition id (matches log_counter.id / topicpartitionN)
  topic_id     INT  NOT NULL,
  -- next offset the consumer group should claim
  next_offset  BIGINT NOT NULL DEFAULT 1,
  PRIMARY KEY (group_id, topic_id)
);

PG as a Queue

In Postgres, a queue can be implemented with SELECT FOR UPDATE SKIP LOCKED. This command selects an unlocked row and locks it. It also skips reading already-locked rows. That’s how mutual exclusion is achieved - a worker can’t get other workers’ jobs.

Postgres has a very popular pgmq library that offers a slick queue API.

Example queue

• Tables

CREATE TABLE queue (
  id BIGSERIAL PRIMARY KEY,
  payload BYTEA NOT NULL,
	created_at TIMESTAMP NOT NULL DEFAULT NOW()
)
 
CREATE TABLE queue_archive (
  id BIGINT,
  payload BYTEA NOT NULL,
  created_at TIMESTAMP NOT NULL, -- ts the event was originally created at
  processed_at TIMESTAMP NOT NULL DEFAULT NOW() -- ts the event was processed at
)

• Writes

INSERT INTO queue (payload) VALUES ($1)

• Readers

BEGIN;
 
SELECT id, payload, created_at
  FROM queue
  ORDER BY id
  FOR UPDATE SKIP LOCKED
  LIMIT 1;
 
-- Your business code "processes" the message. In the benchmark, it's a no-op.
 
DELETE FROM queue WHERE id = $1;
 
INSERT INTO queue_archive (id, payload, created_at, processed_at)
  VALUES ($1,$2,$3,NOW());
 
COMMIT;

Each reader only works with one message at a time per transaction

Cecha	SELECT ... FOR UPDATE SKIP LOCKED	LISTEN/NOTIFY
Niezawodność/Gwarancja	Wysoka. Zadanie jest w tabeli, zawsze można je pobrać.	Niska. Powiadomienia mogą zostać utracone (efemeryczne).
Pobieranie zadania	Atomowe. Wybiera i blokuje w jednym kroku.	Dwa kroki. Odbierz powiadomienie, a potem pobierz zadanie (ryzyko wyścigu).
Wydajność w Konkurencji	Doskonała. Workers pomijają zajęte wiersze bez czekania.	Słaba. Problem "Thundering Herd" (wszystkie workery próbują naraz).
Obciążenie Bazy Danych	Polling. Obciążenie cyklicznymi zapytaniami (ale małe, jeśli nie ma zadań).	Utrzymywanie Połączeń. Ograniczenie liczby połączeń.

W praktyce: NOTIFY jest świetny do powiadamiania użytkowników w czasie rzeczywistym lub odświeżania małych cache'ów. Do budowania kolejek zadań, które muszą być niezawodne, SELECT ... FOR UPDATE SKIP LOCKED jest standardowym i rekomendowanym wzorcem.