PostgreSQL LISTEN/NOTIFY trên production: Nhanh nhưng không phải message queue

PostgreSQL vừa commit một thay đổi và ứng dụng cần đánh thức worker ngay, không muốn polling mỗi giây. LISTEN/NOTIFY giải quyết rất gọn — cho tới khi listener mất kết nối, giữ transaction quá lâu hoặc đội ngũ lầm tưởng notification là một hàng đợi bền vững.

Bài này xây dựng cách dùng LISTEN/NOTIFY an toàn trên production: hiểu thời điểm giao notification, tránh race khi khởi động listener, dùng payload nhỏ, kết hợp bảng outbox để khôi phục sự kiện bị lỡ, thiết kế reconnect, giám sát notification queue và biết khi nào cần message broker thật. Hành vi được đối chiếu với tài liệu PostgreSQL hiện hành về LISTEN, NOTIFY, pg_notification_queue_usage() và libpq asynchronous notifications.

1. LISTEN/NOTIFY thực sự cung cấp điều gì?

Một session chạy:

LISTEN order_events;

Session khác trong cùng database phát:

SELECT pg_notify('order_events', '{"order_id":8421}');

Khi transaction phát notification commit, mọi session đang nghe channel trong database đó nhận channel, PID backend phát và payload. Đây là cơ chế IPC nhẹ giữa các client cùng dùng PostgreSQL.

Nó phù hợp cho:

• đánh thức worker để đọc dữ liệu mới từ bảng;
• invalidate cache theo key hoặc version;
• báo cho WebSocket/SSE gateway rằng có thay đổi;
• refresh materialized view hoặc dashboard nội bộ;
• giảm độ trễ so với polling thuần khi dữ liệu gốc đã ở PostgreSQL.

Nó không cung cấp durable subscription, consumer offset, dead-letter queue, backpressure theo consumer hay replay log. Listener không kết nối tại thời điểm notification được phát sẽ không nhận lại notification đó sau này.

Hãy coi NOTIFY là chuông cửa, không phải kiện hàng. Chuông báo “hãy kiểm tra bảng”; dữ liệu thật và trạng thái xử lý phải nằm trong storage bền vững.

2. Notification tuân theo transaction

Nếu NOTIFY chạy trong transaction, PostgreSQL chỉ giao sự kiện sau khi transaction commit. Rollback đồng nghĩa notification biến mất. Điều này giúp listener không nhìn thấy tín hiệu cho dữ liệu chưa tồn tại.

BEGIN;

INSERT INTO orders (customer_id, status)
VALUES (42, 'pending');

SELECT pg_notify('order_events', 'changed');

COMMIT; -- notification chỉ được giao sau điểm này

Chiều ngược lại cũng quan trọng: nếu session listener đang ở trong transaction khi notification đến, server chỉ chuyển notification tới client sau khi transaction đó kết thúc. Vì vậy connection chuyên LISTEN không nên chạy query dài hoặc giữ transaction idle.

Các transaction dùng LISTEN hoặc NOTIFY cũng có ràng buộc với two-phase commit: chúng không thể được prepare. Nếu hệ thống dùng PREPARE TRANSACTION, phải tách luồng notification khỏi transaction 2PC.

3. Delivery semantics: fan-out, không phải work queue

Mọi session đang nghe cùng channel đều nhận notification. Nếu chạy 10 worker cùng LISTEN jobs, cả 10 đều thức dậy; PostgreSQL không chọn một worker duy nhất như competing-consumer queue.

Trong cùng một transaction, nhiều NOTIFY có cùng channel và payload giống nhau có thể được gộp thành một event. Payload khác nhau vẫn là event riêng. Notification từ các transaction khác nhau không bị gộp và được giao theo thứ tự commit.

Hệ quả:

