Transactional outbox với PostgreSQL và RabbitMQ: ghi bền, gửi lại, không lặp side effect

API lưu đơn thành công nhưng worker không nhận được event. Hoặc event đã tới RabbitMQ trong khi transaction tạo đơn lại rollback. Hai tình huống đối nghịch này thường xuất hiện khi code ghi database và publish message như hai thao tác rời nhau, rồi chỉ xử lý lỗi bằng try/catch.

Lab dưới đây ghi order và event vào PostgreSQL cùng transaction, dùng một relay chuyển event sang RabbitMQ, rồi deduplicate ở consumer. Điểm gây lỗi quan trọng là sau publisher confirm nhưng trước khi đánh dấu event đã gửi. Kết quả cần đạt là có thể giao lại message mà projection không bị cập nhật lặp.

1. Xác định ba nơi lưu trạng thái

Database nguồn giữ order và outbox. Broker giữ message chờ giao. Database consumer giữ dấu event đã xử lý và projection. Trong lab, hai phần database dùng chung PostgreSQL để giảm công cài đặt, nhưng các transaction producer và consumer vẫn tách biệt. Khi triển khai, consumer có thể sở hữu database riêng.

Mẫu transactional outbox của Chris Richardson giải quyết khoảng trống giữa commit dữ liệu và phát event bằng một bản ghi bền vững trong cùng transaction nghiệp vụ. Relay có thể phát trùng khi lỗi ở thời điểm không chắc chắn; vì thế thiết kế này không tuyên bố exactly-once delivery.

Không cần tách toàn bộ ứng dụng thành microservice mới áp dụng outbox. Một monolith có worker gửi thông báo hoặc cập nhật chỉ mục cũng có dual write. Ngược lại, nếu công việc chỉ là cập nhật một bảng trong cùng transaction và không có bên ngoài cần nhận event, thêm broker có thể làm vận hành phức tạp hơn nhu cầu.

2. Chuẩn bị lab và schema

Cần Node.js 24, Docker Compose và các cổng loopback 15432, 25672 đang trống. Cổng 25672 của host được ánh xạ vào cổng AMQP 5672 của container; đây không phải cổng quản trị web. Không kết nối ví dụ vào broker hay database đang dùng cho công việc thật.

Image được ghim bằng digest từ lần chạy lab để tránh tag thay đổi làm lệch kết quả. Chỉ chạy cấu hình này trong môi trường thử nghiệm riêng.

File: outbox.compose.yaml

name: thanhnh-architecture-20261020
services:
  postgres:
    image: postgres:17-alpine@sha256:b0f9560a2de083e2cc7382e75f808c7381a32852a7ec49117deedb300e552b24
    environment:
      POSTGRES_USER: lab
      POSTGRES_PASSWORD: lab-only-password
      POSTGRES_DB: arch_lab
    ports: ["127.0.0.1:15432:5432"]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U lab -d arch_lab"]
      interval: 2s
      timeout: 2s
      retries: 30
  rabbitmq:
    image: rabbitmq:4-management-alpine@sha256:d5c8b200a061242868833c8bbdd0703534a4a06ffe654a853e33acd46c7ce258
    environment:
      RABBITMQ_DEFAULT_USER: lab
      RABBITMQ_DEFAULT_PASS: lab-only-password
    ports: ["127.0.0.1:25672:5672"]
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
      interval: 3s
      timeout: 5s
      retries: 30

File: outbox-setup.sh

node --version  # dùng Node.js 24
npm init -y
npm install pg@8.23.1 amqplib@2.2.0
docker compose up -d --wait

Lưu outbox.sql. Partial index chỉ chứa event chưa publish để relay không phải quét toàn bộ lịch sử. Event ID được tạo một lần cùng order và giữ nguyên qua mọi lần gửi lại.

File: outbox.sql

CREATE TABLE IF NOT EXISTS orders (
  id uuid PRIMARY KEY,
  quantity integer NOT NULL CHECK (quantity > 0)
);
CREATE TABLE IF NOT EXISTS outbox (
  id uuid PRIMARY KEY,
  aggregate_id uuid NOT NULL REFERENCES orders(id),
  payload jsonb NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  published_at timestamptz
);
CREATE INDEX IF NOT EXISTS outbox_pending ON outbox(created_at, id)
  WHERE published_at IS NULL;
