fix(NODE-5052): prevent cursor and changestream close logic from running more than once

.mocharc.json

@@ -3,9 +3,13 @@
   "require": [
     "source-map-support/register",
     "ts-node/register",
-    "test/tools/runner/chai-addons.js"
+    "test/tools/runner/chai-addons.js",
+    "test/tools/runner/hooks/unhandled_checker.ts"
+  ],
+  "extension": [
+    "js",
+    "ts"
   ],
-  "extension": ["js", "ts"],
   "recursive": true,
   "timeout": 60000,
   "failZero": true,

src/cursor/abstract_cursor.ts

@@ -412,8 +412,8 @@ export abstract class AbstractCursor<
 
   async close(): Promise<void> {
     const needsToEmitClosed = !this[kClosed];
-    this[kClosed] = true;
     await cleanupCursorAsync(this, { needsToEmitClosed });
+    this[kClosed] = true;
   }
 
   /**
@@ -778,6 +778,20 @@ function cleanupCursor(
   const error = options?.error;
   const needsToEmitClosed = options?.needsToEmitClosed ?? cursor[kDocuments].length === 0;
 
+  if (cursor[kClosed]) {
+    if (needsToEmitClosed) {
+      cursor[kClosed] = true;
+      cursor[kId] = Long.ZERO;
+      cursor.emit(AbstractCursor.CLOSE);
+    }
+    cursor.removeAllListeners();
+    if (session.owner === cursor && !session.hasEnded) {
+      session.endSession({ error }).finally(callback);
+      return;
+    }
+    return process.nextTick(callback);
+  }
+
   if (error) {
     if (cursor.loadBalanced && error instanceof MongoNetworkError) {
       return completeCleanup();
@@ -790,6 +804,7 @@ function cleanupCursor(
       cursor[kId] = Long.ZERO;
       cursor.emit(AbstractCursor.CLOSE);
     }
+    cursor.removeAllListeners();
 
     if (session) {
       if (session.owner === cursor) {
@@ -812,6 +827,7 @@ function cleanupCursor(
       if (session.owner === cursor) {
         session.endSession({ error }).finally(() => {
           cursor.emit(AbstractCursor.CLOSE);
+          cursor.removeAllListeners();
           callback();
         });
         return;
@@ -823,6 +839,7 @@ function cleanupCursor(
     }
 
     cursor.emit(AbstractCursor.CLOSE);
+    cursor.removeAllListeners();
     return callback();
   }
 

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

@@ -2699,4 +2699,47 @@ describe('ChangeStream resumability', function () {
       expect(changeStream.cursor.maxWireVersion).equal(maxWireVersion);
     }
   );
+
+  it(
+    'should not have the bug from NODE-5052',
+    { requires: { topology: '!single' } },
+    async function () {
+      if (globalThis.AbortSignal?.timeout == null) {
+        this.skipReason = 'test requires AbortSignal.timeout';
+        this.skip();
+      }
+
+      const unhandledRejections: AsyncIterableIterator<[reason: Error, promise: Promise<any>]> = on(
+        process,
+        'unhandledRejection',
+        { signal: AbortSignal.timeout(2000) }
+      );
+
+      changeStream = collection.watch();
+
+      const shouldErrorLoop = (async function () {
+        try {
+          // eslint-disable-next-line @typescript-eslint/no-unused-vars
+          for await (const _change of changeStream) {
+            // ignore
+          }
+          return null;
+        } catch (error) {
+          return error;
+        }
+      })();
+
+      await sleep(200);
+      const closeResult = changeStream.close().catch(error => error);
+      expect(closeResult).to.not.be.instanceOf(Error);
+
+      const result = await shouldErrorLoop;
+      expect(result).to.be.instanceOf(MongoAPIError);
+      expect(result.message).to.match(/ChangeStream is closed/i);
+
+      const noUnhandledPromiseRejections = await unhandledRejections.next().catch(error => error);
+      expect(noUnhandledPromiseRejections).to.be.instanceOf(Error);
+      expect(noUnhandledPromiseRejections).to.have.nested.property('cause.name', 'TimeoutError');
+    }
+  );
 });

test/integration/crud/crud_api.test.ts

@@ -1,4 +1,5 @@
 import { expect } from 'chai';
+import { on } from 'events';
 
 import { MongoClient, MongoError, ObjectId, ReturnDocument } from '../../mongodb';
 import { assert as test } from '../shared';
@@ -60,130 +61,77 @@ describe('CRUD API', function () {
     await client.close();
   });
 
-  it('should correctly execute find method using crud api', function (done) {
-    const db = client.db();
-
-    db.collection('t').insertMany([{ a: 1 }, { a: 1 }, { a: 1 }, { a: 1 }], function (err) {
-      expect(err).to.not.exist;
-
-      //
-      // Cursor
-      // --------------------------------------------------
-      const makeCursor = () => {
-        // Possible methods on the the cursor instance
-        return db
-          .collection('t')
-          .find({})
-          .filter({ a: 1 })
-          .addCursorFlag('noCursorTimeout', true)
-          .addQueryModifier('$comment', 'some comment')
-          .batchSize(2)
-          .comment('some comment 2')
-          .limit(2)
-          .maxTimeMS(50)
-          .project({ a: 1 })
-          .skip(0)
-          .sort({ a: 1 });
-      };
-
-      //
-      // Exercise count method
-      // -------------------------------------------------
-      const countMethod = function () {
-        // Execute the different methods supported by the cursor
-        const cursor = makeCursor();
-        cursor.count(function (err, count) {
-          expect(err).to.not.exist;
-          test.equal(2, count);
-          eachMethod();
-        });
-      };
-
-      //
-      // Exercise legacy method each
-      // -------------------------------------------------
-      const eachMethod = function () {
-        let count = 0;
-
-        const cursor = makeCursor();
-        cursor.forEach(
-          () => {
-            count = count + 1;
-          },
-          err => {
-            expect(err).to.not.exist;
-            test.equal(2, count);
-            toArrayMethod();
-          }
-        );
-      };
+  context('should correctly execute find method using crud api', () => {
+    let db;
 
-      //
-      // Exercise toArray
-      // -------------------------------------------------
-      const toArrayMethod = function () {
-        const cursor = makeCursor();
-        cursor.toArray(function (err, docs) {
-          expect(err).to.not.exist;
-          test.equal(2, docs.length);
-          nextMethod();
-        });
-      };
+    beforeEach(async () => {
+      db = client.db();
+      await db.collection('t').deleteMany({});
+      await db.collection('t').insertMany([{ a: 1 }, { a: 1 }, { a: 1 }, { a: 1 }]);
+    });
 
-      //
-      // Exercise next method
-      // -------------------------------------------------
-      const nextMethod = function () {
-        const cursor = makeCursor();
-        cursor.next(function (err, doc) {
-          expect(err).to.not.exist;
-          test.ok(doc != null);
+    const makeCursor = () => {
+      // Possible methods on the the cursor instance
+      return db
+        .collection('t')
+        .find({})
+        .filter({ a: 1 })
+        .addCursorFlag('noCursorTimeout', true)
+        .addQueryModifier('$comment', 'some comment')
+        .batchSize(1)
+        .comment('some comment 2')
+        .limit(2)
+        .maxTimeMS(50)
+        .project({ a: 1 })
+        .skip(0)
+        .sort({ a: 1 });
+    };
+
+    it('count should be accurate', async () => {
+      const cursor = makeCursor();
+      const res = await cursor.count();
+      expect(res).to.equal(2);
+    });
 
-          cursor.next(function (err, doc) {
-            expect(err).to.not.exist;
-            test.ok(doc != null);
+    it('forEach should run 2 times', async () => {
+      const cursor = makeCursor();
+      let count = 0;
+      await cursor.forEach(() => {
+        count += 1;
+      });
+      expect(count).to.equal(2);
+    });
 
-            cursor.next(function (err, doc) {
-              expect(err).to.not.exist;
-              expect(doc).to.not.exist;
-              streamMethod();
-            });
-          });
-        });
-      };
+    it('toArray should return an array with two documents', async () => {
+      const cursor = makeCursor();
+      const res = await cursor.toArray();
+      expect(res).to.have.lengthOf(2);
+    });
 
-      //
-      // Exercise stream
-      // -------------------------------------------------
-      const streamMethod = function () {
-        let count = 0;
-        const cursor = makeCursor();
-        const stream = cursor.stream();
-        stream.on('data', function () {
-          count = count + 1;
-        });
+    it('next be callable three times without blocking', async () => {
+      const cursor = makeCursor();
+      const doc0 = await cursor.next();
+      expect(doc0).to.exist;
+      const doc1 = await cursor.next();
+      expect(doc1).to.exist;
+      const doc2 = await cursor.next();
+      expect(doc2).to.not.exist;
+    });
 
+    it('stream() should create a node stream that emits 2 data events', async () => {
+      const count = 0;
+      const cursor = makeCursor();
+      const stream = cursor.stream();
+      on(stream, 'data'),
         cursor.once('close', function () {
-          test.equal(2, count);
-          explainMethod();
+          expect(count).to.equal(2);
         });
-      };
-
-      //
-      // Explain method
-      // -------------------------------------------------
-      const explainMethod = function () {
-        const cursor = makeCursor();
-        cursor.explain(function (err, result) {
-          expect(err).to.not.exist;
-          test.ok(result != null);
-
-          client.close(done);
-        });
-      };
+    });
 
-      // Execute all the methods
-      countMethod();
+    it('explain works', async () => {
+      const cursor = makeCursor();
+      const result = await cursor.explain();
+      expect(result).to.exist;
     });
   });
 

test/integration/crud/find_cursor_methods.test.js

@@ -109,20 +109,27 @@ describe('Find Cursor', function () {
   });
 
   context('#close', function () {
-    it('should send a killCursors command when closed before completely iterated', function (done) {
+    beforeEach(async function () {
+      await client
+        .db()
+        .collection('abstract_cursor')
+        .insertMany([{ a: 1 }, { a: 2 }, { a: 3 }, { a: 4 }]);
+    });
+    afterEach(async function () {
+      await client.db().collection('abstract_cursor').deleteMany();
+    });
+
+    it('should send a killCursors command when closed before completely iterated', async function () {
       const commands = [];
       client.on('commandStarted', filterForCommands(['killCursors'], commands));
 
       const coll = client.db().collection('abstract_cursor');
       const cursor = coll.find({}, { batchSize: 2 });
-      cursor.next(err => {
-        expect(err).to.not.exist;
-        cursor.close(err => {
-          expect(err).to.not.exist;
-          expect(commands).to.have.length(1);
-          done();
-        });
-      });
+
+      const doc = await cursor.next();
+      expect(doc).property('a', 1);
+      await cursor.close();
+      expect(commands).to.have.length(1);
     });
 
     it('should not send a killCursors command when closed after completely iterated', function (done) {