Change streams and resume tokens

conceptThe concept

Open the interactive lesson to use this diagram.

db.orders.watch(pipeline, options) returns a cursor of change events: { _id: <resume token>, operationType, ns, documentKey, fullDocument?, updateDescription?, clusterTime }.

• Filter with a pipeline: [{ $match: { operationType: "update", "updateDescription.updatedFields.status": { $exists: true } } }].
• fullDocument: "updateLookup" fetches the current document for updates (it may be newer than the change). fullDocument: "whenAvailable"/"required" use stored post-images (6.0+, when changeStreamPreAndPostImages is enabled on the collection), and fullDocumentBeforeChange gives the pre-image.
• Resume: store the _id of the last processed event and reopen with resumeAfter or startAfter (the latter also works after an invalidate). startAtOperationTime starts from a cluster time.

whyWhy it matters

Change streams replace polling and fragile oplog tailing for event-driven architectures: cache invalidation, search index sync, notifications, audit trails, CDC into Kafka or a warehouse. They work across failovers and in sharded clusters with a global order.

Use caseReal-world scenario

When an order becomes shipped, a notification service sends an email. It watches updates where updatedFields.status is "shipped", processes each event, and stores the resume token in its own collection after each success. After a deploy or crash it resumes exactly where it stopped, as long as the oplog still covers that point.

Try itExamples

Run and edit these in the interactive lesson.

db.orders.updateOne({ orderNo: "ORD-100002" }, { $set: { status: "paid" } })
const cs = db.orders.watch([{ $match: { operationType: "update", "updateDescription.updatedFields.status": { $exists: true } } }], { fullDocument: "updateLookup" })
db.orders.updateOne({ orderNo: "ORD-100002" }, { $set: { status: "shipped" } })
db.orders.updateOne({ orderNo: "ORD-100002" }, { $inc: { total: 0.5 } })
const ev = cs.tryNext()
ev.operationType
ev.updateDescription
ev.fullDocument.orderNo
cs.tryNext()

const s1 = db.orders.watch()
db.orders.insertOne({ orderNo: "ORD-CS-1", status: "pending", createdAt: new Date(), total: 5 })
const first = s1.tryNext()
const token = first._id
s1.close()
db.orders.insertOne({ orderNo: "ORD-CS-2", status: "pending", createdAt: new Date(), total: 6 })
const s2 = db.orders.watch([], { resumeAfter: token })
s2.tryNext().fullDocument.orderNo

const all = db.watch([{ $match: { operationType: { $in: ["insert", "delete"] } } }])
db.reviews.insertOne({ productId: "P001", customerId: 1, rating: 5, title: "Great", createdAt: new Date() })
const e = all.tryNext()
e.ns

internalsUnder the hood

Change streams are an aggregation over the oplog, run with majority read concern so events are never rolled back. On sharded clusters, mongos merges per-shard streams in cluster-time order. A resume token encodes the cluster time and document key; if the oplog has rolled past it (the consumer was down longer than the oplog window), resuming fails with ChangeStreamHistoryLost and you need a full resync. Size the oplog for your longest expected consumer outage.

Avoid / preferDo this, not that

setInterval(() => db.orders.find({ updatedAt: { $gt: last } }), 1000)

const cs = db.orders.watch([
  { $match: { operationType: { $in: ["insert", "update"] } } }
])

cs.on("change", e => handle(e))

cs.on("change", async e => { await handle(e); await saveToken(e._id) })
// on restart: collection.watch(pipeline, { resumeAfter: savedToken })

PitfallsCommon mistakes

• Not persisting resume tokens, so a restart either misses or reprocesses events.
• Consumers down longer than the oplog window: ChangeStreamHistoryLost.
• Assuming updateLookup returns the document as it was right after that change; it returns the current version.
• Heavy processing inside the change stream loop instead of handing work to a queue.
• Opening one change stream per user or per entity; each is a server-side cursor scanning the oplog. Open few streams and fan out in the application.

LimitsWhen not to use it

A change stream is not a message queue with retention. If a consumer is down longer than the oplog window, its resume token is gone and it must resync from scratch. For fan-out to many consumers or long retention, feed a broker (for example with the Kafka connector) instead of opening many streams.

SecuritySecurity notes

Change events can carry full documents and pre-images. Strip sensitive fields with $project in the stream pipeline before they leave the database, and remember that stored pre-images keep old values around longer than the documents themselves.

ChallengeTry it yourself

db.cs_lab.drop()
db.createCollection("cs_lab")
const cs = db.cs_lab.watch()
db.cs_lab.insertOne({ _id: 1, status: "new" })
db.cs_lab.updateOne({ _id: 1 }, { $set: { status: "paid" } })
const seen = []
let ev
while ((ev = cs.tryNext())) seen.push(ev.operationType)
seen

InterviewQuestions you should be able to answer

QuizCheck your understanding

4 questions with instant feedback in the interactive lesson; a wrong answer goes into your review deck.