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);
@@ -1835,6 +1829,78 @@ describe.only('Change Streams', function () {
  });
  });
  });
+
+ describe("NODE-4763 - doesn't produce duplicates after resume", function () {
+ const resumableError = { code: 6, message: 'host unreachable' };
+
+ beforeEach(async function () {
+ 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);
+ });
+
+ afterEach(async function () {
+ await client.db('admin').command({
+ configureFailPoint: is4_2Server(this.configuration.version)
+ ? 'failCommand'
+ : 'failGetMoreAfterCursorCheckout',
+ mode: 'off'
+ } as FailPoint);
+ });
+
+ 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);
+ });
+ });
 });
 
 describe('ChangeStream resumability', function () {

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;