การกลับมาใช้สตรีมการเปลี่ยนแปลงต่อหลังหยุดชะงัก

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.

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.

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.

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.

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.

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.

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.

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.

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).

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.

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.

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.