SERVER-48796 Implement recipientSyncData command in MigrationServiceInstance.

2 files changed, 178 insertions, 17 deletions

diff --git a/jstests/replsets/tenant_migration_conflicting_recipient_sync_data_cmds.js b/jstests/replsets/tenant_migration_conflicting_recipient_sync_data_cmds.js
new file mode 100644
index 00000000000..e9d0208e9e3
--- /dev/null
+++ b/jstests/replsets/tenant_migration_conflicting_recipient_sync_data_cmds.js
@@ -0,0 +1,105 @@
+/**
+ * Test that tenant migration recipient rejects conflicting recipientSyncData commands.
+ *
+ * @tags: [requires_fcv_47, requires_majority_read_concern, incompatible_with_eft]
+ */
+(function() {
+
+"use strict";
+load("jstests/libs/parallel_shell_helpers.js");
+load("jstests/libs/curop_helpers.js");  // for waitForCurOpByFailPoint().
+
+var rst = new ReplSetTest({nodes: 1, nodeOptions: {setParameter: {enableTenantMigrations: true}}});
+rst.startSet();
+rst.initiate();
+
+const primary = rst.getPrimary();
+const configDB = primary.getDB("config");
+const tenantMigrationRecipientStateColl = configDB["tenantMigrationRecipients"];
+const tenantMigrationRecipientStateCollNss = tenantMigrationRecipientStateColl.getFullName();
+
+const tenantId = "test";
+const connectionString = "foo/bar:12345";
+const readPreference = {
+    mode: 'primary'
+};
+
+function checkTenantMigrationRecipientStateCollCount(expectedCount) {
+    let res = tenantMigrationRecipientStateColl.find().toArray();
+    assert.eq(expectedCount,
+              res.length,
+              "'config.tenantMigrationRecipients' collection count mismatch: " + tojson(res));
+}
+
+function startRecipientSyncDataCmd(migrationUuid, tenantId, connectionString, readPreference) {
+    jsTestLog("Starting a recipientSyncDataCmd for migrationUuid: " + migrationUuid +
+              " tenantId: '" + tenantId + "'");
+    assert.commandWorkedOrFailedWithCode(db.adminCommand({
+        recipientSyncData: 1,
+        migrationId: migrationUuid,
+        donorConnectionString: connectionString,
+        tenantId: tenantId,
+        readPreference: readPreference
+    }),
+                                         ErrorCodes.ConflictingOperationInProgress);
+}
+
+// Enable the failpoint to stop the tenant migration after persisting the state doc.
+assert.commandWorked(primary.adminCommand({
+    configureFailPoint: "fpAfterPersistingTenantMigrationRecipientInstanceStateDoc",
+    mode: "alwaysOn",
+    data: {action: "stop"}
+}));
+
+{
+    // Enable the failpoint before inserting the state document by upsert command.
+    assert.commandWorked(primary.adminCommand(
+        {configureFailPoint: "hangBeforeUpsertPerformsInsert", mode: "alwaysOn"}));
+
+    // Sanity check : 'config.tenantMigrationRecipients' collection count should be empty.
+    checkTenantMigrationRecipientStateCollCount(0);
+
+    // Start the  conflicting recipientSyncData cmds.
+    const recipientSyncDataCmd1 = startParallelShell(
+        funWithArgs(startRecipientSyncDataCmd, UUID(), tenantId, connectionString, readPreference),
+        primary.port);
+    const recipientSyncDataCmd2 = startParallelShell(
+        funWithArgs(startRecipientSyncDataCmd, UUID(), tenantId, connectionString, readPreference),
+        primary.port);
+
+    // Wait until both the conflicting instances got started.
+    checkLog.containsWithCount(primary, "Starting tenant migration recipient instance", 2);
+
+    jsTestLog("Waiting for 'hangBeforeUpsertPerformsInsert' failpoint to reach");
+    waitForCurOpByFailPoint(
+        configDB, tenantMigrationRecipientStateCollNss, "hangBeforeUpsertPerformsInsert");
+
+    // Unblock the tenant migration instance from persisting the state doc.
+    assert.commandWorked(
+        primary.adminCommand({configureFailPoint: "hangBeforeUpsertPerformsInsert", mode: "off"}));
+
+    // Wait for both the conflicting instances to complete.
+    recipientSyncDataCmd1();
+    recipientSyncDataCmd2();
+
+    // Only one instance should have succeeded in persisting the state doc, other should have failed
+    // with ErrorCodes.ConflictingOperationInProgress.
+    checkTenantMigrationRecipientStateCollCount(1);
+}
+
+{
+    // Now, again call recipientSyncData cmd  to run on the same tenant "test'. Since, our previous
+    // instance for  tenant "test' wasn't garbage collected, the migration status for that tenant is
+    // considered as active. So, this command should fail with
+    // ErrorCodes.ConflictingOperationInProgress.
+    const recipientSyncDataCmd3 = startParallelShell(
+        funWithArgs(startRecipientSyncDataCmd, UUID(), tenantId, connectionString, readPreference),
+        primary.port);
+    recipientSyncDataCmd3();
+
+    // Collection count should remain the same.
+    checkTenantMigrationRecipientStateCollCount(1);
+}
+
+rst.stopSet();
+})();
diff --git a/jstests/replsets/tenant_migration_invalid_inputs.js b/jstests/replsets/tenant_migration_invalid_inputs.js
index 415c8261270..443cfb2fb14 100644
--- a/jstests/replsets/tenant_migration_invalid_inputs.js
+++ b/jstests/replsets/tenant_migration_invalid_inputs.js
@@ -1,7 +1,7 @@
 /**
- * Tests that the donorStartMigration command throws a error if the provided tenantId
- * is unsupported (i.e. '', 'admin', 'local' or 'config') or if the recipient connection string
- * matches the donor's connection string.
+ * Tests that the donorStartMigration & recipientSyncData command throws an error if the provided
+ * tenantId is unsupported (i.e. '', 'admin', 'local' or 'config') or if the recipient
+ * connection string matches the donor's connection string.
  *
  * @tags: [requires_fcv_47]
  */
