SERVER-66984 config.transactions collection may end up not having the "parentLsid" partial index if the primary steps down or shuts down while setting up the collection

(cherry picked from commit f9a3cf210791c19cc2d8569d0a7837745838d0cb)

3 files changed, 180 insertions, 54 deletions

diff --git a/jstests/multiVersion/internal_transactions_index_setFCV.js b/jstests/multiVersion/internal_transactions_index_setFCV.js
index 783dcaace3f..712e8249a54 100644
--- a/jstests/multiVersion/internal_transactions_index_setFCV.js
+++ b/jstests/multiVersion/internal_transactions_index_setFCV.js
@@ -7,6 +7,11 @@
 (function() {
 "use strict";
 
+// This test does direct writes to config.transactions which requires not using a session.
+TestData.disableImplicitSessions = true;
+
+load("jstests/replsets/rslib.js");
+
 // Verifies both the _id index and partial parent_lsid index exists for config.transactions.
 function assertPartialIndexExists(node) {
     const configDB = node.getDB("config");
@@ -36,7 +41,7 @@ function assertPartialIndexDoesNotExist(node) {
  * Verifies the partial index is dropped/created on FCV transitions and retryable writes work in all
  * FCVs.
  */
-function runTest(setFCVConn, modifyIndexConns, verifyIndexConns) {
+function runTest(setFCVConn, modifyIndexConns, verifyIndexConns, rst) {
     // Start at latest FCV which should have the index.
     assert.commandWorked(setFCVConn.adminCommand({setFeatureCompatibilityVersion: latestFCV}));
     verifyIndexConns.forEach(conn => {
@@ -49,6 +54,25 @@ function runTest(setFCVConn, modifyIndexConns, verifyIndexConns) {
         assertPartialIndexDoesNotExist(conn);
     });
 
+    if (rst) {
+        // On step up to primary the index should not be created.
+
+        let primary = rst.getPrimary();
+        // Clear the collection so we'd try to create the index.
+        assert.commandWorked(primary.getDB("config").transactions.remove({}));
+        assert.commandWorked(
+            primary.adminCommand({replSetStepDown: ReplSetTest.kForeverSecs, force: true}));
+        assert.commandWorked(primary.adminCommand({replSetFreeze: 0}));
+
+        setFCVConn = rst.getPrimary();
+        modifyIndexConns = [rst.getPrimary()];
+        rst.awaitReplication();
+        verifyIndexConns.forEach(conn => {
+            reconnect(conn);
+            assertPartialIndexDoesNotExist(conn);
+        });
+    }
+
     assert.commandWorked(setFCVConn.getDB("foo").runCommand(
         {insert: "bar", documents: [{x: 1}], lsid: {id: UUID()}, txnNumber: NumberLong(11)}));
 
@@ -68,6 +92,25 @@ function runTest(setFCVConn, modifyIndexConns, verifyIndexConns) {
         assertPartialIndexDoesNotExist(conn);
     });
 
+    if (rst) {
+        // On step up to primary the index should not be created.
+
+        let primary = rst.getPrimary();
+        // Clear the collection so we'd try to create the index.
+        assert.commandWorked(primary.getDB("config").transactions.remove({}));
+        assert.commandWorked(
+            primary.adminCommand({replSetStepDown: ReplSetTest.kForeverSecs, force: true}));
+        assert.commandWorked(primary.adminCommand({replSetFreeze: 0}));
+
+        setFCVConn = rst.getPrimary();
+        modifyIndexConns = [rst.getPrimary()];
+        rst.awaitReplication();
+        verifyIndexConns.forEach(conn => {
+            reconnect(conn);
+            assertPartialIndexDoesNotExist(conn);
+        });
+    }
+
     assert.commandWorked(setFCVConn.getDB("foo").runCommand(
         {insert: "bar", documents: [{x: 1}], lsid: {id: UUID()}, txnNumber: NumberLong(11)}));
 
@@ -77,6 +120,25 @@ function runTest(setFCVConn, modifyIndexConns, verifyIndexConns) {
         assertPartialIndexExists(conn);
     });
 
+    if (rst) {
+        // On step up to primary the index should be created.
+
+        let primary = rst.getPrimary();
+        // Clear the collection so we'll try to create the index.
+        assert.commandWorked(primary.getDB("config").transactions.remove({}));
+        assert.commandWorked(
+            primary.adminCommand({replSetStepDown: ReplSetTest.kForeverSecs, force: true}));
+        assert.commandWorked(primary.adminCommand({replSetFreeze: 0}));
+
+        setFCVConn = rst.getPrimary();
+        modifyIndexConns = [rst.getPrimary()];
+        rst.awaitReplication();
+        verifyIndexConns.forEach(conn => {
+            reconnect(conn);
+            assertPartialIndexExists(conn);
+        });
+    }
+
     assert.commandWorked(setFCVConn.getDB("foo").runCommand(
         {insert: "bar", documents: [{x: 1}], lsid: {id: UUID()}, txnNumber: NumberLong(11)}));
 
@@ -126,7 +188,7 @@ function runTest(setFCVConn, modifyIndexConns, verifyIndexConns) {
     rst.initiate();
     // Note setFCV always waits for majority write concern so in a two node cluster secondaries will
     // always have replicated the setFCV writes.
-    runTest(rst.getPrimary(), [rst.getPrimary()], [rst.getPrimary(), rst.getSecondary()]);
+    runTest(rst.getPrimary(), [rst.getPrimary()], [rst.getPrimary(), rst.getSecondary()], rst);
     rst.stopSet();
 }
 
diff --git a/jstests/sharding/internal_txns/partial_index.js b/jstests/sharding/internal_txns/partial_index.js
index 032505660ab..c6fbb997b85 100644
--- a/jstests/sharding/internal_txns/partial_index.js
+++ b/jstests/sharding/internal_txns/partial_index.js
@@ -14,10 +14,88 @@ const st = new ShardingTest({shards: {rs0: {nodes: 2}}});
 const kDbName = "testDb";
 const kCollName = "testColl";
 const kConfigTxnNs = "config.transactions";
+const kPartialIndexName = "parent_lsid";
 
 const mongosTestDB = st.s.getDB(kDbName);
 const shard0PrimaryConfigTxnColl = st.rs0.getPrimary().getCollection(kConfigTxnNs);
 
+function assertPartialIndexExists(node) {
+    const configDB = node.getDB("config");
+    const indexSpecs = assert.commandWorked(configDB.runCommand({"listIndexes": "transactions"}))
+                           .cursor.firstBatch;
+    indexSpecs.sort((index0, index1) => index0.name > index1.name);
+    assert.eq(indexSpecs.length, 2);
+    const idIndexSpec = indexSpecs[0];
+    assert.eq(idIndexSpec.key, {"_id": 1});
+    const partialIndexSpec = indexSpecs[1];
+    assert.eq(partialIndexSpec.key, {"parentLsid": 1, "_id.txnNumber": 1, "_id": 1});
+    assert.eq(partialIndexSpec.partialFilterExpression, {"parentLsid": {"$exists": true}});
+}
+
+function assertFindUsesCoveredQuery(node) {
+    const configTxnColl = node.getCollection(kConfigTxnNs);
+    const childSessionDoc = configTxnColl.findOne({
+        "_id.id": sessionUUID,
+        "_id.txnNumber": childLsid.txnNumber,
+        "_id.txnUUID": childLsid.txnUUID
+    });
+
+    const explainRes = assert.commandWorked(
+        configTxnColl.explain()
+            .find({"parentLsid": parentSessionDoc._id, "_id.txnNumber": childLsid.txnNumber},
+                  {_id: 1})
+            .finish());
+    const winningPlan = getWinningPlan(explainRes.queryPlanner);
+    assert.eq(winningPlan.stage, "PROJECTION_COVERED");
+    assert.eq(winningPlan.inputStage.stage, "IXSCAN");
+
+    const findRes =
+        configTxnColl
+            .find({"parentLsid": parentSessionDoc._id, "_id.txnNumber": childLsid.txnNumber},
+                  {_id: 1})
+            .toArray();
+    assert.eq(findRes.length, 1);
+    assert.eq(findRes[0]._id, childSessionDoc._id);
+}
+
+function assertPartialIndexDoesNotExist(node) {
+    const configDB = node.getDB("config");
+    const indexSpecs = assert.commandWorked(configDB.runCommand({"listIndexes": "transactions"}))
+                           .cursor.firstBatch;
+    assert.eq(indexSpecs.length, 1);
+    const idIndexSpec = indexSpecs[0];
+    assert.eq(idIndexSpec.key, {"_id": 1});
+}
+
+function indexRecreationTest(recreateAfterDrop) {
+    st.rs0.getPrimary().getCollection(kConfigTxnNs).dropIndex(kPartialIndexName);
+    st.rs0.awaitReplication();
+
+    st.rs0.nodes.forEach(node => {
+        assertPartialIndexDoesNotExist(node);
+    });
+
+    let primary = st.rs0.getPrimary();
+    assert.commandWorked(
+        primary.adminCommand({replSetStepDown: ReplSetTest.kForeverSecs, force: true}));
+    assert.commandWorked(primary.adminCommand({replSetFreeze: 0}));
+
+    st.rs0.awaitNodesAgreeOnPrimary();
+    st.rs0.awaitReplication();
+
+    st.rs0.nodes.forEach(node => {
+        if (recreateAfterDrop) {
+            assertPartialIndexExists(node);
+        } else {
+            assertPartialIndexDoesNotExist(node);
+        }
+    });
+}
+
+// If the collection is empty and the index does not exist, we should create the partial index on
+// stepup.
+indexRecreationTest(true /*Recreate after drop*/);
+
 const sessionUUID = UUID();
 const parentLsid = {
     id: sessionUUID
@@ -59,45 +137,6 @@ function runRetryableInternalTransaction(txnNumber) {
     }));
 }
 
-function assertPartialIndexExists(node) {
-    const configDB = node.getDB("config");
-    const indexSpecs = assert.commandWorked(configDB.runCommand({"listIndexes": "transactions"}))
-                           .cursor.firstBatch;
-    indexSpecs.sort((index0, index1) => index0.name > index1.name);
-    assert.eq(indexSpecs.length, 2);
-    const idIndexSpec = indexSpecs[0];
-    assert.eq(idIndexSpec.key, {"_id": 1});
-    const partialIndexSpec = indexSpecs[1];
-    assert.eq(partialIndexSpec.key, {"parentLsid": 1, "_id.txnNumber": 1, "_id": 1});
-    assert.eq(partialIndexSpec.partialFilterExpression, {"parentLsid": {"$exists": true}});
-}
-
-function assertFindUsesCoveredQuery(node) {
-    const configTxnColl = node.getCollection(kConfigTxnNs);
-    const childSessionDoc = configTxnColl.findOne({
-        "_id.id": sessionUUID,
-        "_id.txnNumber": childLsid.txnNumber,
-        "_id.txnUUID": childLsid.txnUUID
-    });
-
-    const explainRes = assert.commandWorked(
-        configTxnColl.explain()
-            .find({"parentLsid": parentSessionDoc._id, "_id.txnNumber": childLsid.txnNumber},
-                  {_id: 1})
-            .finish());
-    const winningPlan = getWinningPlan(explainRes.queryPlanner);
-    assert.eq(winningPlan.stage, "PROJECTION_COVERED");
-    assert.eq(winningPlan.inputStage.stage, "IXSCAN");
-
-    const findRes =
-        configTxnColl
-            .find({"parentLsid": parentSessionDoc._id, "_id.txnNumber": childLsid.txnNumber},
-                  {_id: 1})
-            .toArray();
-    assert.eq(findRes.length, 1);
-    assert.eq(findRes[0]._id, childSessionDoc._id);
-}
-
 runRetryableInternalTransaction(childTxnNumber);
 assert.eq(shard0PrimaryConfigTxnColl.count({"_id.id": sessionUUID}), 2);
 
@@ -121,7 +160,7 @@ st.rs0.nodes.forEach(node => {
 //
 
 const indexConn = st.rs0.getPrimary();
-assert.commandWorked(indexConn.getCollection("config.transactions").dropIndex("parent_lsid"));
+assert.commandWorked(indexConn.getCollection("config.transactions").dropIndex(kPartialIndexName));
 
 // Normal writes don't involve config.transactions, so they succeed.
 assert.commandWorked(indexConn.getDB(kDbName).runCommand(
@@ -246,5 +285,9 @@ assert.commandWorked(indexConn.adminCommand({
     autocommit: false
 }));
 
+// We expect that if the partial index is dropped when the collection isn't empty, then on stepup we
+// should not recreate the collection.
+indexRecreationTest(false /*Don't recreate after drop*/);
+
 st.stop();
 })();
diff --git a/src/mongo/db/session_catalog_mongod.cpp b/src/mongo/db/session_catalog_mongod.cpp
index fcefa1ed69b..fbfa223d80c 100644
--- a/src/mongo/db/session_catalog_mongod.cpp
+++ b/src/mongo/db/session_catalog_mongod.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/create_indexes_gen.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/index_builds_coordinator.h"
+#include "mongo/db/internal_transactions_feature_flag_gen.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/ops/write_ops.h"
@@ -373,25 +374,45 @@ int removeExpiredTransactionSessionsFromDisk(
 }
 
 void createTransactionTable(OperationContext* opCtx) {
-    auto serviceCtx = opCtx->getServiceContext();
     CollectionOptions options;
-    auto createCollectionStatus =
-        repl::StorageInterface::get(serviceCtx)
-            ->createCollection(opCtx, NamespaceString::kSessionTransactionsTableNamespace, options);
+    auto storageInterface = repl::StorageInterface::get(opCtx);
+    auto createCollectionStatus = storageInterface->createCollection(
+        opCtx, NamespaceString::kSessionTransactionsTableNamespace, options);
+
     if (createCollectionStatus == ErrorCodes::NamespaceExists) {
-        return;
+        if (!feature_flags::gFeatureFlagInternalTransactions.isEnabled(
+                serverGlobalParams.featureCompatibility)) {
+            return;
+        }
+
+        AutoGetCollection autoColl(
+            opCtx, NamespaceString::kSessionTransactionsTableNamespace, LockMode::MODE_IS);
+
+        // During failover recovery it is possible that the collection is created, but the partial
+        // index is not since they are recorded as separate oplog entries. If it is already created
+        // or if the collection isn't empty we can return early.
+        if (autoColl->getIndexCatalog()->findIndexByName(
+                opCtx, MongoDSessionCatalog::kConfigTxnsPartialIndexName) ||
+            !autoColl->isEmpty(opCtx)) {
+            return;
+        }
+    } else {
+        uassertStatusOKWithContext(createCollectionStatus,
+                                   str::stream()
+                                       << "Failed to create the "
+                                       << NamespaceString::kSessionTransactionsTableNamespace.ns()
+                                       << " collection");
     }
 
-    uassertStatusOKWithContext(
-        createCollectionStatus,
-        str::stream() << "Failed to create the "
-                      << NamespaceString::kSessionTransactionsTableNamespace.ns() << " collection");
+    if (!feature_flags::gFeatureFlagInternalTransactions.isEnabled(
+            serverGlobalParams.featureCompatibility)) {
+        return;
+    }
 
     auto indexSpec = MongoDSessionCatalog::getConfigTxnPartialIndexSpec();
 
-    const auto createIndexStatus =
-        repl::StorageInterface::get(opCtx)->createIndexesOnEmptyCollection(
-            opCtx, NamespaceString::kSessionTransactionsTableNamespace, {indexSpec});
+    const auto createIndexStatus = storageInterface->createIndexesOnEmptyCollection(
+        opCtx, NamespaceString::kSessionTransactionsTableNamespace, {indexSpec});
     uassertStatusOKWithContext(
         createIndexStatus,
         str::stream() << "Failed to create partial index for the "