SERVER-43123 SERVER-43124 Use ReadWriteConcernDefaults for incoming mongod requests

20 files changed, 298 insertions, 116 deletions

diff --git a/src/mongo/db/commands.cpp b/src/mongo/db/commands.cpp
index e78d3c19773..fd6c28d145c 100644
--- a/src/mongo/db/commands.cpp
+++ b/src/mongo/db/commands.cpp
@@ -52,6 +52,7 @@
 #include "mongo/db/curop.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/read_write_concern_defaults.h"
 #include "mongo/rpc/factory.h"
 #include "mongo/rpc/metadata/client_metadata_ismaster.h"
 #include "mongo/rpc/op_msg_rpc_impls.h"
@@ -628,7 +629,7 @@ private:
         return _command->supportsWriteConcern(cmdObj());
     }
 
-    bool supportsReadConcern(repl::ReadConcernLevel level) const override {
+    ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const override {
         return _command->supportsReadConcern(_dbName, cmdObj(), level);
     }
 
diff --git a/src/mongo/db/commands.h b/src/mongo/db/commands.h
index 6002af5e03e..6f174a40dfa 100644
--- a/src/mongo/db/commands.h
+++ b/src/mongo/db/commands.h
@@ -432,6 +432,46 @@ private:
 };
 
 /**
+ * The result of checking an invocation's support for readConcern.  There are two parts:
+ * - Whether or not the invocation supports the given readConcern.
+ * - Whether or not the invocation permits having the default readConcern applied to it.
+ */
+struct ReadConcernSupportResult {
+    /**
+     * Whether this command invocation supports the requested readConcern level. This only
+     * applies when running outside transactions because all commands that are allowed to run
+     * in a transaction must support all the read concerns that can be used in a transaction.
+     */
+    enum class ReadConcern { kSupported, kNotSupported } readConcern;
+
+    /**
+     * Whether this command invocation supports applying the default readConcern to it.
+     */
+    enum class DefaultReadConcern { kPermitted, kNotPermitted } defaultReadConcern;
+
+    /**
+     * Construct with either the enum value or a bool, where true indicates
+     * ReadConcern::kSupported or DefaultReadConcern::kPermitted (as appropriate).
+     */
+    ReadConcernSupportResult(ReadConcern supported, DefaultReadConcern defaultPermitted)
+        : readConcern(supported), defaultReadConcern(defaultPermitted) {}
+
+    ReadConcernSupportResult(bool supported, DefaultReadConcern defaultPermitted)
+        : readConcern(supported ? ReadConcern::kSupported : ReadConcern::kNotSupported),
+          defaultReadConcern(defaultPermitted) {}
+
+    ReadConcernSupportResult(ReadConcern supported, bool defaultPermitted)
+        : readConcern(supported),
+          defaultReadConcern(defaultPermitted ? DefaultReadConcern::kPermitted
+                                              : DefaultReadConcern::kNotPermitted) {}
+
+    ReadConcernSupportResult(bool supported, bool defaultPermitted)
+        : readConcern(supported ? ReadConcern::kSupported : ReadConcern::kNotSupported),
+          defaultReadConcern(defaultPermitted ? DefaultReadConcern::kPermitted
+                                              : DefaultReadConcern::kNotPermitted) {}
+};
+
+/**
  * Represents a single invocation of a given command.
  */
 class CommandInvocation {
@@ -472,12 +512,11 @@ public:
     virtual bool supportsWriteConcern() const = 0;
 
     /**
-     * Returns true if this command invocation supports the given readConcern level. This only
-     * applies when running outside transactions because all commands that are allowed to run in a
-     * transaction must support all the read concerns that can be used in a transaction.
+     * Returns this invocation's support for readConcern.
      */
-    virtual bool supportsReadConcern(repl::ReadConcernLevel level) const {
-        return level == repl::ReadConcernLevel::kLocalReadConcern;
+    virtual ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const {
+        return {level == repl::ReadConcernLevel::kLocalReadConcern,
+                ReadConcernSupportResult::DefaultReadConcern::kNotPermitted};
     }
 
     /**
@@ -624,10 +663,11 @@ public:
      * the option to the shards as needed. We rely on the shards to fail the commands in the
      * cases where it isn't supported.
      */
-    virtual bool supportsReadConcern(const std::string& dbName,
-                                     const BSONObj& cmdObj,
-                                     repl::ReadConcernLevel level) const {
-        return level == repl::ReadConcernLevel::kLocalReadConcern;
+    virtual ReadConcernSupportResult supportsReadConcern(const std::string& dbName,
+                                                         const BSONObj& cmdObj,
+                                                         repl::ReadConcernLevel level) const {
+        return {level == repl::ReadConcernLevel::kLocalReadConcern,
+                ReadConcernSupportResult::DefaultReadConcern::kNotPermitted};
     }
 
     virtual bool allowsAfterClusterTime(const BSONObj& cmdObj) const {
diff --git a/src/mongo/db/commands/count_cmd.cpp b/src/mongo/db/commands/count_cmd.cpp
index da7b522855b..da2033e7645 100644
--- a/src/mongo/db/commands/count_cmd.cpp
+++ b/src/mongo/db/commands/count_cmd.cpp
@@ -90,10 +90,11 @@ public:
         return false;
     }
 
-    bool supportsReadConcern(const std::string& dbName,
-                             const BSONObj& cmdObj,
-                             repl::ReadConcernLevel level) const override {
-        return true;
+    ReadConcernSupportResult supportsReadConcern(const std::string& dbName,
+                                                 const BSONObj& cmdObj,
+                                                 repl::ReadConcernLevel level) const override {
+        return {ReadConcernSupportResult::ReadConcern::kSupported,
+                ReadConcernSupportResult::DefaultReadConcern::kPermitted};
     }
 
     ReadWriteType getReadWriteType() const override {
diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp
index 645839bb0f0..ff93a8fcfae 100644
--- a/src/mongo/db/commands/distinct.cpp
+++ b/src/mongo/db/commands/distinct.cpp
@@ -84,10 +84,11 @@ public:
         return true;
     }
 
-    bool supportsReadConcern(const std::string& dbName,
-                             const BSONObj& cmdObj,
-                             repl::ReadConcernLevel level) const override {
-        return true;
+    ReadConcernSupportResult supportsReadConcern(const std::string& dbName,
+                                                 const BSONObj& cmdObj,
+                                                 repl::ReadConcernLevel level) const override {
+        return {ReadConcernSupportResult::ReadConcern::kSupported,
+                ReadConcernSupportResult::DefaultReadConcern::kPermitted};
     }
 
     ReadWriteType getReadWriteType() const override {
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index 9da9269284f..5b91aa22420 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -176,8 +176,9 @@ public:
             return false;
         }
 
-        bool supportsReadConcern(repl::ReadConcernLevel level) const final {
-            return true;
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const final {
+            return {ReadConcernSupportResult::ReadConcern::kSupported,
+                    ReadConcernSupportResult::DefaultReadConcern::kPermitted};
         }
 
         bool canIgnorePrepareConflicts() const override {
diff --git a/src/mongo/db/commands/haystack.cpp b/src/mongo/db/commands/haystack.cpp
index 79e6907c28c..a14a4fd3a29 100644
--- a/src/mongo/db/commands/haystack.cpp
+++ b/src/mongo/db/commands/haystack.cpp
@@ -73,10 +73,11 @@ public:
         return AllowedOnSecondary::kAlways;
     }
 
-    bool supportsReadConcern(const std::string& dbName,
-                             const BSONObj& cmdObj,
-                             repl::ReadConcernLevel level) const final {
-        return true;
+    ReadConcernSupportResult supportsReadConcern(const std::string& dbName,
+                                                 const BSONObj& cmdObj,
+                                                 repl::ReadConcernLevel level) const final {
+        return {ReadConcernSupportResult::ReadConcern::kSupported,
+                ReadConcernSupportResult::DefaultReadConcern::kPermitted};
     }
 
     ReadWriteType getReadWriteType() const {
diff --git a/src/mongo/db/commands/pipeline_command.cpp b/src/mongo/db/commands/pipeline_command.cpp
index 4550b138626..85c220d091a 100644
--- a/src/mongo/db/commands/pipeline_command.cpp
+++ b/src/mongo/db/commands/pipeline_command.cpp
@@ -82,15 +82,17 @@ public:
             return true;
         }
 
-        bool supportsReadConcern(repl::ReadConcernLevel level) const override {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const override {
             // Aggregations that are run directly against a collection allow any read concern.
             // Otherwise, if the aggregate is collectionless then the read concern must be 'local'
             // (e.g. $currentOp). The exception to this is a $changeStream on a whole database,
             // which is considered collectionless but must be read concern 'majority'. Further read
             // concern validation is done one the pipeline is parsed.
-            return level == repl::ReadConcernLevel::kLocalReadConcern ||
-                level == repl::ReadConcernLevel::kMajorityReadConcern ||
-                !AggregationRequest::parseNs(_dbName, _request.body).isCollectionlessAggregateNS();
+            return {level == repl::ReadConcernLevel::kLocalReadConcern ||
+                        level == repl::ReadConcernLevel::kMajorityReadConcern ||
+                        !AggregationRequest::parseNs(_dbName, _request.body)
+                             .isCollectionlessAggregateNS(),
+                    ReadConcernSupportResult::DefaultReadConcern::kPermitted};
         }
 
         bool allowsSpeculativeMajorityReads() const override {
diff --git a/src/mongo/db/read_write_concern_defaults.cpp b/src/mongo/db/read_write_concern_defaults.cpp
index 828160e2e41..267ee413045 100644
--- a/src/mongo/db/read_write_concern_defaults.cpp
+++ b/src/mongo/db/read_write_concern_defaults.cpp
@@ -35,15 +35,27 @@
 
 namespace mongo {
 
+namespace {
+
+static constexpr auto kReadConcernLevelsDisallowedAsDefault = {
+    repl::ReadConcernLevel::kSnapshotReadConcern, repl::ReadConcernLevel::kLinearizableReadConcern};
+
+}
+
+bool ReadWriteConcernDefaults::isSuitableReadConcernLevel(repl::ReadConcernLevel level) {
+    for (auto bannedLevel : kReadConcernLevelsDisallowedAsDefault) {
+        if (level == bannedLevel) {
+            return false;
+        }
+    }
+    return true;
+}
+
 void ReadWriteConcernDefaults::checkSuitabilityAsDefault(const ReadConcern& rc) {
     uassert(ErrorCodes::BadValue,
-            str::stream() << "level: '" << ReadConcern::kSnapshotReadConcernStr
-                          << "' is not suitable for the default read concern",
-            rc.getLevel() != repl::ReadConcernLevel::kSnapshotReadConcern);
-    uassert(ErrorCodes::BadValue,
-            str::stream() << "level: '" << ReadConcern::kLinearizableReadConcernStr
+            str::stream() << "level: '" << repl::readConcernLevels::toString(rc.getLevel())
                           << "' is not suitable for the default read concern",
-            rc.getLevel() != repl::ReadConcernLevel::kLinearizableReadConcern);
+            isSuitableReadConcernLevel(rc.getLevel()));
     uassert(ErrorCodes::BadValue,
             str::stream() << "'" << ReadConcern::kAfterOpTimeFieldName
                           << "' is not suitable for the default read concern",
diff --git a/src/mongo/db/read_write_concern_defaults.h b/src/mongo/db/read_write_concern_defaults.h
index f7ce2f9bfee..cb8b6386ae8 100644
--- a/src/mongo/db/read_write_concern_defaults.h
+++ b/src/mongo/db/read_write_concern_defaults.h
@@ -61,6 +61,15 @@ public:
     ReadWriteConcernDefaults() = default;
     ~ReadWriteConcernDefaults() = default;
 
+    /**
+     * Returns true if the RC level is permissible to use as a default, and false if it cannot be a
+     * RC default.
+     */
+    static bool isSuitableReadConcernLevel(repl::ReadConcernLevel level);
+
+    /**
+     * Checks if the given RWC is suitable to use as a default, and uasserts if not.
+     */
     static void checkSuitabilityAsDefault(const ReadConcern& rc);
     static void checkSuitabilityAsDefault(const WriteConcern& wc);
 
diff --git a/src/mongo/db/repl/read_concern_args.cpp b/src/mongo/db/repl/read_concern_args.cpp
index a5de08a0fd7..d7ef1ce6ff1 100644
--- a/src/mongo/db/repl/read_concern_args.cpp
+++ b/src/mongo/db/repl/read_concern_args.cpp
@@ -57,18 +57,20 @@ const ReadConcernArgs& ReadConcernArgs::get(const OperationContext* opCtx) {
 }
 
 
-ReadConcernArgs::ReadConcernArgs() = default;
+ReadConcernArgs::ReadConcernArgs() : _specified(false) {}
 
 ReadConcernArgs::ReadConcernArgs(boost::optional<ReadConcernLevel> level)
-    : _level(std::move(level)) {}
+    : _level(std::move(level)), _specified(_level) {}
 
 ReadConcernArgs::ReadConcernArgs(boost::optional<OpTime> opTime,
                                  boost::optional<ReadConcernLevel> level)
-    : _opTime(std::move(opTime)), _level(std::move(level)) {}
+    : _opTime(std::move(opTime)), _level(std::move(level)), _specified(_opTime || _level) {}
 
 ReadConcernArgs::ReadConcernArgs(boost::optional<LogicalTime> clusterTime,
                                  boost::optional<ReadConcernLevel> level)
-    : _afterClusterTime(std::move(clusterTime)), _level(std::move(level)) {}
+    : _afterClusterTime(std::move(clusterTime)),
+      _level(std::move(level)),
+      _specified(_afterClusterTime || _level) {}
 
 std::string ReadConcernArgs::toString() const {
     return toBSON().toString();
@@ -84,6 +86,10 @@ bool ReadConcernArgs::isEmpty() const {
     return !_afterClusterTime && !_opTime && !_atClusterTime && !_level;
 }
 
+bool ReadConcernArgs::isSpecified() const {
+    return _specified;
+}
+
 ReadConcernLevel ReadConcernArgs::getLevel() const {
     return _level.value_or(ReadConcernLevel::kLocalReadConcern);
 }
@@ -105,6 +111,8 @@ boost::optional<LogicalTime> ReadConcernArgs::getArgsAtClusterTime() const {
 }
 
 Status ReadConcernArgs::initialize(const BSONElement& readConcernElem) {
+    invariant(isEmpty());  // only legal to call on uninitialized object.
+    _specified = false;
     if (readConcernElem.eoo()) {
         return Status::OK();
     }
@@ -158,21 +166,15 @@ Status ReadConcernArgs::parse(const BSONObj& readConcernObj) {
                 return readCommittedStatus;
             }
 
-            if (levelString == kLocalReadConcernStr) {
-                _level = ReadConcernLevel::kLocalReadConcern;
-            } else if (levelString == kMajorityReadConcernStr) {
-                _level = ReadConcernLevel::kMajorityReadConcern;
-            } else if (levelString == kLinearizableReadConcernStr) {
-                _level = ReadConcernLevel::kLinearizableReadConcern;
-            } else if (levelString == kAvailableReadConcernStr) {
-                _level = ReadConcernLevel::kAvailableReadConcern;
-            } else if (levelString == kSnapshotReadConcernStr) {
-                _level = ReadConcernLevel::kSnapshotReadConcern;
-            } else {
+            _level = readConcernLevels::fromString(levelString);
+            if (!_level) {
                 return Status(ErrorCodes::FailedToParse,
                               str::stream() << kReadConcernFieldName << '.' << kLevelFieldName
-                                            << " must be either 'local', 'majority', "
-                                               "'linearizable', 'available', or 'snapshot'");
+                                            << " must be either '" << readConcernLevels::kLocalName
+                                            << "', '" << readConcernLevels::kMajorityName << "', '"
+                                            << readConcernLevels::kLinearizableName << "', '"
+                                            << readConcernLevels::kAvailableName << "', or '"
+                                            << readConcernLevels::kSnapshotName << "'");
             }
         } else {
             return Status(ErrorCodes::InvalidOptions,
@@ -202,22 +204,23 @@ Status ReadConcernArgs::parse(const BSONObj& readConcernObj) {
         return Status(ErrorCodes::InvalidOptions,
                       str::stream()
                           << kAfterClusterTimeFieldName << " field can be set only if "
-                          << kLevelFieldName << " is equal to " << kMajorityReadConcernStr << ", "
-                          << kLocalReadConcernStr << ", or " << kSnapshotReadConcernStr);
+                          << kLevelFieldName << " is equal to " << readConcernLevels::kMajorityName
+                          << ", " << readConcernLevels::kLocalName << ", or "
+                          << readConcernLevels::kSnapshotName);
     }
 
     if (_opTime && getLevel() == ReadConcernLevel::kSnapshotReadConcern) {
         return Status(ErrorCodes::InvalidOptions,
                       str::stream()
                           << kAfterOpTimeFieldName << " field cannot be set if " << kLevelFieldName
-                          << " is equal to " << kSnapshotReadConcernStr);
+                          << " is equal to " << readConcernLevels::kSnapshotName);
     }
 
     if (_atClusterTime && getLevel() != ReadConcernLevel::kSnapshotReadConcern) {
         return Status(ErrorCodes::InvalidOptions,
-                      str::stream()
-                          << kAtClusterTimeFieldName << " field can be set only if "
-                          << kLevelFieldName << " is equal to " << kSnapshotReadConcernStr);
+                      str::stream() << kAtClusterTimeFieldName << " field can be set only if "
+                                    << kLevelFieldName << " is equal to "
+                                    << readConcernLevels::kSnapshotName);
     }
 
     // Make sure that atClusterTime wasn't specified with zero seconds.
@@ -233,6 +236,7 @@ Status ReadConcernArgs::parse(const BSONObj& readConcernObj) {
                       str::stream() << kAfterClusterTimeFieldName << " cannot be a null timestamp");
     }
 
+    _specified = true;
     return Status::OK();
 }
 
@@ -261,33 +265,7 @@ void ReadConcernArgs::appendInfo(BSONObjBuilder* builder) const {
     BSONObjBuilder rcBuilder(builder->subobjStart(kReadConcernFieldName));
 
     if (_level) {
-        StringData levelName;
-        switch (_level.get()) {
-            case ReadConcernLevel::kLocalReadConcern:
-                levelName = kLocalReadConcernStr;
-                break;
-
-            case ReadConcernLevel::kMajorityReadConcern:
-                levelName = kMajorityReadConcernStr;
-                break;
-
-            case ReadConcernLevel::kLinearizableReadConcern:
-                levelName = kLinearizableReadConcernStr;
-                break;
-
-            case ReadConcernLevel::kAvailableReadConcern:
-                levelName = kAvailableReadConcernStr;
-                break;
-
-            case ReadConcernLevel::kSnapshotReadConcern:
-                levelName = kSnapshotReadConcernStr;
-                break;
-
-            default:
-                MONGO_UNREACHABLE;
-        }
-
-        rcBuilder.append(kLevelFieldName, levelName);
+        rcBuilder.append(kLevelFieldName, readConcernLevels::toString(_level.get()));
     }
 
     if (_opTime) {
@@ -305,5 +283,43 @@ void ReadConcernArgs::appendInfo(BSONObjBuilder* builder) const {
     rcBuilder.done();
 }
 
+boost::optional<ReadConcernLevel> readConcernLevels::fromString(StringData levelString) {
+    if (levelString == readConcernLevels::kLocalName) {
+        return ReadConcernLevel::kLocalReadConcern;
+    } else if (levelString == readConcernLevels::kMajorityName) {
+        return ReadConcernLevel::kMajorityReadConcern;
+    } else if (levelString == readConcernLevels::kLinearizableName) {
+        return ReadConcernLevel::kLinearizableReadConcern;
+    } else if (levelString == readConcernLevels::kAvailableName) {
+        return ReadConcernLevel::kAvailableReadConcern;
+    } else if (levelString == readConcernLevels::kSnapshotName) {
+        return ReadConcernLevel::kSnapshotReadConcern;
+    } else {
+        return boost::none;
+    }
+}
+
+StringData readConcernLevels::toString(ReadConcernLevel level) {
+    switch (level) {
+        case ReadConcernLevel::kLocalReadConcern:
+            return kLocalName;
+
+        case ReadConcernLevel::kMajorityReadConcern:
+            return kMajorityName;
+
+        case ReadConcernLevel::kLinearizableReadConcern:
+            return kLinearizableName;
+
+        case ReadConcernLevel::kAvailableReadConcern:
+            return kAvailableName;
+
+        case ReadConcernLevel::kSnapshotReadConcern:
+            return kSnapshotName;
+
+        default:
+            MONGO_UNREACHABLE;
+    }
+}
+
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/repl/read_concern_args.h b/src/mongo/db/repl/read_concern_args.h
index 310cfc59432..110624649bd 100644
--- a/src/mongo/db/repl/read_concern_args.h
+++ b/src/mongo/db/repl/read_concern_args.h
@@ -54,12 +54,6 @@ public:
     static constexpr StringData kAtClusterTimeFieldName = "atClusterTime"_sd;
     static constexpr StringData kLevelFieldName = "level"_sd;
 
-    static constexpr StringData kLocalReadConcernStr = "local"_sd;
-    static constexpr StringData kMajorityReadConcernStr = "majority"_sd;
-    static constexpr StringData kLinearizableReadConcernStr = "linearizable"_sd;
-    static constexpr StringData kAvailableReadConcernStr = "available"_sd;
-    static constexpr StringData kSnapshotReadConcernStr = "snapshot"_sd;
-
     /**
      * Represents the internal mechanism an operation uses to satisfy 'majority' read concern.
      */
@@ -147,6 +141,13 @@ public:
     bool isEmpty() const;
 
     /**
+     * Returns true if this ReadConcernArgs represents a read concern that was actually specified.
+     * If the RC was specified as an empty BSON object this will still be true (unlike isEmpty()).
+     * False represents an absent or missing read concern, ie. one which wasn't present at all.
+     */
+    bool isSpecified() const;
+
+    /**
      *  Returns default kLocalReadConcern if _level is not set.
      */
     ReadConcernLevel getLevel() const;
@@ -187,6 +188,12 @@ private:
      * level is 'majority'.
      */
     MajorityReadMechanism _majorityReadMechanism{MajorityReadMechanism::kMajoritySnapshot};
+
+    /**
+     * True indicates that a read concern has been specified (even if it might be empty), as
+     * opposed to being absent or missing.
+     */
+    bool _specified;
 };
 
 }  // namespace repl
diff --git a/src/mongo/db/repl/read_concern_level.h b/src/mongo/db/repl/read_concern_level.h
index 9b8c1951de9..e0a8f5c3304 100644
--- a/src/mongo/db/repl/read_concern_level.h
+++ b/src/mongo/db/repl/read_concern_level.h
@@ -29,6 +29,8 @@
 
 #pragma once
 
+#include <boost/optional.hpp>
+
 namespace mongo {
 namespace repl {
 
@@ -40,5 +42,24 @@ enum class ReadConcernLevel {
     kSnapshotReadConcern
 };
 
+namespace readConcernLevels {
+
+constexpr std::initializer_list<ReadConcernLevel> all = {ReadConcernLevel::kLocalReadConcern,
+                                                         ReadConcernLevel::kMajorityReadConcern,
+                                                         ReadConcernLevel::kLinearizableReadConcern,
+                                                         ReadConcernLevel::kAvailableReadConcern,
+                                                         ReadConcernLevel::kSnapshotReadConcern};
+
+constexpr StringData kLocalName = "local"_sd;
+constexpr StringData kMajorityName = "majority"_sd;
+constexpr StringData kLinearizableName = "linearizable"_sd;
+constexpr StringData kAvailableName = "available"_sd;
+constexpr StringData kSnapshotName = "snapshot"_sd;
+
+boost::optional<ReadConcernLevel> fromString(StringData levelString);
+StringData toString(ReadConcernLevel level);
+
+}  // namespace readConcernLevels
+
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/service_entry_point_common.cpp b/src/mongo/db/service_entry_point_common.cpp
index 182509fa32a..6fb5557d3d4 100644
--- a/src/mongo/db/service_entry_point_common.cpp
+++ b/src/mongo/db/service_entry_point_common.cpp
@@ -62,6 +62,7 @@
 #include "mongo/db/ops/write_ops_exec.h"
 #include "mongo/db/query/find.h"
 #include "mongo/db/read_concern.h"
+#include "mongo/db/read_write_concern_defaults.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/repl/read_concern_args.h"
 #include "mongo/db/repl/repl_client_info.h"
@@ -226,7 +227,8 @@ private:
  * Given the specified command, returns an effective read concern which should be used or an error
  * if the read concern is not valid for the command.
  */
-StatusWith<repl::ReadConcernArgs> _extractReadConcern(const CommandInvocation* invocation,
+StatusWith<repl::ReadConcernArgs> _extractReadConcern(OperationContext* opCtx,
+                                                      const CommandInvocation* invocation,
                                                       const BSONObj& cmdObj,
                                                       bool startTransaction) {
     repl::ReadConcernArgs readConcernArgs;
@@ -236,7 +238,41 @@ StatusWith<repl::ReadConcernArgs> _extractReadConcern(const CommandInvocation* i
         return readConcernParseStatus;
     }
 
+    auto readConcernSupport = invocation->supportsReadConcern(readConcernArgs.getLevel());
+    if (readConcernSupport.defaultReadConcern ==
+            ReadConcernSupportResult::DefaultReadConcern::kPermitted &&
+        !opCtx->getClient()->isInDirectClient()) {
+        if (serverGlobalParams.clusterRole == ClusterRole::ShardServer ||
+            serverGlobalParams.clusterRole == ClusterRole::ConfigServer) {
+            // ReadConcern should always be explicitly specified by operations received on shard and
+            // config servers, even if it is empty (ie. readConcern: {}).  In this context
+            // (shard/config servers) an empty RC indicates the operation should use the implicit
+            // server defaults.  So, warn if the operation has not specified readConcern and is on a
+            // shard/config server.
+            if (!readConcernArgs.isSpecified()) {
+                // TODO: Disabled until after SERVER-43712, to avoid log spam.
+                // log() << "Missing readConcern on " << invocation->definition()->getName();
+            }
+        } else {
+            // A member in a regular replica set.  Since these servers receive client queries, in
+            // this context empty RC (ie. readConcern: {}) means the same as if absent/unspecified,
+            // which is to apply the CWRWC defaults if present.  This means we just test isEmpty(),
+            // since this covers both isSpecified() && !isSpecified()
+            if (readConcernArgs.isEmpty()) {
+                const auto rcDefault = ReadWriteConcernDefaults::get(opCtx->getServiceContext())
+                                           .getDefaultReadConcern();
+                if (rcDefault) {
+                    readConcernArgs = *rcDefault;
+                    LOG(2) << "Applying default readConcern on "
+                           << invocation->definition()->getName() << " of " << *rcDefault;
+                }
+            }
+        }
+    }
+
     auto readConcernLevel = readConcernArgs.getLevel();
+    // Update the readConcernSupport, in case the default RC was applied.
+    readConcernSupport = invocation->supportsReadConcern(readConcernLevel);
     if (startTransaction && readConcernLevel != repl::ReadConcernLevel::kSnapshotReadConcern &&
         readConcernLevel != repl::ReadConcernLevel::kMajorityReadConcern &&
         readConcernLevel != repl::ReadConcernLevel::kLocalReadConcern) {
@@ -256,7 +292,8 @@ StatusWith<repl::ReadConcernArgs> _extractReadConcern(const CommandInvocation* i
     // There is no need to check if the command supports the read concern while in a transaction
     // because all commands that are allowed to run in a transaction must support all the read
     // concerns that can be used with a transaction.
-    if (!startTransaction && !invocation->supportsReadConcern(readConcernLevel)) {
+    if (!startTransaction &&
+        readConcernSupport.readConcern == ReadConcernSupportResult::ReadConcern::kNotSupported) {
         return {ErrorCodes::InvalidOptions,
                 str::stream() << "Command does not support read concern "
                               << readConcernArgs.toString()};
@@ -564,6 +601,18 @@ bool runCommandImpl(OperationContext* opCtx,
             // the command body.
             behaviors.uassertCommandDoesNotSpecifyWriteConcern(request.body);
         } else {
+            // WriteConcern should always be explicitly specified by operations received on shard
+            // and config servers, even if it is empty (ie. writeConcern: {}).  In this context
+            // (shard/config servers) an empty WC indicates the operation should use the implicit
+            // server defaults.  So, warn if the operation has not specified writeConcern and is on
+            // a shard/config server.
+            if (!opCtx->getClient()->isInDirectClient() &&
+                (serverGlobalParams.clusterRole == ClusterRole::ShardServer ||
+                 serverGlobalParams.clusterRole == ClusterRole::ConfigServer) &&
+                !request.body.hasField(WriteConcernOptions::kWriteConcernField)) {
+                // TODO: Disabled until after SERVER-43712, to avoid log spam.
+                // log() << "Missing writeConcern on " << command->getName();
+            }
             extractedWriteConcern.emplace(
                 uassertStatusOK(extractWriteConcern(opCtx, request.body)));
             if (sessionOptions.getAutocommit()) {
@@ -862,7 +911,7 @@ void execCommandDatabase(OperationContext* opCtx,
         bool startTransaction = static_cast<bool>(sessionOptions.getStartTransaction());
         if (!skipReadConcern) {
             auto newReadConcernArgs = uassertStatusOK(
-                _extractReadConcern(invocation.get(), request.body, startTransaction));
+                _extractReadConcern(opCtx, invocation.get(), request.body, startTransaction));
             {
                 // We must obtain the client lock to set the ReadConcernArgs on the operation
                 // context as it may be concurrently read by CurrentOp.
@@ -976,7 +1025,8 @@ void execCommandDatabase(OperationContext* opCtx,
         // parse it here, so if it is valid it can be used to compute the proper operationTime.
         auto& readConcernArgs = repl::ReadConcernArgs::get(opCtx);
         if (readConcernArgs.isEmpty()) {
-            auto readConcernArgsStatus = _extractReadConcern(invocation.get(), request.body, false);
+            auto readConcernArgsStatus =
+                _extractReadConcern(opCtx, invocation.get(), request.body, false);
             if (readConcernArgsStatus.isOK()) {
                 // We must obtain the client lock to set the ReadConcernArgs on the operation
                 // context as it may be concurrently read by CurrentOp.
diff --git a/src/mongo/db/write_concern.cpp b/src/mongo/db/write_concern.cpp
index 30c01e5a831..88577ae6269 100644
--- a/src/mongo/db/write_concern.cpp
+++ b/src/mongo/db/write_concern.cpp
@@ -38,6 +38,7 @@
 #include "mongo/db/client.h"
 #include "mongo/db/commands/server_status_metric.h"
 #include "mongo/db/operation_context.h"
+#include "mongo/db/read_write_concern_defaults.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/server_options.h"
@@ -79,11 +80,24 @@ StatusWith<WriteConcernOptions> extractWriteConcern(OperationContext* opCtx,
 
     WriteConcernOptions writeConcern = wcResult.getValue();
 
-    // Get the default write concern specified in ReplSetConfig only if no write concern is
-    // specified in the command (when usedDefault is true) to avoid locking the
-    // ReplicationCoordinator mutex unconditionally.
+    // If no write concern is specified in the command (so usedDefault is true), then use the
+    // cluster-wide default WC (if there is one), or else the default WC from the ReplSetConfig
+    // (which takes the ReplicationCoordinator mutex).
     if (writeConcern.usedDefault) {
-        writeConcern = repl::ReplicationCoordinator::get(opCtx)->getGetLastErrorDefault();
+        writeConcern = ([&]() {
+            // WriteConcern defaults can only be applied on regular replica set members.  Operations
+            // received by shard and config servers should always have WC explicitly specified.
+            if (serverGlobalParams.clusterRole != ClusterRole::ShardServer &&
+                serverGlobalParams.clusterRole != ClusterRole::ConfigServer &&
+                !opCtx->getClient()->isInDirectClient()) {
+                auto wcDefault = ReadWriteConcernDefaults::get(opCtx->getServiceContext())
+                                     .getDefaultWriteConcern();
+                if (wcDefault) {
+                    return *wcDefault;
+                }
+            }
+            return repl::ReplicationCoordinator::get(opCtx)->getGetLastErrorDefault();
+        })();
         if (writeConcern.wNumNodes == 0 && writeConcern.wMode.empty()) {
             writeConcern.wNumNodes = 1;
         }
diff --git a/src/mongo/s/commands/cluster_distinct_cmd.cpp b/src/mongo/s/commands/cluster_distinct_cmd.cpp
index c584258344c..3c9c667678d 100644
--- a/src/mongo/s/commands/cluster_distinct_cmd.cpp
+++ b/src/mongo/s/commands/cluster_distinct_cmd.cpp
@@ -72,10 +72,11 @@ public:
         return false;
     }
 
-    bool supportsReadConcern(const std::string& dbName,
-                             const BSONObj& cmdObj,
-                             repl::ReadConcernLevel level) const final {
-        return true;
+    ReadConcernSupportResult supportsReadConcern(const std::string& dbName,
+                                                 const BSONObj& cmdObj,
+                                                 repl::ReadConcernLevel level) const final {
+        return {ReadConcernSupportResult::ReadConcern::kSupported,
+                ReadConcernSupportResult::DefaultReadConcern::kPermitted};
     }
 
     void addRequiredPrivileges(const std::string& dbname,
diff --git a/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp b/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp
index ee72e7e43b1..90442a2b4c2 100644
--- a/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp
+++ b/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp
@@ -153,10 +153,11 @@ public:
         return true;
     }
 
-    bool supportsReadConcern(const std::string& dbName,
-                             const BSONObj& cmdObj,
-                             repl::ReadConcernLevel level) const final {
-        return true;
+    ReadConcernSupportResult supportsReadConcern(const std::string& dbName,
+                                                 const BSONObj& cmdObj,
+                                                 repl::ReadConcernLevel level) const final {
+        return {ReadConcernSupportResult::ReadConcern::kSupported,
+                ReadConcernSupportResult::DefaultReadConcern::kPermitted};
     }
 
     void addRequiredPrivileges(const std::string& dbname,
diff --git a/src/mongo/s/commands/cluster_find_cmd.cpp b/src/mongo/s/commands/cluster_find_cmd.cpp
index e133875439b..091cac8879b 100644
--- a/src/mongo/s/commands/cluster_find_cmd.cpp
+++ b/src/mongo/s/commands/cluster_find_cmd.cpp
@@ -112,8 +112,9 @@ public:
             return false;
         }
 
-        bool supportsReadConcern(repl::ReadConcernLevel level) const final {
-            return true;
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const final {
+            return {ReadConcernSupportResult::ReadConcern::kSupported,
+                    ReadConcernSupportResult::DefaultReadConcern::kPermitted};
         }
 
         NamespaceString ns() const override {
diff --git a/src/mongo/s/commands/cluster_killcursors_cmd.cpp b/src/mongo/s/commands/cluster_killcursors_cmd.cpp
index 6d8671c4ab4..a90a24d3f53 100644
--- a/src/mongo/s/commands/cluster_killcursors_cmd.cpp
+++ b/src/mongo/s/commands/cluster_killcursors_cmd.cpp
@@ -42,11 +42,12 @@ class ClusterKillCursorsCmd final : public KillCursorsCmdBase {
 public:
     ClusterKillCursorsCmd() = default;
 
-    bool supportsReadConcern(const std::string& dbName,
-                             const BSONObj& cmdObj,
-                             repl::ReadConcernLevel level) const final {
+    ReadConcernSupportResult supportsReadConcern(const std::string& dbName,
+                                                 const BSONObj& cmdObj,
+                                                 repl::ReadConcernLevel level) const final {
         // killCursors must support read concerns in order to be run in transactions.
-        return true;
+        return {ReadConcernSupportResult::ReadConcern::kSupported,
+                ReadConcernSupportResult::DefaultReadConcern::kPermitted};
     }
 
     bool run(OperationContext* opCtx,
diff --git a/src/mongo/s/commands/cluster_pipeline_cmd.cpp b/src/mongo/s/commands/cluster_pipeline_cmd.cpp
index 25dbc52d9fc..02a70417d72 100644
--- a/src/mongo/s/commands/cluster_pipeline_cmd.cpp
+++ b/src/mongo/s/commands/cluster_pipeline_cmd.cpp
@@ -70,8 +70,9 @@ public:
             return true;
         }
 
-        bool supportsReadConcern(repl::ReadConcernLevel level) const override {
-            return true;
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const override {
+            return {ReadConcernSupportResult::ReadConcern::kSupported,
+                    ReadConcernSupportResult::DefaultReadConcern::kPermitted};
         }
 
         void _runAggCommand(OperationContext* opCtx,
diff --git a/src/mongo/s/commands/cluster_write_cmd.cpp b/src/mongo/s/commands/cluster_write_cmd.cpp
index 479ed6a81ba..e1b8a7a41e2 100644
--- a/src/mongo/s/commands/cluster_write_cmd.cpp
+++ b/src/mongo/s/commands/cluster_write_cmd.cpp
@@ -543,8 +543,9 @@ private:
         return true;
     }
 
-    bool supportsReadConcern(repl::ReadConcernLevel level) const final {
-        return true;
+    ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const final {
+        return {ReadConcernSupportResult::ReadConcern::kSupported,
+                ReadConcernSupportResult::DefaultReadConcern::kPermitted};
     }
 
     void doCheckAuthorization(OperationContext* opCtx) const final {