Change streams

Watching a collection

1const cs = db.orders.watch(2  [{ $match: { operationType: { $in: ["insert", "update"] } } }],3  { fullDocument: "updateLookup" }4)5while (true) {6  const ev = cs.tryNext()7  if (ev) printjson(ev); else sleep(200)8}

In another shell:

1db.orders.insertOne({ _id: "o-1001", customerId: 42, status: "pending", totalCents: 2500 })2db.orders.updateOne({ _id: "o-1001" }, { $set: { status: "paid" } })

The watcher printed (trimmed, tokens shortened):

1{ "operationType": "insert", "documentKey": { "_id": "o-1001" },2  "fullDocument": { "_id": "o-1001", "customerId": 42, "status": "pending", "totalCents": 2500 },3  "_id": { "_data": "826AD2214B000000012B042C0100296E5A10..." } }4{ "operationType": "update", "documentKey": { "_id": "o-1001" },5  "updateDescription": { "updatedFields": { "status": "paid" }, "removedFields": [], "truncatedArrays": [] },6  "fullDocument": { "_id": "o-1001", "customerId": 42, "status": "paid", "totalCents": 2500 },7  "_id": { "_data": "826AD2214B000000022B042C0100296E5A10..." } }

An update event normally carries only updateDescription, the changed fields. fullDocument: "updateLookup" adds the current document, read when the event is delivered, so it may already include later changes. For the exact before and after state of each change, enable pre- and post-images on the collection (6.0 and later) and use fullDocumentBeforeChange.

Resuming

Each event's _id is its resume token. A consumer that stores the token after processing each event can restart with watch(pipeline, { resumeAfter: token }) and continue exactly after it. That is at-least-once delivery: if the process crashes after acting on an event but before saving its token, it sees the event again, so make the handler idempotent. Do not remove or change _id in your pipeline; the server rejects streams that modify it.

Events are only sent for changes a majority of data-bearing members have, so a consumer never acts on a write that a failover later rolls back.

The oplog window

Resuming only works while the token's position is still in the oplog, which is a capped collection that overwrites old entries. On the test set, db.getReplicationInfo() showed an oplog sized at about 42 GB (by default it takes a share of free disk) but holding only 0.05 hours of history, because the set was new; on a busy production system the window might be hours or days. A consumer down longer than the window gets an error on resume and must rebuild from a full read of the collection. Monitor the window, and size the oplog for your longest expected outage.

Typical uses

Updating a search index or cache, publishing domain events (an outbox: write the order and its event in one transaction, and let a change stream publish the event), and feeding Kafka through the MongoDB Kafka connector.