Thực hành transactional outbox với Postgres và Python: không mất sự kiện, và không trùng khi bên nhận dedup

Thử hình dung đoạn code này ở chỗ khách: lưu đơn hàng vào Postgres, commit, rồi gọi webhook sang ERP của khách. Nếu process chết giữa hai dòng đó, đơn hàng tồn tại mà ERP không bao giờ biết. Đảo thứ tự lại thì rủi ro ngược lại: ERP nhận được một đơn mà database của bạn đã rollback.

Đây là bài toán dual-write. Blog của Debezium nói thẳng: không thể có một transaction chung bao cả database của service lẫn Apache Kafka. Hệ thống của khách cũng vậy, thậm chí còn khó hơn vì bạn không kiểm soát nó.

Với FDE, đây là lỗi điển hình của một tích hợp “chạy được trong demo”. Bài này dựng từ đầu một transactional outbox bằng Postgres và Python để bạn mang thẳng ra chỗ khách.

Bạn sẽ dựng gì, và cần gì trước?

Theo mô tả trên microservices.io, outbox giải quyết đúng câu hỏi: làm sao vừa cập nhật database vừa gửi message một cách nguyên tử. Cách làm là service lưu message vào database trước, chung transaction với dữ liệu nghiệp vụ, rồi một relay chuyển nó đi sau.

Bạn cần Postgres chạy cục bộ, Python 3 và một thư viện kết nối Postgres. Code dưới đây viết theo phong cách psycopg 3 và một HTTP client như httpx. Đây là bản rút gọn để học: thiếu logging, retry backoff và cấu hình, bạn tự bổ sung trước khi đưa vào production.

Bước 1: bảng outbox nằm ngay trong database của bạn

Ý tưởng cốt lõi, như Debezium mô tả, là đặt một bảng “outbox” trong chính database của service. Không có broker trung gian, không có hàng đợi riêng.

CREATE TABLE orders (
  id          uuid PRIMARY KEY,
  customer_id text NOT NULL,
  status      text NOT NULL
);

CREATE TABLE outbox (
  id           uuid PRIMARY KEY,
  aggregate_id uuid NOT NULL,
  event_type   text NOT NULL,
  payload      jsonb NOT NULL,
  created_at   timestamptz NOT NULL DEFAULT now(),
  sent_at      timestamptz
);

CREATE INDEX outbox_unsent ON outbox (created_at)
  WHERE sent_at IS NULL;

Kiểm tra: \d outbox trong psql phải hiện outbox_unsent, một partial index chỉ phủ những hàng chưa gửi. Cột id chính là event ID, thứ mà về sau bên nhận sẽ dùng để dedup.

Bước 2: ghi đơn hàng và sự kiện trong một transaction

Mọi thứ còn lại của pattern đều dựa vào bước này. Hai câu INSERT phải nằm trong cùng một khối transaction.

import uuid, json

def create_order(conn, customer_id):
    order_id = uuid.uuid4()
    event_id = uuid.uuid4()
    with conn.transaction():
        conn.execute(
            "INSERT INTO orders (id, customer_id, status) VALUES (%s, %s, 'created')",
            (order_id, customer_id),
        )
        conn.execute(
            "INSERT INTO outbox (id, aggregate_id, event_type, payload) "
            "VALUES (%s, %s, 'order_created', %s)",
            (event_id, order_id, json.dumps({"order_id": str(order_id), "customer_id": customer_id})),
        )
    return order_id

Kiểm tra bằng cách cố tình raise exception giữa hai câu INSERT. Kết quả đúng là cả orders lẫn outbox đều trống. Có đơn mà không có sự kiện, hoặc ngược lại, nghĩa là bạn đang có hai transaction chứ không phải một.

Chú ý rằng hàm này không gọi mạng. Nó chỉ chạm vào database của bạn, nên nó nhanh và không bao giờ treo vì API của khách chậm.

Bước 3: relay bằng polling và SKIP LOCKED

Microservices.io nêu hai cách dựng relay: tailing transaction log (CDC) hoặc polling publisher. Với Python và Postgres thuần, polling là điểm khởi đầu hợp lý vì không cần thêm hạ tầng.

