fix(NODE-4763): cache resume token in ChangeStream#tryNext() method

Author: tadjik1

src/change_stream.ts

@@ -809,7 +809,11 @@ export class ChangeStream<
       while (true) {
         try {
           const change = await this.cursor.tryNext();
-          return change ?? null;
+          if (!change) {
+            return null;
+          }
+          const processedChange = this._processChange(change);
+          return processedChange;
         } catch (error) {
           try {
             await this._processErrorIteratorMode(error, this.cursor.id != null);

test/integration/change-streams/change_stream.test.ts

@@ -45,7 +45,7 @@ const pipeline = [
   { $addFields: { comment: 'The documentKey field has been projected out of this document.' } }
 ];
 
-describe.only('Change Streams', function () {
+describe('Change Streams', function () {
   let client: MongoClient;
   let collection: Collection;
   let changeStream: ChangeStream;
@@ -377,10 +377,7 @@ describe.only('Change Streams', function () {
 
         async test() {
           await initIteratorMode(changeStream);
-          collection.insertOne({ a: 1 });
-
-          const hasNext = await changeStream.hasNext();
-          expect(hasNext).to.be.true;
+          await collection.insertOne({ a: 1 });
 
           const change = await changeStream.next();
           expect(change).to.have.property('_id').that.deep.equals(changeStream.resumeToken);
@@ -392,10 +389,7 @@ describe.only('Change Streams', function () {
 
         async test() {
           await initIteratorMode(changeStream);
-          collection.insertOne({ a: 1 });
-
-          const hasNext = await changeStream.hasNext();
-          expect(hasNext).to.be.true;
+          await collection.insertOne({ a: 1 });
 
           const change = await changeStream.tryNext();
           expect(change).to.have.property('_id').that.deep.equals(changeStream.resumeToken);
@@ -408,7 +402,7 @@ describe.only('Change Streams', function () {
       async test() {
         const willBeChange = once(changeStream, 'change');
         await once(changeStream.cursor, 'init');
-        collection.insertOne({ a: 1 });
+        await collection.insertOne({ a: 1 });
 
         const [change] = await willBeChange;
         expect(change).to.have.property('_id').that.deep.equals(changeStream.resumeToken);
@@ -2753,3 +2747,90 @@ describe('ChangeStream resumability', function () {
     }
   );
 });
+
+describe('NODE-4763 - handle resume without duplicates', function () {
+  let client: MongoClient;
+  let db: Db;
+  let collection: Collection;
+  let changeStream: ChangeStream;
+  const resumableError = { code: 6, message: 'host unreachable' };
+
+  beforeEach(async function () {
+    client = this.configuration.newClient();
+    await client.connect();
+
+    await client.db('admin').command({
+      configureFailPoint: is4_2Server(this.configuration.version)
+        ? 'failCommand'
+        : 'failGetMoreAfterCursorCheckout',
+      mode: { skip: 1 },
+      data: {
+        failCommands: ['getMore'],
+        errorCode: resumableError.code,
+        errmsg: resumableError.message
+      }
+    } as FailPoint);
+
+    db = client.db('NODE-4763');
+    collection = await db.createCollection('test');
+    changeStream = collection.watch();
+  });
+
+  afterEach(async function () {
+    await client.db('admin').command({
+      configureFailPoint: is4_2Server(this.configuration.version)
+        ? 'failCommand'
+        : 'failGetMoreAfterCursorCheckout',
+      mode: 'off'
+    } as FailPoint);
+
+    await changeStream.close();
+    await db.dropCollection('test');
+    await client?.close();
+  });
+
+  describe('when using iterator form', function () {
+    it('#next', { requires: { topology: 'replicaset' } }, async function test() {
+      await initIteratorMode(changeStream);
+
+      await collection.insertOne({ a: 1 });
+      const change = await changeStream.next();
+      expect(change).to.have.property('operationType', 'insert');
+      expect(change).to.have.nested.property('fullDocument.a', 1);
+
+      await collection.insertOne({ a: 2 });
+      const change2 = await changeStream.next();
+      expect(change2).to.have.property('operationType', 'insert');
+      expect(change2).to.have.nested.property('fullDocument.a', 2);
+    });
+
+    it('#tryNext', { requires: { topology: 'replicaset' } }, async function test() {
+      await initIteratorMode(changeStream);
+
+      await collection.insertOne({ a: 1 });
+      const change = await changeStream.tryNext();
+      expect(change).to.have.property('operationType', 'insert');
+      expect(change).to.have.nested.property('fullDocument.a', 1);
+
+      await collection.insertOne({ a: 2 });
+      const change2 = await changeStream.tryNext();
+      expect(change2).to.have.property('operationType', 'insert');
+      expect(change2).to.have.nested.property('fullDocument.a', 2);
+    });
+  });
+
+  it('in an event listener form', { requires: { topology: 'replicaset' } }, async function () {
+    const willBeChange = on(changeStream, 'change');
+    await once(changeStream.cursor, 'init');
+
+    await collection.insertOne({ a: 1 });
+    const change = await willBeChange.next();
+    expect(change.value[0]).to.have.property('operationType', 'insert');
+    expect(change.value[0]).to.have.nested.property('fullDocument.a', 1);
+
+    await collection.insertOne({ a: 2 });
+    const change2 = await willBeChange.next();
+    expect(change2.value[0]).to.have.property('operationType', 'insert');
+    expect(change2.value[0]).to.have.nested.property('fullDocument.a', 2);
+  });
+});

test/tools/utils.ts

@@ -207,7 +207,12 @@ export function extractAuthFromConnectionString(connectionString: string | any[]
 
 export interface FailPoint {
   configureFailPoint: 'failCommand' | 'failGetMoreAfterCursorCheckout' | 'maxTimeNeverTimeOut';
-  mode: { activationProbability: number } | { times: number } | 'alwaysOn' | 'off';
+  mode:
+    | { activationProbability: number }
+    | { times: number }
+    | { skip: number }
+    | 'alwaysOn'
+    | 'off';
   data: {
     failCommands: string[];
     errorCode?: number;