SERVER-63504 Add shard worker for ecoc compaction

13 files changed, 643 insertions, 133 deletions

diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index 37055dd8a94..db92db44bac 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -85,6 +85,8 @@ env.Library(
     source=[
         'end_sessions_command.cpp',
         'fail_point_cmd.cpp',
+        'fle2_compact.cpp',
+        'fle2_compact.idl',
         'generic.cpp',
         'generic.idl',
         'hashcmd.cpp',
@@ -104,6 +106,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/bson/mutable/mutable_bson',
+        '$BUILD_DIR/mongo/crypto/encrypted_field_config',
         '$BUILD_DIR/mongo/db/auth/auth',
         '$BUILD_DIR/mongo/db/auth/authprivilege',
         '$BUILD_DIR/mongo/db/commands',
@@ -114,9 +117,11 @@ env.Library(
         '$BUILD_DIR/mongo/db/logical_session_id',
         '$BUILD_DIR/mongo/db/logical_session_id_helpers',
         '$BUILD_DIR/mongo/db/mongohasher',
+        '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/ops/write_ops_parsers',
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/idl/basic_types',
+        '$BUILD_DIR/mongo/idl/idl_parser',
         '$BUILD_DIR/mongo/idl/server_parameter',
         '$BUILD_DIR/mongo/rpc/message',
         'test_commands_enabled',
@@ -227,22 +232,6 @@ env.Library(
 )
 
 env.Library(
-    target="fle2_compact",
-    source=[
-        'fle2_compact.cpp',
-        'fle2_compact.idl',
-    ],
-    LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/crypto/encrypted_field_config',
-        '$BUILD_DIR/mongo/db/auth/auth',
-        '$BUILD_DIR/mongo/db/catalog/catalog_helpers',
-        '$BUILD_DIR/mongo/db/catalog/collection_catalog',
-        '$BUILD_DIR/mongo/db/catalog_raii',
-        '$BUILD_DIR/mongo/db/commands',
-    ]
-)
-
-env.Library(
     target="mongod_fsync",
     source=[
         "fsync.cpp",
@@ -545,6 +534,7 @@ env.Library(
         "dbhash.cpp",
         "driverHelpers.cpp",
         "internal_rename_if_options_and_indexes_match_cmd.cpp",
+        "fle2_compact_cmd.cpp",
         "map_reduce_command.cpp",
         "oplog_application_checks.cpp",
         "oplog_note.cpp",
@@ -614,7 +604,6 @@ env.Library(
         '$BUILD_DIR/mongo/idl/idl_parser',
         '$BUILD_DIR/mongo/util/net/ssl_manager',
         'core',
-        'fle2_compact',
         'kill_common',
         'map_reduce_agg',
         'mongod_fcv',
diff --git a/src/mongo/db/commands/fle2_compact.cpp b/src/mongo/db/commands/fle2_compact.cpp
index 7588cbc6735..24256c6bc54 100644
--- a/src/mongo/db/commands/fle2_compact.cpp
+++ b/src/mongo/db/commands/fle2_compact.cpp
@@ -27,70 +27,14 @@
  *    it in the license file.
  */
 
-#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kStorage
-
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/commands/fle2_compact.h"
 
 #include "mongo/crypto/encryption_fields_gen.h"
-#include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/collection_catalog.h"
-#include "mongo/db/catalog/rename_collection.h"
-#include "mongo/db/catalog_raii.h"
-#include "mongo/db/commands.h"
-#include "mongo/logv2/log.h"
 
 namespace mongo {
-namespace {
-
-class CompactStructuredEncryptionDataCmd final
-    : public TypedCommand<CompactStructuredEncryptionDataCmd> {
-public:
-    using Request = CompactStructuredEncryptionData;
-    using Reply = CompactStructuredEncryptionData::Reply;
-    using TC = TypedCommand<CompactStructuredEncryptionDataCmd>;
-
-    class Invocation final : public TC::InvocationBase {
-    public:
-        using TC::InvocationBase::InvocationBase;
-        using TC::InvocationBase::request;
-
-        Reply typedRun(OperationContext* opCtx) {
-            auto swCompactStats = compactEncryptedCompactionCollection(opCtx, request());
-            uassertStatusOK(swCompactStats);
-            return Reply(swCompactStats.getValue());
-        }
-
-    private:
-        bool supportsWriteConcern() const final {
-            return false;
-        }
-
-        void doCheckAuthorization(OperationContext* opCtx) const final {
-            auto* as = AuthorizationSession::get(opCtx->getClient());
-            uassert(ErrorCodes::Unauthorized,
-                    "Not authorized to compact structured encryption data",
-                    as->isAuthorizedForActionsOnResource(
-                        ResourcePattern::forExactNamespace(request().getNamespace()),
-                        ActionType::compactStructuredEncryptionData));
-        }
-
-        NamespaceString ns() const final {
-            return request().getNamespace();
-        }
-    };
-
-    typename TC::AllowedOnSecondary secondaryAllowed(ServiceContext*) const final {
-        return BasicCommand::AllowedOnSecondary::kNever;
-    }
-
-    bool adminOnly() const final {
-        return false;
-    }
-} compactStructuredEncryptionDataCmd;
-
-}  // namespace
 
 StatusWith<EncryptedStateCollectionsNamespaces>
 EncryptedStateCollectionsNamespaces::createFromDataCollection(const Collection& edc) {
@@ -129,60 +73,4 @@ EncryptedStateCollectionsNamespaces::createFromDataCollection(const Collection&
     return namespaces;
 }
 
-StatusWith<CompactStats> compactEncryptedCompactionCollection(
-    OperationContext* opCtx, const CompactStructuredEncryptionData& request) {
-    mongo::ECOCStats ecocStats(0, 0);
-    mongo::ECStats eccStats(0, 0, 0, 0);
-    mongo::ECStats escStats(0, 0, 0, 0);
-    const auto& edcNss = request.getNamespace();
-
-    LOGV2(6319900, "Compacting the encrypted compaction collection", "namespace"_attr = edcNss);
-
-    AutoGetDb autoDb(opCtx, edcNss.db(), MODE_IX);
-    if (!autoDb.getDb()) {
-        return Status(ErrorCodes::NamespaceNotFound, str::stream() << "database does not exist");
-    }
-
-    auto catalog = CollectionCatalog::get(opCtx);
-    Lock::CollectionLock edcLock(opCtx, edcNss, MODE_IS);
-
-    // Check the data collection exists and is not a view
-    auto edc = catalog->lookupCollectionByNamespace(opCtx, edcNss);
-    if (!edc) {
-        if (catalog->lookupView(opCtx, edcNss)) {
-            return Status(ErrorCodes::CommandNotSupportedOnView,
-                          "cannot compact structured encryption data on a view");
-        }
-        return Status(ErrorCodes::NamespaceNotFound, "collection does not exist");
-    }
-
-    uassert(6319903, "Feature flag FLE2 is not enabled", gFeatureFlagFLE2.isEnabledAndIgnoreFCV());
-
-    auto swNamespaces = EncryptedStateCollectionsNamespaces::createFromDataCollection(*edc.get());
-    if (!swNamespaces.isOK()) {
-        return swNamespaces.getStatus();
-    }
-    auto& namespaces = swNamespaces.getValue();
-
-    auto ecoc = catalog->lookupCollectionByNamespace(opCtx, namespaces.ecocNss);
-    auto ecocRename = catalog->lookupCollectionByNamespace(opCtx, namespaces.ecocRenameNss);
-
-    if (ecoc && !ecocRename) {
-        LOGV2(6319901,
-              "Renaming the encrypted compaction collection",
-              "ecocNss"_attr = namespaces.ecocNss,
-              "ecocRenameNss"_attr = namespaces.ecocRenameNss);
-        RenameCollectionOptions renameOpts;
-        validateAndRunRenameCollection(
-            opCtx, namespaces.ecocNss, namespaces.ecocRenameNss, renameOpts);
-        ecoc.reset();
-    }
-
-
-    LOGV2(6319902,
-          "Done compacting the encrypted compaction collection",
-          "namespace"_attr = request.getNamespace());
-    return CompactStats(ecocStats, eccStats, escStats);
-}
-
 }  // namespace mongo
diff --git a/src/mongo/db/commands/fle2_compact.h b/src/mongo/db/commands/fle2_compact.h
index 35308325169..b994e22a5a2 100644
--- a/src/mongo/db/commands/fle2_compact.h
+++ b/src/mongo/db/commands/fle2_compact.h
@@ -30,8 +30,8 @@
 #pragma once
 
 #include "mongo/base/status_with.h"
+#include "mongo/db/catalog/collection.h"
 #include "mongo/db/commands/fle2_compact_gen.h"
-#include "mongo/db/operation_context.h"
 
 namespace mongo {
 
@@ -46,7 +46,4 @@ struct EncryptedStateCollectionsNamespaces {
     NamespaceString ecocRenameNss;
 };
 
-StatusWith<CompactStats> compactEncryptedCompactionCollection(
-    OperationContext* opCtx, const CompactStructuredEncryptionData& request);
-
 }  // namespace mongo
diff --git a/src/mongo/db/commands/fle2_compact_cmd.cpp b/src/mongo/db/commands/fle2_compact_cmd.cpp
new file mode 100644
index 00000000000..ba3d8a7ed22
--- /dev/null
+++ b/src/mongo/db/commands/fle2_compact_cmd.cpp
@@ -0,0 +1,145 @@
+/**
+ *    Copyright (C) 2022-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::kStorage
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/commands/fle2_compact.h"
+
+#include "mongo/crypto/encryption_fields_gen.h"
+#include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/catalog/collection_catalog.h"
+#include "mongo/db/catalog/rename_collection.h"
+#include "mongo/db/catalog_raii.h"
+#include "mongo/db/commands.h"
+#include "mongo/logv2/log.h"
+
+namespace mongo {
+namespace {
+
+CompactStats compactEncryptedCompactionCollection(OperationContext* opCtx,
+                                                  const CompactStructuredEncryptionData& request) {
+    mongo::ECOCStats ecocStats(0, 0);
+    mongo::ECStats eccStats(0, 0, 0, 0);
+    mongo::ECStats escStats(0, 0, 0, 0);
+    const auto& edcNss = request.getNamespace();
+
+    LOGV2(6319900, "Compacting the encrypted compaction collection", "namespace"_attr = edcNss);
+
+    AutoGetDb autoDb(opCtx, edcNss.db(), MODE_IX);
+    uassert(ErrorCodes::NamespaceNotFound,
+            str::stream() << "Database '" << edcNss.db() << "' does not exist",
+            autoDb.getDb());
+
+    auto catalog = CollectionCatalog::get(opCtx);
+    Lock::CollectionLock edcLock(opCtx, edcNss, MODE_IS);
+
+    // Check the data collection exists and is not a view
+    auto edc = catalog->lookupCollectionByNamespace(opCtx, edcNss);
+    if (!edc) {
+        uassert(ErrorCodes::CommandNotSupportedOnView,
+                "Cannot compact structured encryption data on a view",
+                !catalog->lookupView(opCtx, edcNss));
+        uasserted(ErrorCodes::NamespaceNotFound,
+                  str::stream() << "Collection '" << edcNss << "' does not exist");
+    }
+
+    uassert(6319903, "Feature flag FLE2 is not enabled", gFeatureFlagFLE2.isEnabledAndIgnoreFCV());
+
+    auto namespaces =
+        uassertStatusOK(EncryptedStateCollectionsNamespaces::createFromDataCollection(*edc.get()));
+
+
+    auto ecoc = catalog->lookupCollectionByNamespace(opCtx, namespaces.ecocNss);
+    auto ecocRename = catalog->lookupCollectionByNamespace(opCtx, namespaces.ecocRenameNss);
+
+    if (ecoc && !ecocRename) {
+        LOGV2(6319901,
+              "Renaming the encrypted compaction collection",
+              "ecocNss"_attr = namespaces.ecocNss,
+              "ecocRenameNss"_attr = namespaces.ecocRenameNss);
+        RenameCollectionOptions renameOpts;
+        validateAndRunRenameCollection(
+            opCtx, namespaces.ecocNss, namespaces.ecocRenameNss, renameOpts);
+        ecoc.reset();
+    }
+
+    LOGV2(6319902,
+          "Done compacting the encrypted compaction collection",
+          "namespace"_attr = request.getNamespace());
+    return CompactStats(ecocStats, eccStats, escStats);
+}
+
+class CompactStructuredEncryptionDataCmd final
+    : public TypedCommand<CompactStructuredEncryptionDataCmd> {
+public:
+    using Request = CompactStructuredEncryptionData;
+    using Reply = CompactStructuredEncryptionData::Reply;
+    using TC = TypedCommand<CompactStructuredEncryptionDataCmd>;
+
+    class Invocation final : public TC::InvocationBase {
+    public:
+        using TC::InvocationBase::InvocationBase;
+        using TC::InvocationBase::request;
+
+        Reply typedRun(OperationContext* opCtx) {
+            return Reply(compactEncryptedCompactionCollection(opCtx, request()));
+        }
+
+    private:
+        bool supportsWriteConcern() const final {
+            return false;
+        }
+
+        void doCheckAuthorization(OperationContext* opCtx) const final {
+            auto* as = AuthorizationSession::get(opCtx->getClient());
+            uassert(ErrorCodes::Unauthorized,
+                    "Not authorized to compact structured encryption data",
+                    as->isAuthorizedForActionsOnResource(
+                        ResourcePattern::forExactNamespace(request().getNamespace()),
+                        ActionType::compactStructuredEncryptionData));
+        }
+
+        NamespaceString ns() const final {
+            return request().getNamespace();
+        }
+    };
+
+    typename TC::AllowedOnSecondary secondaryAllowed(ServiceContext*) const final {
+        return BasicCommand::AllowedOnSecondary::kNever;
+    }
+
+    bool adminOnly() const final {
+        return false;
+    }
+} compactStructuredEncryptionDataCmd;
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/namespace_string.cpp b/src/mongo/db/namespace_string.cpp
index ab4a511bc69..bd9d3a68d93 100644
--- a/src/mongo/db/namespace_string.cpp
+++ b/src/mongo/db/namespace_string.cpp
@@ -154,6 +154,9 @@ const NamespaceString NamespaceString::kConfigsvrCoordinatorsNamespace(
 const NamespaceString NamespaceString::kUserWritesCriticalSectionsNamespace(
     NamespaceString::kConfigDb, "user_writes_critical_sections");
 
+const NamespaceString NamespaceString::kCompactStructuredEncryptionCoordinatorNamespace(
+    NamespaceString::kConfigDb, "compact_structured_encryption_coordinator");
+
 bool NamespaceString::isListCollectionsCursorNS() const {
     return coll() == listCollectionsCursorCol;
 }
diff --git a/src/mongo/db/namespace_string.h b/src/mongo/db/namespace_string.h
index f511a6a3f04..6a9418f1775 100644
--- a/src/mongo/db/namespace_string.h
+++ b/src/mongo/db/namespace_string.h
@@ -213,6 +213,9 @@ public:
     // Namespace used during the recovery procedure for the config server.
     static const NamespaceString kConfigsvrRestoreNamespace;
 
+    // Namespace used for CompactParticipantCoordinator service.
+    static const NamespaceString kCompactStructuredEncryptionCoordinatorNamespace;
+
     /**
      * Constructs an empty NamespaceString.
      */
diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index 38cae8e2be5..96f60ade98d 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -320,6 +320,8 @@ env.Library(
         'collmod_coordinator.cpp',
         'collmod_coordinator_document.idl',
         'collmod_coordinator_pre60_compatible.cpp',
+        'compact_structured_encryption_data_coordinator.cpp',
+        'compact_structured_encryption_data_coordinator.idl',
         'config/set_cluster_parameter_coordinator.cpp',
         'config/set_cluster_parameter_coordinator_document.idl',
         'config/set_user_write_block_mode_coordinator.cpp',
@@ -399,6 +401,7 @@ env.Library(
         'shardsvr_collmod_command.cpp',
         'shardsvr_collmod_participant_command.cpp',
         'shardsvr_commit_reshard_collection_command.cpp',
+        'shardsvr_compact_structured_encryption_data_command.cpp',
         'shardsvr_create_collection_command.cpp',
         'shardsvr_create_collection_participant_command.cpp',
         'shardsvr_drop_collection_command.cpp',
@@ -428,6 +431,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/commands/core',
         '$BUILD_DIR/mongo/db/commands/create_command',
         '$BUILD_DIR/mongo/db/commands/mongod_fcv',
+        '$BUILD_DIR/mongo/db/commands/rename_collection_idl',
         '$BUILD_DIR/mongo/db/commands/server_status',
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
         '$BUILD_DIR/mongo/db/commands/txn_cmd_request',
diff --git a/src/mongo/db/s/compact_structured_encryption_data_coordinator.cpp b/src/mongo/db/s/compact_structured_encryption_data_coordinator.cpp
new file mode 100644
index 00000000000..ba8c48cbef1
--- /dev/null
+++ b/src/mongo/db/s/compact_structured_encryption_data_coordinator.cpp
@@ -0,0 +1,159 @@
+/**
+ *    Copyright (C) 2022-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/s/compact_structured_encryption_data_coordinator.h"
+
+#include "mongo/base/checked_cast.h"
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/db/catalog/collection_catalog.h"
+#include "mongo/db/commands/rename_collection_gen.h"
+#include "mongo/db/dbdirectclient.h"
+#include "mongo/db/drop_gen.h"
+#include "mongo/db/persistent_task_store.h"
+#include "mongo/logv2/log.h"
+
+namespace mongo {
+namespace {
+
+const auto kMajorityWriteConcern = BSON("writeConcern" << BSON("w"
+                                                               << "majority"));
+/**
+ * Issue a simple success/fail command such as renameCollection or drop
+ * using majority write concern.
+ */
+template <typename Request>
+void doRunCommand(OperationContext* opCtx, StringData dbname, const Request& request) {
+    DBDirectClient client(opCtx);
+    BSONObj cmd = request.toBSON(kMajorityWriteConcern);
+    auto reply = client.runCommand(OpMsgRequest::fromDBAndBody(dbname, cmd))->getCommandReply();
+    uassertStatusOK(getStatusFromCommandResult(reply));
+}
+
+void doRenameOperation(const CompactStructuredEncryptionDataState& state) {
+    const auto& ecocNss = state.getEcocNss();
+    auto opCtx = cc().makeOperationContext();
+    auto catalog = CollectionCatalog::get(opCtx.get());
+    auto uuid = catalog->lookupUUIDByNSS(opCtx.get(), ecocNss);
+    if (!uuid) {
+        // Nothing to rename.
+        LOGV2_DEBUG(6350492,
+                    5,
+                    "Skipping rename stage as there is no source collection",
+                    "ecocNss"_attr = ecocNss);
+        return;
+    } else if (uuid.get() != state.getEcocUuid()) {
+        // The generation of the collection to be compacted is different than the one which was
+        // requested.
+        LOGV2_DEBUG(6350491,
+                    5,
+                    "Skipping rename of mismatched collection uuid",
+                    "ecocNss"_attr = ecocNss,
+                    "uuid"_attr = uuid.get(),
+                    "expectedUUID"_attr = state.getEcocUuid());
+        return;
+    }
+
+    // Otherwise, perform the rename so long as the target namespace does not exist.
+    RenameCollectionCommand cmd(ecocNss, state.getEcocRenameNss());
+    cmd.setDropTarget(false);
+    cmd.setCollectionUUID(state.getEcocUuid());
+
+    doRunCommand(opCtx.get(), ecocNss.db(), cmd);
+}
+
+void doDropOperation(const CompactStructuredEncryptionDataState& state) {
+    auto ecocNss = state.getEcocRenameNss();
+    Drop cmd(ecocNss);
+    cmd.setCollectionUUID(state.getEcocUuid());
+    auto opCtx = cc().makeOperationContext();
+    doRunCommand(opCtx.get(), ecocNss.db(), cmd);
+}
+
+}  // namespace
+
+boost::optional<BSONObj> CompactStructuredEncryptionDataCoordinator::reportForCurrentOp(
+    MongoProcessInterface::CurrentOpConnectionsMode connMode,
+    MongoProcessInterface::CurrentOpSessionsMode sessionMode) noexcept {
+    BSONObjBuilder bob;
+    bob.append("type", "op");
+    bob.append("desc", "CompactStructuredEncryptionDataCoordinator");
+    bob.append("op", "command");
+    bob.append("nss", _doc.getId().getNss().ns());
+    bob.append("ecocNss", _doc.getEcocNss().ns());
+    bob.append("ecocUuid", _doc.getEcocUuid().toString());
+    bob.append("ecocRenameNss", _doc.getEcocRenameNss().ns());
+    bob.append("currentPhase", _doc.getPhase());
+    bob.append("active", true);
+    return bob.obj();
+}
+
+void CompactStructuredEncryptionDataCoordinator::_enterPhase(Phase newPhase) {
+    StateDoc doc(_doc);
+    doc.setPhase(newPhase);
+
+    LOGV2_DEBUG(6350490,
+                2,
+                "Transitioning phase for CompactStructuredEncryptionDataCoordinator",
+                "nss"_attr = _doc.getId().getNss().ns(),
+                "ecocNss"_attr = _doc.getEcocNss().ns(),
+                "ecocUuid"_attr = _doc.getEcocUuid(),
+                "ecocRenameNss"_attr = _doc.getEcocRenameNss().ns(),
+                "compactionTokens"_attr = _doc.getCompactionTokens(),
+                "oldPhase"_attr = CompactStructuredEncryptionDataPhase_serializer(_doc.getPhase()),
+                "newPhase"_attr = CompactStructuredEncryptionDataPhase_serializer(newPhase));
+
+    if (_doc.getPhase() == Phase::kRenameEcocForCompact) {
+        _doc = _insertStateDocument(std::move(doc));
+        return;
+    }
+    auto opCtx = cc().makeOperationContext();
+    _doc = _updateStateDocument(opCtx.get(), std::move(doc));
+}
+
+ExecutorFuture<void> CompactStructuredEncryptionDataCoordinator::_runImpl(
+    std::shared_ptr<executor::ScopedTaskExecutor> executor,
+    const CancellationToken& token) noexcept {
+    return ExecutorFuture<void>(**executor)
+        .then(_executePhase(Phase::kRenameEcocForCompact, doRenameOperation))
+        .then(_executePhase(Phase::kCompactStructuredEncryptionData,
+                            [this, anchor = shared_from_this()](const auto&) {
+                                // This will be implemented in a later phase.
+                                ECOCStats ecocStats(1, 2);
+                                ECStats eccStats(3, 4, 5, 6);
+                                ECStats escStats(7, 8, 9, 10);
+                                _response = CompactStats(ecocStats, eccStats, escStats);
+                            }))
+        .then(_executePhase(Phase::kDropTempCollection, doDropOperation));
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/compact_structured_encryption_data_coordinator.h b/src/mongo/db/s/compact_structured_encryption_data_coordinator.h
new file mode 100644
index 00000000000..cd5a2319250
--- /dev/null
+++ b/src/mongo/db/s/compact_structured_encryption_data_coordinator.h
@@ -0,0 +1,100 @@
+/**
+ *    Copyright (C) 2022-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 <boost/optional.hpp>
+#include <memory>
+
+#include "mongo/bson/bsonobj.h"
+#include "mongo/db/commands/fle2_compact_gen.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/s/compact_structured_encryption_data_coordinator_gen.h"
+#include "mongo/db/s/sharding_ddl_coordinator.h"
+
+namespace mongo {
+
+class CompactStructuredEncryptionDataCoordinator final : public ShardingDDLCoordinator {
+public:
+    static constexpr auto kStateContext = "CompactStructuredEncryptionDataState"_sd;
+    using StateDoc = CompactStructuredEncryptionDataState;
+    using Phase = CompactStructuredEncryptionDataPhaseEnum;
+
+    CompactStructuredEncryptionDataCoordinator(ShardingDDLCoordinatorService* service,
+                                               const BSONObj& doc)
+        : ShardingDDLCoordinator(service, doc), _doc(StateDoc::parse({kStateContext}, doc)) {}
+
+    boost::optional<BSONObj> reportForCurrentOp(
+        MongoProcessInterface::CurrentOpConnectionsMode connMode,
+        MongoProcessInterface::CurrentOpSessionsMode sessionMode) noexcept final;
+
+    CompactStructuredEncryptionDataCommandReply getResponse(OperationContext* opCtx) {
+        getCompletionFuture().get(opCtx);
+        invariant(_response);
+        return *_response;
+    }
+
+    const NamespaceString& nss() const {
+        return _doc.getId().getNss();
+    }
+
+    void checkIfOptionsConflict(const BSONObj& doc) const final {}
+
+private:
+    void _enterPhase(Phase newPhase);
+
+    template <typename Func>
+    auto _executePhase(const Phase& newPhase, Func&& func) {
+        return [=] {
+            const auto& currPhase = _doc.getPhase();
+            if (currPhase > newPhase) {
+                return;
+            }
+            if (currPhase < newPhase) {
+                _enterPhase(newPhase);
+            }
+
+            return func(_doc);
+        };
+    }
+
+private:
+    ShardingDDLCoordinatorMetadata const& metadata() const final {
+        return _doc.getShardingDDLCoordinatorMetadata();
+    }
+
+    ExecutorFuture<void> _runImpl(std::shared_ptr<executor::ScopedTaskExecutor> executor,
+                                  const CancellationToken& token) noexcept final;
+
+private:
+    StateDoc _doc;
+    boost::optional<CompactStructuredEncryptionDataCommandReply> _response;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/compact_structured_encryption_data_coordinator.idl b/src/mongo/db/s/compact_structured_encryption_data_coordinator.idl
new file mode 100644
index 00000000000..a22a235495b
--- /dev/null
+++ b/src/mongo/db/s/compact_structured_encryption_data_coordinator.idl
@@ -0,0 +1,68 @@
+# 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.
+#
+
+global:
+    cpp_namespace: "mongo"
+
+imports:
+    - "mongo/idl/basic_types.idl"
+    - "mongo/db/commands/fle2_compact.idl"
+    - "mongo/db/s/sharding_ddl_coordinator.idl"
+
+enums:
+    CompactStructuredEncryptionDataPhase:
+        description: "The current phase of the compactStructuredEncryptionData pipeline"
+        type: string
+        values:
+            kRenameEcocForCompact: "rename-ecoc-for-compact"
+            kCompactStructuredEncryptionData: "compact-structured-encryption-data"
+            kDropTempCollection: "drop-temp-collection"
+
+structs:
+    CompactStructuredEncryptionDataState:
+        description: "Represents the state of the compactStructuredEncryptionData pipeline"
+        strict: true
+        chained_structs:
+            ShardingDDLCoordinatorMetadata: ShardingDDLCoordinatorMetadata
+        fields:
+            phase:
+                description: "Current phase"
+                type: CompactStructuredEncryptionDataPhase
+                default: kRenameEcocForCompact
+            ecocNss:
+                description: "Collection containing compaction metadata to perform compact with"
+                type: namespacestring
+            ecocUuid:
+                description: "UUID of the collection identified by ecocNss"
+                type: uuid
+            ecocRenameNss:
+                description: "Temporary name to use while performing compaction"
+                type: namespacestring
+            compactionTokens:
+                description: "Compaction tokens for the compact operation"
+                type: object_owned
diff --git a/src/mongo/db/s/sharding_ddl_coordinator.idl b/src/mongo/db/s/sharding_ddl_coordinator.idl
index eb0b7e88590..3a918da262d 100644
--- a/src/mongo/db/s/sharding_ddl_coordinator.idl
+++ b/src/mongo/db/s/sharding_ddl_coordinator.idl
@@ -58,6 +58,7 @@ enums:
             kCollMod: "collMod_V2"
             kReshardCollection: "reshardCollection"
             kReshardCollectionNoResilient: "reshardCollectionNoResilient"
+            kCompactStructuredEncryptionData: "compactStructuredEncryptionData"
 
 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 60e288f2082..5fb5a6fb6ac 100644
--- a/src/mongo/db/s/sharding_ddl_coordinator_service.cpp
+++ b/src/mongo/db/s/sharding_ddl_coordinator_service.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/pipeline/expression_context.h"
 #include "mongo/db/s/collmod_coordinator.h"
 #include "mongo/db/s/collmod_coordinator_pre60_compatible.h"
+#include "mongo/db/s/compact_structured_encryption_data_coordinator.h"
 #include "mongo/db/s/create_collection_coordinator.h"
 #include "mongo/db/s/database_sharding_state.h"
 #include "mongo/db/s/drop_collection_coordinator.h"
@@ -106,6 +107,10 @@ std::shared_ptr<ShardingDDLCoordinator> constructShardingDDLCoordinatorInstance(
             return std::make_shared<ReshardCollectionCoordinator_NORESILIENT>(
                 service, std::move(initialState));
             break;
+        case DDLCoordinatorTypeEnum::kCompactStructuredEncryptionData:
+            return std::make_shared<CompactStructuredEncryptionDataCoordinator>(
+                service, std::move(initialState));
+            break;
         default:
             uasserted(ErrorCodes::BadValue,
                       str::stream()
diff --git a/src/mongo/db/s/shardsvr_compact_structured_encryption_data_command.cpp b/src/mongo/db/s/shardsvr_compact_structured_encryption_data_command.cpp
new file mode 100644
index 00000000000..f777a7fd309
--- /dev/null
+++ b/src/mongo/db/s/shardsvr_compact_structured_encryption_data_command.cpp
@@ -0,0 +1,148 @@
+/**
+ *    Copyright (C) 2022-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::kCommand
+
+#include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/catalog/collection_catalog.h"
+#include "mongo/db/catalog_raii.h"
+#include "mongo/db/commands.h"
+#include "mongo/db/commands/fle2_compact.h"
+#include "mongo/db/commands/fle2_compact_gen.h"
+#include "mongo/db/s/compact_structured_encryption_data_coordinator.h"
+#include "mongo/db/s/compact_structured_encryption_data_coordinator_gen.h"
+#include "mongo/logv2/log.h"
+#include "mongo/s/cluster_commands_helpers.h"
+
+namespace mongo {
+namespace {
+
+class _shardsvrCompactStructuredEncryptionDataCommand final
+    : public TypedCommand<_shardsvrCompactStructuredEncryptionDataCommand> {
+public:
+    using Request = CompactStructuredEncryptionData;
+    using Reply = typename Request::Reply;
+
+    _shardsvrCompactStructuredEncryptionDataCommand()
+        : TypedCommand("_shardsvrCompactStructuredEncryptionData"_sd) {}
+
+    bool skipApiVersionCheck() const final {
+        // Internal command (server to server).
+        return true;
+    }
+
+    std::string help() const final {
+        return "Internal command. Do not call directly. Compacts a ECOC collection.";
+    }
+
+    bool adminOnly() const final {
+        return false;
+    }
+
+    AllowedOnSecondary secondaryAllowed(ServiceContext*) const final {
+        return AllowedOnSecondary::kNever;
+    }
+
+    class Invocation final : public InvocationBase {
+    public:
+        using InvocationBase::InvocationBase;
+
+        Reply typedRun(OperationContext* opCtx) {
+            uassert(6350499,
+                    "Feature flag FLE2 is not enabled",
+                    gFeatureFlagFLE2.isEnabledAndIgnoreFCV());
+
+            auto compact = makeRequest(opCtx);
+            if (!compact) {
+                // Nothing to do.
+                return {{{0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}}};
+            }
+
+            return checked_pointer_cast<CompactStructuredEncryptionDataCoordinator>(
+                       ShardingDDLCoordinatorService::getService(opCtx)->getOrCreateInstance(
+                           opCtx, compact->toBSON()))
+                ->getResponse(opCtx);
+        }
+
+    private:
+        boost::optional<CompactStructuredEncryptionDataState> makeRequest(OperationContext* opCtx) {
+            const auto& req = request();
+            const auto& nss = req.getNamespace();
+
+            AutoGetCollection baseColl(opCtx, nss, MODE_IX);
+            uassert(ErrorCodes::NamespaceNotFound,
+                    str::stream() << "Unknown collection: " << nss,
+                    baseColl.getCollection());
+
+            auto namespaces =
+                uassertStatusOK(EncryptedStateCollectionsNamespaces::createFromDataCollection(
+                    *(baseColl.getCollection().get())));
+
+            AutoGetCollection ecocColl(opCtx, namespaces.ecocNss, MODE_IX);
+            if (!ecocColl.getCollection()) {
+                return boost::none;
+            }
+            auto ecocUUID = ecocColl->uuid();
+
+            // Append UUID to rename namespace.
+            NamespaceString renameNss(namespaces.ecocRenameNss.db(),
+                                      str::stream() << namespaces.ecocRenameNss << '.' << ecocUUID);
+
+            CompactStructuredEncryptionDataState compact;
+            compact.setShardingDDLCoordinatorMetadata(
+                {{nss, DDLCoordinatorTypeEnum::kCompactStructuredEncryptionData}});
+            compact.setEcocNss(namespaces.ecocNss);
+            compact.setEcocUuid(std::move(ecocUUID));
+            compact.setEcocRenameNss(renameNss);
+            compact.setCompactionTokens(req.getCompactionTokens().getOwned());
+
+            return compact;
+        }
+
+        NamespaceString ns() const override {
+            return request().getNamespace();
+        }
+
+        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));
+        }
+    };
+
+} shardsvrCompactStructuredEncryptionDataCommand;
+
+}  // namespace
+}  // namespace mongo