SERVER-56488 Change the default read concern to always be local

48 files changed, 530 insertions, 114 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 6cdc29d7ae4..b7e59d0e7ba 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -903,6 +903,7 @@ env.Library(
         'introspect',
         'lasterror',
         'query_exec',
+        'repl/repl_server_parameters',
         'repl/replica_set_messages',
         'shared_request_handling',
         'transaction',
diff --git a/src/mongo/db/commands.cpp b/src/mongo/db/commands.cpp
index 9501937342c..7f7c5f39155 100644
--- a/src/mongo/db/commands.cpp
+++ b/src/mongo/db/commands.cpp
@@ -880,8 +880,9 @@ private:
         return _command->supportsWriteConcern(cmdObj());
     }
 
-    ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const override {
-        return _command->supportsReadConcern(cmdObj(), level);
+    ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault) const override {
+        return _command->supportsReadConcern(cmdObj(), level, isImplicitDefault);
     }
 
     bool supportsReadMirroring() const override {
diff --git a/src/mongo/db/commands.h b/src/mongo/db/commands.h
index 086f08ddda0..07c08e5000d 100644
--- a/src/mongo/db/commands.h
+++ b/src/mongo/db/commands.h
@@ -658,7 +658,8 @@ public:
     /**
      * Returns this invocation's support for readConcern.
      */
-    virtual ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const {
+    virtual ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                         bool isImplicitDefault) const {
         static const Status kReadConcernNotSupported{ErrorCodes::InvalidOptions,
                                                      "read concern not supported"};
         static const Status kDefaultReadConcernNotPermitted{ErrorCodes::InvalidOptions,
@@ -686,7 +687,8 @@ public:
             return false;
         }
 
-        if (auto result = supportsReadConcern(repl::ReadConcernLevel::kMajorityReadConcern);
+        if (auto result = supportsReadConcern(repl::ReadConcernLevel::kMajorityReadConcern,
+                                              false /* isImplicitDefault */);
             result.readConcernSupport.isOK()) {
             // If the command supports read concern, it has storage and newtork implications.
             return false;
@@ -860,11 +862,12 @@ public:
      * cases where it isn't supported.
      */
     virtual ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
-                                                         repl::ReadConcernLevel level) const {
+                                                         repl::ReadConcernLevel level,
+                                                         bool isImplicitDefault) const {
         static const Status kReadConcernNotSupported{ErrorCodes::InvalidOptions,
                                                      "read concern not supported"};
-        static const Status kDefaultReadConcernNotPermitted{ErrorCodes::InvalidOptions,
-                                                            "default read concern not permitted"};
+        static const Status kDefaultReadConcernNotPermitted{
+            ErrorCodes::InvalidOptions, "cluster wide default read concern not permitted"};
         return {{level != repl::ReadConcernLevel::kLocalReadConcern, kReadConcernNotSupported},
                 {kDefaultReadConcernNotPermitted}};
     }
diff --git a/src/mongo/db/commands/count_cmd.cpp b/src/mongo/db/commands/count_cmd.cpp
index c53ffb44608..2d4769c5f86 100644
--- a/src/mongo/db/commands/count_cmd.cpp
+++ b/src/mongo/db/commands/count_cmd.cpp
@@ -97,7 +97,8 @@ public:
     }
 
     ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
-                                                 repl::ReadConcernLevel level) const override {
+                                                 repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault) const override {
         static const Status kSnapshotNotSupported{ErrorCodes::InvalidOptions,
                                                   "read concern snapshot not supported"};
         return {{level == repl::ReadConcernLevel::kSnapshotReadConcern, kSnapshotNotSupported},
diff --git a/src/mongo/db/commands/dbhash.cpp b/src/mongo/db/commands/dbhash.cpp
index 6ba7a8638dc..a17c4bc77c0 100644
--- a/src/mongo/db/commands/dbhash.cpp
+++ b/src/mongo/db/commands/dbhash.cpp
@@ -88,7 +88,8 @@ public:
     }
 
     ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
-                                                 repl::ReadConcernLevel level) const final {
+                                                 repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault) const final {
 
         static const Status kReadConcernNotSupported{ErrorCodes::InvalidOptions,
                                                      "read concern not supported"};
diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp
index 2c01df647bc..6c9dba828d3 100644
--- a/src/mongo/db/commands/distinct.cpp
+++ b/src/mongo/db/commands/distinct.cpp
@@ -95,7 +95,8 @@ public:
     }
 
     ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
-                                                 repl::ReadConcernLevel level) const override {
+                                                 repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault) const override {
         return ReadConcernSupportResult::allSupportedAndDefaultPermitted();
     }
 
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index 669f133fab5..00d5f57c803 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -198,7 +198,8 @@ public:
             return false;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const final {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const final {
             return ReadConcernSupportResult::allSupportedAndDefaultPermitted();
         }
 
diff --git a/src/mongo/db/commands/getmore_cmd.cpp b/src/mongo/db/commands/getmore_cmd.cpp
index b2b407c65d6..88eb60eb201 100644
--- a/src/mongo/db/commands/getmore_cmd.cpp
+++ b/src/mongo/db/commands/getmore_cmd.cpp
@@ -277,7 +277,8 @@ public:
             return false;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const override {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const override {
             return kSupportsReadConcernResult;
         }
 
diff --git a/src/mongo/db/commands/killcursors_common.h b/src/mongo/db/commands/killcursors_common.h
index 4d906b66777..06ee9c9335d 100644
--- a/src/mongo/db/commands/killcursors_common.h
+++ b/src/mongo/db/commands/killcursors_common.h
@@ -74,11 +74,12 @@ public:
             return false;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const final {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const final {
             if constexpr (Impl::supportsReadConcern) {
                 return ReadConcernSupportResult::allSupportedAndDefaultPermitted();
             } else {
-                return KCV1Gen::InvocationBaseGen::supportsReadConcern(level);
+                return KCV1Gen::InvocationBaseGen::supportsReadConcern(level, isImplicitDefault);
             }
         }
 
diff --git a/src/mongo/db/commands/map_reduce_command_base.h b/src/mongo/db/commands/map_reduce_command_base.h
index bcc0870dfe5..f3806c40766 100644
--- a/src/mongo/db/commands/map_reduce_command_base.h
+++ b/src/mongo/db/commands/map_reduce_command_base.h
@@ -51,7 +51,8 @@ public:
      * the aggregate command.
      */
     virtual ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
-                                                         repl::ReadConcernLevel level) const {
+                                                         repl::ReadConcernLevel level,
+                                                         bool isImplicitDefault) const {
         static const Status kReadConcernNotSupported{ErrorCodes::InvalidOptions,
                                                      "read concern not supported"};
         static const Status kDefaultReadConcernNotPermitted{ErrorCodes::InvalidOptions,
diff --git a/src/mongo/db/commands/pipeline_command.cpp b/src/mongo/db/commands/pipeline_command.cpp
index beed004686e..c3772f5af11 100644
--- a/src/mongo/db/commands/pipeline_command.cpp
+++ b/src/mongo/db/commands/pipeline_command.cpp
@@ -121,9 +121,11 @@ public:
             return true;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const override {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const override {
             return _liteParsedPipeline.supportsReadConcern(
                 level,
+                isImplicitDefault,
                 _aggregationRequest.getExplain(),
                 serverGlobalParams.enableMajorityReadConcern);
         }
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 74874ee9e93..0c7b25d8520 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -472,8 +472,8 @@ boost::intrusive_ptr<ExpressionContext> makeExpressionContext(
  */
 void _adjustChangeStreamReadConcern(OperationContext* opCtx) {
     repl::ReadConcernArgs& readConcernArgs = repl::ReadConcernArgs::get(opCtx);
-    // There is already a read concern level set. Do nothing.
-    if (readConcernArgs.hasLevel()) {
+    // There is already a non-default read concern level set. Do nothing.
+    if (readConcernArgs.hasLevel() && !readConcernArgs.getProvenance().isImplicitDefault()) {
         return;
     }
     // We upconvert an empty read concern to 'majority'.
diff --git a/src/mongo/db/commands/txn_cmds.cpp b/src/mongo/db/commands/txn_cmds.cpp
index 9287537a5e2..9cac96afe54 100644
--- a/src/mongo/db/commands/txn_cmds.cpp
+++ b/src/mongo/db/commands/txn_cmds.cpp
@@ -185,7 +185,8 @@ public:
             return true;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const final {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const final {
             // abortTransaction commences running inside a transaction (even though the transaction
             // will be ended by the time it completes).  Therefore it needs to accept any
             // readConcern which is valid within a transaction.  However it is not appropriate to
diff --git a/src/mongo/db/pipeline/document_source_change_stream.h b/src/mongo/db/pipeline/document_source_change_stream.h
index 8768c7aee30..144dddf84d4 100644
--- a/src/mongo/db/pipeline/document_source_change_stream.h
+++ b/src/mongo/db/pipeline/document_source_change_stream.h
@@ -87,13 +87,14 @@ public:
             }
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const {
             // Change streams require "majority" readConcern. If the client did not specify an
             // explicit readConcern, change streams will internally upconvert the readConcern to
             // majority (so clients can always send aggregations without readConcern). We therefore
             // do not permit the cluster-wide default to be applied.
             return onlySingleReadConcernSupported(
-                kStageName, repl::ReadConcernLevel::kMajorityReadConcern, level);
+                kStageName, repl::ReadConcernLevel::kMajorityReadConcern, level, isImplicitDefault);
         }
 
         void assertSupportsMultiDocumentTransaction() const {
diff --git a/src/mongo/db/pipeline/document_source_current_op.h b/src/mongo/db/pipeline/document_source_current_op.h
index ed2093c449b..49e0be6c35d 100644
--- a/src/mongo/db/pipeline/document_source_current_op.h
+++ b/src/mongo/db/pipeline/document_source_current_op.h
@@ -82,8 +82,9 @@ public:
             return true;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const {
-            return onlyReadConcernLocalSupported(kStageName, level);
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const {
+            return onlyReadConcernLocalSupported(kStageName, level, isImplicitDefault);
         }
 
         void assertSupportsMultiDocumentTransaction() const {
diff --git a/src/mongo/db/pipeline/document_source_list_cached_and_active_users.h b/src/mongo/db/pipeline/document_source_list_cached_and_active_users.h
index 35ab6dfe54c..f1f1e25e55e 100644
--- a/src/mongo/db/pipeline/document_source_list_cached_and_active_users.h
+++ b/src/mongo/db/pipeline/document_source_list_cached_and_active_users.h
@@ -73,8 +73,9 @@ public:
             return false;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const {
-            return onlyReadConcernLocalSupported(kStageName, level);
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const {
+            return onlyReadConcernLocalSupported(kStageName, level, isImplicitDefault);
         }
 
         void assertSupportsMultiDocumentTransaction() const {
diff --git a/src/mongo/db/pipeline/document_source_list_local_sessions.h b/src/mongo/db/pipeline/document_source_list_local_sessions.h
index 4615296a0ed..5306fc1fb7e 100644
--- a/src/mongo/db/pipeline/document_source_list_local_sessions.h
+++ b/src/mongo/db/pipeline/document_source_list_local_sessions.h
@@ -84,8 +84,9 @@ public:
             return false;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const {
-            return onlyReadConcernLocalSupported(kStageName, level);
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const {
+            return onlyReadConcernLocalSupported(kStageName, level, isImplicitDefault);
         }
 
         void assertSupportsMultiDocumentTransaction() const {
diff --git a/src/mongo/db/pipeline/document_source_merge.h b/src/mongo/db/pipeline/document_source_merge.h
index f3c76be7e2f..54e9620793f 100644
--- a/src/mongo/db/pipeline/document_source_merge.h
+++ b/src/mongo/db/pipeline/document_source_merge.h
@@ -88,7 +88,8 @@ public:
             return false;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const final {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const final {
             return {
                 {level == repl::ReadConcernLevel::kLinearizableReadConcern,
                  {ErrorCodes::InvalidOptions,
diff --git a/src/mongo/db/pipeline/document_source_out.h b/src/mongo/db/pipeline/document_source_out.h
index b01182ffcbd..140544897d1 100644
--- a/src/mongo/db/pipeline/document_source_out.h
+++ b/src/mongo/db/pipeline/document_source_out.h
@@ -67,7 +67,8 @@ public:
             return {Privilege(ResourcePattern::forExactNamespace(_foreignNss), actions)};
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const final {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const final {
             return {
                 {level == repl::ReadConcernLevel::kLinearizableReadConcern,
                  {ErrorCodes::InvalidOptions,
diff --git a/src/mongo/db/pipeline/document_source_plan_cache_stats.h b/src/mongo/db/pipeline/document_source_plan_cache_stats.h
index 06aa28d908f..d72b0a22ac8 100644
--- a/src/mongo/db/pipeline/document_source_plan_cache_stats.h
+++ b/src/mongo/db/pipeline/document_source_plan_cache_stats.h
@@ -67,8 +67,9 @@ public:
             return false;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const {
-            return onlyReadConcernLocalSupported(kStageName, level);
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const {
+            return onlyReadConcernLocalSupported(kStageName, level, isImplicitDefault);
         }
 
         void assertSupportsMultiDocumentTransaction() const {
diff --git a/src/mongo/db/pipeline/lite_parsed_document_source.h b/src/mongo/db/pipeline/lite_parsed_document_source.h
index a504eeb9b3c..ceb33521bcc 100644
--- a/src/mongo/db/pipeline/lite_parsed_document_source.h
+++ b/src/mongo/db/pipeline/lite_parsed_document_source.h
@@ -192,7 +192,8 @@ public:
     /**
      * Verifies that this stage is allowed to run with the specified read concern level.
      */
-    virtual ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const {
+    virtual ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                         bool isImplicitDefault) const {
         return ReadConcernSupportResult::allSupportedAndDefaultPermitted();
     }
 
@@ -224,11 +225,11 @@ protected:
                                 << "multi-document transaction.");
     }
 
-    ReadConcernSupportResult onlySingleReadConcernSupported(
-        StringData stageName,
-        repl::ReadConcernLevel supportedLevel,
-        repl::ReadConcernLevel candidateLevel) const {
-        return {{candidateLevel != supportedLevel,
+    ReadConcernSupportResult onlySingleReadConcernSupported(StringData stageName,
+                                                            repl::ReadConcernLevel supportedLevel,
+                                                            repl::ReadConcernLevel candidateLevel,
+                                                            bool isImplicitDefault) const {
+        return {{candidateLevel != supportedLevel && !isImplicitDefault,
                  {ErrorCodes::InvalidOptions,
                   str::stream() << "Aggregation stage " << stageName
                                 << " cannot run with a readConcern other than '"
@@ -241,9 +242,10 @@ protected:
     }
 
     ReadConcernSupportResult onlyReadConcernLocalSupported(StringData stageName,
-                                                           repl::ReadConcernLevel level) const {
+                                                           repl::ReadConcernLevel level,
+                                                           bool isImplicitDefault) const {
         return onlySingleReadConcernSupported(
-            stageName, repl::ReadConcernLevel::kLocalReadConcern, level);
+            stageName, repl::ReadConcernLevel::kLocalReadConcern, level, isImplicitDefault);
     }
 
 private:
diff --git a/src/mongo/db/pipeline/lite_parsed_pipeline.cpp b/src/mongo/db/pipeline/lite_parsed_pipeline.cpp
index 2e5ba95c7d7..437c7d17f5c 100644
--- a/src/mongo/db/pipeline/lite_parsed_pipeline.cpp
+++ b/src/mongo/db/pipeline/lite_parsed_pipeline.cpp
@@ -39,6 +39,7 @@ namespace mongo {
 
 ReadConcernSupportResult LiteParsedPipeline::supportsReadConcern(
     repl::ReadConcernLevel level,
+    bool isImplicitDefault,
     boost::optional<ExplainOptions::Verbosity> explain,
     bool enableMajorityReadConcern) const {
     // Start by assuming that we will support both readConcern and cluster-wide default.
@@ -77,7 +78,7 @@ ReadConcernSupportResult LiteParsedPipeline::supportsReadConcern(
         if (!result.readConcernSupport.isOK() && !result.defaultReadConcernPermit.isOK()) {
             break;
         }
-        result.merge(spec->supportsReadConcern(level));
+        result.merge(spec->supportsReadConcern(level, isImplicitDefault));
     }
 
     return result;
diff --git a/src/mongo/db/pipeline/lite_parsed_pipeline.h b/src/mongo/db/pipeline/lite_parsed_pipeline.h
index be9601150e0..51303f3aa04 100644
--- a/src/mongo/db/pipeline/lite_parsed_pipeline.h
+++ b/src/mongo/db/pipeline/lite_parsed_pipeline.h
@@ -128,6 +128,7 @@ public:
      * Verifies that this pipeline is allowed to run with the specified read concern level.
      */
     ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault,
                                                  boost::optional<ExplainOptions::Verbosity> explain,
                                                  bool enableMajorityReadConcern) const;
 
diff --git a/src/mongo/db/read_concern_support_result.h b/src/mongo/db/read_concern_support_result.h
index 9bca706a15b..672f8f673d1 100644
--- a/src/mongo/db/read_concern_support_result.h
+++ b/src/mongo/db/read_concern_support_result.h
@@ -46,7 +46,7 @@ struct ReadConcernSupportResult {
      * given read concern and permits the default cluster-wide read concern to be applied.
      */
     static ReadConcernSupportResult allSupportedAndDefaultPermitted() {
-        return {Status::OK(), Status::OK()};
+        return {Status::OK(), Status::OK(), Status::OK()};
     }
 
     /**
@@ -59,13 +59,30 @@ struct ReadConcernSupportResult {
      */
     Status defaultReadConcernPermit;
 
+    /*
+     * Whether this permits the implicit default readConcern to be applied (and if not, why not).
+     */
+    Status implicitDefaultReadConcernPermit;
+
     /**
      * Construct with the given Statuses, or default to Status::OK if omitted.
      */
     ReadConcernSupportResult(boost::optional<Status> readConcernStatus,
                              boost::optional<Status> defaultReadConcernStatus)
         : readConcernSupport(readConcernStatus.value_or(Status::OK())),
-          defaultReadConcernPermit(defaultReadConcernStatus.value_or(Status::OK())) {}
+          defaultReadConcernPermit(defaultReadConcernStatus.value_or(Status::OK())),
+          implicitDefaultReadConcernPermit(Status::OK()) {}
+
+    /**
+     * Construct with the given Statuses, or default to Status::OK if omitted.
+     */
+    ReadConcernSupportResult(boost::optional<Status> readConcernStatus,
+                             boost::optional<Status> defaultReadConcernStatus,
+                             boost::optional<Status> implicitDefaultReadConcernStatus)
+        : readConcernSupport(readConcernStatus.value_or(Status::OK())),
+          defaultReadConcernPermit(defaultReadConcernStatus.value_or(Status::OK())),
+          implicitDefaultReadConcernPermit(
+              implicitDefaultReadConcernStatus.value_or(Status::OK())) {}
 
     /**
      * Combine the contents of another ReadConcernSupportResult with this one. The outcome is that,
diff --git a/src/mongo/db/read_write_concern_defaults.cpp b/src/mongo/db/read_write_concern_defaults.cpp
index c982ef78bda..debf69a43f6 100644
--- a/src/mongo/db/read_write_concern_defaults.cpp
+++ b/src/mongo/db/read_write_concern_defaults.cpp
@@ -226,6 +226,16 @@ void ReadWriteConcernDefaults::refreshIfNecessary(OperationContext* opCtx) {
     }
 }
 
+repl::ReadConcernArgs ReadWriteConcernDefaults::getImplicitDefaultReadConcern() {
+    const bool isDefaultRCLocalFeatureFlagEnabled =
+        serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+        repl::feature_flags::gDefaultRCLocal.isEnabled(serverGlobalParams.featureCompatibility);
+    if (!isDefaultRCLocalFeatureFlagEnabled) {
+        return repl::ReadConcernArgs();
+    }
+    return repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern);
+}
+
 boost::optional<ReadWriteConcernDefaults::RWConcernDefaultAndTime>
 ReadWriteConcernDefaults::_getDefaultCWRWCFromDisk(OperationContext* opCtx) {
     auto defaultsHandle = _defaults.acquire(opCtx, Type::kReadWriteConcernEntry);
@@ -243,6 +253,22 @@ ReadWriteConcernDefaults::_getDefaultCWRWCFromDisk(OperationContext* opCtx) {
 ReadWriteConcernDefaults::RWConcernDefaultAndTime ReadWriteConcernDefaults::getDefault(
     OperationContext* opCtx) {
     auto cached = _getDefaultCWRWCFromDisk(opCtx).value_or(RWConcernDefaultAndTime());
+
+    const bool isDefaultRCLocalFeatureFlagEnabled =
+        serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+        repl::feature_flags::gDefaultRCLocal.isEnabled(serverGlobalParams.featureCompatibility);
+
+    // Only overwrite the default read concern and its source if it has already been set on mongos.
+    if (isDefaultRCLocalFeatureFlagEnabled && !cached.getDefaultReadConcernSource()) {
+        if (!cached.getDefaultReadConcern() || cached.getDefaultReadConcern().get().isEmpty()) {
+            auto rcDefault = getImplicitDefaultReadConcern();
+            cached.setDefaultReadConcern(rcDefault);
+            cached.setDefaultReadConcernSource(DefaultReadConcernSourceEnum::kImplicit);
+        } else {
+            cached.setDefaultReadConcernSource(DefaultReadConcernSourceEnum::kGlobal);
+        }
+    }
+
     const bool isDefaultWCMajorityFeatureFlagEnabled =
         serverGlobalParams.featureCompatibility.isVersionInitialized() &&
         repl::feature_flags::gDefaultWCMajority.isEnabled(serverGlobalParams.featureCompatibility);
diff --git a/src/mongo/db/read_write_concern_defaults.h b/src/mongo/db/read_write_concern_defaults.h
index 1ce216c2b80..cf061d9122d 100644
--- a/src/mongo/db/read_write_concern_defaults.h
+++ b/src/mongo/db/read_write_concern_defaults.h
@@ -76,6 +76,11 @@ public:
     boost::optional<ReadConcern> getDefaultReadConcern(OperationContext* opCtx);
     boost::optional<WriteConcern> getDefaultWriteConcern(OperationContext* opCtx);
 
+    /**
+     * Returns the implicit default read concern.
+     */
+    repl::ReadConcernArgs getImplicitDefaultReadConcern();
+
     class RWConcernDefaultAndTime : public RWConcernDefault {
     public:
         RWConcernDefaultAndTime() = default;
diff --git a/src/mongo/db/read_write_concern_defaults.idl b/src/mongo/db/read_write_concern_defaults.idl
index c280d1bb0b8..ceb864d272a 100644
--- a/src/mongo/db/read_write_concern_defaults.idl
+++ b/src/mongo/db/read_write_concern_defaults.idl
@@ -42,6 +42,14 @@ enums:
             kImplicit: "implicit"
             # The default write concern was set globally through setDefaultRWConcern
             kGlobal: "global"
+    DefaultReadConcernSource:
+        description: "The source of the default read concern"
+        type: string
+        values:
+            # The default read concern was set implicitly by the server.
+            kImplicit: "implicit"
+            # The default read concern was set globally through setDefaultRWConcern.
+            kGlobal: "global"
 
 structs:
     RWConcernDefault:
@@ -72,3 +80,7 @@ structs:
                 description: "The source of the default write concern."
                 type: DefaultWriteConcernSource
                 optional: true
+            defaultReadConcernSource:
+                description: "The source of the default read concern"
+                type: DefaultReadConcernSource
+                optional: true
diff --git a/src/mongo/db/read_write_concern_defaults_test.cpp b/src/mongo/db/read_write_concern_defaults_test.cpp
index 73009587628..1d28f29ab57 100644
--- a/src/mongo/db/read_write_concern_defaults_test.cpp
+++ b/src/mongo/db/read_write_concern_defaults_test.cpp
@@ -64,6 +64,10 @@ protected:
         serverGlobalParams.featureCompatibility.isVersionInitialized() &&
         repl::feature_flags::gDefaultWCMajority.isEnabled(serverGlobalParams.featureCompatibility)};
 
+    bool _isDefaultRCLocalEnabled{
+        serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+        repl::feature_flags::gDefaultRCLocal.isEnabled(serverGlobalParams.featureCompatibility)};
+
     ServiceContext::UniqueOperationContext _opCtxHolder{makeOperationContext()};
     OperationContext* const _opCtx{_opCtxHolder.get()};
 };
@@ -74,7 +78,17 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithAbsentCWRWCWithImplicitWC
     // By not calling _lookupMock.setLookupCallReturnValue(), tests _defaults.lookup() returning
     // boost::none.
     auto defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
+
     ASSERT(!defaults.getDefaultWriteConcern());
     ASSERT(!isCWWCSet());
     ASSERT(!defaults.getUpdateOpTime());
@@ -96,7 +110,17 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithAbsentCWRWCWithImplicitWC
     // boost::none.
     ASSERT(!isCWWCSet());
     auto defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
+
     ASSERT(defaults.getDefaultWriteConcern());
     ASSERT_EQ(WriteConcernOptions::kMajority, defaults.getDefaultWriteConcern().get().wMode);
     ASSERT(!defaults.getUpdateOpTime());
@@ -113,7 +137,17 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithCWRWCNeverSetWithImplicit
 
     ASSERT(!isCWWCSet());
     auto defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
+
     ASSERT(!defaults.getDefaultWriteConcern());
     ASSERT(!defaults.getUpdateOpTime());
     ASSERT(!defaults.getUpdateWallClockTime());
@@ -134,7 +168,17 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithCWRWCNeverSetWithImplicit
     ASSERT(!isCWWCSet());
     _lookupMock.setLookupCallReturnValue({});
     auto defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
+
     ASSERT(defaults.getDefaultWriteConcern());
     ASSERT_EQ(WriteConcernOptions::kMajority, defaults.getDefaultWriteConcern().get().wMode);
     ASSERT(!defaults.getUpdateOpTime());
@@ -153,7 +197,17 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithUnsetCWRWCWithImplicitWCW
 
     ASSERT(!isCWWCSet());
     auto defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
+
     ASSERT(!defaults.getDefaultWriteConcern());
     ASSERT_EQ(Timestamp(1, 2), *defaults.getUpdateOpTime());
     ASSERT_EQ(1234, defaults.getUpdateWallClockTime()->toMillisSinceEpoch());
@@ -177,7 +231,17 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithUnsetCWRWCWithImplicitWCM
 
     ASSERT(!isCWWCSet());
     auto defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
+
     ASSERT(defaults.getDefaultWriteConcern());
     ASSERT_EQ(WriteConcernOptions::kMajority, defaults.getDefaultWriteConcern().get().wMode);
     ASSERT_EQ(Timestamp(1, 2), *defaults.getUpdateOpTime());
@@ -208,6 +272,13 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithSetCWRWCWithImplicitWCW1)
     auto defaults = getDefault();
     ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
            repl::ReadConcernLevel::kLocalReadConcern);
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kGlobal);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
+
     ASSERT_EQ(4, defaults.getDefaultWriteConcern()->wNumNodes);
     ASSERT_EQ(Timestamp(1, 2), *defaults.getUpdateOpTime());
     ASSERT_EQ(1234, defaults.getUpdateWallClockTime()->toMillisSinceEpoch());
@@ -243,6 +314,11 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithSetCWRWCWithImplicitWCMaj
     auto defaults = getDefault();
     ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
            repl::ReadConcernLevel::kLocalReadConcern);
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kGlobal);
+    }
+
     ASSERT_EQ(4, defaults.getDefaultWriteConcern()->wNumNodes);
     ASSERT_EQ(Timestamp(1, 2), *defaults.getUpdateOpTime());
     ASSERT_EQ(1234, defaults.getUpdateWallClockTime()->toMillisSinceEpoch());
@@ -268,9 +344,19 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWriteConcernSourceImplicitWit
     _lookupMock.setLookupCallReturnValue(std::move(newDefaults));
 
     ASSERT(!isCWWCSet());
-    // The default write concern source should be set to implicit if wc.usedDefault is true
     auto defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
+
+    // The default write concern source should be set to implicit if wc.usedDefault is true
     ASSERT_EQ(0, defaults.getDefaultWriteConcern()->wNumNodes);
     ASSERT_EQ(WriteConcernOptions::kMajority, defaults.getDefaultWriteConcern().get().wMode);
     ASSERT_EQ(Timestamp(1, 2), *defaults.getUpdateOpTime());
@@ -285,7 +371,7 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithSetAndUnSetCWRCWithImplic
     // Setting only default read concern.
     RWConcernDefault newDefaults;
     newDefaults.setDefaultReadConcern(
-        repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern));
+        repl::ReadConcernArgs(repl::ReadConcernLevel::kAvailableReadConcern));
     newDefaults.setUpdateOpTime(Timestamp(1, 2));
     newDefaults.setUpdateWallClockTime(Date_t::fromMillisSinceEpoch(1234));
     _lookupMock.setLookupCallReturnValue(std::move(newDefaults));
@@ -293,7 +379,11 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithSetAndUnSetCWRCWithImplic
     ASSERT(!isCWWCSet());
     auto defaults = getDefault();
     ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
-           repl::ReadConcernLevel::kLocalReadConcern);
+           repl::ReadConcernLevel::kAvailableReadConcern);
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kGlobal);
+    }
     ASSERT(!defaults.getDefaultWriteConcern());
     ASSERT_EQ(Timestamp(1, 2), *defaults.getUpdateOpTime());
     ASSERT_EQ(1234, defaults.getUpdateWallClockTime()->toMillisSinceEpoch());
@@ -312,7 +402,17 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithSetAndUnSetCWRCWithImplic
 
     ASSERT(!isCWWCSet());
     defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
+
     ASSERT(!defaults.getDefaultWriteConcern());
     ASSERT_EQ(Timestamp(1, 3), *defaults.getUpdateOpTime());
     ASSERT_EQ(1234, defaults.getUpdateWallClockTime()->toMillisSinceEpoch());
@@ -341,6 +441,10 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithSetAndUnSetCWRCWithImplic
     auto defaults = getDefault();
     ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
            repl::ReadConcernLevel::kLocalReadConcern);
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kGlobal);
+    }
     ASSERT(defaults.getDefaultWriteConcern());
     ASSERT_EQ(WriteConcernOptions::kMajority, defaults.getDefaultWriteConcern().get().wMode);
     ASSERT_EQ(Timestamp(1, 2), *defaults.getUpdateOpTime());
@@ -358,8 +462,17 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithSetAndUnSetCWRCWithImplic
 
     ASSERT(!isCWWCSet());
     defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
-    ASSERT(defaults.getDefaultWriteConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
+
     ASSERT_EQ(WriteConcernOptions::kMajority, defaults.getDefaultWriteConcern().get().wMode);
     ASSERT_EQ(Timestamp(1, 3), *defaults.getUpdateOpTime());
     ASSERT_EQ(1234, defaults.getUpdateWallClockTime()->toMillisSinceEpoch());
@@ -385,7 +498,16 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithoutInvalidateDoesNotCallL
 
     ASSERT(!isCWWCSet());
     auto defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
     ASSERT(!defaults.getDefaultWriteConcern());
     ASSERT_EQ(Timestamp(1, 2), *defaults.getUpdateOpTime());
     ASSERT_EQ(1234, defaults.getUpdateWallClockTime()->toMillisSinceEpoch());
@@ -396,7 +518,7 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithoutInvalidateDoesNotCallL
 
     RWConcernDefault newDefaults2;
     newDefaults2.setDefaultReadConcern(
-        repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern));
+        repl::ReadConcernArgs(repl::ReadConcernLevel::kAvailableReadConcern));
     WriteConcernOptions wc;
     wc.wNumNodes = 4;
     wc.usedDefault = false;
@@ -411,7 +533,16 @@ TEST_F(ReadWriteConcernDefaultsTest, TestGetDefaultWithoutInvalidateDoesNotCallL
 
     ASSERT(!isCWWCSet());
     auto defaults2 = getDefault();
-    ASSERT(!defaults2.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults2.getDefaultReadConcern());
+        ASSERT(defaults2.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults2.getDefaultReadConcernSource());
+        ASSERT(defaults2.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults2.getDefaultReadConcern());
+        ASSERT(!defaults2.getDefaultReadConcernSource());
+    }
     ASSERT(!defaults2.getDefaultWriteConcern());
     ASSERT_EQ(Timestamp(1, 2), *defaults2.getUpdateOpTime());
     ASSERT_EQ(1234, defaults2.getUpdateWallClockTime()->toMillisSinceEpoch());
@@ -432,7 +563,16 @@ TEST_F(ReadWriteConcernDefaultsTest, TestInvalidate) {
 
     ASSERT(!isCWWCSet());
     auto defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
     ASSERT(!defaults.getDefaultWriteConcern());
     ASSERT_EQ(Timestamp(1, 2), *defaults.getUpdateOpTime());
     ASSERT_EQ(1234, defaults.getUpdateWallClockTime()->toMillisSinceEpoch());
@@ -443,7 +583,7 @@ TEST_F(ReadWriteConcernDefaultsTest, TestInvalidate) {
 
     RWConcernDefault newDefaults2;
     newDefaults2.setDefaultReadConcern(
-        repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern));
+        repl::ReadConcernArgs(repl::ReadConcernLevel::kAvailableReadConcern));
     WriteConcernOptions wc;
     wc.wNumNodes = 4;
     wc.usedDefault = false;
@@ -460,7 +600,11 @@ TEST_F(ReadWriteConcernDefaultsTest, TestInvalidate) {
     ASSERT(isCWWCSet());
     auto defaults2 = getDefault();
     ASSERT(defaults2.getDefaultReadConcern()->getLevel() ==
-           repl::ReadConcernLevel::kLocalReadConcern);
+           repl::ReadConcernLevel::kAvailableReadConcern);
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults2.getDefaultReadConcernSource());
+        ASSERT(defaults2.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kGlobal);
+    }
     ASSERT_EQ(4, defaults2.getDefaultWriteConcern()->wNumNodes);
     ASSERT_EQ(Timestamp(3, 4), *defaults2.getUpdateOpTime());
     ASSERT_EQ(5678, defaults2.getUpdateWallClockTime()->toMillisSinceEpoch());
@@ -476,7 +620,17 @@ TEST_F(ReadWriteConcernDefaultsTest, TestRefreshDefaultsWithEmptyCacheAndAbsentD
 
     ASSERT(!isCWWCSet());
     auto defaults = getDefault();
-    ASSERT(!defaults.getDefaultReadConcern());
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(defaults.getDefaultReadConcern());
+        ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(defaults.getDefaultReadConcernSource());
+        ASSERT(defaults.getDefaultReadConcernSource() == DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+    }
+
     ASSERT(!defaults.getDefaultWriteConcern());
     ASSERT(!defaults.getUpdateOpTime());
     ASSERT(!defaults.getUpdateWallClockTime());
@@ -583,10 +737,16 @@ protected:
     auto setupOldDefaults() {
         auto defaults = _rwcd.generateNewCWRWCToBeSavedOnDisk(
             operationContext(),
-            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kAvailableReadConcern),
             uassertStatusOK(WriteConcernOptions::parse(BSON("w" << 4))));
+
         ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
-               repl::ReadConcernLevel::kLocalReadConcern);
+               repl::ReadConcernLevel::kAvailableReadConcern);
+        if (_isDefaultRCLocalEnabled) {
+            // default read concern source is not saved on disk.
+            ASSERT(!defaults.getDefaultReadConcernSource());
+        }
+
         ASSERT_EQ(4, defaults.getDefaultWriteConcern()->wNumNodes);
         ASSERT(defaults.getUpdateOpTime());
         ASSERT(defaults.getUpdateWallClockTime());
@@ -614,6 +774,10 @@ protected:
     bool _isDefaultWCMajorityEnabled{
         serverGlobalParams.featureCompatibility.isVersionInitialized() &&
         repl::feature_flags::gDefaultWCMajority.isEnabled(serverGlobalParams.featureCompatibility)};
+
+    bool _isDefaultRCLocalEnabled{
+        serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+        repl::feature_flags::gDefaultRCLocal.isEnabled(serverGlobalParams.featureCompatibility)};
 };
 
 TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
@@ -690,6 +854,8 @@ TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
         uassertStatusOK(WriteConcernOptions::parse(BSON("w" << 5))));
     ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
            repl::ReadConcernLevel::kMajorityReadConcern);
+    ASSERT(!defaults.getDefaultReadConcernSource());
+
     ASSERT_EQ(5, defaults.getDefaultWriteConcern()->wNumNodes);
     ASSERT_LT(*oldDefaults.getUpdateOpTime(), *defaults.getUpdateOpTime());
     ASSERT_LT(*oldDefaults.getUpdateWallClockTime(), *defaults.getUpdateWallClockTime());
@@ -716,6 +882,8 @@ TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
         boost::none);
     ASSERT(defaults.getDefaultReadConcern()->getLevel() ==
            repl::ReadConcernLevel::kMajorityReadConcern);
+    ASSERT(!defaults.getDefaultReadConcernSource());
+
     ASSERT_EQ(oldDefaults.getDefaultWriteConcern()->wNumNodes,
               defaults.getDefaultWriteConcern()->wNumNodes);
     ASSERT_LT(*oldDefaults.getUpdateOpTime(), *defaults.getUpdateOpTime());
@@ -765,6 +933,8 @@ TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
         uassertStatusOK(WriteConcernOptions::parse(BSON("w" << 5))));
     ASSERT(oldDefaults.getDefaultReadConcern()->getLevel() ==
            defaults.getDefaultReadConcern()->getLevel());
+    ASSERT(!defaults.getDefaultReadConcernSource());
+
     ASSERT_EQ(5, defaults.getDefaultWriteConcern()->wNumNodes);
     ASSERT_LT(*oldDefaults.getUpdateOpTime(), *defaults.getUpdateOpTime());
     ASSERT_LT(*oldDefaults.getUpdateWallClockTime(), *defaults.getUpdateWallClockTime());
@@ -834,7 +1004,10 @@ TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
     auto oldDefaults = setupOldDefaults();
     auto defaults = _rwcd.generateNewCWRWCToBeSavedOnDisk(
         operationContext(), repl::ReadConcernArgs(), boost::none);
+    // Assert that the on disk version does not have a read/write concern set.
     ASSERT(!defaults.getDefaultReadConcern());
+    ASSERT(!defaults.getDefaultReadConcernSource());
+
     ASSERT(defaults.getDefaultWriteConcern());
     ASSERT_EQ(oldDefaults.getDefaultWriteConcern()->wNumNodes,
               defaults.getDefaultWriteConcern()->wNumNodes);
@@ -852,6 +1025,19 @@ TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
         ASSERT(newDefaults.getDefaultWriteConcernSource() ==
                DefaultWriteConcernSourceEnum::kGlobal);
     }
+
+    // Test that the implicit default read concern is still used after read concern is unset.
+    if (_isDefaultRCLocalEnabled) {
+        ASSERT(newDefaults.getDefaultReadConcern());
+        ASSERT(newDefaults.getDefaultReadConcern()->getLevel() ==
+               repl::ReadConcernLevel::kLocalReadConcern);
+        ASSERT(newDefaults.getDefaultReadConcernSource());
+        ASSERT(newDefaults.getDefaultReadConcernSource() ==
+               DefaultReadConcernSourceEnum::kImplicit);
+    } else {
+        ASSERT(!newDefaults.getDefaultReadConcern());
+        ASSERT(!newDefaults.getDefaultReadConcernSource());
+    }
 }
 
 TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
@@ -869,6 +1055,7 @@ TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
         ASSERT(defaults.getDefaultReadConcern());
         ASSERT(oldDefaults.getDefaultReadConcern()->getLevel() ==
                defaults.getDefaultReadConcern()->getLevel());
+        ASSERT(!defaults.getDefaultReadConcernSource());
         ASSERT(!defaults.getDefaultWriteConcern());
         ASSERT_LT(*oldDefaults.getUpdateOpTime(), *defaults.getUpdateOpTime());
         ASSERT_LT(*oldDefaults.getUpdateWallClockTime(), *defaults.getUpdateWallClockTime());
@@ -911,7 +1098,10 @@ TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
     } else {
         auto defaults = _rwcd.generateNewCWRWCToBeSavedOnDisk(
             operationContext(), repl::ReadConcernArgs(), WriteConcernOptions());
+        // Assert that the on disk version does not have a read/write concern set.
         ASSERT(!defaults.getDefaultReadConcern());
+        ASSERT(!defaults.getDefaultReadConcernSource());
+
         ASSERT(!defaults.getDefaultWriteConcern());
         ASSERT_LT(*oldDefaults.getUpdateOpTime(), *defaults.getUpdateOpTime());
         ASSERT_LT(*oldDefaults.getUpdateWallClockTime(), *defaults.getUpdateWallClockTime());
@@ -920,6 +1110,18 @@ TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
         _rwcd.refreshIfNecessary(operationContext());
         auto newDefaults = _rwcd.getDefault(operationContext());
         ASSERT_LT(oldDefaults.localUpdateWallClockTime(), newDefaults.localUpdateWallClockTime());
+
+        if (_isDefaultRCLocalEnabled) {
+            ASSERT(newDefaults.getDefaultReadConcern());
+            ASSERT(newDefaults.getDefaultReadConcern()->getLevel() ==
+                   repl::ReadConcernLevel::kLocalReadConcern);
+            ASSERT(newDefaults.getDefaultReadConcernSource());
+            ASSERT(newDefaults.getDefaultReadConcernSource() ==
+                   DefaultReadConcernSourceEnum::kImplicit);
+        } else {
+            ASSERT(!newDefaults.getDefaultReadConcern());
+            ASSERT(!newDefaults.getDefaultReadConcernSource());
+        }
     }
 }
 
@@ -932,6 +1134,7 @@ TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
         uassertStatusOK(WriteConcernOptions::parse(BSON("j" << true))));
     ASSERT(oldDefaults.getDefaultReadConcern()->getLevel() ==
            defaults.getDefaultReadConcern()->getLevel());
+    ASSERT(!defaults.getDefaultReadConcernSource());
     ASSERT_EQ(1, defaults.getDefaultWriteConcern()->wNumNodes);
     ASSERT_EQ(0, defaults.getDefaultWriteConcern()->wTimeout);
     ASSERT(WriteConcernOptions::SyncMode::JOURNAL == defaults.getDefaultWriteConcern()->syncMode);
@@ -960,6 +1163,7 @@ TEST_F(ReadWriteConcernDefaultsTestWithClusterTime,
         uassertStatusOK(WriteConcernOptions::parse(BSON("wtimeout" << 12345))));
     ASSERT(oldDefaults.getDefaultReadConcern()->getLevel() ==
            defaults.getDefaultReadConcern()->getLevel());
+    ASSERT(!defaults.getDefaultReadConcernSource());
     ASSERT_EQ(1, defaults.getDefaultWriteConcern()->wNumNodes);
     ASSERT_EQ(12345, defaults.getDefaultWriteConcern()->wTimeout);
     ASSERT(WriteConcernOptions::SyncMode::UNSET == defaults.getDefaultWriteConcern()->syncMode);
diff --git a/src/mongo/db/read_write_concern_provenance.h b/src/mongo/db/read_write_concern_provenance.h
index 0679b311821..51e49121c78 100644
--- a/src/mongo/db/read_write_concern_provenance.h
+++ b/src/mongo/db/read_write_concern_provenance.h
@@ -116,6 +116,13 @@ public:
     }
 
     /**
+     * Returns true if the RWC was an implicit default.
+     */
+    const bool isImplicitDefault() const {
+        return hasSource() && *getSource() == Source::implicitDefault;
+    }
+
+    /**
      * Sets the source of this provenance.  In order to prevent accidental clobbering of provenance
      * with incorrect values, a source cannot change during the provenance's lifetime, except for
      * the initial transition from kUnset to some other Source value.
diff --git a/src/mongo/db/repl/read_concern_args.cpp b/src/mongo/db/repl/read_concern_args.cpp
index a95cf27880f..a02b5e9b67b 100644
--- a/src/mongo/db/repl/read_concern_args.cpp
+++ b/src/mongo/db/repl/read_concern_args.cpp
@@ -101,6 +101,10 @@ bool ReadConcernArgs::isSpecified() const {
     return _specified;
 }
 
+bool ReadConcernArgs::isImplicitDefault() const {
+    return getProvenance().isImplicitDefault();
+}
+
 ReadConcernLevel ReadConcernArgs::getLevel() const {
     return _level.value_or(ReadConcernLevel::kLocalReadConcern);
 }
diff --git a/src/mongo/db/repl/read_concern_args.h b/src/mongo/db/repl/read_concern_args.h
index 2bc1b5bdee4..15b1489aedd 100644
--- a/src/mongo/db/repl/read_concern_args.h
+++ b/src/mongo/db/repl/read_concern_args.h
@@ -153,6 +153,11 @@ public:
     bool isSpecified() const;
 
     /**
+     * Returns true if this ReadConcernArgs represents an implicit default read concern.
+     */
+    bool isImplicitDefault() const;
+
+    /**
      *  Returns default kLocalReadConcern if _level is not set.
      */
     ReadConcernLevel getLevel() const;
diff --git a/src/mongo/db/repl/repl_server_parameters.idl b/src/mongo/db/repl/repl_server_parameters.idl
index 33b01923304..40be075770f 100644
--- a/src/mongo/db/repl/repl_server_parameters.idl
+++ b/src/mongo/db/repl/repl_server_parameters.idl
@@ -551,3 +551,11 @@ feature_flags:
             document images for retryable find and modifies.
         cpp_varname: feature_flags::gFeatureFlagRetryableFindAndModify
         default: false
+
+    featureFlagDefaultReadConcernLocal:
+        description: >-
+            When enabled, default read concern will be set to local if there is no cluster wide
+            read concern (CWRC).
+        cpp_varname: feature_flags::gDefaultRCLocal
+        default: true
+        version: 5.0
\ No newline at end of file
diff --git a/src/mongo/db/repl/replication_info.cpp b/src/mongo/db/repl/replication_info.cpp
index 77ebda6ca44..48f2682d505 100644
--- a/src/mongo/db/repl/replication_info.cpp
+++ b/src/mongo/db/repl/replication_info.cpp
@@ -268,6 +268,21 @@ public:
         return false;
     }
 
