Merge pull request #11666 from Automattic/6.3

6.3

Author: vkarpov15

.github/workflows/test.yml

@@ -102,3 +102,15 @@ jobs:
         with:
           name: coverage
           path: coverage
+  test-replica-sets:
+    runs-on: ubuntu-latest
+    name: Replica Set tests
+    steps:
+      - uses: actions/checkout@a12a3943b4bdde767164f792f33f40b04645d846 # v3
+      - name: Setup node
+        uses: actions/setup-node@5b52f097d36d4b0b2f94ed6de710023fbb8b2236 # v3.1.0
+        with:
+          node-version: 16
+      - run: npm install
+      - name: Test
+        run: env START_REPLICA_SET=1 npm test

lib/aggregate.js

@@ -6,7 +6,7 @@
 
 const AggregationCursor = require('./cursor/AggregationCursor');
 const Query = require('./query');
-const applyGlobalMaxTimeMS = require('./helpers/query/applyGlobalMaxTimeMS');
+const { applyGlobalMaxTimeMS, applyGlobalDiskUse } = require('./helpers/query/applyGlobalOption');
 const getConstructorName = require('./helpers/getConstructorName');
 const prepareDiscriminatorPipeline = require('./helpers/aggregate/prepareDiscriminatorPipeline');
 const promiseOrCallback = require('./helpers/promiseOrCallback');
@@ -973,6 +973,7 @@ Aggregate.prototype.exec = function(callback) {
   const collection = this._model.collection;
 
   applyGlobalMaxTimeMS(this.options, model);
+  applyGlobalDiskUse(this.options, model);
 
   if (this.options && this.options.cursor) {
     return new AggregationCursor(this);

lib/cursor/ChangeStream.js

@@ -16,6 +16,7 @@ class ChangeStream extends EventEmitter {
 
     this.driverChangeStream = null;
     this.closed = false;
+    this.bindedEvents = false;
     this.pipeline = pipeline;
     this.options = options;
 
@@ -27,21 +28,60 @@ class ChangeStream extends EventEmitter {
       }
 
       this.driverChangeStream = driverChangeStream;
-      this._bindEvents();
       this.emit('ready');
     });
   }
 
   _bindEvents() {
+    if (this.bindedEvents) {
+      return;
+    }
+
+    this.bindedEvents = true;
+
+    if (this.driverChangeStream == null) {
+      this.once('ready', () => {
+        this.driverChangeStream.on('close', () => {
+          this.closed = true;
+        });
+
+        ['close', 'change', 'end', 'error'].forEach(ev => {
+          this.driverChangeStream.on(ev, data => this.emit(ev, data));
+        });
+      });
+
+      return;
+    }
+
     this.driverChangeStream.on('close', () => {
       this.closed = true;
     });
 
     ['close', 'change', 'end', 'error'].forEach(ev => {
-      this.driverChangeStream.on(ev, data => this.emit(ev, data));
+      this.driverChangeStream.on(ev, data => {
+        this.emit(ev, data);
+      });
     });
   }
 
+  hasNext(cb) {
+    return this.driverChangeStream.hasNext(cb);
+  }
+
+  next(cb) {
+    return this.driverChangeStream.next(cb);
+  }
+
+  on(event, handler) {
+    this._bindEvents();
+    return super.on(event, handler);
+  }
+
+  once(event, handler) {
+    this._bindEvents();
+    return super.once(event, handler);
+  }
+
   _queue(cb) {
     this.once('ready', () => cb());
   }

lib/cursor/QueryCursor.js

