SERVER-70831 Relying on the critical section for dropCollection

19 files changed, 487 insertions, 205 deletions

diff --git a/jstests/core/txns/kill_cursors_in_transaction.js b/jstests/core/txns/kill_cursors_in_transaction.js
index a9a41ffc954..c71800f0df8 100644
--- a/jstests/core/txns/kill_cursors_in_transaction.js
+++ b/jstests/core/txns/kill_cursors_in_transaction.js
@@ -60,14 +60,20 @@ assert.soon(
                        {
                            $match: {
                                $or: [
-                                   {'command.drop': collName},
-                                   {'command._shardsvrDropCollectionParticipant': collName}
-                               ],
-                               waitingForLock: true
+                                   {
+                                       $or: [
+                                           {'command.drop': collName},
+                                           // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+                                           {'command._shardsvrDropCollectionParticipant': collName}
+                                       ],
+                                       waitingForLock: true
+                                   },
+                                   {'command._shardsvrParticipantBlock': collName},
+                               ]
                            }
                        }
                    ])
-                   .itcount() === 1;
+                   .itcount() > 0;
     },
     function() {
         return "Failed to find drop in currentOp output: " +
diff --git a/jstests/core/txns/kill_sessions_kills_transaction.js b/jstests/core/txns/kill_sessions_kills_transaction.js
index 7e858c02cc5..cfde983fba1 100644
--- a/jstests/core/txns/kill_sessions_kills_transaction.js
+++ b/jstests/core/txns/kill_sessions_kills_transaction.js
@@ -58,14 +58,20 @@ assert.soon(
                        {
                            $match: {
                                $or: [
-                                   {'command.drop': collName},
-                                   {'command._shardsvrDropCollectionParticipant': collName}
-                               ],
-                               waitingForLock: true
+                                   {
+                                       $or: [
+                                           {'command.drop': collName},
+                                           // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+                                           {'command._shardsvrDropCollectionParticipant': collName}
+                                       ],
+                                       waitingForLock: true
+                                   },
+                                   {'command._shardsvrParticipantBlock': collName},
+                               ]
                            }
                        }
                    ])
-                   .itcount() === 1;
+                   .itcount() > 0;
     },
     function() {
         return "Failed to find drop in currentOp output: " +
diff --git a/jstests/sharding/write_concern_basic.js b/jstests/sharding/write_concern_basic.js
deleted file mode 100644
index dabae78eddd..00000000000
--- a/jstests/sharding/write_concern_basic.js
+++ /dev/null
@@ -1,35 +0,0 @@
-// Test that certain basic commands preserve write concern errors.
-// @tags: [
-// ]
-//
-
-(function() {
-"use strict";
-
-load("jstests/sharding/libs/failpoint_helpers.js");
-
-let st = new ShardingTest({shards: 2});
-
-// Test drop collection
-{
-    let db = st.s.getDB("test");
-    // _shardsvrDropCollectionParticipant required for PM-1965
-    failCommandsWithWriteConcernError(st.rs0, ["drop", "_shardsvrDropCollectionParticipant"]);
-    assert.commandWorked(db.collection.insert({"a": 1}));
-    assert.commandFailedWithCode(db.runCommand({drop: "collection"}), 12345);
-    turnOffFailCommand(st.rs0);
-}
-
-// Test drop database
-{
-    let db = st.s.getDB("test");
-    // _shardsvrDropDatabaseParticipant required for PM-1965
-    failCommandsWithWriteConcernError(st.rs1, ["dropDatabase", "_shardsvrDropDatabaseParticipant"]);
-    assert.commandWorked(db.collection.insert({"a": 1}));
-    st.ensurePrimaryShard("test", st.shard0.shardName);
-    assert.commandFailedWithCode(db.runCommand({"dropDatabase": 1}), 12345);
-    turnOffFailCommand(st.rs1);
-}
-
-st.stop();
-})();
diff --git a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
index 11deb0947d3..83e1784e716 100644
--- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
+++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
@@ -324,6 +324,18 @@ public:
                     ->waitForCoordinatorsOfGivenTypeToComplete(
                         opCtx, DDLCoordinatorTypeEnum::kRenameCollectionPre63Compatible);
             }
+            // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+            if (serverGlobalParams.clusterRole == ClusterRole::ShardServer &&
+                feature_flags::gDropCollectionHoldingCriticalSection.isEnabledOnVersion(
+                    requestedVersion)) {
+                ShardingDDLCoordinatorService::getService(opCtx)
+                    ->waitForCoordinatorsOfGivenTypeToComplete(
+                        opCtx, DDLCoordinatorTypeEnum::kDropCollectionPre70Compatible);
+
+                ShardingDDLCoordinatorService::getService(opCtx)
+                    ->waitForCoordinatorsOfGivenTypeToComplete(
+                        opCtx, DDLCoordinatorTypeEnum::kDropDatabasePre70Compatible);
+            }
 
             return true;
         }
@@ -455,6 +467,19 @@ public:
                     opCtx, DDLCoordinatorTypeEnum::kRenameCollectionPre63Compatible);
         }
 
+        // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+        if (serverGlobalParams.clusterRole == ClusterRole::ShardServer &&
+            requestedVersion > actualVersion &&
+            feature_flags::gDropCollectionHoldingCriticalSection.isEnabledOnVersion(
+                requestedVersion)) {
+            ShardingDDLCoordinatorService::getService(opCtx)
+                ->waitForCoordinatorsOfGivenTypeToComplete(
+                    opCtx, DDLCoordinatorTypeEnum::kDropCollectionPre70Compatible);
+            ShardingDDLCoordinatorService::getService(opCtx)
+                ->waitForCoordinatorsOfGivenTypeToComplete(
+                    opCtx, DDLCoordinatorTypeEnum::kDropDatabasePre70Compatible);
+        }
+
         LOGV2(6744302,
               "setFeatureCompatibilityVersion succeeded",
               "upgradeOrDowngrade"_attr = upgradeOrDowngrade,
@@ -504,6 +529,18 @@ private:
                     opCtx, DDLCoordinatorTypeEnum::kRenameCollection);
         }
 
