SERVER-63762 Track initial number of orphans in range deletion task documents (receiver-side)

6 files changed, 154 insertions, 2 deletions

diff --git a/jstests/sharding/migration_tracks_orphans.js b/jstests/sharding/migration_tracks_orphans.js
new file mode 100644
index 00000000000..0724b3ebcc2
--- /dev/null
+++ b/jstests/sharding/migration_tracks_orphans.js
@@ -0,0 +1,103 @@
+/**
+ * Tests that the recipient in a migration correctly tracks the number of orphans during cloning and
+ * transfer mods. The count of orphan documents should be persisted in the range deletion document
+ * in config.rangeDeletions.
+ *
+ * @tags: [
+ *  requires_fcv_60,
+ * ]
+ */
+
+(function() {
+'use strict';
+
+load("jstests/libs/fail_point_util.js");
+load('jstests/libs/parallel_shell_helpers.js');
+
+const st = new ShardingTest({shards: 2});
+
+// Setup database and collection for test
+const dbName = 'db';
+const db = st.getDB(dbName);
+assert.commandWorked(
+    st.s.adminCommand({enableSharding: dbName, primaryShard: st.shard0.shardName}));
+const coll = db['test'];
+const nss = coll.getFullName();
+assert.commandWorked(st.s.adminCommand({shardCollection: nss, key: {_id: 1}}));
+
+function assertOrphanCountIsCorrect(conn, ns, numOrphans) {
+    const rangeDeletionDoc =
+        conn.getDB("config").getCollection("rangeDeletions").findOne({nss: ns});
+    assert.neq(null,
+               rangeDeletionDoc,
+               "did not find document for namespace " + ns +
+                   ", contents of config.rangeDeletions on " + conn + ": " +
+                   tojson(conn.getDB("config").getCollection("rangeDeletions").find().toArray()));
+    assert.eq(numOrphans,
+              rangeDeletionDoc.numOrphanDocs,
+              "Incorrect count of orphaned documents in config.rangeDeletions on " + conn +
+                  ": expected " + numOrphans +
+                  " orphaned documents but found range deletion document " +
+                  tojson(rangeDeletionDoc));
+}
+
+function assertEventuallyDoesNotHaveRangeDeletionDoc(conn) {
+    assert.soon(() => {
+        return 0 == conn.getDB("config").getCollection("rangeDeletions").find().itcount();
+    });
+}
+
+// Insert some docs into the collection.
+const numDocs = 1000;
+let bulk = coll.initializeUnorderedBulkOp();
+for (let i = 0; i < numDocs; i++) {
+    bulk.insert({_id: i});
+}
+assert.commandWorked(bulk.execute());
+
+// Pause after bulk clone and check number of orphans is equal to numDocs
+let bulkCloneFailpoint = configureFailPoint(st.shard1, "migrateThreadHangAtStep4");
+const awaitResult = startParallelShell(
+    funWithArgs(function(nss, toShardName) {
+        assert.commandWorked(db.adminCommand({moveChunk: nss, find: {_id: 0}, to: toShardName}));
+    }, nss, st.shard1.shardName), st.s.port);
+
+// Assert that the range deletion document is present and has the correct number of orphans.
+bulkCloneFailpoint.wait();
+assertOrphanCountIsCorrect(st.shard1, nss, numDocs);
+
+// Pause after transfer mods and check number of orphans has changed correctly.
+let transferModsFailpoint = configureFailPoint(st.shard1, "migrateThreadHangAtStep5");
+
+// Perform some upserts and deletes to change the number of orphans on the recipient.
+let bulkMods = coll.initializeUnorderedBulkOp();
+const numUpserts = 50;
+for (let i = 0; i < numUpserts; i++) {
+    let key = numDocs + i;
+    bulkMods.find({_id: key}).upsert().update({$set: {_id: key}});
+}
+const numDeletes = 25;
+for (let i = 0; i < numDeletes; i++) {
+    bulkMods.find({_id: i}).removeOne();
+}
+// Perform some updates that shouldn't change the number of orphans.
+const numUpdates = 10;
+for (let i = 0; i < numUpdates; i++) {
+    let key = numDeletes + i;
+    bulkMods.find({_id: key}).update({$set: {x: key}});
+}
+assert.commandWorked(bulkMods.execute());
+
+// Assert that the number of orphans is still correct.
+bulkCloneFailpoint.off();
+transferModsFailpoint.wait();
+const updatedOrphanCount = numDocs + numUpserts - numDeletes;
+assertOrphanCountIsCorrect(st.shard1, nss, updatedOrphanCount);
+transferModsFailpoint.off();
+
+// Allow migration to finish and clean up range deletions
+awaitResult();
+assertEventuallyDoesNotHaveRangeDeletionDoc(st.shard1);
+
+st.stop();
+})();
diff --git a/src/mongo/db/s/migration_destination_manager.cpp b/src/mongo/db/s/migration_destination_manager.cpp
index 364195ee50b..45cec7065e3 100644
--- a/src/mongo/db/s/migration_destination_manager.cpp
+++ b/src/mongo/db/s/migration_destination_manager.cpp
@@ -1316,6 +1316,9 @@ void MigrationDestinationManager::_migrateDriver(OperationContext* outerOpCtx,
                                                        << " failed.");
                     }
 