@@ -225,6 +225,8 @@ QueryCursor.prototype.next = function(callback) {
  * @param {Function} fn
  * @param {Object} [options]
  * @param {Number} [options.parallel] the number of promises to execute in parallel. Defaults to 1.
+ * @param {Number} [options.batchSize] if set, will call `fn()` with arrays of documents with length at most `batchSize`
+ * @param {Boolean} [options.continueOnError=false] if true, `eachAsync()` iterates through all docs even if `fn` throws an error. If false, `eachAsync()` throws an error immediately if the given function `fn()` throws an error.
  * @param {Function} [callback] executed when all docs have been processed
  * @return {Promise}
  * @api public

lib/error/eachAsyncMultiError.js

@@ -0,0 +1,41 @@
+/*!
+ * Module dependencies.
+ */
+
+'use strict';
+
+const MongooseError = require('./');
+
+
+/**
+ * If `eachAsync()` is called with `continueOnError: true`, there can be
+ * multiple errors. This error class contains an `errors` property, which
+ * contains an array of all errors that occurred in `eachAsync()`.
+ *
+ * @api private
+ */
+
+class EachAsyncMultiError extends MongooseError {
+  /**
+   * @param {String} connectionString
+   */
+  constructor(errors) {
+    let preview = errors.map(e => e.message).join(', ');
+    if (preview.length > 50) {
+      preview = preview.slice(0, 50) + '...';
+    }
+    super(`eachAsync() finished with ${errors.length} errors: ${preview}`);
+
+    this.errors = errors;
+  }
+}
+
+Object.defineProperty(EachAsyncMultiError.prototype, 'name', {
+  value: 'EachAsyncMultiError'
+});
+
+/*!
+ * exports
+ */
+
+module.exports = EachAsyncMultiError;

lib/helpers/cursor/eachAsync.js

@@ -4,6 +4,7 @@
  * Module dependencies.
  */
 
+const EachAsyncMultiError = require('../../error/eachAsyncMultiError');
 const immediate = require('../immediate');
 const promiseOrCallback = require('../promiseOrCallback');
 
@@ -24,10 +25,11 @@ const promiseOrCallback = require('../promiseOrCallback');
 module.exports = function eachAsync(next, fn, options, callback) {
   const parallel = options.parallel || 1;
   const batchSize = options.batchSize;
+  const continueOnError = options.continueOnError;
+  const aggregatedErrors = [];
   const enqueue = asyncQueue();
 
   return promiseOrCallback(callback, cb => {
-
     if (batchSize != null) {
       if (typeof batchSize !== 'number') {
         throw new TypeError('batchSize must be a number');
@@ -62,14 +64,22 @@ module.exports = function eachAsync(next, fn, options, callback) {
           return done();
         }
         if (err != null) {
-          error = err;
-          finalCallback(err);
-          return done();
+          if (continueOnError) {
+            aggregatedErrors.push(err);
+          } else {
+            error = err;
+            finalCallback(err);
+            return done();
+          }
         }
         if (doc == null) {
           drained = true;
           if (handleResultsInProgress <= 0) {
-            finalCallback(null);
+            const finalErr = continueOnError ?
+              createEachAsyncMultiError(aggregatedErrors) :
+              error;
+
+            finalCallback(finalErr);
           } else if (batchSize && documentsBatch.length) {
             handleNextResult(documentsBatch, currentDocumentIndex++, handleNextResultCallBack);
           }
@@ -102,11 +112,18 @@ module.exports = function eachAsync(next, fn, options, callback) {
             --handleResultsInProgress;
           }
           if (err != null) {
-            error = err;
-            return finalCallback(err);
+            if (continueOnError) {
+              aggregatedErrors.push(err);
+            } else {
+              error = err;
+              return finalCallback(err);
+            }
           }
           if (drained && handleResultsInProgress <= 0) {
-            return finalCallback(null);
+            const finalErr = continueOnError ?
+              createEachAsyncMultiError(aggregatedErrors) :
+              error;
+            return finalCallback(finalErr);
           }
 
           immediate(() => enqueue(fetch));
@@ -118,11 +135,18 @@ module.exports = function eachAsync(next, fn, options, callback) {
   }
 
   function handleNextResult(doc, i, callback) {
-    const promise = fn(doc, i);
-    if (promise && typeof promise.then === 'function') {
-      promise.then(
+    let maybePromise;
+    try {
+      maybePromise = fn(doc, i);
+    } catch (err) {
+      return callback(err);
+    }
+    if (maybePromise && typeof maybePromise.then === 'function') {
+      maybePromise.then(
         function() { callback(null); },
-        function(error) { callback(error || new Error('`eachAsync()` promise rejected without error')); });
+        function(error) {
+          callback(error || new Error('`eachAsync()` promise rejected without error'));
+        });
     } else {
       callback(null);
     }
@@ -158,3 +182,11 @@ function asyncQueue() {
     }
   }
 }
+
+function createEachAsyncMultiError(aggregatedErrors) {
+  if (aggregatedErrors.length === 0) {
+    return null;
+  }
+
+  return new EachAsyncMultiError(aggregatedErrors);
+}

lib/helpers/model/discriminator.js

@@ -17,7 +17,6 @@ const CUSTOMIZABLE_DISCRIMINATOR_OPTIONS = {
  */
 
 module.exports = function discriminator(model, name, schema, tiedValue, applyPlugins) {
-
   if (!(schema && schema.instanceOfSchema)) {
     throw new Error('You must pass a valid discriminator Schema');
   }
@@ -109,7 +108,7 @@ module.exports = function discriminator(model, name, schema, tiedValue, applyPlu
 
     utils.merge(schema, baseSchema, {
       isDiscriminatorSchemaMerge: true,
-      omit: { discriminators: true, base: true },
+      omit: { discriminators: true, base: true, _applyDiscriminators: true },
       omitNested: conflictingPaths.reduce((cur, path) => {
         cur['tree.' + path] = true;
         return cur;
@@ -141,7 +140,6 @@ module.exports = function discriminator(model, name, schema, tiedValue, applyPlu
     obj[key][schema.options.typeKey] = existingPath ? existingPath.options[schema.options.typeKey] : String;
     schema.add(obj);
 
-
     schema.discriminatorMapping = { key: key, value: value, isRoot: false };
 
     if (baseSchema.options.collection) {

lib/helpers/query/applyGlobalMaxTimeMS.js

@@ -0,0 +1,29 @@
+'use strict';
+
+const utils = require('../../utils');
+
+function applyGlobalMaxTimeMS(options, model) {
+  applyGlobalOption(options, model, 'maxTimeMS');
+}
+
+function applyGlobalDiskUse(options, model) {
+  applyGlobalOption(options, model, 'allowDiskUse');
+}
+
+module.exports = {
+  applyGlobalMaxTimeMS,
+  applyGlobalDiskUse
+};
+
+
+function applyGlobalOption(options, model, optionName) {
+  if (utils.hasUserDefinedProperty(options, optionName)) {
+    return;
+  }
+
+  if (utils.hasUserDefinedProperty(model.db.options, optionName)) {
+    options[optionName] = model.db.options[optionName];
+  } else if (utils.hasUserDefinedProperty(model.base.options, optionName)) {
+    options[optionName] = model.base.options[optionName];
+  }
+}

lib/helpers/query/applyGlobalOption.js

@@ -179,7 +179,9 @@ Mongoose.prototype.driver = driver;
 Mongoose.prototype.set = function(key, value) {
   const _mongoose = this instanceof Mongoose ? this : mongoose;
 
-  if (VALID_OPTIONS.indexOf(key) === -1) throw new Error(`\`${key}\` is an invalid option.`);
+  if (VALID_OPTIONS.indexOf(key) === -1) {
+    throw new Error(`\`${key}\` is an invalid option.`);
+  }
 
   if (arguments.length === 1) {
     return _mongoose.options[key];
@@ -467,6 +469,7 @@ Mongoose.prototype.pluralize = function(fn) {
  */
 
 Mongoose.prototype.model = function(name, schema, collection, options) {
+
   const _mongoose = this instanceof Mongoose ? this : mongoose;
 
   if (typeof schema === 'string') {
@@ -519,7 +522,6 @@ Mongoose.prototype.model = function(name, schema, collection, options) {
   }
 
   const model = _mongoose._model(name, schema, collection, options);
-
   _mongoose.connection.models[name] = model;
   _mongoose.models[name] = model;
 
@@ -561,12 +563,17 @@ Mongoose.prototype._model = function(name, schema, collection, options) {
 
   const connection = options.connection || _mongoose.connection;
   model = _mongoose.Model.compile(model || name, schema, collection, connection, _mongoose);
-
   // Errors handled internally, so safe to ignore error
   model.init(function $modelInitNoop() {});
 
   connection.emit('model', model);
 
+  if (schema._applyDiscriminators != null) {
+    for (const disc of Object.keys(schema._applyDiscriminators)) {
+      model.discriminator(disc, schema._applyDiscriminators[disc]);
+    }
+  }
+
   return model;
 };