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,62 +778,51 @@ function cleanupCursor(
   const error = options?.error;
   const needsToEmitClosed = options?.needsToEmitClosed ?? cursor[kDocuments].length === 0;
 
-  if (error) {
-    if (cursor.loadBalanced && error instanceof MongoNetworkError) {
-      return completeCleanup();
-    }
-  }
-
-  if (cursorId == null || server == null || cursorId.isZero() || cursorNs == null) {
+  const cleanupListeners = () => {
     if (needsToEmitClosed) {
       cursor[kClosed] = true;
       cursor[kId] = Long.ZERO;
       cursor.emit(AbstractCursor.CLOSE);
     }
+    cursor.removeAllListeners();
+  };
 
-    if (session) {
-      if (session.owner === cursor) {
-        session.endSession({ error }).finally(() => {
-          callback();
-        });
-        return;
-      }
-
-      if (!session.inTransaction()) {
-        maybeClearPinnedConnection(session, { error });
-      }
+  const cleanupSession = () => {
+    if (session?.owner === cursor && !session?.hasEnded) {
+      session.endSession({ error }).finally(callback);
+      return;
+    }
+    if (!session?.inTransaction()) {
+      maybeClearPinnedConnection(session, { error });
     }
+    process.nextTick(callback);
+  };
 
-    return callback();
-  }
+  const completeCleanup = () => {
+    cleanupListeners();
+    return cleanupSession();
+  };
 
-  function completeCleanup() {
-    if (session) {
-      if (session.owner === cursor) {
-        session.endSession({ error }).finally(() => {
-          cursor.emit(AbstractCursor.CLOSE);
-          callback();
-        });
-        return;
-      }
+  if (cursor[kClosed]) {
+    return completeCleanup();
+  }
 
-      if (!session.inTransaction()) {
-        maybeClearPinnedConnection(session, { error });
-      }
+  if (error) {
+    if (cursor.loadBalanced && error instanceof MongoNetworkError) {
+      return completeCleanup();
     }
+  }
 
-    cursor.emit(AbstractCursor.CLOSE);
-    return callback();
+  if (cursorId == null || server == null || cursorId.isZero() || cursorNs == null) {
+    return completeCleanup();
   }
 
   cursor[kKilled] = true;
 
   executeOperation(
     cursor[kClient],
     new KillCursorsOperation(cursorId, cursorNs, server, { session })
-  ).finally(() => {
-    completeCleanup();
-  });
+  ).finally(completeCleanup);
   return;
 }
 

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

@@ -2699,4 +2699,47 @@ describe('ChangeStream resumability', function () {
       expect(changeStream.cursor.maxWireVersion).equal(maxWireVersion);
     }
   );
+
+  it(
+    'does not raise unhandled rejection errors',
+    { 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,89 @@ 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 });
-      };
+  context('when creating a cursor with find', () => {
+    let db;
 
-      //
-      // 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;
+    beforeEach(async () => {
+      db = client.db();
+      await db.collection('t').deleteMany({});
+      await db.collection('t').insertMany([{ a: 1 }, { a: 1 }, { a: 1 }, { a: 1 }]);
+    });
 
+    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 });
+    };
+
+    describe('#count()', () => {
+      it('returns the number of documents', async () => {
         const cursor = makeCursor();
-        cursor.forEach(
-          () => {
-            count = count + 1;
-          },
-          err => {
-            expect(err).to.not.exist;
-            test.equal(2, count);
-            toArrayMethod();
-          }
-        );
-      };
+        const res = await cursor.count();
+        expect(res).to.equal(2);
+      });
+    });
 
-      //
-      // Exercise toArray
-      // -------------------------------------------------
-      const toArrayMethod = function () {
+    describe('#forEach()', () => {
+      it('iterates all the documents', async () => {
         const cursor = makeCursor();
-        cursor.toArray(function (err, docs) {
-          expect(err).to.not.exist;
-          test.equal(2, docs.length);
-          nextMethod();
+        let count = 0;
+        await cursor.forEach(() => {
+          count += 1;
         });
-      };
+        expect(count).to.equal(2);
+      });
+    });
 
-      //
-      // Exercise next method
-      // -------------------------------------------------
-      const nextMethod = function () {
+    describe('#toArray()', () => {
+      it('returns an array with all documents', async () => {
         const cursor = makeCursor();
-        cursor.next(function (err, doc) {
-          expect(err).to.not.exist;
-          test.ok(doc != null);
-
-          cursor.next(function (err, doc) {
-            expect(err).to.not.exist;
-            test.ok(doc != null);
+        const res = await cursor.toArray();
+        expect(res).to.have.lengthOf(2);
+      });
+    });
 
-            cursor.next(function (err, doc) {
-              expect(err).to.not.exist;
-              expect(doc).to.not.exist;
-              streamMethod();
-            });
-          });
-        });
-      };
+    describe('#next()', () => {
+      it('is callable 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;
+      });
+    });
 
-      //
-      // Exercise stream
-      // -------------------------------------------------
-      const streamMethod = function () {
-        let count = 0;
+    describe('#stream()', () => {
+      it('creates a node stream that emits data events', async () => {
+        const count = 0;
         const cursor = makeCursor();
         const stream = cursor.stream();
-        stream.on('data', function () {
-          count = count + 1;
-        });
-
+        on(stream, 'data');
         cursor.once('close', function () {
-          test.equal(2, count);
-          explainMethod();
-        });
-      };
-
-      //
-      // 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);
+          expect(count).to.equal(2);
         });
-      };
+      });
+    });
+  });
 
-      // Execute all the methods
-      countMethod();
+  describe('#explain()', () => {
+    it('returns an explain document', async () => {
+      const cursor = makeCursor();
+      const result = await cursor.explain();
+      expect(result).to.exist;
     });
   });