+        // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+        if (actualVersion > requestedVersion &&
+            !feature_flags::gDropCollectionHoldingCriticalSection.isEnabledOnVersion(
+                requestedVersion)) {
+            ShardingDDLCoordinatorService::getService(opCtx)
+                ->waitForCoordinatorsOfGivenTypeToComplete(opCtx,
+                                                           DDLCoordinatorTypeEnum::kDropCollection);
+            ShardingDDLCoordinatorService::getService(opCtx)
+                ->waitForCoordinatorsOfGivenTypeToComplete(opCtx,
+                                                           DDLCoordinatorTypeEnum::kDropDatabase);
+        }
+
         // TODO SERVER-68373 remove once 7.0 becomes last LTS
         if (actualVersion > requestedVersion) {
             // Drain the QE compact coordinator because it persists state that is
diff --git a/src/mongo/db/s/drop_collection_coordinator.cpp b/src/mongo/db/s/drop_collection_coordinator.cpp
index 57ca7f3dab5..8ab0f9c1404 100644
--- a/src/mongo/db/s/drop_collection_coordinator.cpp
+++ b/src/mongo/db/s/drop_collection_coordinator.cpp
@@ -35,6 +35,7 @@
 #include "mongo/db/db_raii.h"
 #include "mongo/db/s/collection_sharding_runtime.h"
 #include "mongo/db/s/global_index_ddl_util.h"
+#include "mongo/db/s/participant_block_gen.h"
 #include "mongo/db/s/range_deletion_util.h"
 #include "mongo/db/s/sharded_index_catalog_commands_gen.h"
 #include "mongo/db/s/sharding_ddl_util.h"
@@ -142,125 +143,40 @@ ExecutorFuture<void> DropCollectionCoordinator::_runImpl(
     std::shared_ptr<executor::ScopedTaskExecutor> executor,
     const CancellationToken& token) noexcept {
     return ExecutorFuture<void>(**executor)
-        .then(_buildPhaseHandler(
-            Phase::kFreezeCollection,
-            [this, anchor = shared_from_this()] {
-                auto opCtxHolder = cc().makeOperationContext();
-                auto* opCtx = opCtxHolder.get();
-                getForwardableOpMetadata().setOn(opCtx);
-
-                try {
-                    auto coll = Grid::get(opCtx)->catalogClient()->getCollection(opCtx, nss());
-                    _doc.setCollInfo(std::move(coll));
-                } catch (const ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
-                    // The collection is not sharded or doesn't exist.
-                    _doc.setCollInfo(boost::none);
-                }
-
-                {
-                    AutoGetCollection coll{
-                        opCtx,
-                        nss(),
-                        MODE_IS,
-                        AutoGetCollection::Options{}
-                            .viewMode(auto_get_collection::ViewMode::kViewsPermitted)
-                            .expectedUUID(_doc.getCollectionUUID())};
-                }
-
-                BSONObjBuilder logChangeDetail;
-                if (_doc.getCollInfo()) {
-                    logChangeDetail.append("collectionUUID",
-                                           _doc.getCollInfo()->getUuid().toBSON());
-                }
-
-                ShardingLogging::get(opCtx)->logChange(
-                    opCtx, "dropCollection.start", nss().ns(), logChangeDetail.obj());
-
-                // Persist the collection info before sticking to using it's uuid. This ensures this
-                // node is still the RS primary, so it was also the primary at the moment we read
-                // the collection metadata.
-                _updateStateDocument(opCtx, StateDoc(_doc));
-
-                if (_doc.getCollInfo()) {
-                    sharding_ddl_util::stopMigrations(opCtx, nss(), _doc.getCollInfo()->getUuid());
-                }
-            }))
-        .then(_buildPhaseHandler(
-            Phase::kDropCollection,
-            [this, executor = executor, anchor = shared_from_this()] {
-                auto opCtxHolder = cc().makeOperationContext();
-                auto* opCtx = opCtxHolder.get();
-                getForwardableOpMetadata().setOn(opCtx);
-
-                if (!_firstExecution) {
-                    // Perform a noop write on the participants in order to advance the txnNumber
-                    // for this coordinator's lsid so that requests with older txnNumbers can no
-                    // longer execute.
-                    _updateSession(opCtx);
-                    _performNoopRetryableWriteOnAllShardsAndConfigsvr(
-                        opCtx, getCurrentSession(), **executor);
-                }
-
-                const auto collIsSharded = bool(_doc.getCollInfo());
-
-                LOGV2_DEBUG(5390504,
-                            2,
-                            "Dropping collection",
-                            "namespace"_attr = nss(),
-                            "sharded"_attr = collIsSharded);
-
-                if (collIsSharded) {
-                    invariant(_doc.getCollInfo());
-                    const auto& coll = _doc.getCollInfo().value();
-                    sharding_ddl_util::removeCollAndChunksMetadataFromConfig(
-                        opCtx, coll, ShardingCatalogClient::kMajorityWriteConcern);
-                }
-
-                // Remove tags even if the collection is not sharded or didn't exist
-                _updateSession(opCtx);
-                sharding_ddl_util::removeTagsMetadataFromConfig(opCtx, nss(), getCurrentSession());
-
-                // get a Lsid and an incremented txnNumber. Ensures we are the primary
-                _updateSession(opCtx);
-
-                const auto primaryShardId = ShardingState::get(opCtx)->shardId();
-
-                // We need to send the drop to all the shards because both movePrimary and
-                // moveChunk leave garbage behind for sharded collections.
-                auto participants = Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx);
-
-                // Remove primary shard from participants
-                participants.erase(
-                    std::remove(participants.begin(), participants.end(), primaryShardId),
-                    participants.end());
-
-                sharding_ddl_util::sendDropCollectionParticipantCommandToShards(
-                    opCtx,
-                    nss(),
-                    participants,
-                    **executor,
-                    getCurrentSession(),
-                    true /*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(),
-                    false /*fromMigrate*/);
-
-                // Remove potential query analyzer document only after purging the collection from
-                // the catalog. This ensures no leftover documents referencing an old incarnation of
-                // a collection.
-                sharding_ddl_util::removeQueryAnalyzerMetadataFromConfig(opCtx, nss(), boost::none);
-
-                ShardingLogging::get(opCtx)->logChange(opCtx, "dropCollection", nss().ns());
-                LOGV2(5390503, "Collection dropped", "namespace"_attr = nss());
-            }))
+        .then([this, executor = executor, anchor = shared_from_this()] {
+            if (_isPre70Compatible())
+                return;
+
+            if (_doc.getPhase() < Phase::kFreezeCollection)
+                _checkPreconditionsAndSaveArgumentsOnDoc();
+        })
+        .then(_buildPhaseHandler(Phase::kFreezeCollection,
+                                 [this, executor = executor, anchor = shared_from_this()] {
+                                     _freezeMigrations(executor);
+                                 }))
+
+        .then([this, executor = executor, anchor = shared_from_this()] {
+            if (_isPre70Compatible())
+                return;
+
+            _buildPhaseHandler(Phase::kEnterCriticalSection,
+                               [this, executor = executor, anchor = shared_from_this()] {
+                                   _enterCriticalSection(executor);
+                               })();
+        })
+        .then(_buildPhaseHandler(Phase::kDropCollection,
+                                 [this, executor = executor, anchor = shared_from_this()] {
+                                     _commitDropCollection(executor);
+                                 }))
+        .then([this, executor = executor, anchor = shared_from_this()] {
+            if (_isPre70Compatible())
+                return;
+
+            _buildPhaseHandler(Phase::kReleaseCriticalSection,
+                               [this, executor = executor, anchor = shared_from_this()] {
+                                   _exitCriticalSection(executor);
+                               })();
+        })
         .onError([this, anchor = shared_from_this()](const Status& status) {
             if (!status.isA<ErrorCategory::NotPrimaryError>() &&
                 !status.isA<ErrorCategory::ShutdownError>()) {
@@ -273,4 +189,180 @@ ExecutorFuture<void> DropCollectionCoordinator::_runImpl(
         });
 }
 
+void DropCollectionCoordinator::_saveCollInfo(OperationContext* opCtx) {
+
+    try {
+        auto coll = Grid::get(opCtx)->catalogClient()->getCollection(opCtx, nss());
+        _doc.setCollInfo(std::move(coll));
+    } catch (const ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
+        // The collection is not sharded or doesn't exist.
+        _doc.setCollInfo(boost::none);
+    }
+}
+
+void DropCollectionCoordinator::_checkPreconditionsAndSaveArgumentsOnDoc() {
+    auto opCtxHolder = cc().makeOperationContext();
+    auto* opCtx = opCtxHolder.get();
+    getForwardableOpMetadata().setOn(opCtx);
+
+    // If the request had an expected UUID for the collection being dropped, we should verify that
+    // it matches the one from the local catalog
+    {
+        AutoGetCollection coll{opCtx,
+                               nss(),
+                               MODE_IS,
+                               AutoGetCollection::Options{}
+                                   .viewMode(auto_get_collection::ViewMode::kViewsPermitted)
+                                   .expectedUUID(_doc.getCollectionUUID())};
+    }
+
+    _saveCollInfo(opCtx);
+}
+
+void DropCollectionCoordinator::_freezeMigrations(
+    std::shared_ptr<executor::ScopedTaskExecutor> executor) {
+    auto opCtxHolder = cc().makeOperationContext();
+    auto* opCtx = opCtxHolder.get();
+    getForwardableOpMetadata().setOn(opCtx);
+
+    if (_isPre70Compatible()) {
+        _saveCollInfo(opCtx);
+
+        // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+        // This check can be removed since it is also present in _checkPreconditions.
+        {
+            AutoGetCollection coll{opCtx,
+                                   nss(),
+                                   MODE_IS,
+                                   AutoGetCollection::Options{}
+                                       .viewMode(auto_get_collection::ViewMode::kViewsPermitted)
+                                       .expectedUUID(_doc.getCollectionUUID())};
+        }
+
+        // Persist the collection info before sticking to using it's uuid. This ensures this node is
+        // still the RS primary, so it was also the primary at the moment we read the collection
+        // metadata.
+        _updateStateDocument(opCtx, StateDoc(_doc));
+    }
+
+    BSONObjBuilder logChangeDetail;
+    if (_doc.getCollInfo()) {
+        logChangeDetail.append("collectionUUID", _doc.getCollInfo()->getUuid().toBSON());
+    }
+
+    ShardingLogging::get(opCtx)->logChange(
+        opCtx, "dropCollection.start", nss().ns(), logChangeDetail.obj());
+
+    if (_doc.getCollInfo()) {
+        sharding_ddl_util::stopMigrations(opCtx, nss(), _doc.getCollInfo()->getUuid());
+    }
+}
+
+void DropCollectionCoordinator::_enterCriticalSection(
+    std::shared_ptr<executor::ScopedTaskExecutor> executor) {
+    LOGV2_DEBUG(7038100, 2, "Acquiring critical section", "namespace"_attr = nss());
+
+    auto opCtxHolder = cc().makeOperationContext();
+    auto* opCtx = opCtxHolder.get();
+    getForwardableOpMetadata().setOn(opCtx);
+
+    _updateSession(opCtx);
+    ShardsvrParticipantBlock blockCRUDOperationsRequest(nss());
+    blockCRUDOperationsRequest.setBlockType(mongo::CriticalSectionBlockTypeEnum::kReadsAndWrites);
+    blockCRUDOperationsRequest.setReason(_critSecReason);
+    blockCRUDOperationsRequest.setAllowViews(true);
+
+    const auto cmdObj =
+        CommandHelpers::appendMajorityWriteConcern(blockCRUDOperationsRequest.toBSON({}));
+    sharding_ddl_util::sendAuthenticatedCommandToShards(
+        opCtx,
+        nss().db(),
+        cmdObj.addFields(getCurrentSession().toBSON()),
+        Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx),
+        **executor);
+
+    LOGV2_DEBUG(7038101, 2, "Acquired critical section", "namespace"_attr = nss());
+}
+
+void DropCollectionCoordinator::_commitDropCollection(
+    std::shared_ptr<executor::ScopedTaskExecutor> executor) {
+    auto opCtxHolder = cc().makeOperationContext();
+    auto* opCtx = opCtxHolder.get();
+    getForwardableOpMetadata().setOn(opCtx);
+
+    const auto collIsSharded = bool(_doc.getCollInfo());
+
+    LOGV2_DEBUG(5390504,
+                2,
+                "Dropping collection",
+                "namespace"_attr = nss(),
+                "sharded"_attr = collIsSharded);
+
+    if (collIsSharded) {
+        invariant(_doc.getCollInfo());
+        const auto& coll = _doc.getCollInfo().value();
+        sharding_ddl_util::removeCollAndChunksMetadataFromConfig(
+            opCtx, coll, ShardingCatalogClient::kMajorityWriteConcern);
+    }
+
+    // Remove tags even if the collection is not sharded or didn't exist
+    _updateSession(opCtx);
+    sharding_ddl_util::removeTagsMetadataFromConfig(opCtx, nss(), getCurrentSession());
+
+    // get a Lsid and an incremented txnNumber. Ensures we are the primary
+    _updateSession(opCtx);
+
+    const auto primaryShardId = ShardingState::get(opCtx)->shardId();
+
+    // We need to send the drop to all the shards because both movePrimary and
+    // moveChunk leave garbage behind for sharded collections.
+    auto participants = Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx);
+    // Remove primary shard from participants
+    participants.erase(std::remove(participants.begin(), participants.end(), primaryShardId),
+                       participants.end());
+
+    sharding_ddl_util::sendDropCollectionParticipantCommandToShards(
+        opCtx, nss(), participants, **executor, getCurrentSession(), true /*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(), false /*fromMigrate*/);
+
+    // Remove potential query analyzer document only after purging the collection from
+    // the catalog. This ensures no leftover documents referencing an old incarnation of
+    // a collection.
+    sharding_ddl_util::removeQueryAnalyzerMetadataFromConfig(opCtx, nss(), boost::none);
+
+    ShardingLogging::get(opCtx)->logChange(opCtx, "dropCollection", nss().ns());
+    LOGV2(5390503, "Collection dropped", "namespace"_attr = nss());
+}
+
+void DropCollectionCoordinator::_exitCriticalSection(
+    std::shared_ptr<executor::ScopedTaskExecutor> executor) {
+    LOGV2_DEBUG(7038102, 2, "Releasing critical section", "namespace"_attr = nss());
+
+    auto opCtxHolder = cc().makeOperationContext();
+    auto* opCtx = opCtxHolder.get();
+    getForwardableOpMetadata().setOn(opCtx);
+
+    _updateSession(opCtx);
+    ShardsvrParticipantBlock unblockCRUDOperationsRequest(nss());
+    unblockCRUDOperationsRequest.setBlockType(CriticalSectionBlockTypeEnum::kUnblock);
+    unblockCRUDOperationsRequest.setReason(_critSecReason);
+    unblockCRUDOperationsRequest.setAllowViews(true);
+
+    const auto cmdObj =
+        CommandHelpers::appendMajorityWriteConcern(unblockCRUDOperationsRequest.toBSON({}));
+    sharding_ddl_util::sendAuthenticatedCommandToShards(
+        opCtx,
+        nss().db(),
+        cmdObj.addFields(getCurrentSession().toBSON()),
+        Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx),
+        **executor);
+
+    LOGV2_DEBUG(7038103, 2, "Released critical section", "namespace"_attr = nss());
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/s/drop_collection_coordinator.h b/src/mongo/db/s/drop_collection_coordinator.h
index 7a8288c4e9c..cbde60fdbed 100644
--- a/src/mongo/db/s/drop_collection_coordinator.h
+++ b/src/mongo/db/s/drop_collection_coordinator.h
@@ -43,7 +43,10 @@ public:
     using Phase = DropCollectionCoordinatorPhaseEnum;
 
     DropCollectionCoordinator(ShardingDDLCoordinatorService* service, const BSONObj& initialState)
