Các giải pháp cho dual write

Gần hai năm trước, trong bài Dual write, mình có giới thiệu một vài pattern để xử lý dual write và nói rằng có thể sẽ viết kỹ hơn về chúng sau. Hơi lâu một chút, nhưng cuối cùng cũng tới lúc rồi. Nếu chưa đọc bài đó, bạn có thể xem lại để hiểu vấn đề và lý do chỉ đổi thứ tự ghi hay retry trong bộ nhớ vẫn chưa giải quyết được.

Ý tưởng chung của các pattern này khá đơn giản: khi không thể gộp hai lần ghi vào hai hệ thống độc lập thành một thao tác nguyên tử, chúng ta dùng bản ghi đã được lưu bền vững của lần ghi này làm cơ sở cho lần ghi còn lại. Trong ví dụ e-commerce ở bài trước, search service sẽ index thông tin đăng bán sau khi thông tin đó đã được lưu vào database, hoặc sau khi một event yêu cầu tạo thông tin đăng bán đã được phát đi và lưu bền vững. Sự khác biệt này quan trọng với pattern listen to yourself, mình sẽ giải thích ở phần sau. Dù theo cách nào, search service cũng cần được thông báo một cách đáng tin cậy để người mua có thể tìm thấy sản phẩm.

Trong bài này, chúng ta sẽ tìm hiểu transactional outbox, listen to yourself và change data capture, cùng những đánh đổi của từng cách. Bắt đầu với transactional outbox nhé.

Transactional outbox pattern⌗

Thay vì ghi trực tiếp vào nhiều hệ thống, chúng ta lưu dữ liệu nghiệp vụ và một message mô tả thay đổi trong cùng một transaction của database. Với catalog service trong ví dụ này, chúng ta tạo bản ghi sản phẩm và bản ghi outbox cùng nhau. Hoặc cả hai cùng commit, hoặc không bản ghi nào được lưu. Sau đó, một process riêng sẽ gửi message tới search service.

Tên gọi này bắt nguồn từ khay thư đi: bạn để thư cần gửi vào đó, rồi có người tới lấy và chuyển đi. Tương tự, outbox giữ các message đang chờ gửi, còn message relay đảm nhận việc chuyển chúng tới nơi nhận.

Chúng ta cần hai thành phần:

• Outbox lưu các message đang chờ gửi. Với database quan hệ, outbox thường là một bảng riêng. Với database dạng document, outbox có thể nằm trong cùng document với dữ liệu nghiệp vụ. Điều quan trọng là việc lưu dữ liệu nghiệp vụ và message phải là một thao tác nguyên tử.
• Message relay lấy các message đang chờ và gửi tới nơi nhận. Relay có thể chạy dưới dạng một tác vụ nền hoặc một service riêng. Trong ví dụ này, relay định kỳ kiểm tra outbox và yêu cầu search service index từng sản phẩm mới.

Để dễ hình dung hơn, chúng ta thử xem một ít SQL. Với tác vụ index đơn giản, chúng ta có thể lưu một cờ ngay trên bản ghi sản phẩm. Việc tạo sản phẩm và đánh dấu sản phẩm cần được index diễn ra trong một câu lệnh:

-- Tạo sản phẩm và đánh dấu cần index.
INSERT INTO
    "product" ("name", "price", "need_search_indexing")
VALUES
    ('Battery', 3.25, TRUE);

-- Relay đọc các sản phẩm đang chờ được index.
SELECT * FROM "product" WHERE "need_search_indexing" IS TRUE;

Cờ này cho relay biết sản phẩm cần được index để tìm kiếm, nhưng không lưu lại từng thay đổi riêng lẻ dưới dạng event. Nếu muốn mỗi event có một bản ghi riêng, chúng ta có thể dùng bảng outbox. Trong ví dụ này, ứng dụng cung cấp ID sản phẩm là p0001:

-- Hai lệnh INSERT phải nằm trong cùng một transaction.
BEGIN;

INSERT INTO
    "product" ("id", "name", "price")
VALUES
    ('p0001', 'Battery', 3.25);

INSERT INTO
    "product_event_outbox" ("product_id", "type", "payload", "processed_at")
VALUES
    ('p0001', 'CREATED', '{"name": "Battery", "price": 3.25}', NULL);

COMMIT;

-- NULL nghĩa là event chưa được đánh dấu đã xử lý.
SELECT * FROM "product_event_outbox" WHERE "processed_at" IS NULL;

