fix: assert update/replace atomic requirements in bulk operations

`checkForAtomicOperators` was refactored into the more versatile
`hasAtomicOperators` and the logic for asserting atomic operator
requirements was streamlined across all crud operations as well as
bulk operations.

NODE-2660

Author: mbroadst

lib/bulk/common.js

@@ -11,6 +11,8 @@ const applyRetryableWrites = require('../utils').applyRetryableWrites;
 const applyWriteConcern = require('../utils').applyWriteConcern;
 const executeLegacyOperation = require('../utils').executeLegacyOperation;
 const isPromiseLike = require('../utils').isPromiseLike;
+const hasAtomicOperators = require('../utils').hasAtomicOperators;
+const maxWireVersion = require('../core/utils').maxWireVersion;
 
 // Error codes
 const WRITE_CONCERN_ERROR = 64;
@@ -641,6 +643,10 @@ class FindOperators {
       document.hint = updateDocument.hint;
     }
 
+    if (!hasAtomicOperators(updateDocument)) {
+      throw new TypeError('Update document requires atomic operators');
+    }
+
     // Clear out current Op
     this.s.currentOp = null;
     return this.s.options.addToOperationsList(this, UPDATE, document);
@@ -650,12 +656,33 @@ class FindOperators {
    * Add a replace one operation to the bulk operation
    *
    * @method
-   * @param {object} updateDocument the new document to replace the existing one with
+   * @param {object} replacement the new document to replace the existing one with
    * @throws {MongoError} If operation cannot be added to bulk write
    * @return {OrderedBulkOperation|UnorderedBulkOperation} A reference to the parent BulkOperation
    */
-  replaceOne(updateDocument) {
-    this.updateOne(updateDocument);
+  replaceOne(replacement) {
+    // Perform upsert
+    const upsert = typeof this.s.currentOp.upsert === 'boolean' ? this.s.currentOp.upsert : false;
+
+    // Establish the update command
+    const document = {
+      q: this.s.currentOp.selector,
+      u: replacement,
+      multi: false,
+      upsert: upsert
+    };
+
+    if (replacement.hint) {
+      document.hint = replacement.hint;
+    }
+
+    if (hasAtomicOperators(replacement)) {
+      throw new TypeError('Replacement document must not use atomic operators');
+    }
+
+    // Clear out current Op
+    this.s.currentOp = null;
+    return this.s.options.addToOperationsList(this, UPDATE, document);
   }
 
   /**
@@ -943,6 +970,12 @@ class BulkOperationBase {
 
     // Crud spec update format
     if (op.updateOne || op.updateMany || op.replaceOne) {
+      if (op.replaceOne && hasAtomicOperators(op[key].replacement)) {
+        throw new TypeError('Replacement document must not use atomic operators');
+      } else if ((op.updateOne || op.updateMany) && !hasAtomicOperators(op[key].update)) {
+        throw new TypeError('Update document requires atomic operators');
+      }
+
       const multi = op.updateOne || op.replaceOne ? false : true;
       const operation = {
         q: op[key].filter,
@@ -960,7 +993,15 @@ class BulkOperationBase {
       } else {
         if (op[key].upsert) operation.upsert = true;
       }
-      if (op[key].arrayFilters) operation.arrayFilters = op[key].arrayFilters;
+      if (op[key].arrayFilters) {
+        // TODO: this check should be done at command construction against a connection, not a topology
+        if (maxWireVersion(this.s.topology) < 6) {
+          throw new TypeError('arrayFilters are only supported on MongoDB 3.6+');
+        }
+
+        operation.arrayFilters = op[key].arrayFilters;
+      }
+
       return this.s.options.addToOperationsList(this, UPDATE, operation);
     }
 

lib/collection.js

@@ -5,7 +5,6 @@ const deprecateOptions = require('./utils').deprecateOptions;
 const checkCollectionName = require('./utils').checkCollectionName;
 const ObjectID = require('./core').BSON.ObjectID;
 const MongoError = require('./core').MongoError;
-const toError = require('./utils').toError;
 const normalizeHintField = require('./utils').normalizeHintField;
 const decorateCommand = require('./utils').decorateCommand;
 const decorateWithCollation = require('./utils').decorateWithCollation;
@@ -23,7 +22,6 @@ const AggregationCursor = require('./aggregation_cursor');
 const CommandCursor = require('./command_cursor');
 
 // Operations
-const checkForAtomicOperators = require('./operations/collection_ops').checkForAtomicOperators;
 const ensureIndex = require('./operations/collection_ops').ensureIndex;
 const group = require('./operations/collection_ops').group;
 const parallelCollectionScan = require('./operations/collection_ops').parallelCollectionScan;
@@ -747,14 +745,6 @@ Collection.prototype.insert = deprecate(function(docs, options, callback) {
  */
 Collection.prototype.updateOne = function(filter, update, options, callback) {
   if (typeof options === 'function') (callback = options), (options = {});
-  options = options || {};
-
-  const err = checkForAtomicOperators(update);
-  if (err) {
-    if (typeof callback === 'function') return callback(err);
-    return this.s.promiseLibrary.reject(err);
-  }
-
   options = Object.assign({}, options);
 
   // Add ignoreUndefined
@@ -763,9 +753,11 @@ Collection.prototype.updateOne = function(filter, update, options, callback) {
     options.ignoreUndefined = this.s.options.ignoreUndefined;
   }
 
-  const updateOneOperation = new UpdateOneOperation(this, filter, update, options);
-
-  return executeOperation(this.s.topology, updateOneOperation, callback);
+  return executeOperation(
+    this.s.topology,
+    new UpdateOneOperation(this, filter, update, options),
+    callback
+  );
 };
 
 /**
@@ -798,9 +790,11 @@ Collection.prototype.replaceOne = function(filter, doc, options, callback) {
     options.ignoreUndefined = this.s.options.ignoreUndefined;
   }
 
-  const replaceOneOperation = new ReplaceOneOperation(this, filter, doc, options);
-
-  return executeOperation(this.s.topology, replaceOneOperation, callback);
+  return executeOperation(
+    this.s.topology,
+    new ReplaceOneOperation(this, filter, doc, options),
+    callback
+  );
 };
 
 /**
@@ -826,14 +820,6 @@ Collection.prototype.replaceOne = function(filter, doc, options, callback) {
  */
 Collection.prototype.updateMany = function(filter, update, options, callback) {
   if (typeof options === 'function') (callback = options), (options = {});
-  options = options || {};
-
-  const err = checkForAtomicOperators(update);
-  if (err) {
-    if (typeof callback === 'function') return callback(err);
-    return this.s.promiseLibrary.reject(err);
-  }
-
   options = Object.assign({}, options);
 
   // Add ignoreUndefined
@@ -842,9 +828,11 @@ Collection.prototype.updateMany = function(filter, update, options, callback) {
     options.ignoreUndefined = this.s.options.ignoreUndefined;
   }
 
-  const updateManyOperation = new UpdateManyOperation(this, filter, update, options);
-
-  return executeOperation(this.s.topology, updateManyOperation, callback);
+  return executeOperation(
+    this.s.topology,
+    new UpdateManyOperation(this, filter, update, options),
+    callback
+  );
 };
 
 /**
@@ -1679,13 +1667,11 @@ Collection.prototype.findOneAndDelete = function(filter, options, callback) {
   if (typeof options === 'function') (callback = options), (options = {});
   options = options || {};
 
-  // Basic validation
-  if (filter == null || typeof filter !== 'object')
-    throw toError('filter parameter must be an object');
-
-  const findOneAndDeleteOperation = new FindOneAndDeleteOperation(this, filter, options);
-
-  return executeOperation(this.s.topology, findOneAndDeleteOperation, callback);
+  return executeOperation(
+    this.s.topology,
+    new FindOneAndDeleteOperation(this, filter, options),
+    callback
+  );
 };
 
 /**
@@ -1714,27 +1700,11 @@ Collection.prototype.findOneAndReplace = function(filter, replacement, options,
   if (typeof options === 'function') (callback = options), (options = {});
   options = options || {};
 
-  // Basic validation
-  if (filter == null || typeof filter !== 'object')
-    throw toError('filter parameter must be an object');
-  if (replacement == null || typeof replacement !== 'object')
-    throw toError('replacement parameter must be an object');
-
-  // Check that there are no atomic operators
-  const keys = Object.keys(replacement);
-
-  if (keys[0] && keys[0][0] === '$') {
-    throw toError('The replacement document must not contain atomic operators.');
-  }
-
-  const findOneAndReplaceOperation = new FindOneAndReplaceOperation(
-    this,
-    filter,
-    replacement,
-    options
+  return executeOperation(
+    this.s.topology,
+    new FindOneAndReplaceOperation(this, filter, replacement, options),
+    callback
   );
-
-  return executeOperation(this.s.topology, findOneAndReplaceOperation, callback);
 };
 
 /**
@@ -1764,21 +1734,11 @@ Collection.prototype.findOneAndUpdate = function(filter, update, options, callba
   if (typeof options === 'function') (callback = options), (options = {});
   options = options || {};
 
-  // Basic validation
-  if (filter == null || typeof filter !== 'object')
-    throw toError('filter parameter must be an object');
-  if (update == null || typeof update !== 'object')
-    throw toError('update parameter must be an object');
-
-  const err = checkForAtomicOperators(update);
-  if (err) {
-    if (typeof callback === 'function') return callback(err);
-    return this.s.promiseLibrary.reject(err);
-  }
-
-  const findOneAndUpdateOperation = new FindOneAndUpdateOperation(this, filter, update, options);
-
-  return executeOperation(this.s.topology, findOneAndUpdateOperation, callback);
+  return executeOperation(
+    this.s.topology,
+    new FindOneAndUpdateOperation(this, filter, update, options),
+    callback
+  );
 };
 
 /**

lib/operations/collection_ops.js

@@ -13,7 +13,6 @@ const indexInformationDb = require('./db_ops').indexInformation;
 const Long = require('../core').BSON.Long;
 const MongoError = require('../core').MongoError;
 const ReadPreference = require('../core').ReadPreference;
-const toError = require('../utils').toError;
 const insertDocuments = require('./common_functions').insertDocuments;
 const updateDocuments = require('./common_functions').updateDocuments;
 
@@ -51,24 +50,6 @@ const updateDocuments = require('./common_functions').updateDocuments;
 const groupFunction =
   'function () {\nvar c = db[ns].find(condition);\nvar map = new Map();\nvar reduce_function = reduce;\n\nwhile (c.hasNext()) {\nvar obj = c.next();\nvar key = {};\n\nfor (var i = 0, len = keys.length; i < len; ++i) {\nvar k = keys[i];\nkey[k] = obj[k];\n}\n\nvar aggObj = map.get(key);\n\nif (aggObj == null) {\nvar newObj = Object.extend({}, key);\naggObj = Object.extend(newObj, initial);\nmap.put(key, aggObj);\n}\n\nreduce_function(obj, aggObj);\n}\n\nreturn { "result": map.values() };\n}';
 
-// Check the update operation to ensure it has atomic operators.
-function checkForAtomicOperators(update) {
-  if (Array.isArray(update)) {
-    return update.reduce((err, u) => err || checkForAtomicOperators(u), null);
-  }
-
-  const keys = Object.keys(update);
-
-  // same errors as the server would give for update doc lacking atomic operators
-  if (keys.length === 0) {
-    return toError('The update operation document must contain at least one atomic operator.');
-  }
-
-  if (keys[0][0] !== '$') {
-    return toError('the update operation document must contain atomic operators.');
-  }
-}
-
 /**
  * Create an index on the db and collection.
  *
@@ -360,7 +341,6 @@ function save(coll, doc, options, callback) {
 }
 
 module.exports = {
-  checkForAtomicOperators,
   createIndex,
   createIndexes,
   ensureIndex,

lib/operations/find_one_and_delete.js

@@ -9,6 +9,11 @@ class FindOneAndDeleteOperation extends FindAndModifyOperation {
     finalOptions.fields = options.projection;
     finalOptions.remove = true;
 
+    // Basic validation
+    if (filter == null || typeof filter !== 'object') {
+      throw new TypeError('Filter parameter must be an object');
+    }
+
     super(collection, filter, finalOptions.sort, null, finalOptions);
   }
 }

lib/operations/find_one_and_replace.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const FindAndModifyOperation = require('./find_and_modify');
+const hasAtomicOperators = require('../utils').hasAtomicOperators;
 
 class FindOneAndReplaceOperation extends FindAndModifyOperation {
   constructor(collection, filter, replacement, options) {
@@ -11,6 +12,18 @@ class FindOneAndReplaceOperation extends FindAndModifyOperation {
     finalOptions.new = options.returnOriginal !== void 0 ? !options.returnOriginal : false;
     finalOptions.upsert = options.upsert !== void 0 ? !!options.upsert : false;
 
+    if (filter == null || typeof filter !== 'object') {
+      throw new TypeError('Filter parameter must be an object');
+    }
+
+    if (replacement == null || typeof replacement !== 'object') {
+      throw new TypeError('Replacement parameter must be an object');
+    }
+
+    if (hasAtomicOperators(replacement)) {
+      throw new TypeError('Replacement document must not contain atomic operators');
+    }
+
     super(collection, filter, finalOptions.sort, replacement, finalOptions);
   }
 }

lib/operations/find_one_and_update.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const FindAndModifyOperation = require('./find_and_modify');
+const { hasAtomicOperators } = require('../utils');
 
 class FindOneAndUpdateOperation extends FindAndModifyOperation {
   constructor(collection, filter, update, options) {
@@ -12,6 +13,18 @@ class FindOneAndUpdateOperation extends FindAndModifyOperation {
       typeof options.returnOriginal === 'boolean' ? !options.returnOriginal : false;
     finalOptions.upsert = typeof options.upsert === 'boolean' ? options.upsert : false;
 
+    if (filter == null || typeof filter !== 'object') {
+      throw new TypeError('Filter parameter must be an object');
+    }
+
+    if (update == null || typeof update !== 'object') {
+      throw new TypeError('Update parameter must be an object');
+    }
+
+    if (!hasAtomicOperators(update)) {
+      throw new TypeError('Update document requires atomic operators');
+    }
+
     super(collection, filter, finalOptions.sort, update, finalOptions);
   }
 }