SERVER-65170 Integrate FLE compact into CompactStructuredEncryptionDataCoordinator

8 files changed, 172 insertions, 43 deletions

diff --git a/src/mongo/db/commands/fle2_compact.cpp b/src/mongo/db/commands/fle2_compact.cpp
index d73585d9b1d..9922b598a12 100644
--- a/src/mongo/db/commands/fle2_compact.cpp
+++ b/src/mongo/db/commands/fle2_compact.cpp
@@ -649,4 +649,14 @@ CompactStats processFLECompact(OperationContext* opCtx,
     return stats;
 }
 
+void validateCompactRequest(const CompactStructuredEncryptionData& request, const Collection& edc) {
+    uassert(6346807,
+            "Target namespace is not an encrypted collection",
+            edc.getCollectionOptions().encryptedFieldConfig);
+
+    // Validate the request contains a compaction token for each encrypted field
+    const auto& efc = edc.getCollectionOptions().encryptedFieldConfig.value();
+    CompactionHelpers::validateCompactionTokens(efc, request.getCompactionTokens());
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/commands/fle2_compact.h b/src/mongo/db/commands/fle2_compact.h
index cf316e1be72..d859f30a20d 100644
--- a/src/mongo/db/commands/fle2_compact.h
+++ b/src/mongo/db/commands/fle2_compact.h
@@ -48,6 +48,11 @@ struct EncryptedStateCollectionsNamespaces {
     NamespaceString ecocRenameNss;
 };
 
+/**
+ * Validate a compact request has the right encryption tokens.
+ */
+void validateCompactRequest(const CompactStructuredEncryptionData& request, const Collection& edc);
+
 CompactStats processFLECompact(OperationContext* opCtx,
                                const CompactStructuredEncryptionData& request,
                                GetTxnCallback getTxn,
diff --git a/src/mongo/db/commands/fle2_compact_cmd.cpp b/src/mongo/db/commands/fle2_compact_cmd.cpp
index d96e9b28dac..6df74b07b6e 100644
--- a/src/mongo/db/commands/fle2_compact_cmd.cpp
+++ b/src/mongo/db/commands/fle2_compact_cmd.cpp
@@ -77,13 +77,7 @@ CompactStats compactEncryptedCompactionCollection(OperationContext* opCtx,
             "FLE 2 is only supported when FCV supports 6.0",
             gFeatureFlagFLE2.isEnabled(serverGlobalParams.featureCompatibility));
 
-    uassert(6346807,
-            "Target namespace is not an encrypted collection",
-            edc->getCollectionOptions().encryptedFieldConfig);
-
-    // Validate the request contains a compaction token for each encrypted field
-    const auto& efc = edc->getCollectionOptions().encryptedFieldConfig.value();
-    CompactionHelpers::validateCompactionTokens(efc, request.getCompactionTokens());
+    validateCompactRequest(request, *edc.get());
 
     auto namespaces =
         uassertStatusOK(EncryptedStateCollectionsNamespaces::createFromDataCollection(*edc.get()));
diff --git a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
index bf5ae0a1469..2ecf3bce9f0 100644
--- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
+++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
@@ -425,6 +425,17 @@ public:
                             opCtx, DDLCoordinatorTypeEnum::kRefineCollectionShardKey);
                 }
 
+                // TODO SERVER-65077: Remove FCV check once 6.0 is released
+                if (actualVersion > requestedVersion &&
+                    !gFeatureFlagFLE2.isEnabledOnVersion(requestedVersion)) {
+                    // No more (recoverable) CompactStructuredEncryptionDataCoordinator will start
+                    // because we have already switched the FCV value to kDowngrading. Wait for the
+                    // ongoing CompactStructuredEncryptionDataCoordinator to finish.
+                    ShardingDDLCoordinatorService::getService(opCtx)
+                        ->waitForCoordinatorsOfGivenTypeToComplete(
+                            opCtx, DDLCoordinatorTypeEnum::kCompactStructuredEncryptionData);
+                }
+
                 // If we are only running phase-1, then we are done
                 return true;
             }
diff --git a/src/mongo/db/s/compact_structured_encryption_data_coordinator.cpp b/src/mongo/db/s/compact_structured_encryption_data_coordinator.cpp
index 47d168ea268..3fcfe10062a 100644
--- a/src/mongo/db/s/compact_structured_encryption_data_coordinator.cpp
+++ b/src/mongo/db/s/compact_structured_encryption_data_coordinator.cpp
@@ -36,6 +36,7 @@
 #include "mongo/base/checked_cast.h"
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/catalog/collection_catalog.h"
+#include "mongo/db/commands/fle2_compact.h"
 #include "mongo/db/commands/rename_collection_gen.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/drop_gen.h"
