SERVER-60007 Implement command to drop collection if its UUID differs from the expected

10 files changed, 261 insertions, 26 deletions

diff --git a/jstests/core/views/views_all_commands.js b/jstests/core/views/views_all_commands.js
index be52f9b4928..15f4fb71e89 100644
--- a/jstests/core/views/views_all_commands.js
+++ b/jstests/core/views/views_all_commands.js
@@ -134,6 +134,7 @@ let viewsCommandTests = {
     _shardsvrAbortReshardCollection: {skip: isAnInternalCommand},
     _shardsvrCloneCatalogData: {skip: isAnInternalCommand},
     _shardsvrDropCollection: {skip: isAnInternalCommand},
+    _shardsvrDropCollectionIfUUIDNotMatching: {skip: isUnrelated},
     _shardsvrDropCollectionParticipant: {skip: isAnInternalCommand},
     _shardsvrCleanupReshardCollection: {skip: isAnInternalCommand},
     _shardsvrCommitReshardCollection: {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 ab8aea876f4..6a5143f1373 100644
--- a/jstests/replsets/db_reads_while_recovering_all_commands.js
+++ b/jstests/replsets/db_reads_while_recovering_all_commands.js
@@ -74,6 +74,7 @@ const allCommands = {
     _shardsvrCommitReshardCollection: {skip: isPrimaryOnly},
     _shardsvrDropCollection: {skip: isPrimaryOnly},
     _shardsvrCreateCollection: {skip: isPrimaryOnly},
+    _shardsvrDropCollectionIfUUIDNotMatching: {skip: isNotAUserDataRead},
     _shardsvrDropCollectionParticipant: {skip: isPrimaryOnly},
     _shardsvrCreateCollectionParticipant: {skip: isPrimaryOnly},
     _shardsvrMovePrimary: {skip: isPrimaryOnly},
diff --git a/jstests/sharding/drop_collection_if_uuid_not_matching.js b/jstests/sharding/drop_collection_if_uuid_not_matching.js
new file mode 100644
index 00000000000..59dde47a364
--- /dev/null
+++ b/jstests/sharding/drop_collection_if_uuid_not_matching.js
@@ -0,0 +1,43 @@
+/**
+ * Tests that the _shardsvrDropCollectionIfUUIDNotMatching command works as expected:
+ * - Noop in case the collection doesn't exist.
+ * - Drop collection if uuid different from the expected.
+ * - Keep the collection if the uuid is exactly the expected one.
+ *
+ * @tags: [
+ *     requires_fcv_51, // The command is not present in v5.0
+ *     does_not_support_stepdowns, // The command is not resilient to stepdowns
+ * ]
+ */
+
+"use strict";
+
+const dbName = "test";
+const collName = "coll";
+const ns = dbName + "." + collName;
+
+const st = new ShardingTest({shards: 1});
+const mongos = st.s;
+db = st.rs0.getPrimary().getDB(dbName);
+
+assert.commandWorked(mongos.adminCommand({enableSharding: dbName}));
+
+// Non-existing collection with a random expected UUID (command succeeds, noop)
+assert.commandWorked(db.runCommand(
+    {_shardsvrDropCollectionIfUUIDNotMatching: collName, expectedCollectionUUID: UUID()}));
+
+// Existing collection with a random expected UUID (command succeeds after successful drop)
+assert.commandWorked(db.getCollection(collName).insert({_id: 0}));
+assert.commandWorked(db.runCommand(
+    {_shardsvrDropCollectionIfUUIDNotMatching: collName, expectedCollectionUUID: UUID()}));
+assert.eq(null, db.getCollection(collName).findOne({_id: 0}));
+
+// Existing collection with the expected UUID (command succeeds but no drop)
+assert.commandWorked(st.s.adminCommand({shardCollection: ns, key: {_id: 1}}));
+const collUUID = st.config.collections.findOne({_id: ns}).uuid;
+assert.commandWorked(db.getCollection(collName).insert({_id: 0}));
+assert.commandWorked(db.runCommand(
+    {_shardsvrDropCollectionIfUUIDNotMatching: collName, expectedCollectionUUID: collUUID}));
+assert.neq(null, db.getCollection(collName).findOne({_id: 0}));
+
+st.stop();
diff --git a/jstests/sharding/read_write_concern_defaults_application.js b/jstests/sharding/read_write_concern_defaults_application.js
index bbb0207ef02..1f1c2f4e69a 100644
--- a/jstests/sharding/read_write_concern_defaults_application.js
+++ b/jstests/sharding/read_write_concern_defaults_application.js
@@ -135,6 +135,7 @@ let testCases = {
     _shardsvrCreateCollection: {skip: "internal command"},
     _shardsvrCreateCollectionParticipant: {skip: "internal command"},
     _shardsvrDropCollection: {skip: "internal command"},
+    _shardsvrDropCollectionIfUUIDNotMatching: {skip: "internal command"},
     _shardsvrDropCollectionParticipant: {skip: "internal command"},
     _shardsvrDropDatabase: {skip: "internal command"},
     _shardsvrDropDatabaseParticipant: {skip: "internal command"},
diff --git a/src/mongo/db/catalog/drop_collection.cpp b/src/mongo/db/catalog/drop_collection.cpp
index 61846605fbe..932b6145757 100644
--- a/src/mongo/db/catalog/drop_collection.cpp
+++ b/src/mongo/db/catalog/drop_collection.cpp
@@ -131,7 +131,8 @@ Status _abortIndexBuildsAndDrop(OperationContext* opCtx,
                                 const NamespaceString& startingNss,
                                 std::function<Status(Database*, const NamespaceString&)>&& dropFn,
                                 DropReply* reply,
-                                bool appendNs = true) {
+                                bool appendNs = true,
+                                boost::optional<UUID> dropIfUUIDNotMatching = boost::none) {
     // We only need to hold an intent lock to send abort signals to the active index builder on this
     // collection.
     boost::optional<AutoGetDb> optionalAutoDb(std::move(autoDb));
@@ -165,6 +166,9 @@ Status _abortIndexBuildsAndDrop(OperationContext* opCtx,
 
     IndexBuildsCoordinator* indexBuildsCoord = IndexBuildsCoordinator::get(opCtx);
     const UUID collectionUUID = coll->uuid();
+    if (dropIfUUIDNotMatching && collectionUUID == *dropIfUUIDNotMatching) {
+        return Status::OK();
+    }
     const NamespaceStringOrUUID dbAndUUID{coll->ns().db().toString(), coll->uuid()};
     const int numIndexes = coll->getIndexCatalog()->numIndexesTotal(opCtx);
 
@@ -233,12 +237,12 @@ Status _abortIndexBuildsAndDrop(OperationContext* opCtx,
     return Status::OK();
 }
 
-Status _dropCollection(OperationContext* opCtx,
-                       Database* db,
-                       const NamespaceString& collectionName,
-                       const repl::OpTime& dropOpTime,
-                       DropCollectionSystemCollectionMode systemCollectionMode,
-                       DropReply* reply) {
+Status _dropCollectionForApplyOps(OperationContext* opCtx,
+                                  Database* db,
+                                  const NamespaceString& collectionName,
+                                  const repl::OpTime& dropOpTime,
+                                  DropCollectionSystemCollectionMode systemCollectionMode,
+                                  DropReply* reply) {
     Lock::CollectionLock collLock(opCtx, collectionName, MODE_X);
     const CollectionPtr& coll =
         CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, collectionName);
@@ -281,23 +285,11 @@ Status _dropCollection(OperationContext* opCtx,
     return Status::OK();
 }
 
-Status dropCollection(OperationContext* opCtx,
-                      const NamespaceString& nss,
-                      DropReply* reply,
-                      DropCollectionSystemCollectionMode systemCollectionMode) {
-    if (!serverGlobalParams.quiet.load()) {
-        LOGV2(518070, "CMD: drop", logAttrs(nss));
-    }
-
-    if (MONGO_unlikely(hangDropCollectionBeforeLockAcquisition.shouldFail())) {
-        LOGV2(518080, "Hanging drop collection before lock acquisition while fail point is set");
-        hangDropCollectionBeforeLockAcquisition.pauseWhileSet();
-    }
-
-    // We rewrite drop of time-series buckets collection to drop of time-series view collection.
-    // This ensures that such drop will delete both collections.
-    const auto collectionName =
-        nss.isTimeseriesBucketsCollection() ? nss.getTimeseriesViewNamespace() : nss;
+Status _dropCollection(OperationContext* opCtx,
+                       const NamespaceString& collectionName,
+                       DropReply* reply,
+                       DropCollectionSystemCollectionMode systemCollectionMode,
+                       boost::optional<UUID> dropIfUUIDNotMatching = boost::none) {
 
     try {
         return writeConflictRetry(opCtx, "drop", collectionName.ns(), [&] {
@@ -326,7 +318,9 @@ Status dropCollection(OperationContext* opCtx,
                         wuow.commit();
                         return Status::OK();
                     },
-                    reply);
+                    reply,
+                    true /* appendNs */,
+                    dropIfUUIDNotMatching);
             }
 
             auto dropTimeseries = [opCtx, &autoDb, &collectionName, &reply](
@@ -393,6 +387,51 @@ Status dropCollection(OperationContext* opCtx,
     }
 }
 
+Status dropCollection(OperationContext* opCtx,
+                      const NamespaceString& nss,
+                      DropReply* reply,
+                      DropCollectionSystemCollectionMode systemCollectionMode) {
+    if (!serverGlobalParams.quiet.load()) {
+        LOGV2(518070, "CMD: drop", logAttrs(nss));
+    }
+
+    if (MONGO_unlikely(hangDropCollectionBeforeLockAcquisition.shouldFail())) {
+        LOGV2(518080, "Hanging drop collection before lock acquisition while fail point is set");
+        hangDropCollectionBeforeLockAcquisition.pauseWhileSet();
+    }
+
+    // We rewrite drop of time-series buckets collection to drop of time-series view collection.
+    // This ensures that such drop will delete both collections.
+    const auto collectionName =
+        nss.isTimeseriesBucketsCollection() ? nss.getTimeseriesViewNamespace() : nss;
+
+    return _dropCollection(opCtx, collectionName, reply, systemCollectionMode);
+}
+
+Status dropCollectionIfUUIDNotMatching(OperationContext* opCtx,
+                                       const NamespaceString& ns,
+                                       const UUID& expectedUUID) {
+    AutoGetDb autoDb(opCtx, ns.db(), MODE_IX);
+    if (autoDb.getDb()) {
+        {
+            Lock::CollectionLock collLock(opCtx, ns, MODE_IS);
+            const auto coll = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, ns);
+            if (!coll || coll->uuid() == expectedUUID) {
+                return Status::OK();
+            }
+        }
+
+        DropReply repl;
+        return _dropCollection(opCtx,
+                               ns,
+                               &repl,
+                               DropCollectionSystemCollectionMode::kDisallowSystemCollectionDrops,
+                               expectedUUID);
+    }
+
+    return Status::OK();
+}
+
 Status dropCollectionForApplyOps(OperationContext* opCtx,
                                  const NamespaceString& collectionName,
                                  const repl::OpTime& dropOpTime,
@@ -419,7 +458,7 @@ Status dropCollectionForApplyOps(OperationContext* opCtx,
         if (!coll) {
             return _dropView(opCtx, db, collectionName, &unusedReply);
         } else {
-            return _dropCollection(
+            return _dropCollectionForApplyOps(
                 opCtx, db, collectionName, dropOpTime, systemCollectionMode, &unusedReply);
         }
     });
diff --git a/src/mongo/db/catalog/drop_collection.h b/src/mongo/db/catalog/drop_collection.h
index 43362d3cb11..e6cd613f260 100644
--- a/src/mongo/db/catalog/drop_collection.h
+++ b/src/mongo/db/catalog/drop_collection.h
@@ -57,6 +57,13 @@ Status dropCollection(OperationContext* opCtx,
                       DropCollectionSystemCollectionMode systemCollectionMode);
 
 /**
+ * Drops the collection "collectionName" only if its uuid is not matching "expectedUUID".
+ */
+Status dropCollectionIfUUIDNotMatching(OperationContext* opCtx,
+                                       const NamespaceString& ns,
+                                       const UUID& expectedUUID);
+
+/**
  * Drops the collection "collectionName". When applying a 'drop' oplog entry on a secondary, the
  * 'dropOpTime' will contain the optime of the oplog entry.
  */
diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index e11ff37a646..1678dbdffb4 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -336,6 +336,7 @@ env.Library(
         'shardsvr_create_collection_command.cpp',
         'shardsvr_create_collection_participant_command.cpp',
         'shardsvr_drop_collection_command.cpp',
+        'shardsvr_drop_collection_if_uuid_not_matching_command.cpp',
         'shardsvr_drop_collection_participant_command.cpp',
         'shardsvr_drop_database_command.cpp',
         'shardsvr_drop_database_participant_command.cpp',
diff --git a/src/mongo/db/s/shardsvr_drop_collection_if_uuid_not_matching_command.cpp b/src/mongo/db/s/shardsvr_drop_collection_if_uuid_not_matching_command.cpp
new file mode 100644
index 00000000000..75920fd592e
--- /dev/null
+++ b/src/mongo/db/s/shardsvr_drop_collection_if_uuid_not_matching_command.cpp
@@ -0,0 +1,92 @@
+/**
+ *    Copyright (C) 2021-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/auth/authorization_session.h"
+#include "mongo/db/catalog/drop_collection.h"
+#include "mongo/db/commands.h"
+#include "mongo/db/s/sharding_state.h"
+#include "mongo/logv2/log.h"
+#include "mongo/s/request_types/drop_collection_if_uuid_not_matching_gen.h"
+
+namespace mongo {
+namespace {
+
+class ShardsvrDropCollectionIfUUIDNotMatchingCommand final
+    : public TypedCommand<ShardsvrDropCollectionIfUUIDNotMatchingCommand> {
+public:
+    bool skipApiVersionCheck() const override {
+        /* Internal command (server to server) */
+        return true;
+    }
+
+    AllowedOnSecondary secondaryAllowed(ServiceContext*) const override {
+        return Command::AllowedOnSecondary::kNever;
+    }
+
+    std::string help() const override {
+        return "Internal command aimed to remove stale entries from the local collection catalog.";
+    }
+
+    using Request = ShardsvrDropCollectionIfUUIDNotMatchingRequest;
+
+    class Invocation final : public InvocationBase {
+    public:
+        using InvocationBase::InvocationBase;
+
+        void typedRun(OperationContext* opCtx) {
+            uassertStatusOK(ShardingState::get(opCtx)->canAcceptShardedCommands());
+            opCtx->setAlwaysInterruptAtStepDownOrUp();
+
+            uassertStatusOK(dropCollectionIfUUIDNotMatching(
+                opCtx, ns(), request().getExpectedCollectionUUID()));
+        }
+
+    private:
+        NamespaceString ns() const override {
+            return request().getNamespace();
+        }
+
+        bool supportsWriteConcern() const override {
+            return false;
+        }
+
+        void doCheckAuthorization(OperationContext* opCtx) const override {
+            uassert(ErrorCodes::Unauthorized,
+                    "Unauthorized",
+                    AuthorizationSession::get(opCtx->getClient())
+                        ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(),
+                                                           ActionType::dropCollection));
+        }
+    };
+} shardSvrDropCollectionIfUUIDNotMatching;
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index 6d4d42b8f6d..be3cf25e9ce 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -187,6 +187,7 @@ env.Library(
         'request_types/commit_chunk_migration_request_type.cpp',
         'request_types/commit_reshard_collection.idl',
         'request_types/configure_collection_auto_split.idl',
+        'request_types/drop_collection_if_uuid_not_matching.idl',
         'request_types/ensure_chunk_version_is_greater_than.idl',
         'request_types/flush_database_cache_updates.idl',
         'request_types/flush_resharding_state_change.idl',
diff --git a/src/mongo/s/request_types/drop_collection_if_uuid_not_matching.idl b/src/mongo/s/request_types/drop_collection_if_uuid_not_matching.idl
new file mode 100644
index 00000000000..7b9f30c7fce
--- /dev/null
+++ b/src/mongo/s/request_types/drop_collection_if_uuid_not_matching.idl
@@ -0,0 +1,49 @@
+# Copyright(C) 2021 - 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.
+#
+
+# This IDL file describes the BSON format for the _shardsvrDropCollectionIfUUIDNotMatching command.
+
+global:
+  cpp_namespace: "mongo"
+
+imports:
+    - "mongo/idl/basic_types.idl"
+
+commands:
+  _shardsvrDropCollectionIfUUIDNotMatching:
+    command_name: _shardsvrDropCollectionIfUUIDNotMatching
+    cpp_name: ShardsvrDropCollectionIfUUIDNotMatchingRequest
+    description: "Internal dropCollectionIfUUIDNotMatching request."
+    strict: false
+    namespace: concatenate_with_db
+    api_version: ""
+    fields:
+        expectedCollectionUUID:
+            type: uuid
+            description: "The expected collection UUID: if the local catalog has a different uuid
+                         associated to the namespace, the collection will be dropped."