SERVER-62266: Serialize setClusterParameter and drop cluster parameter collection during FCV 6.0 downgrade

(cherry picked from commit ad30aeecc3c3afce0c02666cbb06778c0c65f25d)

9 files changed, 229 insertions, 62 deletions

diff --git a/jstests/noPassthrough/libs/user_write_blocking.js b/jstests/noPassthrough/libs/user_write_blocking.js
index faa9f474868..6dc912bce5e 100644
--- a/jstests/noPassthrough/libs/user_write_blocking.js
+++ b/jstests/noPassthrough/libs/user_write_blocking.js
@@ -66,7 +66,12 @@ const UserWriteBlockHelpers = (function() {
         asAdmin(fun) {
             const db = this.adminConn.getDB(jsTestName());
             const coll = db.test;
-            return fun({conn: this.adminConn, db: db, admin: db.getSiblingDB("admin"), coll: coll});
+            return fun({
+                conn: this.adminConn,
+                db: db,
+                admin: db.getSiblingDB("admin"),
+                coll: coll,
+            });
         }
 
         runInParallelShell(asAdmin, funString) {
@@ -96,21 +101,10 @@ const UserWriteBlockHelpers = (function() {
             assert.eq(expectedUserWriteBlockMode, status.repl.userWriteBlockMode);
         }
 
-        _hangTransition(targetConn, failpoint) {
-            let hangWhileSetingUserWriteBlockModeFailPoint =
-                configureFailPoint(targetConn, failpoint);
-
-            const awaitShell = startParallelShell(
-                funWithArgs((username, password) => {
-                    let admin = db.getSiblingDB("admin");
-                    admin.auth(username, password);
-                    assert.commandWorked(
-                        admin.runCommand({setUserWriteBlockMode: 1, global: true}));
-                }, bypassUser, password), this.conn.port);
-
-            hangWhileSetingUserWriteBlockModeFailPoint.wait();
-
-            return {waiter: awaitShell, failpoint: hangWhileSetingUserWriteBlockModeFailPoint};
+        _hangTransition(targetConn, failpoint, awaitShell) {
+            let hangFailPoint = configureFailPoint(targetConn, failpoint);
+            hangFailPoint.wait();
+            return {waiter: awaitShell, failpoint: hangFailPoint};
         }
 
         setFailPoint(failpointName) {
@@ -223,8 +217,8 @@ const UserWriteBlockHelpers = (function() {
 
     class ShardingFixture extends Fixture {
         constructor() {
-            const st = new ShardingTest(
-                {shards: 1, mongos: 1, config: 1, auth: "", other: {keyFile: keyfile}});
+            const st =
+                new ShardingTest({shards: 1, rs: {nodes: 3}, auth: "", other: {keyFile: keyfile}});
 
             super(st.s.port);
             this.st = st;
@@ -236,8 +230,19 @@ const UserWriteBlockHelpers = (function() {
                 backend, keyfile, () => backend.getDB('admin').serverStatus());
         }
 
-        hangTransition() {
-            return this._hangTransition(this.st.shard0, "hangInShardsvrSetUserWriteBlockMode");
+        hangTransition(command, failpoint) {
+            const awaitShell =
+                startParallelShell(funWithArgs((username, password, command) => {
+                                       let admin = db.getSiblingDB("admin");
+                                       admin.auth(username, password);
+                                       assert.commandWorked(admin.runCommand(command));
+                                   }, bypassUser, password, command), this.conn.port);
+            return this._hangTransition(this.st.shard0, failpoint, awaitShell);
+        }
+
+        restartConfigPrimary() {
+            jsTestLog('Restarting config primary');
+            this.st.restartConfigServer(this.st.configRS.getPrimary());
         }
 
         setFailPoint(failpointName) {
diff --git a/jstests/noPassthrough/set_cluster_parameter_fcv.js b/jstests/noPassthrough/set_cluster_parameter_fcv.js
new file mode 100644
index 00000000000..f78018cad44
--- /dev/null
+++ b/jstests/noPassthrough/set_cluster_parameter_fcv.js
@@ -0,0 +1,108 @@
+// Test setClusterParameter command against FCV.
+//
+// @tags: [
+//   creates_and_authenticates_user,
+//   requires_auth,
+//   requires_fcv_60,
+//   requires_non_retryable_commands,
+//   requires_persistence,
+//   requires_replication,
+// ]
+
+load("jstests/noPassthrough/libs/user_write_blocking.js");
+
+(function() {
+'use strict';
+
+const {
+    WriteBlockState,
+    ShardingFixture,
+    ReplicaFixture,
+    bypassUser,
+    noBypassUser,
+    password,
+    keyfile
+} = UserWriteBlockHelpers;
+
+function mapToClusterParamsColl(db) {
+    return db.getSiblingDB('config').clusterParameters;
+}
+
+function runTest(fixture) {
+    // When the cluster is started at FCV 6.0, it is possible to run setClusterParameter.
+    fixture.asAdmin(({admin}) => assert.commandWorked(admin.runCommand(
+                        {setClusterParameter: {testIntClusterParameter: {intData: 102}}})));
+
+    // Check that the config.clusterParameters collection has been created with a document for the
+    // parameter.
+    fixture.asAdmin(
+        ({db}) => assert.eq(1, mapToClusterParamsColl(db).count({_id: "testIntClusterParameter"})));
+
+    // When the cluster is at FCV 6.0 without an ongoing setClusterParameter operation in progress,
+    // it should be possible to downgrade the cluster.
+    fixture.asAdmin(({admin}) => assert.commandWorked(
+                        admin.runCommand({setFeatureCompatibilityVersion: "5.0"})));
+
+    // After downgrade, config.clusterParameters should not exist.
+    fixture.asAdmin(({db}) => assert.isnull(mapToClusterParamsColl(db).exists()));
+    fixture.asAdmin(
+        ({db}) => assert.eq(
+            0, mapToClusterParamsColl(db).count({_id: "testIntClusterParameter", intData: 102})));
+
+    // While the cluster is downgraded, it should not be possible to run setClusterParameter.
+    fixture.asAdmin(({admin}) => assert.commandFailed(admin.runCommand(
+                        {setClusterParameter: {testIntClusterParameter: {intData: 102}}})));
+
+    // Upgrading the cluster back to 6.0 should permit setClusterParameter to work again.
+    fixture.asAdmin(({admin}) => assert.commandWorked(
+                        admin.runCommand({setFeatureCompatibilityVersion: "6.0"})));
+    fixture.asAdmin(({admin}) => assert.commandWorked(admin.runCommand(
+                        {setClusterParameter: {testIntClusterParameter: {intData: 103}}})));
+
+    // Set a failpoint to make setClusterParameter hang on a sharded cluster. FCV downgrade should
+    // fail while setClusterParameter is in progress.
+    if (fixture.hangTransition) {
+        let hangWaiter =
+            fixture.hangTransition({setClusterParameter: {testIntClusterParameter: {intData: 105}}},
+                                   'hangInShardsvrSetClusterParameter');
+
+        fixture.asAdmin(({admin}) => assert.commandFailedWithCode(
+                            admin.runCommand({setFeatureCompatibilityVersion: "5.0"}),
+                            ErrorCodes.CannotDowngrade));
+
+        // Restart the config server primary and verify that FCV downgrade still fails.
+        fixture.restartConfigPrimary();
+        fixture.asAdmin(({admin}) => assert.commandFailedWithCode(
+                            admin.runCommand({setFeatureCompatibilityVersion: "5.0"}),
+                            ErrorCodes.CannotDowngrade));
+
+        // Turn off the failpoint and wait for the hung setClusterParameter operation to drain.
+        hangWaiter.failpoint.off();
+        hangWaiter.waiter();
+
+        // Verify that the updated value was successfully updated and is visible despite the restart
+        // and failed FCV downgrade attempts.
+        fixture.asAdmin(({admin}) => assert.eq(
+                            105,
+                            admin.runCommand({getClusterParameter: "testIntClusterParameter"})
+                                .clusterParameters[0]
+                                .intData));
+
+        // Verify that FCV downgrade succeeds after the setClusterParameter operation has drained.
+        fixture.asAdmin(({admin}) => assert.commandWorked(
+                            admin.runCommand({setFeatureCompatibilityVersion: "5.0"})));
+    }
+}
+
+{
+    const rst = new ReplicaFixture();
+    runTest(rst);
+    rst.stop();
+}
+
+{
+    const st = new ShardingFixture();
+    runTest(st);
+    st.stop();
+}
+}());
diff --git a/jstests/noPassthrough/user_write_block_fcv.js b/jstests/noPassthrough/user_write_block_fcv.js
index 5193ddb17d1..f59a1544659 100644
--- a/jstests/noPassthrough/user_write_block_fcv.js
+++ b/jstests/noPassthrough/user_write_block_fcv.js
@@ -59,11 +59,20 @@ function runTest(fixture) {
     // This does not need to be tested on replicasets. FCV locks are uninterruptable. Waiting on a
     // failpoint on a replicaset primary would prevent FCV from ever completing.
     if (fixture.hangTransition) {
-        let hangWaiter = fixture.hangTransition();
+        let hangWaiter = fixture.hangTransition({setUserWriteBlockMode: 1, global: true},
+                                                "hangInShardsvrSetUserWriteBlockMode");
 
         fixture.asAdmin(({admin}) => assert.commandFailedWithCode(
                             admin.runCommand({setFeatureCompatibilityVersion: "5.0"}),
                             ErrorCodes.CannotDowngrade));
+
+        // Restart the config server primary while in the middle of activating write block mode. FCV
+        // downgrade should still fail.
+        fixture.restartConfigPrimary();
+        fixture.asAdmin(({admin}) => assert.commandFailedWithCode(
+                            admin.runCommand({setFeatureCompatibilityVersion: "5.0"}),
+                            ErrorCodes.CannotDowngrade));
+
         hangWaiter.failpoint.off();
         hangWaiter.waiter();
     }
diff --git a/src/mongo/db/commands/set_cluster_parameter_command.cpp b/src/mongo/db/commands/set_cluster_parameter_command.cpp
index 31df926ce68..005f17d0c06 100644
--- a/src/mongo/db/commands/set_cluster_parameter_command.cpp
+++ b/src/mongo/db/commands/set_cluster_parameter_command.cpp
@@ -34,6 +34,7 @@
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/cluster_server_parameter_cmds_gen.h"
+#include "mongo/db/commands/feature_compatibility_version.h"
 #include "mongo/db/commands/set_cluster_parameter_invocation.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/idl/cluster_server_parameter_gen.h"
@@ -68,15 +69,16 @@ public:
         using InvocationBase::InvocationBase;
 
         void typedRun(OperationContext* opCtx) {
+            uassert(ErrorCodes::ErrorCodes::NotImplemented,
+                    "setClusterParameter can only run on mongos in sharded clusters",
+                    (serverGlobalParams.clusterRole == ClusterRole::None));
+
+            FixedFCVRegion fcvRegion(opCtx);
             uassert(
                 ErrorCodes::IllegalOperation,
                 "Cannot set cluster parameter, gFeatureFlagClusterWideConfig is not enabled",
                 gFeatureFlagClusterWideConfig.isEnabled(serverGlobalParams.featureCompatibility));
 
-            uassert(ErrorCodes::ErrorCodes::NotImplemented,
-                    "setClusterParameter can only run on mongos in sharded clusters",
-                    (serverGlobalParams.clusterRole == ClusterRole::None));
-
             // TODO SERVER-65249: This will eventually be made specific to the parameter being set
             // so that some parameters will be able to use setClusterParameter even on standalones.
             uassert(ErrorCodes::IllegalOperation,
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 2ecf3bce9f0..a56f54ebbcb 100644
--- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
+++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
@@ -82,6 +82,7 @@
 #include "mongo/db/session_txn_record_gen.h"
 #include "mongo/db/timeseries/timeseries_index_schema_conversion_functions.h"
 #include "mongo/db/vector_clock.h"
+#include "mongo/idl/cluster_server_parameter_gen.h"
 #include "mongo/logv2/log.h"
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/s/pm2423_feature_flags_gen.h"
@@ -370,7 +371,7 @@ public:
                             ErrorCodes::CannotDowngrade,
                             "Cannot downgrade while user write blocking is being changed",
                             ConfigsvrCoordinatorService::getService(opCtx)
-                                ->isAnyCoordinatorOfGivenTypeRunning(
+                                ->areAllCoordinatorsOfTypeFinished(
                                     opCtx, ConfigsvrCoordinatorTypeEnum::kSetUserWriteBlockMode));
                     }
 
@@ -384,6 +385,30 @@ public:
                             !isBlockingUserWrites);
                 }
 
+                // TODO (SERVER-65572): Remove setClusterParameter serialization and collection
+                // drop after this is backported to 6.0.
+                if (!gFeatureFlagClusterWideConfig.isEnabledOnVersion(requestedVersion)) {
+                    if (serverGlobalParams.clusterRole == ClusterRole::ConfigServer) {
+                        uassert(ErrorCodes::CannotDowngrade,
+                                "Cannot downgrade while cluster server parameter is being set",
+                                ConfigsvrCoordinatorService::getService(opCtx)
+                                    ->areAllCoordinatorsOfTypeFinished(
+                                        opCtx, ConfigsvrCoordinatorTypeEnum::kSetClusterParameter));
+                    }
+
+                    DropReply dropReply;
+                    const auto dropStatus = dropCollection(
+                        opCtx,
+                        NamespaceString::kClusterParametersNamespace,
+                        &dropReply,
+                        DropCollectionSystemCollectionMode::kAllowSystemCollectionDrops);
+                    uassert(
+                        dropStatus.code(),
+                        str::stream() << "Failed to drop the cluster server parameters collection"
+                                      << causedBy(dropStatus.reason()),
+                        dropStatus.isOK() || dropStatus.code() == ErrorCodes::NamespaceNotFound);
+                }
+
                 FeatureCompatibilityVersion::updateFeatureCompatibilityVersionDocument(
                     opCtx,
                     actualVersion,
diff --git a/src/mongo/db/s/config/configsvr_coordinator_service.cpp b/src/mongo/db/s/config/configsvr_coordinator_service.cpp
index b3ba449ac0f..d7ff2d006e0 100644
--- a/src/mongo/db/s/config/configsvr_coordinator_service.cpp
+++ b/src/mongo/db/s/config/configsvr_coordinator_service.cpp
@@ -93,10 +93,11 @@ ConfigsvrCoordinatorService::constructInstance(BSONObj initialState) {
     }
 }
 
-bool ConfigsvrCoordinatorService::isAnyCoordinatorOfGivenTypeRunning(
+bool ConfigsvrCoordinatorService::areAllCoordinatorsOfTypeFinished(
     OperationContext* opCtx, ConfigsvrCoordinatorTypeEnum coordinatorType) {
 
-    const auto instances = getAllInstances(opCtx);
+    // First, check if all in-memory ConfigsvrCoordinators are finished.
+    const auto& instances = getAllInstances(opCtx);
     for (const auto& instance : instances) {
         auto typedInstance = checked_pointer_cast<ConfigsvrCoordinator>(instance);
         if (typedInstance->coordinatorType() == coordinatorType) {
@@ -106,7 +107,14 @@ bool ConfigsvrCoordinatorService::isAnyCoordinatorOfGivenTypeRunning(
         }
     }
 
-    return true;
+    // If the POS has just been rebuilt on a newly-elected primary, there is a chance that the
+    // the coordinator instance does not exist yet. Query the state document namespace for any
+    // documents that will be built into instances.
+    DBDirectClient client(opCtx);
+    FindCommandRequest findStateDocs{NamespaceString::kConfigsvrCoordinatorsNamespace};
+    findStateDocs.setFilter(BSON("_id" << BSON("coordinatorType" << coordinatorType)));
+
+    return !client.find(std::move(findStateDocs))->more();
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/config/configsvr_coordinator_service.h b/src/mongo/db/s/config/configsvr_coordinator_service.h
index 7b8f30a3ed3..dfc2ccddb38 100644
--- a/src/mongo/db/s/config/configsvr_coordinator_service.h
+++ b/src/mongo/db/s/config/configsvr_coordinator_service.h
@@ -68,8 +68,8 @@ public:
 
     std::shared_ptr<Instance> constructInstance(BSONObj initialState) override;
 
-    bool isAnyCoordinatorOfGivenTypeRunning(OperationContext* opCtx,
-                                            ConfigsvrCoordinatorTypeEnum coordinatorType);
+    bool areAllCoordinatorsOfTypeFinished(OperationContext* opCtx,
+                                          ConfigsvrCoordinatorTypeEnum coordinatorType);
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp b/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp
index 6a0109194f6..da76faa6593 100644
--- a/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp
+++ b/src/mongo/db/s/config/configsvr_set_cluster_parameter_command.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/commands.h"
+#include "mongo/db/commands/feature_compatibility_version.h"
 #include "mongo/db/commands/set_cluster_parameter_invocation.h"
 #include "mongo/db/s/config/configsvr_coordinator_service.h"
 #include "mongo/db/s/config/set_cluster_parameter_coordinator.h"
@@ -58,46 +59,51 @@ public:
                     str::stream() << Request::kCommandName << " can only be run on config servers",
                     serverGlobalParams.clusterRole == ClusterRole::ConfigServer);
 
-            uassert(
-                ErrorCodes::IllegalOperation,
-                "featureFlagClusterWideConfig not enabled",
-                gFeatureFlagClusterWideConfig.isEnabled(serverGlobalParams.featureCompatibility));
+            const auto coordinatorCompletionFuture = [&]() -> SharedSemiFuture<void> {
+                FixedFCVRegion fcvRegion(opCtx);
+                uassert(ErrorCodes::IllegalOperation,
+                        "featureFlagClusterWideConfig not enabled",
+                        gFeatureFlagClusterWideConfig.isEnabled(
+                            serverGlobalParams.featureCompatibility));
 
-            // Validate parameter before creating coordinator.
-            {
-                BSONObj cmdParamObj = request().getCommandParameter();
-                BSONElement commandElement = cmdParamObj.firstElement();
-                StringData parameterName = commandElement.fieldName();
-                std::unique_ptr<ServerParameterService> sps =
-                    std::make_unique<ClusterParameterService>();
-                const ServerParameter* serverParameter = sps->getIfExists(parameterName);
+                // Validate parameter before creating coordinator.
+                {
+                    BSONObj cmdParamObj = request().getCommandParameter();
+                    BSONElement commandElement = cmdParamObj.firstElement();
+                    StringData parameterName = commandElement.fieldName();
+                    std::unique_ptr<ServerParameterService> sps =
+                        std::make_unique<ClusterParameterService>();
+                    const ServerParameter* serverParameter = sps->getIfExists(parameterName);
 
-                uassert(ErrorCodes::IllegalOperation,
-                        str::stream() << "Unknown Cluster Parameter " << parameterName,
-                        serverParameter != nullptr);
+                    uassert(ErrorCodes::IllegalOperation,
+                            str::stream() << "Unknown Cluster Parameter " << parameterName,
+                            serverParameter != nullptr);
 
-                uassert(ErrorCodes::IllegalOperation,
-                        "Cluster parameter value must be an object",
-                        BSONType::Object == commandElement.type());
+                    uassert(ErrorCodes::IllegalOperation,
+                            "Cluster parameter value must be an object",
+                            BSONType::Object == commandElement.type());
+
+                    BSONObjBuilder clusterParamBuilder;
+                    clusterParamBuilder << "_id" << parameterName;
+                    clusterParamBuilder.appendElements(commandElement.Obj());
 
-                BSONObjBuilder clusterParamBuilder;
-                clusterParamBuilder << "_id" << parameterName;
-                clusterParamBuilder.appendElements(commandElement.Obj());
+                    BSONObj clusterParam = clusterParamBuilder.obj();
 
-                BSONObj clusterParam = clusterParamBuilder.obj();
+                    uassertStatusOK(serverParameter->validate(clusterParam));
+                }
 
-                uassertStatusOK(serverParameter->validate(clusterParam));
-            }
+                SetClusterParameterCoordinatorDocument coordinatorDoc;
+                coordinatorDoc.setConfigsvrCoordinatorMetadata(
+                    {ConfigsvrCoordinatorTypeEnum::kSetClusterParameter});
+                coordinatorDoc.setParameter(request().getCommandParameter());
 
-            SetClusterParameterCoordinatorDocument coordinatorDoc;
-            coordinatorDoc.setConfigsvrCoordinatorMetadata(
-                {ConfigsvrCoordinatorTypeEnum::kSetClusterParameter});
-            coordinatorDoc.setParameter(request().getCommandParameter());
+                const auto service = ConfigsvrCoordinatorService::getService(opCtx);
+                const auto instance = service->getOrCreateService(opCtx, coordinatorDoc.toBSON());
 
-            const auto service = ConfigsvrCoordinatorService::getService(opCtx);
-            const auto instance = service->getOrCreateService(opCtx, coordinatorDoc.toBSON());
+                return instance->getCompletionFuture();
+            }();
 
-            instance->getCompletionFuture().get(opCtx);
+            coordinatorCompletionFuture.get(opCtx);
         }
 
     private:
diff --git a/src/mongo/db/s/shardsvr_set_cluster_parameter_command.cpp b/src/mongo/db/s/shardsvr_set_cluster_parameter_command.cpp
index c84ad1dd27e..08ddd4d5a01 100644
--- a/src/mongo/db/s/shardsvr_set_cluster_parameter_command.cpp
+++ b/src/mongo/db/s/shardsvr_set_cluster_parameter_command.cpp
@@ -45,6 +45,8 @@
 namespace mongo {
 namespace {
 
+MONGO_FAIL_POINT_DEFINE(hangInShardsvrSetClusterParameter);
+
 const WriteConcernOptions kLocalWriteConcern{
     1, WriteConcernOptions::SyncMode::UNSET, WriteConcernOptions::kNoTimeout};
 
@@ -64,6 +66,8 @@ public:
             CommandHelpers::uassertCommandRunWithMajority(Request::kCommandName,
                                                           opCtx->getWriteConcern());
 
+            hangInShardsvrSetClusterParameter.pauseWhileSet();
+
             SetClusterParameter setClusterParameterRequest(request().getCommandParameter());
             setClusterParameterRequest.setDbName(NamespaceString::kAdminDb);
             std::unique_ptr<ServerParameterService> parameterService =