CREATE TABLE IF NOT EXISTS processed_events (id uuid PRIMARY KEY);
CREATE TABLE IF NOT EXISTS order_projection (
  order_id uuid PRIMARY KEY,
  quantity integer NOT NULL
);

File: outbox-load.sh

docker compose exec -T postgres psql -v ON_ERROR_STOP=1 -U lab -d arch_lab < outbox.sql

Tạo db.mjs cho kết nối lab. Giá trị mật khẩu trong mẫu là credential dùng một lần cho container cục bộ, không phải secret production. Database và AMQP production cần xác thực, TLS, giới hạn quyền và mạng riêng theo môi trường thực tế.

File: db.mjs

import pg from 'pg';
// Fixed local lab target; this file never reads production .env.
export const pool = new pg.Pool({
  connectionString: 'postgres://lab:lab-only-password@127.0.0.1:15432/arch_lab',
  max: 10, connectionTimeoutMillis: 2000, statement_timeout: 3000,
});

3. Producer, relay và hàm xử lý consumer

Lưu file outbox.mjs. Code giữ phần xử lý nghiệp vụ nhỏ để nhìn rõ ranh giới transaction. Hàm createOrderWithEvent chỉ ghi database; khi broker ngừng hoạt động, order vẫn có thể commit và event nằm chờ trong outbox.

File: outbox.mjs

import { randomUUID } from 'node:crypto';
import amqp from 'amqplib';
import { pool } from './db.mjs';
export async function createOrderWithEvent(quantity) {
  if (!Number.isInteger(quantity) || quantity < 1) throw new Error('invalid quantity');
  const c = await pool.connect();
  const orderId = randomUUID(), eventId = randomUUID();
  try {
    await c.query('BEGIN');
    await c.query('INSERT INTO orders VALUES ($1,$2)', [orderId, quantity]);
    await c.query('INSERT INTO outbox(id,aggregate_id,payload) VALUES ($1,$2,$3)',
      [eventId, orderId, { type: 'order.created.v1', orderId, quantity }]);
    await c.query('COMMIT'); return { orderId, eventId };
  } catch (error) { await c.query('ROLLBACK'); throw error; }
  finally { c.release(); }
}
export async function openBroker() {
  const connection = await amqp.connect('amqp://lab:lab-only-password@127.0.0.1:25672');
  connection.on('error', () => {}); // Supervisor must reconnect in a real worker.
  const channel = await connection.createConfirmChannel();
  channel.on('error', () => {});
  await channel.assertQueue('arch.orders', { durable: true, arguments: { 'x-queue-type': 'quorum' } });
  await channel.prefetch(8);
  return { connection, channel };
}
function confirmedSend(channel, row) {
  return new Promise((resolve, reject) => {
    let returned = false;
    const onReturn = (msg) => { if (msg.properties.messageId === row.id) returned = true; };
    channel.on('return', onReturn);
    const timer = setTimeout(() => {
      channel.removeListener('return', onReturn);
      void channel.close().catch(() => {});
      reject(new Error('publisher confirm timeout'));
    }, 2000);
    channel.sendToQueue('arch.orders', Buffer.from(JSON.stringify(row.payload)),
      { persistent: true, mandatory: true, messageId: row.id, contentType: 'application/json' },
      (error) => {
        clearTimeout(timer); channel.removeListener('return', onReturn);
        if (error || returned) reject(error || new Error('unroutable message'));
        else resolve();
      });
  });
}
export async function publishOne(channel, crashAfterConfirm = false) {
  const c = await pool.connect();
  try {
    await c.query('BEGIN');
    const { rows: [row] } = await c.query(`SELECT id,payload FROM outbox
      WHERE published_at IS NULL ORDER BY created_at,id
      FOR UPDATE SKIP LOCKED LIMIT 1`);
    if (!row) { await c.query('COMMIT'); return false; }
    await confirmedSend(channel, row);
    if (crashAfterConfirm) throw new Error('injected crash after confirm');
    await c.query('UPDATE outbox SET published_at=now() WHERE id=$1', [row.id]);
    await c.query('COMMIT'); return true;
  } catch (error) { await c.query('ROLLBACK'); throw error; }
  finally { c.release(); }
}
export async function applyEvent(message) {
  const event = JSON.parse(message.content.toString());
  const id = message.properties.messageId;
  if (!/^[0-9a-f-]{36}$/.test(id || '') || event.type !== 'order.created.v1' ||
      !/^[0-9a-f-]{36}$/.test(event.orderId || '') || !Number.isInteger(event.quantity) || event.quantity < 1) {
    throw new Error('invalid event');
  }
  const c = await pool.connect();
  try {
    await c.query('BEGIN');
    const inserted = await c.query('INSERT INTO processed_events VALUES ($1) ON CONFLICT DO NOTHING RETURNING id', [id]);
    if (inserted.rowCount) await c.query('INSERT INTO order_projection VALUES ($1,$2)', [event.orderId, event.quantity]);
    await c.query('COMMIT'); return Boolean(inserted.rowCount);
  } catch (error) { await c.query('ROLLBACK'); throw error; }
  finally { c.release(); }
}