-        : RecoverableShardingDDLCoordinator(service, "DropCollectionCoordinator", initialState) {}
+        : RecoverableShardingDDLCoordinator(service, "DropCollectionCoordinator", initialState),
+          _critSecReason(BSON("command"
+                              << "dropCollection"
+                              << "ns" << originalNss().toString())) {}
 
     ~DropCollectionCoordinator() = default;
 
@@ -60,12 +63,37 @@ public:
                                       bool fromMigrate);
 
 private:
+    const BSONObj _critSecReason;
+
     StringData serializePhase(const Phase& phase) const override {
         return DropCollectionCoordinatorPhase_serializer(phase);
     }
 
+    bool _mustAlwaysMakeProgress() override {
+        return !_isPre70Compatible() && _doc.getPhase() > Phase::kUnset;
+    }
+
+    // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+    bool _isPre70Compatible() const {
+        return operationType() == DDLCoordinatorTypeEnum::kDropCollectionPre70Compatible;
+    }
+
     ExecutorFuture<void> _runImpl(std::shared_ptr<executor::ScopedTaskExecutor> executor,
                                   const CancellationToken& token) noexcept override;
+
+    // TODO SERVER-73627: inline this function once 7.0 becomes last LTS since it should be only
+    // called once.
+    void _saveCollInfo(OperationContext* opCtx);
+
+    void _checkPreconditionsAndSaveArgumentsOnDoc();
+
+    void _freezeMigrations(std::shared_ptr<executor::ScopedTaskExecutor> executor);
+
+    void _enterCriticalSection(std::shared_ptr<executor::ScopedTaskExecutor> executor);
+
+    void _commitDropCollection(std::shared_ptr<executor::ScopedTaskExecutor> executor);
+
+    void _exitCriticalSection(std::shared_ptr<executor::ScopedTaskExecutor> executor);
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/drop_collection_coordinator_document.idl b/src/mongo/db/s/drop_collection_coordinator_document.idl
index 4c71ec33d5c..1cf65cc5c82 100644
--- a/src/mongo/db/s/drop_collection_coordinator_document.idl
+++ b/src/mongo/db/s/drop_collection_coordinator_document.idl
@@ -45,7 +45,9 @@ enums:
         values:
             kUnset: "unset"
             kFreezeCollection: "freezeCollection"
