SERVER-55882: Invalidate in-memory transactions when fetching committed transactions in tenant migrations

(cherry picked from commit c5ae3db5845814fe5e6ecd304d797c8f655fa697)

5 files changed, 143 insertions, 13 deletions

diff --git a/buildscripts/resmokeconfig/suites/tenant_migration_multi_stmt_txn_jscore_passthrough.yml b/buildscripts/resmokeconfig/suites/tenant_migration_multi_stmt_txn_jscore_passthrough.yml
index 0500363c080..8da743c03c1 100644
--- a/buildscripts/resmokeconfig/suites/tenant_migration_multi_stmt_txn_jscore_passthrough.yml
+++ b/buildscripts/resmokeconfig/suites/tenant_migration_multi_stmt_txn_jscore_passthrough.yml
@@ -314,9 +314,6 @@ selector:
   # Does not support tojson of command objects.
   - jstests/core/SERVER-23626.js
 
-  # TODO(SERVER-55882): Investigate why this test is failing.
-  - jstests/core/wildcard_index_multikey.js
-
   exclude_with_any_tags:
   - assumes_standalone_mongod
   # These tests run getMore commands which are not supported in the tenant migration passthrough.
diff --git a/buildscripts/resmokelib/testing/hooks/tenant_migration.py b/buildscripts/resmokelib/testing/hooks/tenant_migration.py
index b9f6189f9b1..1bf3bce18f8 100644
--- a/buildscripts/resmokelib/testing/hooks/tenant_migration.py
+++ b/buildscripts/resmokelib/testing/hooks/tenant_migration.py
@@ -205,7 +205,7 @@ class _TenantMigrationOptions:
 class _TenantMigrationThread(threading.Thread):  # pylint: disable=too-many-instance-attributes
     THREAD_NAME = "TenantMigrationThread"
 
-    WAIT_SECS_RANGES = [[0.1, 0.5], [1, 5], [5, 15]]
+    WAIT_SECS_RANGES = [[0.05, 0.1], [0.1, 0.5], [1, 5], [5, 15]]
     POLL_INTERVAL_SECS = 0.1
 
     NO_SUCH_MIGRATION_ERR_CODE = 327
@@ -262,6 +262,13 @@ class _TenantMigrationThread(threading.Thread):  # pylint: disable=too-many-inst
                     donor_rs_index + 1) % self._tenant_migration_fixture.get_num_replsets()
                 migration_opts = self._create_migration_opts(donor_rs_index, recipient_rs_index)
 
+                # Briefly wait to let the test run before starting the tenant migration, so that
+                # the first migration is more likely to have data to migrate.
+                wait_secs = random.uniform(
+                    *self.WAIT_SECS_RANGES[migration_num % len(self.WAIT_SECS_RANGES)])
+                self.logger.info("Waiting for %.3f seconds before starting migration.", wait_secs)
+                self.__lifecycle.wait_for_tenant_migration_interval(wait_secs)
+
                 self.logger.info("Starting tenant migration: %s.", str(migration_opts))
                 start_time = time.time()
                 is_committed = self._run_migration(migration_opts)
@@ -273,10 +280,6 @@ class _TenantMigrationThread(threading.Thread):  # pylint: disable=too-many-inst
                 if found_idle_request:
                     continue
 
-                wait_secs = random.uniform(
-                    *self.WAIT_SECS_RANGES[migration_num % len(self.WAIT_SECS_RANGES)])
-                self.__lifecycle.wait_for_tenant_migration_interval(wait_secs)
-
                 if is_committed:
                     donor_rs_index = recipient_rs_index
                 migration_num += 1
