SERVER-69603 Do not report drop events as part of shardCollection

(cherry picked from commit 22afc187e2a5ed450cf608458393830eb40b913a)

16 files changed, 173 insertions, 58 deletions

diff --git a/jstests/sharding/change_stream_no_drop.js b/jstests/sharding/change_stream_no_drop.js
new file mode 100644
index 00000000000..d378421cd18
--- /dev/null
+++ b/jstests/sharding/change_stream_no_drop.js
@@ -0,0 +1,62 @@
+/**
+ * DDL coordinator are responsible for dropping temporary collections, especially after failures.
+ * However, the change stream should not be aware of those events.
+ * @tags: [
+ *  # Requires all nodes to be running the latest binary.
+ *  multiversion_incompatible,
+ * ]
+ */
+function assertNoDrop(changeStream) {
+    while (changeStream.hasNext()) {
+        assert.neq(changeStream.next().operationType, 'drop');
+    }
+}
+
+function emptyChangeStream(changeStream) {
+    while (changeStream.hasNext()) {
+        changeStream.next();
+    }
+}
+
+(function() {
+
+const dbName = 'db';
+
+load('jstests/libs/fail_point_util.js');  // For configureFailPoint
+
+// Enable explicitly the periodic no-op writer to allow the router to process change stream events
+// coming from all shards. This is enabled for production clusters by default.
+const st = new ShardingTest({
+    mongos: 1,
+    config: 1,
+    shards: 2,
+    rs: {nodes: 1, setParameter: {writePeriodicNoops: true, periodicNoopIntervalSecs: 1}},
+    other: {enableBalancer: true}
+});
+
+// create a database and a change stream on it
+jsTest.log('Creating a change stream on ' + dbName);
+assert.commandWorked(
+    st.s.adminCommand({enableSharding: dbName, primaryShard: st.shard0.shardName}));
+let changeStream = st.s.getDB('db').watch();
+
+// setFeatureCompatibilityVersion might cause dropping of deprecated collections
+emptyChangeStream(changeStream);
+
+jsTest.log(
+    'The shard_collection_coordinator at second attempt (after failure) should not report drop events for orphaned');
+{
+    configureFailPoint(st.shard0,
+                       'failAtCommitCreateCollectionCoordinator',
+                       data = {},
+                       failPointMode = {times: 1});
+
+    collectionName = dbName + '.coll';
+    assert.commandWorked(st.s.adminCommand(
+        {shardCollection: collectionName, key: {_id: "hashed"}, numInitialChunks: 10}));
+
+    assertNoDrop(changeStream);
+}
+
+st.stop();
+}());
diff --git a/src/mongo/db/s/create_collection_coordinator.cpp b/src/mongo/db/s/create_collection_coordinator.cpp
index e0fb9f87ff2..31d302b9989 100644
--- a/src/mongo/db/s/create_collection_coordinator.cpp
+++ b/src/mongo/db/s/create_collection_coordinator.cpp
@@ -59,6 +59,11 @@
 #include "mongo/s/grid.h"
 #include "mongo/s/sharding_feature_flags_gen.h"
 
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
+MONGO_FAIL_POINT_DEFINE(failAtCommitCreateCollectionCoordinator);
+
+
 namespace mongo {
 namespace {
 
@@ -342,8 +347,8 @@ void broadcastDropCollection(OperationContext* opCtx,
                              const NamespaceString& nss,
                              const std::shared_ptr<executor::TaskExecutor>& executor,
                              const OperationSessionInfo& osi) {
-    const auto primaryShardId = ShardingState::get(opCtx)->shardId();
     const ShardsvrDropCollectionParticipant dropCollectionParticipant(nss);
+    const auto primaryShardId = ShardingState::get(opCtx)->shardId();
 
     auto participants = Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx);
     // Remove primary shard from participants
@@ -351,7 +356,7 @@ void broadcastDropCollection(OperationContext* opCtx,
                        participants.end());
 
     sharding_ddl_util::sendDropCollectionParticipantCommandToShards(
-        opCtx, nss, participants, executor, osi);
+        opCtx, nss, participants, executor, osi, true /* fromMigrate */);
 }
 
 /**
@@ -873,6 +878,12 @@ void CreateCollectionCoordinator::_createCollectionOnNonPrimaryShards(
 void CreateCollectionCoordinator::_commit(OperationContext* opCtx) {
     LOGV2_DEBUG(5277906, 2, "Create collection _commit", "namespace"_attr = nss());
 
+    if (MONGO_unlikely(failAtCommitCreateCollectionCoordinator.shouldFail())) {
+        LOGV2_DEBUG(6960301, 2, "About to hit failAtCommitCreateCollectionCoordinator fail point");
+        uasserted(ErrorCodes::InterruptedAtShutdown,
+                  "failAtCommitCreateCollectionCoordinator fail point");
+    }
+
     // Upsert Chunks.
     _doc = _updateSession(opCtx, _doc);
     insertChunks(opCtx, _initialChunks->chunks, getCurrentSession(_doc));
diff --git a/src/mongo/db/s/drop_collection_coordinator.cpp b/src/mongo/db/s/drop_collection_coordinator.cpp
index a4abb140d6c..76b86dfeec8 100644
--- a/src/mongo/db/s/drop_collection_coordinator.cpp
+++ b/src/mongo/db/s/drop_collection_coordinator.cpp
@@ -76,8 +76,9 @@ boost::optional<BSONObj> DropCollectionCoordinator::reportForCurrentOp(
     return bob.obj();
 }
 
-DropReply DropCollectionCoordinator::dropCollectionLocally(OperationContext* opCtx,
-                                                           const NamespaceString& nss) {
+void DropCollectionCoordinator::dropCollectionLocally(OperationContext* opCtx,
+                                                      const NamespaceString& nss,
+                                                      bool fromMigrate) {
     {
         // Clear CollectionShardingRuntime entry
         Lock::DBLock dbLock(opCtx, nss.db(), MODE_IX);
@@ -86,17 +87,21 @@ DropReply DropCollectionCoordinator::dropCollectionLocally(OperationContext* opC
         csr->clearFilteringMetadataForDroppedCollection(opCtx);
     }
 
-    DropReply result;
-    uassertStatusOK(dropCollection(
-        opCtx, nss, &result, DropCollectionSystemCollectionMode::kDisallowSystemCollectionDrops));
+    DropReply unused;
+    if (fromMigrate)
+        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, nss);
+    else
+        uassertStatusOK(
+            dropCollection(opCtx,
+                           nss,
+                           &unused,
+                           DropCollectionSystemCollectionMode::kDisallowSystemCollectionDrops));
 
     // Force the refresh of the catalog cache to purge outdated information
     const auto catalog = Grid::get(opCtx)->catalogCache();
     uassertStatusOK(catalog->getCollectionRoutingInfoWithRefresh(opCtx, nss));
     CatalogCacheLoader::get(opCtx).waitForCollectionFlush(opCtx, nss);
     repl::ReplClientInfo::forClient(opCtx->getClient()).setLastOpToSystemLastOpTime(opCtx);
-
-    return result;
 }
 
 void DropCollectionCoordinator::_enterPhase(Phase newPhase) {
@@ -215,13 +220,23 @@ ExecutorFuture<void> DropCollectionCoordinator::_runImpl(
                     participants.end());
 
                 sharding_ddl_util::sendDropCollectionParticipantCommandToShards(
-                    opCtx, nss(), participants, **executor, getCurrentSession(_doc));
+                    opCtx,
+                    nss(),
+                    participants,
+                    **executor,
+                    getCurrentSession(_doc),
+                    false /*fromMigrate*/);
 
                 // The sharded collection must be dropped on the primary shard after it has been
                 // dropped on all of the other shards to ensure it can only be re-created as
                 // unsharded with a higher optime than all of the drops.
                 sharding_ddl_util::sendDropCollectionParticipantCommandToShards(
-                    opCtx, nss(), {primaryShardId}, **executor, getCurrentSession(_doc));
+                    opCtx,
+                    nss(),
+                    {primaryShardId},
+                    **executor,
+                    getCurrentSession(_doc),
+                    false /*fromMigrate*/);
 
                 ShardingLogging::get(opCtx)->logChange(opCtx, "dropCollection", nss().ns());
                 LOGV2(5390503, "Collection dropped", "namespace"_attr = nss());