@@ -59,42 +60,122 @@ void doRunCommand(OperationContext* opCtx, StringData dbname, const Request& req
     uassertStatusOK(getStatusFromCommandResult(reply));
 }
 
-void doRenameOperation(const CompactStructuredEncryptionDataState& state) {
+void doRenameOperation(const CompactStructuredEncryptionDataState& state,
+                       bool* skipCompact,
+                       boost::optional<UUID>* newEcocRenameUuid) {
     const auto& ecocNss = state.getEcocNss();
+    const auto& ecocRenameNss = state.getEcocRenameNss();
     auto opCtx = cc().makeOperationContext();
     auto catalog = CollectionCatalog::get(opCtx.get());
-    auto uuid = catalog->lookupUUIDByNSS(opCtx.get(), ecocNss);
-    if (!uuid) {
+    auto ecocUuid = catalog->lookupUUIDByNSS(opCtx.get(), ecocNss);
+    auto ecocRenameUuid = catalog->lookupUUIDByNSS(opCtx.get(), ecocRenameNss);
+
+    auto hasEcocBefore = state.getEcocUuid().has_value();
+    auto hasEcocNow = !!ecocUuid;
+    auto hasEcocRenameBefore = state.getEcocRenameUuid().has_value();
+    auto hasEcocRenameNow = !!ecocRenameUuid;
+
+    *skipCompact = false;
+    *newEcocRenameUuid = state.getEcocRenameUuid();
+
+    if (hasEcocRenameBefore != hasEcocRenameNow) {
+        LOGV2_DEBUG(6517001,
+                    1,
+                    "Skipping compaction due to change in collection state",
+                    "ecocRenameNss"_attr = ecocRenameNss,
+                    "hasEcocRenameBefore"_attr = hasEcocRenameBefore,
+                    "hasEcocRenameNow"_attr = hasEcocRenameNow);
+        *skipCompact = true;
+        return;
+    } else if (hasEcocRenameNow) {
+        if (ecocRenameUuid.get() != state.getEcocRenameUuid().value()) {
+            LOGV2_DEBUG(6517002,
+                        1,
+                        "Skipping compaction due to mismatched collection uuid",
+                        "ecocRenameNss"_attr = ecocRenameNss,
+                        "uuid"_attr = ecocRenameUuid.get(),
+                        "expectedUUID"_attr = state.getEcocRenameUuid().value());
+            *skipCompact = true;
+        }
+        // the temp ecoc from a previous compact still exists, so skip rename
+        return;
+    }
+
+    if (!hasEcocNow) {
         // Nothing to rename.
         LOGV2_DEBUG(6350492,
-                    5,
+                    1,
                     "Skipping rename stage as there is no source collection",
                     "ecocNss"_attr = ecocNss);
+        *skipCompact = true;
+        return;
+    } else if (!hasEcocBefore) {
+        // mismatch of before/after state
+        LOGV2_DEBUG(6517003,
+                    1,
+                    "Skipping compaction due to change in collection state",
+                    "ecocNss"_attr = ecocNss);
+        *skipCompact = true;
         return;
-    } else if (uuid.get() != state.getEcocUuid()) {
+    } else if (ecocUuid.get() != state.getEcocUuid().value()) {
         // The generation of the collection to be compacted is different than the one which was
         // requested.
         LOGV2_DEBUG(6350491,
-                    5,
+                    1,
                     "Skipping rename of mismatched collection uuid",
                     "ecocNss"_attr = ecocNss,
-                    "uuid"_attr = uuid.get(),
-                    "expectedUUID"_attr = state.getEcocUuid());
+                    "uuid"_attr = ecocUuid.get(),
+                    "expectedUUID"_attr = state.getEcocUuid().value());
+        *skipCompact = true;
         return;
     }
 
+    LOGV2(6517004,
+          "Renaming the encrypted compaction collection",
+          "ecocNss"_attr = ecocNss,
+          "ecocUuid"_attr = ecocUuid.get(),
+          "ecocRenameNss"_attr = ecocRenameNss);
+
     // Otherwise, perform the rename so long as the target namespace does not exist.
-    RenameCollectionCommand cmd(ecocNss, state.getEcocRenameNss());
+    RenameCollectionCommand cmd(ecocNss, ecocRenameNss);
     cmd.setDropTarget(false);
-    cmd.setCollectionUUID(state.getEcocUuid());
+    cmd.setCollectionUUID(state.getEcocUuid().value());
 
     doRunCommand(opCtx.get(), ecocNss.db(), cmd);
+    *newEcocRenameUuid = state.getEcocUuid();
+}
+
+CompactStats doCompactOperation(const CompactStructuredEncryptionDataState& state) {
+    if (state.getSkipCompact()) {
+        LOGV2_DEBUG(6517005, 1, "Skipping compaction");
+        return CompactStats(ECOCStats(), ECStats(), ECStats());
+    }
+
+    EncryptedStateCollectionsNamespaces namespaces;
+    namespaces.edcNss = state.getId().getNss();
+    namespaces.eccNss = state.getEccNss();
+    namespaces.escNss = state.getEscNss();
+    namespaces.ecocNss = state.getEcocNss();
+    namespaces.ecocRenameNss = state.getEcocRenameNss();
+    auto opCtx = cc().makeOperationContext();
+    CompactStructuredEncryptionData request(namespaces.edcNss, state.getCompactionTokens());
+
+    return processFLECompact(opCtx.get(), request, &getTransactionWithRetriesForMongoS, namespaces);
 }
 
 void doDropOperation(const CompactStructuredEncryptionDataState& state) {
+    if (state.getSkipCompact()) {
+        LOGV2_DEBUG(6517006, 1, "Skipping drop of temporary encrypted compaction collection");
+        return;
+    }
+
+    uassert(6517007,
+            "Cannot drop temporary encrypted compaction collection due to missing collection UUID",
+            state.getEcocRenameUuid().has_value());
+
     auto ecocNss = state.getEcocRenameNss();
     Drop cmd(ecocNss);
-    cmd.setCollectionUUID(state.getEcocUuid());
+    cmd.setCollectionUUID(state.getEcocRenameUuid().value());
     auto opCtx = cc().makeOperationContext();
     doRunCommand(opCtx.get(), ecocNss.db(), cmd);
 }
@@ -109,9 +190,13 @@ boost::optional<BSONObj> CompactStructuredEncryptionDataCoordinator::reportForCu
     bob.append("desc", "CompactStructuredEncryptionDataCoordinator");
     bob.append("op", "command");
     bob.append("nss", _doc.getId().getNss().ns());
+    bob.append("escNss", _doc.getEscNss().ns());
+    bob.append("eccNss", _doc.getEccNss().ns());
     bob.append("ecocNss", _doc.getEcocNss().ns());
-    bob.append("ecocUuid", _doc.getEcocUuid().toString());
+    bob.append("ecocUuid", _doc.getEcocUuid() ? _doc.getEcocUuid().value().toString() : "none");
     bob.append("ecocRenameNss", _doc.getEcocRenameNss().ns());
+    bob.append("ecocRenameUuid",
+               _doc.getEcocRenameUuid() ? _doc.getEcocRenameUuid().value().toString() : "none");
     bob.append("currentPhase", _doc.getPhase());
     bob.append("active", true);
     return bob.obj();
@@ -125,17 +210,24 @@ void CompactStructuredEncryptionDataCoordinator::_enterPhase(Phase newPhase) {
                 2,
                 "Transitioning phase for CompactStructuredEncryptionDataCoordinator",
                 "nss"_attr = _doc.getId().getNss().ns(),
+                "escNss"_attr = _doc.getEscNss().ns(),
+                "eccNss"_attr = _doc.getEccNss().ns(),
                 "ecocNss"_attr = _doc.getEcocNss().ns(),
                 "ecocUuid"_attr = _doc.getEcocUuid(),
                 "ecocRenameNss"_attr = _doc.getEcocRenameNss().ns(),
+                "ecocRenameUuid"_attr = _doc.getEcocRenameUuid(),
+                "skipCompact"_attr = _doc.getSkipCompact(),
                 "compactionTokens"_attr = _doc.getCompactionTokens(),
                 "oldPhase"_attr = CompactStructuredEncryptionDataPhase_serializer(_doc.getPhase()),
                 "newPhase"_attr = CompactStructuredEncryptionDataPhase_serializer(newPhase));
 
     if (_doc.getPhase() == Phase::kRenameEcocForCompact) {
+        doc.setSkipCompact(_skipCompact);
+        doc.setEcocRenameUuid(_ecocRenameUuid);
         _doc = _insertStateDocument(std::move(doc));
         return;
     }
+
     auto opCtx = cc().makeOperationContext();
     _doc = _updateStateDocument(opCtx.get(), std::move(doc));
 }