Khi muốn chạy nhiều worker, bạn cần SKIP LOCKED. Tài liệu Postgres giải thích rằng với tuỳ chọn này, những hàng không khoá được ngay sẽ bị bỏ qua, và nó dùng được để tránh tranh chấp khoá khi nhiều consumer cùng đọc một bảng kiểu hàng đợi.

def relay_once(conn, http, url, batch=50):
    with conn.transaction():
        rows = conn.execute(
            "SELECT id, event_type, payload FROM outbox "
            "WHERE sent_at IS NULL ORDER BY created_at "
            "LIMIT %s FOR UPDATE SKIP LOCKED",
            (batch,),
        ).fetchall()
        for event_id, event_type, payload in rows:
            resp = http.post(
                url,
                json={"id": str(event_id), "type": event_type, "data": payload},
                headers={"Idempotency-Key": str(event_id)},
                timeout=10,
            )
            resp.raise_for_status()
            conn.execute("UPDATE outbox SET sent_at = now() WHERE id = %s", (event_id,))

Chạy hàm này trong vòng lặp có sleep ngắn. Kiểm tra: mở hai terminal chạy hai worker, tạo 200 đơn, rồi đếm ở phía nhận. Mỗi worker phải nhận một phần khác nhau, không worker nào đứng chờ worker kia.

Đây là bản đơn giản hoá có chủ đích: cả batch nằm trong một transaction dài, giữ khoá hàng suốt các lần gọi HTTP. Tài liệu Postgres cũng cảnh báo SKIP LOCKED cho ra góc nhìn không nhất quán về dữ liệu, nên chỉ dùng cho bảng kiểu hàng đợi như outbox, đừng dùng cho truy vấn thông thường.

Vì sao “không mất” vẫn đồng nghĩa với “có thể trùng”?

Đi qua một trường hợp cụ thể với bản trên. Batch có 50 sự kiện, 30 sự kiện đầu POST thành công, sự kiện thứ 31 trả về lỗi 500. raise_for_status ném lỗi, transaction rollback, và cả 30 lệnh UPDATE trước đó cũng mất theo.

Lần poll sau, cả 30 sự kiện đã tới tay khách sẽ được gửi lại. Kịch bản process bị kill giữa POST và UPDATE cũng cho kết quả tương tự. Microservices.io nói rõ hệ quả: relay có thể publish một message nhiều lần, nên consumer phải idempotent, chẳng hạn bằng cách ghi lại ID các message đã xử lý.

Đó là lý do header Idempotency-Key mang đúng event ID. Bạn không xoá được bản gửi trùng, nhưng bạn có thể khiến nó vô hại.

Bản tốt hơn: mỗi sự kiện một transaction

Trùng là không tránh được, nhưng gửi lại 30 sự kiện vì một lỗi là thừa. Cách sửa đơn giản nhất là commit theo từng sự kiện: mỗi transaction chỉ khoá đúng một hàng, gửi, đánh dấu, rồi commit.

def relay_one(conn, http, url):
    with conn.transaction():
        row = conn.execute(
            "SELECT id, event_type, payload FROM outbox "
            "WHERE sent_at IS NULL ORDER BY created_at "
            "LIMIT 1 FOR UPDATE SKIP LOCKED"
        ).fetchone()
        if row is None:
            return False
        event_id, event_type, payload = row
        resp = http.post(
            url,
            json={"id": str(event_id), "type": event_type, "data": payload},
            headers={"Idempotency-Key": str(event_id)},
            timeout=10,
        )
        resp.raise_for_status()
        conn.execute("UPDATE outbox SET sent_at = now() WHERE id = %s", (event_id,))
    return True

Chạy lại kịch bản cũ: 30 sự kiện đầu đã commit sent_at, chỉ sự kiện thứ 31 quay lại hàng đợi. Kill giữa POST và UPDATE giờ chỉ gây trùng đúng một sự kiện. Lưu ý nếu cứ ORDER BY created_at thì sự kiện lỗi sẽ được thử lại ngay, nên bạn cần thêm backoff.

Bản này vẫn giữ khoá một hàng trong lúc chờ HTTP. Bước tiếp theo, nếu API của khách chậm, là “claim trước”: một transaction ngắn đánh dấu hàng đã được nhận kèm thời điểm, commit, gửi bên ngoài transaction, rồi mới cập nhật sent_at. Khi đó bạn phải tự thêm cột và quy tắc thu hồi những hàng bị claim quá lâu, nên chỉ làm khi thật sự cần.

