Reprendre les flux de changements après une interruption
Les apprenants conserveront le jeton de reprise et redémarreront un flux de changements à partir du dernier événement traité afin de garantir une livraison au moins une fois.
Reprendre les flux de changements après une interruption est une leçon MongoDB Academy gratuite sur CoddyKit. Ceci est la leçon 4 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage MongoDB Academy, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours MongoDB Academy comprend 4 leçons au total.
Certaines parties de cette leçon n'ont pas encore été traduites et s'affichent en anglais.
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.
Questions Fréquemment Posées
La leçon « Reprendre les flux de changements après une interruption » est-elle gratuite ?
Oui — le texte complet de « Reprendre les flux de changements après une interruption » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours MongoDB Academy, passe à CoddyKit PRO. Le cours MongoDB Academy comprend 4 leçons au total.
Qu'est-ce que j'apprendrai dans « Reprendre les flux de changements après une interruption » ?
Les apprenants conserveront le jeton de reprise et redémarreront un flux de changements à partir du dernier événement traité afin de garantir une livraison au moins une fois. Tu pratiques MongoDB Academy avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.
Dois-je avoir de l'expérience pour commencer MongoDB Academy ?
Aucune expérience préalable n'est requise. MongoDB Academy sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 4 sur 4.
Combien de temps prend la leçon « Reprendre les flux de changements après une interruption » ?
La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.
Peux-tu écrire et exécuter du code dans cette leçon MongoDB Academy ?
Oui. Chaque leçon MongoDB Academy inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.
Toutes les leçons de ce cours
- Ouvrir un flux de changements sur une collection
- Structure du document d’un événement de changement
- Filtrer des événements avec une chaîne de traitement d’agrégation
- Reprendre les flux de changements après une interruption