SERVER-6558 Add writeConcern option to findAndModify command

alabid · alabid · commit 21c4f06e356d · 2015-03-17T11:57:00.000-04:00
diff --git a/jstests/replsets/find_and_modify_wc.js b/jstests/replsets/find_and_modify_wc.js
@@ -0,0 +1,79 @@
+//
+// Tests writeConcerns with findAndModify command
+//
+(function() {
+    'use strict';
+
+    var nodeCount = 3;
+    var rst = new ReplSetTest({ nodes: nodeCount });
+    rst.startSet({ nojournal: "" });
+    rst.initiate();
+
+    var primary = rst.getPrimary();
+    var coll = primary.getCollection("test.find_and_modify_wc");
+    coll.remove({});
+
+    // insert some documents
+    var docs = [];
+    for (var i = 1; i <= 5; ++i) {
+        docs.push({ i: i, j: 2*i });
+    }
+    var res = coll.runCommand({ insert: coll.getName(),
+                                documents: docs,
+                                writeConcern: { w: nodeCount } });
+    assert(res.ok);
+    assert.eq(5, coll.count());
+
+    // use for updates in subsequent runCommand calls
+    var reqUpdate = {
+        findAndModify: coll.getName(),
+        query: { i: 3 },
+        update: { $inc: { j: 1 } },
+        writeConcern: { w: 'majority' }
+    };
+
+    // Verify findAndModify returns old document new: false
+    var res = coll.runCommand(reqUpdate);
+    assert(res.ok);
+    assert(res.value);
+    // (2 * res.value.i) == 6 == res.value.j (old document)
+    assert.eq(2 * res.value.i, res.value.j);
+    assert(!res.writeConcernError);
+
+    // Verify findAndModify returns new document with new: true
+    reqUpdate.new = true;
+    res = coll.runCommand(reqUpdate);
+    assert(res.ok);
+    assert(res.value);
+    // (2 * res.value.i + 2) == 8 == res.value.j (new document after two updates)
+    assert.eq(2 * res.value.i + 2, res.value.j);
+    assert(!res.writeConcernError);
+
+    // Verify findAndModify remove works
+    res = coll.runCommand({
+        findAndModify: coll.getName(),
+        sort: { i: 1 },
+        remove: true,
+        writeConcern: { w: nodeCount }
+    });
+    assert.eq(res.value.i, 1);
+    assert.eq(coll.count(), 4);
+    assert(!res.writeConcernError);
+
+    // Verify findAndModify returns writeConcernError
+    // when given invalid writeConcerns
+    [
+        { w: 'invalid' },
+        { w: nodeCount + 1 }
+    ].forEach(function(wc) {
+        reqUpdate.writeConcern = wc;
+        res = coll.runCommand(reqUpdate);
+
+        assert(res.writeConcernError);
+        assert(res.writeConcernError.code);
+        assert(res.writeConcernError.errmsg);
+    });
+
+    rst.stopSet();
+
+})();
diff --git a/src/mongo/db/commands.cpp b/src/mongo/db/commands.cpp
@@ -51,6 +51,7 @@
 #include "mongo/db/jsobj.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/server_parameters.h"
+#include "mongo/s/write_ops/wc_error_detail.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -261,6 +262,15 @@ namespace mongo {
         }
     }
 
+    void Command::appendCommandWCStatus(BSONObjBuilder& result, const Status& status) {
+        if (!status.isOK()) {
+            WCErrorDetail wcError;
+            wcError.setErrCode(status.code());
+            wcError.setErrMessage(status.reason());
+            result.append("writeConcernError", wcError.toBSON());
+        }
+    }
+
     Status Command::getStatusFromCommandResult(const BSONObj& result) {
         return mongo::getStatusFromCommandResult(result);
     }
diff --git a/src/mongo/db/commands.h b/src/mongo/db/commands.h
@@ -48,8 +48,8 @@ namespace mongo {
     class BSONObjBuilder;
     class Client;
     class Database;
-    class Timer;
     class OperationContext;
+    class Timer;
 
 namespace mutablebson {
     class Document;
@@ -315,6 +315,12 @@ namespace mutablebson {
                                                BSONArray firstBatch,
                                                BSONObjBuilder* builder);
 
+        /**
+         * Helper for setting a writeConcernError field in the command result object if
+         * a writeConcern error occurs.
+         */
+        static void appendCommandWCStatus(BSONObjBuilder& result, const Status& status);
+
         // Set by command line.  Controls whether or not testing-only commands should be available.
         static int testCommandsEnabled;
 
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
@@ -47,6 +47,7 @@
 #include "mongo/db/ops/update_lifecycle_impl.h"
 #include "mongo/db/query/get_executor.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
+#include "mongo/db/write_concern.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -114,6 +115,12 @@ namespace mongo {
                 return false;
             }
 
+            StatusWith<WriteConcernOptions> wcResult = extractWriteConcern(cmdObj);
+            if (!wcResult.isOK()) {
+                return appendCommandStatus(result, wcResult.getStatus());
+            }
+            setupSynchronousCommit(wcResult.getValue(), txn);
+
             bool ok = false;
             MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
                 errmsg = "";
@@ -175,6 +182,13 @@ namespace mongo {
                              errmsg);
             }
 
