SERVER-42192 Enable moveChunk FSM workloads to run in stepdown suites.

Adds automatic retry logic to ChunkHelper.moveChunk() to handle when the
CSRS or replica set shard primary being killed, terminated, or stepped
down leads to the moveChunk command being interrupted.

Exposes replica set connections as part of the "connection cache" so
that DBClientRS may be used to track the current primary of the CSRS or
replica set shard.

Introduces an fsm.forceRunningOutsideTransaction() utility function to
prevent a state function from running inside a multi-statement
transaction as part of the concurrency_*_multi_stmt_txn*.yml test
suites.

2 files changed, 51 insertions, 4 deletions

diff --git a/jstests/concurrency/fsm_libs/fsm.js b/jstests/concurrency/fsm_libs/fsm.js
index 3aa73477425..e3b8fc0c16b 100644
--- a/jstests/concurrency/fsm_libs/fsm.js
+++ b/jstests/concurrency/fsm_libs/fsm.js
@@ -1,6 +1,18 @@
 'use strict';
 
 var fsm = (function() {
+    const kIsRunningInsideTransaction = Symbol('isRunningInsideTransaction');
+
+    function forceRunningOutsideTransaction(data) {
+        if (data[kIsRunningInsideTransaction]) {
+            const err =
+                new Error('Intentionally thrown to stop state function from running inside of a' +
+                          ' multi-statement transaction');
+            err.isNotSupported = true;
+            throw err;
+        }
+    }
+
     // args.data = 'this' object of the state functions
     // args.db = database object
     // args.collName = collection name
@@ -9,6 +21,7 @@ var fsm = (function() {
     // args.startState = name of initial state function
     // args.states = state functions of the form
     //               { stateName: function(db, collName) { ... } }
+    // args.tid = the thread identifier
     // args.transitions = transitions between state functions of the form
     //                    { stateName: { nextState1: probability,
     //                                   nextState2: ... } }
@@ -40,14 +53,41 @@ var fsm = (function() {
                 return conn;
             };
 
-            connCache = {mongos: [], config: [], shards: {}};
+            const getReplSetName = (conn) => {
+                const res = assert.commandWorked(conn.getDB('admin').runCommand({isMaster: 1}));
+                assert.eq('string',
+                          typeof res.setName,
+                          () => `not connected to a replica set: ${tojson(res)}`);
+                return res.setName;
+            };
+
+            const makeReplSetConnWithExistingSession = (connStrList, replSetName) => {
+                const conn = makeNewConnWithExistingSession(`mongodb://${
+                    connStrList.join(',')}/?appName=tid:${args.tid}&replicaSet=${replSetName}`);
+
+                return conn;
+            };
+
+            connCache =
+                {mongos: [], config: [], shards: {}, rsConns: {config: undefined, shards: {}}};
             connCache.mongos = args.cluster.mongos.map(makeNewConnWithExistingSession);
             connCache.config = args.cluster.config.map(makeNewConnWithExistingSession);
+            connCache.rsConns.config = makeReplSetConnWithExistingSession(
+                args.cluster.config, getReplSetName(connCache.config[0]));
+
+            // We set _isConfigServer=true on the Mongo connection object so
+            // set_read_preference_secondary.js knows to avoid overriding the read preference as the
+            // concurrency suite may be running with a 1-node CSRS.
+            connCache.rsConns.config._isConfigServer = true;
 
             var shardNames = Object.keys(args.cluster.shards);
 
-            shardNames.forEach(name => (connCache.shards[name] = args.cluster.shards[name].map(
-                                            makeNewConnWithExistingSession)));
+            shardNames.forEach(name => {
+                connCache.shards[name] =
+                    args.cluster.shards[name].map(makeNewConnWithExistingSession);
+                connCache.rsConns.shards[name] = makeReplSetConnWithExistingSession(
+                    args.cluster.shards[name], getReplSetName(connCache.shards[name][0]));
+            });
         }
 
         for (var i = 0; i < args.iterations; ++i) {
@@ -63,8 +103,10 @@ var fsm = (function() {
                     let data;
                     withTxnAndAutoRetry(args.db.getSession(), () => {
                         data = TransactionsUtil.deepCopyObject({}, args.data);
+                        data[kIsRunningInsideTransaction] = true;
                         fn.call(data, args.db, args.collName, connCache);
                     });
+                    delete data[kIsRunningInsideTransaction];
                     args.data = data;
                 } catch (e) {
                     // Retry state functions that threw OperationNotSupportedInTransaction or
@@ -128,5 +170,9 @@ var fsm = (function() {
         assert(false, 'not reached');
     }
 
-    return {run: runFSM, _getWeightedRandomChoice: getWeightedRandomChoice};
+    return {
+        forceRunningOutsideTransaction,
+        run: runFSM,
+        _getWeightedRandomChoice: getWeightedRandomChoice,
+    };
 })();
diff --git a/jstests/concurrency/fsm_libs/worker_thread.js b/jstests/concurrency/fsm_libs/worker_thread.js
index 58eeed3e66f..7e237e6257b 100644
--- a/jstests/concurrency/fsm_libs/worker_thread.js
+++ b/jstests/concurrency/fsm_libs/worker_thread.js
@@ -205,6 +205,7 @@ var workerThread = (function() {
                     passConnectionCache: config.passConnectionCache,
                     startState: config.startState,
                     states: config.states,
+                    tid: args.tid,
                     transitions: config.transitions
                 };
             });