diff --git a/jstests/replsets/tenant_migration_recipient_invalidates_in_memory_txns.js b/jstests/replsets/tenant_migration_recipient_invalidates_in_memory_txns.js
new file mode 100644
index 00000000000..e916e36b7d3
--- /dev/null
+++ b/jstests/replsets/tenant_migration_recipient_invalidates_in_memory_txns.js
@@ -0,0 +1,116 @@
+/**
+ * Tests that tenant migrations will invalid transactions that are stored in-memory when fetching
+ * committed transactions from the donor. We do this by first running a transaction with txnNum 0 on
+ * the donor and migrating it to the recipient. After the migration commits, we then run a read-only
+ * transaction on the recipient. We expect the read-only transaction to only advance the in-memory
+ * state of the transaction -- it should not update the 'config.transaction' entry on the recipient.
+ * Then, we run a second migration. By this point, the recipient has an in-memory transaction number
+ * of 1 for the session, which is ahead of the donor. However, the migration should still complete,
+ * because the recipient will invalidate its in-memory understanding and refetch the on-disk
+ * transaction state instead.
+ *
+ * Note: this test is designed to emulate a back-and-forth migration from donor to recipient,
+ * recipient to donor, then donor to recipient again.
+ *
+ * @tags: [requires_fcv_49, requires_majority_read_concern, incompatible_with_eft,
+ * incompatible_with_windows_tls, incompatible_with_macos, requires_persistence]
+ */
+
+(function() {
+"use strict";
+
+load("jstests/replsets/libs/tenant_migration_test.js");
+load("jstests/replsets/rslib.js");
+load("jstests/libs/uuid_util.js");
+
+const setParameterOpts = {
+    tenantMigrationGarbageCollectionDelayMS: 3 * 1000
+};
+const tenantMigrationTest =
+    new TenantMigrationTest({name: jsTestName(), sharedOptions: {setParameter: setParameterOpts}});
+if (!tenantMigrationTest.isFeatureFlagEnabled()) {
+    jsTestLog("Skipping test because the tenant migrations feature flag is disabled");
+    tenantMigrationTest.stop();
+    return;
+}
+
+const tenantId = "testTenantId";
+const tenantDB = tenantMigrationTest.tenantDB(tenantId, "testDB");
+const collName = "testColl";
+const transactionsNS = "config.transactions";
+
+const donorPrimary = tenantMigrationTest.getDonorPrimary();
+const recipientPrimary = tenantMigrationTest.getRecipientPrimary();
+
+const commitTransaction = (node, lsid, txnNumber, readOnly = false) => {
+    txnNumber = NumberLong(txnNumber);
+
+    const cmd = readOnly ? {find: collName} : {insert: collName, documents: [{x: 1}]};
+    const cmdObj = Object.assign(cmd, {
+        lsid,
+        txnNumber,
+        stmtId: NumberInt(0),
+        startTransaction: true,
+        autocommit: false,
+    });
+    assert.commandWorked(node.getDB(tenantDB).runCommand(cmdObj));
+
+    assert.commandWorked(node.adminCommand({
+        commitTransaction: 1,
+        lsid,
+        txnNumber,
+        stmtId: NumberInt(0),
+        autocommit: false,
+        writeConcern: {w: "majority"}
+    }));
+};
+
+const session = donorPrimary.startSession({causalConsistency: false});
+const lsid = session.getSessionId();
+
+jsTestLog("Committing transaction number 0 on donor");
+commitTransaction(donorPrimary, lsid, 0, false /* readOnly */);
+
+jsTestLog("Running the first migration");
+const migrationId1 = UUID();
+const migrationOpts1 = {
+    migrationIdString: extractUUIDFromObject(migrationId1),
+    tenantId,
+};
+let migrationRes = assert.commandWorked(tenantMigrationTest.runMigration(migrationOpts1));
+assert.eq(migrationRes.state, TenantMigrationTest.State.kCommitted);
+tenantMigrationTest.waitForMigrationGarbageCollection(migrationId1, tenantId);
+
+// Verify that the config.transaction entry was migrated successfully.
+let recipientTxnEntry = recipientPrimary.getCollection(transactionsNS).find().toArray()[0];
+assert.eq(lsid.id, recipientTxnEntry._id.id);
+assert.eq(0, recipientTxnEntry.txnNum);
+
+// Run a read-only txn on the recipient with the same sessionId. The read-only
+// transaction should not update the on-disk 'config.transaction' entry, but it will update the
+// in-memory txnNum to 1 for the session.
+jsTestLog("Committing transaction number 1 on recipient");
+commitTransaction(recipientPrimary, lsid, 1, true /* readOnly */);
+
+recipientTxnEntry = recipientPrimary.getCollection(transactionsNS).find().toArray()[0];
+assert.eq(lsid.id, recipientTxnEntry._id.id);
+// The transaction number should still be 0, since transaction with txnNum 1 was read-only and thus
+// will only be updated in memory.
+assert.eq(0, recipientTxnEntry.txnNum);
+
+jsTestLog("Dropping tenant DB on recipient");
+assert.commandWorked(recipientPrimary.getDB(tenantDB).dropDatabase());
+
+jsTestLog("Running the second migration");
+const migrationId2 = UUID();
+const migrationOpts2 = {
+    migrationIdString: extractUUIDFromObject(migrationId2),
+    tenantId,
+};
+migrationRes = assert.commandWorked(tenantMigrationTest.runMigration(migrationOpts2));
+// The migration should have committed successfully even though the in-memory transaction number was
+// higher, since the higher number should have been invalidated.
+assert.eq(migrationRes.state, TenantMigrationTest.State.kCommitted);
+
+tenantMigrationTest.stop();
+})();
diff --git a/jstests/replsets/tenant_migration_sync_source_too_stale.js b/jstests/replsets/tenant_migration_sync_source_too_stale.js
index 2096394f3fc..cf315683e5f 100644
--- a/jstests/replsets/tenant_migration_sync_source_too_stale.js
+++ b/jstests/replsets/tenant_migration_sync_source_too_stale.js
@@ -47,7 +47,7 @@ const tenantMigrationTest = new TenantMigrationTest({
     donorRst,
     // Set a low value for excluding donor hosts so that the test doesn't take long to retry a sync
     // source.
-    sharedOptions: {setParamter: {tenantMigrationExcludeDonorHostTimeoutMS: 1000}}
+    sharedOptions: {setParameter: {tenantMigrationExcludeDonorHostTimeoutMS: 1000}}
 });
 
 const tenantId = "testTenantId";
