Description

Change Streams include a feature that allows the stream to resume after certain failures (for example, when a host is unreachable or a cursor is not found). The full list of resumable errors is defined in the spec).

This feature works by having ChangeStream cache the resume token returned by the server on successful operations, and then pass its value in either the startAfter or resumeAfter field when attempting to resume. The server then knows from which point in the oplog to return changes to the client.

The caching mechanism is handled internally in both the event-listener form and the iterator form, and it updates the stream’s resumeToken property. For example:

const changeStream = collection.watch([]);
while (true) {
 await changeStream.next();
 console.log(changeStream.resumeToken); // resumeToken updates on each change
}
// this can be rewritten into for..of form which uses .next() internally
// for await (const _ of changeStream) {

However, not all exposed methods cache the token. In particular, .tryNext() ignores it, so the following code logs the same resumeToken for each change:

const changeStream = collection.watch([]);
while (true) {
 const change = await changeStream.tryNext();
 if (change) {
 console.log(changeStream.resumeToken); // resumeToken does not update
 }
 await scheduler.wait(1000); // add delay since tryNext() does not wait for changes
}

If the stream encounters a resumable error in the snippets above, it will attempt to resume. Without the latest resume token (.tryNext() case), the server might return changes that were already consumed, resulting in duplicates on the application side. This issue does not occur with .next() or with the 'change' event listener, both of which update the resumeToken correctly.

The fix

Handle changes returned by .tryNext() the same way as in .next() by using the private ._processChange(), except when the change is null. Because .tryNext() does not wait for server changes, a null indicates "no changes," whereas .next() would treat null as a signal to close the stream.

Testing

Resume functionality is covered by a combination of unit and integration tests that verify token caching and the use of startAfter/resumeAfter during resume.

Additionally, a test case was added for this specific user scenario:

• Create a change stream
• Get the first change
• Attempt to get the second change; fail on getMore
• Resume with the token
• Get the second change (no duplicates)

Release Highlight

ChangeStream .tryNext() now updates resumeToken to prevent duplicates after resume

When .tryNext() returns a change document, the driver now caches its resumeToken, aligning its behavior with .next() and the 'change' event. If .tryNext() returns null (no new changes), nothing is cached, which is unchanged from previous behavior.

Previously, .tryNext() did not update the resumeToken, so a resumable error could cause a resume from an older token and re-deliver already processed changes. With this release, resumes continue from the latest token observed via .tryNext(), preventing duplicates.

const changeStream = collection.watch([]);
while (true) {
 const change = await changeStream.tryNext(); // prior versions could return duplicates
 await scheduler.wait(1000); // delay since tryNext() does not wait for changes
}

Applications that poll change streams with .tryNext() in non-blocking loops benefit directly. There are no API changes; if you previously tracked and passed resumeAfter or startAfter manually, you can now rely on the driver’s built-in token caching.

Huge thanks to @rkistner for bringing this bug to our attention and for sharing code to reproduce it. Huge thanks as well to @Omnicpie for investigating and implementing a fix.

Double check the following

• Ran npm run check:lint script
• Self-review completed using the steps outlined here
• PR title follows the correct format: type(NODE-xxxx)[!]: description
• Changes are covered by tests
• New TODOs have a related JIRA ticket