• không đếm nghiệp vụ bằng số notification;
• không giả định một NOTIFY tương ứng đúng một row;
• consumer phải chịu được wakeup trùng;
• worker group cần cơ chế claim công việc ở bảng, ví dụ FOR UPDATE SKIP LOCKED;
• nhiều subscriber độc lập cần trạng thái xử lý riêng nếu mỗi bên phải nhận đủ sự kiện.

4. Payload nhỏ và không chứa bí mật

Trong cấu hình mặc định, payload phải ngắn hơn khoảng 8000 byte. Payload là text, không phải nơi phù hợp cho JSON lớn, file hoặc dữ liệu nhạy cảm. Tài liệu PostgreSQL khuyến nghị lưu dữ liệu lớn trong bảng rồi gửi key của record.

Thiết kế payload tốt:

{"event_id":91284}

Hoặc chỉ cần một tín hiệu:

SELECT pg_notify('outbox_wakeup', '');

Notification nhìn thấy bởi mọi database user có thể LISTEN channel; channel không phải ranh giới bí mật. Không đưa access token, email, dữ liệu khách hàng hoặc nội dung nghiệp vụ nhạy cảm vào payload. Listener phải đọc row bằng query chịu kiểm soát quyền.

Khi channel/payload động, dùng pg_notify(text, text) với parameter binding. Không nối chuỗi SQL từ input không tin cậy.

5. Race khi listener khởi động

LISTEN chỉ có hiệu lực khi transaction chứa nó commit. Giữa lúc ứng dụng đọc snapshot ban đầu và bắt đầu lắng nghe có thể có transaction khác commit. Quy trình đúng trong tài liệu PostgreSQL là:

1. Kết nối bằng một session dành riêng cho listener.
2. Chạy LISTEN channel và commit.
3. Trong transaction mới, đọc trạng thái/outbox hiện tại để catch up.
4. Sau đó xử lý notification cho thay đổi tiếp theo.

Một vài notification đầu có thể trỏ tới dữ liệu đã thấy trong lần catch-up; consumer phải idempotent. Trình tự ngược — đọc trước rồi mới LISTEN — tạo khoảng trống thật sự và có thể mất wakeup.

-- Bước 1, session listener
LISTEN outbox_wakeup;
COMMIT;

-- Bước 2, sau khi LISTEN đã có hiệu lực
SELECT *
FROM outbox_events
WHERE processed_at IS NULL
ORDER BY id
LIMIT 100;

6. Dedicated connection là yêu cầu kiến trúc

LISTEN là state của database session và tự mất khi session kết thúc. Connection phải sống lâu, đọc socket đều và tự reconnect. Không mượn connection bất kỳ từ pool rồi trả lại vì lần xử lý sau có thể nhận session khác.

Với pooler ở chế độ transaction pooling, session client không gắn cố định với một backend nên LISTEN state không phải contract ổn định. Dùng dedicated/session-pinned connection hoặc một listener service riêng; query nghiệp vụ thông thường vẫn đi qua pool.

Listener loop cần:

• keepalive và timeout hợp lý;
• đọc input từ socket thay vì busy polling;
• heartbeat/health check;
• reconnect với backoff và jitter;
• chạy lại LISTEN rồi catch up từ bảng sau mỗi reconnect;
• shutdown có kiểm soát và xử lý tín hiệu hệ điều hành.

Với libpq, PQnotifies() chỉ lấy notification đã được hấp thụ vào client; ứng dụng cần PQconsumeInput() khi socket readable rồi gọi PQnotifies() cho tới khi hết.

7. Mẫu bền vững: transactional outbox + NOTIFY

Tạo outbox trong cùng PostgreSQL transaction với dữ liệu nghiệp vụ:

CREATE TABLE outbox_events (
    id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    topic        text NOT NULL,
    aggregate_id text NOT NULL,
    payload      jsonb NOT NULL,
    created_at   timestamptz NOT NULL DEFAULT clock_timestamp(),
    processed_at timestamptz
);

CREATE INDEX outbox_events_pending_idx
ON outbox_events (id)
WHERE processed_at IS NULL;

Ghi nghiệp vụ và event atomically:

