SERVER-78149: Fsync with lock option mongos command

(cherry picked from commit 972abe01272e894b8318dea03cf453e9d1e7918e)

4 files changed, 135 insertions, 78 deletions

diff --git a/jstests/sharding/cluster_fsync_basic.js b/jstests/sharding/cluster_fsync_basic.js
deleted file mode 100644
index 60886f20b55..00000000000
--- a/jstests/sharding/cluster_fsync_basic.js
+++ /dev/null
@@ -1,44 +0,0 @@
-/**
- * Testing the mongos fsyncUnlock functionality.
- * @tags: [
- *   requires_fsync,
- *   featureFlagClusterFsyncLock
- * ]
- */
-(function() {
-"use strict";
-
-const dbName = "test";
-const collName = "collTest";
-const ns = dbName + "." + collName;
-const st = new ShardingTest({
-    shards: 2,
-    mongos: 1,
-    mongosOptions: {setParameter: {featureFlagClusterFsyncLock: true}},
-    config: 1
-});
-const adminDB = st.s.getDB("admin");
-
-jsTest.log("Insert some data.");
-const coll = st.s0.getDB(dbName)[collName];
-let bulk = coll.initializeUnorderedBulkOp();
-for (let i = -50; i < 50; i++) {
-    bulk.insert({_id: i});
-}
-assert.commandWorked(bulk.execute());
-
-jsTest.log("Create a sharded collection with one chunk on each of the two shards.");
-st.ensurePrimaryShard(dbName, st.shard0.shardName);
-assert.commandWorked(st.s.adminCommand({enableSharding: dbName}));
-assert.commandWorked(st.s.adminCommand({shardCollection: ns, key: {_id: 1}}));
-assert.commandWorked(st.s.adminCommand({split: ns, middle: {_id: 0}}));
-assert.commandWorked(st.s.adminCommand(
-    {moveChunk: ns, find: {_id: 0}, to: st.shard1.shardName, _waitForDelete: true}));
-
-// TODO (SERVER-78149): Fsync Lock Command will be called before Fsync Unlock Command.
-let ret = assert.commandFailed(st.s.adminCommand({fsyncUnlock: 1}));
-const errmsg = "fsyncUnlock called when not locked";
-assert.eq(ret.errmsg.includes(errmsg), true);
-
-st.stop();
-}());
diff --git a/jstests/sharding/features3.js b/jstests/sharding/features3.js
index 9019c15ad19..a16626b168a 100644
--- a/jstests/sharding/features3.js
+++ b/jstests/sharding/features3.js
@@ -8,6 +8,7 @@
 // @tags: [
 //   expects_explicit_underscore_id_index,
 // ]
+import {FeatureFlagUtil} from "jstests/libs/feature_flag_util.js";
 (function() {
 'use strict';
 
@@ -151,8 +152,12 @@ x = dbForTest._adminCommand("fsync");
 assert(x.ok == 1, "fsync failed: " + tojson(x));
 
 // test fsync+lock on admin db
-x = dbForTest._adminCommand({"fsync": 1, lock: true});
-assert(!x.ok, "lock should fail: " + tojson(x));
+const featureFlagClusterFsyncLock =
+    FeatureFlagUtil.isEnabled(s.configRS.getPrimary().getDB('admin'), "ClusterFsyncLock");
+if (!featureFlagClusterFsyncLock) {
+    x = dbForTest._adminCommand({"fsync": 1, lock: true});
+    assert(!x.ok, "lock should fail: " + tojson(x));
+}
 
 s.stop();
 })();
diff --git a/jstests/sharding/fsync_lock_unlock.js b/jstests/sharding/fsync_lock_unlock.js
new file mode 100644
index 00000000000..af66af3f22e
--- /dev/null
+++ b/jstests/sharding/fsync_lock_unlock.js
@@ -0,0 +1,75 @@
+/**
+ * Verifies the fsync with lock+unlock command on mongos.
+ * @tags: [
+ *   requires_fsync,
+ *   featureFlagClusterFsyncLock,
+ *   uses_parallel_shell,
+ * ]
+ */
+(function() {
+"use strict";
+
+const dbName = "test";
+const collName = "collTest";
+const ns = dbName + "." + collName;
+const st = new ShardingTest({
+    shards: 2,
+    mongos: 1,
+    mongosOptions: {setParameter: {featureFlagClusterFsyncLock: true}},
+    config: 1
+});
+const adminDB = st.s.getDB('admin');
+
+function waitUntilOpCountIs(opFilter, num, st) {
+    assert.soon(() => {
+        let ops = st.s.getDB('admin')
+                      .aggregate([
+                          {$currentOp: {}},
+                          {$match: opFilter},
+                      ])
+                      .toArray();
+        if (ops.length != num) {
+            jsTest.log("Num operations: " + ops.length + ", expected: " + num);
+            jsTest.log(ops);
+            return false;
+        }
+        return true;
+    });
+}
+
+jsTest.log("Insert some data.");
+const coll = st.s0.getDB(dbName)[collName];
+assert.commandWorked(coll.insert({x: 1}));
+
+// unlock before lock should fail
+let ret = assert.commandFailed(st.s.adminCommand({fsyncUnlock: 1}));
+const errmsg = "fsyncUnlock called when not locked";
+assert.eq(ret.errmsg.includes(errmsg), true);
+
+// lock then unlock
+assert.commandWorked(st.s.adminCommand({fsync: 1, lock: true}));
+
+// Make sure writes are blocked. Spawn a write operation in a separate shell and make sure it
+// is blocked. There is really no way to do that currently, so just check that the write didn't
+// go through.
+let codeToRun = () => {
+    assert.commandWorked(db.getSiblingDB("test").getCollection("collTest").insert({x: 1}));
+};
+
+let writeOpHandle = startParallelShell(codeToRun, st.s.port);
+
+waitUntilOpCountIs({op: 'insert', ns: 'test.collTest', waitingForLock: true}, 1, st);
+
+// Make sure reads can still run even though there is a pending write and also that the write
+// didn't get through.
+assert.eq(1, coll.find({}).itcount());
+assert.commandWorked(st.s.adminCommand({fsyncUnlock: 1}));
+
+writeOpHandle();
+
+// ensure writers are allowed after the cluster is unlocked
+assert.commandWorked(coll.insert({x: 1}));
+assert.eq(coll.count(), 3);
+
+st.stop();
+}());
diff --git a/src/mongo/s/commands/cluster_fsync_cmd.cpp b/src/mongo/s/commands/cluster_fsync_cmd.cpp
index e82469acee3..d88ab19b382 100644
--- a/src/mongo/s/commands/cluster_fsync_cmd.cpp
+++ b/src/mongo/s/commands/cluster_fsync_cmd.cpp
@@ -31,14 +31,26 @@
 
 #include "mongo/client/read_preference.h"
 #include "mongo/client/remote_command_targeter.h"