Vì sao không đánh dấu published trước khi gửi?

Nếu đặt published_at rồi COMMIT trước khi broker nhận, process có thể chết ở khoảng trống đó và event bị bỏ sót. Mẫu làm ngược lại: chờ broker confirm rồi mới đánh dấu trong transaction relay. Lỗi ở giữa hai bước sẽ khiến gửi lại, một lựa chọn có thể xử lý bằng event ID thay vì mất dấu công việc cần làm.

Theo tài liệu RabbitMQ về confirms và acknowledgements, publisher confirm và consumer ACK là hai cơ chế độc lập. Confirm không chứng minh consumer đã cập nhật dữ liệu. Với message không route được, mandatory cho phép broker trả basic.return; code kiểm tra cả trường hợp này thay vì coi mọi confirm là thành công nghiệp vụ.

SKIP LOCKED giúp chia việc, không bảo đảm thứ tự

Relay khóa một dòng pending trong lúc publish và đợi confirm tối đa hai giây. Relay khác có thể bỏ qua dòng đang khóa để chọn dòng khác. PostgreSQL mô tả SKIP LOCKED phù hợp việc phân phối hàng đợi nhưng có thể tạo một góc nhìn không nhất quán cho truy vấn tổng quát.

Đổi lại sự đơn giản, mẫu giữ transaction DB trong một lượt gọi mạng. Khi tăng tải, cân nhắc claim có lease hoặc CDC; phải xử lý lease hết hạn và worker cũ quay lại. Nếu một aggregate có nhiều event cần đúng thứ tự, thêm sequence và chiến lược phân vùng theo aggregate. ORDER BY created_at không đủ bảo đảm thứ tự xử lý giữa nhiều relay và consumer.

4. Cố tình lỗi sau confirm và xem event bị giao lại

Trong thư mục lab mới, lưu outbox-demo.mjs rồi chạy. Hãy bắt đầu với queue và outbox sạch của riêng lab; demo không phải chương trình drain backlog có sẵn. Nó tạo một order, cố tình rollback relay sau confirm, gửi lại rồi xử lý hai bản cùng event ID.

File: outbox-demo.mjs

import { createOrderWithEvent, openBroker, publishOne, applyEvent } from './outbox.mjs';
import { pool } from './db.mjs';
const { connection, channel } = await openBroker();
try {
  const order = await createOrderWithEvent(2);
  try { await publishOne(channel, true); } catch (error) { console.log(error.message); }
  await publishOne(channel); // Resends the same event ID after the injected failure.
  const applied = [];
  for (let i = 0; i < 2; i++) {
    let message;
    for (let attempt = 0; attempt < 30; attempt++) {
      message = await channel.get('arch.orders', { noAck: false });
      if (message) break;
      await new Promise((resolve) => setTimeout(resolve, 100));
    }
    if (!message) throw new Error('message not received');
    applied.push(await applyEvent(message));
    channel.ack(message); // Only after the database transaction committed.
  }
  console.log({ order, applied });
} finally { await connection.close(); await pool.end(); }

File: outbox-run.sh

node outbox-demo.mjs