+    ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
+                                                 repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault) const final {
+        static const Status kReadConcernNotSupported{ErrorCodes::InvalidOptions,
+                                                     "read concern not supported"};
+        static const Status kDefaultReadConcernNotPermitted{
+            ErrorCodes::InvalidOptions, "cluster wide default read concern not permitted"};
+
+        static const Status kImplicitDefaultReadConcernNotPermitted{
+            ErrorCodes::InvalidOptions, "implicit default read concern not permitted"};
+        return {{level != repl::ReadConcernLevel::kLocalReadConcern, kReadConcernNotSupported},
+                {kDefaultReadConcernNotPermitted},
+                {kImplicitDefaultReadConcernNotPermitted}};
+    }
+
     void addRequiredPrivileges(const std::string& dbname,
                                const BSONObj& cmdObj,
                                std::vector<Privilege>* out) const final {}  // No auth required
diff --git a/src/mongo/db/service_entry_point_common.cpp b/src/mongo/db/service_entry_point_common.cpp
index a7fb9c60ce3..b5c8bfd4d83 100644
--- a/src/mongo/db/service_entry_point_common.cpp
+++ b/src/mongo/db/service_entry_point_common.cpp
@@ -69,6 +69,7 @@
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/repl/read_concern_args.h"
 #include "mongo/db/repl/repl_client_info.h"
