From 51f8c398a20d7d6d1da2cdea9bb4e47af96476fd Mon Sep 17 00:00:00 2001
From: Durran Jordan <durran@gmail.com>
Date: Wed, 7 Aug 2024 17:18:07 +0200
Subject: [PATCH] fix(NODE-6255): cursor throws exhausted errors after explicit
 close

---
 src/cursor/abstract_cursor.ts                 | 15 ++++++++++++---
 .../change_streams.prose.test.ts              |  6 +++++-
 .../node-specific/abstract_cursor.test.ts     | 19 +++++++++++++++++--
 3 files changed, 34 insertions(+), 6 deletions(-)

diff --git a/src/cursor/abstract_cursor.ts b/src/cursor/abstract_cursor.ts
index da08f1a1a66..dbbb5ec47ff 100644
--- a/src/cursor/abstract_cursor.ts
+++ b/src/cursor/abstract_cursor.ts
@@ -146,6 +146,8 @@ export abstract class AbstractCursor<
   /** @internal */
   private isClosed: boolean;
   /** @internal */
+  private isForceClosed: boolean;
+  /** @internal */
   private isKilled: boolean;
   /** @internal */
   protected readonly cursorOptions: InternalAbstractCursorOptions;
@@ -169,6 +171,7 @@ export abstract class AbstractCursor<
     this.cursorId = null;
     this.initialized = false;
     this.isClosed = false;
+    this.isForceClosed = false;
     this.isKilled = false;
     this.cursorOptions = {
       readPreference:
@@ -369,7 +372,7 @@ export abstract class AbstractCursor<
   }
 
   async hasNext(): Promise<boolean> {
-    if (this.cursorId === Long.ZERO) {
+    if (this.cursorId === Long.ZERO || this.isForceClosed) {
       return false;
     }
 
@@ -385,7 +388,7 @@ export abstract class AbstractCursor<
 
   /** Get the next available document from the cursor, returns null if no more documents are available. */
   async next(): Promise<TSchema | null> {
-    if (this.cursorId === Long.ZERO) {
+    if (this.cursorId === Long.ZERO || this.isForceClosed) {
       throw new MongoCursorExhaustedError();
     }
 
@@ -405,7 +408,7 @@ export abstract class AbstractCursor<
    * Try to get the next available document from the cursor or `null` if an empty batch is returned
    */
   async tryNext(): Promise<TSchema | null> {
-    if (this.cursorId === Long.ZERO) {
+    if (this.cursorId === Long.ZERO || this.isForceClosed) {
       throw new MongoCursorExhaustedError();
     }
 
@@ -447,6 +450,12 @@ export abstract class AbstractCursor<
   }
 
   async close(): Promise<void> {
+    // We flag that an explicit call to close the cursor has happened, so no matter
+    // what the current state is it can no longer be used. This is do to areas in the
+    // cursor that are setting the isClosed flag or cursor id to zero without going
+    // through this path.
+    this.isForceClosed = true;
+    this.documents?.clear();
     await this.cleanup();
   }
 
diff --git a/test/integration/change-streams/change_streams.prose.test.ts b/test/integration/change-streams/change_streams.prose.test.ts
index 55c11591c51..8a4f5a37af2 100644
--- a/test/integration/change-streams/change_streams.prose.test.ts
+++ b/test/integration/change-streams/change_streams.prose.test.ts
@@ -869,6 +869,7 @@ describe('Change Stream prose tests', function () {
         changeStream.on('change', change => {
           if (change.operationType === 'invalidate') {
             startAfter = change._id;
+            console.log('closing');
             changeStream.close(done);
           }
         });
@@ -890,7 +891,10 @@ describe('Change Stream prose tests', function () {
         const events = [];
         client.on('commandStarted', e => recordEvent(events, e));
         const changeStream = coll.watch([], { startAfter });
-        this.defer(() => changeStream.close());
+        this.defer(() => {
+          console.log('defer');
+          changeStream.close();
+        });
 
         changeStream.once('change', change => {
           expect(change).to.containSubset({
diff --git a/test/integration/node-specific/abstract_cursor.test.ts b/test/integration/node-specific/abstract_cursor.test.ts
index bee2333db94..6890a94fbeb 100644
--- a/test/integration/node-specific/abstract_cursor.test.ts
+++ b/test/integration/node-specific/abstract_cursor.test.ts
@@ -348,12 +348,27 @@ describe('class AbstractCursor', function () {
       });
     });
 
+    context('when the cursor is closed', function () {
+      context('when calling next()', function () {
+        it('raises a cursor exhausted error', async function () {
+          cursor = client.db().collection('test').find({});
+          await cursor.next();
+          await cursor.close();
+          const error = await cursor.next().catch(error => error);
+          expect(error).to.be.instanceOf(MongoCursorExhaustedError);
+          expect(cursor.id.isZero()).to.be.true;
+          expect(cursor).to.have.property('closed', true);
+          expect(cursor).to.have.property('killed', false);
+        });
+      });
+    });
+
     describe('when some documents have been iterated and the cursor is closed', () => {
-      it('has a zero id and is not closed and is killed', async function () {
+      it('has a zero id and is closed and is killed', async function () {
         cursor = client.db().collection('test').find({}, { batchSize: 2 });
         await cursor.next();
         await cursor.close();
-        expect(cursor).to.have.property('closed', false);
+        expect(cursor).to.have.property('closed', true);
         expect(cursor).to.have.property('killed', true);
         expect(cursor.id.isZero()).to.be.true;
         const error = await cursor.next().catch(error => error);