A change stream is a cursor that returns an event every time a document in a collection, database or whole cluster changes. It’s how you keep a search index, a cache or another service in step with MongoDB without polling. Opening one takes a line of code. Keeping one correct across restarts, failovers and deploys is the real subject, and it all comes down to the resume token.
Change streams need a replica set or sharded cluster, and there isn’t one running where this was written, so treat the event documents below as illustrative. Their shape follows the MongoDB 7.0 and 8.0 manual.
What an event looks like
const stream = db.orders.watch();An update to an order produces something like:
{
_id: { _data: '8266F3A1B2000000012B042C0100296E5A1004...' },
operationType: 'update',
clusterTime: Timestamp({ t: 1729420210, i: 1 }),
wallTime: ISODate('2026-10-20T10:30:10.412Z'),
ns: { db: 'shop', coll: 'orders' },
documentKey: { _id: ObjectId('6714f0c2a1b2c3d4e5f60718') },
updateDescription: {
updatedFields: { status: 'shipped', 'shipping.carrier': 'DHL' },
removedFields: [],
truncatedArrays: []
}
}Three parts matter most. _id is the resume token, an opaque value that marks this event’s position in the stream. documentKey identifies which document changed (it includes the shard key on sharded collections). And updateDescription holds only the delta: the fields that changed, not the whole document.
Change streams are built on the oplog, the replica set’s log of operations, and they only return changes that a majority of members have committed. That’s a useful guarantee: if a primary fails and rolls back writes that never reached a majority, your consumer never saw them.
Filtering and asking for the full document
Because watch() accepts an aggregation pipeline, you can filter on the server instead of discarding events in your code:
const pipeline = [
{ $match: { operationType: { $in: ['insert', 'update', 'replace', 'delete'] } } },
];
const stream = db.collection('orders').watch(pipeline, { fullDocument: 'updateLookup' });Don’t add stages that modify or remove _id. The server needs it to resume the stream, and the manual says so explicitly.
fullDocument: 'updateLookup' asks the server to attach the current version of the document to update events. Interviewers like this option because its behaviour is subtle. The lookup happens when the event is processed, not when the update happened. If an order was updated to shipped and then to delivered a moment later, the event for the first update can arrive carrying the delivered document. If the order was deleted in between, fullDocument is null. For a search index that only needs the latest state, that’s fine. For an audit log that must know what the document looked like at each step, it’s wrong.
For that, MongoDB 6.0 and later support pre- and post-images. Enable them on the collection, then request them:
db.runCommand({ collMod: 'orders', changeStreamPreAndPostImages: { enabled: true } });
const stream = db.collection('orders').watch([], {
fullDocument: 'whenAvailable', // post-image as of this event
fullDocumentBeforeChange: 'whenAvailable', // pre-image
});Images are stored in a separate system collection and cost extra writes and storage, so enable them only where you need them, and set a retention time.
Resume tokens and where to store them
A process that watches a stream will restart: deploys, crashes, node failures. When it comes back, it should continue exactly where it left off. That’s what the resume token is for: open the stream with resumeAfter: token and the server replays everything after that event.
The key question is when you save the token. Here’s a consumer that keeps an Elasticsearch index in sync, using the Node.js driver:
const checkpoints = db.collection('stream_checkpoints');
async function run() {
const saved = await checkpoints.findOne({ _id: 'orders-to-search' });
const stream = db.collection('orders').watch(pipeline, {
fullDocument: 'updateLookup',
...(saved ? { resumeAfter: saved.token } : {}),
});
for await (const event of stream) {
const id = event.documentKey._id.toString();
if (event.operationType === 'delete' || event.fullDocument === null) {
await search.delete({ index: 'orders', id }).catch(ignoreNotFound);
} else {
await search.index({ index: 'orders', id, document: toSearchDoc(event.fullDocument) });
}
await checkpoints.updateOne(
{ _id: 'orders-to-search' },
{ $set: { token: event._id, at: event.wallTime } },
{ upsert: true },
);
}
}The order of the last two steps is the important decision. The token is saved after the side effect. If the process dies between indexing and checkpointing, it restarts from the previous token and indexes that order again. Saving it before the side effect would mean a crash in between skips the event forever.
So delivery is at least once, and the consumer has to tolerate duplicates. Here that’s easy: indexing a document by its _id is idempotent, and so is deleting it. For a side effect that isn’t naturally idempotent, like sending a “your order shipped” email, store a dedupe key such as the event’s token or (documentKey, clusterTime) and skip events you’ve already handled.
Writing a checkpoint per event doubles your write load on busy streams. Checkpointing every N events or every few seconds is common, at the cost of replaying a few more on restart. The driver also exposes stream.resumeToken, which can advance even when your $match filters events out, so checkpointing it periodically stops a quiet, heavily filtered stream from falling behind the oplog.
Within a single process, the driver already handles brief failures. If the primary steps down, a compliant driver treats the error as resumable and reopens the stream from the last token it saw, without your code noticing. Your own checkpointing is for when the process restarts.
When a resume token stops working
Two situations break resumption, and you should be ready for both.
The oplog has moved on. The oplog is a capped collection with a fixed size, so it holds a limited window of history: maybe 48 hours on a quiet cluster, maybe 3 hours under heavy load. If your consumer is down longer than that window, its saved token points at an entry that no longer exists, and opening the stream fails with a ChangeStreamHistoryLost error (the exact message varies by version). There is no way to replay the missing changes. You need a full resync: note the current cluster time, copy every document into the target, then open a new stream with startAtOperationTime set to that time so nothing that changed during the copy is missed.
To make that rare, size the oplog to cover your worst realistic outage (replSetResizeOplog, or minRetentionHours to keep entries for a minimum time), and alert when the consumer’s lag, now - checkpoint.at, approaches the window.
An invalidate event. Dropping or renaming a watched collection ends a collection-level stream with an invalidate event. You can’t resumeAfter an invalidate token; use startAfter with it to open a new stream past that point. Teams sometimes meet this when a migration script renames a collection and the sync service quietly stops.
One more edge: an event larger than 16 MB, which can happen with pre- and post-images on big documents, fails the stream. MongoDB 7.0 (and 6.0.9) added the $changeStreamSplitLargeEvent stage, which splits such events into fragments you reassemble.
Interview exercise: the Monday morning failure
The order-sync service was stopped on Friday evening for maintenance and restarted on Monday morning. It immediately crashes with a “resume point may no longer be in the oplog” error, and the search index is missing every order since Friday. The cluster’s oplog window is about 30 hours. What happened, how do you recover, and what would you change?
Answer
The service was down for roughly 60 hours, twice the oplog window. Its saved resume token refers to an oplog entry from Friday that has since been overwritten, so the server can’t reconstruct the stream from that point. Nothing is wrong with the code; the history is simply gone.
Recovery is a resync, done in an order that doesn’t lose writes. First record the current cluster time (for example from db.runCommand({ hello: 1 }).operationTime). Then rebuild the index from a full scan of orders, ideally into a new index that you swap in when it’s complete. Finally open a new change stream with startAtOperationTime set to the recorded time, so changes made during the rebuild are replayed. Some of those will repeat what the scan already wrote, which is harmless because indexing by _id is idempotent.
To prevent it: make the oplog cover the longest planned downtime plus a margin, using replSetResizeOplog or minRetentionHours. Monitor consumer lag against the oplog window and alert well before they meet. And put the resync path in the service as a documented, tested procedure, because some day it will be needed again.