diff --git a/src/mongo/db/s/drop_collection_coordinator.h b/src/mongo/db/s/drop_collection_coordinator.h
index 140013e41e1..b77b8b55cc1 100644
--- a/src/mongo/db/s/drop_collection_coordinator.h
+++ b/src/mongo/db/s/drop_collection_coordinator.h
@@ -52,8 +52,12 @@ public:
     /**
      * Locally drops a collection, cleans its CollectionShardingRuntime metadata and refreshes the
      * catalog cache.
+     * The oplog entry associated with the drop collection will be generated with the fromMigrate
+     * flag.
      */
-    static DropReply dropCollectionLocally(OperationContext* opCtx, const NamespaceString& nss);
+    static void dropCollectionLocally(OperationContext* opCtx,
+                                      const NamespaceString& nss,
+                                      bool fromMigrate);
 
 private:
     ShardingDDLCoordinatorMetadata const& metadata() const override {
diff --git a/src/mongo/db/s/drop_database_coordinator.cpp b/src/mongo/db/s/drop_database_coordinator.cpp
index 3645f186d72..a83a50ce75a 100644
--- a/src/mongo/db/s/drop_database_coordinator.cpp
+++ b/src/mongo/db/s/drop_database_coordinator.cpp
@@ -133,13 +133,13 @@ void DropDatabaseCoordinator::_dropShardedCollection(
     participants.erase(std::remove(participants.begin(), participants.end(), primaryShardId),
                        participants.end());
     sharding_ddl_util::sendDropCollectionParticipantCommandToShards(
-        opCtx, nss, participants, **executor, getCurrentSession(_doc));
+        opCtx, nss, participants, **executor, getCurrentSession(_doc), false /* fromMigrate */);
 
     // The sharded collection must be dropped on the primary shard after it has been dropped on all
     // of the other shards to ensure it can only be re-created as unsharded with a higher optime
     // than all of the drops.
     sharding_ddl_util::sendDropCollectionParticipantCommandToShards(
-        opCtx, nss, {primaryShardId}, **executor, getCurrentSession(_doc));
+        opCtx, nss, {primaryShardId}, **executor, getCurrentSession(_doc), false /* fromMigrate */);
 }
 
 DropDatabaseCoordinator::DropDatabaseCoordinator(ShardingDDLCoordinatorService* service,
diff --git a/src/mongo/db/s/rename_collection_participant_service.cpp b/src/mongo/db/s/rename_collection_participant_service.cpp
index c00d81e4e20..a7f27d01bb9 100644
--- a/src/mongo/db/s/rename_collection_participant_service.cpp
+++ b/src/mongo/db/s/rename_collection_participant_service.cpp
@@ -61,7 +61,7 @@ const Backoff kExponentialBackoff(Seconds(1), Milliseconds::max());
 void dropCollectionLocally(OperationContext* opCtx, const NamespaceString& nss) {
     bool knownNss = [&]() {
         try {
-            DropCollectionCoordinator::dropCollectionLocally(opCtx, nss);
+            DropCollectionCoordinator::dropCollectionLocally(opCtx, nss, false /* fromMigrate */);
             return true;
         } catch (const ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
             return false;
diff --git a/src/mongo/db/s/resharding/resharding_data_copy_util.cpp b/src/mongo/db/s/resharding/resharding_data_copy_util.cpp
index 9893b2b0f2e..6fec5cb7061 100644
--- a/src/mongo/db/s/resharding/resharding_data_copy_util.cpp
+++ b/src/mongo/db/s/resharding/resharding_data_copy_util.cpp
@@ -44,6 +44,7 @@
 #include "mongo/db/s/resharding/resharding_txn_cloner_progress_gen.h"
 #include "mongo/db/s/resharding/resharding_util.h"
 #include "mongo/db/s/session_catalog_migration.h"
+#include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/db/session_catalog_mongod.h"
 #include "mongo/db/session_txn_record_gen.h"
 #include "mongo/db/storage/write_unit_of_work.h"
@@ -71,28 +72,6 @@ void ensureCollectionExists(OperationContext* opCtx,
     });
 }
 
-void ensureCollectionDropped(OperationContext* opCtx,
-                             const NamespaceString& nss,
-                             const boost::optional<UUID>& uuid) {
-    invariant(!opCtx->lockState()->isLocked());
-    invariant(!opCtx->lockState()->inAWriteUnitOfWork());
-
-    writeConflictRetry(
-        opCtx, "resharding::data_copy::ensureCollectionDropped", nss.toString(), [&] {
-            AutoGetCollection coll(opCtx, nss, MODE_X);
-            if (!coll || (uuid && coll->uuid() != uuid)) {
-                // If the collection doesn't exist or exists with a different UUID, then the
-                // requested collection has been dropped already.
-                return;
-            }
-
-            WriteUnitOfWork wuow(opCtx);
-            uassertStatusOK(coll.getDb()->dropCollectionEvenIfSystem(
-                opCtx, nss, {} /* dropOpTime */, true /* markFromMigrate */));
-            wuow.commit();
-        });
-}
-
 void ensureOplogCollectionsDropped(OperationContext* opCtx,
                                    const UUID& reshardingUUID,
                                    const UUID& sourceUUID,
@@ -119,11 +98,11 @@ void ensureOplogCollectionsDropped(OperationContext* opCtx,
 
         // Drop the conflict stash collection for this donor.
         auto stashNss = getLocalConflictStashNamespace(sourceUUID, donor.getShardId());
-        ensureCollectionDropped(opCtx, stashNss);
+        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, stashNss);
 
         // Drop the oplog buffer collection for this donor.
         auto oplogBufferNss = getLocalOplogBufferNamespace(sourceUUID, donor.getShardId());
-        ensureCollectionDropped(opCtx, oplogBufferNss);
+        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx, oplogBufferNss);
     }
 }
 
