Change stream — CDC trên oplog, resume token, outbox và những event không bao giờ tới

Một relay Laravel đọc change stream của outbox, đẩy 3 000 đơn sang Redis Stream, bị kill -9 sáu lần: 0 message mất; lưu token mỗi 100 event thì 394 bản trùng, thêm khoá chống trùng thì 0. Cùng tải, dual write (ghi DB rồi publish) mất 7 message, không exception nào. 180 write w:1 bị rollback sau khi cắt mạng primary: 0 event — change stream chỉ đọc tới commit point; cũng vì thế khi 2 secondary bị fsyncLock, write vẫn ack mà event đứng im 21,7 s. fullDocument: 'updateLookup' với consumer chậm 10 ms/event: 32–69 % event mang document sai bản. Consumer theo dõi một collection ít ghi thì mất lịch sử dù chưa từng tắt (ChangeStreamHistoryLost). Driver Node 7.7 mất 30 s mới nghe tiếp sau failover dù primary mới có sau 10 s; worker PHP viết ngây thơ mất 95–136 event mỗi lần failover. Oplog, commit point, rollback: bài 7; outbox ghi cùng transaction: bài 8; time series không watch được: bài 2; shard: bài 9; Redis Stream: series Redis.

Thuật ngữ dùng trong bài
change stream (cursor trả về từng thay đổi đã commit, theo thứ tự) · CDC — change data capture (bắt thay đổi từ log của DB thay vì quét bảng) · oplog (log thao tác của replica set, collection capped — bài 7) · commit point (vị trí oplog mà đa số node đã ghi, không bao giờ bị rollback) · resume token (dấu trang: vị trí của một event, nằm ở field _id) · high-water-mark token (token chỉ một mốc thời gian, không gắn với event nào; server gửi kèm mọi batch, kể cả batch rỗng, dưới tên postBatchResumeToken) · cửa sổ oplog (khoảng thời gian oplog còn giữ) · invalidate (event báo stream phải đóng vì collection bị drop/rename) · pre-image / post-image (bản trước / sau khi đổi) · outbox ("việc cần gửi" ghi cùng transaction) · relay (process đọc outbox, đẩy sang broker) · dual write (ghi DB rồi gửi broker, không có transaction chung) · at-least-once (không mất, có thể trùng) · idempotent (làm 2 lần = 1 lần) · replication slot (PostgreSQL: vị trí của consumer, DB giữ WAL tới khi nó đọc).

Kiểm chứng: Context7 (/mongodb/mongo, /websites/mongodb_manual, /mongodb/node-mongodb-native, /mongodb/laravel-mongodb); source r9.0.2 (docs/change_streams.md, document_source_change_stream*, change_stream_rewrite_helpers.cpp, oplog_cap_maintainer_thread.cpp, noop_writer.cpp); source driver Node 7.7.0 và mongodb/mongodb 2.5.0; release notes 9.0, SERVER-135099, SERVER-134970. Output: labs/mongodb/out/10-*.txt, môi trường ở Phụ lục. Máy lab dùng chung: số ms và event/s chỉ để so tỉ lệ.

Trả lời 30 giây

Change stream là một aggregate $changeStream. Server tự mở rộng nó thành pipeline đọc oplog: lọc theo namespace và $match của bạn, đổi từng entry thành event, gắn resume token (mã hoá clusterTime + vị trí). Nó chỉ trả thay đổi đã majority-commit, nên không bao giờ phát event cho write sau đó bị rollback — đổi lại, khi đa số không ghi được thì stream đứng im. Token cho phép nghe tiếp sau khi consumer chết, nhưng chỉ khi vị trí đó còn trong oplog; oplog là capped và quay vòng, offline lâu hơn cửa sổ thì mất lịch sử, phải đồng bộ lại từ đầu. Giao tới consumer là at-least-once (lưu token sau khi xử lý → chết giữa chừng thì xử lý lại), nên sink phải idempotent. updateLookup đọc bản hiện tại, không phải bản lúc đổi; cần đúng bản thì bật pre/post image. Mỗi cursor tự quét oplog: 50 consumer là 50 lần quét. Mẫu chuẩn để phát event ra ngoài: outbox ghi cùng transaction + relay change stream + consumer idempotent.

1. Bài toán gốc và bối cảnh ra đời

Trước 3.6 (cuối 2017), muốn biết dữ liệu đổi gì thì phải tail local.oplog.rs bằng tailable cursor (cursor không đóng khi hết dữ liệu): Meteor (mỗi app server tail cả oplog rồi tự lọc), mongo-connector đẩy sang Elasticsearch. Percona (2018) liệt kê cái giá: oplog là log của cả server nên consumer phải "read and throw away" mọi thứ không cần; định dạng là nội bộ, tên field khó hiểu và có thể đổi (5.0 đổi update sang diff $v:2 — bài 7); đọc cả entry chưa đủ đa số nên có thể xử lý một write về sau bị rollback; trên sharded cluster phải tail từng shard rồi tự trộn, còn chunk migration hiện ra như insert/delete giả; không có cách "nghe tiếp từ chỗ cũ" chuẩn.

3.6 đưa ra change stream (MongoDB version history). Nó không phải API mới mà là một stage aggregate, nên dùng lại được auth, $match, $project, và chạy qua mongos. Ba quyết định thiết kế đi cùng: chỉ trả thay đổi đã majority-commit; mỗi event mang resume token; mongos trộn các shard theo clusterTime để ra một thứ tự toàn cục. Về sau có thêm: pre/post image (6.0), $changeStreamSplitLargeEvent (6.0.9 / 7.0), bộ metric metrics.changeStreams.* và vị trí cursor trong $currentOp (9.0). Reader v2 ("precise shard targeting" — chỉ mở cursor trên shard thật sự giữ dữ liệu) có trong code 9.0 nhưng bị tắt ở 9.0.2 (SERVER-135099). Chính MongoDB cũng chạy trên change stream: resharding, mongosync, Atlas Search (docs/change_streams.md); Meteor 3.5 đổi mặc định từ oplog tailing sang change stream (Meteor docs).

