SERVER-57114 Change refine collection shard key command to use ShardingDDLCoordinator

...  instead of ShardinDDLCoordinator_NORESILIENT

10 files changed, 161 insertions, 61 deletions

diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index 825d057fc3e..49498dde8d5 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -342,6 +342,7 @@ env.Library(
         'sharded_rename_collection.idl',
         'rename_collection_participant_service.cpp',
         'refine_collection_shard_key_coordinator.cpp',
+        'refine_collection_shard_key_coordinator_document.idl',
         'reshard_collection_coordinator.cpp',
         'resharding_test_commands.cpp',
         'resharding_test_commands.idl',
diff --git a/src/mongo/db/s/create_collection_coordinator.cpp b/src/mongo/db/s/create_collection_coordinator.cpp
index 388366a8632..625361e7f20 100644
--- a/src/mongo/db/s/create_collection_coordinator.cpp
+++ b/src/mongo/db/s/create_collection_coordinator.cpp
@@ -348,7 +348,7 @@ boost::optional<BSONObj> CreateCollectionCoordinator::reportForCurrentOp(
 void CreateCollectionCoordinator::checkIfOptionsConflict(const BSONObj& doc) const {
     // If we have two shard collections on the same namespace, then the arguments must be the same.
     const auto otherDoc = CreateCollectionCoordinatorDocument::parse(
-        IDLParserErrorContext("RenameCollectionCoordinatorDocument"), doc);
+        IDLParserErrorContext("CreateCollectionCoordinatorDocument"), doc);
 
     uassert(ErrorCodes::ConflictingOperationInProgress,
             "Another create collection with different arguments is already running for the same "
diff --git a/src/mongo/db/s/refine_collection_shard_key_coordinator.cpp b/src/mongo/db/s/refine_collection_shard_key_coordinator.cpp
index d4eae301de6..d56ef28f19f 100644
--- a/src/mongo/db/s/refine_collection_shard_key_coordinator.cpp
+++ b/src/mongo/db/s/refine_collection_shard_key_coordinator.cpp
@@ -41,61 +41,88 @@
 namespace mongo {
 
 RefineCollectionShardKeyCoordinator::RefineCollectionShardKeyCoordinator(
-    OperationContext* opCtx, const NamespaceString& nss, const KeyPattern newShardKey)
-    : ShardingDDLCoordinator_NORESILIENT(opCtx, nss),
-      _serviceContext(opCtx->getServiceContext()),
-      _newShardKey(std::move(newShardKey)){};
-
-SemiFuture<void> RefineCollectionShardKeyCoordinator::runImpl(
-    std::shared_ptr<executor::TaskExecutor> executor) {
-    return ExecutorFuture<void>(executor, Status::OK())
-        .then([this, anchor = shared_from_this()]() {
-            ThreadClient tc{"RefineCollectionShardKeyCoordinator", _serviceContext};
-            auto opCtxHolder = tc->makeOperationContext();
+    ShardingDDLCoordinatorService* service, const BSONObj& initialState)
+    : ShardingDDLCoordinator(service, initialState),
+      _doc(RefineCollectionShardKeyCoordinatorDocument::parse(
+          IDLParserErrorContext("RefineCollectionShardKeyCoordinatorDocument"), initialState)),
+      _newShardKey(_doc.getNewShardKey()) {}
+
+
+void RefineCollectionShardKeyCoordinator::checkIfOptionsConflict(const BSONObj& doc) const {
+    // If we have two shard collections on the same namespace, then the arguments must be the same.
+    const auto otherDoc = RefineCollectionShardKeyCoordinatorDocument::parse(
+        IDLParserErrorContext("RefineCollectionShardKeyCoordinatorDocument"), doc);
+
+    uassert(ErrorCodes::ConflictingOperationInProgress,
+            "Another create collection with different arguments is already running for the same "
+            "namespace",
+            SimpleBSONObjComparator::kInstance.evaluate(
+                _doc.getRefineCollectionShardKeyRequest().toBSON() ==
+                otherDoc.getRefineCollectionShardKeyRequest().toBSON()));
+}
+
+boost::optional<BSONObj> RefineCollectionShardKeyCoordinator::reportForCurrentOp(
+    MongoProcessInterface::CurrentOpConnectionsMode connMode,
+    MongoProcessInterface::CurrentOpSessionsMode sessionMode) noexcept {
+    BSONObjBuilder cmdBob;
+    if (const auto& optComment = getForwardableOpMetadata().getComment()) {
+        cmdBob.append(optComment.get().firstElement());
+    }
+    cmdBob.appendElements(_doc.getRefineCollectionShardKeyRequest().toBSON());
+
+    BSONObjBuilder bob;
+    bob.append("type", "op");
+    bob.append("desc", "RefineCollectionShardKeyCoordinator");
+    bob.append("op", "command");
+    bob.append("ns", nss().toString());
+    bob.append("command", cmdBob.obj());
+    bob.append("active", true);
+    return bob.obj();
+}
+
+ExecutorFuture<void> RefineCollectionShardKeyCoordinator::_runImpl(
+    std::shared_ptr<executor::ScopedTaskExecutor> executor,
+    const CancellationToken& token) noexcept {
+    return ExecutorFuture<void>(**executor)
+        .then([this, anchor = shared_from_this()] {
+            auto opCtxHolder = cc().makeOperationContext();
             auto* opCtx = opCtxHolder.get();
-            _forwardableOpMetadata.setOn(opCtx);
-
-            auto distLockManager = DistLockManager::get(opCtx->getServiceContext());
-            const auto dbDistLock =
-                uassertStatusOK(distLockManager->lock(opCtx,
-                                                      _nss.db(),
-                                                      "RefineCollectionShardKey",
-                                                      DistLockManager::kDefaultLockTimeout));
-            const auto collDistLock =
-                uassertStatusOK(distLockManager->lock(opCtx,
-                                                      _nss.ns(),
-                                                      "RefineCollectionShardKey",
-                                                      DistLockManager::kDefaultLockTimeout));
+            getForwardableOpMetadata().setOn(opCtx);
 
             const auto cm = uassertStatusOK(
-                Grid::get(opCtx)->catalogCache()->getShardedCollectionRoutingInfoWithRefresh(opCtx,
-                                                                                             _nss));
+                Grid::get(opCtx)->catalogCache()->getShardedCollectionRoutingInfoWithRefresh(
+                    opCtx, nss()));
             ConfigsvrRefineCollectionShardKey configsvrRefineCollShardKey(
-                _nss, _newShardKey.toBSON(), cm.getVersion().epoch());
-            configsvrRefineCollShardKey.setDbName(_nss.db().toString());
+                nss(), _newShardKey.toBSON(), cm.getVersion().epoch());
+            configsvrRefineCollShardKey.setDbName(nss().db().toString());
             // TODO SERVER-54810 don't set `setIsFromPrimaryShard` once 5.0 becomes last-LTS
             configsvrRefineCollShardKey.setIsFromPrimaryShard(true);
 
             auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
-            auto cmdResponse = uassertStatusOK(configShard->runCommandWithFixedRetryAttempts(
-                opCtx,
-                ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                "admin",
-                CommandHelpers::appendMajorityWriteConcern(configsvrRefineCollShardKey.toBSON({}),
-                                                           opCtx->getWriteConcern()),
-                Shard::RetryPolicy::kIdempotent));
-
-            uassertStatusOK(cmdResponse.commandStatus);
-            uassertStatusOK(cmdResponse.writeConcernStatus);
+
+            try {
+                auto cmdResponse = uassertStatusOK(configShard->runCommandWithFixedRetryAttempts(
+                    opCtx,
+                    ReadPreferenceSetting(ReadPreference::PrimaryOnly),
+                    NamespaceString::kAdminDb.toString(),
+                    CommandHelpers::appendMajorityWriteConcern(
+                        configsvrRefineCollShardKey.toBSON({}), opCtx->getWriteConcern()),
+                    Shard::RetryPolicy::kIdempotent));
+
+                uassertStatusOK(cmdResponse.commandStatus);
+                uassertStatusOK(cmdResponse.writeConcernStatus);
+            } catch (const DBException&) {
+                _completeOnError = true;
+                throw;
+            }
         })
         .onError([this, anchor = shared_from_this()](const Status& status) {
             LOGV2_ERROR(5277700,
                         "Error running refine collection shard key",
-                        "namespace"_attr = _nss,
+                        "namespace"_attr = nss(),
                         "error"_attr = redact(status));
             return status;
-        })
-        .semi();
+        });
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/refine_collection_shard_key_coordinator.h b/src/mongo/db/s/refine_collection_shard_key_coordinator.h
index 417278bd008..6a0468facd3 100644
--- a/src/mongo/db/s/refine_collection_shard_key_coordinator.h
+++ b/src/mongo/db/s/refine_collection_shard_key_coordinator.h
@@ -30,23 +30,27 @@
 #pragma once
 
 #include "mongo/db/keypattern.h"
+#include "mongo/db/s/refine_collection_shard_key_coordinator_document_gen.h"
 #include "mongo/db/s/sharding_ddl_coordinator.h"
 
 namespace mongo {
 
-class RefineCollectionShardKeyCoordinator final
-    : public ShardingDDLCoordinator_NORESILIENT,
-      public std::enable_shared_from_this<RefineCollectionShardKeyCoordinator> {
+class RefineCollectionShardKeyCoordinator final : public ShardingDDLCoordinator {
 public:
-    RefineCollectionShardKeyCoordinator(OperationContext* opCtx,
-                                        const NamespaceString& nss,
-                                        const KeyPattern newShardKey);
+    RefineCollectionShardKeyCoordinator(ShardingDDLCoordinatorService* service,
+                                        const BSONObj& initialState);
 
-private:
-    SemiFuture<void> runImpl(std::shared_ptr<executor::TaskExecutor> executor) override;
+    void checkIfOptionsConflict(const BSONObj& coorDoc) const override;
+
+    boost::optional<BSONObj> reportForCurrentOp(
+        MongoProcessInterface::CurrentOpConnectionsMode connMode,
+        MongoProcessInterface::CurrentOpSessionsMode sessionMode) noexcept override;
 
-    ServiceContext* _serviceContext;
+private:
+    ExecutorFuture<void> _runImpl(std::shared_ptr<executor::ScopedTaskExecutor> executor,
+                                  const CancellationToken& token) noexcept override;
 
+    RefineCollectionShardKeyCoordinatorDocument _doc;
     const KeyPattern _newShardKey;
 };
 
diff --git a/src/mongo/db/s/refine_collection_shard_key_coordinator_document.idl b/src/mongo/db/s/refine_collection_shard_key_coordinator_document.idl
new file mode 100644
index 00000000000..4fa8173b6b2
--- /dev/null
+++ b/src/mongo/db/s/refine_collection_shard_key_coordinator_document.idl
@@ -0,0 +1,47 @@
+# 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 file defines the format of documents stored in config.ddl.refineCollectionShardKey, used by the
+# shard coordinator to serialize with other DDL operations.
+
+global:
+    cpp_namespace: "mongo"
+
+imports:
+    - "mongo/idl/basic_types.idl"
+    - "mongo/db/s/sharding_ddl_coordinator.idl"
+    - "mongo/s/request_types/sharded_ddl_commands.idl"
+
+structs:
+    RefineCollectionShardKeyCoordinatorDocument:
+        description: "Object with neccessary fields to refine a collection's shard key"
+        generate_comparison_operators: false
+        strict: true
+        chained_structs:
+            ShardingDDLCoordinatorMetadata: ShardingDDLCoordinatorMetadata
+            RefineCollectionShardKeyRequest: RefineCollectionShardKeyRequest
diff --git a/src/mongo/db/s/sharding_ddl_coordinator.idl b/src/mongo/db/s/sharding_ddl_coordinator.idl
index 3a28d9f2e67..57ff72c6a87 100644
--- a/src/mongo/db/s/sharding_ddl_coordinator.idl
+++ b/src/mongo/db/s/sharding_ddl_coordinator.idl
@@ -46,6 +46,7 @@ enums:
             kDropCollection: "dropCollection"
             kRenameCollection: "renameCollection"
             kCreateCollection: "createCollection"
+            kRefineCollectionShardKey: "refineCollectionShardKey"
 
 types:
     ForwardableOperationMetadata:
diff --git a/src/mongo/db/s/sharding_ddl_coordinator_service.cpp b/src/mongo/db/s/sharding_ddl_coordinator_service.cpp
index ae27f9ff9eb..1cbd546a6f5 100644
--- a/src/mongo/db/s/sharding_ddl_coordinator_service.cpp
+++ b/src/mongo/db/s/sharding_ddl_coordinator_service.cpp
@@ -45,6 +45,7 @@
 #include "mongo/db/s/create_collection_coordinator.h"
 #include "mongo/db/s/drop_collection_coordinator.h"
 #include "mongo/db/s/drop_database_coordinator.h"
+#include "mongo/db/s/refine_collection_shard_key_coordinator.h"
 #include "mongo/db/s/rename_collection_coordinator.h"
 
 namespace mongo {
@@ -67,6 +68,10 @@ std::shared_ptr<ShardingDDLCoordinator> constructShardingDDLCoordinatorInstance(
         case DDLCoordinatorTypeEnum::kCreateCollection:
             return std::make_shared<CreateCollectionCoordinator>(service, std::move(initialState));
             break;
+        case DDLCoordinatorTypeEnum::kRefineCollectionShardKey:
+            return std::make_shared<RefineCollectionShardKeyCoordinator>(service,
+                                                                         std::move(initialState));
+            break;
         default:
             uasserted(ErrorCodes::BadValue,
                       str::stream()
diff --git a/src/mongo/db/s/shardsvr_refine_collection_shard_key_command.cpp b/src/mongo/db/s/shardsvr_refine_collection_shard_key_command.cpp
index f67a70f8312..e3e4d41d6a3 100644
--- a/src/mongo/db/s/shardsvr_refine_collection_shard_key_command.cpp
+++ b/src/mongo/db/s/shardsvr_refine_collection_shard_key_command.cpp
@@ -61,9 +61,16 @@ public:
         using InvocationBase::InvocationBase;
 
         void typedRun(OperationContext* opCtx) {
-            auto refineCoordinator = std::make_shared<RefineCollectionShardKeyCoordinator>(
-                opCtx, ns(), request().getNewShardKey());
-            refineCoordinator->run(opCtx).get(opCtx);
+            auto coordinatorDoc = RefineCollectionShardKeyCoordinatorDocument();
+            coordinatorDoc.setShardingDDLCoordinatorMetadata(
+                {{ns(), DDLCoordinatorTypeEnum::kRefineCollectionShardKey}});
+            coordinatorDoc.setRefineCollectionShardKeyRequest(
+                request().getRefineCollectionShardKeyRequest());
+
+            auto service = ShardingDDLCoordinatorService::getService(opCtx);
+            auto refineCoordinator = checked_pointer_cast<RefineCollectionShardKeyCoordinator>(
+                service->getOrCreateInstance(opCtx, coordinatorDoc.toBSON()));
+            refineCoordinator->getCompletionFuture().get(opCtx);
         }
 
         bool supportsWriteConcern() const override {
diff --git a/src/mongo/s/commands/cluster_refine_collection_shard_key_cmd.cpp b/src/mongo/s/commands/cluster_refine_collection_shard_key_cmd.cpp
index ecd8266b483..ce3fa693d57 100644
--- a/src/mongo/s/commands/cluster_refine_collection_shard_key_cmd.cpp
+++ b/src/mongo/s/commands/cluster_refine_collection_shard_key_cmd.cpp
@@ -66,8 +66,10 @@ public:
             }
 
             // Send it to the primary shard
-            ShardsvrRefineCollectionShardKey refineCollectionShardKeyCommand(nss,
-                                                                             request().getKey());
+            RefineCollectionShardKeyRequest requestParamObj;
+            requestParamObj.setNewShardKey(request().getKey());
+            ShardsvrRefineCollectionShardKey refineCollectionShardKeyCommand(nss);
+            refineCollectionShardKeyCommand.setRefineCollectionShardKeyRequest(requestParamObj);
 
             auto cmdResponse = executeCommandAgainstDatabasePrimary(
                 opCtx,
diff --git a/src/mongo/s/request_types/sharded_ddl_commands.idl b/src/mongo/s/request_types/sharded_ddl_commands.idl
index 81d2950189c..e736dedec09 100644
--- a/src/mongo/s/request_types/sharded_ddl_commands.idl
+++ b/src/mongo/s/request_types/sharded_ddl_commands.idl
@@ -125,6 +125,15 @@ structs:
                 description: "Latest version of the collection"
                 optional: false
 
+    RefineCollectionShardKeyRequest:
+        description: "Parameters sent by the router"
+        strict: false
+        fields:
+            newShardKey:
+                type: KeyPattern
+                description: "The index specification document to use as the new shard key."
+                optional: false
+
 commands:
 
     _shardsvrCreateCollection:
@@ -243,11 +252,8 @@ commands:
         namespace: concatenate_with_db
         api_version: ""
         strict: false
-        fields:
-            newShardKey:
-                type: KeyPattern
-                description: "The index specification document to use as the new shard key."
-                optional: false
+        chained_structs: 
+            RefineCollectionShardKeyRequest: RefineCollectionShardKeyRequest
 
     _configsvrCreateDatabase:
         description: "The internal createDatabase command on the config server"