fix(cursor): make aggregation cursors handle destroy() and handle destroy() before cursor created

Fix Automattic#14966

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 = false;
+      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

@@ -248,7 +248,19 @@ QueryCursor.prototype.close = async function close() {
  */
 
 QueryCursor.prototype._destroy = function _destroy(_err, callback) {
-  this.cursor.close()
+  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 = false;
       callback();

test/query.cursor.test.js

@@ -936,6 +936,19 @@ describe('QueryCursor', function() {
     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) {