SERVER-73938 Make sure chunk migration can handle a retryable internal transaction whose oplog entries have been truncated

(cherry picked from commit b9f6d6beb02df55ede1276222a56279e7b2d48f2)

3 files changed, 110 insertions, 5 deletions

diff --git a/etc/backports_required_for_multiversion_tests.yml b/etc/backports_required_for_multiversion_tests.yml
index c66453c204b..b936889ae67 100644
--- a/etc/backports_required_for_multiversion_tests.yml
+++ b/etc/backports_required_for_multiversion_tests.yml
@@ -308,6 +308,8 @@ last-continuous:
     ticket: SERVER-73110
   - test_file: jstests/core/timeseries/timeseries_groupby_reorder.js
     ticket: SERVER-73822
+  - test_file: jstests/sharding/internal_txns/incomplete_transaction_history_during_migration.js
+    ticket: SERVER-73938
   suites: null
 last-lts:
   all:
@@ -691,4 +693,6 @@ last-lts:
     ticket: SERVER-73110
   - test_file: jstests/core/timeseries/timeseries_groupby_reorder.js
     ticket: SERVER-73822
+  - test_file: jstests/sharding/internal_txns/incomplete_transaction_history_during_migration.js
+    ticket: SERVER-73938
   suites: null
diff --git a/jstests/sharding/internal_txns/incomplete_transaction_history_during_migration.js b/jstests/sharding/internal_txns/incomplete_transaction_history_during_migration.js
new file mode 100644
index 00000000000..92f1ab8a311
--- /dev/null
+++ b/jstests/sharding/internal_txns/incomplete_transaction_history_during_migration.js
@@ -0,0 +1,89 @@
+/*
+ * Test that chunk migration can migrate a retryable internal transaction whose oplog entries have
+ * been truncated.
+ *
+ * @tags: [uses_transactions, requires_persistence]
+ */
+(function() {
+'use strict';
+
+// This test involves writing directly to the config.transactions collection which is not allowed
+// in a session.
+TestData.disableImplicitSessions = true;
+
+const st = new ShardingTest({shards: 2, rs: {nodes: 1}});
+
+const dbName = 'testDb';
+const collName = 'testColl';
+const ns = dbName + '.' + collName;
+const testDB = st.s.getDB(dbName);
+
+assert.commandWorked(st.s.adminCommand({enableSharding: dbName}));
+st.ensurePrimaryShard(dbName, st.shard0.name);
+
+assert.commandWorked(st.s.getCollection(ns).createIndex({x: 1}));
+assert.commandWorked(st.s.adminCommand({shardCollection: ns, key: {x: 1}}));
+
+assert.commandWorked(st.s.getDB(dbName).runCommand({insert: collName, documents: [{x: 1}]}));
+
+const parentLsid = {
+    id: UUID()
+};
+const parentTxnNumber = NumberLong(35);
+
+const originalChildLsid = {
+    id: parentLsid.id,
+    txnNumber: parentTxnNumber,
+    txnUUID: UUID()
+};
+const childTxnNumber = NumberLong(1);
+
+const updateCmdObj = {
+    update: collName,
+    updates: [{q: {x: 1}, u: {$set: {y: 1}}}],
+    stmtId: NumberInt(0),
+};
+const res0 = assert.commandWorked(testDB.runCommand(Object.assign({}, updateCmdObj, {
+    lsid: originalChildLsid,
+    txnNumber: childTxnNumber,
+    autocommit: false,
+    startTransaction: true
+})));
+assert.eq(res0.nModified, 1, res0);
+assert.commandWorked(st.s.adminCommand({
+    commitTransaction: 1,
+    lsid: originalChildLsid,
+    txnNumber: childTxnNumber,
+    autocommit: false,
+}));
+
+// Manually update the config.transactions document for the retryable internal transaction to point
+// to an invalid op time.
+const shard0ConfigTxnsColl = st.rs0.getPrimary().getCollection("config.transactions");
+const res1 = assert.commandWorked(shard0ConfigTxnsColl.update(
+    {"_id.txnUUID": originalChildLsid.txnUUID},
+    {$set: {lastWriteOpTime: {ts: new Timestamp(100, 1), t: NumberLong(1)}}}));
+assert.eq(res1.nModified, 1, res1);
+
+assert.commandWorked(st.s.adminCommand({split: ns, middle: {x: 0}}));
+assert.commandWorked(st.s.adminCommand({moveChunk: ns, find: {x: 0}, to: st.shard1.shardName}));
+
+assert.commandFailedWithCode(testDB.runCommand(Object.assign(
+                                 {}, updateCmdObj, {lsid: parentLsid, txnNumber: parentTxnNumber})),
+                             ErrorCodes.IncompleteTransactionHistory);
+
+const retryChildLsid = {
+    id: parentLsid.id,
+    txnNumber: parentTxnNumber,
+    txnUUID: UUID()
+};
+assert.commandFailedWithCode(testDB.runCommand(Object.assign({}, updateCmdObj, {
+    lsid: retryChildLsid,
+    txnNumber: childTxnNumber,
+    autocommit: false,
+    startTransaction: true
+})),
+                             ErrorCodes.IncompleteTransactionHistory);
+
+st.stop();
+})();
diff --git a/src/mongo/db/s/session_catalog_migration_source.cpp b/src/mongo/db/s/session_catalog_migration_source.cpp
index efcfa3f1bad..b060e810de1 100644
--- a/src/mongo/db/s/session_catalog_migration_source.cpp
+++ b/src/mongo/db/s/session_catalog_migration_source.cpp
@@ -506,11 +506,23 @@ bool SessionCatalogMigrationSource::_handleWriteHistory(WithLock lk, OperationCo
             // Determine if this oplog entry should be migrated. If so, add the oplog entry or the
             // oplog entries derived from it to the oplog buffer.
             if (isInternalSessionForRetryableWrite(*nextOplog->getSessionId())) {
-                invariant(nextOplog->getCommandType() == repl::OplogEntry::CommandType::kApplyOps);
-                // Derive retryable write oplog entries from this retryable internal transaction
-                // applyOps oplog entry, and add them to the oplog buffer.
-                _extractOplogEntriesForInternalTransactionForRetryableWrite(
-                    lk, *nextOplog, &_unprocessedOplogBuffer);
+                if (nextOplog->getCommandType() == repl::OplogEntry::CommandType::kApplyOps) {
+                    // Derive retryable write oplog entries from this retryable internal transaction
+                    // applyOps oplog entry, and add them to the oplog buffer.
+                    _extractOplogEntriesForInternalTransactionForRetryableWrite(
+                        lk, *nextOplog, &_unprocessedOplogBuffer);
+                } else {
+                    tassert(7393800,
+                            str::stream() << "Found an oplog entry for a retrayble internal "
+                                             "transaction with an unexpected type"
+                                          << redact(nextOplog->toBSONForLogging()),
+                            nextOplog->getOpType() == repl::OpTypeEnum::kNoop);
+                    if (!nextOplog->getStatementIds().empty() &&
+                        !shouldSkipOplogEntry(nextOplog.value(), _keyPattern, _chunkRange)) {
+                        _unprocessedOplogBuffer.emplace_back(*nextOplog);
+                    }
+                }
+
                 continue;
             }