diff --git a/src/mongo/db/s/resharding/resharding_data_copy_util.h b/src/mongo/db/s/resharding/resharding_data_copy_util.h
index 9f2a332ef6c..0858fa2d0e4 100644
--- a/src/mongo/db/s/resharding/resharding_data_copy_util.h
+++ b/src/mongo/db/s/resharding/resharding_data_copy_util.h
@@ -60,16 +60,6 @@ void ensureCollectionExists(OperationContext* opCtx,
                             const CollectionOptions& options);
 
 /**
- * Drops the specified collection or returns without error if the collection has already been
- * dropped. A particular incarnation of the collection can be dropped by specifying its UUID.
- *
- * This functions assumes the collection being dropped doesn't have any two-phase index builds
- * active on it.
- */
-void ensureCollectionDropped(OperationContext* opCtx,
-                             const NamespaceString& nss,
-                             const boost::optional<UUID>& uuid = boost::none);
-/**
  * Removes documents from the oplog applier progress and transaction applier progress collections
  * that are associated with an in-progress resharding operation. Also drops all oplog buffer
  * collections and conflict stash collections that are associated with the in-progress resharding
diff --git a/src/mongo/db/s/resharding/resharding_donor_service.cpp b/src/mongo/db/s/resharding/resharding_donor_service.cpp
index 1e65452d040..6a7962ac6ea 100644
--- a/src/mongo/db/s/resharding/resharding_donor_service.cpp
+++ b/src/mongo/db/s/resharding/resharding_donor_service.cpp
@@ -55,6 +55,7 @@
 #include "mongo/db/s/resharding/resharding_metrics.h"
 #include "mongo/db/s/resharding/resharding_server_parameters_gen.h"
 #include "mongo/db/s/resharding/resharding_util.h"
+#include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/db/write_block_bypass.h"
 #include "mongo/db/write_concern_options.h"
@@ -798,7 +799,7 @@ void ReshardingDonorService::DonorStateMachine::_dropOriginalCollectionThenTrans
         // Allow bypassing user write blocking. The check has already been performed on the
         // db-primary shard's ReshardCollectionCoordinator.
         WriteBlockBypass::get(opCtx.get()).set(true);
-        resharding::data_copy::ensureCollectionDropped(
+        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(
             opCtx.get(), _metadata.getSourceNss(), _metadata.getSourceUUID());
     }
 
diff --git a/src/mongo/db/s/resharding/resharding_donor_service_test.cpp b/src/mongo/db/s/resharding/resharding_donor_service_test.cpp
index 663b8c28136..95e34eacd58 100644
--- a/src/mongo/db/s/resharding/resharding_donor_service_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_donor_service_test.cpp
@@ -53,6 +53,7 @@
 #include "mongo/db/s/resharding/resharding_donor_service.h"
 #include "mongo/db/s/resharding/resharding_service_test_helpers.h"
 #include "mongo/db/s/resharding/resharding_util.h"
+#include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/unittest/death_test.h"
@@ -161,7 +162,8 @@ public:
     void createSourceCollection(OperationContext* opCtx, const ReshardingDonorDocument& donorDoc) {
         CollectionOptions options;
         options.uuid = donorDoc.getSourceUUID();
-        resharding::data_copy::ensureCollectionDropped(opCtx, donorDoc.getSourceNss());
+        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx,
+                                                                       donorDoc.getSourceNss());
         resharding::data_copy::ensureCollectionExists(opCtx, donorDoc.getSourceNss(), options);
     }
 
@@ -169,7 +171,8 @@ public:
                                              const ReshardingDonorDocument& donorDoc) {
         CollectionOptions options;
         options.uuid = donorDoc.getReshardingUUID();
-        resharding::data_copy::ensureCollectionDropped(opCtx, donorDoc.getTempReshardingNss());
+        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(
+            opCtx, donorDoc.getTempReshardingNss());
         resharding::data_copy::ensureCollectionExists(
             opCtx, donorDoc.getTempReshardingNss(), options);
     }
diff --git a/src/mongo/db/s/resharding/resharding_recipient_service.cpp b/src/mongo/db/s/resharding/resharding_recipient_service.cpp
index 7c07564a688..c66f19b3aaa 100644
--- a/src/mongo/db/s/resharding/resharding_recipient_service.cpp
+++ b/src/mongo/db/s/resharding/resharding_recipient_service.cpp
@@ -56,6 +56,7 @@
 #include "mongo/db/s/resharding/resharding_recipient_service_external_state.h"
 #include "mongo/db/s/resharding/resharding_server_parameters_gen.h"
 #include "mongo/db/s/shard_key_util.h"
+#include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/db/write_block_bypass.h"
 #include "mongo/executor/network_interface_factory.h"
@@ -760,7 +761,7 @@ void ReshardingRecipientService::RecipientStateMachine::_cleanupReshardingCollec
         opCtx.get(), _metadata.getReshardingUUID(), _metadata.getSourceUUID(), _donorShards);
 
     if (aborted) {
-        resharding::data_copy::ensureCollectionDropped(
+        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(
             opCtx.get(), _metadata.getTempReshardingNss(), _metadata.getReshardingUUID());
     }
 }
diff --git a/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp b/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp
index d922b0f25e5..d4682d561fe 100644
--- a/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_recipient_service_test.cpp
@@ -48,6 +48,7 @@
 #include "mongo/db/s/resharding/resharding_recipient_service.h"
 #include "mongo/db/s/resharding/resharding_recipient_service_external_state.h"
 #include "mongo/db/s/resharding/resharding_service_test_helpers.h"
+#include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/logv2/log.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/util/clock_source_mock.h"
@@ -266,7 +267,8 @@ public:
                                 const ReshardingRecipientDocument& recipientDoc) {
         CollectionOptions options;
         options.uuid = recipientDoc.getSourceUUID();
-        resharding::data_copy::ensureCollectionDropped(opCtx, recipientDoc.getSourceNss());
+        mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent(opCtx,
+                                                                       recipientDoc.getSourceNss());
         resharding::data_copy::ensureCollectionExists(opCtx, recipientDoc.getSourceNss(), options);
     }
 
diff --git a/src/mongo/db/s/sharding_ddl_util.cpp b/src/mongo/db/s/sharding_ddl_util.cpp
index 564fb64009e..af0dee01943 100644
--- a/src/mongo/db/s/sharding_ddl_util.cpp
+++ b/src/mongo/db/s/sharding_ddl_util.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/commands/feature_compatibility_version.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/repl/repl_client_info.h"
@@ -595,8 +596,11 @@ void sendDropCollectionParticipantCommandToShards(OperationContext* opCtx,
                                                   const NamespaceString& nss,
                                                   const std::vector<ShardId>& shardIds,
                                                   std::shared_ptr<executor::TaskExecutor> executor,
-                                                  const OperationSessionInfo& osi) {
-    const ShardsvrDropCollectionParticipant dropCollectionParticipant(nss);
+                                                  const OperationSessionInfo& osi,
+                                                  bool fromMigrate) {
+    ShardsvrDropCollectionParticipant dropCollectionParticipant(nss);
+    dropCollectionParticipant.setFromMigrate(fromMigrate);
+
     const auto cmdObj =
         CommandHelpers::appendMajorityWriteConcern(dropCollectionParticipant.toBSON({}));
 
@@ -618,5 +622,28 @@ BSONObj getCriticalSectionReasonForRename(const NamespaceString& from, const Nam
                 << "from" << from.toString() << "to" << to.toString());
 }
 
+void ensureCollectionDroppedNoChangeEvent(OperationContext* opCtx,
+                                          const NamespaceString& nss,
+                                          const boost::optional<UUID>& uuid) {
+    invariant(!opCtx->lockState()->isLocked());
+    invariant(!opCtx->lockState()->inAWriteUnitOfWork());
+
+    writeConflictRetry(opCtx,
+                       "mongo::sharding_ddl_util::ensureCollectionDroppedNoChangeEvent",
+                       nss.toString(),
+                       [&] {
+                           AutoGetCollection coll(opCtx, nss, MODE_X);
+                           if (!coll || (uuid && coll->uuid() != uuid)) {
+                               // If the collection doesn't exist or exists with a different UUID,
+                               // then the requested collection has been dropped already.
+                               return;
+                           }
+
+                           WriteUnitOfWork wuow(opCtx);
+                           uassertStatusOK(coll.getDb()->dropCollectionEvenIfSystem(
+                               opCtx, nss, {} /* dropOpTime */, true /* markFromMigrate */));
+                           wuow.commit();
+                       });
+}
 }  // namespace sharding_ddl_util
 }  // namespace mongo
