fix: ensure implicit sessions are ended consistently

This is a fix and a refator, which ensures that implicit sessions
are ended as soon as possible when cursor commands are executed.
The current code path calls `_endSessions` in a confusing number of
locations, and in some cases can accidentally call the method twice
in a row. Reducing the number of calls to the method, and ensuring
its called only after server command responses and the `close`
method should prevent these errors

NODE-2630

Author: mbroadst

lib/core/cursor.js

@@ -411,7 +411,15 @@ class CoreCursor extends Readable {
       batchSize = this.cursorState.limit - this.cursorState.currentLimit;
     }
 
-    this.server.getMore(this.ns, this.cursorState, batchSize, this.options, callback);
+    const cursorState = this.cursorState;
+    this.server.getMore(this.ns, cursorState, batchSize, this.options, (err, result, conn) => {
+      // NOTE: `getMore` modifies `cursorState`, would be very ideal not to do so in the future
+      if (err || (cursorState.cursorId && cursorState.cursorId.isZero())) {
+        this._endSession();
+      }
+
+      callback(err, result, conn);
+    });
   }
 
   _initializeCursor(callback) {
@@ -432,18 +440,15 @@ class CoreCursor extends Readable {
     }
 
     function done(err, result) {
-      if (
-        cursor.cursorState.cursorId &&
-        cursor.cursorState.cursorId.isZero() &&
-        cursor._endSession
-      ) {
+      const cursorState = cursor.cursorState;
+      if (err || (cursorState.cursorId && cursorState.cursorId.isZero())) {
         cursor._endSession();
       }
 
       if (
-        cursor.cursorState.documents.length === 0 &&
-        cursor.cursorState.cursorId &&
-        cursor.cursorState.cursorId.isZero() &&
+        cursorState.documents.length === 0 &&
+        cursorState.cursorId &&
+        cursorState.cursorId.isZero() &&
         !cursor.cmd.tailable &&
         !cursor.cmd.awaitData
       ) {
@@ -689,8 +694,8 @@ function _setCursorNotifiedImpl(self, callback) {
   self.cursorState.documents = [];
   self.cursorState.cursorIndex = 0;
 
-  if (self._endSession) {
-    self._endSession(undefined, () => callback());
+  if (self.cursorState.session) {
+    self._endSession(callback);
     return;
   }
 
@@ -776,10 +781,6 @@ function nextFunction(self, callback) {
         return handleCallback(callback, err);
       }
 
-      if (self.cursorState.cursorId && self.cursorState.cursorId.isZero() && self._endSession) {
-        self._endSession();
-      }
-
       // Save the returned connection to ensure all getMore's fire over the same connection
       self.connection = connection;
 

lib/cursor.js

@@ -832,9 +832,7 @@ class Cursor extends CoreCursor {
       const fetchDocs = () => {
         cursor._next((err, doc) => {
           if (err) {
-            return cursor._endSession
-              ? cursor._endSession(() => handleCallback(cb, err))
-              : handleCallback(cb, err);
+            return handleCallback(cb, err);
           }
 
           if (doc == null) {
@@ -914,38 +912,18 @@ class Cursor extends CoreCursor {
     if (typeof options === 'function') (callback = options), (options = {});
     options = Object.assign({}, { skipKillCursors: false }, options);
 
-    this.s.state = CursorState.CLOSED;
-    if (!options.skipKillCursors) {
-      // Kill the cursor
-      this.kill();
-    }
-
-    const completeClose = () => {
-      // Emit the close event for the cursor
-      this.emit('close');
-
-      // Callback if provided
-      if (typeof callback === 'function') {
-        return handleCallback(callback, null, this);
-      }
-
-      // Return a Promise
-      return new this.s.promiseLibrary(resolve => {
-        resolve();
-      });
-    };
-
-    if (this.cursorState.session) {
-      if (typeof callback === 'function') {
-        return this._endSession(() => completeClose());
+    return maybePromise(this, callback, cb => {
+      this.s.state = CursorState.CLOSED;
+      if (!options.skipKillCursors) {
+        // Kill the cursor
+        this.kill();
       }
 
-      return new this.s.promiseLibrary(resolve => {
-        this._endSession(() => completeClose().then(resolve));
+      this._endSession(() => {
+        this.emit('close');
+        cb(null, this);
       });
-    }
-
-    return completeClose();
+    });
   }
 
   /**

lib/operations/cursor_ops.js

@@ -134,10 +134,9 @@ function toArray(cursor, callback) {
   const fetchDocs = () => {
     cursor._next((err, doc) => {
       if (err) {
-        return cursor._endSession
-          ? cursor._endSession(() => handleCallback(callback, err))
-          : handleCallback(callback, err);
+        return handleCallback(callback, err);
       }
+
       if (doc == null) {
         return cursor.close({ skipKillCursors: true }, () => handleCallback(callback, null, items));
       }