+            kEnterCriticalSection: "enterCriticalSection"
             kDropCollection: "dropCollection"
+            kReleaseCriticalSection: "releaseCriticalSection"
 
 types:
     CollectionInfo:
diff --git a/src/mongo/db/s/drop_database_coordinator.cpp b/src/mongo/db/s/drop_database_coordinator.cpp
index 5f9c6ba3cb9..228490b0ebb 100644
--- a/src/mongo/db/s/drop_database_coordinator.cpp
+++ b/src/mongo/db/s/drop_database_coordinator.cpp
@@ -36,6 +36,7 @@
 #include "mongo/db/cluster_transaction_api.h"
 #include "mongo/db/persistent_task_store.h"
 #include "mongo/db/s/database_sharding_state.h"
+#include "mongo/db/s/participant_block_gen.h"
 #include "mongo/db/s/shard_metadata_util.h"
 #include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/db/s/sharding_logging.h"
@@ -202,6 +203,12 @@ bool isDbAlreadyDropped(OperationContext* opCtx,
     return false;
 }
 
+BSONObj getReasonForDropCollection(const NamespaceString& nss) {
+    return BSON("command"
+                << "dropCollection fromDropDatabase"
+                << "nss" << nss.ns());
+}
+
 }  // namespace
 
 void DropDatabaseCoordinator::_dropShardedCollection(
@@ -216,6 +223,23 @@ void DropDatabaseCoordinator::_dropShardedCollection(
     auto collDDLLock = DDLLockManager::get(opCtx)->lock(
         opCtx, nss.ns(), coorName, DDLLockManager::kDefaultLockTimeout);
 
+    if (!_isPre70Compatible()) {
+        _updateSession(opCtx);
+        ShardsvrParticipantBlock blockCRUDOperationsRequest(nss);
+        blockCRUDOperationsRequest.setBlockType(
+            mongo::CriticalSectionBlockTypeEnum::kReadsAndWrites);
+        blockCRUDOperationsRequest.setReason(getReasonForDropCollection(nss));
+        blockCRUDOperationsRequest.setAllowViews(true);
+        const auto cmdObj =
+            CommandHelpers::appendMajorityWriteConcern(blockCRUDOperationsRequest.toBSON({}));
+        sharding_ddl_util::sendAuthenticatedCommandToShards(
+            opCtx,
+            nss.db(),
+            cmdObj.addFields(getCurrentSession().toBSON()),
+            Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx),
+            **executor);
+    }
+
     sharding_ddl_util::removeCollAndChunksMetadataFromConfig(
         opCtx, coll, ShardingCatalogClient::kMajorityWriteConcern);
 
@@ -242,6 +266,23 @@ void DropDatabaseCoordinator::_dropShardedCollection(
 
     // Remove collection's query analyzer configuration document, if it exists.
     sharding_ddl_util::removeQueryAnalyzerMetadataFromConfig(opCtx, nss, coll.getUuid());
+
+    if (!_isPre70Compatible()) {
+        _updateSession(opCtx);
+        ShardsvrParticipantBlock unblockCRUDOperationsRequest(nss);
+        unblockCRUDOperationsRequest.setBlockType(CriticalSectionBlockTypeEnum::kUnblock);
+        unblockCRUDOperationsRequest.setReason(getReasonForDropCollection(nss));
+        unblockCRUDOperationsRequest.setAllowViews(true);
+
+        const auto cmdObj =
+            CommandHelpers::appendMajorityWriteConcern(unblockCRUDOperationsRequest.toBSON({}));
+        sharding_ddl_util::sendAuthenticatedCommandToShards(
+            opCtx,
+            nss.db(),
+            cmdObj.addFields(getCurrentSession().toBSON()),
+            Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx),
+            **executor);
+    }
 }
 
 void DropDatabaseCoordinator::_clearDatabaseInfoOnPrimary(OperationContext* opCtx) {
diff --git a/src/mongo/db/s/drop_database_coordinator.h b/src/mongo/db/s/drop_database_coordinator.h
index 1d5cf77e028..447031516f2 100644
--- a/src/mongo/db/s/drop_database_coordinator.h
+++ b/src/mongo/db/s/drop_database_coordinator.h
@@ -54,6 +54,15 @@ private:
         return DropDatabaseCoordinatorPhase_serializer(phase);
     }
 