BEGIN;

UPDATE orders
SET status = 'paid'
WHERE id = 8421;

INSERT INTO outbox_events (topic, aggregate_id, payload)
VALUES (
    'order.paid',
    '8421',
    '{"order_id":8421}'::jsonb
);

SELECT pg_notify('outbox_wakeup', '');

COMMIT;

Nếu transaction rollback, cả business change, outbox row và notification đều không tồn tại. Nếu listener offline lúc commit, nó bỏ lỡ tiếng chuông nhưng vẫn tìm thấy outbox row khi reconnect hoặc trong lượt polling dự phòng.

8. Trigger giúp không quên NOTIFY

Có thể đặt notification trong trigger của outbox:

CREATE OR REPLACE FUNCTION notify_outbox_insert()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
    PERFORM pg_notify('outbox_wakeup', '');
    RETURN NEW;
END;
$$;

CREATE TRIGGER outbox_insert_notify
AFTER INSERT ON outbox_events
FOR EACH ROW
EXECUTE FUNCTION notify_outbox_insert();

Vì payload giống nhau trong một transaction có thể được gộp, batch insert chỉ cần đánh thức listener; listener đọc toàn bộ row pending. Đây là hành vi mong muốn, không phải mất event, vì bảng mới là nguồn chân lý.

Trigger phải nhẹ: không gọi HTTP, không xử lý JSON phức tạp và không làm công việc bên ngoài. Nó chạy trên đường commit của nghiệp vụ.

9. Claim công việc với nhiều worker

Khi nhiều worker cùng được đánh thức, claim row bằng khóa ở database:

BEGIN;

SELECT id, topic, aggregate_id, payload
FROM outbox_events
WHERE processed_at IS NULL
ORDER BY id
FOR UPDATE SKIP LOCKED
LIMIT 50;

-- xử lý/đánh dấu theo strategy của hệ thống

UPDATE outbox_events
SET processed_at = clock_timestamp()
WHERE id = ANY(:claimed_ids);

COMMIT;

Không giữ transaction mở trong lúc gọi API chậm nếu có thể tránh. Một mẫu thường dùng là claim với trạng thái/lease ngắn, commit, thực hiện side effect, rồi đánh dấu hoàn tất. Nhưng crash giữa side effect và đánh dấu có thể tạo retry; consumer downstream vẫn phải idempotent.

“Exactly once” xuyên database và hệ thống ngoài thường không đạt được chỉ bằng LISTEN/NOTIFY. Thiết kế theo at-least-once, idempotency key, unique constraint và reconciliation.

10. Cẩn thận với cursor dựa trên bigserial

Sequence cấp ID trước commit. Hai transaction có thể lấy ID 100 và 101 nhưng transaction 101 commit trước. Nếu consumer lưu last_id=101 ngay rồi chỉ đọc id > 101, row 100 commit muộn có thể bị bỏ qua.

Giải pháp phụ thuộc mô hình:

• dùng bảng pending/processed và query theo trạng thái thay vì high-water mark;
• serialize bước publish bằng một dispatcher;
• duy trì per-consumer delivery records với unique constraint;
• nếu dùng cursor, chứng minh rõ rằng thứ tự cấp key khớp thứ tự publish/commit hoặc có cơ chế xử lý gap.

Thứ tự delivery của NOTIFY không biến sequence ID thành commit-ordered log. Đây là lý do một bảng outbox cần semantics xử lý rõ ràng, không chỉ thêm cột id.

11. Queue nội bộ có thể đầy

PostgreSQL giữ notification chưa được mọi listener xử lý trong một queue. Một session đã LISTEN rồi giữ transaction rất lâu có thể ngăn cleanup. Khi queue đầy, transaction gọi NOTIFY có thể thất bại tại commit — tức lỗi notification có thể tác động trực tiếp luồng ghi nghiệp vụ.

Giám sát:

SELECT pg_notification_queue_usage();