+                    migrationutil::persistUpdatedNumOrphans(
+                        opCtx, BSON("_id" << _migrationId.get()), batchNumCloned);
+
                     {
                         stdx::lock_guard<Latch> statsLock(_mutex);
                         _numCloned += batchNumCloned;
@@ -1635,6 +1638,7 @@ void MigrationDestinationManager::_migrateDriver(OperationContext* outerOpCtx,
 
 bool MigrationDestinationManager::_applyMigrateOp(OperationContext* opCtx, const BSONObj& xfer) {
     bool didAnything = false;
+    int changeInOrphans = 0;
 
     // Deleted documents
     if (xfer["deleted"].isABSONObj()) {
@@ -1678,6 +1682,7 @@ bool MigrationDestinationManager::_applyMigrateOp(OperationContext* opCtx, const
                               true /* fromMigrate */);
             });
 
+            changeInOrphans--;
             didAnything = true;
         }
     }
@@ -1726,13 +1731,20 @@ bool MigrationDestinationManager::_applyMigrateOp(OperationContext* opCtx, const
 
             // We are in write lock here, so sure we aren't killing
             writeConflictRetry(opCtx, "transferModsUpdates", _nss.ns(), [&] {
-                Helpers::upsert(opCtx, _nss.ns(), updatedDoc, true);
+                auto res = Helpers::upsert(opCtx, _nss.ns(), updatedDoc, true);
+                if (!res.upsertedId.isEmpty()) {
+                    changeInOrphans++;
+                }
             });
 
             didAnything = true;
         }
     }
 
+    if (changeInOrphans != 0) {
+        migrationutil::persistUpdatedNumOrphans(
+            opCtx, BSON("_id" << _migrationId.get()), changeInOrphans);
+    }
     return didAnything;
 }
 
diff --git a/src/mongo/db/s/migration_util.cpp b/src/mongo/db/s/migration_util.cpp
index 219236c5e7a..dffa086da60 100644
--- a/src/mongo/db/s/migration_util.cpp
+++ b/src/mongo/db/s/migration_util.cpp
@@ -667,6 +667,17 @@ void persistRangeDeletionTaskLocally(OperationContext* opCtx,
     }
 }
 
+void persistUpdatedNumOrphans(OperationContext* opCtx,
+                              const BSONObj& rangeDeletionQuery,
+                              const int& changeInOrphans) {
+    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
+    store.update(
+        opCtx,
+        rangeDeletionQuery,
+        BSON("$inc" << BSON(RangeDeletionTask::kNumOrphanDocsFieldName << changeInOrphans)),
+        WriteConcernOptions());
+}
+
 void persistCommitDecision(OperationContext* opCtx,
                            const MigrationCoordinatorDocument& migrationDoc) {
     invariant(migrationDoc.getDecision() &&
diff --git a/src/mongo/db/s/migration_util.h b/src/mongo/db/s/migration_util.h
index ff2add3ba09..43d2c71cff4 100644
--- a/src/mongo/db/s/migration_util.h
+++ b/src/mongo/db/s/migration_util.h
@@ -144,6 +144,14 @@ void persistRangeDeletionTaskLocally(OperationContext* opCtx,
                                      const WriteConcernOptions& writeConcern);
 
 /**
+ * Updates the range deletion task document to increase or decrease numOrphanedDocs and waits for
+ * write concern.
+ */
+void persistUpdatedNumOrphans(OperationContext* opCtx,
+                              const BSONObj& rangeDeletionQuery,
+                              const int& changeInOrphans);
+
+/**
  * Updates the migration coordinator document to set the decision field to "committed" and waits for
  * majority writeConcern.
  */
diff --git a/src/mongo/db/s/migration_util_test.cpp b/src/mongo/db/s/migration_util_test.cpp
index ff85cb6858c..61d78b8359b 100644
--- a/src/mongo/db/s/migration_util_test.cpp
+++ b/src/mongo/db/s/migration_util_test.cpp
@@ -339,6 +339,24 @@ TEST_F(MigrationUtilsTest, TestInvalidUUID) {
     ASSERT_FALSE(migrationutil::checkForConflictingDeletions(opCtx, range, wrongUuid));
 }
 
+TEST_F(MigrationUtilsTest, TestUpdateNumberOfOrphans) {
+    auto opCtx = operationContext();
+    const auto uuid = UUID::gen();
+    PersistentTaskStore<RangeDeletionTask> store(NamespaceString::kRangeDeletionNamespace);
+    auto rangeDeletionDoc = createDeletionTask(opCtx, kTestNss, uuid, 0, 10);
+    store.add(opCtx, rangeDeletionDoc);
+
+    auto rangeDeletionQuery = BSON("_id" << rangeDeletionDoc.getId());
+
+    migrationutil::persistUpdatedNumOrphans(opCtx, rangeDeletionQuery, 5);
+    rangeDeletionDoc.setNumOrphanDocs(5);
+    ASSERT_EQ(store.count(opCtx, rangeDeletionDoc.toBSON().removeField("timestamp")), 1);
+
+    migrationutil::persistUpdatedNumOrphans(opCtx, rangeDeletionQuery, -5);
+    rangeDeletionDoc.setNumOrphanDocs(0);
+    ASSERT_EQ(store.count(opCtx, rangeDeletionDoc.toBSON().removeField("timestamp")), 1);
+}
+
 /**
  * Fixture that uses a mocked CatalogCacheLoader and CatalogClient to allow metadata refreshes
  * without using the mock network.
diff --git a/src/mongo/db/s/range_deletion_task.idl b/src/mongo/db/s/range_deletion_task.idl
index 2c368b5fe90..611e2170cb6 100644
--- a/src/mongo/db/s/range_deletion_task.idl
+++ b/src/mongo/db/s/range_deletion_task.idl
@@ -80,5 +80,5 @@ structs:
                 optional: true
             numOrphanDocs:
                 type: safeInt64
+                default: 0
                 description: "The estimated number of orphaned documents in the range"
-                optional: true  ## TODO SERVER-63819 non-optional once 6.0 branches out