Bước 4: phía nhận dedup theo event ID

Trong ví dụ của Debezium, việc đầu tiên handler làm là kiểm tra sự kiện có UUID đó đã được xử lý chưa. Nếu bạn viết luôn endpoint nhận, hoặc khách cho phép bạn đặt một lớp đệm trước hệ thống của họ, hãy làm như sau:

CREATE TABLE processed_events (
  event_id     uuid PRIMARY KEY,
  processed_at timestamptz NOT NULL DEFAULT now()
);

def handle_event(conn, event):
    with conn.transaction():
        cur = conn.execute(
            "INSERT INTO processed_events (event_id) VALUES (%s) ON CONFLICT DO NOTHING",
            (event["id"],),
        )
        if cur.rowcount == 0:
            return  # đã xử lý trước đó, bỏ qua
        apply_business_change(conn, event)

Chi tiết quan trọng là câu INSERT dedup và thay đổi nghiệp vụ nằm chung một transaction. Tách ra thì bạn lại tự dựng ra một bài toán dual-write thu nhỏ.

Kiểm tra lại kịch bản ở trên: kill relay ngay sau POST. Sau khi relay khởi động lại, bảng processed_events vẫn chỉ có một hàng cho event đó, và thay đổi nghiệp vụ chỉ áp dụng một lần.

Khi khách đã có API idempotency, hãy học theo Stripe

Nhiều khi bạn không chạm được vào code của khách. Khi đó hãy hỏi API của họ có hỗ trợ idempotency key hay không. Stripe là mẫu tham khảo tốt: client tự sinh key để server nhận ra các lần retry, Stripe gợi ý dùng UUID V4, và key dài tối đa 255 ký tự.

Stripe lưu response đầu tiên của mỗi key và trả lại đúng response đó cho các lần retry, kể cả lỗi 500. Key bị dọn sau 24 giờ, và nếu một key cũ quay lại với tham số khác, Stripe báo lỗi để chặn dùng nhầm.

Hai chi tiết đó cho bạn hai câu hỏi phải đặt ra với khách. Họ giữ key bao lâu? Nếu relay của bạn có thể dừng lâu hơn khoảng đó, sự kiện gửi lại sẽ bị coi là mới. Và đừng bao giờ sinh key mới cho mỗi lần retry: key phải gắn với sự kiện, không gắn với request.

Những lỗi hay gặp

Lỗi phổ biến nhất là gọi API ngay trong hàm create_order “cho nhanh”, rồi coi outbox như phương án dự phòng. Làm thế là quay lại dual-write. Lỗi thứ hai là dùng ID tự tăng của request làm idempotency key, nên mỗi lần retry lại ra một key mới.

Lỗi thứ ba là để bảng outbox phình mãi. Partial index giữ truy vấn polling nhanh, nhưng bạn vẫn cần một job định kỳ dọn các hàng đã gửi, chẳng hạn:

DELETE FROM outbox WHERE sent_at < now() - interval '7 days';

Con số 7 ngày ở đây chỉ là ví dụ. Quy tắc là thời gian giữ phải dài hơn khoảng giữ key của phía nhận, để khi cần điều tra một lần gửi trùng bạn vẫn còn dữ liệu đối chiếu.

Ở chỗ khách, outbox bắt đầu từ một lời than phiền

Thường thì đó là câu: “thỉnh thoảng có đơn không sang được hệ thống kế toán”. Việc đầu tiên nên làm là tìm mọi chỗ code ghi database rồi gọi ra ngoài ngay sau đó, và hỏi khách xem endpoint của họ xử lý request lặp lại ra sao.

Khi đọc JD cho vị trí FDE hay integration engineer, hãy để ý các cụm như “at-least-once”, “idempotency”, “webhook”, “CDC”. Trên CV, một dòng như “thay dual-write bằng transactional outbox, có kiểm thử kill-giữa-chừng chứng minh không mất và không xử lý trùng” thuyết phục hơn nhiều so với chữ “event-driven” chung chung.

Một repo nhỏ chứa đúng bốn bước trên, cả hai phiên bản relay và một script mô phỏng sự cố, là thứ đáng đem ra trong buổi phỏng vấn. Người phỏng vấn sẽ hỏi relay chết giữa chừng thì sao, và bạn sẽ có sẵn câu trả lời bằng code.