text: /**
   *
   * Test read committed functionality following a following a rollback. Currently we require that all
   * snapshots be dropped during rollback, therefore committed reads will block until a new committed
   * snapshot is available.
   */
  
  load("jstests/replsets/rslib.js");  // For startSetIfSupportsReadMajority.
  
  (function() {
      "use strict";
  
      function assertCommittedReadsBlock(coll) {
          var res =
              coll.runCommand('find', {"readConcern": {"level": "majority"}, "maxTimeMS": 3000});
          assert.commandFailedWithCode(
              res,
              ErrorCodes.ExceededTimeLimit,
              "Expected read of " + coll.getFullName() + ' on ' + coll.getMongo().host + " to block");
      }
  
      function doCommittedRead(coll) {
          var res =
              coll.runCommand('find', {"readConcern": {"level": "majority"}, "maxTimeMS": 10000});
          assert.commandWorked(res,
                               'reading from ' + coll.getFullName() + ' on ' + coll.getMongo().host);
          return new DBCommandCursor(coll.getMongo(), res).toArray()[0].state;
      }
  
      function doDirtyRead(coll) {
          var res = coll.runCommand('find', {"readConcern": {"level": "local"}});
          assert.commandWorked(res,
                               'reading from ' + coll.getFullName() + ' on ' + coll.getMongo().host);
          return new DBCommandCursor(coll.getMongo(), res).toArray()[0].state;
      }
  
      // Set up a set and grab things for later.
      var name = "read_committed_after_rollback";
      var replTest = new ReplSetTest(
          {name: name, nodes: 5, useBridge: true, nodeOptions: {enableMajorityReadConcern: ''}});
  
      if (!startSetIfSupportsReadMajority(replTest)) {
          jsTest.log("skipping test since storage engine doesn't support committed reads");
          return;
      }
  
      var nodes = replTest.nodeList();
      var config = {
          "_id": name,
          "members": [
              {"_id": 0, "host": nodes[0]},
              {"_id": 1, "host": nodes[1]},
              {"_id": 2, "host": nodes[2], priority: 0},
              // Note: using two arbiters to ensure that a host that can't talk to any other
              // data-bearing node can still be elected. This also means that a write isn't considered
              // committed until it is on all 3 data-bearing nodes, not just 2.
              {"_id": 3, "host": nodes[3], arbiterOnly: true},
              {"_id": 4, "host": nodes[4], arbiterOnly: true},
          ]
      };
  
      replTest.initiate(config);
  
      // Get connections.
      var oldPrimary = replTest.getPrimary();
      var newPrimary = replTest.liveNodes.slaves[0];
      var pureSecondary = replTest.liveNodes.slaves[1];
      var arbiters = [replTest.nodes[3], replTest.nodes[4]];
  
      // This is the collection that all of the tests will use.
      var collName = name + '.collection';
      var oldPrimaryColl = oldPrimary.getCollection(collName);
      var newPrimaryColl = newPrimary.getCollection(collName);
  
      // Set up initial state.
      assert.writeOK(oldPrimaryColl.insert({_id: 1, state: 'old'},
                                           {writeConcern: {w: 'majority', wtimeout: 30000}}));
      assert.eq(doDirtyRead(oldPrimaryColl), 'old');
      assert.eq(doCommittedRead(oldPrimaryColl), 'old');
      assert.eq(doDirtyRead(newPrimaryColl), 'old');
      // Note that we can't necessarily do a committed read from newPrimaryColl and get 'old', since
      // delivery of the commit level to secondaries isn't synchronized with anything
      // (we would have to hammer to reliably prove that it eventually would work).
  
      // Partition the world such that oldPrimary is still primary but can't replicate to anyone.
      // newPrimary is disconnected from the arbiters first to ensure that it can't be elected.
      newPrimary.disconnect(arbiters);
      oldPrimary.disconnect([newPrimary, pureSecondary]);
      assert.eq(doDirtyRead(newPrimaryColl), 'old');
  
      // This write will only make it to oldPrimary and will never become committed.
      assert.writeOK(oldPrimaryColl.save({_id: 1, state: 'INVALID'}));
      assert.eq(doDirtyRead(oldPrimaryColl), 'INVALID');
      assert.eq(doCommittedRead(oldPrimaryColl), 'old');
  
      // Change the partitioning so that oldPrimary is isolated, and newPrimary can be elected.
      oldPrimary.setSlaveOk();
      oldPrimary.disconnect(arbiters);
      newPrimary.reconnect(arbiters);
      assert.soon(() => newPrimary.adminCommand('isMaster').ismaster, '', 60 * 1000);
      assert.soon(function() {
          try {
              return !oldPrimary.adminCommand('isMaster').ismaster;
          } catch (e) {
              return false;  // ignore disconnect errors.
          }
      });
  
      // Stop applier on pureSecondary to ensure that writes to newPrimary won't become committed yet.
      assert.commandWorked(
          pureSecondary.adminCommand({configureFailPoint: "rsSyncApplyStop", mode: "alwaysOn"}));
      assert.writeOK(newPrimaryColl.save({_id: 1, state: 'new'}));
      assert.eq(doDirtyRead(newPrimaryColl), 'new');
      // Note that we still can't do a committed read from the new primary and reliably get anything,
      // since we never proved that it learned about the commit level from the old primary before
      // the new primary got elected.  The new primary cannot advance the commit level until it
      // commits a write in its own term.  This includes learning that a majority of nodes have
      // received such a write.
      assert.eq(doCommittedRead(oldPrimaryColl), 'old');
  
      // Reconnect oldPrimary to newPrimary, inducing rollback of the 'INVALID' write. This causes
      // oldPrimary to drop all snapshots. oldPrimary still won't be connected to enough hosts to
      // allow it to be elected, so newPrimary should stay primary for the rest of this test.
      oldPrimary.reconnect(newPrimary);
      assert.soon(function() {
          try {
              return oldPrimary.adminCommand('isMaster').secondary &&
                  doDirtyRead(oldPrimaryColl) == 'new';
          } catch (e) {
              return false;  // ignore disconnect errors.
          }
      }, '', 60 * 1000);
      assert.eq(doDirtyRead(oldPrimaryColl), 'new');
      assertCommittedReadsBlock(oldPrimaryColl);
  
      // Try asserts again after sleeping to make sure state doesn't change while pureSecondary isn't
      // replicating.
      sleep(1000);
      assert.eq(doDirtyRead(oldPrimaryColl), 'new');
      assertCommittedReadsBlock(oldPrimaryColl);
  
      // Resume oplog application on pureSecondary to allow the 'new' write to be committed. It should
      // now be visible as a committed read to both oldPrimary and newPrimary.
      assert.commandWorked(
          pureSecondary.adminCommand({configureFailPoint: "rsSyncApplyStop", mode: "off"}));
      // 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');
  }());
  