+#include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/commands.h"
+#include "mongo/db/database_name.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/service_context.h"
+#include "mongo/db/shard_id.h"
+#include "mongo/logv2/log.h"
 #include "mongo/s/client/shard.h"
 #include "mongo/s/client/shard_registry.h"
+#include "mongo/s/cluster_commands_helpers.h"
 #include "mongo/s/grid.h"
+#include "mongo/s/sharding_feature_flags_gen.h"
+#include "mongo/util/assert_util.h"
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand
 
 namespace mongo {
 namespace {
 
+constexpr auto kRawFieldName = "raw"_sd;
 class FsyncCommand : public ErrmsgCommandDeprecated {
 public:
     FsyncCommand() : ErrmsgCommandDeprecated("fsync") {}
@@ -67,53 +79,62 @@ public:
         out->push_back(Privilege(ResourcePattern::forClusterResource(), actions));
     }
 
+    void unlockLockedShards(const std::set<ShardId> lockedShards,
+                            OperationContext* opCtx,
+                            const std::string& dbname) {
+        std::vector<AsyncRequestsSender::Request> requests;
+
+        for (const ShardId& shardId : lockedShards) {
+            requests.emplace_back(shardId, BSON("fsyncUnlock" << 1));
+        }
+        auto responses = gatherResponses(opCtx,
+                                         dbname,
+                                         ReadPreferenceSetting(ReadPreference::PrimaryOnly),
+                                         Shard::RetryPolicy::kIdempotent,
+                                         requests);
+        std::string errmsg;
+        BSONObjBuilder rawResult;
+        const auto response = appendRawResponses(opCtx, &errmsg, &rawResult, responses);
+        if (!response.responseOK) {
+            LOGV2_WARNING(781491, "Unlocking of shards failed: {error}", "error"_attr = errmsg);
+        }
+    }
+
     bool errmsgRun(OperationContext* opCtx,
                    const std::string& dbname,
                    const BSONObj& cmdObj,
                    std::string& errmsg,
                    BSONObjBuilder& result) override {
-        if (cmdObj["lock"].trueValue()) {
+
+        if (cmdObj["lock"].trueValue() &&
+            !feature_flags::gClusterFsyncLock.isEnabled(serverGlobalParams.featureCompatibility)) {
             errmsg = "can't do lock through mongos";
             return false;
         }
 
-        BSONObjBuilder sub;
-
-        bool ok = true;
+        auto shardResults = scatterGatherUnversionedTargetAllShards(
+            opCtx,
+            dbname,
+            applyReadWriteConcern(
+                opCtx, this, CommandHelpers::filterCommandRequestForPassthrough(cmdObj)),
+            ReadPreferenceSetting(ReadPreference::PrimaryOnly),
+            Shard::RetryPolicy::kIdempotent);
 
-        auto const shardRegistry = Grid::get(opCtx)->shardRegistry();
-        const auto shardIds = shardRegistry->getAllShardIdsNoReload();
-
-        for (const ShardId& shardId : shardIds) {
-            auto shardStatus = shardRegistry->getShard(opCtx, shardId);
-            if (!shardStatus.isOK()) {
-                continue;
-            }
-            const auto s = shardStatus.getValue();
-
-            auto response = uassertStatusOK(s->runCommandWithFixedRetryAttempts(
-                opCtx,
-                ReadPreferenceSetting{ReadPreference::PrimaryOnly},
-                "admin",
-                BSON("fsync" << 1),
-                Shard::RetryPolicy::kIdempotent));
-            uassertStatusOK(response.commandStatus);
-            BSONObj x = std::move(response.response);
-
-            sub.append(s->getId().toString(), x);
-
-            if (!x["ok"].trueValue()) {
-                ok = false;
-                errmsg = x["errmsg"].String();
-            }
-        }
+        BSONObjBuilder rawResult;
+        const auto response = appendRawResponses(opCtx, &errmsg, &rawResult, shardResults);
 
         // This field has had dummy value since MMAP went away. It is undocumented.
         // Maintaining it so as not to cause unnecessary user pain across upgrades.
         result.append("numFiles", 1);
-        result.append("all", sub.obj());
+        result.append("all", rawResult.obj()[kRawFieldName].Obj());
+        if (!response.responseOK) {
+            if (cmdObj["lock"].trueValue()) {
+                unlockLockedShards(response.shardsWithSuccessResponses, opCtx, dbname);
+            }
+            return false;
+        }
 
-        return ok;
+        return true;
     }
 
 } clusterFsyncCmd;