SERVER-65641 Add test coverage for reaping a session with active TransactionParticipant and TransactionRouter yielders

(cherry picked from commit 879d439c78c5c72735af6fbfdb2690eb32f74559)

4 files changed, 155 insertions, 10 deletions

diff --git a/jstests/sharding/internal_txns/reap_sessions_with_active_yielders.js b/jstests/sharding/internal_txns/reap_sessions_with_active_yielders.js
new file mode 100644
index 00000000000..eae0f3ee24b
--- /dev/null
+++ b/jstests/sharding/internal_txns/reap_sessions_with_active_yielders.js
@@ -0,0 +1,130 @@
+/*
+ * Test that the logical session cache reaper does not reap sessions with active TransactionRouter
+ * yielders.
+ *
+ * @tags: [requires_fcv_60, uses_transactions]
+ */
+(function() {
+"use strict";
+
+load("jstests/libs/fail_point_util.js");
+load("jstests/libs/parallelTester.js");
+load("jstests/libs/uuid_util.js");
+load("jstests/sharding/libs/sharded_transactions_helpers.js");
+
+// This test runs the reapLogicalSessionCacheNow command. That can lead to direct writes to the
+// config.transactions collection, which cannot be performed on a session.
+TestData.disableImplicitSessions = true;
+
+const st = new ShardingTest({
+    shards: 2,
+    rs: {
+        setParameter: {
+            TransactionRecordMinimumLifetimeMinutes: 0,
+        }
+    }
+});
+const shard0Primary = st.rs0.getPrimary();
+
+// Set up a sharded collection with two chunks:
+// shard0: [MinKey, 0]
+// shard1: [0, MaxKey]
+const dbName = "testDb";
+const collName = "testColl";
+const ns = dbName + "." + collName;
+const mongosTestDB = st.s.getDB(dbName);
+const mongosTestColl = mongosTestDB.getCollection(collName);
+
+assert.commandWorked(st.s.adminCommand({enableSharding: dbName}));
+assert.commandWorked(st.s.adminCommand({shardCollection: ns, key: {x: 1}}));
+assert.commandWorked(st.s.adminCommand({split: ns, middle: {x: 0}}));
+assert.commandWorked(
+    st.s.adminCommand({moveChunk: ns, find: {x: MinKey}, to: st.shard0.shardName}));
+assert.commandWorked(st.s.adminCommand({moveChunk: ns, find: {x: 1}, to: st.shard1.shardName}));
+
+const sessionsColl = st.s.getCollection("config.system.sessions");
+const transactionsCollOnShard0 = shard0Primary.getCollection("config.transactions");
+const imageCollOnShard0 = shard0Primary.getCollection("config.image_collection");
+
+function assertNumEntries(
+    {sessionUUID, numSessionsCollEntries, numTransactionsCollEntries, numImageCollEntries}) {
+    const filter = {"_id.id": sessionUUID};
+    assert.eq(numSessionsCollEntries,
+              sessionsColl.find(filter).itcount(),
+              tojson(sessionsColl.find().toArray()));
+    assert.eq(numTransactionsCollEntries,
+              transactionsCollOnShard0.find(filter).itcount(),
+              tojson(transactionsCollOnShard0.find().toArray()));
+    assert.eq(numImageCollEntries,
+              imageCollOnShard0.find().itcount(),
+              tojson(imageCollOnShard0.find().toArray()));
+}
+
+jsTest.log("Test reaping when there is an internal transaction with an active " +
+           "TransactionRouter yielder");
+
+const parentLsid = {
+    id: UUID()
+};
+
+const runInternalTxn = (shard0PrimaryHost, parentLsidUUIDString, dbName, collName) => {
+    const shard0Primary = new Mongo(shard0PrimaryHost);
+    const testInternalTxnCmdObj = {
+        testInternalTransactions: 1,
+        commandInfos: [{
+            dbName,
+            command: {
+                insert: collName,
+                documents: [{x: -10}, {x: 10}],
+                stmtIds: [NumberInt(1), NumberInt(2)]
+            }
+        }],
+        useClusterClient: true,
+        lsid: {id: UUID(parentLsidUUIDString)},
+    };
+    assert.commandWorked(shard0Primary.adminCommand(testInternalTxnCmdObj));
+};
+
+// Use {skip:2} to pause the transaction before unyielding after committing instead of after
+// executing the two insert statements.
+let fp = configureFailPoint(shard0Primary, "hangBeforeUnyieldingTransactionRouter", {}, {skip: 2});
+const internalTxnThread = new Thread(
+    runInternalTxn, shard0Primary.host, extractUUIDFromObject(parentLsid.id), dbName, collName);
+internalTxnThread.start();
+fp.wait();
+
+assert.commandWorked(shard0Primary.adminCommand({refreshLogicalSessionCacheNow: 1}));
+assertNumEntries({
+    sessionUUID: parentLsid.id,
+    numSessionsCollEntries: 1,
+    numTransactionsCollEntries: 1,
+    numImageCollEntries: 0
+});
+
+// Force the logical session cache to reap, and verify that the config.transactions entry for
+// the internal transaction does not get reaped.
+assert.commandWorked(sessionsColl.remove({"_id.id": parentLsid.id}));
+assert.commandWorked(shard0Primary.adminCommand({reapLogicalSessionCacheNow: 1}));
+assertNumEntries({
+    sessionUUID: parentLsid.id,
+    numSessionsCollEntries: 0,
+    numTransactionsCollEntries: 1,
+    numImageCollEntries: 0
+});
+
+fp.off();
+internalTxnThread.join();
+
+assert.eq(mongosTestColl.find({x: -10}).itcount(), 1);
+assert.eq(mongosTestColl.find({x: 10}).itcount(), 1);
+
+assert.commandWorked(shard0Primary.adminCommand({reapLogicalSessionCacheNow: 1}));
+assertNumEntries({
+    sessionUUID: parentLsid.id,
+    numSessionsCollEntries: 0,
+    numTransactionsCollEntries: 0,
+    numImageCollEntries: 0
+});
+
+st.stop();
+})();
diff --git a/src/mongo/db/transaction_participant.cpp b/src/mongo/db/transaction_participant.cpp
index ba0dd452462..bbbf8a7c93e 100644
--- a/src/mongo/db/transaction_participant.cpp
+++ b/src/mongo/db/transaction_participant.cpp
@@ -1404,26 +1404,35 @@ void TransactionParticipant::Participant::unstashTransactionResources(OperationC
     invariant(!opCtx->getClient()->isInDirectClient());
     invariant(opCtx->getTxnNumber());
 
-    uassert(ErrorCodes::NoSuchTransaction,
-            str::stream() << "The requested transaction number is different than the "
-                             "active transaction. Requested: "
-                          << *opCtx->getTxnNumber()
-                          << ". Active: " << o().activeTxnNumberAndRetryCounter.getTxnNumber(),
-            *opCtx->getTxnNumber() == o().activeTxnNumberAndRetryCounter.getTxnNumber());
-
+    // Verify that transaction number and mode are as expected.
     if (opCtx->inMultiDocumentTransaction()) {
+        uassert(ErrorCodes::NoSuchTransaction,
+                str::stream() << "Attempted to run '" << cmdName << "' inside a transaction with "
+                              << "session id" << _sessionId() << " and transaction number "
+                              << *opCtx->getTxnNumber()
+                              << " but the active transaction number on the session is "
+                              << o().activeTxnNumberAndRetryCounter.getTxnNumber(),
+                *opCtx->getTxnNumber() == o().activeTxnNumberAndRetryCounter.getTxnNumber());
+
         uassert(6611000,
                 str::stream() << "Attempted to use the active transaction number "
                               << o().activeTxnNumberAndRetryCounter.getTxnNumber() << " in session "
                               << _sessionId()
-                              << " for a transaction but it corresponds to a retryable write",
+                              << " to run a transaction but it corresponds to a retryable write",
                 !o().txnState.isInRetryableWriteMode());
     } else {
+        uassert(6564100,
+                str::stream() << "Attempted to run '" << cmdName << "' as a retryable write with "
+                              << "session id" << _sessionId() << " and transaction number "
+                              << *opCtx->getTxnNumber()
+                              << " but the active transaction number on the session is "
+                              << o().activeTxnNumberAndRetryCounter.getTxnNumber(),
+                *opCtx->getTxnNumber() == o().activeTxnNumberAndRetryCounter.getTxnNumber());
         uassert(6611001,
                 str::stream() << "Attempted to use the active transaction number "
                               << o().activeTxnNumberAndRetryCounter.getTxnNumber() << " in session "
                               << _sessionId()
-                              << " for a retryable write but it corresponds to a transaction",
+                              << " to run a retryable write but it corresponds to a transaction",
                 o().txnState.isInRetryableWriteMode());
     }
 
diff --git a/src/mongo/db/transaction_participant_test.cpp b/src/mongo/db/transaction_participant_test.cpp
index 0c549ad4f33..65c6419b370 100644
--- a/src/mongo/db/transaction_participant_test.cpp
+++ b/src/mongo/db/transaction_participant_test.cpp
@@ -5339,7 +5339,7 @@ TEST_F(TxnParticipantTest, UnstashRetryableWriteAfterActiveTxnNumberHasChanged)
         auto txnParticipant = TransactionParticipant::get(opCtx);
         ASSERT_THROWS_CODE(txnParticipant.unstashTransactionResources(opCtx, "insert"),
                            AssertionException,
-                           ErrorCodes::NoSuchTransaction);
+                           6564100);
     }
 }
 