diff --git a/src/mongo/db/s/sharding_ddl_util.h b/src/mongo/db/s/sharding_ddl_util.h
index 157a65399a1..dd02fe5035a 100644
--- a/src/mongo/db/s/sharding_ddl_util.h
+++ b/src/mongo/db/s/sharding_ddl_util.h
@@ -196,9 +196,21 @@ void sendDropCollectionParticipantCommandToShards(OperationContext* opCtx,
                                                   const NamespaceString& nss,
                                                   const std::vector<ShardId>& shardIds,
                                                   std::shared_ptr<executor::TaskExecutor> executor,
-                                                  const OperationSessionInfo& osi);
+                                                  const OperationSessionInfo& osi,
+                                                  bool fromMigrate);
 
 BSONObj getCriticalSectionReasonForRename(const NamespaceString& from, const NamespaceString& to);
 
+/**
+ * Drops the specified collection or returns without error if the collection has already been
+ * dropped. A particular incarnation of the collection can be dropped by specifying its UUID.
+ *
+ * This functions assumes the collection being dropped doesn't have any two-phase index builds
+ * active on it.
+ */
+void ensureCollectionDroppedNoChangeEvent(OperationContext* opCtx,
+                                          const NamespaceString& nss,
+                                          const boost::optional<UUID>& uuid = boost::none);
+
 }  // namespace sharding_ddl_util
 }  // namespace mongo
