Resume with tokens

Continue a change stream where you left it after your process crashes or restarts. Each event carries a resume token that identifies it.

Requirements

• A user with the find and changeStream privileges on what you watch.
• A place to store the last token, such as a collection of your own.

Store the token of each event

for await (const change of db.orders.watch()) {
  await handle(change);
  await db.tokens.updateOne({ _id: "orders" }, { $set: { token: change._id } }, { upsert: true });
}

Resume

const saved = await db.tokens.findOne({ _id: "orders" });
const stream = db.orders.watch([], { resumeAfter: saved.token });

startAfter works the same as resumeAfter and also serves to continue after an invalidate event.

Check

If you stop the process, insert a document and start it again, the loop receives that event.

Common problems

• You cannot resume because the token is older than the history the cluster keeps, 24 hours by default. Raise teselaHistoryRetention if your processes can be stopped for longer.

See also

• Open a change stream
• Invalidate a cache with change streams