SERVER-59495 Donor and recipient tenant migration state machines will persist the migration protocol info and tenantId info will be an empty string for 'Merge' protocol."

17 files changed, 376 insertions, 36 deletions

diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index a9281408827..491be419999 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -624,7 +624,11 @@ env.Library(
         '$BUILD_DIR/mongo/client/connection_string',
         '$BUILD_DIR/mongo/client/read_preference',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
+        '$BUILD_DIR/mongo/db/repl/repl_server_parameters',
         '$BUILD_DIR/mongo/db/repl/tenant_migration_state_machine_idl',
+        '$BUILD_DIR/mongo/db/server_options_core',
+        '$BUILD_DIR/mongo/db/serverless/serverless_types_idl',
+        '$BUILD_DIR/mongo/idl/feature_flag',
         '$BUILD_DIR/mongo/idl/idl_parser',
     ]
 )
diff --git a/src/mongo/db/commands/tenant_migration_donor_cmds.cpp b/src/mongo/db/commands/tenant_migration_donor_cmds.cpp
index 63c62bcd47d..4e37821036f 100644
--- a/src/mongo/db/commands/tenant_migration_donor_cmds.cpp
+++ b/src/mongo/db/commands/tenant_migration_donor_cmds.cpp
@@ -26,6 +26,7 @@
  *    exception statement from all source files in the program, then also delete
  *    it in the license file.
  */
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand
 
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/commands.h"
@@ -36,10 +37,13 @@
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl/tenant_migration_access_blocker_util.h"
 #include "mongo/db/repl/tenant_migration_donor_service.h"
