fix: track position, helper, tailable interrupt

Author: nbbeeken

src/cursor/abstract_cursor.ts

@@ -264,10 +264,6 @@ export abstract class AbstractCursor<
       throw new MongoRuntimeError('Cursor must be constructed with MongoClient');
     }
     this.cursorClient = client;
-
-    this.cursorClient.s.activeCursors.add(this);
-    this.once('close', removeActiveCursor);
-
     this.cursorNamespace = namespace;
     this.cursorId = null;
     this.initialized = false;
@@ -360,6 +356,7 @@ export abstract class AbstractCursor<
     };
 
     this.timeoutContext = options.timeoutContext;
+    this.trackCursor();
   }
 
   /**
@@ -439,6 +436,14 @@ export abstract class AbstractCursor<
     await this.close();
   }
 
+  /** Adds cursor to client's tracking so it will be closed by MongoClient.close() */
+  private trackCursor() {
+    this.cursorClient.s.activeCursors.add(this);
+    if (!this.listeners('close').includes(removeActiveCursor)) {
+      this.once('close', removeActiveCursor);
+    }
+  }
+
   /** Returns current buffered documents length */
   bufferedCount(): number {
     return this.documents?.length ?? 0;
@@ -832,21 +837,14 @@ export abstract class AbstractCursor<
     this.isClosed = false;
     this.isKilled = false;
     this.initialized = false;
+    this.trackCursor();
 
-    this.cursorClient.s.activeCursors.add(this);
-    if (!this.listeners('close').includes(removeActiveCursor)) {
-      this.once('close', removeActiveCursor);
-    }
-
-    const session = this.cursorSession;
-    if (session) {
-      // We only want to end this session if we created it, and it hasn't ended yet
-      if (session.explicit === false) {
-        if (!session.hasEnded) {
-          session.endSession().then(undefined, squashError);
-        }
-        this.cursorSession = this.cursorClient.startSession({ owner: this, explicit: false });
+    // We only want to end this session if we created it, and it hasn't ended yet
+    if (this.cursorSession.explicit === false) {
+      if (!this.cursorSession.hasEnded) {
+        this.cursorSession.endSession().then(undefined, squashError);
       }
+      this.cursorSession = this.cursorClient.startSession({ owner: this, explicit: false });
     }
   }
 
@@ -982,7 +980,6 @@ export abstract class AbstractCursor<
   /** @internal */
   private async cleanup(timeoutMS?: number, error?: Error) {
     this.isClosed = true;
-    const session = this.cursorSession;
     const timeoutContextForKillCursors = (): CursorTimeoutContext | undefined => {
       if (timeoutMS != null) {
         this.timeoutContext?.clear();
@@ -1004,7 +1001,7 @@ export abstract class AbstractCursor<
         !this.cursorId.isZero() &&
         this.cursorNamespace &&
         this.selectedServer &&
-        !session.hasEnded
+        !this.cursorSession.hasEnded
       ) {
         this.isKilled = true;
         const cursorId = this.cursorId;
@@ -1013,7 +1010,7 @@ export abstract class AbstractCursor<
         await executeOperation(
           this.cursorClient,
           new KillCursorsOperation(cursorId, this.cursorNamespace, this.selectedServer, {
-            session
+            session: this.cursorSession
           }),
           timeoutContextForKillCursors()
         );
@@ -1022,11 +1019,11 @@ export abstract class AbstractCursor<
       squashError(error);
     } finally {
       try {
-        if (session?.owner === this) {
-          await session.endSession({ error });
+        if (this.cursorSession?.owner === this) {
+          await this.cursorSession.endSession({ error });
         }
-        if (!session?.inTransaction()) {
-          maybeClearPinnedConnection(session, { error });
+        if (!this.cursorSession?.inTransaction()) {
+          maybeClearPinnedConnection(this.cursorSession, { error });
         }
       } finally {
         this.emitClose();

test/integration/crud/misc_cursors.test.js

@@ -10,7 +10,7 @@ const sinon = require('sinon');
 const { Writable } = require('stream');
 const { once, on } = require('events');
 const { setTimeout } = require('timers');
-const { ReadPreference, MongoExpiredSessionError } = require('../../mongodb');
+const { ReadPreference } = require('../../mongodb');
 const { ServerType } = require('../../mongodb');
 const { formatSort } = require('../../mongodb');
 
@@ -1872,7 +1872,7 @@ describe('Cursor', function () {
     expect(cursor).to.have.property('closed', true);
 
     const error = await rejectedEarlyBecauseClientClosed;
-    expect(error).to.be.instanceOf(MongoExpiredSessionError);
+    expect(error).to.be.null; // TODO: is this API or just "how it behaved at the time"
   });
 
   it('shouldAwaitData', {