• Tối ưu cho: đẩy thay đổi đã commit, đúng thứ tự commit, sang hệ khác gần thời gian thực (vô hiệu cache, đồng bộ search index, kho phân tích), phát domain event qua outbox, CDC sang Kafka (MongoDB Kafka connector, Debezium) — và nghe tiếp được sau khi chết.
• Không tối ưu cho: lưu lịch sử lâu dài để phát lại tuỳ ý (chỉ dài bằng cửa sổ oplog); hàng đợi việc (không có consumer group, ack từng message, chia việc song song theo key); fan-out lớn (mỗi cursor tự quét oplog); biết "ý định" của lệnh (event dựng từ entry oplog, $inc 3 thành "giá trị cuối"); time series, view, standalone.

2. Mô hình tư duy cốt lõi

Change stream = một con trỏ đọc oplog, chạy sau commit point, mỗi event mang vị trí của chính nó. Hầu hết hành vi — cả tốt lẫn xấu — suy ra từ câu này:

3. Internals và quyết định thiết kế

3.1 Server tự dựng pipeline từ một stage

Trên replica set, $changeStream được mở rộng thành các stage nội bộ (docs/change_streams.md):

flowchart LR
  A["OplogMatch
quét local.oplog.rs
lọc ns + $match đã viết lại"] --> B["UnwindTransaction
applyOps → N event"]
  B --> C["Transform
entry → event
gắn resume token"]
  C --> D["CheckInvalidate
drop/rename → invalidate"]
  D --> E["CheckResumability
token còn trong oplog?
không → HistoryLost"]
  E --> F["AddPreImage / AddPostImage
updateLookup hoặc
pre-image + delta"]
  F --> G["$match / $project
của người dùng"]
  G --> H["SplitLargeEvent
(nếu có)"]

$match của bạn được viết lại thành điều kiện trên field oplog (change_stream_rewrite_helpers.cpp: operationType → op, documentKey → o/o2, fullDocument → o cho insert/replace, ns, updateDescription…) và đẩy xuống lúc quét, nên bớt việc dựng event. Nhưng con trỏ vẫn phải đi qua mọi entry: lab 6.2 đo docsExamined mỗi cursor bằng tổng entry oplog của cả collection không ai watch, dù $match chỉ giữ 1 %.

Event được dựng từ entry oplog, nên hình dạng của nó theo oplog chứ không theo câu lệnh. Lab 10-events (rút gọn):

insert        ct=1792201960.2   {"documentKey":{"_id":1},"fullDocument":{…}}
update        ct=1792201960.3   {"updatedFields":{"price":120,"meta.level":2},"removedFields":["name"],"truncatedArrays":[]}
update        ct=1792201961.5   {"updatedFields":{"meta.0":"y"},…,"disambiguatedPaths":{"meta.0":["meta","0"]}}   ← field tên "0", không phải index
insert ×3     ct=1792201961.7   (insertMany = 1 entry applyOps → 3 event CÙNG clusterTime, token dài 166/168/168)
update ×4     ct=…961.8 … .11   (updateMany = mỗi document 1 entry, mỗi event 1 clusterTime)
insert+update ct=1792201961.12  (transaction: cùng clusterTime, có lsid + txnNumber)
$set price:120 khi price đã là 120 → không có entry oplog → KHÔNG có event
createIndexes / modify / rename / drop → chỉ thấy khi showExpandedEvents: true

Cùng là "mảng ngắn lại" nhưng diff trong oplog khác nhau, nên event khác nhau — consumer áp diff sang hệ khác (Elasticsearch, bảng SQL) phải xử lý đủ các dạng:

3.2 Resume token: vị trí, không phải id

_id._data là chuỗi hex của KeyString (resume_token.h): byte 0x82 + 8 byte clusterTime (t, i) + version (hiện là 2) + loại token + vị trí trong transaction + cờ invalidate + UUID collection + documentKey. Lab giải mã 2 nhóm 8 ký tự hex đầu: khớp clusterTime ở cả 24 event, và chuỗi token tăng dần khi so như chuỗi (KeyString giữ thứ tự) — tiện để đo độ trễ consumer, nhưng docs yêu cầu coi token là opaque (không dựa vào cấu trúc).

• Event token và high-water-mark token. Mỗi batch (kể cả rỗng) có postBatchResumeToken; driver trả nó qua getResumeToken() / resumeToken. Token này ngắn (lab: 32 ký tự hex) và tiến theo oplog dù collection của bạn không có ghi — phải lưu nó, không chỉ _id của event cuối (mục 5.5).
• Ba cách mở lại: resumeAfter (sau token), startAfter (như trên nhưng dùng được với token của invalidate), startAtOperationTime (một timestamp). Không truyền gì thì bắt đầu ở lastApplied + 1 tick (getStartTimeForNewStream): worker khởi động lại không token là mất khúc giữa (mục 5.7).
• Token không mang được sang pipeline khác. Đổi $match rồi resume bằng token cũ: event của token bị filter mới loại thì server không tìm thấy → ChangeStreamHistoryLost hoặc ChangeStreamFatalError (docs/change_streams.md).

3.3 Chỉ trả thay đổi đã majority-commit

"Change streams only deliver events after they have been committed to a majority of nodes and durably persisted, ensuring they will not be rolled back" (docs/change_streams.md). Với pre-image: chờ cả entry update lẫn bản pre-image cùng đạt majority. Đổi lại sự sạch sẽ đó: độ trễ event ≥ độ trễ commit point (bài 7 mục 3.3), và khi commit point đứng yên (PSA mất secondary, secondary bị khoá, lag lớn) thì stream đứng yên theo — mục 5.6.

3.4 fullDocument: đọc lại, hoặc dựng lại

• updateLookup = thêm một lần đọc theo _id (shard key trong documentKey để định tuyến), read concern afterClusterTime = clusterTime của event: ra bản majority mới nhất tại lúc lookup, không phải lúc đổi; không còn document thì null (change_stream_update_lookup_stage.cpp). 9.0.2 tắt nhánh tối ưu featureFlagChangeStreamOptimizedUpdateLookup (SERVER-134970).
• Pre/post image (6.0+): bật changeStreamPreAndPostImages theo từng collection; mỗi update/replace/delete ghi bản trước vào config.system.preimages cùng lúc với entry oplog; post-image = pre-image + delta của event. Job dọn chạy mỗi expiredChangeStreamPreImageRemovalJobSleepSecs (10 s), xoá bản quá expireAfterSeconds (cluster parameter changeStreamOptions, mặc định off = giữ tới khi entry oplog tương ứng bị cắt). required mà thiếu image → lỗi; whenAvailable → null.

3.5 Sharded cluster: mongos trộn theo clusterTime

Qua mongos, mỗi shard chạy phần pipeline ở trên, mongos chạy $mergeCursors (AsyncResultsMerger) và trả event theo thứ tự clusterTime. Muốn trả event có clusterTime T, mongos phải chắc mọi cursor đã qua T. Shard bận tự báo qua postBatchResumeToken; shard rảnh chỉ tiến khi oplog của nó có entry mới — periodic noop mỗi periodicNoopIntervalSecs (10 s, chỉ đặt được lúc khởi động: lab nhận IllegalOperation: not allowed to change [periodicNoopIntervalSecs] at runtime). 9.0.2 dùng reader v1: mở cursor tới mọi shard + config server (config server để biết shard mới), dù collection chỉ nằm trên một shard — lab đọc featureFlagChangeStreamReaderV2: {value: false} trên cả mongos lẫn shard.

flowchart LR
  S1["shard1 (bận)
event T=105, 106"] --> M["mongos
AsyncResultsMerger
trả event T khi MỌI cursor đã qua T"]
  S2["shard2 (rảnh)
chỉ tiến nhờ noop
mỗi 10 s"] --> M
  CFG["config server
(theo dõi shard mới)"] --> M
  M --> APP["consumer
1 cursor, 1 thứ tự"]

3.6 Event lớn: 16 MB tính cả image

Event là một document BSON nên chịu giới hạn 16 MB (nội bộ thêm 16 KB: maximum 16793600). Lab 10-large: document 16 777 183 byte insert được, event của nó 16 777 533 byte vẫn qua; document 9 MB update 1 field (entry oplog chỉ 321 byte) — xin post-image: event 9,00 MB; xin cả pre lẫn post: BSONObjectTooLarge (10334): … Size 18874905 exceeds maximum 16793600. $changeStreamSplitLargeEvent (phải là stage cuối, đặt chỗ khác → Location7182802) cắt thành mảnh splitEvent {fragment: 1, of: 2}, mỗi mảnh 9 MB và một token riêng; consumer tự ghép, và resume giữa chừng thì phải bỏ mảnh lẻ.

4. Đảm bảo và giới hạn

Theo khung PACELC: change stream chọn C ở cả hai nhánh. Khi partition (P) nó dừng phát thay vì phát event chưa chắc chắn (bỏ A); lúc bình thường (E) nó chấp nhận trễ thêm bằng độ trễ commit point (bỏ L) để không bao giờ phát event "ảo". Ngược với oplog tailing kiểu cũ: nhanh hơn một nhịp nhưng có thể phát write về sau bị rollback.

5. Failure modes

5.1 Consumer chết giữa chừng: không mất, nhưng trùng — nhiều hay ít là do bạn chọn

Lab 10-outbox: producer Laravel ghi 3 000 đơn, mỗi đơn một transaction (đơn + outbox). Relay (worker PHP chạy dài) watch cs_outbox, đẩy sang Redis Stream rồi lưu token vào cs_relay_state. Trong lúc chạy: kill -9 relay 6 lần (khởi động lại ngay), docker pause Redis 3 s.

Broker treo 3 s không gây mất: relay thử lại (broker lỗi lần 0: socket error on read socket) và không lưu token khi chưa gửi được. Trùng chỉ có thể là các event từ token đã lưu tới lúc chết, nên lưu mỗi N event thì mỗi lần chết trùng tối đa N − 1. Lưu sau mỗi event thì trùng gần 0 nhưng chậm 13 lần — mỗi event thêm một write w:majority. Bộ nhớ worker phẳng suốt 12 000 event (+75 KB với sink Redis, +1 018 KB một lần với transaction, peak 21,5–22,5 MB): vòng lặp next() không rò, khác với khi worker dựng model Eloquent.

Bẫy, theo mức hay dính:

1. Lưu token trước khi xử lý → chết giữa chừng là mất event (at-most-once). Luôn: xử lý xong → lưu token.
2. Tin "exactly-once" của docs ("every event … is emitted exactly once") — đúng cho server, không cho đầu-cuối. Exactly-once thật chỉ khi sink là chính MongoDB và kết quả + token commit cùng nhau (dòng cuối bảng); sink ngoài thì idempotent theo id event (id outbox), không theo token.
3. Chống trùng có hạn. IDMP chỉ nhớ IDMP-DURATION (mặc định 100 s, Redis Stream mục 3.6): relay chết lâu hơn thì trùng lọt; khoá bền là unique index ở consumer cuối.

5.2 Relay bị kill giữa transaction: chết lặp lại đúng 60 s

Dòng cuối bảng 5.1 đúng số, nhưng log cho thấy cái giá. Relay bị kill -9 khi transaction ($inc + lưu token) đang mở; process mới khởi động lại thì:

02:24:59 relay-mongotxn bắt đầu từ token đã lưu (n=91)
02:24:59 relay-mongotxn CHẾT sau 0.0 s: BulkWriteException 112: Caused by :: Write conflict during plan execution and yielding is disabled. :: Please retry your operation or multi-document transaction.
   [supervisor: relay đã thoát → chạy lại lúc 04:25:18] … (22 lần chết liên tiếp)
02:25:59 relay-mongotxn xử lý xong event đầu tiên sau 0.0 s kể từ lúc khởi động
đơn 3000 · outbox 3000 · bộ đếm cs_proj 3000 (lệch +0)

Server vẫn giữ transaction của process đã chết, và nó đang giữ write trên document trạng thái + bộ đếm. Transaction mới đụng đúng hai document đó → WriteConflict ngay (bài 8 mục 3.4); transaction() của laravel-mongodb mặc định 1 lượt → ném lỗi → worker chết, supervisor chạy lại, lại đụng — tới khi server tự huỷ transaction mồ côi: 02:24:59 → 02:25:59 = transactionLifetimeLimitSeconds 60. Chữa: thử lại có backoff đủ dài, hoặc bỏ transaction — ghi kết quả idempotent theo id event rồi mới lưu token.

5.3 Dual write: mất im lặng

Cùng tải, producer kiểu dual write (insert đơn w:majority rồi XADD), bị kill -9 8 lần + broker treo 3 s:

đơn 3000 · outbox 0 · message 2993 (id khác nhau 2993) · TRÙNG 0 · MẤT 7 · message không có đơn 0

7 đơn nằm trong DB mà không bao giờ có message; không exception, không log — chỉ đối soát mới thấy. Đảo thứ tự (publish trước, insert sau) chỉ đổi loại lỗi: message cho đơn không tồn tại. Phân tích đủ các biến thể trên PostgreSQL + Laravel queue: gửi mail at-least-once.

5.4 updateLookup trả document sai bản

Lab 10-lookup: event là $inc v, so fullDocument.v với updatedFields.v của chính event đó.

Tỉ lệ sai = độ nóng của document × độ trễ của consumer: theo kịp thì gần 0 nên test không bắt được; chậm đi một chút (deploy, GC, broker treo) là nhảy vọt — đúng lúc không ai nhìn. Debezium ghi cùng cảnh báo: "if multiple updates are closely following one after the other, all update events will contain the same after value" (Debezium MongoDB connector).

Giá của pre-image: 343–355 byte mỗi update (document ~231 byte), throughput w:1 369–384 so với 352–387 update/s khi tắt — không khác đo được trên máy chung. Hết hạn (expireAfterSeconds: 5, chờ 25 s, job dọn xoá 15 779 bản):

'required': NoMatchingDocument (47): … Change stream was configured to require a pre-image for all update, delete and replace events, but the pre-image was not found for event: {operationType: "update", …
'whenAvailable': 50 event · fullDocumentBeforeChange null 50 · fullDocument null 50

Bẫy: (1) dùng updateLookup để dựng lịch sử / audit — sai bản, nhất là khi consumer chậm; (2) whenAvailable trả null im lặng khi image hết hạn — consumer phải coi null là "không biết", không phải "đã xoá"; (3) expireAfterSeconds ngắn hơn thời gian consumer có thể offline → resume được nhưng image đã mất (required thì stream chết với lỗi 47); (4) bật pre-image không trả lại image cho thay đổi trước lúc bật.

5.5 ChangeStreamHistoryLost: token rơi khỏi oplog — kể cả khi consumer chưa từng tắt

Lab 10-history: node riêng --oplogSize 5, ghi 30 MB vào collection khác. (A) consumer lưu token rồi tắt:

resumeAfter token cũ: ChangeStreamHistoryLost (286): Executor error during aggregate command on namespace: lab.cs_h :: caused by :: Resume of change stream was not possible,
  as the resume point may no longer be in the oplog (resumeTimestamp: Timestamp(1792243069, 1))
startAfter / startAtOperationTime cũ: cùng lỗi 286

(B) consumer vẫn chạy trên cs_quiet (một collection ít ghi), trong khi cs_busy ghi 30 MB. Kiểu 1 chỉ lưu _id của event cuối; kiểu 2 lưu getResumeToken() sau mỗi batch, kể cả batch rỗng:

token event cuối = 1792243129.1 · token high-water-mark = 1792243134.20 (dài 32 ký tự hex)
kiểu 1 — token của event cuối: ChangeStreamHistoryLost (286): … (resumeTimestamp: Timestamp(1792243129, 1))
kiểu 2 — high-water-mark token: resume OK

Đây là bẫy nổ trễ kinh điển: collection ít ghi (cấu hình, bảng giá) trong một cluster bận. Consumer khoẻ suốt, nhưng lần restart đầu tiên sau khi oplog quay hết một vòng là mất vị trí. MongoDB Kafka connector có hẳn trang xử lý lỗi này cho "infrequently updated namespace" và khuyên bật heartbeat.interval.ms (Invalid Resume Token).

(C) replSetResizeOplog {minRetentionHours: 1}: oplog được phép vượt kích thước — 34,4 rồi 64,2 MB cho cấu hình 5 MB — và token cũ resume được; đặt về 0 thì 32 s sau oplog về 4,7 MB. Cái giá là đĩa: giữ N giờ ở giờ ghi cao điểm.

Oplog không bị cắt ngay khi đầy. Thread cắt oplog chờ với hạn oplogTruncationCheckPeriodSeconds = 300 (cả 8.0.32 lẫn 9.0.2) và chỉ cắt tới mốc checkpoint ổn định. Ghi 30 MB một mạch rồi dừng: oplog 29,7 MB / 5 MB giữ nguyên 290 s (8.0.32), 296 s (9.0.2), log cắt đúng nhịp 5 phút (02:48:07 Looking for excess documents → 02:53:07 Oplog truncation finished). Hạ chu kỳ xuống 1 s (phải đặt lúc khởi động) vẫn mất 52–58 s ≈ nhịp checkpoint 60 s. Vậy cửa sổ thực tế co giãn: đo bằng rs.printReplicationInfo() ở giờ cao điểm, đừng suy từ MB. (Ghi liên tục không ngừng: chưa đo.)

5.6 Đa số không ghi được: write vẫn ack, event đứng im; write bị rollback: không bao giờ có event

Lab 10-majority: writer w:1 mỗi giây; từ giây 10,8 tới 33,7 hai secondary bị fsyncLock (kịch bản thật: backup kiểu snapshot, đĩa secondary đầy, PSA mất secondary):

primary vẫn là tmp-mdb-r1:27017; commit point trễ 6.865 s; lastApplied của primary trễ 1 s
 seq | ghi lúc (s) | event tới lúc (s) | trễ (s)
  10 |   10.1 |   10.2 |   0.1
  11 |   11.1 |   32.8 |  21.7      ← 22 event cùng tới một lúc khi mở khoá
  …
  32 |   32.3 |   32.8 |   0.5
  33 |   33.3 |   33.3 |   0.1
writer maj: 40 ack, 0 lỗi

Lab 10-rollback: cắt mạng primary r1 25 s; writer w:1 nối thẳng r1 (cùng network namespace); watcher A nối thẳng r1, watcher B qua chuỗi replica set:

writer rb: 318 ack, 277 lỗi (đầu tiên: NotWritablePrimary: not primary, lúc +10073 ms sau khi cắt)
watcher rbA: 138 event · watcher rbB: 138 event      ← 180 write đã ack sau lúc cắt: 0 event ở cả hai
document trên r2 sau rollback: 139 · số document trong file rollback: 180

Watcher A đọc trên chính node nhận 180 write đó mà vẫn không thấy gì, vì commit point của nó không tiến được trong lúc bị cô lập. Đây là lý do change stream sạch hơn oplog tailing: thứ không có event là thứ đã mất thật (nằm trong file rollback). Bẫy ngược lại: chỉ theo dõi độ trễ ghi thì không thấy sự cố này — phải theo dõi độ trễ event (mục 6.1).

5.7 Failover: driver nghe tiếp được, nhưng không như nhau

Lab 10-failover (3 lần chạy, tổng 5 vòng): writer w:majority mỗi 20 ms; giây 10 docker kill primary; giây 40 bật lại (priority 2 giành lại primary — failover lần hai). Bốn consumer cùng lúc:

• Node chờ 30 s dù driver đã thấy primary mới ở +10,4 s (SDAM: không có primary (+142 ms) → tmp-mdb-r3 (+10379 ms)). Source _processErrorIteratorMode: await this.cursor.close() rồi mới _resume() → selectServer. connectTimeoutMS 3 s không đổi gì, serverSelectionTimeoutMS 5 s thì resume ở +10,7 s — khớp với "đóng cursor cũ chờ chọn lại server đã chết tới hết hạn" (suy từ số đo + source). Hạn ngắn thì lỗi gốc lọt ra app (+10,1 s): app vẫn phải tự mở lại từ token.
• Lỗi không có nhãn ResumableChangeStreamError thì driver không tự nghe tiếp. Lúc primary cũ giành lại vị trí: InterruptedDueToReplStateChange: Change stream readPreference '{ mode: "primary" }' can no longer be satisfied after a replica set election, labels=[] → lọt ra app ở cả 4 lần chạy Node.
• PHP mặc định serverSelectionTryOnce: lần resume nội bộ duy nhất chọn server một lần rồi ném No suitable servers found (`serverSelectionTryOnce` set): [connection refused calling hello on 'tmp-mdb-r1:27017']. Topology type: ReplicaSetNoPrimary. Worker không bắt lỗi thì chết; supervisor chạy lại watch() không token → bắt đầu từ "bây giờ" → mất toàn bộ event ghi trong lúc chết.

5.8 Sharded: một shard rảnh làm chậm event của cả cluster

Lab 10-shard (2 shard 1 node, config server, mongos; collection chia 2 chunk): ghi 1 document rồi đo tới lúc event qua mongos.

-- periodicNoopIntervalSecs 10 (mặc định)
chỉ ghi vào shard1 (shard2 rảnh): p50 9438 ms, p90 9756 ms, max 9759 ms · [8108 9498 9322 9413 9614 9094 9544 …]
xen kẽ shard1 / shard2:            p50 7212 ms, p90 14900 ms, max 15731 ms · [7212 3641 14896 4455 14496 4084 …]
thứ tự: 200/200 event · đảo seq 0 · đảo clusterTime 0
-- periodicNoopIntervalSecs 1 (shard + config server khởi động với --setParameter)
chỉ ghi vào shard1 (shard2 rảnh): p50 839 ms, p90 1440 ms, max 1476 ms
xen kẽ shard1 / shard2:            p50 855 ms, p90 1127 ms, max 1295 ms

Mỗi event phải chờ shard rảnh (và config server) có entry oplog mới qua clusterTime của nó. Phép đo tuần tự khoá pha theo nhịp noop nên gần như luôn chờ trọn chu kỳ. Ghi xen kẽ còn tệ hơn (tới 15,7 s): noop writer chỉ ghi khi "there has been no writes to this replica set within write interval" (noop_writer.cpp), nên shard vừa có write thật sẽ bỏ lượt noop kế tiếp → chờ tới gần 2 chu kỳ. Manual khuyên hạ periodicNoopIntervalSecs cho cluster có shard "cold". Thứ tự toàn cục thì giữ đúng.

5.9 invalidate: drop/rename đóng stream

Collection-level stream gặp rename/drop: … → rename → invalidate, rồi cursor đóng. Mở lại bằng token của invalidate:

resumeAfter: InvalidResumeToken (260): Attempting to resume a change stream using 'resumeAfter' is not allowed from an invalidate notification
startAfter: mở được

Bẫy: job "đổi tên collection để swap dữ liệu" ($out, renameCollection với dropTarget) giết mọi consumer của collection đích. Watch ở mức database (thấy rename/drop như event thường, không invalidate) nếu có kiểu job này.

6. Vận hành trong production

6.1 Theo dõi gì

Lưu ý: metrics.changeStreams.cursor.bytesRead gần như không đổi trong lab (0–13 287 byte trong lượt đọc hàng trăm MB) — không dùng nó để đo lượng oplog đã quét; dùng docsExamined.

6.2 Chi phí: mỗi cursor một lần quét

Lab 10-cost (Node, writer 140–200 insert/s vào cs_cost + gấp 4 vào cs_noise không ai watch, 15 s, 2 lượt):

docsExamined ≈ số entry oplog của cả hai collection (insertMany 10/40 document = 1 entry) ở mọi cấu hình: filter không giảm phần quét, chỉ giảm phần dựng + gửi event. 50 cursor ≈ 0,75 core. Cần nhiều consumer cùng loại thì mở 1 stream rồi fan-out trong app/broker, đừng 1 stream cho mỗi user/websocket.

6.3 Tham số ảnh hưởng lớn

6.4 Worker Laravel: vòng lặp đúng

Ghép và rút gọn từ laravel/10-outbox.php (relay) + 10-watch.php (vòng bắt lỗi "robust", đã chạy qua failover thật); phần lọc mã lỗi 286/260/47 là thêm cho production, lab chưa chạy đúng đoạn đó. Ở app thật, đặt vào handle() của một artisan command chạy dưới supervisor:

$outbox = DB::connection('mongodb')->getCollection('outbox');      // MongoDB\Collection gốc: laravel-mongodb không có watch()
$state  = DB::connection('mongodb')->getCollection('relay_state');
$token  = $state->findOne(['_id' => 'orders-relay'])['token'] ?? null;
$open   = fn ($t) => $outbox->watch([['$match' => ['operationType' => 'insert']]],
              ['maxAwaitTimeMS' => 1000, 'batchSize' => 500] + ($t ? ['resumeAfter' => $t] : []));
$cs = $open($token); $cs->rewind();
while (! $this->shouldStop()) {                                   // SIGTERM → dừng sau batch hiện tại
    try {
        if ($cs === null) { $cs = $open($token); $cs->rewind(); }
        if ($cs->valid()) {
            $e = $cs->current();
            $publisher->publish((string) $e['documentKey']['_id'], $e['fullDocument']);  // idempotent theo id outbox
        }
        $token = $cs->getResumeToken();                          // cả khi batch rỗng: high-water-mark
        $state->updateOne(['_id' => 'orders-relay'], ['$set' => ['token' => $token]], ['upsert' => true]);  // lưu SAU khi xử lý
        $cs->next();                                              // chờ tối đa maxAwaitTimeMS
    } catch (MongoDB\Driver\Exception\ServerException $ex) {
        if (in_array($ex->getCode(), [286, 260, 47], true)) throw $ex;   // HistoryLost / invalidate / thiếu image: cần người
        $cs = null; usleep(500_000);                              // 11602, NotPrimary…: mở lại từ $token
    } catch (MongoDB\Driver\Exception\ConnectionException $ex) {     // failover: "No suitable servers found (serverSelectionTryOnce)"
        $cs = null; usleep(500_000);
    }
}

Lưu token mỗi vòng là bản đơn giản; muốn nhanh thì lưu mỗi N event và chấp nhận trùng tối đa N − 1 (mục 5.1). PHP không có thread nền: next() chặn tới maxAwaitTimeMS, nên SIGTERM của deploy chỉ được xử lý giữa hai lần next() — dùng pcntl_async_signals(true) + cờ dừng, đừng exit trong signal handler khi chưa lưu token.

6.5 Runbook: khi gặp ChangeStreamHistoryLost

1. Dừng consumer (đừng để supervisor chạy lại vô hạn với token chết).
2. Ghi mốc T = db.adminCommand({hello: 1}).operationTime, rồi đồng bộ lại toàn bộ (snapshot / quét collection) sang đích — ghi đích phải là upsert idempotent.
3. Mở stream bằng startAtOperationTime: T (T phải còn trong oplog — quét lâu hơn cửa sổ oplog thì quay lại bước 2 với oplog lớn hơn / minRetentionHours).
4. Sau sự cố: tăng cửa sổ oplog, bật lưu high-water-mark token, thêm cảnh báo độ trễ consumer. Debezium làm đúng quy trình này với snapshot mode when_needed.

7. So sánh và ma trận quyết định

Khác biệt gốc giữa change stream và PostgreSQL logical decoding là ai giữ lịch sử (Hình 4). Lab 10-pg trên container tạm PostgreSQL 18.4 (wal_level=logical — my-test-pgdb dùng chung đang replica: ERROR: logical decoding requires "wal_level" >= "logical"):

chưa có slot: pg_wal 16 MB (max_wal_size 64MB)
slot test_decoding, consumer ngừng đọc, mỗi vòng 200 000 dòng + CHECKPOINT:
  wal_status extended · slot giữ 123 MB · pg_wal 144 MB
  wal_status extended · slot giữ 247 MB · pg_wal 272 MB
  wal_status extended · slot giữ 371 MB · pg_wal 384 MB
max_slot_wal_keep_size = 64MB: wal_status lost · invalidation_reason wal_removed · pg_wal 64 MB
  → ERROR: can no longer access replication slot "tmp_mdb_slot"  DETAIL: This replication slot has been invalidated due to "wal_removed".
idle_replication_slot_timeout = 1min (PG 18), chờ 70 s + CHECKPOINT: wal_status lost · invalidation_reason idle_timeout

Mặc định PostgreSQL chọn "consumer không bao giờ mất vị trí, DB trả giá bằng đĩa" (slot chết còn ghim cả XID — transaction ID wraparound); MongoDB chọn "DB không bao giờ đầy vì consumer, consumer trả giá bằng lịch sử". Đặt giới hạn cho slot thì PostgreSQL rơi về đúng hành vi của MongoDB.

Dùng change stream khi: cần phát thay đổi đã commit của MongoDB sang hệ khác theo đúng thứ tự, với vài consumer; outbox nằm trong MongoDB; đội chấp nhận viết worker có resume + idempotent. Tránh khi: cần lưu và phát lại lịch sử dài (dùng Kafka làm log, change stream chỉ là nguồn); hàng chục/hàng trăm consumer độc lập (1 relay → broker → fan-out); cần phân việc song song kiểu consumer group (Redis Stream / Kafka); cluster PSA hoặc thường xuyên mất đa số; cần "ý định" của lệnh thay vì trạng thái (ghi event tường minh vào outbox). Debezium MongoDB connector cũng chỉ đọc change stream ("does not read the oplog directly") nên mọi giới hạn ở trên vẫn còn, chỉ thêm snapshot tự động và Kafka phía sau.

8. Góc nhìn tech lead: tác động tổ chức

• Kiến trúc: watch thẳng collection nghiệp vụ thì hình dạng document/diff thành hợp đồng với mọi consumer — đổi schema là vỡ. Outbox với event tường minh (type, version, payload) làm hợp đồng công khai; collection nghiệp vụ là chi tiết nội bộ.
• Đội và trực: runbook ChangeStreamHistoryLost (6.5), cảnh báo độ trễ consumer, người hiểu commit point ("event đứng im" hay bị đổ cho consumer). Test bắt buộc: kill -9 relay, failover, broker treo, consumer chậm — không thì test xanh mà production lệch (5.4).
• Mặc định cho cả team (1 thư viện nội bộ, không để mỗi service tự làm): lưu token sau xử lý, lưu cả high-water-mark, sink idempotent theo id event, không updateLookup cho lịch sử, bắt lỗi trong vòng lặp worker.
• Capacity: cửa sổ oplog ≥ 2 × thời gian consumer có thể offline (deploy chậm, sự cố broker cuối tuần), đo ở giờ cao điểm; số cursor tỉ lệ thuận với CPU primary.
• Lock-in / exit strategy: token, updateDescription, ngữ nghĩa invalidate là riêng của MongoDB. Lối thoát: consumer đọc outbox với payload tự định nghĩa (đổi nguồn sang Debezium/PostgreSQL thì chỉ thay relay), hoặc đi qua Debezium ngay từ đầu (định dạng event chung giữa nhiều DB). Ghi vào ADR: nguồn event, chỗ giữ vị trí, ngữ nghĩa giao hàng (at-least-once), cách resync.
• Rủi ro dài hạn: lưu lượng oplog tăng theo dữ liệu → cửa sổ co lại âm thầm; 9.0.x đang đổi cách nhắm shard (v2 bị tắt ở 9.0.2, bật lại ở bản sau thì hành vi trễ / cursor đổi) — kiểm tra lại sau mỗi lần nâng bản.

9. Anti-patterns và bài học thực tế

1. Dùng change stream làm hàng đợi việc: không ack, không retry từng việc, không chia tải theo key — relay sang broker có consumer group.
2. Không lưu token, hoặc chỉ lưu _id của event cuối — mất khúc giữa khi restart (5.7), mất lịch sử ở collection ít ghi (5.5).
3. Side effect không idempotent — trùng sau mỗi lần chết (5.1).
4. updateLookup để dựng audit / lịch sử giá — sai bản khi consumer chậm (5.4).
5. Một stream cho mỗi user / websocket / tenant — mỗi cursor một lần quét oplog (6.2).
6. Đổi $match của consumer rồi resume bằng token cũ — có thể mất vị trí (3.2).
7. Định cỡ oplog theo tải thường ngày — job backfill làm cửa sổ còn 2,4–4,2 phút (bài 7 mục 5.6).
8. Dual write "cho nhanh" — mất im lặng (5.3).
9. PSA + change stream — mất secondary là mọi consumer đứng (5.6); manual nói thẳng "change streams may remain idle".
10. Worker không bắt lỗi trong vòng lặp — PHP serverSelectionTryOnce làm mọi failover thành worker chết (5.7).

Bài học công khai: oplog driver của Meteor tail cả oplog rồi so với từng query đang theo dõi — "the oplog records everything … whether your app's publications care about those documents or not" (Galaxy blog, 2026-07); 3.5 chuyển sang change stream để lọc phía server, nhưng docs vẫn cảnh báo nó "can also increase work on busy collections if your selectors are broad" (Meteor docs). MongoDB Kafka connector có trang riêng cho lỗi ResumeToken not found … error 286 (ChangeStreamHistoryLost), nguyên nhân thứ hai là "the monitored collection isn't updated frequently enough" — đúng bẫy 5.5, chữa bằng heartbeat.interval.ms. Debezium: "Kafka can only guarantee that consumers receive every message at least once … consumers must be able to handle duplicate messages". Chưa tìm thấy postmortem công khai có số liệu về sự cố change stream — phần này dựa trên tài liệu của các công cụ và lab.

10. Câu hỏi phụ interviewer hay hỏi tiếp

Change stream khác tail oplog thế nào?

Cùng nguồn là oplog, nhưng change stream lọc phía server, định dạng ổn định, chỉ trả thứ đã majority-commit (không bao giờ thấy write bị rollback — lab 0/180), có resume token, và qua mongos thì có thứ tự toàn cục.

Exactly-once được không?

Server phát mỗi event một lần; tới sink thì at-least-once. Exactly-once chỉ khi sink là chính MongoDB và kết quả + token commit trong cùng một transaction (lab 3 000 / 3 000). Sink ngoài: idempotent theo id event.

Consumer tắt 2 ngày thì sao?

Nếu vị trí còn trong oplog: nghe tiếp, nhận đủ. Nếu không: ChangeStreamHistoryLost (286), phải resync (snapshot + startAtOperationTime từ mốc trước snapshot). Phòng: cửa sổ oplog ≥ 2 × thời gian offline tối đa, oplogMinRetentionHours.

Vì sao event chậm dù write nhanh?

Replica set: commit point trễ (secondary lag, PSA). Sharded: mongos chờ shard rảnh, tới ~2 × periodicNoopIntervalSecs (lab 9,4 s p50, max 15,7 s).

fullDocument: updateLookup có đáng tin?

Không cho lịch sử: nó là một lần đọc lúc giao event, trả bản hiện tại (lab 32–69 % sai khi consumer chậm, null nếu đã xoá). Cần đúng bản: pre/post image (6.0+), trả bằng dung lượng config.system.preimages.

Outbox đọc bằng change stream hay polling?

Change stream: trễ ms, không quét bảng, có thứ tự commit; polling: đơn giản, không phụ thuộc cửa sổ oplog, nhưng thứ tự theo id dễ lệch khi transaction song song. Cả hai đều cần consumer idempotent.

Làm sao chạy nhiều relay song song?

Một stream là một luồng tuần tự. Chia bằng $match (hash của key % N) cho N relay, mỗi relay giữ token riêng; mất thứ tự giữa các nhóm, giữ thứ tự trong nhóm. Mỗi relay vẫn quét toàn bộ oplog.

So với PostgreSQL logical decoding?

Slot giữ WAL cho consumer (consumer chết → đĩa đầy: lab 16 → 384 MB); oplog quay vòng (consumer chết → mất lịch sử). max_slot_wal_keep_size / idle_replication_slot_timeout làm slot "mất lịch sử" giống MongoDB.

Phụ lục: lab đã chạy

Môi trường: MongoDB 9.0.2 — rs0 = tmp-mdb-r1 (priority 2), r2, r3, mỗi node --memory 1g, cache 0,25 GB; node oplog nhỏ tmp-mdb-h1 (--oplogSize 5 --setParameter oplogTruncationCheckPeriodSeconds=1); sharded: config server + 2 shard 1 node + mongos (lần 2 với periodicNoopIntervalSecs=1); MongoDB 8.0.32 để so (tmp-mdb-80). Node 22 + driver mongodb 7.7.0; PHP 8.3 + ext-mongodb 2.5.3 + mongodb/mongodb 2.5.0 + laravel-mongodb 5.11.0 (Laravel 13.34) + phpredis 6.3.0; Redis 8.8.0 tạm (tmp-mdb-redis) làm broker; PostgreSQL 18.4 tạm (tmp-mdb-pg, đã xoá; slot tmp_mdb_* đã drop, my-test-pgdb không đổi gì). Chạy lại:

cd labs/mongodb
./10-change-stream.sh            # tất cả (~60 phút), hoặc từng phần:
./10-change-stream.sh events|outbox|lookup|history|majority|rollback|failover|cost|large|v80|shard|pg
ROUNDS=1 ./10-change-stream.sh failover
source ./lib.sh && mdb_down $RS_NODES tmp-mdb-redis   # dọn

Output: out/10-*.txt — mỗi phần một file; 10-history-run1.txt + 10-history-trunc.txt: lần chạy với chu kỳ cắt oplog mặc định 300 s và log mongod; 10-failover-run1/run2.txt: hai lần chạy trước (chưa có / có đo SDAM). Script: 10-change-stream.sh, js/10-*.js, node/10-*.mjs, laravel/10-*.php, sql/10-pg.sql.

Bẫy khi dựng lab: oplog 5 MB không bị cắt suốt 120 s (oplogTruncationCheckPeriodSeconds 300, phải đặt lúc khởi động); relay chết lặp 60 s làm script chờ mãi nếu không có vòng "supervisor"; increment() của query builder không upsert; my-test-pgdb có wal_level=replica nên logical slot chạy trên container tạm.

Chưa verify: reader v2 (precise shard targeting) — bị tắt ở 9.0.2, chưa thử bản 9.0.4; cluster nhiều shard thật, shard nhiều node; Atlas; ghi liên tục (không dừng) với oplog nhỏ — nhịp cắt oplog trong trường hợp đó; Debezium và MongoDB Kafka connector chạy thật; driver khác Node/PHP (Java, Go, Python) lúc failover; độ trễ thêm của Kafka; PSA thật (lab dùng fsyncLock 2 secondary để giả lập mất đa số).