Resume Tokens

Resume Tokens

Resume Tokens and Streaming Changes to Clients

Every event's _id is an opaque resume token. Store the token of the last event you finished processing, and a new stream opened with resumeAfter: token delivers everything after it, including changes made while your process was down. That is what makes a change stream a dependable feed for a search index, a cache or an audit log rather than a best-effort notification.

Resuming a change stream after a restartJavaScript

const filter = [{ $match: { operationType: { $in: ['insert', 'update'] } } }];
const first = orders.watch(filter);
await first.tryNext();                         // opens the cursor
await orders.insertMany([{ _id: 1, s: 'new' }, { _id: 2, s: 'new' }]);
let token;
for (let i = 0; i < 2; i++) { const e = await first.next(); token = e._id;
  console.log('live   ', e.operationType, JSON.stringify(e.documentKey)); }
await first.close();                           // the app is now offline
await orders.insertOne({ _id: 3, s: 'new' });
await orders.updateOne({ _id: 1 }, { $set: { s: 'paid' } });
await orders.deleteOne({ _id: 2 });            // filtered out by the pipeline
const again = orders.watch(filter, { resumeAfter: token, fullDocument: 'updateLookup' });
for (let i = 0; i < 2; i++) { const e = await again.next();
  console.log('resumed', e.operationType, JSON.stringify(e.fullDocument)); }

Output

live    insert {"_id":1}
live    insert {"_id":2}
resumed insert {"_id":3,"s":"new"}
resumed update {"_id":1,"s":"paid"}

Both changes made while the stream was closed arrived on resume, in order, and the delete was dropped by the $match — the filter applies to resumed events too. Three rules follow. Persist the token after the side effect succeeds, in the same store if you can, or a crash will skip events. Treat it as opaque: a hand-made _data string is rejected with code 50854. And a token older than the oplog window is unrecoverable, so a consumer down longer than your retention must resynchronize from the collection; rs.printReplicationInfo() reports the hours covered.

Use startAfter instead when the last event was an invalidate (a drop or rename closes the stream), and startAtOperationTime when you have a timestamp but no token. To push changes to browsers, an Express 24,430 route holding a text/event-stream response open and writing one data: line per event is enough — but keep one stream per process and fan out, since each occupies a pooled connection.

MENU