SERVER-52811 Implement the new drop database path in _shardsvrDropDatabase

20 files changed, 417 insertions, 276 deletions

diff --git a/buildscripts/resmokeconfig/suites/sharding_continuous_config_stepdown.yml b/buildscripts/resmokeconfig/suites/sharding_continuous_config_stepdown.yml
index 9962b3e597c..c9ce5220476 100644
--- a/buildscripts/resmokeconfig/suites/sharding_continuous_config_stepdown.yml
+++ b/buildscripts/resmokeconfig/suites/sharding_continuous_config_stepdown.yml
@@ -76,7 +76,6 @@ selector:
   - jstests/sharding/zone_changes_hashed.js
   - jstests/sharding/zone_changes_range.js
   # No retries on direct writes to the config/admin databases on the config servers
-  - jstests/sharding/database_versioning_safe_secondary_reads.js
   - jstests/sharding/listDatabases.js
   - jstests/sharding/bulk_insert.js
   - jstests/sharding/printShardingStatus.js
diff --git a/buildscripts/resmokeconfig/suites/sharding_misc.yml b/buildscripts/resmokeconfig/suites/sharding_misc.yml
index 8bc4a25947c..9a6c46a88e7 100644
--- a/buildscripts/resmokeconfig/suites/sharding_misc.yml
+++ b/buildscripts/resmokeconfig/suites/sharding_misc.yml
@@ -163,7 +163,6 @@ selector:
   - jstests/sharding/coll_epoch_test2.js
   - jstests/sharding/cursor1.js
   - jstests/sharding/change_streams/lookup_change_stream_post_image_id_shard_key.js
-  - jstests/sharding/database_versioning_safe_secondary_reads.js
   - jstests/sharding/change_streams_shards_start_in_sync.js
   - jstests/sharding/auth_add_shard.js
   - jstests/sharding/change_streams/lookup_change_stream_post_image_compound_shard_key.js