Điều mình thích ở transactional outbox là ý tưởng cơ bản khá đơn giản và linh hoạt. Chúng ta có thể triển khai bằng database hiện có và một tác vụ nền, không cần thêm message broker hay dùng framework. Nếu muốn, chúng ta vẫn có thể cho relay gửi message tới message broker, hoặc dùng framework để quản lý outbox.

Tất nhiên, chúng ta sẽ có thêm code cần bảo trì, cùng các cột hoặc bảng bổ sung. Nếu dùng bảng outbox riêng, chúng ta có thể sẽ cần một tác vụ dọn dẹp định kỳ, chẳng hạn cron job, cùng chính sách quy định thời gian giữ lại các event đã xử lý để bảng không tăng mãi. Các event vẫn đang chờ gửi thì cần được giữ lại trong outbox.

Vì message được gửi bất đồng bộ, dữ liệu trong database và search index chỉ nhất quán theo mô hình eventual consistency: sản phẩm có thể đã được lưu nhưng chưa tìm kiếm được ngay. Nếu relay kiểm tra outbox theo định kỳ, khoảng thời gian giữa các lần kiểm tra cũng làm tăng độ trễ này.

Listen to yourself pattern⌗

Listen to yourself là một pattern xử lý theo event. Vẫn với ví dụ sản phẩm ở trên, khi người dùng tạo sản phẩm, catalog service sẽ gửi event event.product.submitted tới message broker trước, thay vì lưu ngay sản phẩm vào database. Sau khi broker xác nhận event đã được lưu bền vững, service báo cho người dùng rằng yêu cầu đã được tiếp nhận. Phản hồi này chưa có nghĩa là sản phẩm đã có trong database hay đã tìm kiếm được.

Sau đó, catalog service nhận lại chính event mình đã gửi và lưu sản phẩm vào database. Tên gọi của pattern cũng bắt nguồn từ đây: service lắng nghe các event do chính nó phát ra. Search service có subscription riêng để cập nhật search index, còn notification service có thể nhận cùng event đó để gửi thông báo. Mỗi service cần một subscription hoặc consumer group độc lập để đều nhận được event, thay vì tranh nhau xử lý một bản sao duy nhất.

Nếu sau này có tính năng mới cần cùng event, chúng ta chỉ cần thêm consumer mà không phải sửa các consumer hiện có. Nhờ vậy, việc thêm tính năng và scale từng consumer độc lập khá linh hoạt. Việc ghi database cũng có thể được retry riêng qua catalog consumer, thay vì phụ thuộc vào phần xử lý request để ghi vào cả hai hệ thống.

Đánh đổi là chúng ta cần thêm message broker nếu chưa có, và cách tổ chức ứng dụng theo event có thể còn lạ lẫm. Ngay cả việc cập nhật database của chính catalog service cũng diễn ra bất đồng bộ, nên nếu đọc sản phẩm ngay sau khi yêu cầu được tiếp nhận, có thể vẫn chưa thấy dữ liệu. Phản hồi ban đầu có thể nhanh vì chỉ cần chờ broker, nhưng vẫn cần thời gian để sản phẩm xuất hiện trong database và search index.

Change data capture⌗

Đúng như tên gọi, change data capture (CDC) ghi nhận các thay đổi trong database và gửi chúng tới hệ thống khác. Ở đây, chúng ta sẽ tập trung vào CDC dựa trên log, tận dụng các log mà database duy trì để phục hồi dữ liệu hoặc replication. Ví dụ có write-ahead log (WAL) của PostgreSQL, binlog của MySQL và oplog của MongoDB. Trong ví dụ sản phẩm, catalog service lưu sản phẩm vào database, rồi pipeline CDC đọc thay đổi đã commit và gửi tới các hệ thống phía sau.

Định dạng log có ảnh hưởng tới cách CDC hoạt động. Chẳng hạn, MySQL hỗ trợ replication theo câu lệnh (statement-based) và theo dòng dữ liệu (row-based). Cách thứ nhất ghi lại câu lệnh SQL để replica chạy lại. Log có thể gọn, nhưng một số thao tác cho kết quả không cố định, như tạo UUID, không an toàn khi chạy lại theo cách này. Cách thứ hai ghi lại giá trị của các dòng bị thay đổi thay vì chạy lại câu lệnh ban đầu. Log có thể lớn hơn, nhất là khi một câu lệnh thay đổi nhiều dòng, nhưng connector CDC sẽ có được dữ liệu thực tế đã thay đổi. Connector MySQL của Debezium yêu cầu log theo dạng row-based.

