SERVER-66777 Ensure that internal transactions do not get interrupted by logical session reaper

(cherry picked from commit 01938cf7239dc4eb6a2fa79b31743cd815d4d92c)

7 files changed, 406 insertions, 217 deletions

diff --git a/jstests/replsets/internal_sessions_reaping_basic.js b/jstests/replsets/internal_sessions_reaping_basic.js
index f2f5d5a86c8..ca2dbc06597 100644
--- a/jstests/replsets/internal_sessions_reaping_basic.js
+++ b/jstests/replsets/internal_sessions_reaping_basic.js
@@ -1,7 +1,11 @@
 /**
- * Tests that the lifetime of the config.transactions and config.image_collection entries for
- * child sessions is tied to the lifetime of the config.system.sessions entry for their parent
- * sessions.
+ * Tests that the reaper does not reap expired internal transaction sessions for non-retryable
+ * writes or non-internal transaction sessions until the logical sessions that they correspond to
+ * have expired.
+ *
+ * Tests that the logical session cache reaper reaps expired internal transaction sessions for old
+ * retryable writes even when the config.system.sessions entries for the logical sessions that they
+ * correspond to still exist (i.e. the logical sessions still haven't expired).
  *
  * @tags: [requires_fcv_60, uses_transactions]
  */
@@ -9,8 +13,10 @@
 (function() {
 "use strict";
 
-// This test makes assertions about the number of sessions, which are not compatible with
-// implicit sessions.
+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 rst = new ReplSetTest({
@@ -18,6 +24,7 @@ const rst = new ReplSetTest({
     nodeOptions: {
         setParameter: {
             maxSessions: 1,
+            // Make transaction records expire immediately.
             TransactionRecordMinimumLifetimeMinutes: 0,
             storeFindAndModifyImagesInSideCollection: true
         }
@@ -37,162 +44,184 @@ const transactionsColl = primary.getCollection(kConfigTxnsNs);
 const imageColl = primary.getCollection(kImageCollNs);
 const oplogColl = primary.getCollection(kOplogCollNs);
 
-const kDbName = "testDb";
-const kCollName = "testColl";
-const testDB = primary.getDB(kDbName);
+const dbName = "testDb";
+const collName = "testColl";
+const testDB = primary.getDB(dbName);
+const testColl = testDB.getCollection(collName);
 
-assert.commandWorked(testDB.createCollection(kCollName));
-assert.commandWorked(primary.adminCommand({refreshLogicalSessionCacheNow: 1}));
+assert.commandWorked(testDB.runCommand({
+    insert: collName,
+    documents: [{_id: 0}, {_id: 1}],
+}));
 
 const sessionUUID = UUID();
 const parentLsid = {
     id: sessionUUID
 };
-
-const kInternalTxnNumber = NumberLong(0);
-
-let numTransactionsCollEntries = 0;
-let numImageCollEntries = 0;
-
-assert.commandWorked(
-    testDB.runCommand({insert: kCollName, documents: [{_id: 0}], lsid: parentLsid}));
-
-const childLsid0 = {
-    id: sessionUUID,
-    txnUUID: UUID()
-};
-assert.commandWorked(testDB.runCommand({
-    update: kCollName,
-    updates: [{q: {_id: 0}, u: {$set: {a: 0}}}],
-    lsid: childLsid0,
-    txnNumber: kInternalTxnNumber,
-    startTransaction: true,
-    autocommit: false
-}));
-assert.commandWorked(testDB.adminCommand(
-    {commitTransaction: 1, lsid: childLsid0, txnNumber: kInternalTxnNumber, autocommit: false}));
-numTransactionsCollEntries++;
-assert.eq(numTransactionsCollEntries, transactionsColl.find().itcount());
-
-jsTest.log("Verify that the config.transactions entry for the internal transaction for " +
-           "the non-retryable update did not get reaped after command returned");
-assert.eq(numTransactionsCollEntries, transactionsColl.find().itcount());
-
-const parentTxnNumber1 = NumberLong(1);
-
-assert.commandWorked(testDB.runCommand({
-    update: kCollName,
-    updates: [{q: {_id: 0}, u: {$set: {b: 0}}}],
-    lsid: parentLsid,
-    txnNumber: parentTxnNumber1,
-    stmtId: NumberInt(0)
-}));
-numTransactionsCollEntries++;
-
-const childLsid1 = {
-    id: sessionUUID,
-    txnNumber: parentTxnNumber1,
-    txnUUID: UUID()
-};
-assert.commandWorked(testDB.runCommand({
-    update: kCollName,
-    updates: [{q: {_id: 0}, u: {$set: {c: 0}}}],
-    lsid: childLsid1,
-    txnNumber: kInternalTxnNumber,
-    stmtId: NumberInt(1),
-    startTransaction: true,
-    autocommit: false
-}));
-assert.commandWorked(testDB.adminCommand(
-    {commitTransaction: 1, lsid: childLsid1, txnNumber: kInternalTxnNumber, autocommit: false}));
-numTransactionsCollEntries++;
-
-const parentTxnNumber2 = NumberLong(2);
-
-assert.commandWorked(testDB.runCommand({
-    findAndModify: kCollName,
-    query: {_id: 0},
-    update: {$set: {d: 0}},
-    lsid: parentLsid,
-    txnNumber: parentTxnNumber2,
-    stmtId: NumberInt(0)
-}));
-numImageCollEntries++;
-
-jsTest.log("Verify that the config.transactions entry for the retryable internal transaction for " +
-           "the update did not get reaped although there is already a new retryable write");
-assert.eq(numTransactionsCollEntries, transactionsColl.find().itcount());
-
-const childLsid2 = {
-    id: sessionUUID,
-    txnNumber: parentTxnNumber2,
-    txnUUID: UUID()
-};
-assert.commandWorked(testDB.runCommand({
-    findAndModify: kCollName,
-    query: {_id: 0},
-    update: {$set: {e: 0}},
-    lsid: childLsid2,
-    txnNumber: kInternalTxnNumber,
-    stmtId: NumberInt(1),
-    startTransaction: true,
-    autocommit: false
-}));
-assert.commandWorked(testDB.adminCommand(
-    {commitTransaction: 1, lsid: childLsid2, txnNumber: kInternalTxnNumber, autocommit: false}));
-numTransactionsCollEntries++;
-numImageCollEntries++;
-
-const parentTxnNumber3 = NumberLong(3);
-
-assert.commandWorked(testDB.runCommand({
-    insert: kCollName,
-    documents: [{_id: 1}],
-    lsid: parentLsid,
-    txnNumber: parentTxnNumber3,
-    stmtId: NumberInt(0)
-}));
-
-jsTest.log("Verify that the config.transactions entry for the retryable internal transaction for " +
-           "the findAndModify did not get reaped although there is already a new retryable write");
-assert.eq(numTransactionsCollEntries, transactionsColl.find().itcount());
-assert.eq(numImageCollEntries, imageColl.find().itcount());
-
-assert.eq({_id: 0, a: 0, b: 0, c: 0, d: 0, e: 0},
-          testDB.getCollection(kCollName).findOne({_id: 0}));
-assert.eq({_id: 1}, testDB.getCollection(kCollName).findOne({_id: 1}));
-
-assert.commandWorked(primary.adminCommand({refreshLogicalSessionCacheNow: 1}));
-
-assert.eq(1, sessionsColl.find({"_id.id": sessionUUID}).itcount());
-assert.eq(numTransactionsCollEntries, transactionsColl.find().itcount());
-assert.eq(numImageCollEntries, imageColl.find().itcount());
-
-assert.commandWorked(primary.adminCommand({reapLogicalSessionCacheNow: 1}));
-
-jsTest.log("Verify that the config.transactions entries for internal transactions did not get " +
-           "reaped although they are expired since the config.system.sessions entry for the " +
-           "parent session still has not been deleted");
-
-assert.eq(1, sessionsColl.find({"_id.id": sessionUUID}).itcount());
-assert.eq(numTransactionsCollEntries,
-          transactionsColl.find().itcount(),
-          tojson(transactionsColl.find().toArray()));
-assert.eq(numImageCollEntries, imageColl.find().itcount());
-
-// Remove the session doc so the parent session gets reaped when reapLogicalSessionCacheNow is run.
-assert.commandWorked(sessionsColl.remove({}));
-assert.commandWorked(primary.adminCommand({reapLogicalSessionCacheNow: 1}));
-
-jsTest.log("Verify that the config.transactions entries got reaped since the " +
-           "config.system.sessions entry for the parent session had already been deleted");
-assert.eq(0, sessionsColl.find().itcount());
-assert.eq(0, transactionsColl.find().itcount(), tojson(transactionsColl.find().toArray()));
-assert.eq(0, imageColl.find().itcount());
+const parentLsidFilter = makeLsidFilter(parentLsid, "_id");
+let parentTxnNumber = 0;
+const childTxnNumber = NumberLong(0);
+
+let numTransactionsCollEntriesReaped = 0;
+
+{
+    jsTest.log("Test reaping when there is an expired internal transaction session for a " +
+               "non-retryable write without an open transaction");
+
+    parentTxnNumber++;
+    assert.commandWorked(testDB.runCommand({
+        findAndModify: collName,
+        query: {_id: 0},
+        update: {$set: {x: 0}},
+        lsid: parentLsid,
+        txnNumber: NumberLong(parentTxnNumber),
+    }));
+    assert.commandWorked(primary.adminCommand({refreshLogicalSessionCacheNow: 1}));
+    assert.eq(1, sessionsColl.find({"_id.id": sessionUUID}).itcount());
+    assert.eq(1, transactionsColl.find(parentLsidFilter).itcount());
+    assert.eq(1, imageColl.find(parentLsidFilter).itcount());
+
+    const childLsid = {id: sessionUUID, txnUUID: UUID()};
+    const childLsidFilter = makeLsidFilter(childLsid, "_id");
+    assert.commandWorked(testDB.runCommand({
+        findAndModify: collName,
+        query: {_id: 0},
+        update: {$set: {y: 0}},
+        lsid: childLsid,
+        txnNumber: childTxnNumber,
+        startTransaction: true,
+        autocommit: false
+    }));
+    assert.commandWorked(
+        testDB.adminCommand(makeCommitTransactionCmdObj(childLsid, childTxnNumber)));
+
+    assert.eq({_id: 0, x: 0, y: 0}, testColl.findOne({_id: 0}));
+
+    // Verify that the config.transactions entry for the internal transaction session for
+    // non-retryable write does not get reaped automatically when the transaction committed.
+    assert.eq(1, sessionsColl.find({"_id.id": sessionUUID}).itcount());
+    assert.eq(1, transactionsColl.find(parentLsidFilter).itcount());
+    assert.eq(1, transactionsColl.find(childLsidFilter).itcount());
+    assert.eq(1, imageColl.find(parentLsidFilter).itcount());
+    assert.eq(0, imageColl.find(childLsidFilter).itcount());
+
+    // Force the logical session cache to reap, and verify that the config.transactions entries for
+    // the internal transaction session for non-retryable write and the non-internal transaction
+    // session do not get reaped because the config.system.sessions entry still has not been
+    // deleted.
+    assert.commandWorked(primary.adminCommand({reapLogicalSessionCacheNow: 1}));
+    assert.eq(1, sessionsColl.find({"_id.id": sessionUUID}).itcount());
+    assert.eq(1, transactionsColl.find(parentLsidFilter).itcount());
+    assert.eq(1, transactionsColl.find(childLsidFilter).itcount());
+    assert.eq(1, imageColl.find(parentLsidFilter).itcount());
+    assert.eq(0, imageColl.find(childLsidFilter).itcount());
+
+    // Delete the config.system.sessions entry, force the logical session cache to reap again, and
+    // verify that the config.transactions entries for both sessions do get reaped this time.
+    assert.commandWorked(sessionsColl.remove({}));
+    assert.commandWorked(primary.adminCommand({reapLogicalSessionCacheNow: 1}));
+    assert.eq(0, sessionsColl.find({"_id.id": sessionUUID}).itcount());
+    assert.eq(0, transactionsColl.find(parentLsidFilter).itcount());
+    assert.eq(0, transactionsColl.find(childLsidFilter).itcount());
+    assert.eq(0, imageColl.find(parentLsidFilter).itcount());
+    assert.eq(0, imageColl.find(parentLsidFilter).itcount());
+    numTransactionsCollEntriesReaped += 2;
+}
+
+{
+    jsTest.log("Test reaping when there is an expired internal transaction session for a " +
+               "previous retryable write (i.e. with an old txnNumber)");
+
+    parentTxnNumber++;
+    assert.commandWorked(testDB.runCommand({
+        findAndModify: collName,
+        query: {_id: 1},
+        update: {$set: {x: 1}},
+        lsid: parentLsid,
+        txnNumber: NumberLong(parentTxnNumber),
+    }));
+    assert.commandWorked(primary.adminCommand({refreshLogicalSessionCacheNow: 1}));
+    assert.eq(1, sessionsColl.find({"_id.id": sessionUUID}).itcount());
+    assert.eq(1, transactionsColl.find(parentLsidFilter).itcount());
+
+    parentTxnNumber++;
+    const childLsid = {id: sessionUUID, txnNumber: NumberLong(parentTxnNumber), txnUUID: UUID()};
+    const childLsidFilter = makeLsidFilter(childLsid, "_id");
+    assert.commandWorked(testDB.runCommand({
+        findAndModify: collName,
+        query: {_id: 1},
+        update: {$set: {y: 1}},
+        lsid: childLsid,
+        txnNumber: childTxnNumber,
+        startTransaction: true,
+        autocommit: false
+    }));
+    assert.commandWorked(
+        testDB.adminCommand(makeCommitTransactionCmdObj(childLsid, childTxnNumber)));
+
+    assert.eq({_id: 1, x: 1, y: 1}, testColl.findOne({_id: 1}));
+
+    parentTxnNumber++;
+    assert.commandWorked(testDB.runCommand({
+        findAndModify: collName,
+        query: {_id: 1},
+        update: {$set: {y: 1}},
+        lsid: parentLsid,
+        txnNumber: NumberLong(parentTxnNumber),
+        startTransaction: true,
+        autocommit: false
+    }));
+
+    // Verify that the the config.transactions entry and config.image_collection entry for the
+    // internal transaction session for the previous retryable write do not get reaped automatically
+    // when the new txnNumber started.
+    assert.eq(1, sessionsColl.find({"_id.id": sessionUUID}).itcount());
+    assert.eq(1, transactionsColl.find(parentLsidFilter).itcount());
+    assert.eq(1, transactionsColl.find(childLsidFilter).itcount());
+    assert.eq(1, imageColl.find(parentLsidFilter).itcount());
+    assert.eq(1, imageColl.find(childLsidFilter).itcount());
+
+    // Force the logical session cache to reap, and verify that the config.transactions entry and
+    // config.image_collection entry for the internal transaction session for the previous
+    // retryable write do get reaped although the config.system.sessions entry still has not been
+    // deleted.
+    assert.commandWorked(primary.adminCommand({reapLogicalSessionCacheNow: 1}));
+    assert.eq(1, sessionsColl.find({"_id.id": sessionUUID}).itcount());
+    assert.eq(1, transactionsColl.find(parentLsidFilter).itcount());
+    assert.eq(0, transactionsColl.find(childLsidFilter).itcount());
+    assert.eq(1, imageColl.find(parentLsidFilter).itcount());
+    assert.eq(0, imageColl.find(childLsidFilter).itcount());
+    numTransactionsCollEntriesReaped++;
+
+    assert.commandWorked(
+        testDB.adminCommand(makeCommitTransactionCmdObj(parentLsid, parentTxnNumber)));
+    assert.eq({_id: 1, x: 1, y: 1}, testColl.findOne({_id: 1}));
+    assert.eq(1, sessionsColl.find({"_id.id": sessionUUID}).itcount());
+    assert.eq(1, transactionsColl.find(parentLsidFilter).itcount());
+    assert.eq(1, imageColl.find(parentLsidFilter).itcount());
+
+    // Force the logical session cache to reap, and verify that the config.transactions entry and
+    // config.image_collection entry for the non-internal transaction session do not get reaped
+    // because the config.system.sessions entry still has not been deleted.
+    assert.commandWorked(primary.adminCommand({reapLogicalSessionCacheNow: 1}));
+    assert.eq(1, sessionsColl.find({"_id.id": sessionUUID}).itcount());
+    assert.eq(1, transactionsColl.find(parentLsidFilter).itcount());
+    assert.eq(1, imageColl.find(parentLsidFilter).itcount());
+
+    // Delete the config.system.sessions entry, force the logical session cache to reap again, and
+    // verify that the config.transactions entry for the expired transaction session does get
+    // reaped this time.
+    assert.commandWorked(sessionsColl.remove({}));
+    assert.commandWorked(primary.adminCommand({reapLogicalSessionCacheNow: 1}));
+    assert.eq(0, sessionsColl.find({"_id.id": sessionUUID}).itcount());
+    assert.eq(0, transactionsColl.find(parentLsidFilter).itcount());
+    assert.eq(0, imageColl.find(parentLsidFilter).itcount());
+    numTransactionsCollEntriesReaped++;
+}
 
 // Validate that writes to config.transactions do not generate oplog entries, with the exception of
 // deletions.
-assert.eq(numTransactionsCollEntries, oplogColl.find({op: 'd', ns: kConfigTxnsNs}).itcount());
+assert.eq(numTransactionsCollEntriesReaped, oplogColl.find({op: 'd', ns: kConfigTxnsNs}).itcount());
 assert.eq(0, oplogColl.find({op: {'$ne': 'd'}, ns: kConfigTxnsNs}).itcount());
 
 rst.stopSet();
diff --git a/jstests/replsets/internal_sessions_reaping_interrupt.js b/jstests/replsets/internal_sessions_reaping_interrupt.js
new file mode 100644
index 00000000000..827d21c8b9e
--- /dev/null
+++ b/jstests/replsets/internal_sessions_reaping_interrupt.js
@@ -0,0 +1,85 @@
+/*
+ * Tests that reaping expired internal transaction sessions does not cause the operations on the
+ * corresponding logical sessions to be interrupted.
+ *
+ * @tags: [requires_fcv_60, uses_transactions]
+ */
+(function() {
+"use strict";
+
+const logicalSessionRefreshMillis = 1000;
+const rst = new ReplSetTest({
+    nodes: 2,
+    nodeOptions: {
+        setParameter: {
+            // Disable the TTL monitor to ensure that the config.system.sessions entry for the
+            // test session is always around.
+            ttlMonitorEnabled: false,
+            disableLogicalSessionCacheRefresh: false,
+            TransactionRecordMinimumLifetimeMinutes: 0,
+            logicalSessionRefreshMillis
+        }
+    }
+});
+rst.startSet();
+rst.initiate();
+
+const primary = rst.getPrimary();
+
+const dbName = "testDb";
+const collName = "testColl";
+const minReapTimes = 5;
+const minDuration = minReapTimes * logicalSessionRefreshMillis;
+
+const sessionsColl = primary.getCollection("config.system.sessions");
+
+function runTest(isRetryableWriteSession, runTxns) {
+    jsTest.log(`Start testing with ${tojson({isRetryableWriteSession, runTxns})}`);
+    const session = primary.startSession({retryWrites: isRetryableWriteSession});
+    const db = session.getDatabase(dbName);
+    const coll = db.getCollection(collName);
+
+    assert.commandWorked(coll.remove({}));
+    assert.commandWorked(primary.adminCommand({refreshLogicalSessionCacheNow: 1}));
+    assert.eq(1, sessionsColl.find({"_id.id": session.getSessionId().id}).itcount());
+
+    const startTime = new Date();
+    let currTime = new Date();
+
+    while (currTime - startTime < minDuration) {
+        const isTxn = runTxns && Math.random() > 0.5;
+        if (isTxn) {
+            session.startTransaction();
+        }
+
+        const doc = {_id: UUID()};
+        const insertOp = {
+            insert: collName,
+            documents: [doc],
+        };
+        if (isRetryableWriteSession && !isTxn) {
+            insertOp.stmtId = NumberInt(1);
+        }
+        assert.commandWorked(db.adminCommand(
+            {testInternalTransactions: 1, commandInfos: [{dbName: dbName, command: insertOp}]}));
+
+        if (isTxn) {
+            assert.commandWorked(session.commitTransaction_forTesting());
+        }
+        assert.eq(coll.find(doc).itcount(), 1);
+        currTime = new Date();
+    }
+
+    const endTime = new Date();
+    jsTest.log(`Finished testing with ${
+        tojson({isRetryableWriteSession, timeTaken: (endTime - startTime)})}`);
+}
+
+for (let isRetryableWriteSession of [true, false]) {
+    for (let runTxns of [true, false]) {
+        runTest(isRetryableWriteSession, runTxns);
+    }
+}
+
+rst.stopSet();
+})();
diff --git a/jstests/replsets/internal_sessions_reaping_retryable_writes.js b/jstests/replsets/internal_sessions_reaping_retryable_writes.js
index ad584d55c84..d0c68261d80 100644
--- a/jstests/replsets/internal_sessions_reaping_retryable_writes.js
+++ b/jstests/replsets/internal_sessions_reaping_retryable_writes.js
@@ -1,7 +1,6 @@
 /*
- * Test that the logical cache reaper reaps Session/TransactionParticipant objects and the
- * config.transactions and config.image_collection entries that correspond to the same retryable
- * write atomically.
+ * Test that the logical session cache reaper reaps transaction sessions that correspond to the same
+ * retryable write atomically.
  *
  * @tags: [requires_fcv_60, uses_transactions]
  */
@@ -72,31 +71,18 @@ function makeSessionOptsForTest() {
     };
 }
 
-function assertNumSessionsCollEntries(sessionOpts, expectedNum) {
+function assertNumEntries(
+    sessionOpts, {numSessionsCollEntries, numTransactionsCollEntries, numImageCollEntries}) {
     const filter = {"_id.id": sessionOpts.parentLsid.id};
-    assert.eq(expectedNum,
-              sessionsColl.find(filter).itcount(),
-              tojson(sessionsColl.find(filter).toArray()));
-}
 
-function assertNumTransactionsCollEntries(sessionOpts, expectedNum) {
-    const filter = {"_id.id": sessionOpts.parentLsid.id};
-    assert.eq(expectedNum,
-              transactionsColl.find(filter).itcount(),
-              tojson(transactionsColl.find(filter).toArray()));
-}
+    const sessionsCollEntries = sessionsColl.find(filter).toArray();
+    assert.eq(numSessionsCollEntries, sessionsCollEntries.length, sessionsCollEntries);
 
-function assertNumImagesCollEntries(sessionOpts, expectedNum) {
-    const filter = {"_id.id": sessionOpts.parentLsid.id};
-    assert.eq(
-        expectedNum, imageColl.find(filter).itcount(), tojson(imageColl.find(filter).toArray()));
-}
+    const transactionsCollEntries = transactionsColl.find(filter).toArray();
+    assert.eq(numTransactionsCollEntries, transactionsCollEntries.length, transactionsCollEntries);
 
-function assertNumEntries(
-    sessionOpts, {numSessionsCollEntries, numTransactionsCollEntries, numImageCollEntries}) {
-    assertNumSessionsCollEntries(sessionOpts, numSessionsCollEntries);
-    assertNumTransactionsCollEntries(sessionOpts, numTransactionsCollEntries);
-    assertNumImagesCollEntries(sessionOpts, numImageCollEntries);
+    const imageCollEntries = imageColl.find(filter).toArray();
+    assert.eq(numImageCollEntries, imageCollEntries.length, imageCollEntries);
 }
 
 // Test reaping when neither the external session nor the internal sessions are checked out.
diff --git a/src/mongo/db/session_catalog.cpp b/src/mongo/db/session_catalog.cpp
index 032e9d8b90e..8e66b4b1732 100644
--- a/src/mongo/db/session_catalog.cpp
+++ b/src/mongo/db/session_catalog.cpp
@@ -249,7 +249,9 @@ SessionCatalog::KillToken SessionCatalog::killSession(const LogicalSessionId& ls
 
     auto sri = _getSessionRuntimeInfo(lg, lsid);
     uassert(ErrorCodes::NoSuchSession, "Session not found", sri);
-    return ObservableSession(lg, sri, &sri->parentSession).kill();
+    auto session = sri->getSession(lsid);
+    uassert(ErrorCodes::NoSuchSession, "Session not found", session);
+    return ObservableSession(lg, sri, session).kill();
 }
 
 size_t SessionCatalog::size() const {
@@ -348,12 +350,17 @@ SessionCatalog::KillToken ObservableSession::kill(ErrorCodes::Error reason) cons
     const bool firstKiller = (0 == _sri->killsRequested);
     ++_sri->killsRequested;
 
-    // For currently checked-out sessions, interrupt the operation context so that the current owner
-    // can release the session
     if (firstKiller && hasCurrentOperation()) {
+        // Interrupt the current OperationContext if its running on the transaction session
+        // that is being killed or if we are killing the parent transaction session.
         invariant(_clientLock.owns_lock());
-        const auto serviceContext = _sri->checkoutOpCtx->getServiceContext();
-        serviceContext->killOperation(_clientLock, _sri->checkoutOpCtx, reason);
+        const auto checkedOutLsid = _sri->checkoutOpCtx->getLogicalSessionId();
+        const auto lsidToKill = getSessionId();
+        const bool isKillingParentSession = !getParentSessionId(lsidToKill);
+        if ((checkedOutLsid == lsidToKill) || isKillingParentSession) {
+            const auto serviceContext = _sri->checkoutOpCtx->getServiceContext();
+            serviceContext->killOperation(_clientLock, _sri->checkoutOpCtx, reason);
+        }
     }
 
     return SessionCatalog::KillToken(getSessionId());
diff --git a/src/mongo/db/session_catalog_mongod.cpp b/src/mongo/db/session_catalog_mongod.cpp
index 427569bba01..7170228af2a 100644
--- a/src/mongo/db/session_catalog_mongod.cpp
+++ b/src/mongo/db/session_catalog_mongod.cpp
@@ -210,23 +210,64 @@ const auto kLastWriteDateFieldName = SessionTxnRecord::kLastWriteDateFieldName;
 
 /**
  * Removes the the config.transactions and the config.image_collection entries for the transaction
- * sessions in 'possiblyExpiredTransactionSessionIds' that are actually expired. Returns the number
+ * sessions in 'expiredTransactionSessionIdsNotInUse' that are safe to reap. Returns the number
  * of transaction sessions whose entries were removed.
  */
 int removeSessionsTransactionRecords(
     OperationContext* opCtx,
     SessionsCollection& sessionsCollection,
-    const LogicalSessionIdSet& possiblyExpiredTransactionSessionIds) {
-    if (possiblyExpiredTransactionSessionIds.empty()) {
+    const LogicalSessionIdSet& expiredTransactionSessionIdsNotInUse) {
+    if (expiredTransactionSessionIdsNotInUse.empty()) {
         return 0;
     }
 
-    // From the possibly expired transaction session ids, find the ones which are actually
-    // expired/removed.
-    auto expiredTransactionSessionIds =
-        sessionsCollection.findRemovedSessions(opCtx, possiblyExpiredTransactionSessionIds);
+    // From the expired transaction session ids that are no longer in use, find the ones that are
+    // safe to reap.
+    LogicalSessionIdSet transactionSessionIdsToReap;
+    {
+        LogicalSessionIdSet possiblyExpiredLogicalSessionIds;
+        LogicalSessionIdMap<LogicalSessionIdSet>
+            transactionSessionIdsToReapIfLogicalSessionsExpired;
+
+        for (const auto& transactionSessionId : expiredTransactionSessionIdsNotInUse) {
+            if (isInternalSessionForRetryableWrite(transactionSessionId)) {
+                // It is safe to reap an internal transaction session for retryable write if it
+                // its transaction record has already expired since by design internal transaction
+                // sessions for retryable write are never reused and reaping them would not
+                // interrupt operations on other transaction sessions for the logical sessions that
+                // they correspond to.
+                transactionSessionIdsToReap.insert(transactionSessionId);
+            } else {
+                // It not safe to reap an internal transaction session for non-retryable write until
+                // the logical session that it corresponds to has expired, even if its transaction
+                // record has already expired. The reason is that each internal transaction session
+                // for non-retryable write is kept in the internal session pool and is reusable
+                // as long as the logical session that its correspond to has not expired and so
+                // reaping it would interrupt any operation that is running on it. The same applies
+                // to a parent transaction session since reaping it would interrupt all operations
+                // on that logical session.
+                auto logicalSessionId = castToParentSessionId(transactionSessionId);
+                possiblyExpiredLogicalSessionIds.insert(logicalSessionId);
+                transactionSessionIdsToReapIfLogicalSessionsExpired[logicalSessionId].insert(
+                    transactionSessionId);
+            }
+        }
+
+        if (!transactionSessionIdsToReapIfLogicalSessionsExpired.empty()) {
+            auto expiredLogicalSessionIds =
+                sessionsCollection.findRemovedSessions(opCtx, possiblyExpiredLogicalSessionIds);
+            for (const auto& [logicalSessionId, transactionSessionIds] :
+                 transactionSessionIdsToReapIfLogicalSessionsExpired) {
+                if (expiredLogicalSessionIds.find(logicalSessionId) !=
+                    expiredLogicalSessionIds.end()) {
+                    transactionSessionIdsToReap.insert(transactionSessionIds.begin(),
+                                                       transactionSessionIds.end());
+                }
+            }
+        }
+    }
 
-    if (expiredTransactionSessionIds.empty()) {
+    if (transactionSessionIdsToReap.empty()) {
         return 0;
     }
 
@@ -247,7 +288,7 @@ int removeSessionsTransactionRecords(
         }());
         imageDeleteOp.setDeletes([&] {
             std::vector<write_ops::DeleteOpEntry> entries;
-            for (const auto& transactionSessionId : expiredTransactionSessionIds) {
+            for (const auto& transactionSessionId : transactionSessionIdsToReap) {
                 entries.emplace_back(
                     BSON(LogicalSessionRecord::kIdFieldName << transactionSessionId.toBSON()),
                     false /* multi = false */);
@@ -268,7 +309,7 @@ int removeSessionsTransactionRecords(
         }());
         sessionDeleteOp.setDeletes([&] {
             std::vector<write_ops::DeleteOpEntry> entries;
-            for (const auto& transactionSessionId : expiredTransactionSessionIds) {
+            for (const auto& transactionSessionId : transactionSessionIdsToReap) {
                 entries.emplace_back(
                     BSON(LogicalSessionRecord::kIdFieldName << transactionSessionId.toBSON()),
                     false /* multi = false */);
@@ -303,7 +344,7 @@ int removeExpiredTransactionSessionsFromDisk(
     // limit.
     const int kMaxBatchSize = 10'000;
 
-    LogicalSessionIdSet possiblyExpiredTransactionSessionIds;
+    LogicalSessionIdSet expiredTransactionSessionIdsNotInUse;
     int numReaped = 0;
     while (cursor->more()) {
         auto transactionSession = SessionsCollectionFetchResultIndividualResult::parse(
@@ -315,15 +356,15 @@ int removeExpiredTransactionSessionsFromDisk(
             continue;
         }
 
-        possiblyExpiredTransactionSessionIds.insert(transactionSessionId);
-        if (possiblyExpiredTransactionSessionIds.size() > kMaxBatchSize) {
+        expiredTransactionSessionIdsNotInUse.insert(transactionSessionId);
+        if (expiredTransactionSessionIdsNotInUse.size() > kMaxBatchSize) {
             numReaped += removeSessionsTransactionRecords(
-                opCtx, sessionsCollection, possiblyExpiredTransactionSessionIds);
-            possiblyExpiredTransactionSessionIds.clear();
+                opCtx, sessionsCollection, expiredTransactionSessionIdsNotInUse);
+            expiredTransactionSessionIdsNotInUse.clear();
         }
     }
     numReaped += removeSessionsTransactionRecords(
-        opCtx, sessionsCollection, possiblyExpiredTransactionSessionIds);
+        opCtx, sessionsCollection, expiredTransactionSessionIdsNotInUse);
 
     return numReaped;
 }
diff --git a/src/mongo/db/session_catalog_test.cpp b/src/mongo/db/session_catalog_test.cpp
index 653149af99f..ef9100ea5ae 100644
--- a/src/mongo/db/session_catalog_test.cpp
+++ b/src/mongo/db/session_catalog_test.cpp
@@ -1126,7 +1126,7 @@ TEST_F(SessionCatalogTest, KillSessionWhenChildSessionIsNotCheckedOut) {
     runTest(parentLsid, makeLogicalSessionIdWithTxnUUIDForTest(parentLsid));
 }
 
-TEST_F(SessionCatalogTest, KillingChildSessionInterruptsParentSession) {
+TEST_F(SessionCatalogTest, KillingChildSessionDoesNotInterruptParentSession) {
     auto runTest = [&](const LogicalSessionId& parentLsid, const LogicalSessionId& childLsid) {
         auto killToken = [this, &parentLsid, &childLsid] {
             assertCanCheckoutSession(childLsid);
@@ -1137,9 +1137,8 @@ TEST_F(SessionCatalogTest, KillingChildSessionInterruptsParentSession) {
 
             auto killToken = catalog()->killSession(childLsid);
 
-            // Make sure the owning operation context is interrupted
-            ASSERT_THROWS_CODE(
-                opCtx->checkForInterrupt(), AssertionException, ErrorCodes::Interrupted);
+            // Make sure the owning operation context is not interrupted.
+            opCtx->checkForInterrupt();
 
             // Make sure that the checkOutForKill call will wait for the owning operation context to
             // check the session back in
@@ -1369,7 +1368,7 @@ TEST_F(SessionCatalogTest, MarkSessionAsKilledCanBeCalledMoreThanOnce) {
     runTest(makeLogicalSessionIdWithTxnUUIDForTest());
 }
 
-TEST_F(SessionCatalogTest, MarkSessionsAsKilledWhenSessionDoesNotExist) {
+TEST_F(SessionCatalogTest, MarkNonExistentSessionAsKilled) {
     auto runTest = [&](const LogicalSessionId& nonExistentLsid) {
         ASSERT_THROWS_CODE(
             catalog()->killSession(nonExistentLsid), AssertionException, ErrorCodes::NoSuchSession);
@@ -1380,6 +1379,61 @@ TEST_F(SessionCatalogTest, MarkSessionsAsKilledWhenSessionDoesNotExist) {
     runTest(makeLogicalSessionIdWithTxnUUIDForTest());
 }
 
+TEST_F(SessionCatalogTest, MarkNonExistentChildSessionAsKilledWhenParentSessionExists) {
+    auto runTest = [&](const LogicalSessionId& parentLsid,
+                       const LogicalSessionId& nonExistentChildLsid) {
+        createSession(parentLsid);
+        ASSERT_THROWS_CODE(catalog()->killSession(nonExistentChildLsid),
+                           AssertionException,
+                           ErrorCodes::NoSuchSession);
+    };
+
+    {
+        auto parentLsid = makeLogicalSessionIdForTest();
+        runTest(parentLsid, makeLogicalSessionIdWithTxnNumberAndUUIDForTest(parentLsid));
+    }
+
+    {
+        auto parentLsid = makeLogicalSessionIdForTest();
+        runTest(parentLsid, makeLogicalSessionIdWithTxnUUIDForTest(parentLsid));
+    }
+}
+
+
+TEST_F(SessionCatalogTest, MarkNonExistentChildSessionAsKilledWhenOtherChildSessionExists) {
+    auto runTest = [&](const LogicalSessionId& existentChildLsid,
+                       const LogicalSessionId& nonExistentChildLsid) {
+        createSession(existentChildLsid);
+        ASSERT_THROWS_CODE(catalog()->killSession(nonExistentChildLsid),
+                           AssertionException,
+                           ErrorCodes::NoSuchSession);
+    };
+
+    {
+        auto parentLsid = makeLogicalSessionIdForTest();
+        runTest(makeLogicalSessionIdWithTxnNumberAndUUIDForTest(parentLsid),
+                makeLogicalSessionIdWithTxnNumberAndUUIDForTest(parentLsid));
+    }
+
+    {
+        auto parentLsid = makeLogicalSessionIdForTest();
+        runTest(makeLogicalSessionIdWithTxnUUIDForTest(parentLsid),
+                makeLogicalSessionIdWithTxnUUIDForTest(parentLsid));
+    }
+
+    {
+        auto parentLsid = makeLogicalSessionIdForTest();
+        runTest(makeLogicalSessionIdWithTxnNumberAndUUIDForTest(parentLsid),
+                makeLogicalSessionIdWithTxnUUIDForTest(parentLsid));
+    }
+
+    {
+        auto parentLsid = makeLogicalSessionIdForTest();
+        runTest(makeLogicalSessionIdWithTxnUUIDForTest(parentLsid),
+                makeLogicalSessionIdWithTxnNumberAndUUIDForTest(parentLsid));
+    }
+}
+
 TEST_F(SessionCatalogTestWithDefaultOpCtx, SessionDiscarOperationContextAfterCheckIn) {
     auto runTest = [&](const LogicalSessionId& lsid) {
         _opCtx->setLogicalSessionId(lsid);
diff --git a/src/mongo/db/sessions_collection.cpp b/src/mongo/db/sessions_collection.cpp
index 697cd611cfc..45c59c3631d 100644
--- a/src/mongo/db/sessions_collection.cpp
+++ b/src/mongo/db/sessions_collection.cpp
@@ -231,8 +231,7 @@ LogicalSessionIdSet SessionsCollection::_doFindRemoved(
         batch.push_back(record);
     };
 
-
-    LogicalSessionIdSet activeSessions;
+    LogicalSessionIdSet removed{sessions.begin(), sessions.end()};
 
     auto wrappedSend = [&](BSONObj batch) {
         BSONObjBuilder batchWithReadConcernLocal(batch);
@@ -244,7 +243,7 @@ LogicalSessionIdSet SessionsCollection::_doFindRemoved(
             SessionsCollectionFetchResult::parse("SessionsCollectionFetchResult"_sd, swBatchResult);
 
         for (const auto& lsid : result.getCursor().getFirstBatch()) {
-            activeSessions.insert(lsid.get_id());
+            removed.erase(lsid.get_id());
         }
     };
 
@@ -267,18 +266,6 @@ LogicalSessionIdSet SessionsCollection::_doFindRemoved(
 
     runBulkGeneric(makeT, add, sendLocal, sessions);
 
-    LogicalSessionIdSet removed;
-    for (const auto& session : sessions) {
-        if (activeSessions.find(session) != activeSessions.end()) {
-            continue;
-        }
-        if (auto parentSession = getParentSessionId(session);
-            parentSession && (activeSessions.find(*parentSession) != activeSessions.end())) {
-            continue;
-        }
-        removed.insert(session);
-    }
-
     return removed;
 }