@@ -144,23 +236,13 @@ 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::kRenameEcocForCompact,
+                            [this, anchor = shared_from_this()](const auto& state) {
+                                doRenameOperation(state, &_skipCompact, &_ecocRenameUuid);
+                            }))
         .then(_executePhase(Phase::kCompactStructuredEncryptionData,
-                            [this, anchor = shared_from_this()](const auto&) {
-                                // This will be implemented in a later phase.
-                                ECOCStats ecocStats;
-                                ECStats eccStats, escStats;
-                                ecocStats.setRead(1);
-                                ecocStats.setDeleted(2);
-                                eccStats.setRead(3);
-                                eccStats.setInserted(4);
-                                eccStats.setUpdated(5);
-                                eccStats.setDeleted(6);
-                                escStats.setRead(7);
-                                escStats.setInserted(8);
-                                escStats.setUpdated(9);
-                                escStats.setDeleted(10);
-                                _response = CompactStats(ecocStats, eccStats, escStats);
+                            [this, anchor = shared_from_this()](const auto& state) {
+                                _response = doCompactOperation(state);
                             }))
         .then(_executePhase(Phase::kDropTempCollection, doDropOperation));
 }
diff --git a/src/mongo/db/s/compact_structured_encryption_data_coordinator.h b/src/mongo/db/s/compact_structured_encryption_data_coordinator.h
index cd5a2319250..f6a2bc9d922 100644
--- a/src/mongo/db/s/compact_structured_encryption_data_coordinator.h
+++ b/src/mongo/db/s/compact_structured_encryption_data_coordinator.h
@@ -95,6 +95,8 @@ private:
 private:
     StateDoc _doc;
     boost::optional<CompactStructuredEncryptionDataCommandReply> _response;
