SERVER-7551 do not convert updates to upserts if primary is pre-2.2

Converting updates to upserts during replication was added for version 2.2.0.
This code was dependent on all operators being idempotent.
Some operators in 2.0.x are not fully idempotent, so if we are reading an
oplog generated by a 2.0.x node, we shouldn't activate this new feature.

2 files changed, 27 insertions, 2 deletions

diff --git a/src/mongo/db/repl/rs_sync.cpp b/src/mongo/db/repl/rs_sync.cpp
index 90d3a252754..7c49b89ad70 100644
--- a/src/mongo/db/repl/rs_sync.cpp
+++ b/src/mongo/db/repl/rs_sync.cpp
@@ -109,11 +109,16 @@ namespace replset {
     // This free function is used by the writer threads to apply each op
     void multiSyncApply(const std::vector<BSONObj>& ops, SyncTail* st) {
         initializeWriterThread();
+
+        // convert update operations only for 2.2.1 or greater, because we need guaranteed
+        // idempotent operations for this to work.  See SERVER-6825
+        bool convertUpdatesToUpserts = theReplSet->oplogVersion > 1 ? true : false;
+
         for (std::vector<BSONObj>::const_iterator it = ops.begin();
              it != ops.end();
              ++it) {
             try {
-                fassert(16359, st->syncApply(*it, true));
+                fassert(16359, st->syncApply(*it, convertUpdatesToUpserts));
             } catch (DBException& e) {
                 error() << "writer worker caught exception: " << e.what() 
                         << " on: " << it->toString() << endl;
@@ -420,7 +425,7 @@ namespace replset {
         if (!peek_success) {
             // if we don't have anything in the queue, wait a bit for something to appear
             if (ops->empty()) {
-                // block 1 second
+                // block up to 1 second
                 _networkQueue->waitForMore();
                 return false;
             }
diff --git a/src/mongo/dbtests/replsettests.cpp b/src/mongo/dbtests/replsettests.cpp
index 4711e098f20..c11114f2339 100644
--- a/src/mongo/dbtests/replsettests.cpp
+++ b/src/mongo/dbtests/replsettests.cpp
@@ -466,6 +466,18 @@ namespace ReplSetTests {
                     "timestamp" << 1334810820))), &id);
         }
 
+        void addConflictingUpdates() {
+            BSONObj first = BSON("_id" << "asdfasdfasdf");
+            addOp("i", first);
+
+            BSONObj filter = BSON("_id" << "asdfasdfasdf" << "sp" << BSON("$size" << 2));
+            // Test an op with no version, op is ignored and replication continues (code assumes 
+            // version 1)
+            addOp("u", BSON("$push" << BSON("sp" << 42)), &filter, NULL, NULL);
+            // The following line generates an fassert because it's version 2
+            //addOp("u", BSON("$push" << BSON("sp" << 42)), &filter, NULL, 2);
+        }
+
         void addUniqueIndex() {
             addOp("i", BSON("ns" << ns() << "key" << BSON("x" << 1) << "name" << "x1" << "unique" << true), 0, "unittests.system.indexes");
             addInserts(2);
@@ -501,6 +513,14 @@ namespace ReplSetTests {
             ASSERT_EQUALS(1334810820, obj["requests"]["100002_1"]["timestamp"].number());
 
             drop();
+            
+            // test converting updates to upserts but only for version 2.2.1 and greater,
+            // which means oplog version 2 and greater.
+            addConflictingUpdates();
+            applyOplog();
+
+            drop();
+
         }
     };