Kết quả mong đợi gồm thông báo injected crash và applied có một true, một false. True nghĩa event mới đã tạo projection, false nghĩa dấu processed_events đã có. Không dùng số lần nhận message làm số lần thực hiện side effect nghiệp vụ.

Kiểm thử riêng của bài còn giả lập consumer commit xong nhưng chưa ACK bằng cách đưa message trở lại queue. Các lần nhận sau không tạo thêm projection. Trường hợp publish mandatory không có queue cũng đã được kiểm tra: relay giữ event pending thay vì đánh dấu thành công.

5. ACK sau commit, và chỉ deduplicate phần có cùng transaction

INSERT processed_events và INSERT projection nằm trong cùng transaction. Nếu consumer chết trước commit, lần giao sau vẫn được phép xử lý. Nếu chết sau commit nhưng trước ACK, unique event ID làm lần sau bỏ qua side effect. Chuyển ACK lên trước COMMIT sẽ mở lại nguy cơ broker xóa message trong khi database chưa có kết quả.

Hàm applyEvent không tự ACK; adapter AMQP chỉ gọi ACK khi hàm resolve. Ví dụ dùng channel.get để dễ quan sát từng message. Một worker thường dùng channel.consume, prefetch hữu hạn và cơ chế shutdown đợi việc đang chạy; không sao chép demo một cách nguyên xi thành daemon vô hạn.

Dấu dedup chỉ bảo vệ các thao tác DB trong transaction đó. Nếu worker gửi email hay gọi dịch vụ ngoài, INSERT processed_events không tự làm lời gọi ngoài thành idempotent. Cần key ổn định ở dịch vụ nhận, một outbox tiếp theo hoặc quy trình đối soát phù hợp.

6. Retry, poison message và giới hạn của lab một node

Mẫu khai báo quorum queue durable và message persistent. RabbitMQ mô tả điều kiện an toàn của quorum queues theo số replica còn khả dụng. Một node trong Docker không phải hệ thống HA; kiểm thử này không xác nhận khả năng chịu mất node, mất đĩa hoặc lỗi nhiều vùng.

Không đặt nack requeue=true trong mọi catch rồi lặp ngay. Một JSON sai schema hoặc event type không hỗ trợ sẽ quay lại liên tục, tiêu thụ CPU mà không tiến triển. Tách lỗi dữ liệu sang nơi điều tra có giới hạn truy cập; lỗi tạm như DB unavailable cần backoff và giới hạn thử phù hợp. Mẫu chưa triển khai DLQ hoặc retry scheduler, nên phải bổ sung chính sách trước khi vận hành worker liên tục.

RabbitMQ Reliability Guide cũng lưu ý khả năng gửi trùng khi xác nhận bị mất trên đường truyền. Đó là lý do event ID phải đến từ outbox, không được sinh UUID mới cho mỗi lần publish. Khi reconnect, tạo lại channel, xác nhận topology và chỉ tiếp tục sau khi trạng thái kết nối đã rõ.

7. Chỉ số cần có và kế hoạch đưa lên production

1. Theo dõi tuổi event pending lâu nhất, không chỉ đếm số dòng outbox. Mười event mắc nhiều giờ khác với một nghìn event vừa được ghi.
2. Đo thời gian chờ confirm, số basic.return, số retry và lỗi transaction. Giới hạn số publisher đang giữ connection DB để broker chậm không làm cạn pool.
3. Theo dõi queue ready, unacknowledged, consumer capacity và tỷ lệ dedup. Dedup tăng sau một đợt reconnect có thể là hành vi phục hồi dự kiến.
4. Chọn retention cho outbox đã gửi và processed_events theo cửa sổ replay. Xóa dấu dedup trước khi chạy replay sẽ có thể thực hiện lại side effect.
5. Canary một loại event, kiểm tra reconciliation giữa order và projection. Rollback relay không được xóa event pending hoặc purge queue công việc thật.

Code lab đã kiểm tra hành vi ghi bền, gửi lại và chống trùng trong database; nó chưa kiểm chứng ordering toàn hệ thống, HA hay hiệu năng tải lớn. Khi mở rộng, giữ nguyên các điểm commit và dấu event ổn định trước, rồi mới tối ưu batch và số consumer.