Một pipeline CDC thường có ba phần:

• Source connector đọc các thay đổi đã commit từ log của database và phát ra change event.
• Thành phần trung chuyển, thường là message broker, giữ các event cho tới khi consumer phía sau xử lý.
• Sink connector hoặc consumer của ứng dụng đọc event và áp dụng thay đổi vào nơi nhận, chẳng hạn search index hoặc một database khác.

Kafka và Debezium là một cách kết hợp phổ biến, nhưng CDC không bắt buộc phải dùng bộ công cụ này. Các công cụ và giải pháp tích hợp với database khác cũng có thể cung cấp pipeline tương tự. Đánh đổi là phần hạ tầng và công sức vận hành: connector, thành phần trung chuyển event và consumer đều cần được quản lý. Mình thường sẽ nghiêng về CDC hơn khi hạ tầng đó đã có sẵn và team đã quen với cách vận hành.

Một đánh đổi khác là thay đổi trong bảng dữ liệu nghiệp vụ chỉ mô tả dữ liệu được lưu, chứ không thể hiện trực tiếp ý định nghiệp vụ phía sau. Change event có thể cho biết một dòng vừa được thêm, cập nhật hay xóa. So sánh giá trị trước và sau khi cập nhật cũng có thể cho biết giá hoặc danh mục đã thay đổi, nhưng chưa chắc giải thích được lý do. Ví dụ dưới đây là một update event theo cấu trúc của Debezium đã được rút gọn, bỏ phần schema và một số metadata:

{
  "payload": {
    "before": {
      "id": "p0001",
      "name": "Battery",
      "price": 3.25,
      "category_id": "c001"
    },
    "after": {
      "id": "p0001",
      "name": "Battery",
      "price": 2.95,
      "category_id": "c001"
    },
    "source": {
      "connector": "mysql",
      "db": "catalog",
      "table": "product"
    },
    "op": "u"
  }
}

Ở đây, op: "u" cho biết đây là thao tác cập nhật, còn so sánh before với after cho thấy giá giảm từ 3.25 xuống 2.95. Nhưng event không cho biết đó là khuyến mãi hay sửa lại giá nhập sai. Nếu consumer cần event thể hiện rõ nghiệp vụ, chúng ta có thể kết hợp CDC với outbox: ứng dụng ghi event vào outbox, còn CDC đảm nhận việc gửi đi.

Điều mình thích ở CDC là nó tách logic gửi các thay đổi dữ liệu ra khỏi code nghiệp vụ. Catalog service xử lý nghiệp vụ và ghi vào database, còn pipeline CDC đọc log của database rồi gửi các thay đổi đi. Khi cần tích hợp thêm hệ thống phía sau, chúng ta có thể thêm consumer mà không phải thêm logic gửi dữ liệu cho từng hệ thống vào catalog service.

Kết luận⌗

Khác biệt chính nằm ở việc chúng ta lưu bền vững dữ liệu nào trước, và ứng dụng có thể đảm bảo điều gì sau khi lần ghi đó thành công:

Các cách này cho chúng ta một nơi lưu bền vững để tiếp tục công việc còn dở, nhưng việc gửi message vẫn cần retry và phục hồi khi có lỗi. Chẳng hạn, relay có thể crash sau khi search service đã index sản phẩm nhưng trước khi đánh dấu event đã gửi. Khi chạy lại, relay sẽ gửi event đó lần nữa. Consumer cần xử lý an toàn khi nhận event trùng, ví dụ bằng cách lưu ID của các event đã xử lý, hoặc thiết kế thao tác ghi sao cho chạy lại nhiều lần vẫn cho cùng kết quả. Đó là phần idempotency mà chúng ta vẫn phải triển khai.

Chọn cách nào còn tùy service có cần lưu dữ liệu của mình đồng bộ hay không, consumer cần dữ liệu gì, và team sẵn sàng vận hành những hạ tầng nào.

Đọc thêm⌗

• Transactional outbox
• Understanding the Dual-Write Problem and Its Solutions
• Distributed Data for Microservices: Event Sourcing vs. Change Data Capture
• Debezium connector for MySQL