diff --git a/src/mongo/db/s/shardsvr_drop_collection_participant_command.cpp b/src/mongo/db/s/shardsvr_drop_collection_participant_command.cpp
index a9b18cf3f61..950f2d296d7 100644
--- a/src/mongo/db/s/shardsvr_drop_collection_participant_command.cpp
+++ b/src/mongo/db/s/shardsvr_drop_collection_participant_command.cpp
@@ -76,7 +76,10 @@ public:
             opCtx->setAlwaysInterruptAtStepDownOrUp();
 
             try {
-                DropCollectionCoordinator::dropCollectionLocally(opCtx, ns());
+                bool fromMigrate =
+                    request().getFromMigrate() ? request().getFromMigrate().value() : false;
+
+                DropCollectionCoordinator::dropCollectionLocally(opCtx, ns(), fromMigrate);
             } catch (const ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
                 LOGV2_DEBUG(5280920,
                             1,
diff --git a/src/mongo/s/request_types/sharded_ddl_commands.idl b/src/mongo/s/request_types/sharded_ddl_commands.idl
index 9700c2f52d5..2791c5a1b65 100644
--- a/src/mongo/s/request_types/sharded_ddl_commands.idl
+++ b/src/mongo/s/request_types/sharded_ddl_commands.idl
@@ -315,6 +315,11 @@ commands:
         api_version: ""
         cpp_name: ShardsvrDropCollectionParticipant
         strict: false
+        fields:
+            fromMigrate: 
+                type: bool 
+                description: "Whether the drop comes as a result of an interrupted migration process."
+                optional: true
 
     _shardsvrRenameCollection:
         command_name: _shardsvrRenameCollection