+#include "mongo/logv2/log.h"
 
 namespace mongo {
 namespace {
 
+MONGO_FAIL_POINT_DEFINE(returnResponseCommittedForDonorStartMigrationCmd);
+
 class DonorStartMigrationCmd : public TypedCommand<DonorStartMigrationCmd> {
 public:
     using Request = DonorStartMigration;
@@ -68,16 +72,11 @@ public:
                 !serverGlobalParams.featureCompatibility.isUpgradingOrDowngrading());
 
             const auto& cmd = request();
+            const auto migrationProtocol = cmd.getProtocol().value_or(kDefaulMigrationProtocol);
 
-            if (cmd.getProtocol().value_or(MigrationProtocolEnum::kMultitenantMigrations) ==
-                MigrationProtocolEnum::kShardMerge) {
-                uassert(5949300,
-                        "protocol \"shard merge\" not supported",
-                        repl::feature_flags::gShardMerge.isEnabled(
-                            serverGlobalParams.featureCompatibility));
-            }
+            tenant_migration_util::protocolTenantIdCompatibilityCheck(migrationProtocol,
+                                                                      cmd.getTenantId().toString());
 
-            // TODO (SERVER-59494): tenantId should be optional in the state doc. Include protocol.
             TenantMigrationDonorDocument stateDoc(cmd.getMigrationId(),
                                                   cmd.getRecipientConnectionString().toString(),
                                                   cmd.getReadPreference(),
@@ -96,8 +95,18 @@ public:
                 stateDoc.setRecipientCertificateForDonor(cmd.getRecipientCertificateForDonor());
             }
 
+            stateDoc.setProtocol(migrationProtocol);
+
             const auto stateDocBson = stateDoc.toBSON();
 
+            if (MONGO_unlikely(returnResponseCommittedForDonorStartMigrationCmd.shouldFail())) {
+                LOGV2(5949401,
+                      "Immediately returning committed because "
+                      "'returnResponseCommittedForDonorStartMigrationCmd' failpoint is enabled",
+                      "tenantMigrationDonorInstance"_attr = stateDoc.toBSON());
+                return Response(TenantMigrationDonorStateEnum::kCommitted);
+            }
+
             auto donorService =
                 repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext())
                     ->lookupServiceByName(TenantMigrationDonorService::kServiceName);
diff --git a/src/mongo/db/commands/tenant_migration_donor_cmds.idl b/src/mongo/db/commands/tenant_migration_donor_cmds.idl
index 750504590f9..3117ef7a9ba 100644
--- a/src/mongo/db/commands/tenant_migration_donor_cmds.idl
+++ b/src/mongo/db/commands/tenant_migration_donor_cmds.idl
@@ -35,17 +35,10 @@ global:
 imports:
   - "mongo/client/read_preference_setting.idl"
   - "mongo/db/repl/tenant_migration_state_machine.idl"
+  - "mongo/db/serverless/serverless_types.idl"
   - "mongo/idl/basic_types.idl"
   - "mongo/s/sharding_types.idl"
 
-enums:
-    MigrationProtocol:
-        description: "Determines which tenant migration protocol to use."
-        type: string
-        values:
-            kMultitenantMigrations: "multitenant migrations"
-            kShardMerge: "shard merge"
-
 structs:
     DonorStartMigrationResponse:
         description: "Response of the donorStartMigration command"
@@ -78,6 +71,7 @@ commands:
       tenantId:
         description: "The prefix from which the migrating database will be matched. The prefixes 'admin', 'local', 'config', the empty string, are not allowed."
         type: string
+        default: '""'
         validator:
           callback: "tenant_migration_util::validateDatabasePrefix"
       readPreference:
@@ -101,6 +95,8 @@ commands:
         description: "Which migration protocol to use, default 'multitenant migrations'."
         type: MigrationProtocol
         optional: true
+        validator:
+          callback: "tenant_migration_util::validateProtocolFCVCompatibility"
 
   donorForgetMigration:
     description: "Parser for the 'donorForgetMigration' command."
diff --git a/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp b/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp
index b854b45924e..6120097c424 100644
--- a/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp
+++ b/src/mongo/db/commands/tenant_migration_recipient_cmds.cpp
@@ -34,7 +34,6 @@
 #include "mongo/db/commands/tenant_migration_donor_cmds_gen.h"
 #include "mongo/db/commands/tenant_migration_recipient_cmds_gen.h"
 #include "mongo/db/repl/primary_only_service.h"
-#include "mongo/db/repl/repl_server_parameters_gen.h"
 #include "mongo/db/repl/tenant_migration_recipient_service.h"
 #include "mongo/logv2/log.h"
 
@@ -42,6 +41,7 @@ namespace mongo {
 namespace {
 
 MONGO_FAIL_POINT_DEFINE(returnResponseOkForRecipientSyncDataCmd);
+MONGO_FAIL_POINT_DEFINE(returnResponseOkForRecipientForgetMigrationCmd);
 
 class RecipientSyncDataCmd : public TypedCommand<RecipientSyncDataCmd> {
 public:
@@ -70,6 +70,10 @@ public:
                 !serverGlobalParams.featureCompatibility.isUpgradingOrDowngrading());
 
             const auto& cmd = request();
+            const auto migrationProtocol = cmd.getProtocol().value_or(kDefaulMigrationProtocol);
+
+            tenant_migration_util::protocolTenantIdCompatibilityCheck(migrationProtocol,
+                                                                      cmd.getTenantId().toString());
 
             TenantMigrationRecipientDocument stateDoc(cmd.getMigrationId(),
                                                       cmd.getDonorConnectionString().toString(),
@@ -77,6 +81,7 @@ public:
                                                       cmd.getStartMigrationDonorTimestamp(),
                                                       cmd.getReadPreference());
 
+
             if (!repl::tenantMigrationDisableX509Auth) {
                 uassert(ErrorCodes::InvalidOptions,
                         str::stream() << "'" << Request::kRecipientCertificateForDonorFieldName
@@ -85,11 +90,14 @@ public:
                 stateDoc.setRecipientCertificateForDonor(cmd.getRecipientCertificateForDonor());
             }
 
+            stateDoc.setProtocol(migrationProtocol);
+
             const auto stateDocBson = stateDoc.toBSON();
 
             if (MONGO_unlikely(returnResponseOkForRecipientSyncDataCmd.shouldFail())) {
                 LOGV2(4879608,
-                      "'returnResponseOkForRecipientSyncDataCmd' failpoint enabled.",
+                      "Immediately returning OK because 'returnResponseOkForRecipientSyncDataCmd' "
+                      "failpoint is enabled.",
                       "tenantMigrationRecipientInstance"_attr = stateDoc.toBSON());
                 return Response(repl::OpTime());
             }
@@ -181,6 +189,10 @@ public:
                         serverGlobalParams.clusterRole == ClusterRole::ShardServer);
 
             const auto& cmd = request();
+            const auto migrationProtocol = cmd.getProtocol().value_or(kDefaulMigrationProtocol);
+
+            tenant_migration_util::protocolTenantIdCompatibilityCheck(migrationProtocol,
+                                                                      cmd.getTenantId().toString());
 
             opCtx->setAlwaysInterruptAtStepDownOrUp();
             auto recipientService =
@@ -205,12 +217,21 @@ public:
                         cmd.getRecipientCertificateForDonor());
                 stateDoc.setRecipientCertificateForDonor(cmd.getRecipientCertificateForDonor());
             }
-
+            stateDoc.setProtocol(migrationProtocol);
             // Set the state to 'kDone' so that we don't create a recipient access blocker
             // unnecessarily if this recipientForgetMigration command is received before a
             // recipientSyncData command or after the state doc is garbage collected.
             stateDoc.setState(TenantMigrationRecipientStateEnum::kDone);
 
+
+            if (MONGO_unlikely(returnResponseOkForRecipientForgetMigrationCmd.shouldFail())) {
+                LOGV2(5949502,
+                      "Immediately returning ok because "
+                      "'returnResponseOkForRecipientForgetMigrationCmd' failpoint is enabled",
+                      "tenantMigrationRecipientInstance"_attr = stateDoc.toBSON());
+                return;
+            }
+
             auto recipientInstance = repl::TenantMigrationRecipientService::Instance::getOrCreate(
                 opCtx, recipientService, stateDoc.toBSON());
 
diff --git a/src/mongo/db/commands/tenant_migration_recipient_cmds.idl b/src/mongo/db/commands/tenant_migration_recipient_cmds.idl
index 6a4b44c9752..f6e4f4531e6 100644
--- a/src/mongo/db/commands/tenant_migration_recipient_cmds.idl
+++ b/src/mongo/db/commands/tenant_migration_recipient_cmds.idl
@@ -36,6 +36,7 @@ global:
 imports:
   - "mongo/client/read_preference_setting.idl"
   - "mongo/db/repl/tenant_migration_pem_payload.idl"
+  - "mongo/db/serverless/serverless_types.idl"
   - "mongo/idl/basic_types.idl"
   - "mongo/s/sharding_types.idl"
   - "mongo/db/repl/replication_types.idl"
@@ -68,6 +69,7 @@ structs:
             The prefix from which the migrating database will be matched. The prefixes 'admin',
             'local', 'config', the empty string, are not allowed.
           type: string
+          default: '""'
           validator:
             callback: "tenant_migration_util::validateDatabasePrefix"
         readPreference:
@@ -81,6 +83,12 @@ structs:
           type: TenantMigrationPEMPayload
           # TODO (SERVER-54085): Remove server parameter tenantMigrationDisableX509Auth.
           optional: true
+        protocol:
+          description: "Which migration protocol to use, default 'multitenant migrations'."
+          type: MigrationProtocol
+          optional: true
+          validator:
+            callback: "tenant_migration_util::validateProtocolFCVCompatibility"
 
 commands:
   recipientSyncData:
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index b9a222f5706..dbc111fd53c 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -1292,6 +1292,7 @@ env.Library(
         '$BUILD_DIR/mongo/client/connection_string',
         '$BUILD_DIR/mongo/client/read_preference',
         '$BUILD_DIR/mongo/db/commands/mongod_fcv',
+        '$BUILD_DIR/mongo/db/serverless/serverless_types_idl',
         '$BUILD_DIR/mongo/idl/idl_parser',
         '$BUILD_DIR/mongo/util/net/ssl_manager',
         'optime',
diff --git a/src/mongo/db/repl/tenant_migration_donor_service.cpp b/src/mongo/db/repl/tenant_migration_donor_service.cpp
index 7cf45761185..2c8e242e86b 100644
--- a/src/mongo/db/repl/tenant_migration_donor_service.cpp
+++ b/src/mongo/db/repl/tenant_migration_donor_service.cpp
@@ -249,10 +249,11 @@ TenantMigrationDonorService::Instance::Instance(ServiceContext* const serviceCon
       _serviceContext(serviceContext),
       _donorService(donorService),
       _stateDoc(tenant_migration_access_blocker::parseDonorStateDocument(initialState)),
-      _instanceName(kServiceName + "-" + _stateDoc.getTenantId()),
+      _instanceName(kServiceName + "-" + _stateDoc.getId().toString()),
       _recipientUri(
           uassertStatusOK(MongoURI::parse(_stateDoc.getRecipientConnectionString().toString()))),
       _tenantId(_stateDoc.getTenantId()),
+      _protocol(_stateDoc.getProtocol().value_or(MigrationProtocolEnum::kMultitenantMigrations)),
       _recipientConnectionString(_stateDoc.getRecipientConnectionString()),
       _readPreference(_stateDoc.getReadPreference()),
       _migrationUuid(_stateDoc.getId()),
@@ -260,6 +261,7 @@ TenantMigrationDonorService::Instance::Instance(ServiceContext* const serviceCon
       _recipientCertificateForDonor(_stateDoc.getRecipientCertificateForDonor()),
       _sslMode(repl::tenantMigrationDisableX509Auth ? transport::kGlobalSSLMode
                                                     : transport::kEnableSSL) {
+
     _recipientCmdExecutor = _makeRecipientCmdExecutor();
     _recipientCmdExecutor->startup();
 
@@ -397,8 +399,9 @@ Status TenantMigrationDonorService::Instance::checkIfOptionsConflict(
     const TenantMigrationDonorDocument& stateDoc) {
     stdx::lock_guard<Latch> lg(_mutex);
     invariant(stateDoc.getId() == _migrationUuid);
+    invariant(stateDoc.getProtocol());
 
-    if (stateDoc.getTenantId() == _tenantId &&
+    if (stateDoc.getProtocol().value() == _protocol && stateDoc.getTenantId() == _tenantId &&
         stateDoc.getRecipientConnectionString() == _recipientConnectionString &&
         stateDoc.getReadPreference().equals(_readPreference) &&
         stateDoc.getDonorCertificateForRecipient() == _donorCertificateForRecipient &&
@@ -731,12 +734,18 @@ ExecutorFuture<void> TenantMigrationDonorService::Instance::_sendRecipientSyncDa
         request.setDbName(NamespaceString::kAdminDb);
 
         MigrationRecipientCommonData commonData(
-            _migrationUuid, donorConnString.toString(), _tenantId, _readPreference);
+            _migrationUuid, donorConnString.toString(), _readPreference);
         commonData.setRecipientCertificateForDonor(_recipientCertificateForDonor);
-        request.setMigrationRecipientCommonData(commonData);
+        // TODO SERVER-59794: Pass tenantId only for 'kMultitenantMigrations' protocol.
+        commonData.setTenantId(_tenantId);
 
         stdx::lock_guard<Latch> lg(_mutex);
 
+        if (_isAtLeastFCV52AtStart) {
+            commonData.setProtocol(_protocol);
+        }
+        request.setMigrationRecipientCommonData(commonData);
+
         invariant(_stateDoc.getStartMigrationDonorTimestamp());
         request.setStartMigrationDonorTimestamp(*_stateDoc.getStartMigrationDonorTimestamp());
         request.setReturnAfterReachingDonorTimestamp(_stateDoc.getBlockTimestamp());
@@ -758,8 +767,18 @@ ExecutorFuture<void> TenantMigrationDonorService::Instance::_sendRecipientForget
     request.setDbName(NamespaceString::kAdminDb);
 
     MigrationRecipientCommonData commonData(
-        _migrationUuid, donorConnString.toString(), _tenantId, _readPreference);
+        _migrationUuid, donorConnString.toString(), _readPreference);
     commonData.setRecipientCertificateForDonor(_recipientCertificateForDonor);
+    // TODO SERVER-59794: Pass tenantId only for 'kMultitenantMigrations' protocol.
+    commonData.setTenantId(_tenantId);
+    auto isAtLeastFCV52AtStart = [&]() -> bool {
+        stdx::lock_guard<Latch> lg(_mutex);
+        return _isAtLeastFCV52AtStart;
+    };
+
+    if (isAtLeastFCV52AtStart()) {
+        commonData.setProtocol(_protocol);
+    }
     request.setMigrationRecipientCommonData(commonData);
 
     return _sendCommandToRecipient(executor, recipientTargeterRS, request.toBSON(BSONObj()), token);
@@ -779,6 +798,33 @@ CancellationToken TenantMigrationDonorService::Instance::_initAbortMigrationSour
     return _abortMigrationSource->token();
 }
 
+bool TenantMigrationDonorService::Instance::_checkifProtocolRemainsFCVCompatible() {
+    stdx::lock_guard<Latch> lg(_mutex);
+
+    // Ensure that the on-disk protocol and cached value remains the same.
+    invariant(!_stateDoc.getProtocol() || _stateDoc.getProtocol().value() == getProtocol());
+
+    _isAtLeastFCV52AtStart = serverGlobalParams.featureCompatibility.isGreaterThanOrEqualTo(
+        multiversion::FeatureCompatibilityVersion::kVersion_5_2);
+    if (_isAtLeastFCV52AtStart) {
+        // When the instance is started using state doc < 5.2 FCV format, _stateDoc._protocol field
+        // won't be set. In that case, the cached value Instance::_protocol will be set to
+        // "kMultitenantMigrations".
+        return true;
+    }
+
+    if (getProtocol() == MigrationProtocolEnum::kShardMerge) {
+        LOGV2(5949503,
+              "Must abort tenant migration as 'Merge' protocol is not supported for FCV "
+              "below 5.2");
+        return false;
+    }
+    // For backward compatibility, ensure that the 'protocol' field is not set in the
+    // document.
+    _stateDoc.setProtocol(boost::none);
+    return true;
+}
+
 SemiFuture<void> TenantMigrationDonorService::Instance::run(
     std::shared_ptr<executor::ScopedTaskExecutor> executor,
     const CancellationToken& token) noexcept {
@@ -791,13 +837,27 @@ SemiFuture<void> TenantMigrationDonorService::Instance::run(
         }
     }
 
-    // We must abort the migration if we try to start or resume while upgrading or downgrading.
-    // (Generic FCV reference): This FCV check should exist across LTS binary versions.
-    if (serverGlobalParams.featureCompatibility.isUpgradingOrDowngrading()) {
-        LOGV2(5356302, "Must abort tenant migration as donor is upgrading or downgrading");
+    auto isFCVUpgradingOrDowngrading = [&]() -> bool {
+        // We must abort the migration if we try to start or resume while upgrading or downgrading.
+        // (Generic FCV reference): This FCV check should exist across LTS binary versions.
+        if (serverGlobalParams.featureCompatibility.isUpgradingOrDowngrading()) {
+            LOGV2(5356302, "Must abort tenant migration as donor is upgrading or downgrading");
+            return true;
+        }
+        return false;
+    };
+
+
+    // Tenant migrations gets aborted on FCV upgrading or downgrading state. But,
+    // due to race between between Instance::getOrCreate() and
+    // SetFeatureCompatibilityVersionCommand::_cancelTenantMigrations(), we might miss aborting this
+    // tenant migration and FCV might have updated or downgraded at this point. So, need to ensure
+    // that the protocol is still compatible with FCV.
+    if (isFCVUpgradingOrDowngrading() || !_checkifProtocolRemainsFCVCompatible()) {
         onReceiveDonorAbortMigration();
     }
 
+    // Any FCV changes after this point will abort this migration.
     auto abortToken = _initAbortMigrationSource(token);
 
     auto recipientTargeterRS = std::make_shared<RemoteCommandTargeterRS>(
diff --git a/src/mongo/db/repl/tenant_migration_donor_service.h b/src/mongo/db/repl/tenant_migration_donor_service.h
index 9663b63d006..ccb233ce526 100644
--- a/src/mongo/db/repl/tenant_migration_donor_service.h
+++ b/src/mongo/db/repl/tenant_migration_donor_service.h
@@ -33,7 +33,6 @@
 #include "mongo/client/fetcher.h"
 #include "mongo/client/remote_command_targeter_rs.h"
 #include "mongo/db/repl/primary_only_service.h"
-#include "mongo/db/repl/repl_server_parameters_gen.h"
 #include "mongo/db/repl/tenant_migration_access_blocker_util.h"
 #include "mongo/executor/thread_pool_task_executor.h"
 #include "mongo/util/cancellation.h"
@@ -144,6 +143,10 @@ public:
             return _stateDoc.getRecipientConnectionString();
         }
 
+        const MigrationProtocolEnum& getProtocol() const {
+            return _protocol;
+        }
+
     private:
         const NamespaceString _stateDocumentsNS = NamespaceString::kTenantMigrationDonorsNamespace;
 
@@ -264,6 +267,12 @@ public:
          */
         CancellationToken _initAbortMigrationSource(const CancellationToken& token);
 
+        /*
+         * Returns false if the protocol is FCV incompatible. Also, resets the 'protocol' field in
+         * the _stateDoc to boost::none for FCV < 5.2.
+         */
+        bool _checkifProtocolRemainsFCVCompatible();
+
         ServiceContext* const _serviceContext;
         const TenantMigrationDonorService* const _donorService;
 
@@ -274,6 +283,7 @@ public:
         // This data is provided in the initial state doc and never changes.  We keep copies to
         // avoid having to obtain the mutex to access them.
         const std::string _tenantId;
+        const MigrationProtocolEnum _protocol;
         const std::string _recipientConnectionString;
         const ReadPreferenceSetting _readPreference;
         const UUID _migrationUuid;
@@ -321,6 +331,10 @@ public:
         // interrupting the instance, e.g. receiving donorAbortMigration. Initialized in
         // _initAbortMigrationSource().
         boost::optional<CancellationSource> _abortMigrationSource;
+
+        // Value is set at the beginning of run() method. Mainly used to determine if the 'protocol'
+        // field needs to be added to recipient migration commands and state document.
+        bool _isAtLeastFCV52AtStart = false;
     };
 
 private:
diff --git a/src/mongo/db/repl/tenant_migration_donor_service_test.cpp b/src/mongo/db/repl/tenant_migration_donor_service_test.cpp
index 53588c3c158..3210b8038a6 100644
--- a/src/mongo/db/repl/tenant_migration_donor_service_test.cpp
+++ b/src/mongo/db/repl/tenant_migration_donor_service_test.cpp
@@ -187,6 +187,7 @@ TEST_F(TenantMigrationDonorServiceTest, CheckSettingMigrationStartDate) {
         "donor-rs/localhost:12345",
         ReadPreferenceSetting(ReadPreference::PrimaryOnly, TagSet::primaryOnly()),
         "tenantA");
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setDonorCertificateForRecipient(kDonorPEMPayload);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
diff --git a/src/mongo/db/repl/tenant_migration_recipient_entry_helpers_test.cpp b/src/mongo/db/repl/tenant_migration_recipient_entry_helpers_test.cpp
index 980f5243591..9a916aa8a16 100644
--- a/src/mongo/db/repl/tenant_migration_recipient_entry_helpers_test.cpp
+++ b/src/mongo/db/repl/tenant_migration_recipient_entry_helpers_test.cpp
@@ -117,6 +117,7 @@ TEST_F(TenantMigrationRecipientEntryHelpersTest, AddTenantMigrationRecipientStat
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    activeTenantAStateDoc.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     activeTenantAStateDoc.setRecipientCertificateForDonor(kRecipientPEMPayload);
     ASSERT_OK(insertStateDoc(opCtx.get(), activeTenantAStateDoc));
     ASSERT_TRUE(checkStateDocPersisted(opCtx.get(), activeTenantAStateDoc));
@@ -127,6 +128,7 @@ TEST_F(TenantMigrationRecipientEntryHelpersTest, AddTenantMigrationRecipientStat
                                                "tenantA",
                                                kDefaultStartMigrationTimestamp,
                                                ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    stateDoc1.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     stateDoc1.setRecipientCertificateForDonor(kRecipientPEMPayload);
     auto status = insertStateDoc(opCtx.get(), stateDoc1);
     ASSERT_EQUALS(ErrorCodes::ConflictingOperationInProgress, status.code());
@@ -138,6 +140,7 @@ TEST_F(TenantMigrationRecipientEntryHelpersTest, AddTenantMigrationRecipientStat
                                                "tenantB",
                                                kDefaultStartMigrationTimestamp,
                                                ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    stateDoc2.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     stateDoc2.setRecipientCertificateForDonor(kRecipientPEMPayload);
     ASSERT_THROWS_CODE(
         insertStateDoc(opCtx.get(), stateDoc2), DBException, ErrorCodes::DuplicateKey);
@@ -149,6 +152,7 @@ TEST_F(TenantMigrationRecipientEntryHelpersTest, AddTenantMigrationRecipientStat
                                                "tenantA",
                                                kDefaultStartMigrationTimestamp,
                                                ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    stateDoc3.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     stateDoc3.setRecipientCertificateForDonor(kRecipientPEMPayload);
     status = insertStateDoc(opCtx.get(), stateDoc3);
     ASSERT_EQUALS(ErrorCodes::ConflictingOperationInProgress, status.code());
@@ -160,6 +164,7 @@ TEST_F(TenantMigrationRecipientEntryHelpersTest, AddTenantMigrationRecipientStat
                                                "tenantB",
                                                kDefaultStartMigrationTimestamp,
                                                ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    stateDoc4.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     stateDoc4.setRecipientCertificateForDonor(kRecipientPEMPayload);
     ASSERT_OK(insertStateDoc(opCtx.get(), stateDoc4));
     ASSERT_TRUE(checkStateDocPersisted(opCtx.get(), stateDoc4));
@@ -176,6 +181,7 @@ TEST_F(TenantMigrationRecipientEntryHelpersTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    inactiveTenantAStateDoc.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     inactiveTenantAStateDoc.setRecipientCertificateForDonor(kRecipientPEMPayload);
     inactiveTenantAStateDoc.setExpireAt(Date_t::now());
     ASSERT_OK(insertStateDoc(opCtx.get(), inactiveTenantAStateDoc));
@@ -187,6 +193,7 @@ TEST_F(TenantMigrationRecipientEntryHelpersTest,
                                                "tenantA",
                                                kDefaultStartMigrationTimestamp,
                                                ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    stateDoc1.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     stateDoc1.setRecipientCertificateForDonor(kRecipientPEMPayload);
     ASSERT_THROWS_CODE(
         insertStateDoc(opCtx.get(), stateDoc1), DBException, ErrorCodes::DuplicateKey);
@@ -198,6 +205,7 @@ TEST_F(TenantMigrationRecipientEntryHelpersTest,
                                                "tenantB",
                                                kDefaultStartMigrationTimestamp,
                                                ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    stateDoc2.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     stateDoc2.setRecipientCertificateForDonor(kRecipientPEMPayload);
     ASSERT_THROWS_CODE(
         insertStateDoc(opCtx.get(), stateDoc2), DBException, ErrorCodes::DuplicateKey);
@@ -209,6 +217,7 @@ TEST_F(TenantMigrationRecipientEntryHelpersTest,
                                                "tenantA",
                                                kDefaultStartMigrationTimestamp,
                                                ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    stateDoc3.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     stateDoc3.setRecipientCertificateForDonor(kRecipientPEMPayload);
     ASSERT_OK(insertStateDoc(opCtx.get(), stateDoc3));
     ASSERT_TRUE(checkStateDocPersisted(opCtx.get(), stateDoc3));
@@ -219,6 +228,7 @@ TEST_F(TenantMigrationRecipientEntryHelpersTest,
                                                "tenantC",
                                                kDefaultStartMigrationTimestamp,
                                                ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    stateDoc4.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     stateDoc4.setRecipientCertificateForDonor(kRecipientPEMPayload);
     ASSERT_OK(insertStateDoc(opCtx.get(), stateDoc4));
     ASSERT_TRUE(checkStateDocPersisted(opCtx.get(), stateDoc4));
diff --git a/src/mongo/db/repl/tenant_migration_recipient_service.cpp b/src/mongo/db/repl/tenant_migration_recipient_service.cpp
index 30c98b48af7..16bda5828f4 100644
--- a/src/mongo/db/repl/tenant_migration_recipient_service.cpp
+++ b/src/mongo/db/repl/tenant_migration_recipient_service.cpp
@@ -313,6 +313,7 @@ TenantMigrationRecipientService::Instance::Instance(
       _stateDoc(TenantMigrationRecipientDocument::parse(IDLParserErrorContext("recipientStateDoc"),
                                                         stateDoc)),
       _tenantId(_stateDoc.getTenantId().toString()),
+      _protocol(_stateDoc.getProtocol().value_or(MigrationProtocolEnum::kMultitenantMigrations)),
       _migrationUuid(_stateDoc.getId()),
       _donorConnectionString(_stateDoc.getDonorConnectionString().toString()),
       _donorUri(uassertStatusOK(MongoURI::parse(_stateDoc.getDonorConnectionString().toString()))),
@@ -415,7 +416,7 @@ Status TenantMigrationRecipientService::Instance::checkIfOptionsConflict(
     stdx::lock_guard<Latch> lg(_mutex);
     invariant(stateDoc.getId() == _migrationUuid);
 
-    if (stateDoc.getTenantId() == _tenantId &&
+    if (stateDoc.getProtocol() == _protocol && stateDoc.getTenantId() == _tenantId &&
         stateDoc.getDonorConnectionString() == _donorConnectionString &&
         stateDoc.getReadPreference().equals(_readPreference) &&
         stateDoc.getRecipientCertificateForDonor() == _recipientCertificateForDonor) {
@@ -1951,6 +1952,33 @@ void TenantMigrationRecipientService::Instance::_compareRecipientAndDonorFCV() c
     }
 }
 
+bool TenantMigrationRecipientService::Instance::_checkifProtocolRemainsFCVCompatible() {
+    stdx::lock_guard<Latch> lg(_mutex);
+
+    // Ensure that the on-disk protocol and cached value remains the same.
+    invariant(!_stateDoc.getProtocol() || _stateDoc.getProtocol().value() == getProtocol());
+
+    auto isAtLeastFCV52AtStart = serverGlobalParams.featureCompatibility.isGreaterThanOrEqualTo(
+        multiversion::FeatureCompatibilityVersion::kVersion_5_2);
+    if (isAtLeastFCV52AtStart) {
+        // When the instance is started using state doc < 5.2 FCV format, _stateDoc._protocol field
+        // won't be set. In that case, the cached value Instance::_protocol will be set to
+        // "kMultitenantMigrations".
+        return true;
+    }
+
+    if (getProtocol() == MigrationProtocolEnum::kShardMerge) {
+        LOGV2(5949504,
+              "Must abort tenant migration as 'Merge' protocol is not supported for FCV "
+              "below 5.2");
+        return false;
+    }
+    // For backward compatibility, ensure that the 'protocol' field is not set in the
+    // document.
+    _stateDoc.setProtocol(boost::none);
+    return true;
+}
+
 SemiFuture<void> TenantMigrationRecipientService::Instance::run(
     std::shared_ptr<executor::ScopedTaskExecutor> executor,
     const CancellationToken& token) noexcept {
@@ -1968,16 +1996,29 @@ SemiFuture<void> TenantMigrationRecipientService::Instance::run(
     pauseBeforeRunTenantMigrationRecipientInstance.pauseWhileSet();
 
     bool cancelWhenDurable = false;
+    auto isFCVUpgradingOrDowngrading = [&]() -> bool {
+        // We must abort the migration if we try to start or resume while upgrading or downgrading.
+        // We defer this until after the state doc is persisted in a started so as to make sure it
+        // it safe to abort and forget the migration. (Generic FCV reference): This FCV check should
+        // exist across LTS binary versions.
+        if (serverGlobalParams.featureCompatibility.isUpgradingOrDowngrading()) {
+            LOGV2(5356304, "Must abort tenant migration as recipient is upgrading or downgrading");
+            return true;
+        }
+        return false;
+    };
 
-    // We must abort the migration if we try to start or resume while upgrading or downgrading.
-    // We defer this until after the state doc is persisted in a started so as to make sure it it
-    // safe to abort and forget the migration.
-    // (Generic FCV reference): This FCV check should exist across LTS binary versions.
-    if (serverGlobalParams.featureCompatibility.isUpgradingOrDowngrading()) {
-        LOGV2(5356304, "Must abort tenant migration as recipient is upgrading or downgrading");
+    // Tenant migrations gets aborted on FCV upgrading or downgrading state. But,
+    // due to race between between Instance::getOrCreate() and
+    // SetFeatureCompatibilityVersionCommand::_cancelTenantMigrations(), we might miss aborting this
+    // tenant migration and FCV might have updated or downgraded at this point. So, need to ensure
+    // that the protocol is still compatible with FCV.
+    if (isFCVUpgradingOrDowngrading() || !_checkifProtocolRemainsFCVCompatible()) {
         cancelWhenDurable = true;
     }
 
+    // Any FCV changes after this point will abort this migration.
+    //
     // The 'AsyncTry' is run on the cleanup executor as opposed to the scoped executor  as we rely
     // on the 'PrimaryService' to interrupt the operation contexts based on thread pool and not the
     // executor.
@@ -2475,5 +2516,9 @@ const std::string& TenantMigrationRecipientService::Instance::getTenantId() cons
     return _tenantId;
 }
 
+const MigrationProtocolEnum& TenantMigrationRecipientService::Instance::getProtocol() const {
+    return _protocol;
+}
+
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/repl/tenant_migration_recipient_service.h b/src/mongo/db/repl/tenant_migration_recipient_service.h
index 7f7889542af..98a6f7d96f2 100644
--- a/src/mongo/db/repl/tenant_migration_recipient_service.h
+++ b/src/mongo/db/repl/tenant_migration_recipient_service.h
@@ -141,6 +141,11 @@ public:
          */
         const std::string& getTenantId() const;
 
+        /*
+         *  Returns the migration protocol.
+         */
+        const MigrationProtocolEnum& getProtocol() const;
+
         /**
          * To be called on the instance returned by PrimaryOnlyService::getOrCreate(). Returns an
          * error if the options this Instance was created with are incompatible with the options
@@ -305,6 +310,12 @@ public:
         };
 
         /*
+         * Returns false if the protocol is FCV incompatible. Also, resets the 'protocol' field in
+         * the _stateDoc to boost::none for FCV < 5.2.
+         */
+        bool _checkifProtocolRemainsFCVCompatible();
+
+        /*
          * Helper for interrupt().
          * The _receivedForgetMigrationPromise is resolved when skipWaitingForForgetMigration is
          * set (e.g. stepDown/shutDown). And we use skipWaitingForForgetMigration=false for
@@ -511,6 +522,7 @@ public:
         // This data is provided in the initial state doc and never changes.  We keep copies to
         // avoid having to obtain the mutex to access them.
         const std::string _tenantId;                                                     // (R)
+        const MigrationProtocolEnum _protocol;                                           // (R)
         const UUID _migrationUuid;                                                       // (R)
         const std::string _donorConnectionString;                                        // (R)
         const MongoURI _donorUri;                                                        // (R)
diff --git a/src/mongo/db/repl/tenant_migration_recipient_service_test.cpp b/src/mongo/db/repl/tenant_migration_recipient_service_test.cpp
index b30dee191a0..cb55a6cbaf7 100644
--- a/src/mongo/db/repl/tenant_migration_recipient_service_test.cpp
+++ b/src/mongo/db/repl/tenant_migration_recipient_service_test.cpp
@@ -480,6 +480,7 @@ TEST_F(TenantMigrationRecipientServiceTest, BasicTenantMigrationRecipientService
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly, TagSet::primaryOnly()));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -505,6 +506,7 @@ TEST_F(TenantMigrationRecipientServiceTest, InstanceReportsErrorOnFailureWhilePe
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly, TagSet::primaryOnly()));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -540,6 +542,7 @@ TEST_F(TenantMigrationRecipientServiceTest, TenantMigrationRecipientConnection_P
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -591,6 +594,7 @@ TEST_F(TenantMigrationRecipientServiceTest, TenantMigrationRecipientConnection_S
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::SecondaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -643,6 +647,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Hang the migration before attempting to connect to clients.
@@ -701,6 +706,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Hang the migration before attempting to connect to clients.
@@ -773,6 +779,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::Nearest));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Hang the migration before attempting to connect to clients.
@@ -834,6 +841,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryPreferred));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Hang the migration before attempting to connect to clients.
@@ -903,6 +911,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryPreferred));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Hang the migration before attempting to connect to clients.
@@ -975,6 +984,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::SecondaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Hang the migration before attempting to connect to clients.
@@ -1036,6 +1046,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::SecondaryPreferred));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Hang the migration before attempting to connect to clients.
@@ -1115,6 +1126,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryPreferred));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
     initialStateDocument.setStartApplyingDonorOpTime(startApplyingOpTime);
 
@@ -1173,6 +1185,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         startMigrationDonorTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryPreferred));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -1234,6 +1247,7 @@ TEST_F(TenantMigrationRecipientServiceTest, TenantMigrationRecipientConnection_P
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryPreferred));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -1278,6 +1292,7 @@ TEST_F(TenantMigrationRecipientServiceTest, TenantMigrationRecipientConnection_B
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -1300,6 +1315,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -1326,6 +1342,7 @@ TEST_F(TenantMigrationRecipientServiceTest, TenantMigrationRecipientGetStartOpTi
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -1365,6 +1382,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -1410,6 +1428,7 @@ TEST_F(TenantMigrationRecipientServiceTest, TenantMigrationRecipientGetStartOpTi
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -1457,6 +1476,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -1493,6 +1513,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.  Fail to populate the remote oplog mock.
@@ -1528,6 +1549,7 @@ TEST_F(TenantMigrationRecipientServiceTest, TenantMigrationRecipientStartOplogFe
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     auto opCtx = makeOperationContext();
@@ -1583,6 +1605,7 @@ TEST_F(TenantMigrationRecipientServiceTest, TenantMigrationRecipientStartsCloner
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     auto opCtx = makeOperationContext();
@@ -1638,6 +1661,7 @@ TEST_F(TenantMigrationRecipientServiceTest, OplogFetcherFailsDuringOplogApplicat
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Skip the cloners in this test, so we provide an empty list of databases.
@@ -1694,6 +1718,7 @@ TEST_F(TenantMigrationRecipientServiceTest, OplogFetcherResumesFromTopOfOplogBuf
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // We skip cloning here as a way to simulate that the recipient service has detected an existing
@@ -1801,6 +1826,7 @@ TEST_F(TenantMigrationRecipientServiceTest, OplogFetcherNoDocInBufferToResumeFro
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // We skip cloning here as a way to simulate that the recipient service has detected an existing
@@ -1910,6 +1936,7 @@ TEST_F(TenantMigrationRecipientServiceTest, OplogApplierResumesFromLastNoOpOplog
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // We skip cloning here as a way to simulate that the recipient service has detected an existing
@@ -2035,6 +2062,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // We skip cloning here as a way to simulate that the recipient service has detected an existing
@@ -2208,6 +2236,7 @@ TEST_F(TenantMigrationRecipientServiceTest, OplogApplierResumesFromStartDonorApp
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // We skip cloning here as a way to simulate that the recipient service has detected an existing
@@ -2353,6 +2382,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // We skip cloning here as a way to simulate that the recipient service has detected an existing
@@ -2459,6 +2489,7 @@ TEST_F(TenantMigrationRecipientServiceTest, OplogApplierFails) {
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Skip the cloners in this test, so we provide an empty list of databases.
@@ -2525,6 +2556,7 @@ TEST_F(TenantMigrationRecipientServiceTest, StoppingApplierAllowsCompletion) {
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Skip the cloners in this test, so we provide an empty list of databases.
@@ -2580,6 +2612,7 @@ TEST_F(TenantMigrationRecipientServiceTest, TenantMigrationRecipientAddResumeTok
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Skip the cloners in this test, so we provide an empty list of databases.
@@ -2683,6 +2716,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientForgetMigration_BeforeRun)
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     auto fp = globalFailPointRegistry().find("pauseBeforeRunTenantMigrationRecipientInstance");
@@ -2721,6 +2755,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientForgetMigration_FailToIniti
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     auto opCtx = makeOperationContext();
@@ -2755,6 +2790,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientForgetMigration_WaitUntilSt
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     auto fp = globalFailPointRegistry().find("pauseAfterRunTenantMigrationRecipientInstance");
@@ -2839,6 +2875,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientForgetMigration_AfterStartO
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -2902,6 +2939,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientForgetMigration_AfterConsis
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Skip the cloners in this test, so we provide an empty list of databases.
@@ -2990,6 +3028,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientForgetMigration_AfterFail)
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Skip the cloners in this test, so we provide an empty list of databases.
@@ -3072,6 +3111,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientForgetMigration_FailToMarkG
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -3124,6 +3164,7 @@ TEST_F(TenantMigrationRecipientServiceTest, TenantMigrationRecipientServiceRecor
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -3159,6 +3200,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Add an FCV value as if it was from a previous attempt.
@@ -3198,6 +3240,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Add an FCV value as if it was from a previous attempt, making sure we set a different
@@ -3245,6 +3288,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -3280,6 +3324,7 @@ TEST_F(TenantMigrationRecipientServiceTest, WaitUntilMigrationReachesReturnAfter
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Skip the cloners in this test, so we provide an empty list of databases.
@@ -3340,6 +3385,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientReceivesRetriableFetcherErr
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -3404,6 +3450,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientReceivesNonRetriableFetcher
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -3463,6 +3510,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientWillNotRetryOnExternalInter
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -3521,6 +3569,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientReceivesRetriableClonerErro
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -3593,6 +3642,7 @@ TEST_F(TenantMigrationRecipientServiceTest, RecipientReceivesNonRetriableClonerE
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
 
     // Create and start the instance.
@@ -3647,6 +3697,7 @@ TEST_F(TenantMigrationRecipientServiceTest, IncrementNumRestartsDueToRecipientFa
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
     // Starting a migration where the state is not 'kUninitialized' indicates that we are restarting
     // from failover.
@@ -3698,6 +3749,7 @@ TEST_F(TenantMigrationRecipientServiceTest,
         "tenantA",
         kDefaultStartMigrationTimestamp,
         ReadPreferenceSetting(ReadPreference::PrimaryOnly));
+    initialStateDocument.setProtocol(MigrationProtocolEnum::kMultitenantMigrations);
     initialStateDocument.setRecipientCertificateForDonor(kRecipientPEMPayload);
     // Starting a migration where the state is not 'kUninitialized' indicates that we are restarting
     // from failover.
diff --git a/src/mongo/db/repl/tenant_migration_state_machine.idl b/src/mongo/db/repl/tenant_migration_state_machine.idl
index f8a6291d5a6..c39dc376418 100644
--- a/src/mongo/db/repl/tenant_migration_state_machine.idl
+++ b/src/mongo/db/repl/tenant_migration_state_machine.idl
@@ -36,6 +36,7 @@ imports:
     - "mongo/client/read_preference_setting.idl"
     - "mongo/db/repl/replication_types.idl"
     - "mongo/db/repl/tenant_migration_pem_payload.idl"
+    - "mongo/db/serverless/serverless_types.idl"
     - "mongo/idl/basic_types.idl"
 
 enums:
@@ -137,6 +138,10 @@ structs:
                 description: >-
                     The wall-clock time at which the migration has started.
                 optional: true
+            protocol:
+                description: "Which migration protocol to use."
+                type: MigrationProtocol
+                optional: true
 
     tenantMigrationRecipientDocument:
         description: "Represents an in-progress tenant migration on the migration recipient."
@@ -171,6 +176,10 @@ structs:
                 description: >-
                     The read preference setting that the recipient will use to determine
                     which node in the donor replica set to clone from.
+            protocol:
+                description: "Which migration protocol to use."
+                type: MigrationProtocol
+                optional: true
             state:
                 type: TenantMigrationRecipientState
                 description: "The state of the tenant migration."
diff --git a/src/mongo/db/repl/tenant_migration_util.h b/src/mongo/db/repl/tenant_migration_util.h
index 18d303d6984..0e3f8284ff8 100644
--- a/src/mongo/db/repl/tenant_migration_util.h
+++ b/src/mongo/db/repl/tenant_migration_util.h
@@ -38,13 +38,17 @@
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/keys_collection_document_gen.h"
 #include "mongo/db/pipeline/pipeline.h"
+#include "mongo/db/repl/repl_server_parameters_gen.h"
 #include "mongo/db/repl/replication_coordinator.h"
+#include "mongo/db/serverless/serverless_types_gen.h"
 #include "mongo/executor/scoped_task_executor.h"
 #include "mongo/util/net/ssl_util.h"
 #include "mongo/util/str.h"
 
 namespace mongo {
 
+constexpr auto kDefaulMigrationProtocol = MigrationProtocolEnum::kMultitenantMigrations;
+
 namespace {
 
 const std::set<std::string> kUnsupportedTenantIds{"", "admin", "local", "config"};
@@ -63,6 +67,26 @@ inline Status validateDatabasePrefix(const std::string& tenantId) {
                  str::stream() << "cannot migrate databases for tenant \'" << tenantId << "'");
 }
 
+inline Status validateProtocolFCVCompatibility(
+    const boost::optional<MigrationProtocolEnum>& protocol) {
+    if (!protocol)
+        return Status::OK();
+
+    if (!serverGlobalParams.featureCompatibility.isGreaterThanOrEqualTo(
+            multiversion::FeatureCompatibilityVersion::kVersion_5_2)) {
+        return Status(ErrorCodes::InvalidOptions,
+                      str::stream() << "'protocol' field is not supported for FCV below 5.2'");
+    }
+
+    if (*protocol == MigrationProtocolEnum::kShardMerge &&
+        !repl::feature_flags::gShardMerge.isEnabled(serverGlobalParams.featureCompatibility)) {
+        return Status(ErrorCodes::IllegalOperation,
+                      str::stream() << "protocol '" << MigrationProtocol_serializer(*protocol)
+                                    << "' not supported");
+    }
+    return Status::OK();
+}
+
 inline Status validateTimestampNotNull(const Timestamp& ts) {
     return (!ts.isNull())
         ? Status::OK()
@@ -128,6 +152,28 @@ inline Status validatePrivateKeyPEMPayload(const StringData& payload) {
 #endif
 }
 
+inline void protocolTenantIdCompatibilityCheck(const MigrationProtocolEnum& protocol,
+                                               const std::string& tenantId) {
+    switch (protocol) {
+        case MigrationProtocolEnum::kShardMerge: {
+            // TODO SERVER-59794: Add a check to ensure tenantId is not provided for 'Merge'
+            // protocol.
+            break;
+        }
+        case MigrationProtocolEnum::kMultitenantMigrations: {
+            uassert(ErrorCodes::InvalidOptions,
+                    str::stream() << "'tenantId' is required for protocol '"
+                                  << MigrationProtocol_serializer(protocol) << "'",
+                    !tenantId.empty());
+
+
+            break;
+        }
+        default:
+            MONGO_UNREACHABLE;
+    }
+}
+
 /*
  * Creates an ExternalKeysCollectionDocument representing an config.external_validation_keys
  * document from the given the admin.system.keys document BSONObj.
diff --git a/src/mongo/db/serverless/SConscript b/src/mongo/db/serverless/SConscript
index 0f3259271f6..b2ebb792ae7 100644
--- a/src/mongo/db/serverless/SConscript
+++ b/src/mongo/db/serverless/SConscript
@@ -4,6 +4,18 @@ Import("env")
 env = env.Clone()
 
 env.Library(
+    target='serverless_types_idl',
+    source=[
+        'serverless_types.idl',
+    ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/idl/basic_types',
+        '$BUILD_DIR/mongo/idl/idl_parser'
+    ],
+)
+
+env.Library(
     target='tenant_split_state_machine',
     source=[
         'tenant_split_state_machine.idl',
diff --git a/src/mongo/db/serverless/serverless_types.idl b/src/mongo/db/serverless/serverless_types.idl
new file mode 100644
index 00000000000..e906e7a52c3
--- /dev/null
+++ b/src/mongo/db/serverless/serverless_types.idl
@@ -0,0 +1,40 @@
+# Copyright (C) 2021-present MongoDB, Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the Server Side Public License, version 1,
+# as published by MongoDB, Inc.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# Server Side Public License for more details.
+#
+# You should have received a copy of the Server Side Public License
+# along with this program. If not, see
+# <http://www.mongodb.com/licensing/server-side-public-license>.
+#
+# As a special exception, the copyright holders give permission to link the
+# code of portions of this program with the OpenSSL library under certain
+# conditions as described in each individual source file and distribute
+# linked combinations including the program with the OpenSSL library. You
+# must comply with the Server Side Public License in all respects for
+# all of the code used other than as permitted herein. If you modify file(s)
+# with this exception, you may extend this exception to your version of the
+# file(s), but you are not obligated to do so. If you do not wish to do so,
+# delete this exception statement from your version. If you delete this
+# exception statement from all source files in the program, then also delete
+# it in the license file.
+#
+global:
+    cpp_namespace: "mongo"
+
+imports:
+    - "mongo/idl/basic_types.idl"
+
+enums:
+    MigrationProtocol:
+        description: "Determines which tenant migration protocol to use."
+        type: string
+        values:
+            kMultitenantMigrations: "multitenant migrations"
+            kShardMerge: "shard merge"