Merge pull request Automattic#14982 from Automattic/vkarpov15/Automatticgh-14966

fix(cursor): close underlying query cursor when calling destroy()

Author: vkarpov15

lib/cursor/aggregationCursor.js

@@ -196,6 +196,37 @@ AggregationCursor.prototype.close = async function close() {
   this.emit('close');
 };
 
+/**
+ * Marks this cursor as destroyed. Will stop streaming and subsequent calls to
+ * `next()` will error.
+ *
+ * @return {this}
+ * @api private
+ * @method _destroy
+ */
+
+AggregationCursor.prototype._destroy = function _destroy(_err, callback) {
+  let waitForCursor = null;
+  if (!this.cursor) {
+    waitForCursor = new Promise((resolve) => {
+      this.once('cursor', resolve);
+    });
+  } else {
+    waitForCursor = Promise.resolve();
+  }
+
+  waitForCursor
+    .then(() => this.cursor.close())
+    .then(() => {
+      this._closed = true;
+      callback();
+    })
+    .catch(error => {
+      callback(error);
+    });
+  return this;
+};
+
 /**
  * Get the next document from this cursor. Will return `null` when there are
  * no documents left.

lib/cursor/queryCursor.js

@@ -238,6 +238,39 @@ QueryCursor.prototype.close = async function close() {
   }
 };
 
+/**
+ * Marks this cursor as destroyed. Will stop streaming and subsequent calls to
+ * `next()` will error.
+ *
+ * @return {this}
+ * @api private
+ * @method _destroy
+ */
+
+QueryCursor.prototype._destroy = function _destroy(_err, callback) {
+  let waitForCursor = null;
+  if (!this.cursor) {
+    waitForCursor = new Promise((resolve) => {
+      this.once('cursor', resolve);
+    });
+  } else {
+    waitForCursor = Promise.resolve();
+  }
+
+  waitForCursor
+    .then(() => {
+      this.cursor.close();
+    })
+    .then(() => {
+      this._closed = true;
+      callback();
+    })
+    .catch(error => {
+      callback(error);
+    });
+  return this;
+};
+
 /**
  * Rewind this cursor to its uninitialized state. Any options that are present on the cursor will
  * remain in effect. Iterating this cursor will cause new queries to be sent to the server, even

test/query.cursor.test.js

@@ -4,6 +4,7 @@
 
 'use strict';
 
+const { once } = require('events');
 const start = require('./common');
 
 const assert = require('assert');
@@ -920,6 +921,34 @@ describe('QueryCursor', function() {
     assert.ok(cursor.cursor);
     assert.equal(driverCursor, cursor.cursor);
   });
+
+  it('handles destroy() (gh-14966)', async function() {
+    db.deleteModel(/Test/);
+    const TestModel = db.model('Test', mongoose.Schema({ name: String }));
+
+    const stream = await TestModel.find().cursor();
+    await once(stream, 'cursor');
+    assert.ok(!stream.cursor.closed);
+
+    stream.destroy();
+
+    await once(stream.cursor, 'close');
+    assert.ok(stream.destroyed);
+    assert.ok(stream.cursor.closed);
+  });
+
+  it('handles destroy() before cursor is created (gh-14966)', async function() {
+    db.deleteModel(/Test/);
+    const TestModel = db.model('Test', mongoose.Schema({ name: String }));
+
+    const stream = await TestModel.find().cursor();
+    assert.ok(!stream.cursor);
+    stream.destroy();
+
+    await once(stream, 'cursor');
+    assert.ok(stream.destroyed);
+    assert.ok(stream.cursor.closed);
+  });
 });
 
 async function delay(ms) {

types/cursor.d.ts

@@ -26,6 +26,12 @@ declare module 'mongoose' {
      */
     close(): Promise<void>;
 
+    /**
+     * Destroy this cursor, closing the underlying cursor. Will stop streaming
+     * and subsequent calls to `next()` will error.
+     */
+    destroy(): this;
+
     /**
      * Rewind this cursor to its uninitialized state. Any options that are present on the cursor will
      * remain in effect. Iterating this cursor will cause new queries to be sent to the server, even