+#include "mongo/db/repl/repl_server_parameters_gen.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl/speculative_majority_read_info.h"
 #include "mongo/db/repl/storage_interface.h"
@@ -358,12 +359,28 @@ StatusWith<repl::ReadConcernArgs> _extractReadConcern(OperationContext* opCtx,
 
     bool clientSuppliedReadConcern = readConcernArgs.isSpecified();
     bool customDefaultWasApplied = false;
-    auto readConcernSupport = invocation->supportsReadConcern(readConcernArgs.getLevel());
-    if (readConcernSupport.defaultReadConcernPermit.isOK() &&
-        (startTransaction || !opCtx->inMultiDocumentTransaction()) &&
+    auto readConcernSupport = invocation->supportsReadConcern(readConcernArgs.getLevel(),
+                                                              readConcernArgs.isImplicitDefault());
+
+    auto applyDefaultReadConcern = [&](const repl::ReadConcernArgs rcDefault) -> void {
+        LOGV2_DEBUG(21955,
+                    2,
+                    "Applying default readConcern on {command} of {readConcernDefault} "
+                    "on {command}",
+                    "Applying default readConcern on command",
+                    "readConcernDefault"_attr = rcDefault,
+                    "command"_attr = invocation->definition()->getName());
+        readConcernArgs = std::move(rcDefault);
+        // Update the readConcernSupport, since the default RC was applied.
+        readConcernSupport =
+            invocation->supportsReadConcern(readConcernArgs.getLevel(), !customDefaultWasApplied);
+    };
+
+    auto shouldApplyDefaults = (startTransaction || !opCtx->inMultiDocumentTransaction()) &&
         repl::ReplicationCoordinator::get(opCtx)->isReplEnabled() &&
-        !opCtx->getClient()->isInDirectClient()) {
+        !opCtx->getClient()->isInDirectClient();
 
+    if (readConcernSupport.defaultReadConcernPermit.isOK() && shouldApplyDefaults) {
         if (isInternalClient) {
             // ReadConcern should always be explicitly specified by operations received from
             // internal clients (ie. from a mongos or mongod), even if it is empty (ie.
@@ -386,26 +403,35 @@ StatusWith<repl::ReadConcernArgs> _extractReadConcern(OperationContext* opCtx,
             // 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(opCtx);
+                const auto rwcDefaults =
+                    ReadWriteConcernDefaults::get(opCtx->getServiceContext()).getDefault(opCtx);
+                const auto rcDefault = rwcDefaults.getDefaultReadConcern();
                 if (rcDefault) {
-                    customDefaultWasApplied = true;
-                    readConcernArgs = std::move(*rcDefault);
-                    LOGV2_DEBUG(21955,
-                                2,
-                                "Applying default readConcern on {command} of {readConcernDefault} "
-                                "on {command}",
-                                "Applying default readConcern on command",
-                                "readConcernDefault"_attr = *rcDefault,
-                                "command"_attr = invocation->definition()->getName());
-                    // Update the readConcernSupport, since the default RC was applied.
-                    readConcernSupport =
-                        invocation->supportsReadConcern(readConcernArgs.getLevel());
+                    const bool isDefaultRCLocalFeatureFlagEnabled =
+                        serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+                        repl::feature_flags::gDefaultRCLocal.isEnabled(
+                            serverGlobalParams.featureCompatibility);
+                    const auto readConcernSource = rwcDefaults.getDefaultReadConcernSource();
+                    customDefaultWasApplied = !isDefaultRCLocalFeatureFlagEnabled ||
+                        (readConcernSource &&
+                         readConcernSource.get() == DefaultReadConcernSourceEnum::kGlobal);
+
+                    applyDefaultReadConcern(*rcDefault);
                 }
             }
         }
     }
 
+    // Apply the implicit default read concern even if the command does not support a cluster wide
+    // read concern.
+    if (!readConcernSupport.defaultReadConcernPermit.isOK() &&
+        readConcernSupport.implicitDefaultReadConcernPermit.isOK() && shouldApplyDefaults &&
+        !isInternalClient && readConcernArgs.isEmpty()) {
+        auto rcDefault = ReadWriteConcernDefaults::get(opCtx->getServiceContext())
+                             .getImplicitDefaultReadConcern();
+        applyDefaultReadConcern(rcDefault);
+    }
+
     // It's fine for clients to provide any provenance value to mongod. But if they haven't, then an
     // appropriate provenance needs to be determined.
     auto& provenance = readConcernArgs.getProvenance();
@@ -987,7 +1013,8 @@ void CheckoutSessionAndInvokeCommand::_checkOutSession() {
         // `createIndexes` do not support readConcern inside transactions.
         // TODO(SERVER-46971): Consider how to extend this check to other commands.
         auto cmdName = command->getName();
-        auto readConcernSupport = invocation->supportsReadConcern(readConcernArgs.getLevel());
+        auto readConcernSupport = invocation->supportsReadConcern(
+            readConcernArgs.getLevel(), readConcernArgs.isImplicitDefault());
         if (readConcernArgs.hasLevel() &&
             (cmdName == "create"_sd || cmdName == "createIndexes"_sd)) {
             if (!readConcernSupport.readConcernSupport.isOK()) {
diff --git a/src/mongo/s/cluster_commands_helpers.cpp b/src/mongo/s/cluster_commands_helpers.cpp
index e23f1d63c89..8dfb3f20b4c 100644
--- a/src/mongo/s/cluster_commands_helpers.cpp
+++ b/src/mongo/s/cluster_commands_helpers.cpp
@@ -335,7 +335,8 @@ BSONObj applyReadWriteConcern(OperationContext* opCtx,
                               CommandInvocation* invocation,
                               const BSONObj& cmdObj) {
     const auto& readConcernArgs = repl::ReadConcernArgs::get(opCtx);
-    const auto readConcernSupport = invocation->supportsReadConcern(readConcernArgs.getLevel());
+    const auto readConcernSupport = invocation->supportsReadConcern(
+        readConcernArgs.getLevel(), readConcernArgs.isImplicitDefault());
     return applyReadWriteConcern(opCtx,
                                  readConcernSupport.readConcernSupport.isOK(),
                                  invocation->supportsWriteConcern(),
@@ -346,7 +347,8 @@ BSONObj applyReadWriteConcern(OperationContext* opCtx,
                               BasicCommandWithReplyBuilderInterface* cmd,
                               const BSONObj& cmdObj) {
     const auto& readConcernArgs = repl::ReadConcernArgs::get(opCtx);
-    const auto readConcernSupport = cmd->supportsReadConcern(cmdObj, readConcernArgs.getLevel());
+    const auto readConcernSupport = cmd->supportsReadConcern(
+        cmdObj, readConcernArgs.getLevel(), readConcernArgs.isImplicitDefault());
     return applyReadWriteConcern(opCtx,
                                  readConcernSupport.readConcernSupport.isOK(),
                                  cmd->supportsWriteConcern(cmdObj),
diff --git a/src/mongo/s/commands/SConscript b/src/mongo/s/commands/SConscript
index 8a109f999f4..d56a8ae52d3 100644
--- a/src/mongo/s/commands/SConscript
+++ b/src/mongo/s/commands/SConscript
@@ -141,6 +141,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/read_write_concern_defaults',
         '$BUILD_DIR/mongo/db/repl/hello_auth',
         '$BUILD_DIR/mongo/db/repl/hello_command',
+        '$BUILD_DIR/mongo/db/repl/repl_server_parameters',
         '$BUILD_DIR/mongo/db/shared_request_handling',
         '$BUILD_DIR/mongo/db/stats/api_version_metrics',
         '$BUILD_DIR/mongo/db/stats/counters',
@@ -180,6 +181,7 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/db/auth/authmocks',
         '$BUILD_DIR/mongo/db/auth/saslauth',
         '$BUILD_DIR/mongo/db/pipeline/process_interface/mongos_process_interface_factory',
+        '$BUILD_DIR/mongo/db/read_write_concern_defaults_mock',
         '$BUILD_DIR/mongo/s/query/cluster_aggregate',
         '$BUILD_DIR/mongo/s/sharding_router_test_fixture',
         'cluster_commands',
diff --git a/src/mongo/s/commands/cluster_abort_transaction_cmd.cpp b/src/mongo/s/commands/cluster_abort_transaction_cmd.cpp
index 0aaaa722157..b2581fc59eb 100644
--- a/src/mongo/s/commands/cluster_abort_transaction_cmd.cpp
+++ b/src/mongo/s/commands/cluster_abort_transaction_cmd.cpp
@@ -81,7 +81,8 @@ public:
     }
 
     ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
-                                                 repl::ReadConcernLevel level) const override {
+                                                 repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault) const override {
         // abortTransaction commences running inside a transaction (even though the transaction will
         // be ended by the time it completes).  Therefore it needs to accept any readConcern which
         // is valid within a transaction.  However it is not appropriate to apply the default
diff --git a/src/mongo/s/commands/cluster_command_test_fixture.cpp b/src/mongo/s/commands/cluster_command_test_fixture.cpp
index b2a0131d3a5..c7573969bd7 100644
--- a/src/mongo/s/commands/cluster_command_test_fixture.cpp
+++ b/src/mongo/s/commands/cluster_command_test_fixture.cpp
@@ -39,6 +39,7 @@
 #include "mongo/db/keys_collection_manager.h"
 #include "mongo/db/logical_session_cache_noop.h"
 #include "mongo/db/logical_time_validator.h"
+#include "mongo/db/read_write_concern_defaults.h"
 #include "mongo/db/vector_clock.h"
 #include "mongo/s/cluster_last_error_info.h"
 #include "mongo/util/fail_point.h"
@@ -74,6 +75,10 @@ void ClusterCommandTestFixture::setUp() {
 
     _staleVersionAndSnapshotRetriesBlock = std::make_unique<FailPointEnableBlock>(
         "enableStaleVersionAndSnapshotRetriesWithinTransactions");
+
+    // The ReadWriteConcernDefaults decoration on the service context won't always be created,
+    // so we should manually instantiate it to ensure it exists in our tests.
+    ReadWriteConcernDefaults::create(getServiceContext(), _lookupMock.getFetchDefaultsFn());
 }
 
 BSONObj ClusterCommandTestFixture::_makeCmd(BSONObj cmdObj, bool includeAfterClusterTime) {
diff --git a/src/mongo/s/commands/cluster_command_test_fixture.h b/src/mongo/s/commands/cluster_command_test_fixture.h
index 6a2f869afe6..3f379052ff9 100644
--- a/src/mongo/s/commands/cluster_command_test_fixture.h
+++ b/src/mongo/s/commands/cluster_command_test_fixture.h
@@ -31,6 +31,7 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/db/read_write_concern_defaults_cache_lookup_mock.h"
 #include "mongo/s/catalog_cache_test_fixture.h"
 #include "mongo/s/commands/strategy.h"
 
@@ -115,6 +116,8 @@ private:
     // errors for the duration of each test.
     // TODO SERVER-39704: Remove this failpoint block.
     std::unique_ptr<FailPointEnableBlock> _staleVersionAndSnapshotRetriesBlock;
+
+    ReadWriteConcernDefaultsLookupMock _lookupMock;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/s/commands/cluster_count_cmd.cpp b/src/mongo/s/commands/cluster_count_cmd.cpp
index 3d6e8672211..d22b8279a17 100644
--- a/src/mongo/s/commands/cluster_count_cmd.cpp
+++ b/src/mongo/s/commands/cluster_count_cmd.cpp
@@ -66,7 +66,8 @@ public:
     }
 
     ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
-                                                 repl::ReadConcernLevel level) const override {
+                                                 repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault) const override {
         static const Status kSnapshotNotSupported{ErrorCodes::InvalidOptions,
                                                   "read concern snapshot not supported"};
         return {{level == repl::ReadConcernLevel::kSnapshotReadConcern, kSnapshotNotSupported},
diff --git a/src/mongo/s/commands/cluster_distinct_cmd.cpp b/src/mongo/s/commands/cluster_distinct_cmd.cpp
index 7e695d526e2..31741e57bba 100644
--- a/src/mongo/s/commands/cluster_distinct_cmd.cpp
+++ b/src/mongo/s/commands/cluster_distinct_cmd.cpp
@@ -78,7 +78,8 @@ public:
     }
 
     ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
-                                                 repl::ReadConcernLevel level) const final {
+                                                 repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault) const final {
         return ReadConcernSupportResult::allSupportedAndDefaultPermitted();
     }
 
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 df3ff3bf894..8f7ed0e2ee2 100644
--- a/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp
+++ b/src/mongo/s/commands/cluster_find_and_modify_cmd.cpp
@@ -191,7 +191,8 @@ public:
     }
 
     ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
-                                                 repl::ReadConcernLevel level) const override {
+                                                 repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault) const override {
         return {{level != repl::ReadConcernLevel::kLocalReadConcern &&
                      level != repl::ReadConcernLevel::kSnapshotReadConcern,
                  {ErrorCodes::InvalidOptions, "read concern not supported"}},
diff --git a/src/mongo/s/commands/cluster_find_cmd.cpp b/src/mongo/s/commands/cluster_find_cmd.cpp
index 7230f6b2095..797391c29fd 100644
--- a/src/mongo/s/commands/cluster_find_cmd.cpp
+++ b/src/mongo/s/commands/cluster_find_cmd.cpp
@@ -129,7 +129,8 @@ public:
             return false;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const final {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const final {
             return ReadConcernSupportResult::allSupportedAndDefaultPermitted();
         }
 
diff --git a/src/mongo/s/commands/cluster_getmore_cmd.cpp b/src/mongo/s/commands/cluster_getmore_cmd.cpp
index 3370ae4ab1e..7b7c9a8898b 100644
--- a/src/mongo/s/commands/cluster_getmore_cmd.cpp
+++ b/src/mongo/s/commands/cluster_getmore_cmd.cpp
@@ -84,7 +84,8 @@ public:
             return false;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const override {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const override {
             return kSupportsReadConcernResult;
         }
 
diff --git a/src/mongo/s/commands/cluster_hello_cmd.cpp b/src/mongo/s/commands/cluster_hello_cmd.cpp
index f6cc0344504..cdb05cf2c9d 100644
--- a/src/mongo/s/commands/cluster_hello_cmd.cpp
+++ b/src/mongo/s/commands/cluster_hello_cmd.cpp
@@ -78,6 +78,20 @@ public:
         return false;
     }
 
+    ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
+                                                 repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault) const final {
+        static const Status kReadConcernNotSupported{ErrorCodes::InvalidOptions,
+                                                     "read concern not supported"};
+        static const Status kDefaultReadConcernNotPermitted{
+            ErrorCodes::InvalidOptions, "cluster wide default read concern not permitted"};
+        static const Status kImplicitDefaultReadConcernNotPermitted{
+            ErrorCodes::InvalidOptions, "implicit default read concern not permitted"};
+        return {{level != repl::ReadConcernLevel::kLocalReadConcern, kReadConcernNotSupported},
+                {kDefaultReadConcernNotPermitted},
+                {kImplicitDefaultReadConcernNotPermitted}};
+    }
+
     AllowedOnSecondary secondaryAllowed(ServiceContext*) const final {
         return AllowedOnSecondary::kAlways;
     }
diff --git a/src/mongo/s/commands/cluster_pipeline_cmd.cpp b/src/mongo/s/commands/cluster_pipeline_cmd.cpp
index ba0fba16362..87dbb0e05c1 100644
--- a/src/mongo/s/commands/cluster_pipeline_cmd.cpp
+++ b/src/mongo/s/commands/cluster_pipeline_cmd.cpp
@@ -108,9 +108,11 @@ public:
             return true;
         }
 
-        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const override {
+        ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                     bool isImplicitDefault) const override {
             return _liteParsedPipeline.supportsReadConcern(
                 level,
+                isImplicitDefault,
                 _aggregationRequest.getExplain(),
                 serverGlobalParams.enableMajorityReadConcern);
         }
diff --git a/src/mongo/s/commands/cluster_write_cmd.cpp b/src/mongo/s/commands/cluster_write_cmd.cpp
index 04d9961217d..331c6d78024 100644
--- a/src/mongo/s/commands/cluster_write_cmd.cpp
+++ b/src/mongo/s/commands/cluster_write_cmd.cpp
@@ -613,7 +613,8 @@ private:
         return true;
     }
 
-    ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level) const final {
+    ReadConcernSupportResult supportsReadConcern(repl::ReadConcernLevel level,
+                                                 bool isImplicitDefault) const final {
         return ReadConcernSupportResult::allSupportedAndDefaultPermitted();
     }
 
diff --git a/src/mongo/s/commands/strategy.cpp b/src/mongo/s/commands/strategy.cpp
index e16e8c7340a..dfe5d30c981 100644
--- a/src/mongo/s/commands/strategy.cpp
+++ b/src/mongo/s/commands/strategy.cpp
@@ -60,6 +60,7 @@
 #include "mongo/db/query/getmore_command_gen.h"
 #include "mongo/db/query/query_request_helper.h"
 #include "mongo/db/read_write_concern_defaults.h"
+#include "mongo/db/repl/repl_server_parameters_gen.h"
 #include "mongo/db/stats/api_version_metrics.h"
 #include "mongo/db/stats/counters.h"
 #include "mongo/db/transaction_validation.h"
@@ -719,33 +720,57 @@ Status ParseAndRunCommand::RunInvocation::_setup() {
     bool clientSuppliedReadConcern = readConcernArgs.isSpecified();
     bool customDefaultReadConcernWasApplied = false;
 
-    auto readConcernSupport = invocation->supportsReadConcern(readConcernArgs.getLevel());
-    if (readConcernSupport.defaultReadConcernPermit.isOK() &&
-        (startTransaction || !TransactionRouter::get(opCtx))) {
+    auto readConcernSupport = invocation->supportsReadConcern(readConcernArgs.getLevel(),
+                                                              readConcernArgs.isImplicitDefault());
+
+    auto applyDefaultReadConcern = [&](const repl::ReadConcernArgs rcDefault) -> void {
+        // We must obtain the client lock to set ReadConcernArgs, because it's an
+        // in-place reference to the object on the operation context, which may be
+        // concurrently used elsewhere (eg. read by currentOp).
+        stdx::lock_guard<Client> lk(*opCtx->getClient());
+        LOGV2_DEBUG(22767,
+                    2,
+                    "Applying default readConcern on {command} of {readConcern}",
+                    "Applying default readConcern on command",
+                    "command"_attr = invocation->definition()->getName(),
+                    "readConcern"_attr = rcDefault);
+        readConcernArgs = std::move(rcDefault);
+        // Update the readConcernSupport, since the default RC was applied.
+        readConcernSupport = invocation->supportsReadConcern(readConcernArgs.getLevel(),
+                                                             !customDefaultReadConcernWasApplied);
+    };
+
+    auto shouldApplyDefaults = startTransaction || !TransactionRouter::get(opCtx);
+    if (readConcernSupport.defaultReadConcernPermit.isOK() && shouldApplyDefaults) {
         if (readConcernArgs.isEmpty()) {
-            const auto rcDefault = ReadWriteConcernDefaults::get(opCtx->getServiceContext())
-                                       .getDefaultReadConcern(opCtx);
+            const auto rwcDefaults =
+                ReadWriteConcernDefaults::get(opCtx->getServiceContext()).getDefault(opCtx);
+            const auto rcDefault = rwcDefaults.getDefaultReadConcern();
             if (rcDefault) {
-                {
-                    // We must obtain the client lock to set ReadConcernArgs, because it's an
-                    // in-place reference to the object on the operation context, which may be
-                    // concurrently used elsewhere (eg. read by currentOp).
-                    stdx::lock_guard<Client> lk(*opCtx->getClient());
-                    readConcernArgs = std::move(*rcDefault);
-                }
-                customDefaultReadConcernWasApplied = true;
-                LOGV2_DEBUG(22767,
-                            2,
-                            "Applying default readConcern on {command} of {readConcern}",
-                            "Applying default readConcern on command",
-                            "command"_attr = invocation->definition()->getName(),
-                            "readConcern"_attr = *rcDefault);
-                // Update the readConcernSupport, since the default RC was applied.
-                readConcernSupport = invocation->supportsReadConcern(readConcernArgs.getLevel());
+                const bool isDefaultRCLocalFeatureFlagEnabled =
+                    serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+                    repl::feature_flags::gDefaultRCLocal.isEnabled(
+                        serverGlobalParams.featureCompatibility);
+                const auto readConcernSource = rwcDefaults.getDefaultReadConcernSource();
+                customDefaultReadConcernWasApplied = !isDefaultRCLocalFeatureFlagEnabled ||
+                    (readConcernSource &&
+                     readConcernSource.get() == DefaultReadConcernSourceEnum::kGlobal);
+
+                applyDefaultReadConcern(*rcDefault);
             }
         }
     }
 
+    // Apply the implicit default read concern even if the command does not support a cluster wide
+    // read concern.
+    if (!readConcernSupport.defaultReadConcernPermit.isOK() &&
+        readConcernSupport.implicitDefaultReadConcernPermit.isOK() && shouldApplyDefaults &&
+        readConcernArgs.isEmpty()) {
+        const auto rcDefault = ReadWriteConcernDefaults::get(opCtx->getServiceContext())
+                                   .getImplicitDefaultReadConcern();
+        applyDefaultReadConcern(rcDefault);
+    }
+
     auto& provenance = readConcernArgs.getProvenance();
 
     // ClientSupplied is the only provenance that clients are allowed to pass to mongos.