Use same options for exec and cursor in Model.aggregate

Author: IchirokuXVI

lib/aggregate.js

@@ -87,6 +87,24 @@ function Aggregate(pipeline, model) {
 
 Aggregate.prototype.options;
 
+/**
+ * Returns default options for this aggregate.
+ *
+ * @param {Model} model
+ * @api private
+ */
+
+Aggregate.prototype._optionsForExec = function() {
+  const options = clone(this.options || {});
+
+  const asyncLocalStorage = this.model()?.db?.base.transactionAsyncLocalStorage?.getStore();
+  if (!options.hasOwnProperty('session') && asyncLocalStorage?.session != null) {
+    options.session = asyncLocalStorage.session;
+  }
+
+  return options;
+};
+
 /**
  * Get/set the model that this aggregation will execute on.
  *
@@ -1022,10 +1040,7 @@ Aggregate.prototype.exec = async function exec() {
   applyGlobalMaxTimeMS(this.options, model.db.options, model.base.options);
   applyGlobalDiskUse(this.options, model.db.options, model.base.options);
 
-  const asyncLocalStorage = this.model()?.db?.base.transactionAsyncLocalStorage?.getStore();
-  if (!this.options.hasOwnProperty('session') && asyncLocalStorage?.session != null) {
-    this.options.session = asyncLocalStorage.session;
-  }
+  const options = this._optionsForExec()
 
   if (this.options && this.options.cursor) {
     return new AggregationCursor(this);
@@ -1051,7 +1066,6 @@ Aggregate.prototype.exec = async function exec() {
     throw new MongooseError('Aggregate has empty pipeline');
   }
 
-  const options = clone(this.options || {});
   let result;
   try {
     const cursor = await collection.aggregate(this._pipeline, options);

lib/cursor/aggregationCursor.js

@@ -44,6 +44,7 @@ function AggregationCursor(agg) {
   const model = agg._model;
   delete agg.options.cursor.useMongooseAggCursor;
   this._mongooseOptions = {};
+  this.options = {};
 
   _init(model, this, agg);
 }
@@ -57,21 +58,25 @@ util.inherits(AggregationCursor, Readable);
 function _init(model, c, agg) {
   if (!model.collection.buffer) {
     model.hooks.execPre('aggregate', agg, function() {
+      Object.assign(c.options, agg._optionsForExec());
+
       if (typeof agg.options?.cursor?.transform === 'function') {
         c._transforms.push(agg.options.cursor.transform);
       }
 
-      c.cursor = model.collection.aggregate(agg._pipeline, agg.options || {});
+      c.cursor = model.collection.aggregate(agg._pipeline, c.options || {});
       c.emit('cursor', c.cursor);
     });
   } else {
     model.collection.emitter.once('queue', function() {
       model.hooks.execPre('aggregate', agg, function() {
+        Object.assign(c.options, agg._optionsForExec());
+
         if (typeof agg.options?.cursor?.transform === 'function') {
           c._transforms.push(agg.options.cursor.transform);
         }
 
-        c.cursor = model.collection.aggregate(agg._pipeline, agg.options || {});
+        c.cursor = model.collection.aggregate(agg._pipeline, c.options || {});
         c.emit('cursor', c.cursor);
       });
     });