diff --git a/src/mongo/s/transaction_router_resource_yielder.cpp b/src/mongo/s/transaction_router_resource_yielder.cpp
index 6d5a5e0c2a4..b0621f1792d 100644
--- a/src/mongo/s/transaction_router_resource_yielder.cpp
+++ b/src/mongo/s/transaction_router_resource_yielder.cpp
@@ -36,6 +36,10 @@
 
 namespace mongo {
 
+namespace {
+MONGO_FAIL_POINT_DEFINE(hangBeforeUnyieldingTransactionRouter);
+}
+
 std::unique_ptr<TransactionRouterResourceYielder>
 TransactionRouterResourceYielder::makeForLocalHandoff() {
     return std::make_unique<TransactionRouterResourceYielder>();
@@ -61,6 +65,8 @@ void TransactionRouterResourceYielder::yield(OperationContext* opCtx) {
 
 void TransactionRouterResourceYielder::unyield(OperationContext* opCtx) {
     if (_yielded) {
+        hangBeforeUnyieldingTransactionRouter.pauseWhileSet();
+
         // Code that uses the TransactionRouter assumes it will only run with it, so check back out
         // the session ignoring interruptions, except at global shutdown to prevent stalling
         // shutdown. Unyield should always run with no resources held, so there shouldn't be a risk