+    bool _skipCompact{false};
+    boost::optional<UUID> _ecocRenameUuid;
 };
 
 }  // 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
index a22a235495b..445df053814 100644
--- a/src/mongo/db/s/compact_structured_encryption_data_coordinator.idl
+++ b/src/mongo/db/s/compact_structured_encryption_data_coordinator.idl
@@ -54,15 +54,30 @@ structs:
                 description: "Current phase"
                 type: CompactStructuredEncryptionDataPhase
                 default: kRenameEcocForCompact
+            skipCompact:
+                description: "Whether to skip the compaction operation"
+                type: bool
+                default: false
+            escNss:
+                description: "Collection containing insertions metadata"
+                type: namespacestring
+            eccNss:
+                description: "Collection containing deletions metadata"
+                type: namespacestring
             ecocNss:
                 description: "Collection containing compaction metadata to perform compact with"
                 type: namespacestring
             ecocUuid:
                 description: "UUID of the collection identified by ecocNss"
                 type: uuid
+                optional: true
             ecocRenameNss:
                 description: "Temporary name to use while performing compaction"
                 type: namespacestring
+            ecocRenameUuid:
+                description: "UUID of the collection identified by ecocRenameNss"
+                type: uuid
+                optional: true
             compactionTokens:
                 description: "Compaction tokens for the compact operation"
                 type: object_owned
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
index b29e42a741e..10dcaf4c41b 100644
--- a/src/mongo/db/s/shardsvr_compact_structured_encryption_data_command.cpp
+++ b/src/mongo/db/s/shardsvr_compact_structured_encryption_data_command.cpp
@@ -33,6 +33,7 @@
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/commands.h"
+#include "mongo/db/commands/feature_compatibility_version.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"
@@ -78,6 +79,7 @@ public:
             uassert(6350499,
                     "FLE 2 is only supported when FCV supports 6.0",
                     gFeatureFlagFLE2.isEnabled(serverGlobalParams.featureCompatibility));
+            FixedFCVRegion fixedFcvRegion(opCtx);
 
             auto compact = makeRequest(opCtx);
             if (!compact) {
@@ -101,26 +103,34 @@ public:
                     str::stream() << "Unknown collection: " << nss,
                     baseColl.getCollection());
 
+            validateCompactRequest(req, *(baseColl.getCollection().get()));
+
             auto namespaces =
                 uassertStatusOK(EncryptedStateCollectionsNamespaces::createFromDataCollection(
                     *(baseColl.getCollection().get())));
 
             AutoGetCollection ecocColl(opCtx, namespaces.ecocNss, MODE_IX);
-            if (!ecocColl.getCollection()) {
+            AutoGetCollection ecocTempColl(opCtx, namespaces.ecocRenameNss, MODE_IX);
+
+            if (!ecocColl.getCollection() && !ecocTempColl.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;
+
+            if (ecocColl.getCollection()) {
+                compact.setEcocUuid(ecocColl->uuid());
+            }
+            if (ecocTempColl.getCollection()) {
+                compact.setEcocRenameUuid(ecocTempColl->uuid());
+            }
+
             compact.setShardingDDLCoordinatorMetadata(
                 {{nss, DDLCoordinatorTypeEnum::kCompactStructuredEncryptionData}});
+            compact.setEscNss(namespaces.escNss);
+            compact.setEccNss(namespaces.eccNss);
             compact.setEcocNss(namespaces.ecocNss);
-            compact.setEcocUuid(std::move(ecocUUID));
-            compact.setEcocRenameNss(renameNss);
+            compact.setEcocRenameNss(namespaces.ecocRenameNss);
             compact.setCompactionTokens(req.getCompactionTokens().getOwned());
 
             return compact;