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

@@ -828,13 +828,16 @@ function cleanupCursor(
 
   cursor[kKilled] = true;
 
+  if (session.hasEnded) {
+    return completeCleanup();
+  }
+
   executeOperation(
     cursor[kClient],
     new KillCursorsOperation(cursorId, cursorNs, server, { session })
-  ).finally(() => {
-    completeCleanup();
-  });
-  return;
+  )
+    .catch(() => null)
+    .finally(completeCleanup);
 }
 
 /** @internal */

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

@@ -1019,6 +1019,30 @@ describe('Change Streams', function () {
             }
           }
         );
+
+        it(
+          'when closed throws "ChangeStream is closed"',
+          { requires: { topology: '!single' } },
+          async function () {
+            changeStream = collection.watch();
+
+            const loop = (async function () {
+              // eslint-disable-next-line @typescript-eslint/no-unused-vars
+              for await (const _change of changeStream) {
+                return 'loop entered'; // loop should never be entered
+              }
+              return 'loop ended without error'; // loop should not finish without error
+            })();
+
+            await sleep(1);
+            const closeResult = changeStream.close().catch(error => error);
+            expect(closeResult).to.not.be.instanceOf(Error);
+
+            const result = await loop.catch(error => error);
+            expect(result).to.be.instanceOf(MongoAPIError);
+            expect(result.message).to.match(/ChangeStream is closed/i);
+          }
+        );
       });
 
       describe('#return', function () {

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,92 @@ 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 collection;
 
-      //
-      // 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();
-        });
-      };
+    beforeEach(async () => {
+      collection = client.db().collection('t');
+      await collection.drop().catch(() => null);
+      await collection.insertMany([{ a: 1 }, { a: 1 }, { a: 1 }, { a: 1 }]);
+    });
 
-      //
-      // Exercise legacy method each
-      // -------------------------------------------------
-      const eachMethod = function () {
-        let count = 0;
+    afterEach(async () => {
+      await collection?.drop().catch(() => null);
+    });
 
+    const makeCursor = () => {
+      // Possible methods on the the cursor instance
+      return collection
+        .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();
+          expect(count).to.equal(2);
         });
-      };
+      });
+    });
 
-      //
-      // Explain method
-      // -------------------------------------------------
-      const explainMethod = function () {
+    describe('#explain()', () => {
+      it('returns an explain document', async () => {
         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();
+        const result = await cursor.explain();
+        expect(result).to.exist;
+      });
     });
   });
 

test/integration/crud/find_cursor_methods.test.js

@@ -108,50 +108,58 @@ describe('Find Cursor', function () {
     });
   });
 
-  context('#close', function () {
-    it('should send a killCursors command when closed before completely iterated', function (done) {
-      const commands = [];
-      client.on('commandStarted', filterForCommands(['killCursors'], commands));
+  describe('#close', function () {
+    let collection;
 
-      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();
-        });
-      });
+    beforeEach(async function () {
+      collection = client.db().collection('abstract_cursor');
+      await collection.drop().catch(() => null);
+      await collection.insertMany([{ a: 1 }, { a: 2 }, { a: 3 }, { a: 4 }]);
     });
 
-    it('should not send a killCursors command when closed after completely iterated', function (done) {
-      const commands = [];
-      client.on('commandStarted', filterForCommands(['killCursors'], commands));
+    afterEach(async function () {
+      await collection?.drop().catch(() => null);
+    });
 
-      const coll = client.db().collection('abstract_cursor');
-      const cursor = coll.find({}, { batchSize: 2 });
-      cursor.toArray(err => {
-        expect(err).to.not.exist;
+    context('when closed before completely iterated', () => {
+      it('sends a killCursors command', async () => {
+        const killCursorsCommands = [];
+        client.on('commandStarted', filterForCommands(['killCursors'], killCursorsCommands));
 
-        cursor.close(err => {
-          expect(err).to.not.exist;
-          expect(commands).to.have.length(0);
-          done();
-        });
+        const cursor = collection.find({}, { batchSize: 2 });
+
+        const doc = await cursor.next();
+        expect(doc).property('a', 1);
+
+        expect(killCursorsCommands).to.have.length(0);
+        await cursor.close();
+        expect(killCursorsCommands).to.have.length(1);
       });
     });
 
-    it('should not send a killCursors command when closed before initialization', function (done) {
-      const commands = [];
-      client.on('commandStarted', filterForCommands(['killCursors'], commands));
+    context('when closed after completely iterated', () => {
+      it('does not send a killCursors command', async () => {
+        const killCursorsCommands = [];
+        client.on('commandStarted', filterForCommands(['killCursors'], killCursorsCommands));
 
-      const coll = client.db().collection('abstract_cursor');
-      const cursor = coll.find({}, { batchSize: 2 });
-      cursor.close(err => {
-        expect(err).to.not.exist;
-        expect(commands).to.have.length(0);
-        done();
+        const cursor = collection.find();
+        await cursor.toArray();
+        expect(killCursorsCommands).to.have.length(0);
+        await cursor.close();
+        expect(killCursorsCommands).to.have.length(0);
+      });
+    });
+
+    context('when closed before initialization', () => {
+      it('does not send a killCursors command', async () => {
+        const killCursorsCommands = [];
+        client.on('commandStarted', filterForCommands(['killCursors'], killCursorsCommands));
+
+        const cursor = collection.find();
+
+        expect(killCursorsCommands).to.have.length(0);
+        await cursor.close();
+        expect(killCursorsCommands).to.have.length(0);
       });
     });
   });