fix lint and refactor

Author: baileympearson

src/cursor/abstract_cursor.ts

@@ -1,5 +1,5 @@
 import { Readable, Transform } from 'stream';
-import { callbackify, promisify } from 'util';
+import { promisify } from 'util';
 
 import { type BSONSerializeOptions, type Document, Long, pluckBSONSerializeOptions } from '../bson';
 import {
@@ -708,7 +708,10 @@ async function next<T>(
       try {
         return cursor[kTransform](doc);
       } catch (error) {
-        await cleanupCursorAsync(cursor, { error, needsToEmitClosed: true });
+        await cleanupCursorAsync(cursor, { error, needsToEmitClosed: true }).catch(() => {
+          // `cleanupCursorAsync` should never throw, but if it does we want to throw the original
+          // error instead.
+        });
         throw error;
       }
     }
@@ -725,7 +728,9 @@ async function next<T>(
 
   if (cursorIsDead(cursor)) {
     // if the cursor is dead, we clean it up
-    await cleanupCursorAsync(cursor);
+    // cleanupCursorAsync should never throw, but if it does it indicates a bug in the driver
+    // and we should surface the error
+    await cleanupCursorAsync(cursor, {});
     return null;
   }
 
@@ -740,7 +745,10 @@ async function next<T>(
     response = await getMore(batchSize);
   } catch (error) {
     if (error) {
-      await cleanupCursorAsync(cursor, { error });
+      await cleanupCursorAsync(cursor, { error }).catch(() => {
+        // `cleanupCursorAsync` should never throw, but if it does we want to throw the original
+        // error instead.
+      });
       throw error;
     }
   }
@@ -762,7 +770,10 @@ async function next<T>(
     // we intentionally clean up the cursor to release its session back into the pool before the cursor
     // is iterated.  This prevents a cursor that is exhausted on the server from holding
     // onto a session indefinitely until the AbstractCursor is iterated.
-    await cleanupCursorAsync(cursor);
+    //
+    // cleanupCursorAsync should never throw, but if it does it indicates a bug in the driver
+    // and we should surface the error
+    await cleanupCursorAsync(cursor, {});
   }
 
   if (cursor[kDocuments].length === 0 && blocking === false) {
@@ -777,20 +788,7 @@ function cursorIsDead(cursor: AbstractCursor): boolean {
   return !!cursorId && cursorId.isZero();
 }
 
-const cleanupCursorAsyncInternal = promisify(cleanupCursor);
-
-async function cleanupCursorAsync<T>(
-  cursor: AbstractCursor<T>,
-  options: { needsToEmitClosed?: boolean; error?: AnyError } = {}
-): Promise<void> {
-  try {
-    await cleanupCursorAsyncInternal(cursor, options);
-  } catch {
-    // `cleanupCursor` never throws but we can't really test that.
-    // so this is a hack to ensure that any upstream consumers
-    // can safely guarantee on this wrapper never throwing.
-  }
-}
+const cleanupCursorAsync = promisify(cleanupCursor);
 
 function cleanupCursor(
   cursor: AbstractCursor,
@@ -802,6 +800,10 @@ function cleanupCursor(
   const server = cursor[kServer];
   const session = cursor[kSession];
   const error = options?.error;
+
+  // Cursors only emit closed events once the client-side cursor has been exhausted fully or there
+  // was an error.  Notably, when the server returns a cursor id of 0 and a non-empty batch, we
+  // cleanup the cursor but don't emit a `close` event.
   const needsToEmitClosed = options?.needsToEmitClosed ?? cursor[kDocuments].length === 0;
 
   if (error) {

test/integration/crud/misc_cursors.test.js

@@ -1708,7 +1708,6 @@ describe('Cursor', function () {
     expect(cursor).property('closed', false);
 
     const willClose = once(cursor, 'close');
-    const willEnd = once(stream, 'close');
 
     const dataEvents = on(stream, 'data');
 
@@ -1722,16 +1721,12 @@ describe('Cursor', function () {
     // After 5 successful data events, destroy stream
     stream.destroy();
 
-    // We should get a a close event on the stream and a close event on the cursor
-    // We should **not** get an 'error' event,
+    // We should get a close event on the stream and a close event on the cursor
+    // We should **not** get an 'error' or an 'end' event,
     // the following will throw if either stream or cursor emitted an 'error' event
-    await Promise.race([
-      willEnd,
-      sleep(100, { ref: false }).then(() => Promise.reject(new Error('end event never emitted')))
-    ]);
     await Promise.race([
       willClose,
-      sleep(100, { ref: false }).then(() => Promise.reject(new Error('close event never emitted')))
+      sleep(100).then(() => Promise.reject(new Error('close event never emitted')))
     ]);
   });
 
@@ -3606,7 +3601,6 @@ describe('Cursor', function () {
 
     await collection.insertMany(docs);
 
-    // TODO - talk to Neal about this test
     const cursor = await collection.find({}, { batchSize: 3 });
     for (let i = 0; i < 3; ++i) {
       await cursor.next();

test/integration/node-specific/abstract_cursor.test.ts

@@ -117,7 +117,6 @@ describe('class AbstractCursor', function () {
         expect(doc.name).to.equal('JOHN DOE');
       });
 
-      // skipped because these tests fail after throwing uncaught exceptions
       it(`when the transform throws, Cursor.stream() propagates the error to the user`, async () => {
         const cursor = collection
           .find()

test/integration/node-specific/cursor_stream.test.js

@@ -1,7 +1,7 @@
 'use strict';
 const { expect } = require('chai');
 const { Binary } = require('../../mongodb');
-const { setTimeout, setImmediate } = require('timers');
+const { setTimeout } = require('timers');
 
 describe('Cursor Streams', function () {
   let client;
@@ -296,7 +296,6 @@ describe('Cursor Streams', function () {
         const stream = cursor.stream();
         stream.on('error', err => (error = err));
         cursor.on('close', function () {
-          // NOTE: use `setImmediate` here because the stream implementation uses `nextTick` to emit the error
           setTimeout(() => {
             expect(error).to.exist;
             client.close(done);