+    bool _mustAlwaysMakeProgress() override {
+        return !_isPre70Compatible() && _doc.getPhase() > Phase::kUnset;
+    }
+
+    // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+    bool _isPre70Compatible() const {
+        return operationType() == DDLCoordinatorTypeEnum::kDropDatabasePre70Compatible;
+    }
+
     ExecutorFuture<void> _runImpl(std::shared_ptr<executor::ScopedTaskExecutor> executor,
                                   const CancellationToken& token) noexcept override;
 
diff --git a/src/mongo/db/s/participant_block.idl b/src/mongo/db/s/participant_block.idl
index a2473c8700e..0ccf831d177 100644
--- a/src/mongo/db/s/participant_block.idl
+++ b/src/mongo/db/s/participant_block.idl
@@ -55,3 +55,7 @@ commands:
             reason:
                 type: object
                 optional: true
+            allowViews:
+                # WARNING: This flag can be used only by coordinators running exclusively in FCV >= 7.0
+                # TODO SERVER-68084: remove this flag
+                type: optionalBool
\ No newline at end of file
diff --git a/src/mongo/db/s/shard_server_op_observer.cpp b/src/mongo/db/s/shard_server_op_observer.cpp
index 396a6c1f3fa..d5f66667365 100644
--- a/src/mongo/db/s/shard_server_op_observer.cpp
+++ b/src/mongo/db/s/shard_server_op_observer.cpp
@@ -308,7 +308,7 @@ void ShardServerOpObserver::onInserts(OperationContext* opCtx,
                             lockCollectionIfNotPrimary.emplace(
                                 opCtx,
                                 insertedNss,
-                                MODE_IX,
+                                fixLockModeForSystemDotViewsChanges(insertedNss, MODE_IX),
                                 AutoGetCollection::Options{}.viewMode(
                                     auto_get_collection::ViewMode::kViewsPermitted));
                         }
@@ -484,7 +484,7 @@ void ShardServerOpObserver::onUpdate(OperationContext* opCtx, const OplogUpdateE
                         lockCollectionIfNotPrimary.emplace(
                             opCtx,
                             updatedNss,
-                            MODE_IX,
+                            fixLockModeForSystemDotViewsChanges(updatedNss, MODE_IX),
                             AutoGetCollection::Options{}.viewMode(
                                 auto_get_collection::ViewMode::kViewsPermitted));
                     }
@@ -719,7 +719,7 @@ void ShardServerOpObserver::onDelete(OperationContext* opCtx,
                         lockCollectionIfNotPrimary.emplace(
                             opCtx,
                             deletedNss,
-                            MODE_IX,
+                            fixLockModeForSystemDotViewsChanges(deletedNss, MODE_IX),
                             AutoGetCollection::Options{}.viewMode(
                                 auto_get_collection::ViewMode::kViewsPermitted));
                     }
