SERVER-27123 Only update the commit point as a secondary from oplog queries against your sync source

6 files changed, 176 insertions, 44 deletions

diff --git a/jstests/replsets/double_rollback.js b/jstests/replsets/double_rollback.js
index 42aa40fae6a..342775c2545 100644
--- a/jstests/replsets/double_rollback.js
+++ b/jstests/replsets/double_rollback.js
@@ -12,7 +12,9 @@
 
 (function() {
     'use strict';
+
     load("jstests/libs/check_log.js");
+    load("jstests/replsets/rslib.js");
 
     var name = "double_rollback";
     var dbName = "test";
@@ -32,14 +34,6 @@
     var nodes = rst.startSet();
     rst.initiate();
 
-    function waitForState(node, state) {
-        assert.soonNoExcept(function() {
-            assert.commandWorked(node.adminCommand(
-                {replSetTest: 1, waitForMemberState: state, timeoutMillis: rst.kDefaultTimeoutMS}));
-            return true;
-        });
-    }
-
     function stepUp(rst, node) {
         var primary = rst.getPrimary();
         if (primary != node) {
diff --git a/jstests/replsets/read_committed_after_rollback.js b/jstests/replsets/read_committed_after_rollback.js
index e434483f99b..033a83ad2d9 100644
--- a/jstests/replsets/read_committed_after_rollback.js
+++ b/jstests/replsets/read_committed_after_rollback.js
@@ -143,10 +143,10 @@ load("jstests/replsets/rslib.js");  // For startSetIfSupportsReadMajority.
     // now be visible as a committed read to both oldPrimary and newPrimary.
     assert.commandWorked(
         pureSecondary.adminCommand({configureFailPoint: "rsSyncApplyStop", mode: "off"}));
-    var gleResponse =
-        newPrimaryColl.runCommand({getLastError: 1, w: 'majority', wtimeout: 5 * 1000 * 60});
-    assert.commandWorked(gleResponse);
-    assert.eq(null, gleResponse.err, "GLE detected write error: " + tojson(gleResponse));
+    // Do a write to the new primary so that the old primary can establish a sync source to learn
+    // about the new commit.
+    assert.writeOK(newPrimary.getDB(name).unrelatedCollection.insert(
+        {a: 1}, {writeConcern: {w: 'majority', wtimeout: replTest.kDefaultTimeoutMS}}));
     assert.eq(doCommittedRead(newPrimaryColl), 'new');
     assert.eq(doCommittedRead(oldPrimaryColl), 'new');
 }());
diff --git a/jstests/replsets/read_committed_stale_history.js b/jstests/replsets/read_committed_stale_history.js
new file mode 100644
index 00000000000..9793a47576e
--- /dev/null
+++ b/jstests/replsets/read_committed_stale_history.js
@@ -0,0 +1,148 @@
+/*
+ * Tests that a node on a stale branch of history won't incorrectly mark its ops as committed even
+ * when hearing about a commit point with a higher optime from a new primary.
+ */
+(function() {
+    'use strict';
+
+    load("jstests/libs/check_log.js");
+    load("jstests/libs/write_concern_util.js");
+    load("jstests/replsets/rslib.js");
+
+    var name = "readCommittedStaleHistory";
+    var dbName = "wMajorityCheck";
+    var collName = "stepdown";
+
+    var rst = new ReplSetTest({
+        name: name,
+        nodes: [
+            {},
+            {},
+            {rsConfig: {priority: 0}},
+        ],
+        nodeOptions: {enableMajorityReadConcern: ""},
+        useBridge: true
+    });
+
+    if (!startSetIfSupportsReadMajority(rst)) {
+        jsTest.log("skipping test since storage engine doesn't support committed reads");
+        return;
+    }
+
+    var nodes = rst.nodes;
+    rst.initiate();
+
+    /**
+     * Waits for the given node to be in state primary *and* have finished drain mode and thus
+     * be available for writes.
+     */
+    function waitForPrimary(node) {
+        assert.soon(function() {
+            return node.adminCommand('ismaster').ismaster;
+        });
+    }
+
+    function stepUp(node) {
+        var primary = rst.getPrimary();
+        if (primary != node) {
+            assert.throws(function() {
+                primary.adminCommand({replSetStepDown: 60 * 5});
+            });
+        }
+        waitForPrimary(node);
+    }
+
+    // Asserts that the given document is not visible in the committed snapshot on the given node.
+    function checkDocNotCommitted(node, doc) {
+        var docs =
+            node.getDB(dbName).getCollection(collName).find(doc).readConcern('majority').toArray();
+        assert.eq(0, docs.length, tojson(docs));
+    }
+
+    jsTestLog("Make sure node 0 is primary.");
+    rst.getPrimary();
+    stepUp(nodes[0]);
+    var primary = rst.getPrimary();
+    var secondaries = rst.getSecondaries();
+    assert.eq(nodes[0], primary);
+    // Wait for all data bearing nodes to get up to date.
+    assert.writeOK(nodes[0].getDB(dbName).getCollection(collName).insert(
+        {a: 1}, {writeConcern: {w: 3, wtimeout: rst.kDefaultTimeoutMS}}));
+
+    // Stop the secondaries from replicating.
+    stopServerReplication(secondaries);
+    // Stop the primary from being able to complete stepping down.
+    assert.commandWorked(
+        nodes[0].adminCommand({configureFailPoint: 'blockHeartbeatStepdown', mode: 'alwaysOn'}));
+
+    jsTestLog("Do a write that won't ever reach a majority of nodes");
+    assert.writeOK(nodes[0].getDB(dbName).getCollection(collName).insert({a: 2}));
+
+    // Ensure that the write that was just done is not visible in the committed snapshot.
+    checkDocNotCommitted(nodes[0], {a: 2});
+
+    // Prevent the primary from rolling back later on.
+    assert.commandWorked(
+        nodes[0].adminCommand({configureFailPoint: 'rollbackHangBeforeStart', mode: 'alwaysOn'}));
+
+    jsTest.log("Disconnect primary from all secondaries");
+    nodes[0].disconnect(nodes[1]);
+    nodes[0].disconnect(nodes[2]);
+
+    jsTest.log("Wait for a new primary to be elected");
+    // Allow the secondaries to replicate again.
+    restartServerReplication(secondaries);
+
+    waitForPrimary(nodes[1]);
+
+    jsTest.log("Do a write to the new primary");
+    assert.writeOK(nodes[1].getDB(dbName).getCollection(collName).insert(
+        {a: 3}, {writeConcern: {w: 2, wtimeout: rst.kDefaultTimeoutMS}}));
+
+    jsTest.log("Reconnect the old primary to the rest of the nodes");
+    nodes[1].reconnect(nodes[0]);
+    nodes[2].reconnect(nodes[0]);
+
+    // Sleep 10 seconds to allow some heartbeats to be processed, so we can verify that the
+    // heartbeats don't cause the stale primary to incorrectly advance the commit point.
+    sleep(10000);
+
+    checkDocNotCommitted(nodes[0], {a: 2});
+
+    jsTest.log("Allow the old primary to finish stepping down and become secondary");
+    var res = null;
+    try {
+        res = nodes[0].adminCommand({configureFailPoint: 'blockHeartbeatStepdown', mode: 'off'});
+    } catch (e) {
+        // Expected - once we disable the fail point the stepdown will proceed and it's racy whether
+        // the stepdown closes all connections before or after the configureFailPoint command
+        // returns
+    }
+    if (res) {
+        assert.commandWorked(res);
+    }
+    rst.waitForState(nodes[0], ReplSetTest.State.SECONDARY);
+
+    // At this point the former primary will attempt to go into rollback, but the
+    // 'rollbackHangBeforeStart' will prevent it from doing so.
+    checkDocNotCommitted(nodes[0], {a: 2});
+    checkLog.contains(nodes[0], 'rollback - rollbackHangBeforeStart fail point enabled');
+    checkDocNotCommitted(nodes[0], {a: 2});
+
+    jsTest.log("Allow the original primary to roll back its write and catch up to the new primary");
+    assert.commandWorked(
+        nodes[0].adminCommand({configureFailPoint: 'rollbackHangBeforeStart', mode: 'off'}));
+
+    assert.soonNoExcept(function() {
+        return null == nodes[0].getDB(dbName).getCollection(collName).findOne({a: 2});
+    }, "Original primary never rolled back its write");
+
+    rst.awaitReplication();
+
+    // Ensure that the old primary got the write that the new primary did and sees it as committed.
+    assert.neq(
+        null,
+        nodes[0].getDB(dbName).getCollection(collName).find({a: 3}).readConcern('majority').next());
+
+    rst.stopSet();
+}());
diff --git a/jstests/replsets/rslib.js b/jstests/replsets/rslib.js
index beebf99159a..03ce30de2ab 100644
--- a/jstests/replsets/rslib.js
+++ b/jstests/replsets/rslib.js
@@ -7,6 +7,7 @@ var reconfig;
 var awaitOpTime;
 var startSetIfSupportsReadMajority;
 var waitUntilAllNodesCaughtUp;
+var waitForState;
 var reInitiateWithoutThrowingOnAbortedMember;
 var awaitRSClientHosts;
 var getLastOpTime;
@@ -212,6 +213,17 @@ var getLastOpTime;
     };
 
     /**
+     * Waits for the given node to reach the given state, ignoring network errors.
+     */
+    waitForState = function(node, state) {
+        assert.soonNoExcept(function() {
+            assert.commandWorked(node.adminCommand(
+                {replSetTest: 1, waitForMemberState: state, timeoutMillis: 60 * 1000 * 5}));
+            return true;
+        });
+    };
+
+    /**
      * Starts each node in the given replica set if the storage engine supports readConcern
      *'majority'.
      * Returns true if the replica set was started successfully and false otherwise.
diff --git a/jstests/replsets/write_concern_after_stepdown.js b/jstests/replsets/write_concern_after_stepdown.js
index c8493ea4fb6..95a08a005e0 100644
--- a/jstests/replsets/write_concern_after_stepdown.js
+++ b/jstests/replsets/write_concern_after_stepdown.js
@@ -5,6 +5,9 @@
 (function() {
     'use strict';
 
+    load("jstests/replsets/rslib.js");
+    load("jstests/libs/write_concern_util.js");
+
     var name = "writeConcernStepDownAndBackUp";
     var dbName = "wMajorityCheck";
     var collName = "stepdownAndBackUp";
@@ -21,14 +24,6 @@
     var nodes = rst.startSet();
     rst.initiate();
 
-    function waitForState(node, state) {
-        assert.soonNoExcept(function() {
-            assert.commandWorked(node.adminCommand(
-                {replSetTest: 1, waitForMemberState: state, timeoutMillis: rst.kDefaultTimeoutMS}));
-            return true;
-        });
-    }
-
     function waitForPrimary(node) {
         assert.soon(function() {
             return node.adminCommand('ismaster').ismaster;
@@ -55,10 +50,7 @@
         {a: 1}, {writeConcern: {w: 3, wtimeout: rst.kDefaultTimeoutMS}}));
 
     // Stop the secondaries from replicating.
-    secondaries.forEach(function(node) {
-        assert.commandWorked(
-            node.adminCommand({configureFailPoint: 'rsSyncApplyStop', mode: 'alwaysOn'}));
-    });
+    stopServerReplication(secondaries);
     // Stop the primary from being able to complete stepping down.
     assert.commandWorked(
         nodes[0].adminCommand({configureFailPoint: 'blockHeartbeatStepdown', mode: 'alwaysOn'}));
@@ -79,10 +71,7 @@
 
     jsTest.log("Wait for a new primary to be elected");
     // Allow the secondaries to replicate again.
-    secondaries.forEach(function(node) {
-        assert.commandWorked(
-            node.adminCommand({configureFailPoint: 'rsSyncApplyStop', mode: 'off'}));
-    });
+    restartServerReplication(secondaries);
 
     waitForPrimary(nodes[1]);
 
diff --git a/jstests/replsets/write_concern_after_stepdown_and_stepup.js b/jstests/replsets/write_concern_after_stepdown_and_stepup.js
index a5a0f0b6d64..e2667d4f66a 100644
--- a/jstests/replsets/write_concern_after_stepdown_and_stepup.js
+++ b/jstests/replsets/write_concern_after_stepdown_and_stepup.js
@@ -6,6 +6,9 @@
 (function() {
     'use strict';
 
+    load("jstests/replsets/rslib.js");
+    load("jstests/libs/write_concern_util.js");
+
     var name = "writeConcernStepDownAndBackUp";
     var dbName = "wMajorityCheck";
     var collName = "stepdownAndBackUp";
@@ -22,14 +25,6 @@
     var nodes = rst.startSet();
     rst.initiate();
 
-    function waitForState(node, state) {
-        assert.soonNoExcept(function() {
-            assert.commandWorked(node.adminCommand(
-                {replSetTest: 1, waitForMemberState: state, timeoutMillis: rst.kDefaultTimeoutMS}));
-            return true;
-        });
-    }
-
     function waitForPrimary(node) {
         assert.soon(function() {
             return node.adminCommand('ismaster').ismaster;
@@ -56,10 +51,7 @@
         {a: 1}, {writeConcern: {w: 3, wtimeout: rst.kDefaultTimeoutMS}}));
 
     // Stop the secondaries from replicating.
-    secondaries.forEach(function(node) {
-        assert.commandWorked(
-            node.adminCommand({configureFailPoint: 'stopOplogFetcher', mode: 'alwaysOn'}));
-    });
+    stopServerReplication(secondaries);
     // Stop the primary from calling into awaitReplication()
     assert.commandWorked(nodes[0].adminCommand(
         {configureFailPoint: 'hangBeforeWaitingForWriteConcern', mode: 'alwaysOn'}));
@@ -86,10 +78,7 @@
 
     jsTest.log("Wait for a new primary to be elected");
     // Allow the secondaries to replicate again.
-    secondaries.forEach(function(node) {
-        assert.commandWorked(
-            node.adminCommand({configureFailPoint: 'stopOplogFetcher', mode: 'off'}));
-    });
+    restartServerReplication(secondaries);
 
     waitForPrimary(nodes[1]);