fix(NODE-5502): recursive calls to next cause memory leak

Author: nbbeeken

src/cursor/abstract_cursor.ts

@@ -701,96 +701,90 @@ async function next<T>(
     transform: boolean;
   }
 ): Promise<T | null> {
-  const cursorId = cursor[kId];
   if (cursor.closed) {
     return null;
   }
 
-  if (cursor[kDocuments].length !== 0) {
-    const doc = cursor[kDocuments].shift();
-
-    if (doc != null && transform && cursor[kTransform]) {
-      try {
-        return cursor[kTransform](doc);
-      } catch (error) {
-        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;
-      }
+  do {
+    if (cursor[kId] == null) {
+      // All cursors must operate within a session, one must be made implicitly if not explicitly provided
+      await promisify(cb => cursor[kInit](cb))();
     }
 
-    return doc;
-  }
+    if (cursor[kDocuments].length !== 0) {
+      const doc = cursor[kDocuments].shift();
 
-  if (cursorId == null) {
-    // All cursors must operate within a session, one must be made implicitly if not explicitly provided
-    const init = promisify(cb => cursor[kInit](cb));
-    await init();
-    return next(cursor, { blocking, transform });
-  }
+      if (doc != null && transform && cursor[kTransform]) {
+        try {
+          return cursor[kTransform](doc);
+        } catch (error) {
+          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;
+        }
+      }
 
-  if (cursorIsDead(cursor)) {
-    // if the cursor is dead, we clean it up
-    // 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;
-  }
+      return doc;
+    }
 
-  // otherwise need to call getMore
-  const batchSize = cursor[kOptions].batchSize || 1000;
-  const getMore = promisify((batchSize: number, cb: Callback<Document | undefined>) =>
-    cursor._getMore(batchSize, cb)
-  );
+    if (cursorIsDead(cursor)) {
+      // if the cursor is dead, we clean it up
+      // 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;
+    }
+
+    // otherwise need to call getMore
+    const batchSize = cursor[kOptions].batchSize || 1000;
 
-  let response: Document | undefined;
-  try {
-    response = await getMore(batchSize);
-  } catch (error) {
-    if (error) {
+    try {
+      const response = await promisify((batchSize: number, cb: Callback<Document>) =>
+        cursor._getMore(batchSize, cb)
+      )(batchSize);
+
+      if (response) {
+        const cursorId =
+          typeof response.cursor.id === 'number'
+            ? Long.fromNumber(response.cursor.id)
+            : typeof response.cursor.id === 'bigint'
+            ? Long.fromBigInt(response.cursor.id)
+            : response.cursor.id;
+
+        cursor[kDocuments].pushMany(response.cursor.nextBatch);
+        cursor[kId] = cursorId;
+      }
+    } catch (error) {
       await cleanupCursorAsync(cursor, { error }).catch(() => {
         // `cleanupCursorAsync` should never throw, but if it does we want to throw the original
         // error instead.
       });
       throw error;
     }
-  }
 
-  if (response) {
-    const cursorId =
-      typeof response.cursor.id === 'number'
-        ? Long.fromNumber(response.cursor.id)
-        : typeof response.cursor.id === 'bigint'
-        ? Long.fromBigInt(response.cursor.id)
-        : response.cursor.id;
-
-    cursor[kDocuments].pushMany(response.cursor.nextBatch);
-    cursor[kId] = cursorId;
-  }
-
-  if (cursorIsDead(cursor)) {
-    // If we successfully received a response from a cursor BUT the cursor indicates that it is exhausted,
-    // 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.
-    //
-    // 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 (cursorIsDead(cursor)) {
+      // If we successfully received a response from a cursor BUT the cursor indicates that it is exhausted,
+      // 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.
+      //
+      // 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) {
-    return null;
-  }
+    if (cursor[kDocuments].length === 0 && blocking === false) {
+      return null;
+    }
+  } while (!cursorIsDead(cursor) || cursor[kDocuments].length !== 0);
 
-  return next(cursor, { blocking, transform });
+  return null;
 }
 
 function cursorIsDead(cursor: AbstractCursor): boolean {
-  const cursorId = cursor[kId];
-  return !!cursorId && cursorId.isZero();
+  return cursor[kId]?.isZero() ?? true;
 }
 
 const cleanupCursorAsync = promisify(cleanupCursor);
@@ -882,7 +876,7 @@ export function assertUninitialized(cursor: AbstractCursor): void {
 
 class ReadableCursorStream extends Readable {
   private _cursor: AbstractCursor;
-  private _readInProgress = false;
+  private _readInProgress: Promise<Document> | null = null;
 
   constructor(cursor: AbstractCursor) {
     super({
@@ -895,8 +889,7 @@ class ReadableCursorStream extends Readable {
 
   // eslint-disable-next-line @typescript-eslint/no-unused-vars
   override _read(size: number): void {
-    if (!this._readInProgress) {
-      this._readInProgress = true;
+    if (this._readInProgress == null) {
       this._readNext();
     }
   }
@@ -909,8 +902,13 @@ class ReadableCursorStream extends Readable {
   }
 
   private _readNext() {
-    next(this._cursor, { blocking: true, transform: true }).then(
+    if (this._readInProgress) return;
+
+    this._readInProgress = next(this._cursor, { blocking: true, transform: true });
+
+    this._readInProgress.then(
       result => {
+        this._readInProgress = null;
         if (result == null) {
           this.push(null);
         } else if (this.destroyed) {
@@ -919,11 +917,10 @@ class ReadableCursorStream extends Readable {
           if (this.push(result)) {
             return this._readNext();
           }
-
-          this._readInProgress = false;
         }
       },
       err => {
+        this._readInProgress = null;
         // NOTE: This is questionable, but we have a test backing the behavior. It seems the
         //       desired behavior is that a stream ends cleanly when a user explicitly closes
         //       a client during iteration. Alternatively, we could do the "right" thing and

src/cursor/find_cursor.ts

@@ -1,4 +1,4 @@
-import type { Document } from '../bson';
+import { type Document, Long } from '../bson';
 import { MongoInvalidArgumentError, MongoTailableCursorError } from '../error';
 import type { ExplainVerbosityLike } from '../explain';
 import type { MongoClient } from '../mongo_client';
@@ -101,7 +101,9 @@ export class FindCursor<TSchema = any> extends AbstractCursor<TSchema> {
         limit && limit > 0 && numReturned + batchSize > limit ? limit - numReturned : batchSize;
 
       if (batchSize <= 0) {
-        this.close().finally(() => callback());
+        this.close().finally(() =>
+          callback(undefined, { cursor: { id: Long.ZERO, nextBatch: [] } })
+        );
         return;
       }
     }

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

@@ -448,7 +448,7 @@ describe('Change Streams', function () {
       const willBeChange = once(changeStream, 'change');
       await once(changeStream.cursor, 'init');
 
-      collection.insertOne({ a: 1 });
+      await collection.insertOne({ a: 1 });
 
       const [change] = await willBeChange;
       expect(change).to.have.property('operationType', 'insert');