diff --git a/src/mongo/db/s/sharding_ddl_coordinator.idl b/src/mongo/db/s/sharding_ddl_coordinator.idl
index 7f9b7c72a14..26d5702ec1e 100644
--- a/src/mongo/db/s/sharding_ddl_coordinator.idl
+++ b/src/mongo/db/s/sharding_ddl_coordinator.idl
@@ -46,8 +46,12 @@ enums:
             # TODO (SERVER-71309): Remove once 7.0 becomes last LTS.
             kMovePrimaryNoResilient: "movePrimaryNoResilient"
             kMovePrimary: "movePrimary"
-            kDropDatabase: "dropDatabase"
-            kDropCollection: "dropCollection"
+            kDropDatabase: "dropDatabase_V2"
+            # TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+            kDropDatabasePre70Compatible: "dropDatabase"
+            kDropCollection: "dropCollection_V2"
+            # TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+            kDropCollectionPre70Compatible: "dropCollection"
             kRenameCollection: "renameCollection_V2"
             # TODO SERVER-72796: Remove once gGlobalIndexesShardingCatalog is enabled.
             kRenameCollectionPre63Compatible: "renameCollection"
diff --git a/src/mongo/db/s/sharding_ddl_coordinator_service.cpp b/src/mongo/db/s/sharding_ddl_coordinator_service.cpp
index 77799dbd322..78afdfe4849 100644
--- a/src/mongo/db/s/sharding_ddl_coordinator_service.cpp
+++ b/src/mongo/db/s/sharding_ddl_coordinator_service.cpp
@@ -69,10 +69,14 @@ std::shared_ptr<ShardingDDLCoordinator> constructShardingDDLCoordinatorInstance(
         case DDLCoordinatorTypeEnum::kMovePrimary:
             return std::make_shared<MovePrimaryCoordinator>(service, std::move(initialState));
             break;
+        // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
         case DDLCoordinatorTypeEnum::kDropDatabase:
+        case DDLCoordinatorTypeEnum::kDropDatabasePre70Compatible:
             return std::make_shared<DropDatabaseCoordinator>(service, std::move(initialState));
             break;
+        // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
         case DDLCoordinatorTypeEnum::kDropCollection:
+        case DDLCoordinatorTypeEnum::kDropCollectionPre70Compatible:
             return std::make_shared<DropCollectionCoordinator>(service, std::move(initialState));
             break;
         case DDLCoordinatorTypeEnum::kRenameCollection:
diff --git a/src/mongo/db/s/sharding_recovery_service.cpp b/src/mongo/db/s/sharding_recovery_service.cpp
index 0805904d423..458f11e76af 100644
--- a/src/mongo/db/s/sharding_recovery_service.cpp
+++ b/src/mongo/db/s/sharding_recovery_service.cpp
@@ -44,6 +44,7 @@
 #include "mongo/db/s/database_sharding_state.h"
 #include "mongo/db/s/shard_authoritative_catalog_gen.h"
 #include "mongo/db/s/sharding_migration_critical_section.h"
+#include "mongo/db/server_options.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/catalog/type_collection.h"
 #include "mongo/s/write_ops/batched_command_request.h"
@@ -143,7 +144,8 @@ void ShardingRecoveryService::acquireRecoverableCriticalSectionBlockWrites(
     OperationContext* opCtx,
     const NamespaceString& nss,
     const BSONObj& reason,
-    const WriteConcernOptions& writeConcern) {
+    const WriteConcernOptions& writeConcern,
+    bool allowViews) {
     LOGV2_DEBUG(5656600,
                 3,
                 "Acquiring recoverable critical section blocking writes",
@@ -167,7 +169,12 @@ void ShardingRecoveryService::acquireRecoverableCriticalSectionBlockWrites(
         } else {
             // TODO SERVER-68084 add the AutoGetCollectionViewMode::kViewsPermitted parameter to
             // construct collLock.
-            collLock.emplace(opCtx, nss, MODE_S);
+            collLock.emplace(opCtx,
+                             nss,
+                             MODE_S,
+                             (allowViews ? AutoGetCollection::Options{}.viewMode(
+                                               auto_get_collection::ViewMode::kViewsPermitted)
+                                         : AutoGetCollection::Options{}));
         }
 
         DBDirectClient dbClient(opCtx);
@@ -250,7 +257,8 @@ void ShardingRecoveryService::promoteRecoverableCriticalSectionToBlockAlsoReads(
     OperationContext* opCtx,
     const NamespaceString& nss,
     const BSONObj& reason,
-    const WriteConcernOptions& writeConcern) {
+    const WriteConcernOptions& writeConcern,
+    bool allowViews) {
     LOGV2_DEBUG(5656603,
                 3,
                 "Promoting recoverable critical section to also block reads",
@@ -273,7 +281,12 @@ void ShardingRecoveryService::promoteRecoverableCriticalSectionToBlockAlsoReads(
         } else {
             // TODO SERVER-68084 add the AutoGetCollectionViewMode::kViewsPermitted parameter to
             // construct collLock.
-            collLock.emplace(opCtx, nss, MODE_X);
+            collLock.emplace(opCtx,
+                             nss,
+                             MODE_X,
+                             (allowViews ? AutoGetCollection::Options{}.viewMode(
+                                               auto_get_collection::ViewMode::kViewsPermitted)
+                                         : AutoGetCollection::Options{}));
         }
 
         DBDirectClient dbClient(opCtx);
@@ -372,6 +385,7 @@ void ShardingRecoveryService::releaseRecoverableCriticalSection(
     const NamespaceString& nss,
     const BSONObj& reason,
     const WriteConcernOptions& writeConcern,
+    bool allowViews,
     bool throwIfReasonDiffers) {
     LOGV2_DEBUG(5656606,
                 3,
@@ -395,7 +409,12 @@ void ShardingRecoveryService::releaseRecoverableCriticalSection(
         } else {
             // TODO SERVER-68084 add the AutoGetCollectionViewMode::kViewsPermitted parameter to
             // construct collLock.
-            collLock.emplace(opCtx, nss, MODE_X);
+            collLock.emplace(opCtx,
+                             nss,
+                             MODE_X,
+                             (allowViews ? AutoGetCollection::Options{}.viewMode(
+                                               auto_get_collection::ViewMode::kViewsPermitted)
+                                         : AutoGetCollection::Options{}));
         }
 
         DBDirectClient dbClient(opCtx);
@@ -496,18 +515,14 @@ void ShardingRecoveryService::recoverRecoverableCriticalSections(OperationContex
 
     // Release all in-memory critical sections
     for (const auto& nss : CollectionShardingState::getCollectionNames(opCtx)) {
-        try {
-            AutoGetCollection collLock(opCtx, nss, MODE_X);
-            auto scopedCsr =
-                CollectionShardingRuntime::assertCollectionLockedAndAcquireExclusive(opCtx, nss);
-            scopedCsr->exitCriticalSectionNoChecks();
-        } catch (const ExceptionFor<ErrorCodes::CommandNotSupportedOnView>&) {
-            LOGV2_DEBUG(6050800,
-                        2,
-                        "Skipping attempting to exit critical section for view in "
-                        "recoverRecoverableCriticalSections",
-                        "namespace"_attr = nss);
-        }
+        AutoGetCollection collLock(
+            opCtx,
+            nss,
+            MODE_X,
+            AutoGetCollection::Options{}.viewMode(auto_get_collection::ViewMode::kViewsPermitted));
+        auto scopedCsr =
+            CollectionShardingRuntime::assertCollectionLockedAndAcquireExclusive(opCtx, nss);
+        scopedCsr->exitCriticalSectionNoChecks();
     }
     for (const auto& dbName : DatabaseShardingState::getDatabaseNames(opCtx)) {
         AutoGetDb dbLock(opCtx, dbName, MODE_X);
@@ -530,7 +545,11 @@ void ShardingRecoveryService::recoverRecoverableCriticalSections(OperationContex
                     scopedDss->enterCriticalSectionCommitPhase(opCtx, doc.getReason());
                 }
             } else {
-                AutoGetCollection collLock(opCtx, nss, MODE_X);
+                AutoGetCollection collLock(opCtx,
+                                           nss,
+                                           MODE_X,
+                                           AutoGetCollection::Options{}.viewMode(
+                                               auto_get_collection::ViewMode::kViewsPermitted));
                 auto scopedCsr =
                     CollectionShardingRuntime::assertCollectionLockedAndAcquireExclusive(opCtx,
                                                                                          nss);
diff --git a/src/mongo/db/s/sharding_recovery_service.h b/src/mongo/db/s/sharding_recovery_service.h
index 4ffdc846035..930662df0d5 100644
--- a/src/mongo/db/s/sharding_recovery_service.h
+++ b/src/mongo/db/s/sharding_recovery_service.h
@@ -63,11 +63,18 @@ public:
      * Do nothing if the critical section is taken for that namespace and reason, and will invariant
      * otherwise since it is the responsibility of the caller to ensure that only one thread is
      * taking the critical section.
+     *
+     * allowViews states whether views should be supported. By default it is disabled since binaries
+     * older than 7.0 don't properly support views. It is responsability of the caller to guarantee
+     * the correct usage of this flag.
      */
-    void acquireRecoverableCriticalSectionBlockWrites(OperationContext* opCtx,
-                                                      const NamespaceString& nss,
-                                                      const BSONObj& reason,
-                                                      const WriteConcernOptions& writeConcern);
+    void acquireRecoverableCriticalSectionBlockWrites(
+        OperationContext* opCtx,
+        const NamespaceString& nss,
+        const BSONObj& reason,
+        const WriteConcernOptions& writeConcern,
+        // TODO SERVER-68084 remove allowViews parameter
+        bool allowViews = false);
 
     /**
      * Advances the recoverable critical section from the catch-up phase (i.e. blocking writes) to
@@ -78,11 +85,19 @@ public:
      * It updates a doc from `config.collectionCriticalSections` with `writeConcern` write concern.
      *
      * Do nothing if the critical section is already taken in commit phase.
+     *
+     * allowViews states whether views should be supported. By default it is disabled since binaries
+     * older than 7.0 don't properly support views. It is responsability of the caller to guarantee
+     * the correct usage of this flag.
      */
-    void promoteRecoverableCriticalSectionToBlockAlsoReads(OperationContext* opCtx,
-                                                           const NamespaceString& nss,
-                                                           const BSONObj& reason,
-                                                           const WriteConcernOptions& writeConcern);
+    void promoteRecoverableCriticalSectionToBlockAlsoReads(
+        OperationContext* opCtx,
+        const NamespaceString& nss,
+        const BSONObj& reason,
+        const WriteConcernOptions& writeConcern,
+        // TODO SERVER-68084 remove allowViews parameter
+        bool allowViews = false);
+
     /**
      * Releases the recoverable critical section for the given namespace and reason.
      *
@@ -92,13 +107,20 @@ public:
      *
      * Do nothing if the critical section is not taken for that namespace and reason.
      *
+     * allowViews states whether views should be supported. By default it is disabled since binaries
+     * older than 7.0 don't properly support views. It is responsability of the caller to guarantee
+     * the correct usage of this flag.
+     *
      * Throw an invariant in case the collection critical section is already taken by another
      * operation with a different reason unless the flag 'throwIfReasonDiffers' is set to false.
+     *
      */
     void releaseRecoverableCriticalSection(OperationContext* opCtx,
                                            const NamespaceString& nss,
                                            const BSONObj& reason,
                                            const WriteConcernOptions& writeConcern,
+                                           // TODO SERVER-68084 remove allowViews parameter
+                                           bool allowViews = false,
                                            bool throwIfReasonDiffers = true);
 
     /**
diff --git a/src/mongo/db/s/shardsvr_drop_collection_command.cpp b/src/mongo/db/s/shardsvr_drop_collection_command.cpp
index 503d16b1802..8ac48003f7b 100644
--- a/src/mongo/db/s/shardsvr_drop_collection_command.cpp
+++ b/src/mongo/db/s/shardsvr_drop_collection_command.cpp
@@ -33,6 +33,7 @@
 #include "mongo/db/commands/feature_compatibility_version.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/s/drop_collection_coordinator.h"
+#include "mongo/db/s/sharding_ddl_coordinator_gen.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
@@ -104,9 +105,14 @@ public:
                     }
                     return ns();
                 }();
+                // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+                const DDLCoordinatorTypeEnum coordType =
+                    feature_flags::gDropCollectionHoldingCriticalSection.isEnabled(*fixedFcvRegion)
+                    ? DDLCoordinatorTypeEnum::kDropCollection
+                    : DDLCoordinatorTypeEnum::kDropCollectionPre70Compatible;
+
                 auto coordinatorDoc = DropCollectionCoordinatorDocument();
-                coordinatorDoc.setShardingDDLCoordinatorMetadata(
-                    {{targetNss, DDLCoordinatorTypeEnum::kDropCollection}});
+                coordinatorDoc.setShardingDDLCoordinatorMetadata({{targetNss, coordType}});
                 coordinatorDoc.setCollectionUUID(request().getCollectionUUID());
 
                 auto service = ShardingDDLCoordinatorService::getService(opCtx);
diff --git a/src/mongo/db/s/shardsvr_drop_database_command.cpp b/src/mongo/db/s/shardsvr_drop_database_command.cpp
index 847a25783c4..3f23ffa2fa8 100644
--- a/src/mongo/db/s/shardsvr_drop_database_command.cpp
+++ b/src/mongo/db/s/shardsvr_drop_database_command.cpp
@@ -31,6 +31,7 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/commands.h"
+#include "mongo/db/commands/feature_compatibility_version.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/s/drop_database_coordinator.h"
 #include "mongo/db/s/operation_sharding_state.h"
@@ -39,6 +40,7 @@
 #include "mongo/logv2/log.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/request_types/sharded_ddl_commands_gen.h"
+#include "mongo/s/sharding_feature_flags_gen.h"
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
 
@@ -81,19 +83,31 @@ public:
             CurOp::get(opCtx)->raiseDbProfileLevel(
                 CollectionCatalog::get(opCtx)->getDatabaseProfileLevel(ns().dbName()));
 
-            auto coordinatorDoc = DropDatabaseCoordinatorDocument();
-            coordinatorDoc.setShardingDDLCoordinatorMetadata(
-                {{ns(), DDLCoordinatorTypeEnum::kDropDatabase}});
             auto service = ShardingDDLCoordinatorService::getService(opCtx);
             const auto requestVersion = OperationShardingState::get(opCtx).getDbVersion(ns().db());
             auto dropDatabaseCoordinator = [&]() {
                 while (true) {
+                    // TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+                    boost::optional<FixedFCVRegion> fixedFcvRegion;
+                    fixedFcvRegion.emplace(opCtx);
+
+                    DropDatabaseCoordinatorDocument coordinatorDoc;
+                    const DDLCoordinatorTypeEnum coordType =
+                        feature_flags::gDropCollectionHoldingCriticalSection.isEnabled(
+                            **fixedFcvRegion)
+                        ? DDLCoordinatorTypeEnum::kDropDatabase
+                        : DDLCoordinatorTypeEnum::kDropDatabasePre70Compatible;
+
+                    coordinatorDoc.setShardingDDLCoordinatorMetadata({{ns(), coordType}});
+
                     auto currentCoordinator = checked_pointer_cast<DropDatabaseCoordinator>(
                         service->getOrCreateInstance(opCtx, coordinatorDoc.toBSON()));
                     const auto currentDbVersion = currentCoordinator->getDatabaseVersion();
                     if (currentDbVersion == requestVersion) {
                         return currentCoordinator;
                     }
+
+                    fixedFcvRegion.reset();
                     LOGV2_DEBUG(6073000,
                                 2,
                                 "DbVersion mismatch, waiting for existing coordinator to finish",
diff --git a/src/mongo/db/s/shardsvr_participant_block_command.cpp b/src/mongo/db/s/shardsvr_participant_block_command.cpp
index efb223fc22c..7f1176176a1 100644
--- a/src/mongo/db/s/shardsvr_participant_block_command.cpp
+++ b/src/mongo/db/s/shardsvr_participant_block_command.cpp
@@ -87,21 +87,39 @@ public:
                                                             << "ns" << ns().toString()));
                 auto blockType = request().getBlockType().get_value_or(
                     CriticalSectionBlockTypeEnum::kReadsAndWrites);
+
+                bool allowViews = request().getAllowViews();
                 auto service = ShardingRecoveryService::get(opCtx);
                 switch (blockType) {
                     case CriticalSectionBlockTypeEnum::kUnblock:
                         service->releaseRecoverableCriticalSection(
-                            opCtx, ns(), reason, ShardingCatalogClient::kLocalWriteConcern);
+                            opCtx,
+                            ns(),
+                            reason,
+                            ShardingCatalogClient::kLocalWriteConcern,
+                            allowViews);
                         break;
                     case CriticalSectionBlockTypeEnum::kWrites:
                         service->acquireRecoverableCriticalSectionBlockWrites(
-                            opCtx, ns(), reason, ShardingCatalogClient::kLocalWriteConcern);
+                            opCtx,
+                            ns(),
+                            reason,
+                            ShardingCatalogClient::kLocalWriteConcern,
+                            allowViews);
                         break;
                     default:
                         service->acquireRecoverableCriticalSectionBlockWrites(
-                            opCtx, ns(), reason, ShardingCatalogClient::kLocalWriteConcern);
+                            opCtx,
+                            ns(),
+                            reason,
+                            ShardingCatalogClient::kLocalWriteConcern,
+                            allowViews);
                         service->promoteRecoverableCriticalSectionToBlockAlsoReads(
-                            opCtx, ns(), reason, ShardingCatalogClient::kLocalWriteConcern);
+                            opCtx,
+                            ns(),
+                            reason,
+                            ShardingCatalogClient::kLocalWriteConcern,
+                            allowViews);
                 };
             };
 
diff --git a/src/mongo/s/sharding_feature_flags.idl b/src/mongo/s/sharding_feature_flags.idl
index f01b6816291..41586885c02 100644
--- a/src/mongo/s/sharding_feature_flags.idl
+++ b/src/mongo/s/sharding_feature_flags.idl
@@ -98,3 +98,8 @@ feature_flags:
     description: "Feature flag for checking metadata consistency in the cluster, database or collection"
     cpp_varname: feature_flags::gCheckMetadataConsistency
     default: false
+  # TODO SERVER-73627: Remove once 7.0 becomes last LTS.
+  featureFlagDropCollectionHoldingCriticalSection:
+    description: "Feature flag for enabling the new implementation of the dropCollection DDL operation."
+    cpp_varname: feature_flags::gDropCollectionHoldingCriticalSection
+    default: false