@@ -111,7 +111,7 @@ tenantMigrationTest.insertDonorDB(tenantDB, collName);
 hangRecipientPrimaryAfterRetrievingStartOpTimes.off();
 hangRecipientPrimaryBeforeConsistentState.wait();
 
-const hangRecipientPrimaryAfterRestart = configureFailPoint(
+const hangAfterRetrievingOpTimesAfterRestart = configureFailPoint(
     recipientPrimary, 'fpAfterRetrievingStartOpTimesMigrationRecipientInstance', {action: "hang"});
 
 jsTestLog("Stopping donorSecondary");
@@ -128,7 +128,7 @@ assert.soon(() => {
     return recipientDoc[0].numRestartsDueToDonorConnectionFailure == 1;
 });
 
-hangRecipientPrimaryAfterRestart.wait();
+hangAfterRetrievingOpTimesAfterRestart.wait();
 
 res = recipientPrimary.adminCommand({currentOp: true, desc: "tenant recipient migration"});
 currOp = res.inprog[0];
@@ -138,19 +138,26 @@ assert.eq(currOp.dataSyncCompleted, false, tojson(res));
 // Since 'donorSecondary' was shut down, the sync source can only be 'delayedSecondary'.
 assert.eq(delayedSecondary.host, currOp.donorSyncSource, tojson(res));
 
-hangRecipientPrimaryAfterRestart.off();
+const hangAfterPersistingTenantMigrationRecipientInstanceStateDoc =
+    configureFailPoint(recipientPrimary,
+                       "fpAfterPersistingTenantMigrationRecipientInstanceStateDoc",
+                       {action: "hang"});
+hangAfterRetrievingOpTimesAfterRestart.off();
 
 assert.soon(() => {
     // Verify that the recipient eventually has to restart again, since its lastFetched is ahead of
     // the last OpTime on 'delayedSecondary'.
     const recipientDoc = configRecipientNs.find({"_id": migrationId}).toArray();
-    return recipientDoc[0].numRestartsDueToDonorConnectionFailure == 2;
+    return recipientDoc[0].numRestartsDueToDonorConnectionFailure >= 2;
 });
 
+hangAfterPersistingTenantMigrationRecipientInstanceStateDoc.wait();
+
 // Let 'delayedSecondary' catch back up to the recipient's lastFetched OpTime.
 donorRst.remove(donorSecondary);
 restartServerReplication(delayedSecondary);
 donorRst.awaitReplication();
+hangAfterPersistingTenantMigrationRecipientInstanceStateDoc.off();
 
 // Verify that the migration eventually commits successfully.
 const migrationRes =
diff --git a/src/mongo/db/repl/tenant_migration_recipient_service.cpp b/src/mongo/db/repl/tenant_migration_recipient_service.cpp
index 7e910a2a30d..94d7a0a4f70 100644
--- a/src/mongo/db/repl/tenant_migration_recipient_service.cpp
+++ b/src/mongo/db/repl/tenant_migration_recipient_service.cpp
@@ -939,6 +939,13 @@ void TenantMigrationRecipientService::Instance::_processCommittedTransactionEntr
                           << txnNumber << " on session " << sessionId,
             txnParticipant);
 
+    // The in-memory transaction state may have been updated past the on-disk transaction state. For
+    // instance, this might happen in an unprepared read-only transaction, which updates in-memory
+    // but not on-disk. To prevent potential errors, we use the on-disk state for the following
+    // transaction number checks.
+    txnParticipant.invalidate(opCtx);
+    txnParticipant.refreshFromStorageIfNeeded(opCtx);
+
     // If the entry's transaction number is stale/older than the current active transaction number
     // on the participant, fail the migration.
     uassert(ErrorCodes::TransactionTooOld,