+            WriteConcernResult res;
+            wcResult = waitForWriteConcern(txn,
+                                           wcResult.getValue(),
+                                           txn->getClient()->getLastOp(),
+                                           &res);
+            appendCommandWCStatus(result, wcResult.getStatus());
+
             return ok;
         }
 
diff --git a/src/mongo/db/commands/write_commands/batch_executor.cpp b/src/mongo/db/commands/write_commands/batch_executor.cpp
@@ -756,14 +756,6 @@ namespace mongo {
         Collection* _collection;
     };
 
-    void setupSynchronousCommit( const WriteConcernOptions& writeConcern,
-                                 OperationContext* txn ) {
-        if ( writeConcern.syncMode == WriteConcernOptions::JOURNAL ||
-             writeConcern.syncMode == WriteConcernOptions::FSYNC ) {
-            txn->recoveryUnit()->goingToAwaitCommit();
-        }
-    }
-
     void WriteBatchExecutor::bulkExecute( const BatchedCommandRequest& request,
                                           const WriteConcernOptions& writeConcern,
                                           std::vector<BatchedUpsertDetail*>* upsertedIds,
diff --git a/src/mongo/db/write_concern.cpp b/src/mongo/db/write_concern.cpp
@@ -31,6 +31,7 @@
 #include "mongo/db/write_concern.h"
 
 #include "mongo/base/counter.h"
+#include "mongo/bson/util/bson_extract.h"
 #include "mongo/db/commands/server_status_metric.h"
 #include "mongo/db/global_environment_experiment.h"
 #include "mongo/db/operation_context.h"
@@ -52,6 +53,41 @@ namespace mongo {
     static ServerStatusMetricField<Counter64> gleWtimeoutsDisplay("getLastError.wtimeouts",
                                                                   &gleWtimeouts );
 
+    void setupSynchronousCommit(const WriteConcernOptions& writeConcern,
+                                OperationContext* txn) {
+        if ( writeConcern.syncMode == WriteConcernOptions::JOURNAL ||
+             writeConcern.syncMode == WriteConcernOptions::FSYNC ) {
+            txn->recoveryUnit()->goingToAwaitCommit();
+        }
+    }
+
+    StatusWith<WriteConcernOptions> extractWriteConcern(const BSONObj& cmdObj) {
+        BSONElement writeConcernElement;
+        Status wcStatus = bsonExtractTypedField(cmdObj,
+                                                "writeConcern",
+                                                Object,
+                                                &writeConcernElement);
+
+        if (!wcStatus.isOK()) {
+            if (wcStatus == ErrorCodes::NoSuchKey) {
+                return repl::getGlobalReplicationCoordinator()->getGetLastErrorDefault();
+            }
+            return wcStatus;
+        }
+
+        WriteConcernOptions writeConcern;
+        wcStatus = writeConcern.parse(writeConcernElement.Obj());
+
+        if (wcStatus.isOK()) {
+            wcStatus = validateWriteConcern(writeConcern);
+        }
+        if (!wcStatus.isOK()) {
+            return wcStatus;
+        }
+
+        return writeConcern;
+    }
+
     Status validateWriteConcern( const WriteConcernOptions& writeConcern ) {
         const bool isJournalEnabled = getGlobalEnvironment()->getGlobalStorageEngine()->isDurable();
 
diff --git a/src/mongo/db/write_concern.h b/src/mongo/db/write_concern.h
@@ -34,6 +34,23 @@
 namespace mongo {
 
     class OperationContext;
+    template <typename T> class StatusWith;
+
+    /**
+     * If "writeConcern" indicates a durable commit level,
+     * marks the RecoveryUnit associated with "txn" appropriately.
+     * Provides a hint to the storage engine that
+     * particular operations will be waiting for their changes to become durable.
+     */
+    void setupSynchronousCommit(const WriteConcernOptions& writeConcern,
+                                OperationContext* txn);
+
+    /**
+     * Attempts to extract a writeConcern from cmdObj.
+     * Verifies that the writeConcern is of type Object (BSON type) and
+     * that the resulting writeConcern is valid for this particular host.
+     */
+    StatusWith<WriteConcernOptions> extractWriteConcern(const BSONObj& cmdObj);
 
     /**
      * Verifies that a WriteConcern is valid for this particular host.