SELECT pid,
       usename,
       application_name,
       state,
       xact_start,
       query_start,
       wait_event_type,
       wait_event
FROM pg_stat_activity
WHERE xact_start IS NOT NULL
ORDER BY xact_start;

pg_notification_queue_usage() trả tỷ lệ từ 0 tới 1. Alert từ ngưỡng thấp phù hợp với baseline, đồng thời truy tìm listener transaction dài. Không chữa triệu chứng bằng cách chỉ tăng max_notify_queue_pages; sửa lifecycle của session và tốc độ xử lý trước.

12. Reconnect, failover và deploy

Database restart, failover, network flap hoặc deploy đều làm dedicated connection mất. Listener phải xem reconnect là luồng bình thường:

1. Đánh dấu health là degraded.
2. Reconnect với exponential backoff và jitter.
3. Chạy lại LISTEN, xác nhận command commit.
4. Catch up bảng durable từ trạng thái xử lý.
5. Quay lại vòng chờ notification.

Không có “resume token” cho notification thuần. Khả năng phục hồi đến từ bảng/outbox. Trong rolling deploy, hai phiên listener có thể cùng hoạt động và cùng thức dậy; claim/idempotency phải xử lý được.

13. Polling dự phòng vẫn cần thiết

Dù LISTEN/NOTIFY giảm latency, chạy một timer định kỳ để quét pending rows giúp phục hồi khi:

• notification bị lỡ trong khoảng disconnect;
• listener code bị treo nhưng connection chưa đóng;
• trigger/đường phát notification bị lỗi cấu hình;
• một deploy mới có bug subscribe sai channel;
• cần reconciliation dữ liệu cũ.

Interval có thể dài hơn polling-only vì notification xử lý đường nhanh. Ví dụ wakeup ngay khi có NOTIFY và sweep 30–60 giây làm safety net, tùy SLO.

14. Bảo mật và phân quyền

• Dùng role listener chỉ có SELECT trên bảng/view cần thiết.
• Không nhúng bí mật vào channel hoặc payload.
• Validate payload như input không tin cậy; notification không phải authentication.
• Giới hạn kết nối listener và đặt application_name rõ để quan sát.
• Không cho payload quyết định SQL identifier/path tùy ý.
• Nếu tenant isolation quan trọng, enforce tại query/RLS chứ không dựa vào tên channel.

15. Khi nào nên dùng message broker?

Khi bảng outbox trở thành broker tự chế với partition, retention, nhiều offset, retry scheduler và DLQ phức tạp, đó là tín hiệu đánh giá Kafka, RabbitMQ, NATS JetStream, cloud queue hoặc nền tảng tương đương theo yêu cầu cụ thể.

16. Checklist production

• Dùng NOTIFY như wakeup; dữ liệu thật nằm trong bảng durable.
• LISTEN và commit trước, sau đó catch up state trong transaction mới.
• Dùng dedicated connection và tự reconnect/re-LISTEN.
• Giữ listener ngoài transaction dài.
• Payload chỉ chứa ID/hint nhỏ, không chứa bí mật.
• Worker claim row atomically và xử lý idempotent.
• Không dùng số notification làm số event.
• Giám sát pg_notification_queue_usage() và long transaction.
• Có polling/reconciliation dự phòng.
• Load test burst, disconnect, failover và rolling deploy.

Kết luận

LISTEN/NOTIFY là một cơ chế đánh thức có độ trễ thấp, transaction-aware và rất tiện khi PostgreSQL đã là trung tâm dữ liệu. Nó trở nên nguy hiểm khi bị dùng như một durable queue. Mẫu production đáng tin cậy là: ghi outbox cùng transaction nghiệp vụ, phát một tín hiệu nhỏ sau commit, để listener đọc/claim dữ liệu bền vững, và luôn catch up sau reconnect. Khi giới hạn này được tôn trọng, LISTEN/NOTIFY giúp giảm polling mà không biến notification tạm thời thành nguồn chân lý.