Resuming Change Streams After Interruption
Learners will persist the resume token and restart a change stream from the last processed event to guarantee at-least-once delivery.
Resuming Change Streams After Interruption is a free MongoDB Academy lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the MongoDB Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.
The Problem: What Happens After a Crash?
A change stream consumer process can be interrupted by network failures, deployments, crashes, or intentional restarts. Without a mechanism to resume from the last processed event, your application would restart the stream from the current moment and miss all events that occurred during the downtime. MongoDB's resume tokens solve this problem by letting you reopen a stream from an exact position in the oplog history.
What Is a Resume Token?
Every change event document has an _id field that serves as its resume token—a binary opaque value that uniquely identifies the event's position in the oplog. It looks like { _data: '...' } where the data is a hex-encoded internal identifier. You do not need to understand its contents; you only need to save it and pass it back to MongoDB to resume from that exact position.
// A resume token is the _id field of a change event
// Example structure (your values will differ):
// {
// _id: {
// _data: '8264CE3B7200000002463C6F5A100...'
// }
// }
// Store it as-is — do not parse or modify the _data valuePersisting the Resume Token
After processing each event, save the resume token to a persistent store (database, file, Redis) before acknowledging that the event was processed. This 'checkpoint' pattern ensures you can pick up from the last successfully processed event after a restart. Save after processing, not before, to avoid skipping events if your processor crashes during handling.
const CHECKPOINT_COLLECTION = 'changeStreamCheckpoints';
const STREAM_ID = 'orderProcessorV1';
for await (const change of changeStream) {
// Process the event
await processOrderChange(change);
// Save token AFTER successful processing
await db.collection(CHECKPOINT_COLLECTION).updateOne(
{ streamId: STREAM_ID },
{ $set: { resumeToken: change._id, processedAt: new Date() } },
{ upsert: true }
);
}Resuming With resumeAfter
To resume from a saved token, pass it to watch() via the resumeAfter option. MongoDB will replay events from after the event identified by the token—your last processed event is not replayed. If the token refers to an event still in the oplog, MongoDB starts delivering from the next event. This provides exactly-once delivery when combined with idempotent event processing.
async function startOrResumeStream() {
// Load the last saved resume token
const checkpoint = await db.collection('changeStreamCheckpoints')
.findOne({ streamId: 'orderProcessorV1' });
const watchOptions = checkpoint
? { resumeAfter: checkpoint.resumeToken }
: {}; // Start from current position if no token
const changeStream = db.collection('orders').watch([], watchOptions);
for await (const change of changeStream) {
await processOrderChange(change);
await saveResumeToken(change._id);
}
}startAfter vs resumeAfter
MongoDB provides two similar options: resumeAfter and startAfter. resumeAfter resumes from a normal operation token but throws an error if given an 'invalidate' event token. startAfter works the same way but can also resume after an invalidate event—useful when you want to reopen a stream on a collection after it was dropped and recreated. For most use cases, resumeAfter is the correct choice.
// resumeAfter — standard use case, does not work after invalidate tokens
db.collection('orders').watch([], { resumeAfter: savedToken });
// startAfter — can also start after an invalidate event
// (use when the collection may have been dropped and recreated)
db.collection('orders').watch([], { startAfter: savedToken });startAtOperationTime for Time-Based Resumption
If you do not have a resume token but know the timestamp from which you want to start, you can use startAtOperationTime with a MongoDB Timestamp object. This opens the stream at a specific cluster time rather than a specific event. It is useful for replay scenarios—for example, reprocessing all events since a deployment started—but requires that the oplog still contains history back to that time.
const { Timestamp } = require('mongodb');
// Replay events since a specific time
const startTime = new Timestamp({ t: Math.floor(Date.now() / 1000) - 3600, i: 1 }); // 1 hour ago
const changeStream = db.collection('orders').watch([], {
startAtOperationTime: startTime
});
// Events from 1 hour ago will be deliveredOplog Retention and the Horizon
Change stream resumption only works if the token's corresponding event is still in the oplog. MongoDB's oplog is a capped collection with finite size—older entries are overwritten as new ones arrive. On Atlas, you can configure oplog retention (typically 24–72 hours). If your application was down longer than the oplog retention window, the token points to an expired position and MongoDB will return an ChangeStreamHistoryLost error (code 286). Your application must handle this by starting fresh.
async function startOrResumeWithFallback() {
const checkpoint = await loadResumeToken();
try {
const options = checkpoint ? { resumeAfter: checkpoint } : {};
const stream = db.collection('orders').watch([], options);
for await (const change of stream) {
await processChange(change);
await saveResumeToken(change._id);
}
} catch (error) {
if (error.code === 286) { // ChangeStreamHistoryLost
console.warn('Resume token expired — starting from now and running catch-up scan');
await deleteResumeToken();
await catchUpScan(); // full collection scan to catch missed changes
await startOrResumeWithFallback();
} else {
throw error;
}
}
}Idempotent Event Processing
Because change streams provide at-least-once delivery (the same event might be delivered more than once after a resume), your event handlers must be idempotent—processing the same event twice must produce the same result as processing it once. Techniques include: checking if the document already reflects the change before applying it, using upsert instead of insert, or storing processed event IDs in a 'processed events' set and skipping duplicates.
async function processOrderChange(change) {
if (change.operationType === 'insert') {
const { fullDocument } = change;
// Idempotent upsert — safe to replay:
await db.collection('orderSummaries').updateOne(
{ _id: fullDocument._id }, // match by _id
{ $set: { ...summarize(fullDocument) } }, // idempotent set
{ upsert: true } // create if not exists
);
}
}Automatic Driver-Level Resume
The MongoDB Node.js driver (and other official drivers) includes automatic resume for transient network errors. When the connection to the MongoDB server is temporarily lost, the driver transparently reopens the change stream from the last received event token without your code needing to handle it. This automatic resume handles many common interruption scenarios—you only need to implement manual resume logic for application-level restarts (process crashes, deployments).
// The driver automatically resumes after network blips
// No code needed in your loop for this case:
for await (const change of changeStream) {
// If the network drops and reconnects, the driver resumes automatically
// and continues delivering events from where it left off.
await processChange(change);
await saveCheckpoint(change._id); // Still save tokens for process restarts
}Production-Ready Change Stream Handler
A production change stream consumer should combine: resume token persistence (to survive process restarts), oplog expiry handling (fallback to a full scan), idempotent processing (safe to replay events), and graceful shutdown (close streams on SIGTERM). These four elements together achieve reliable event processing even in the face of network failures, deployments, and extended downtime.
class ReliableChangeStreamConsumer {
constructor(collection, handler) {
this.collection = collection;
this.handler = handler;
this.running = false;
}
async start() {
this.running = true;
while (this.running) {
const token = await loadToken();
const stream = this.collection.watch([], token ? { resumeAfter: token } : {});
try {
for await (const change of stream) {
await this.handler(change);
await saveToken(change._id);
}
} catch (e) {
if (e.code === 286) { await deleteToken(); continue; } // expired, restart fresh
if (!this.running) break; // shutting down
throw e;
} finally {
await stream.close();
}
}
}
stop() { this.running = false; }
}Graceful Shutdown and Token Flush
On a planned shutdown (deployment, restart), always flush the latest resume token before closing the change stream. If your application processes events in batches (buffering for performance), ensure the token is saved after every batch—not after every individual event. Track the token of the last successfully processed event, not the last received one. This distinction matters: if you receive an event, save its token, but then crash before processing it, you will miss that event when you resume.
// Correct: save token AFTER processing, not before
for await (const change of changeStream) {
await processChange(change); // process first
await saveToken(change._id); // token saved only after success
// If crash happens here, the event was already processed — safe
}
// Wrong: saving token before processing
for await (const change of changeStream) {
await saveToken(change._id); // token saved
await processChange(change); // if crash here, event is skipped on resume!
}Quick Check
Test your understanding of MongoDB & NoSQL Databases concepts from this lesson.
Lesson Recap
In this lesson you learned: the resume token is stored in change._id and must be persisted after processing each event, pass the token to watch() as resumeAfter to restart the stream from the last processed event, and handle ChangeStreamHistoryLost (code 286) when the token has expired from the oplog by restarting from current time and running a catch-up scan. Next up we explore creating Atlas Search indexes for full-text search.
Frequently asked questions
Is the “Resuming Change Streams After Interruption” lesson free?
Yes — the full text of “Resuming Change Streams After Interruption” is free to read here on the web, and the MongoDB Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the MongoDB Academy course, upgrade to CoddyKit PRO.
What will I learn in “Resuming Change Streams After Interruption”?
Learners will persist the resume token and restart a change stream from the last processed event to guarantee at-least-once delivery. You practise MongoDB Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start MongoDB Academy?
No prior experience is required. MongoDB Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.
How long does the “Resuming Change Streams After Interruption” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this MongoDB Academy lesson?
Yes. Every MongoDB Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.
All lessons in this course
- Opening a Change Stream on a Collection
- Change Event Document Structure
- Filtering Events With an Aggregation Pipeline
- Resuming Change Streams After Interruption