@@ -14,17 +14,23 @@ const rst =
 rst.startSet();
 rst.initiate();
 const primary = rst.getPrimary();
+const tenantId = "test";
+const connectionString = "foo/bar:12345";
+const readPreference = {
+    mode: 'primary'
+};
 
-// Test unsupported tenantIds.
-const unsupportedTenantIds = ['', 'admin', 'local', 'config'];
+jsTestLog("Testing 'donorStartMigration' command provided with invalid options.");
 
-unsupportedTenantIds.forEach((tenantId) => {
+// Test unsupported database prefixes.
+const unsupportedtenantIds = ['', 'admin', 'local', 'config'];
+unsupportedtenantIds.forEach((invalidTenantId) => {
     assert.commandFailedWithCode(primary.adminCommand({
         donorStartMigration: 1,
         migrationId: UUID(),
-        recipientConnectionString: "testRecipientConnString",
-        tenantId: tenantId,
-        readPreference: {mode: "primary"}
+        recipientConnectionString: connectionString,
+        tenantId: invalidTenantId,
+        readPreference: readPreference
     }),
                                  ErrorCodes.BadValue);
 });
@@ -34,20 +40,70 @@ assert.commandFailedWithCode(primary.adminCommand({
     donorStartMigration: 1,
     migrationId: UUID(),
     recipientConnectionString: rst.getURL(),
-    tenantId: "testTenantId",
-    readPreference: {mode: "primary"}
+    tenantId: tenantId,
+    readPreference: readPreference
 }),
-                             ErrorCodes.InvalidOptions);
+                             ErrorCodes.BadValue);
 
-// Test migrating a database to a recipient that has one or more same hosts as donor
-const conflictingRecipientConnectionString = "foo/bar:12345," + primary.host;
+// Test migrating a database to a recipient that has one or more same hosts as donor.
+const conflictingRecipientConnectionString = connectionString + "," + primary.host;
 assert.commandFailedWithCode(primary.adminCommand({
     donorStartMigration: 1,
     migrationId: UUID(),
     recipientConnectionString: conflictingRecipientConnectionString,
-    tenantId: "testTenantId",
-    readPreference: {mode: "primary"}
+    tenantId: tenantId,
+    readPreference: readPreference
+}),
+                             ErrorCodes.BadValue);
+
+jsTestLog("Testing 'recipientSyncData' command provided with invalid options.");
+
+// Test unsupported database prefixes.
+unsupportedtenantIds.forEach((invalidTenantId) => {
+    assert.commandFailedWithCode(primary.adminCommand({
+        recipientSyncData: 1,
+        migrationId: UUID(),
+        donorConnectionString: connectionString,
+        tenantId: invalidTenantId,
+        readPreference: readPreference
+    }),
+                                 ErrorCodes.BadValue);
+});
+
+// Test migrating a database from recipient itself.
+assert.commandFailedWithCode(primary.adminCommand({
+    recipientSyncData: 1,
+    migrationId: UUID(),
+    donorConnectionString: rst.getURL(),
+    tenantId: tenantId,
+    readPreference: readPreference
 }),
-                             ErrorCodes.InvalidOptions);
+                             ErrorCodes.BadValue);
+
+// Test migrating a database from a donor that has one or more same hosts as recipient.
+const conflictingDonorConnectionString = connectionString + "," + primary.host;
+assert.commandFailedWithCode(primary.adminCommand({
+    recipientSyncData: 1,
+    migrationId: UUID(),
+    donorConnectionString: conflictingDonorConnectionString,
+    tenantId: tenantId,
+    readPreference: readPreference
+}),
+                             ErrorCodes.BadValue);
+
+// Test 'returnAfterReachingTimestamp' can' be null.
+const nullTimestamps = [Timestamp(0, 0), Timestamp(0, 1)];
+nullTimestamps.forEach((nullTs) => {
+    assert.commandFailedWithCode(primary.adminCommand({
+        recipientSyncData: 1,
+        migrationId: UUID(),
+        donorConnectionString: connectionString,
+        tenantId: tenantId,
+        readPreference: readPreference,
+        returnAfterReachingTimestamp: nullTs
+    }),
+                                 ErrorCodes.BadValue);
+});
+
 rst.stopSet();
 })();
\ No newline at end of file