diff --git a/jstests/core/views/views_all_commands.js b/jstests/core/views/views_all_commands.js
index e82e5b1abb3..8072aed23a1 100644
--- a/jstests/core/views/views_all_commands.js
+++ b/jstests/core/views/views_all_commands.js
@@ -120,6 +120,7 @@ let viewsCommandTests = {
     _shardsvrDropCollectionParticipant: {skip: isAnInternalCommand},
     _shardsvrCreateCollection: {skip: isAnInternalCommand},
     _shardsvrDropDatabase: {skip: isAnInternalCommand},
+    _shardsvrDropDatabaseParticipant: {skip: isAnInternalCommand},
     _shardsvrMovePrimary: {skip: isAnInternalCommand},
     _shardsvrRefineCollectionShardKey: {skip: isAnInternalCommand},
     _shardsvrRenameCollection: {skip: isAnInternalCommand},
diff --git a/jstests/replsets/db_reads_while_recovering_all_commands.js b/jstests/replsets/db_reads_while_recovering_all_commands.js
index 5e4721a6ddd..d6e7d545b8d 100644
--- a/jstests/replsets/db_reads_while_recovering_all_commands.js
+++ b/jstests/replsets/db_reads_while_recovering_all_commands.js
@@ -70,6 +70,7 @@ const allCommands = {
     _shardsvrMovePrimary: {skip: isPrimaryOnly},
     _shardsvrRenameCollection: {skip: isPrimaryOnly},
     _shardsvrDropDatabase: {skip: isPrimaryOnly},
+    _shardsvrDropDatabaseParticipant: {skip: isPrimaryOnly},
     _shardsvrShardCollection: {skip: isPrimaryOnly},
     _shardsvrRefineCollectionShardKey: {skip: isPrimaryOnly},
     _transferMods: {skip: isPrimaryOnly},
diff --git a/jstests/sharding/database_versioning_safe_secondary_reads.js b/jstests/sharding/database_versioning_safe_secondary_reads.js
deleted file mode 100644
index f22eca34ebb..00000000000
--- a/jstests/sharding/database_versioning_safe_secondary_reads.js
+++ /dev/null
@@ -1,238 +0,0 @@
-/**
- * Tests that shards' cached in-memory and on-disk database versions are updated on primary and
- * secondary nodes when:
- * - the node does not have a cached in-memory version
- * - the node's cached in-memory version is lower than the version sent by a client
- * - the movePrimary critical section is entered on the primary node
- */
-(function() {
-"use strict";
-load("jstests/libs/database_versioning.js");
-
-const dbName = "test";
-
-const st = new ShardingTest({
-    mongos: 2,
-    rs0: {nodes: [{rsConfig: {votes: 1}}, {rsConfig: {priority: 0}}]},
-    rs1: {nodes: [{rsConfig: {votes: 1}}, {rsConfig: {priority: 0}}]},
-    verbose: 2
-});
-
-// Before creating the database, none of the nodes have a cached entry for the database either
-// in memory or on disk.
-checkInMemoryDatabaseVersion(st.rs0.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs0.getSecondary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getSecondary(), dbName, {});
-checkOnDiskDatabaseVersion(st.rs0.getPrimary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs1.getPrimary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs0.getSecondary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs1.getSecondary(), dbName, undefined);
-
-// Manually insert an entry for 'dbName' into config.databases so that we can choose shard1 as
-// the primary shard.
-assert.commandWorked(st.s.getDB("config").getCollection("databases").insert({
-    _id: dbName,
-    primary: st.rs1.name,
-    partitioned: false,
-    version: {uuid: new UUID(), lastMod: NumberInt(1)},
-}));
-
-// Check that a command that attaches databaseVersion returns empty results, even though the
-// database does not actually exist on any shard.
-assert.commandWorked(st.s.getDB(dbName).runCommand({listCollections: 1}));
-
-// Ensure the current primary shard's primary has written the new database entry to disk.
-st.rs1.getPrimary().adminCommand({_flushDatabaseCacheUpdates: dbName, syncFromConfig: false});
-
-// Ensure the database entry on the current primary shard has replicated to the secondary.
-st.rs1.awaitReplication();
-
-const dbEntry0 = st.s.getDB("config").getCollection("databases").findOne({_id: dbName});
-
-// The primary shard's primary should have refreshed its in-memory and on-disk caches. The
-// primary shard's secondary should have refreshed its on-disk entry.
-checkInMemoryDatabaseVersion(st.rs0.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getPrimary(), dbName, dbEntry0.version);
-checkInMemoryDatabaseVersion(st.rs0.getSecondary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getSecondary(), dbName, {});
-checkOnDiskDatabaseVersion(st.rs0.getPrimary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs1.getPrimary(), dbName, dbEntry0);
-checkOnDiskDatabaseVersion(st.rs0.getSecondary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs1.getSecondary(), dbName, dbEntry0);
-
-assert.commandWorked(st.s.getDB(dbName).runCommand(
-    {listCollections: 1, $readPreference: {mode: "secondary"}, readConcern: {level: "local"}}));
-
-// The primary shard's secondary should have refreshed its in-memory cache (its on-disk cache
-// was updated when the primary refreshed, above).
-checkInMemoryDatabaseVersion(st.rs0.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getPrimary(), dbName, dbEntry0.version);
-checkInMemoryDatabaseVersion(st.rs0.getSecondary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getSecondary(), dbName, dbEntry0.version);
-checkOnDiskDatabaseVersion(st.rs0.getPrimary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs1.getPrimary(), dbName, dbEntry0);
-checkOnDiskDatabaseVersion(st.rs0.getSecondary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs1.getSecondary(), dbName, dbEntry0);
-
-// Use 'movePrimary' to ensure shard0 is the primary shard.
-st.ensurePrimaryShard(dbName, st.shard0.shardName);
-const dbEntry1 = st.s.getDB("config").getCollection("databases").findOne({_id: dbName});
-assert.eq(dbEntry1.version.uuid, dbEntry0.version.uuid);
-assert.eq(dbEntry1.version.lastMod, dbEntry0.version.lastMod + 1);
-
-// Ensure the database actually gets created on the primary shard by creating a collection in
-// it.
-assert.commandWorked(st.s.getDB(dbName).runCommand({create: "foo"}));
-
-// Run a command that attaches databaseVersion to cause the current primary shard's primary to
-// refresh its in-memory cached database version.
-jsTest.log("About to do listCollections with readPref=primary");
-assert.commandWorked(st.s.getDB(dbName).runCommand({listCollections: 1}));
-
-// Ensure the current primary shard's primary has written the new database entry to disk.
-st.rs0.getPrimary().adminCommand({_flushDatabaseCacheUpdates: dbName, syncFromConfig: false});
-
-// Ensure the database entry on the current primary shard has replicated to the secondary.
-st.rs0.awaitReplication();
-
-// The primary shard's primary should have refreshed its in-memory cache and on-disk entry. The
-// primary shard's secondary will have refreshed its on-disk entry.
-checkInMemoryDatabaseVersion(st.rs0.getPrimary(), dbName, dbEntry1.version);
-checkInMemoryDatabaseVersion(st.rs1.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs0.getSecondary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getSecondary(), dbName, {});
-checkOnDiskDatabaseVersion(st.rs0.getPrimary(), dbName, dbEntry1);
-checkOnDiskDatabaseVersion(st.rs1.getPrimary(), dbName, dbEntry0);
-checkOnDiskDatabaseVersion(st.rs0.getSecondary(), dbName, dbEntry1);
-checkOnDiskDatabaseVersion(st.rs1.getSecondary(), dbName, dbEntry0);
-
-// Now run a command that attaches databaseVersion with readPref=secondary to make the current
-// primary shard's secondary refresh its in-memory database version from its on-disk entry.
-jsTest.log("About to do listCollections with readPref=secondary");
-assert.commandWorked(st.s.getDB(dbName).runCommand(
-    {listCollections: 1, $readPreference: {mode: "secondary"}, readConcern: {level: "local"}}));
-
-// The primary shard's secondary should have refreshed its in memory-cache.
-checkInMemoryDatabaseVersion(st.rs0.getPrimary(), dbName, dbEntry1.version);
-checkInMemoryDatabaseVersion(st.rs1.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs0.getSecondary(), dbName, dbEntry1.version);
-checkInMemoryDatabaseVersion(st.rs1.getSecondary(), dbName, {});
-checkOnDiskDatabaseVersion(st.rs0.getPrimary(), dbName, dbEntry1);
-checkOnDiskDatabaseVersion(st.rs1.getPrimary(), dbName, dbEntry0);
-checkOnDiskDatabaseVersion(st.rs0.getSecondary(), dbName, dbEntry1);
-checkOnDiskDatabaseVersion(st.rs1.getSecondary(), dbName, dbEntry0);
-
-// Make "staleMongos" load the stale database info into memory.
-const freshMongos = st.s0;
-const staleMongos = st.s1;
-staleMongos.getDB(dbName).runCommand({listCollections: 1});
-
-// Run movePrimary to ensure the movePrimary critical section clears the in-memory cache on the
-// old primary shard.
-jsTest.log("About to do movePrimary");
-assert.commandWorked(freshMongos.adminCommand({movePrimary: dbName, to: st.shard1.shardName}));
-const dbEntry2 = freshMongos.getDB("config").getCollection("databases").findOne({_id: dbName});
-assert.eq(dbEntry2.version.uuid, dbEntry1.version.uuid);
-assert.eq(dbEntry2.version.lastMod, dbEntry1.version.lastMod + 1);
-
-// Ensure the old primary shard's primary has written the 'enterCriticalSectionSignal' flag to
-// its on-disk database entry.
-st.rs0.getPrimary().adminCommand({_flushDatabaseCacheUpdates: dbName, syncFromConfig: false});
-
-// Ensure 'enterCriticalSectionSignal' flag has replicated to the secondary.
-st.rs0.awaitReplication();
-
-// The in-memory cached version should have been cleared on the old primary shard's primary and
-// secondary nodes.
-checkInMemoryDatabaseVersion(st.rs0.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs0.getSecondary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getSecondary(), dbName, {});
-checkOnDiskDatabaseVersion(st.rs0.getPrimary(), dbName, dbEntry1);
-checkOnDiskDatabaseVersion(st.rs1.getPrimary(), dbName, dbEntry0);
-checkOnDiskDatabaseVersion(st.rs0.getSecondary(), dbName, dbEntry1);
-checkOnDiskDatabaseVersion(st.rs1.getSecondary(), dbName, dbEntry0);
-
-// Run listCollections with readPref=secondary from the stale mongos. First, this should cause
-// the old primary shard's secondary to provoke the old primary shard's primary to refresh. Then
-// once the stale mongos refreshes, it should cause the new primary shard's secondary to provoke
-// the new primary shard's primary to refresh.
-jsTest.log("About to do listCollections with readPref=secondary after movePrimary");
-assert.commandWorked(staleMongos.getDB(dbName).runCommand(
-    {listCollections: 1, $readPreference: {mode: "secondary"}, readConcern: {level: "local"}}));
-
-// All nodes should have refreshed.
-checkInMemoryDatabaseVersion(st.rs0.getPrimary(), dbName, dbEntry2.version);
-checkInMemoryDatabaseVersion(st.rs1.getPrimary(), dbName, dbEntry2.version);
-checkInMemoryDatabaseVersion(st.rs0.getSecondary(), dbName, dbEntry2.version);
-checkInMemoryDatabaseVersion(st.rs1.getSecondary(), dbName, dbEntry2.version);
-checkOnDiskDatabaseVersion(st.rs0.getPrimary(), dbName, dbEntry2);
-checkOnDiskDatabaseVersion(st.rs1.getPrimary(), dbName, dbEntry2);
-checkOnDiskDatabaseVersion(st.rs0.getSecondary(), dbName, dbEntry2);
-checkOnDiskDatabaseVersion(st.rs1.getSecondary(), dbName, dbEntry2);
-
-// Drop the database
-jsTest.log("About to drop database from the cluster");
-assert.commandWorked(freshMongos.getDB(dbName).runCommand({dropDatabase: 1}));
-
-// Ensure the drop has replicated to all nodes.
-st.rs0.awaitReplication();
-st.rs1.awaitReplication();
-
-// The in-memory caches will be cleared on both shard's primary nodes and the on-disk database
-// entries will be cleared on all shards.
-checkInMemoryDatabaseVersion(st.rs0.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs0.getSecondary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getSecondary(), dbName, {});
-checkOnDiskDatabaseVersion(st.rs0.getPrimary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs1.getPrimary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs0.getSecondary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs1.getSecondary(), dbName, undefined);
-
-// Manually insert an entry for 'dbName' into config.databases so that we can choose shard0 as
-// the primary shard.
-assert.commandWorked(st.s.getDB("config").getCollection("databases").insert({
-    _id: dbName,
-    primary: st.rs0.name,
-    partitioned: false,
-    version: {uuid: new UUID(), lastMod: NumberInt(1)},
-}));
-
-const dbEntry = st.s.getDB("config").getCollection("databases").findOne({_id: dbName});
-
-assert.commandWorked(st.s.getDB(dbName).runCommand({listCollections: 1}));
-
-// Ensure the current primary shard's primary has written the new database entry to disk.
-st.rs0.getPrimary().adminCommand({_flushDatabaseCacheUpdates: dbName, syncFromConfig: false});
-
-// Ensure the database entry on the current primary shard has replicated to the secondary.
-st.rs0.awaitReplication();
-
-// The primary shard's primary should have refreshed its in-memory and on-disk caches.
-checkInMemoryDatabaseVersion(st.rs0.getPrimary(), dbName, dbEntry.version);
-checkInMemoryDatabaseVersion(st.rs1.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs0.getSecondary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs1.getSecondary(), dbName, {});
-checkOnDiskDatabaseVersion(st.rs0.getPrimary(), dbName, dbEntry);
-checkOnDiskDatabaseVersion(st.rs1.getPrimary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs0.getSecondary(), dbName, dbEntry);
-checkOnDiskDatabaseVersion(st.rs1.getSecondary(), dbName, undefined);
-
-assert.commandWorked(st.s.getDB(dbName).runCommand(
-    {listCollections: 1, $readPreference: {mode: "secondary"}, readConcern: {level: "local"}}));
-
-// The primary shard's secondary should have refreshed its in-memory cache (its on-disk cache
-// was already updated when the primary refreshed).
-checkInMemoryDatabaseVersion(st.rs0.getPrimary(), dbName, dbEntry.version);
-checkInMemoryDatabaseVersion(st.rs1.getPrimary(), dbName, {});
-checkInMemoryDatabaseVersion(st.rs0.getSecondary(), dbName, dbEntry.version);
-checkInMemoryDatabaseVersion(st.rs1.getSecondary(), dbName, {});
-checkOnDiskDatabaseVersion(st.rs0.getPrimary(), dbName, dbEntry);
-checkOnDiskDatabaseVersion(st.rs1.getPrimary(), dbName, undefined);
-checkOnDiskDatabaseVersion(st.rs0.getSecondary(), dbName, dbEntry);
-checkOnDiskDatabaseVersion(st.rs1.getSecondary(), dbName, undefined);
-
-st.stop();
-})();
diff --git a/jstests/sharding/libs/mongos_api_params_util.js b/jstests/sharding/libs/mongos_api_params_util.js
index ba6aa1244e0..cdd8a4961fb 100644
--- a/jstests/sharding/libs/mongos_api_params_util.js
+++ b/jstests/sharding/libs/mongos_api_params_util.js
@@ -385,8 +385,7 @@ let MongosAPIParametersUtil = (function() {
             commandName: "dropDatabase",
             run: {
                 inAPIVersion1: true,
-                configServerCommandName: "_configsvrDropDatabase",
-                shardCommandName: "dropDatabase",
+                shardCommandName: "_shardsvrDropDatabase",
                 permittedInTxn: false,
                 command: () => ({dropDatabase: 1})
             }
diff --git a/jstests/sharding/read_write_concern_defaults_application.js b/jstests/sharding/read_write_concern_defaults_application.js
index ab8ccb42eb8..a7496f2b62e 100644
--- a/jstests/sharding/read_write_concern_defaults_application.js
+++ b/jstests/sharding/read_write_concern_defaults_application.js
@@ -118,6 +118,7 @@ let testCases = {
     _shardsvrDropCollection: {skip: "internal command"},
     _shardsvrDropCollectionParticipant: {skip: "internal command"},
     _shardsvrDropDatabase: {skip: "internal command"},
+    _shardsvrDropDatabaseParticipant: {skip: "internal command"},
     _shardsvrMovePrimary: {skip: "internal command"},
     _shardsvrRefineCollectionShardKey: {skip: "internal command"},
     _shardsvrRenameCollection: {skip: "internal command"},
diff --git a/src/mongo/db/catalog/drop_database.h b/src/mongo/db/catalog/drop_database.h
index 0e3b6e59125..365a7a61e0a 100644
--- a/src/mongo/db/catalog/drop_database.h
+++ b/src/mongo/db/catalog/drop_database.h
@@ -27,6 +27,8 @@
  *    it in the license file.
  */
 
+#pragma once
+
 #include "mongo/base/status.h"
 
 namespace mongo {
diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index 1163deccec7..fa69d7b1cc3 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -327,6 +327,7 @@ env.Library(
         'config/configsvr_split_chunk_command.cpp',
         'config/configsvr_update_zone_key_range_command.cpp',
         'drop_collection_coordinator.cpp',
+        'drop_database_coordinator.cpp',
         'flush_database_cache_updates_command.cpp',
         'flush_routing_table_cache_updates_command.cpp',
         'get_database_version_command.cpp',
@@ -344,6 +345,7 @@ env.Library(
         'shardsvr_drop_collection_command.cpp',
         'shardsvr_drop_collection_participant_command.cpp',
         'shardsvr_drop_database_command.cpp',
+        'shardsvr_drop_database_participant_command.cpp',
         'shardsvr_refine_collection_shard_key_command.cpp',
         'shardsvr_rename_collection.cpp',
         'shardsvr_shard_collection_command.cpp',
diff --git a/src/mongo/db/s/drop_collection_coordinator.cpp b/src/mongo/db/s/drop_collection_coordinator.cpp
index 18220e4e0d7..99883326ab2 100644
--- a/src/mongo/db/s/drop_collection_coordinator.cpp
+++ b/src/mongo/db/s/drop_collection_coordinator.cpp
@@ -37,6 +37,8 @@
 #include "mongo/db/concurrency/lock_manager_defs.h"
 #include "mongo/db/s/database_sharding_state.h"
 #include "mongo/db/s/dist_lock_manager.h"
+#include "mongo/db/s/drop_collection_coordinator.h"
+#include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/s/catalog/type_chunk.h"
@@ -49,8 +51,6 @@
 
 namespace mongo {
 
-static constexpr int kMaxNumStaleShardVersionRetries = 10;
-
 DropCollectionCoordinator::DropCollectionCoordinator(OperationContext* opCtx,
                                                      const NamespaceString& nss)
     : ShardingDDLCoordinator(nss), _serviceContext(opCtx->getServiceContext()) {
@@ -83,34 +83,6 @@ void DropCollectionCoordinator::_sendDropCollToParticipants(OperationContext* op
     }
 }
 
-void DropCollectionCoordinator::_removeCollMetadataFromConfig(OperationContext* opCtx) {
-    IgnoreAPIParametersBlock ignoreApiParametersBlock(opCtx);
-    const auto catalogClient = Grid::get(opCtx)->catalogClient();
-
-    ON_BLOCK_EXIT([this, opCtx] {
-        Grid::get(opCtx)->catalogCache()->invalidateCollectionEntry_LINEARIZABLE(_nss);
-    });
-
-    // Remove chunk data
-    uassertStatusOK(
-        catalogClient->removeConfigDocuments(opCtx,
-                                             ChunkType::ConfigNS,
-                                             BSON(ChunkType::ns(_nss.ns())),
-                                             ShardingCatalogClient::kMajorityWriteConcern));
-    // Remove tag data
-    uassertStatusOK(
-        catalogClient->removeConfigDocuments(opCtx,
-                                             TagsType::ConfigNS,
-                                             BSON(TagsType::ns(_nss.ns())),
-                                             ShardingCatalogClient::kMajorityWriteConcern));
-    // Remove coll metadata
-    uassertStatusOK(
-        catalogClient->removeConfigDocuments(opCtx,
-                                             CollectionType::ConfigNS,
-                                             BSON(CollectionType::kNssFieldName << _nss.ns()),
-                                             ShardingCatalogClient::kMajorityWriteConcern));
-}
-
 void DropCollectionCoordinator::_stopMigrations(OperationContext* opCtx) {
     // TODO SERVER-53861 this will not stop current ongoing migrations
     uassertStatusOK(Grid::get(opCtx)->catalogClient()->updateConfigDocument(
@@ -144,7 +116,7 @@ SemiFuture<void> DropCollectionCoordinator::runImpl(
             const auto routingInfo = uassertStatusOK(
                 Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfoWithRefresh(opCtx, _nss));
 
-            _removeCollMetadataFromConfig(opCtx);
+            sharding_ddl_util::removeCollMetadataFromConfig(opCtx, _nss);
 
             if (routingInfo.isSharded()) {
                 _participants = Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx);
diff --git a/src/mongo/db/s/drop_collection_coordinator.h b/src/mongo/db/s/drop_collection_coordinator.h
index 02347408d5b..933e1e1c4f7 100644
--- a/src/mongo/db/s/drop_collection_coordinator.h
+++ b/src/mongo/db/s/drop_collection_coordinator.h
@@ -45,7 +45,6 @@ private:
     SemiFuture<void> runImpl(std::shared_ptr<executor::TaskExecutor> executor) override;
 
     void _stopMigrations(OperationContext* opCtx);
-    void _removeCollMetadataFromConfig(OperationContext* opCtx);
     void _sendDropCollToParticipants(OperationContext* opCtx);
 
     ServiceContext* _serviceContext;
diff --git a/src/mongo/db/s/drop_database_coordinator.cpp b/src/mongo/db/s/drop_database_coordinator.cpp
new file mode 100644
index 00000000000..f341e4ed6c4
--- /dev/null
+++ b/src/mongo/db/s/drop_database_coordinator.cpp
@@ -0,0 +1,164 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
+
+#include "mongo/db/s/drop_database_coordinator.h"
+
+#include "mongo/db/api_parameters.h"
+#include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/catalog_raii.h"
+#include "mongo/db/concurrency/lock_manager_defs.h"
+#include "mongo/db/s/database_sharding_state.h"
+#include "mongo/db/s/dist_lock_manager.h"
+#include "mongo/db/s/drop_collection_coordinator.h"
+#include "mongo/db/s/sharding_ddl_util.h"
+#include "mongo/logv2/log.h"
+#include "mongo/s/catalog/sharding_catalog_client.h"
+#include "mongo/s/catalog/type_chunk.h"
+#include "mongo/s/catalog/type_shard.h"
+#include "mongo/s/catalog/type_tags.h"
+#include "mongo/s/client/shard_registry.h"
+#include "mongo/s/grid.h"
+#include "mongo/s/request_types/sharded_ddl_commands_gen.h"
+#include "mongo/util/assert_util.h"
+
+namespace mongo {
+namespace {
+
+void sendCommandToAllShards(OperationContext* opCtx,
+                            StringData dbName,
+                            StringData cmdName,
+                            BSONObj cmd) {
+    auto* const shardRegistry = Grid::get(opCtx)->shardRegistry();
+    const auto participants = shardRegistry->getAllShardIds(opCtx);
+
+    for (const auto& shardId : participants) {
+        const auto& shard = uassertStatusOK(shardRegistry->getShard(opCtx, shardId));
+
+        const auto swDropResult = shard->runCommandWithFixedRetryAttempts(
+            opCtx,
+            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+            dbName.toString(),
+            CommandHelpers::appendMajorityWriteConcern(cmd),
+            Shard::RetryPolicy::kIdempotent);
+
+        uassertStatusOKWithContext(
+            Shard::CommandResponse::getEffectiveStatus(std::move(swDropResult)),
+            str::stream() << "Error processing " << cmdName << " on shard " << shardId);
+    }
+}
+
+void removeDatabaseMetadataFromConfig(OperationContext* opCtx, StringData dbName) {
+    IgnoreAPIParametersBlock ignoreApiParametersBlock(opCtx);
+    const auto catalogClient = Grid::get(opCtx)->catalogClient();
+
+    ON_BLOCK_EXIT([&, dbName = dbName.toString()] {
+        Grid::get(opCtx)->catalogCache()->purgeDatabase(dbName);
+    });
+
+    // Remove the database entry from the metadata.
+    const Status status =
+        catalogClient->removeConfigDocuments(opCtx,
+                                             DatabaseType::ConfigNS,
+                                             BSON(DatabaseType::name(dbName.toString())),
+                                             ShardingCatalogClient::kMajorityWriteConcern);
+    uassertStatusOKWithContext(status,
+                               str::stream()
+                                   << "Could not remove database metadata from config server for '"
+                                   << dbName << "'.");
+}
+
+}  // namespace
+
+DropDatabaseCoordinator::DropDatabaseCoordinator(OperationContext* opCtx, StringData dbName)
+    : ShardingDDLCoordinator({dbName, ""}), _serviceContext(opCtx->getServiceContext()) {
+    auto authSession = AuthorizationSession::get(opCtx->getClient());
+    _users =
+        userNameIteratorToContainer<std::vector<UserName>>(authSession->getImpersonatedUserNames());
+    _roles =
+        roleNameIteratorToContainer<std::vector<RoleName>>(authSession->getImpersonatedRoleNames());
+}
+
+SemiFuture<void> DropDatabaseCoordinator::runImpl(
+    std::shared_ptr<executor::TaskExecutor> executor) {
+    return ExecutorFuture<void>(executor, Status::OK())
+        .then([this, anchor = shared_from_this()]() {
+            ThreadClient tc{"DropDatabaseCoordinator", _serviceContext};
+            auto opCtxHolder = tc->makeOperationContext();
+            auto* opCtx = opCtxHolder.get();
+
+            auto authSession = AuthorizationSession::get(opCtx->getClient());
+            authSession->setImpersonatedUserData(_users, _roles);
+
+            const auto dbName = _nss.db();
+            auto distLockManager = DistLockManager::get(_serviceContext);
+            const auto dbDistLock = uassertStatusOK(distLockManager->lock(
+                opCtx, dbName, "DropDatabase", DistLockManager::kDefaultLockTimeout));
+
+            // Drop all collections under this DB
+            auto const catalogClient = Grid::get(opCtx)->catalogClient();
+            const auto allCollectionsForDb = catalogClient->getAllShardedCollectionsForDb(
+                opCtx, dbName, repl::ReadConcernLevel::kMajorityReadConcern);
+
+            for (const auto& nss : allCollectionsForDb) {
+                // TODO SERVER-53905 to support failovers here we need to store the
+                // current namespace of this loop before to delete it from config server
+                // so that on step-up we will remmeber to resume the drop collection for that
+                // namespace.
+                sharding_ddl_util::removeCollMetadataFromConfig(opCtx, nss);
+                const auto dropCollParticipantCmd = ShardsvrDropCollectionParticipant(nss);
+                sendCommandToAllShards(opCtx,
+                                       dbName,
+                                       ShardsvrDropCollectionParticipant::kCommandName,
+                                       dropCollParticipantCmd.toBSON({}));
+            }
+
+            // Drop the DB itself.
+            // The DistLockManager will prevent to re-create the database before each shard
+            // have actually dropped it locally.
+            removeDatabaseMetadataFromConfig(opCtx, dbName);
+            auto dropDatabaseParticipantCmd = ShardsvrDropDatabaseParticipant();
+            dropDatabaseParticipantCmd.setDbName(dbName);
+            sendCommandToAllShards(opCtx,
+                                   dbName,
+                                   ShardsvrDropDatabaseParticipant::kCommandName,
+                                   dropDatabaseParticipantCmd.toBSON({}));
+        })
+        .onError([this, anchor = shared_from_this()](const Status& status) {
+            LOGV2_ERROR(5281131,
+                        "Error running drop database",
+                        "database"_attr = _nss.db(),
+                        "error"_attr = redact(status));
+            return status;
+        })
+        .semi();
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/drop_database_coordinator.h b/src/mongo/db/s/drop_database_coordinator.h
new file mode 100644
index 00000000000..2a66c832175
--- /dev/null
+++ b/src/mongo/db/s/drop_database_coordinator.h
@@ -0,0 +1,51 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/auth/user_name.h"
+#include "mongo/db/s/sharding_ddl_coordinator.h"
+#include "mongo/s/shard_id.h"
+
+namespace mongo {
+
+class DropDatabaseCoordinator final : public ShardingDDLCoordinator,
+                                      public std::enable_shared_from_this<DropDatabaseCoordinator> {
+public:
+    DropDatabaseCoordinator(OperationContext* opCtx, StringData dbName);
+
+private:
+    SemiFuture<void> runImpl(std::shared_ptr<executor::TaskExecutor> executor) override;
+
+    ServiceContext* _serviceContext;
+    std::vector<UserName> _users;
+    std::vector<RoleName> _roles;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/sharding_ddl_util.cpp b/src/mongo/db/s/sharding_ddl_util.cpp
index 231b6d91863..b36aa2ffd35 100644
--- a/src/mongo/db/s/sharding_ddl_util.cpp
+++ b/src/mongo/db/s/sharding_ddl_util.cpp
@@ -47,6 +47,34 @@ namespace mongo {
 
 namespace sharding_ddl_util {
 
+void removeCollMetadataFromConfig(OperationContext* opCtx, NamespaceString nss) {
+
+    IgnoreAPIParametersBlock ignoreApiParametersBlock(opCtx);
+    const auto catalogClient = Grid::get(opCtx)->catalogClient();
+
+    ON_BLOCK_EXIT(
+        [&] { Grid::get(opCtx)->catalogCache()->invalidateCollectionEntry_LINEARIZABLE(nss); });
+
+    // Remove chunk data
+    uassertStatusOK(
+        catalogClient->removeConfigDocuments(opCtx,
+                                             ChunkType::ConfigNS,
+                                             BSON(ChunkType::ns(nss.ns())),
+                                             ShardingCatalogClient::kMajorityWriteConcern));
+    // Remove tag data
+    uassertStatusOK(
+        catalogClient->removeConfigDocuments(opCtx,
+                                             TagsType::ConfigNS,
+                                             BSON(TagsType::ns(nss.ns())),
+                                             ShardingCatalogClient::kMajorityWriteConcern));
+    // Remove coll metadata
+    uassertStatusOK(
+        catalogClient->removeConfigDocuments(opCtx,
+                                             CollectionType::ConfigNS,
+                                             BSON(CollectionType::kNssFieldName << nss.ns()),
+                                             ShardingCatalogClient::kMajorityWriteConcern));
+}
+
 void shardedRenameMetadata(OperationContext* opCtx,
                            const NamespaceString& fromNss,
                            const NamespaceString& toNss) {
diff --git a/src/mongo/db/s/sharding_ddl_util.h b/src/mongo/db/s/sharding_ddl_util.h
index 2161eebb199..16cdb020090 100644
--- a/src/mongo/db/s/sharding_ddl_util.h
+++ b/src/mongo/db/s/sharding_ddl_util.h
@@ -34,6 +34,12 @@ namespace mongo {
 namespace sharding_ddl_util {
 
 /**
+ * Erase collection metadata from config server and invalidate the locally cached once.
+ * In particular remove chunks, tags, and the description associated with the given namespace.
+ */
+void removeCollMetadataFromConfig(OperationContext* opCtx, NamespaceString nss);
+
+/**
  * Rename sharded collection metadata as part of a renameCollection operation.
  *
  * Transaction:
diff --git a/src/mongo/db/s/shardsvr_drop_database_command.cpp b/src/mongo/db/s/shardsvr_drop_database_command.cpp
index 516bb6a2b79..98bdcfed9ad 100644
--- a/src/mongo/db/s/shardsvr_drop_database_command.cpp
+++ b/src/mongo/db/s/shardsvr_drop_database_command.cpp
@@ -29,12 +29,18 @@
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
 
+#include "mongo/platform/basic.h"
+
 #include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/commands.h"
+#include "mongo/db/curop.h"
+#include "mongo/db/s/drop_database_coordinator.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/request_types/sharded_ddl_commands_gen.h"
+#include "mongo/s/sharded_collections_ddl_parameters_gen.h"
 
 namespace mongo {
 namespace {
@@ -86,7 +92,31 @@ public:
                                   << opCtx->getWriteConcern().wMode,
                     opCtx->getWriteConcern().wMode == WriteConcernOptions::kMajority);
 
-            return dropDatabaseLegacy(opCtx, request().getDbName());
+            const auto dbName = request().getDbName();
+
+            if (!feature_flags::gShardingFullDDLSupport.isEnabled(
+                    serverGlobalParams.featureCompatibility) ||
+                feature_flags::gDisableIncompleteShardingDDLSupport.isEnabled(
+                    serverGlobalParams.featureCompatibility)) {
+
+                LOGV2_DEBUG(
+                    5281110, 1, "Running legacy drop database procedure", "database"_attr = dbName);
+                return dropDatabaseLegacy(opCtx, dbName);
+            }
+
+            LOGV2_DEBUG(
+                5281111, 1, "Running new drop database procedure", "database"_attr = dbName);
+
+            // Since this operation is not directly writing locally we need to force its db
+            // profile level increase in order to be logged in "<db>.system.profile"
+            CurOp::get(opCtx)->raiseDbProfileLevel(
+                CollectionCatalog::get(opCtx)->getDatabaseProfileLevel(dbName));
+
+            auto dropDatabaseCoordinator = std::make_shared<DropDatabaseCoordinator>(opCtx, dbName);
+            dropDatabaseCoordinator->run(opCtx).get();
+
+            // The following response can be omitted once 5.0 became last LTS
+            return Response();
         }
 
     private:
diff --git a/src/mongo/db/s/shardsvr_drop_database_participant_command.cpp b/src/mongo/db/s/shardsvr_drop_database_participant_command.cpp
new file mode 100644
index 00000000000..25b40a522a1
--- /dev/null
+++ b/src/mongo/db/s/shardsvr_drop_database_participant_command.cpp
@@ -0,0 +1,117 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/catalog/drop_database.h"
+#include "mongo/db/catalog_raii.h"
+#include "mongo/db/commands.h"
+#include "mongo/db/s/database_sharding_state.h"
+#include "mongo/db/s/sharding_state.h"
+#include "mongo/logv2/log.h"
+#include "mongo/s/request_types/sharded_ddl_commands_gen.h"
+
+namespace mongo {
+namespace {
+
+class ShardsvrDropDatabaseParticipantCommand final
+    : public TypedCommand<ShardsvrDropDatabaseParticipantCommand> {
+public:
+    bool acceptsAnyApiVersionParameters() const override {
+        return true;
+    }
+
+    AllowedOnSecondary secondaryAllowed(ServiceContext*) const override {
+        return Command::AllowedOnSecondary::kNever;
+    }
+
+    std::string help() const override {
+        return "Internal command, which is exported by secondary sharding servers. Do not call "
+               "directly. Participates in droping a database.";
+    }
+
+    using Request = ShardsvrDropDatabaseParticipant;
+
+    class Invocation final : public InvocationBase {
+    public:
+        using InvocationBase::InvocationBase;
+
+        void typedRun(OperationContext* opCtx) {
+            uassertStatusOK(ShardingState::get(opCtx)->canAcceptShardedCommands());
+            uassert(ErrorCodes::InvalidOptions,
+                    str::stream() << Request::kCommandName
+                                  << " must be called with majority writeConcern, got "
+                                  << opCtx->getWriteConcern().wMode,
+                    opCtx->getWriteConcern().wMode == WriteConcernOptions::kMajority);
+
+            const auto& dbName = request().getDbName();
+
+            try {
+                uassertStatusOK(dropDatabase(opCtx, dbName.toString()));
+            } catch (const ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
+                LOGV2_DEBUG(5281101,
+                            1,
+                            "Received a ShardsvrDropDatabaseParticipant but did not find the "
+                            "database locally",
+                            "database"_attr = dbName);
+            }
+
+            {
+                // Clear CollectionShardingRuntime entry
+                UninterruptibleLockGuard noInterrupt(opCtx->lockState());
+                Lock::DBLock dbLock(opCtx, dbName, MODE_X);
+                auto dss = DatabaseShardingState::get(opCtx, dbName);
+                dss->clearDatabaseInfo(opCtx);
+            }
+        }
+
+    private:
+        NamespaceString ns() const override {
+            return {request().getDbName(), ""};
+        }
+
+        bool supportsWriteConcern() const override {
+            return true;
+        }
+
+        void doCheckAuthorization(OperationContext* opCtx) const override {
+            uassert(ErrorCodes::Unauthorized,
+                    "Unauthorized",
+                    AuthorizationSession::get(opCtx->getClient())
+                        ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(),
+                                                           ActionType::internal));
+        }
+    };
+} sharsvrdDropCollectionParticipantCommand;
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/s/commands/SConscript b/src/mongo/s/commands/SConscript
index e04d7dca413..bf2e8ab0297 100644
--- a/src/mongo/s/commands/SConscript
+++ b/src/mongo/s/commands/SConscript
@@ -42,7 +42,7 @@ env.Library(
         'cluster_data_size_cmd.cpp',
         'cluster_db_stats_cmd.cpp',
         'cluster_distinct_cmd.cpp',
-        'cluster_drop_cmd.cpp',
+        'cluster_drop_collection_cmd.cpp',
         'cluster_drop_database_cmd.cpp',
         'cluster_drop_indexes_cmd.cpp',
         'cluster_enable_sharding_cmd.cpp',
diff --git a/src/mongo/s/commands/cluster_drop_cmd.cpp b/src/mongo/s/commands/cluster_drop_collection_cmd.cpp
index 2892b9c67c2..2892b9c67c2 100644
--- a/src/mongo/s/commands/cluster_drop_cmd.cpp
+++ b/src/mongo/s/commands/cluster_drop_collection_cmd.cpp
diff --git a/src/mongo/s/request_types/sharded_ddl_commands.idl b/src/mongo/s/request_types/sharded_ddl_commands.idl
index f7382b78c3e..5c858a54f24 100644
--- a/src/mongo/s/request_types/sharded_ddl_commands.idl
+++ b/src/mongo/s/request_types/sharded_ddl_commands.idl
@@ -103,6 +103,14 @@ commands:
         # The reply can completely removed once 5.0 became last lts
         reply_type: DropDatabaseReply
 
+    _shardsvrDropDatabaseParticipant:
+        description: "Internal command sent to participants shards to drop a database."
+        command_name: _shardsvrDropDatabaseParticipant
+        namespace: ignored
+        api_version: ""
+        cpp_name: ShardsvrDropDatabaseParticipant
+        strict: false
+
     _shardsvrDropCollection:
         description: "Parser for the _shardsvrDropCollection command"
         command_name: _shardsvrDropCollection