New upstream version 6.0.20upstream

28 files changed, 1020 insertions, 177 deletions

diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index 22eb5f4cbb4..bf3ff27ed2e 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -40,6 +40,16 @@ env.Library(
 )
 
 env.Library(
+    target='oplog_constraint_violation_logger',
+    source=[
+        'oplog_constraint_violation_logger.cpp',
+    ],
+    LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/base',
+    ],
+)
+
+env.Library(
     target='oplog',
     source=[
         'apply_ops.cpp',
@@ -55,6 +65,7 @@ env.Library(
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/catalog/catalog_helpers',
         '$BUILD_DIR/mongo/db/catalog/database_holder',
+        '$BUILD_DIR/mongo/db/catalog/health_log_interface',
         '$BUILD_DIR/mongo/db/catalog/import_collection_oplog_entry',
         '$BUILD_DIR/mongo/db/catalog/index_build_oplog_entry',
         '$BUILD_DIR/mongo/db/catalog/local_oplog_info',
@@ -77,6 +88,7 @@ env.Library(
         '$BUILD_DIR/mongo/rpc/command_status',
         'dbcheck',
         'image_collection_entry',
+        'oplog_constraint_violation_logger',
         'repl_coordinator_interface',
         'repl_server_parameters',
         'repl_settings',
@@ -196,6 +208,7 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/bson/bson_validate',
         '$BUILD_DIR/mongo/db/catalog/health_log_interface',
         '$BUILD_DIR/mongo/db/db_raii',
         '$BUILD_DIR/mongo/idl/idl_parser',
@@ -205,6 +218,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/query_exec',
         '$BUILD_DIR/mongo/db/record_id_helpers',
         '$BUILD_DIR/mongo/util/md5',
+        'repl_server_parameters',
     ],
 )
 
@@ -589,7 +603,7 @@ env.Library(
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/auth/authorization_manager_global',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
-        '$BUILD_DIR/mongo/db/curop',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/query_exec',
         '$BUILD_DIR/mongo/db/stats/timer_stats',
         '$BUILD_DIR/mongo/db/storage/storage_options',
@@ -1475,11 +1489,11 @@ env.Library(
         '$BUILD_DIR/mongo/db/auth/auth',
         '$BUILD_DIR/mongo/db/cloner',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
-        '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/kill_sessions_local',
         '$BUILD_DIR/mongo/db/logical_time',
         '$BUILD_DIR/mongo/db/not_primary_error_tracker',
         '$BUILD_DIR/mongo/db/op_observer',
+        '$BUILD_DIR/mongo/db/query/query_stats/query_stats',
         '$BUILD_DIR/mongo/db/query_exec',
         '$BUILD_DIR/mongo/db/service_context',
         '$BUILD_DIR/mongo/db/stats/counters',
@@ -1562,6 +1576,8 @@ if wiredtiger:
         LIBDEPS=[
             '$BUILD_DIR/mongo/db/catalog/database_holder',
             '$BUILD_DIR/mongo/db/catalog/document_validation',
+            '$BUILD_DIR/mongo/db/catalog/health_log',
+            '$BUILD_DIR/mongo/db/catalog/health_log_interface',
             '$BUILD_DIR/mongo/db/index_builds_coordinator_interface',
             '$BUILD_DIR/mongo/db/multitenancy',
             '$BUILD_DIR/mongo/db/service_context_d_test_fixture',
@@ -1678,6 +1694,7 @@ if wiredtiger:
             '$BUILD_DIR/mongo/db/auth/authmocks',
             '$BUILD_DIR/mongo/db/auth/authorization_manager_global',
             '$BUILD_DIR/mongo/db/catalog/catalog_helpers',
+            '$BUILD_DIR/mongo/db/catalog/health_log',
             '$BUILD_DIR/mongo/db/catalog_raii',
             '$BUILD_DIR/mongo/db/commands/feature_compatibility_parsers',
             '$BUILD_DIR/mongo/db/commands/mongod_fcv',
diff --git a/src/mongo/db/repl/collection_bulk_loader_impl.cpp b/src/mongo/db/repl/collection_bulk_loader_impl.cpp
index 73f86c67c59..eb39fdadd55 100644
--- a/src/mongo/db/repl/collection_bulk_loader_impl.cpp
+++ b/src/mongo/db/repl/collection_bulk_loader_impl.cpp
@@ -82,49 +82,54 @@ Status CollectionBulkLoaderImpl::init(const std::vector<BSONObj>& secondaryIndex
         // locks as yielding a MODE_X/MODE_S lock isn't allowed.
         _secondaryIndexesBlock->setIndexBuildMethod(IndexBuildMethod::kForeground);
         _idIndexBlock->setIndexBuildMethod(IndexBuildMethod::kForeground);
-        return writeConflictRetry(
-            _opCtx.get(),
-            "CollectionBulkLoader::init",
-            _collection->getNss().ns(),
-            [&secondaryIndexSpecs, this] {
-                WriteUnitOfWork wuow(_opCtx.get());
-                // All writes in CollectionBulkLoaderImpl should be unreplicated.
-                // The opCtx is accessed indirectly through _secondaryIndexesBlock.
-                UnreplicatedWritesBlock uwb(_opCtx.get());
-                // This enforces the buildIndexes setting in the replica set configuration.
-                CollectionWriter collWriter(_opCtx.get(), *_collection);
-                auto indexCatalog = collWriter.getWritableCollection()->getIndexCatalog();
-                auto specs = indexCatalog->removeExistingIndexesNoChecks(
-                    _opCtx.get(), collWriter.get(), secondaryIndexSpecs);
-                if (specs.size()) {
-                    _secondaryIndexesBlock->ignoreUniqueConstraint();
-                    auto status =
-                        _secondaryIndexesBlock
-                            ->init(_opCtx.get(), collWriter, specs, MultiIndexBlock::kNoopOnInitFn)
-                            .getStatus();
-                    if (!status.isOK()) {
-                        return status;
-                    }
-                } else {
-                    _secondaryIndexesBlock.reset();
-                }
-                if (!_idIndexSpec.isEmpty()) {
-                    auto status = _idIndexBlock
-                                      ->init(_opCtx.get(),
-                                             collWriter,
-                                             _idIndexSpec,
-                                             MultiIndexBlock::kNoopOnInitFn)
-                                      .getStatus();
-                    if (!status.isOK()) {
-                        return status;
-                    }
-                } else {
-                    _idIndexBlock.reset();
-                }
-
-                wuow.commit();
-                return Status::OK();
-            });
+        return writeConflictRetry(_opCtx.get(),
+                                  "CollectionBulkLoader::init",
+                                  _collection->getNss().ns(),
+                                  [&secondaryIndexSpecs, this] {
+                                      WriteUnitOfWork wuow(_opCtx.get());
+                                      // All writes in CollectionBulkLoaderImpl should be
+                                      // unreplicated. The opCtx is accessed indirectly through
+                                      // _secondaryIndexesBlock.
+                                      UnreplicatedWritesBlock uwb(_opCtx.get());
+                                      // This enforces the buildIndexes setting in the replica set
+                                      // configuration.
+                                      CollectionWriter collWriter(_opCtx.get(), *_collection);
+                                      auto indexCatalog =
+                                          collWriter.getWritableCollection()->getIndexCatalog();
+                                      auto specs = indexCatalog->removeExistingIndexesNoChecks(
+                                          _opCtx.get(), collWriter.get(), secondaryIndexSpecs);
+                                      if (specs.size()) {
+                                          _secondaryIndexesBlock->ignoreUniqueConstraint();
+                                          auto status = _secondaryIndexesBlock
+                                                            ->init(_opCtx.get(),
+                                                                   collWriter,
+                                                                   specs,
+                                                                   MultiIndexBlock::kNoopOnInitFn,
+                                                                   /*forRecovery=*/false)
+                                                            .getStatus();
+                                          if (!status.isOK()) {
+                                              return status;
+                                          }
+                                      } else {
+                                          _secondaryIndexesBlock.reset();
+                                      }
+                                      if (!_idIndexSpec.isEmpty()) {
+                                          auto status = _idIndexBlock
+                                                            ->init(_opCtx.get(),
+                                                                   collWriter,
+                                                                   _idIndexSpec,
+                                                                   MultiIndexBlock::kNoopOnInitFn)
+                                                            .getStatus();
+                                          if (!status.isOK()) {
+                                              return status;
+                                          }
+                                      } else {
+                                          _idIndexBlock.reset();
+                                      }
+
+                                      wuow.commit();
+                                      return Status::OK();
+                                  });
     });
 }
 
diff --git a/src/mongo/db/repl/dbcheck.cpp b/src/mongo/db/repl/dbcheck.cpp
index c02ebb9512b..cc140a71817 100644
--- a/src/mongo/db/repl/dbcheck.cpp
+++ b/src/mongo/db/repl/dbcheck.cpp
@@ -27,6 +27,8 @@
  *    it in the license file.
  */
 
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand
+
 #include "mongo/platform/basic.h"
 
 #include "mongo/bson/simple_bsonelement_comparator.h"
@@ -43,10 +45,13 @@
 #include "mongo/db/repl/dbcheck_gen.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/optime.h"
+#include "mongo/db/repl/repl_server_parameters_gen.h"
+#include "mongo/logv2/log.h"
 
 namespace mongo {
 
 MONGO_FAIL_POINT_DEFINE(SleepDbCheckInBatch);
+MONGO_FAIL_POINT_DEFINE(hangAfterGeneratingHashForExtraIndexKeysCheck);
 
 namespace {
 
@@ -409,6 +414,14 @@ Status dbCheckBatchOnSecondary(OperationContext* opCtx,
             // every N batches.
             HealthLogInterface::get(opCtx)->log(*logEntry);
         }
+
+        if (MONGO_unlikely(hangAfterGeneratingHashForExtraIndexKeysCheck.shouldFail())) {
+            LOGV2_DEBUG(3083200,
+                        3,
+                        "Hanging due to hangAfterGeneratingHashForExtraIndexKeysCheck failpoint");
+            // hangAfterGeneratingHashForExtraIndexKeysCheck.pauseWhileSet(opCtx);
+            opCtx->sleepFor(Milliseconds(1000));
+        }
     } catch (const DBException& exception) {
         // In case of an error, report it to the health log,
         auto logEntry = dbCheckErrorHealthLogEntry(
@@ -438,7 +451,6 @@ Status dbCheckOplogCommand(OperationContext* opCtx,
     IDLParserErrorContext ctx("o");
 
     auto skipDbCheck = mode != OplogApplication::Mode::kSecondary;
-    auto severity = skipDbCheck ? SeverityEnum::Warning : SeverityEnum::Info;
     std::string oplogApplicationMode;
     if (mode == OplogApplication::Mode::kInitialSync) {
         oplogApplicationMode = "initial sync";
@@ -459,7 +471,7 @@ Status dbCheckOplogCommand(OperationContext* opCtx,
             // TODO SERVER-78399: Clean up handling minKey/maxKey once feature flag is removed.
             // If the dbcheck oplog entry doesn't contain batchStart, convert minKey to a BSONObj to
             // be used as batchStart.
-            BSONObj batchStart, batchEnd;
+            BSONObj batchStart, batchEnd, batchId;
             if (!invocation.getBatchStart()) {
                 batchStart = BSON("_id" << invocation.getMinKey().elem());
             } else {
@@ -472,21 +484,42 @@ Status dbCheckOplogCommand(OperationContext* opCtx,
             }
             */
 
-            if (!skipDbCheck) {
+            if (!skipDbCheck && !repl::skipApplyingDbCheckBatchOnSecondary.load()) {
                 return dbCheckBatchOnSecondary(opCtx, opTime, invocation);
             }
 
+            // TODO SERVER-89921: Uncomment once the relevant tickets are backported.
             /*
+            if (invocation.getBatchId()) {
+                batchId = invocation.getBatchId().get().toBSON();
+            }
+
             BSONObjBuilder data;
             data.append("batchStart", batchStart);
             data.append("batchEnd", batchEnd);
+
+            if (!batchId.isEmpty()) {
+                data.append("batchId", batchId);
+            }
             */
+
+            auto warningMsg = "cannot execute dbcheck due to ongoing " + oplogApplicationMode;
+            if (repl::skipApplyingDbCheckBatchOnSecondary.load()) {
+                warningMsg =
+                    "skipping applying dbcheck batch because the "
+                    "'skipApplyingDbCheckBatchOnSecondary' parameter is on";
+            }
+
+            LOGV2_DEBUG(8888500, 3, "skipping applying dbcheck batch", "reason"_attr = warningMsg);
+            // TODO SERVER-89921: Uncomment these logging attributes once the relevant tickets are
+            // backported.
+            //"batchStart"_attr = batchStart,
+            //"batchEnd"_attr = batchEnd,
+            //"batchId"_attr = batchId);
+
             auto healthLogEntry = mongo::dbCheckHealthLogEntry(
-                invocation.getNss(),
-                SeverityEnum::Warning,
-                "cannot execute dbcheck due to ongoing " + oplogApplicationMode,
-                type,
-                boost::none /*data*/);
+                invocation.getNss(), SeverityEnum::Warning, warningMsg, type, boost::none /*data*/);
+
             HealthLogInterface::get(Client::getCurrent()->getServiceContext())
                 ->log(*healthLogEntry);
             return Status::OK();
@@ -500,7 +533,7 @@ Status dbCheckOplogCommand(OperationContext* opCtx,
         case OplogEntriesEnum::Stop:
             const auto healthLogEntry = mongo::dbCheckHealthLogEntry(
                 boost::none /*nss*/,
-                severity,
+                skipDbCheck ? SeverityEnum::Warning : SeverityEnum::Info,
                 skipDbCheck ? "cannot execute dbcheck due to ongoing " + oplogApplicationMode : "",
                 type,
                 boost::none /*data*/
diff --git a/src/mongo/db/repl/member_config.h b/src/mongo/db/repl/member_config.h
index cc668fef79c..94d76991b50 100644
--- a/src/mongo/db/repl/member_config.h
+++ b/src/mongo/db/repl/member_config.h
@@ -118,7 +118,7 @@ public:
      * Gets the horizon name for which the parameters (captured during the first `hello`)
      * correspond.
      */
-    StringData determineHorizon(const SplitHorizon::Parameters& params) const {
+    std::string determineHorizon(const SplitHorizon::Parameters& params) const {
         return _splitHorizon.determineHorizon(params);
     }
 
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 9edc5f48b4f..3e79a96362e 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -45,6 +45,7 @@
 #include "mongo/db/auth/action_type.h"
 #include "mongo/db/auth/authorization_manager.h"
 #include "mongo/db/auth/privilege.h"
+#include "mongo/db/catalog/backwards_compatible_collection_options_util.h"
 #include "mongo/db/catalog/capped_utils.h"
 #include "mongo/db/catalog/coll_mod.h"
 #include "mongo/db/catalog/collection.h"
@@ -55,6 +56,7 @@
 #include "mongo/db/catalog/drop_collection.h"
 #include "mongo/db/catalog/drop_database.h"
 #include "mongo/db/catalog/drop_indexes.h"
+#include "mongo/db/catalog/health_log_interface.h"
 #include "mongo/db/catalog/import_collection_oplog_entry_gen.h"
 #include "mongo/db/catalog/local_oplog_info.h"
 #include "mongo/db/catalog/multi_index_block.h"
@@ -987,7 +989,8 @@ const StringMap<ApplyOpMetadata> kOpsMap = {
       {ErrorCodes::NamespaceNotFound}}},
     {"collMod",
      {[](OperationContext* opCtx, const OplogEntry& entry, OplogApplication::Mode mode) -> Status {
-          const auto& cmd = entry.getObject();
+          const auto cmd =
+              backwards_compatible_collection_options::parseCollModCmdFromOplogEntry(entry);
           auto opMsg = OpMsgRequest::fromDBAndBody(entry.getNss().db(), cmd);
           auto collModCmd = CollMod::parse(IDLParserErrorContext("collModOplogEntry"), opMsg);
           const auto nssOrUUID([&collModCmd, &entry, mode]() -> NamespaceStringOrUUID {
@@ -1217,6 +1220,36 @@ void OplogApplication::checkOnOplogFailureForRecovery(OperationContext* opCtx,
     }
 }
 
+// Logger for oplog constraint violations.
+OplogConstraintViolationLogger* oplogConstraintViolationLogger;
+
+MONGO_INITIALIZER(CreateOplogConstraintViolationLogger)(InitializerContext* context) {
+    oplogConstraintViolationLogger = new OplogConstraintViolationLogger();
+}
+
+void logOplogConstraintViolation(OperationContext* opCtx,
+                                 const NamespaceString& nss,
+                                 OplogConstraintViolationEnum type,
+                                 const std::string& operation,
+                                 const BSONObj& opObj,
+                                 boost::optional<Status> status) {
+    // Log the violation.
+    oplogConstraintViolationLogger->logViolationIfReady(type, opObj, status);
+
+    // Write a new entry to the health log.
+    HealthLogEntry entry;
+    entry.setNss(nss);
+    entry.setTimestamp(Date_t::now());
+    // Oplog constraint violations should always be marked as warning.
+    entry.setSeverity(SeverityEnum::Warning);
+    entry.setScope(ScopeEnum::Document);
+    entry.setMsg(toString(type));
+    entry.setOperation(operation);
+    entry.setData(opObj);
+
+    HealthLogInterface::get(opCtx->getServiceContext())->log(entry);
+}
+
 // @return failure status if an update should have happened and the document DNE.
 // See replset initial sync code.
 Status applyOperation_inlock(OperationContext* opCtx,
@@ -1495,7 +1528,17 @@ Status applyOperation_inlock(OperationContext* opCtx,
                             return status;
                         }
                         if (mode == OplogApplication::Mode::kSecondary) {
+                            const auto& opObj = redact(op.toBSONForLogging());
+
                             opCounters->gotInsertOnExistingDoc();
+                            logOplogConstraintViolation(
+                                opCtx,
+                                op.getNss(),
+                                OplogConstraintViolationEnum::kInsertOnExistingDoc,
+                                "insert",
+                                opObj,
+                                boost::none /* status */);
+
                             if (oplogApplicationEnforcesSteadyStateConstraints) {
                                 return status;
                             }
@@ -1719,10 +1762,15 @@ Status applyOperation_inlock(OperationContext* opCtx,
                            !ur.upsertedId.isEmpty() && !(collection && collection->isCapped())) {
                     // This indicates we upconverted an update to an upsert, and it did indeed
                     // upsert.  In steady state mode this is unexpected.
-                    LOGV2_WARNING(2170001,
-                                  "update needed to be converted to upsert",
-                                  "op"_attr = redact(op.toBSONForLogging()));
+                    const auto& opObj = redact(op.toBSONForLogging());
+
                     opCounters->gotUpdateOnMissingDoc();
+                    logOplogConstraintViolation(opCtx,
+                                                op.getNss(),
+                                                OplogConstraintViolationEnum::kUpdateOnMissingDoc,
+                                                "update",
+                                                opObj,
+                                                boost::none /* status */);
 
                     // We shouldn't be doing upserts in secondary mode when enforcing steady state
                     // constraints.
@@ -1858,13 +1906,15 @@ Status applyOperation_inlock(OperationContext* opCtx,
                 // It is legal for a delete operation on the pre-images collection to delete zero
                 // documents - pre-image collections are not guaranteed to contain the same set of
                 // documents at all times.
+                //
+                // It is also legal for a delete operation on the config.image_collection (used for
+                // find-and-modify retries) to delete zero documents.  Since we do not write updates
+                // to this collection which are in the same batch as later deletes, a rollback to
+                // the middle of a batch with both an update and a delete may result in a missing
+                // document, which may be later deleted.
                 if (result.nDeleted == 0 && mode == OplogApplication::Mode::kSecondary &&
-                    !requestNss.isChangeStreamPreImagesCollection()) {
-                    LOGV2_WARNING(2170002,
-                                  "Applied a delete which did not delete anything in steady state "
-                                  "replication",
-                                  "op"_attr = redact(op.toBSONForLogging()));
-
+                    !requestNss.isChangeStreamPreImagesCollection() &&
+                    !requestNss.isConfigImagesCollection()) {
                     // In FCV 4.4, each node is responsible for deleting the excess documents in
                     // capped collections. This implies that capped deletes may not be synchronized
                     // between nodes at times. When upgraded to FCV 5.0, the primary will generate
@@ -1877,11 +1927,25 @@ Status applyOperation_inlock(OperationContext* opCtx,
                     // capped collections when oplog application is enforcing steady state
                     // constraints.
                     bool isCapped = false;
+                    const auto& opObj = redact(op.toBSONForLogging());
                     if (collection) {
                         isCapped = collection->isCapped();
                         opCounters->gotDeleteWasEmpty();
+                        logOplogConstraintViolation(opCtx,
+                                                    op.getNss(),
+                                                    OplogConstraintViolationEnum::kDeleteWasEmpty,
+                                                    "delete",
+                                                    opObj,
+                                                    boost::none /* status */);
                     } else {
                         opCounters->gotDeleteFromMissingNamespace();
+                        logOplogConstraintViolation(
+                            opCtx,
+                            op.getNss(),
+                            OplogConstraintViolationEnum::kDeleteOnMissingNs,
+                            "delete",
+                            opObj,
+                            boost::none /* status */);
                     }
 
                     if (!isCapped) {
@@ -2134,12 +2198,15 @@ Status applyCommand_inlock(OperationContext* opCtx,
 
                 if (mode == OplogApplication::Mode::kSecondary &&
                     status.code() != ErrorCodes::IndexNotFound) {
-                    LOGV2_WARNING(2170000,
-                                  "Acceptable error during oplog application",
-                                  "db"_attr = nss.db(),
-                                  "error"_attr = status,
-                                  "oplogEntry"_attr = redact(entry.toBSONForLogging()));
+                    const auto& opObj = redact(entry.toBSONForLogging());
                     opCounters->gotAcceptableErrorInCommand();
+                    logOplogConstraintViolation(
+                        opCtx,
+                        entry.getNss(),
+                        OplogConstraintViolationEnum::kAcceptableErrorInCommand,
+                        "command",
+                        opObj,
+                        status);
                 } else {
                     LOGV2_DEBUG(51776,
                                 1,
diff --git a/src/mongo/db/repl/oplog.h b/src/mongo/db/repl/oplog.h
index 1752c2e64e6..3df367954b0 100644
--- a/src/mongo/db/repl/oplog.h
+++ b/src/mongo/db/repl/oplog.h
@@ -38,6 +38,7 @@
 #include "mongo/bson/timestamp.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/logical_session_id.h"
+#include "mongo/db/repl/oplog_constraint_violation_logger.h"
 #include "mongo/db/repl/oplog_entry.h"
 #include "mongo/db/repl/oplog_entry_or_grouped_inserts.h"
 #include "mongo/db/repl/optime.h"
@@ -220,6 +221,16 @@ inline std::ostream& operator<<(std::ostream& s, OplogApplication::Mode mode) {
 }
 
 /**
+ * Logs an oplog constraint violation and writes an entry into the health log.
+ */
+void logOplogConstraintViolation(OperationContext* opCtx,
+                                 const NamespaceString& nss,
+                                 OplogConstraintViolationEnum type,
+                                 const std::string& operation,
+                                 const BSONObj& opObj,
+                                 boost::optional<Status> status);
+
+/**
  * Used for applying from an oplog entry or grouped inserts.
  * @param opOrGroupedInserts a single oplog entry or grouped inserts to be applied.
  * @param alwaysUpsert convert some updates to upserts for idempotency reasons
diff --git a/src/mongo/db/repl/oplog_applier_impl.cpp b/src/mongo/db/repl/oplog_applier_impl.cpp
index 4f8c3f9427e..3ce566c713b 100644
--- a/src/mongo/db/repl/oplog_applier_impl.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl.cpp
@@ -103,6 +103,11 @@ Status finishAndLogApply(OperationContext* opCtx,
 
             attrs.add("duration", Milliseconds(opDuration));
 
+            // Obtain storage specific statistics and log them if they exist.
+            CurOp::get(opCtx)->debug().storageStats =
+                opCtx->recoveryUnit()->computeOperationStatisticsSinceLastCall();
+            CurOp::get(opCtx)->debug().reportStorageStats(&attrs);
+
             LOGV2(51801, "Applied op", attrs);
         }
     }
diff --git a/src/mongo/db/repl/oplog_applier_impl_test.cpp b/src/mongo/db/repl/oplog_applier_impl_test.cpp
index 68b1b6fdb47..26e142f6372 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl_test.cpp
@@ -67,7 +67,6 @@
 #include "mongo/db/session_txn_record_gen.h"
 #include "mongo/db/stats/counters.h"
 #include "mongo/db/transaction_participant_gen.h"
-#include "mongo/idl/server_parameter_test_util.h"
 #include "mongo/platform/mutex.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
@@ -129,7 +128,7 @@ TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
     NamespaceString otherNss("test.othername");
     auto op = makeOplogEntry(OpTypeEnum::kDelete, otherNss, {});
     int prevDeleteFromMissing = replOpCounters.getDeleteFromMissingNamespace()->load();
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, false);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, otherNss, false);
     auto postDeleteFromMissing = replOpCounters.getDeleteFromMissingNamespace()->load();
     ASSERT_EQ(1, postDeleteFromMissing - prevDeleteFromMissing);
 
@@ -167,7 +166,7 @@ TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
     NamespaceString otherNss(nss.getSisterNS("othername"));
     auto op = makeOplogEntry(OpTypeEnum::kDelete, otherNss, kUuid);
     int prevDeleteFromMissing = replOpCounters.getDeleteFromMissingNamespace()->load();
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, false);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, nss, false);
     auto postDeleteFromMissing = replOpCounters.getDeleteFromMissingNamespace()->load();
     ASSERT_EQ(1, postDeleteFromMissing - prevDeleteFromMissing);
 
@@ -211,7 +210,7 @@ TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
     // implicitly create the collection.
     auto op = makeOplogEntry(OpTypeEnum::kDelete, nss, {});
     int prevDeleteFromMissing = replOpCounters.getDeleteFromMissingNamespace()->load();
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, false);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, nss, false);
     ASSERT_FALSE(collectionExists(_opCtx.get(), nss));
     auto postDeleteFromMissing = replOpCounters.getDeleteFromMissingNamespace()->load();
     ASSERT_EQ(1, postDeleteFromMissing - prevDeleteFromMissing);
@@ -239,7 +238,7 @@ TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentCollec
     const NamespaceString nss("test.t");
     repl::createCollection(_opCtx.get(), nss, {});
     auto op = makeOplogEntry(OpTypeEnum::kInsert, nss, {});
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, true);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, nss, true);
 }
 
 TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
@@ -248,7 +247,7 @@ TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
     repl::createCollection(_opCtx.get(), nss, {});
     auto op = makeOplogEntry(OpTypeEnum::kDelete, nss, {});
     int prevDeleteWasEmpty = replOpCounters.getDeleteWasEmpty()->load();
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, false);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, nss, false);
     auto postDeleteWasEmpty = replOpCounters.getDeleteWasEmpty()->load();
     ASSERT_EQ(1, postDeleteWasEmpty - prevDeleteWasEmpty);
 
@@ -274,7 +273,7 @@ TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentCollec
     createCollection(_opCtx.get(), nss, createRecordPreImageCollectionOptions());
     ASSERT_OK(getStorageInterface()->insertDocument(_opCtx.get(), nss, {BSON("_id" << 0)}, 0));
     auto op = makeOplogEntry(OpTypeEnum::kDelete, nss, {});
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, true);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, nss, true);
 }
 
 TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
@@ -284,7 +283,7 @@ TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
     ASSERT_OK(getStorageInterface()->insertDocument(_opCtx.get(), nss, {BSON("_id" << 0)}, 0));
     auto op = makeOplogEntry(OpTypeEnum::kInsert, nss, uuid);
     int prevInsertOnExistingDoc = replOpCounters.getInsertOnExistingDoc()->load();
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, false);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, nss, false);
     auto postInsertOnExistingDoc = replOpCounters.getInsertOnExistingDoc()->load();
     ASSERT_EQ(1, postInsertOnExistingDoc - prevInsertOnExistingDoc);
 
@@ -301,7 +300,7 @@ TEST_F(OplogApplierImplTestEnableSteadyStateConstraints,
     auto uuid = createCollectionWithUuid(_opCtx.get(), nss);
     ASSERT_OK(getStorageInterface()->insertDocument(_opCtx.get(), nss, {BSON("_id" << 0)}, 0));
     auto op = makeOplogEntry(OpTypeEnum::kInsert, nss, uuid);
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::DuplicateKey, op, false);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::DuplicateKey, op, nss, false);
 }
 
 TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
@@ -311,7 +310,7 @@ TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
     auto op = makeOplogEntry(
         repl::OpTypeEnum::kUpdate, nss, uuid, BSON("$set" << BSON("a" << 1)), BSON("_id" << 0));
     int prevUpdateOnMissingDoc = replOpCounters.getUpdateOnMissingDoc()->load();
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, true);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, nss, true);
     auto postUpdateOnMissingDoc = replOpCounters.getUpdateOnMissingDoc()->load();
     ASSERT_EQ(1, postUpdateOnMissingDoc - prevUpdateOnMissingDoc);
 
@@ -328,7 +327,8 @@ TEST_F(OplogApplierImplTestEnableSteadyStateConstraints,
     auto uuid = createCollectionWithUuid(_opCtx.get(), nss);
     auto op = makeOplogEntry(
         repl::OpTypeEnum::kUpdate, nss, uuid, BSON("$set" << BSON("a" << 1)), BSON("_id" << 0));
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::UpdateOperationFailed, op, false);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(
+        ErrorCodes::UpdateOperationFailed, op, nss, false);
 }
 
 TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentCollectionLockedByUUID) {
@@ -337,7 +337,7 @@ TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentCollec
     // Test that the collection to lock is determined by the UUID and not the 'ns' field.
     NamespaceString otherNss(nss.getSisterNS("othername"));
     auto op = makeOplogEntry(OpTypeEnum::kInsert, otherNss, uuid);
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, true);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, nss, true);
 }
 
 TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
@@ -351,7 +351,7 @@ TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
     NamespaceString otherNss(nss.getSisterNS("othername"));
     auto op = makeOplogEntry(OpTypeEnum::kDelete, otherNss, options.uuid);
     int prevDeleteWasEmpty = replOpCounters.getDeleteWasEmpty()->load();
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, false);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, nss, false);
     auto postDeleteWasEmpty = replOpCounters.getDeleteWasEmpty()->load();
     ASSERT_EQ(1, postDeleteWasEmpty - prevDeleteWasEmpty);
 
@@ -388,7 +388,7 @@ TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentCollec
     // Test that the collection to lock is determined by the UUID and not the 'ns' field.
     NamespaceString otherNss(nss.getSisterNS("othername"));
     auto op = makeOplogEntry(OpTypeEnum::kDelete, otherNss, options.uuid);
-    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, true);
+    _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, nss, true);
 }
 
 TEST_F(OplogApplierImplTest, applyOplogEntryToRecordChangeStreamPreImages) {
diff --git a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
index 35f06edd12e..4afe05e7bc5 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
@@ -32,6 +32,8 @@
 #include "mongo/db/repl/oplog_applier_impl_test_fixture.h"
 
 #include "mongo/db/catalog/document_validation.h"
+#include "mongo/db/catalog/health_log.h"
+#include "mongo/db/catalog/health_log_interface.h"
 #include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
@@ -182,14 +184,22 @@ void OplogApplierImplTest::setUp() {
     // This is necessary to generate ghost timestamps for index builds that are not 0, since 0 is an
     // invalid timestamp.
     VectorClockMutable::get(_opCtx.get())->tickClusterTimeTo(LogicalTime(Timestamp(1, 0)));
+
+    HealthLogInterface::set(serviceContext, std::make_unique<HealthLog>());
+    HealthLogInterface::get(serviceContext)->startup();
 }
 
 void OplogApplierImplTest::tearDown() {
+    HealthLogInterface::get(serviceContext)->shutdown();
     _opCtx.reset();
     _consistencyMarkers = {};
     DropPendingCollectionReaper::set(serviceContext, {});
     StorageInterface::set(serviceContext, {});
     ServiceContextMongoDTest::tearDown();
+
+    for (auto serverParamController : _serverParamControllers) {
+        serverParamController.reset();
+    }
 }
 
 ReplicationConsistencyMarkers* OplogApplierImplTest::getConsistencyMarkers() const {
@@ -213,7 +223,10 @@ Status OplogApplierImplTest::_applyOplogEntryOrGroupedInsertsWrapper(
 }
 
 void OplogApplierImplTest::_testApplyOplogEntryOrGroupedInsertsCrudOperation(
-    ErrorCodes::Error expectedError, const OplogEntry& op, bool expectedApplyOpCalled) {
+    ErrorCodes::Error expectedError,
+    const OplogEntry& op,
+    const NamespaceString& targetNss,
+    bool expectedApplyOpCalled) {
     bool applyOpCalled = false;
 
     auto checkOpCtx = [](OperationContext* opCtx) {
@@ -228,9 +241,14 @@ void OplogApplierImplTest::_testApplyOplogEntryOrGroupedInsertsCrudOperation(
 
     _opObserver->onInsertsFn =
         [&](OperationContext* opCtx, const NamespaceString& nss, const std::vector<BSONObj>& docs) {
+            // Other threads may be calling into the opObserver. Only assert if we are writing to
+            // the target ns, otherwise skip these asserts.
+            if (targetNss != nss) {
+                return Status::OK();
+            }
+
             applyOpCalled = true;
             checkOpCtx(opCtx);
-            ASSERT_EQUALS(NamespaceString("test.t"), nss);
             ASSERT_EQUALS(1U, docs.size());
             // For upserts we don't know the intended value of the document.
             if (op.getOpType() == repl::OpTypeEnum::kInsert) {
@@ -244,18 +262,28 @@ void OplogApplierImplTest::_testApplyOplogEntryOrGroupedInsertsCrudOperation(
                                   const boost::optional<UUID>& uuid,
                                   StmtId stmtId,
                                   const OplogDeleteEntryArgs& args) {
+        // Other threads may be calling into the opObserver. Only assert if we are writing to
+        // the target ns, otherwise skip these asserts.
+        if (targetNss != nss) {
+            return Status::OK();
+        }
+
         applyOpCalled = true;
         checkOpCtx(opCtx);
-        ASSERT_EQUALS(NamespaceString("test.t"), nss);
         ASSERT(args.deletedDoc);
         ASSERT_BSONOBJ_EQ(op.getObject(), *(args.deletedDoc));
         return Status::OK();
     };
 
     _opObserver->onUpdateFn = [&](OperationContext* opCtx, const OplogUpdateEntryArgs& args) {
+        // Other threads may be calling into the opObserver. Only assert if we are writing to
+        // the target ns, otherwise skip these asserts.
+        if (targetNss != args.nss) {
+            return Status::OK();
+        }
+
         applyOpCalled = true;
         checkOpCtx(opCtx);
-        ASSERT_EQUALS(NamespaceString("test.t"), args.nss);
         return Status::OK();
     };
 
diff --git a/src/mongo/db/repl/oplog_applier_impl_test_fixture.h b/src/mongo/db/repl/oplog_applier_impl_test_fixture.h
index e1b188232ae..28709a29b2c 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test_fixture.h
+++ b/src/mongo/db/repl/oplog_applier_impl_test_fixture.h
@@ -38,6 +38,7 @@
 #include "mongo/db/repl/replication_consistency_markers.h"
 #include "mongo/db/service_context_d_test_fixture.h"
 #include "mongo/db/session_txn_record_gen.h"
+#include "mongo/idl/server_parameter_test_util.h"
 
 namespace mongo {
 
@@ -202,6 +203,7 @@ protected:
 
     void _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::Error expectedError,
                                                            const OplogEntry& op,
+                                                           const NamespaceString& targetNss,
                                                            bool expectedApplyOpCalled);
 
     Status _applyOplogEntryOrGroupedInsertsWrapper(OperationContext* opCtx,
@@ -213,6 +215,11 @@ protected:
     ServiceContext* serviceContext;
     OplogApplierImplOpObserver* _opObserver = nullptr;
 
+    template <typename T>
+    inline void setServerParameter(const std::string& name, T value) {
+        _serverParamControllers.push_back(ServerParameterControllerForTest(name, value));
+    }
+
     OpTime nextOpTime() {
         static long long lastSecond = 1;
         return OpTime(Timestamp(Seconds(lastSecond++), 0), 1LL);
@@ -232,6 +239,8 @@ protected:
     Status runOpsInitialSync(std::vector<OplogEntry> ops);
 
     UUID kUuid{UUID::gen()};
+
+    std::vector<ServerParameterControllerForTest> _serverParamControllers;
 };
 
 // Utility class to allow easily scanning a collection.  Scans in forward order, returns
diff --git a/src/mongo/db/repl/oplog_applier_utils.cpp b/src/mongo/db/repl/oplog_applier_utils.cpp
index 5029e0b57d6..5b1061e3998 100644
--- a/src/mongo/db/repl/oplog_applier_utils.cpp
+++ b/src/mongo/db/repl/oplog_applier_utils.cpp
@@ -256,7 +256,15 @@ Status OplogApplierUtils::applyOplogEntryOrGroupedInsertsCommon(
                         !oplogApplicationEnforcesSteadyStateConstraints &&
                         oplogApplicationMode == OplogApplication::Mode::kSecondary) {
                         if (opCounters) {
+                            const auto& opObj = redact(op.toBSONForLogging());
                             opCounters->gotDeleteFromMissingNamespace();
+                            logOplogConstraintViolation(
+                                opCtx,
+                                op.getNss(),
+                                OplogConstraintViolationEnum::kDeleteOnMissingNs,
+                                "delete",
+                                opObj,
+                                boost::none /* status */);
                         }
                         return Status::OK();
                     }
diff --git a/src/mongo/db/repl/oplog_constraint_violation_logger.cpp b/src/mongo/db/repl/oplog_constraint_violation_logger.cpp
new file mode 100644
index 00000000000..f887393dbdf
--- /dev/null
+++ b/src/mongo/db/repl/oplog_constraint_violation_logger.cpp
@@ -0,0 +1,90 @@
+/**
+ *    Copyright (C) 2023-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kReplication
+
+#include "mongo/db/repl/oplog_constraint_violation_logger.h"
+#include "mongo/logv2/log.h"
+
+namespace mongo {
+namespace repl {
+
+// Default interval set to 10 minutes.
+const Seconds OplogConstraintViolationLogger::kPeriodicLogTimeout(60 * 10);
+
+StringData toString(OplogConstraintViolationEnum type) {
+    switch (type) {
+        case OplogConstraintViolationEnum::kInsertOnExistingDoc:
+            return kInsertOnExistingDocMsg;
+        case OplogConstraintViolationEnum::kUpdateOnMissingDoc:
+            return kUpdateOnMissingDocMsg;
+        case OplogConstraintViolationEnum::kDeleteWasEmpty:
+            return kDeleteWasEmptyMsg;
+        case OplogConstraintViolationEnum::kDeleteOnMissingNs:
+            return kDeleteOnMissingNs;
+        case OplogConstraintViolationEnum::kAcceptableErrorInCommand:
+            return kAcceptableErrorInCommand;
+        default:
+            return "";
+    }
+}
+
+void OplogConstraintViolationLogger::logViolationIfReady(OplogConstraintViolationEnum type,
+                                                         const BSONObj& obj,
+                                                         boost::optional<Status> status) {
+    const auto index = static_cast<int>(type);
+
+    stdx::lock_guard lk(_mutex);
+    const auto lastLog = _lastLogTimes[index];
+    const auto now = Date_t::now();
+
+    if (now < lastLog + OplogConstraintViolationLogger::kPeriodicLogTimeout) {
+        // We have logged this violation already within the last 10 minutes.
+        return;
+    }
+
+    if (!status) {
+        LOGV2_WARNING(7149000,
+                      "Potential replication constraint violation during steady state replication",
+                      "msg"_attr = toString(type),
+                      "obj"_attr = obj);
+    } else {
+        LOGV2_WARNING(7149001,
+                      "Potential replication constraint violation during steady state replication",
+                      "msg"_attr = toString(type),
+                      "obj"_attr = obj,
+                      "status"_attr = *status);
+    }
+
+    // Update the last log time to now.
+    _lastLogTimes[index] = now;
+}
+
+}  // namespace repl
+}  // namespace mongo
diff --git a/src/mongo/db/repl/oplog_constraint_violation_logger.h b/src/mongo/db/repl/oplog_constraint_violation_logger.h
new file mode 100644
index 00000000000..321544e28d9
--- /dev/null
+++ b/src/mongo/db/repl/oplog_constraint_violation_logger.h
@@ -0,0 +1,85 @@
+/**
+ *    Copyright (C) 2023-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include <boost/optional.hpp>
+
+#include "mongo/base/status.h"
+#include "mongo/bson/bsonobj.h"
+#include "mongo/platform/mutex.h"
+#include "mongo/util/time_support.h"
+
+namespace mongo {
+namespace repl {
+
+static constexpr StringData kInsertOnExistingDocMsg = "attempted to insert on existing doc"_sd;
+static constexpr StringData kUpdateOnMissingDocMsg =
+    "ran update as upsert and failed to match any documents"_sd;
+static constexpr StringData kDeleteWasEmptyMsg = "applied a delete that did not delete anything"_sd;
+static constexpr StringData kDeleteOnMissingNs = "applied a delete on missing namespace"_sd;
+static constexpr StringData kAcceptableErrorInCommand =
+    "received an acceptable error during oplog application"_sd;
+
+enum class OplogConstraintViolationEnum {
+    kInsertOnExistingDoc = 0,
+    kUpdateOnMissingDoc,
+    kDeleteWasEmpty,
+    kDeleteOnMissingNs,
+    kAcceptableErrorInCommand,
+    NUM_VIOLATION_TYPES,
+};
+
+// Returns a string describing the constraint violation of the given type.
+StringData toString(OplogConstraintViolationEnum type);
+
+/**
+ * Logs oplog constraint violation occurrences.
+ *
+ * To avoid flooding the logs if continuous oplog constraint violations occur, we will only log
+ * once every 10 minutes per each oplog constraint violation type.
+ */
+class OplogConstraintViolationLogger {
+public:
+    // Minimum period of time before logging another warning log message, set to 10min.
+    static const Seconds kPeriodicLogTimeout;
+
+    void logViolationIfReady(OplogConstraintViolationEnum type,
+                             const BSONObj& obj,
+                             boost::optional<Status> status);
+
+private:
+    mutable Mutex _mutex = MONGO_MAKE_LATCH("OplogConstraintViolationLogger::mutex");
+
+    std::vector<Date_t> _lastLogTimes = std::vector<Date_t>(
+        static_cast<int>(OplogConstraintViolationEnum::NUM_VIOLATION_TYPES));  // (M)
+};
+
+}  // namespace repl
+}  // namespace mongo
diff --git a/src/mongo/db/repl/oplog_entry.cpp b/src/mongo/db/repl/oplog_entry.cpp
index 7f7363be06e..77fba82039f 100644
--- a/src/mongo/db/repl/oplog_entry.cpp
+++ b/src/mongo/db/repl/oplog_entry.cpp
@@ -764,5 +764,32 @@ int OplogEntry::getRawObjSizeBytes() const {
     return _entry.getRawObjSizeBytes();
 }
 
+OplogEntryParserNonStrict::OplogEntryParserNonStrict(const BSONObj& oplogEntry)
+    : _oplogEntryObject{oplogEntry.getOwned()} {}
+
+repl::OpTime OplogEntryParserNonStrict::getOpTime() const {
+    return uassertStatusOKWithContext(repl::OpTime::parseFromOplogEntry(_oplogEntryObject),
+                                      str::stream() << "Failed to parse opTime");
+}
+
+repl::OpTypeEnum OplogEntryParserNonStrict::getOpType() const {
+    auto opTypeElement = _oplogEntryObject[repl::OplogEntry::kOpTypeFieldName];
+    uassert(8881100,
+            str::stream() << "Invalid '" << repl::OplogEntry::kOpTypeFieldName
+                          << "' field type (expected String)",
+            opTypeElement.type() == BSONType::String);
+    return repl::OpType_parse(IDLParserErrorContext("ChangeStreamEntry.op"),
+                              opTypeElement.checkAndGetStringData());
+}
+
+BSONObj OplogEntryParserNonStrict::getObject() const {
+    auto objectElement = _oplogEntryObject[repl::OplogEntry::kObjectFieldName];
+    uassert(8881101,
+            str::stream() << "Invalid '" << repl::OplogEntry::kObjectFieldName
+                          << "' field type (expected Object)",
+            objectElement.isABSONObj());
+    return objectElement.Obj();
+}
+
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/repl/oplog_entry.h b/src/mongo/db/repl/oplog_entry.h
index 7552faa3f9e..f41898f8c6b 100644
--- a/src/mongo/db/repl/oplog_entry.h
+++ b/src/mongo/db/repl/oplog_entry.h
@@ -801,6 +801,40 @@ private:
     bool _isForCappedCollection = false;
 };
 
+/**
+ * Oplog entry document parser. This parser can parse only the key fields. It parses fields on
+ * demand. This parser should be used only in cases when to be parsed oplog entry data structure
+ * version may not match the one that is used by the current server version (this can happen with
+ * past or future versions of oplog entries), otherwise 'OplogEntry::parse()' is supposed to be
+ * used.
+ */
+class OplogEntryParserNonStrict {
+public:
+    /**
+     * Constructs the parser with to be parsed oplog entry document 'oplogEntry'.
+     */
+    OplogEntryParserNonStrict(const BSONObj& oplogEntry);
+
+    /**
+     * Parses and returns "opTime" field.
+     */
+    repl::OpTime getOpTime() const;
+
+    /**
+     * Parses and returns the type of operation field.
+     */
+    repl::OpTypeEnum getOpType() const;
+
+    /**
+     * Parses and returns the "operation applied" field.
+     */
+    BSONObj getObject() const;
+
+private:
+    // Oplog entry as BSON object to be parsed.
+    const BSONObj _oplogEntryObject;
+};
+
 std::ostream& operator<<(std::ostream& s, const DurableOplogEntry& o);
 std::ostream& operator<<(std::ostream& s, const OplogEntry& o);
 
diff --git a/src/mongo/db/repl/oplog_entry_test.cpp b/src/mongo/db/repl/oplog_entry_test.cpp
index c034bd3c517..6cfd6958519 100644
--- a/src/mongo/db/repl/oplog_entry_test.cpp
+++ b/src/mongo/db/repl/oplog_entry_test.cpp
@@ -135,7 +135,75 @@ TEST(OplogEntryTest, OpTimeBaseNonStrictParsing) {
         40414);
 }
 
+TEST(OplogEntryParserTest, ParseOpTimeSuccess) {
+    repl::OpTime opTime{Timestamp{2}, 1};
+    auto const oplogEntry = opTime.toBSON();
+    OplogEntryParserNonStrict parser{oplogEntry};
+    ASSERT_EQ(opTime, parser.getOpTime()) << oplogEntry.toString();
+}
+
+TEST(OplogEntryParserTest, ParseOpTimeFailure) {
+    auto const oplogEntry = BSON("a" << 1);
+    OplogEntryParserNonStrict parser{oplogEntry};
+    ASSERT_THROWS_CODE_AND_WHAT(parser.getOpTime(),
+                                AssertionException,
+                                40414,
+                                "Failed to parse opTime :: caused by :: "
+                                "BSON field 'OpTimeBase.ts' is missing but a required field");
+}
+
+TEST(OplogEntryParserTest, ParseOpTypeSuccess) {
+    auto const oplogEntry =
+        BSON(OplogEntry::kOpTypeFieldName << OpType_serializer(repl::OpTypeEnum::kDelete));
+    OplogEntryParserNonStrict parser{oplogEntry};
+    ASSERT_TRUE(repl::OpTypeEnum::kDelete == parser.getOpType()) << oplogEntry.toString();
+}
 
+TEST(OplogEntryParserTest, ParseOpTypeFailure) {
+    {
+        auto const oplogEntry = BSON(OplogEntry::kOpTypeFieldName << "zz");
+        OplogEntryParserNonStrict parser{oplogEntry};
+        ASSERT_THROWS_CODE_AND_WHAT(
+            parser.getOpType(),
+            AssertionException,
+            ErrorCodes::BadValue,
+            "Enumeration value 'zz' for field 'ChangeStreamEntry.op' is not a valid value.");
+    }
+    {
+        auto const oplogEntry = BSON(OplogEntry::kOpTypeFieldName << 1);
+        OplogEntryParserNonStrict parser{oplogEntry};
+        ASSERT_THROWS_CODE_AND_WHAT(parser.getOpType(),
+                                    AssertionException,
+                                    8881100,
+                                    "Invalid 'op' field type (expected String)");
+    }
+}
+
+TEST(OplogEntryParserTest, ParseObjectSuccess) {
+    auto const objectFieldValue = BSON("a" << 1);
+    auto const oplogEntry = BSON(OplogEntry::kObjectFieldName << objectFieldValue);
+    OplogEntryParserNonStrict parser{oplogEntry};
+    ASSERT_BSONOBJ_BINARY_EQ(objectFieldValue, parser.getObject());
+}
+
+TEST(OplogEntryParserTest, ParseObjectFailure) {
+    {
+        auto const oplogEntry = BSON(OplogEntry::kObjectFieldName << "string");
+        OplogEntryParserNonStrict parser{oplogEntry};
+        ASSERT_THROWS_CODE_AND_WHAT(parser.getObject(),
+                                    AssertionException,
+                                    8881101,
+                                    "Invalid 'o' field type (expected Object)");
+    }
+    {
+        auto const oplogEntry = BSON("a" << 1);
+        OplogEntryParserNonStrict parser{oplogEntry};
+        ASSERT_THROWS_CODE_AND_WHAT(parser.getObject(),
+                                    AssertionException,
+                                    8881101,
+                                    "Invalid 'o' field type (expected Object)");
+    }
+}
 }  // namespace
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/repl/repl_server_parameters.idl b/src/mongo/db/repl/repl_server_parameters.idl
index 3f4f5aac440..e178b4f4154 100644
--- a/src/mongo/db/repl/repl_server_parameters.idl
+++ b/src/mongo/db/repl/repl_server_parameters.idl
@@ -665,6 +665,15 @@ server_parameters:
         default: ""
         validator: { callback: 'validateHostAndPort' }
 
+    skipApplyingDbCheckBatchOnSecondary:
+        description: >-
+            Parameter for whether dbcheck batches should be applied on secondaries.
+        set_at: [ startup, runtime ]
+        cpp_vartype: AtomicWord<bool>
+        cpp_varname: skipApplyingDbCheckBatchOnSecondary
+        default: false
+
+
 feature_flags:
     featureFlagRetryableFindAndModify:
         description: >-
diff --git a/src/mongo/db/repl/replication_coordinator_impl.cpp b/src/mongo/db/repl/replication_coordinator_impl.cpp
index 8074192cb6a..fb0c8fa81fc 100644
--- a/src/mongo/db/repl/replication_coordinator_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl.cpp
@@ -111,7 +111,6 @@
 #include "mongo/util/assert_util.h"
 #include "mongo/util/fail_point.h"
 #include "mongo/util/scopeguard.h"
-#include "mongo/util/stacktrace.h"
 #include "mongo/util/testing_proctor.h"
 #include "mongo/util/time_support.h"
 #include "mongo/util/timer.h"
@@ -184,6 +183,13 @@ ServerStatusMetricField<Counter64> displayNumAutoReconfigs(
     "repl.reconfig.numAutoReconfigsForRemovalOfNewlyAddedFields",
     &numAutoReconfigsForRemovalOfNewlyAddedFields);
 
+Atomic64Metric replicationWaiterListMetric;
+ServerStatusMetricField<Atomic64Metric> displayReplicationWaiterListMetric(
+    "repl.waiters.replication", &replicationWaiterListMetric);
+Atomic64Metric opTimeWaiterListMetric;
+ServerStatusMetricField<Atomic64Metric> displayOpTimeWaiterListMetric("repl.waiters.opTime",
+                                                                      &opTimeWaiterListMetric);
+
 using namespace fmt::literals;
 
 using CallbackArgs = executor::TaskExecutor::CallbackArgs;
@@ -221,15 +227,24 @@ constexpr StringData kQuiesceModeShutdownMessage =
 
 }  // namespace
 
+ReplicationCoordinatorImpl::WaiterList::WaiterList(Atomic64Metric& waiterCountMetric)
+    : _waiterCountMetric(waiterCountMetric) {}
+
+void ReplicationCoordinatorImpl::WaiterList::_updateMetric_inlock() {
+    _waiterCountMetric.set(_waiters.size());
+}
+
 void ReplicationCoordinatorImpl::WaiterList::add_inlock(const OpTime& opTime,
                                                         SharedWaiterHandle waiter) {
     _waiters.emplace(opTime, std::move(waiter));
+    _updateMetric_inlock();
 }
 
 SharedSemiFuture<void> ReplicationCoordinatorImpl::WaiterList::add_inlock(
     const OpTime& opTime, boost::optional<WriteConcernOptions> wc) {
     auto pf = makePromiseFuture<void>();
     _waiters.emplace(opTime, std::make_shared<Waiter>(std::move(pf.promise), std::move(wc)));
+    _updateMetric_inlock();
     return std::move(pf.future);
 }
 
@@ -237,6 +252,7 @@ bool ReplicationCoordinatorImpl::WaiterList::remove_inlock(SharedWaiterHandle wa
     for (auto iter = _waiters.begin(); iter != _waiters.end(); iter++) {
         if (iter->second == waiter) {
             _waiters.erase(iter);
+            _updateMetric_inlock();
             return true;
         }
     }
@@ -260,6 +276,7 @@ void ReplicationCoordinatorImpl::WaiterList::setValueIf_inlock(Func&& func,
             it = _waiters.erase(it);
         }
     }
+    _updateMetric_inlock();
 }
 
 void ReplicationCoordinatorImpl::WaiterList::setValueAll_inlock() {
@@ -267,6 +284,7 @@ void ReplicationCoordinatorImpl::WaiterList::setValueAll_inlock() {
         waiter->promise.emplaceValue();
     }
     _waiters.clear();
+    _updateMetric_inlock();
 }
 
 void ReplicationCoordinatorImpl::WaiterList::setErrorAll_inlock(Status status) {
@@ -275,6 +293,7 @@ void ReplicationCoordinatorImpl::WaiterList::setErrorAll_inlock(Status status) {
         waiter->promise.setError(status);
     }
     _waiters.clear();
+    _updateMetric_inlock();
 }
 
 namespace {
@@ -330,6 +349,8 @@ ReplicationCoordinatorImpl::ReplicationCoordinatorImpl(
       _topCoord(std::move(topCoord)),
       _replExecutor(std::move(executor)),
       _externalState(std::move(externalState)),
+      _replicationWaiterList(replicationWaiterListMetric),
+      _opTimeWaiterList(opTimeWaiterListMetric),
       _inShutdown(false),
       _memberState(MemberState::RS_STARTUP),
       _rsConfigState(kConfigPreStart),
@@ -423,6 +444,10 @@ executor::TaskExecutor::CallbackHandle ReplicationCoordinatorImpl::getCatchupTak
     return _catchupTakeoverCbh;
 }
 
+int64_t ReplicationCoordinatorImpl::getLastHorizonChange_forTest() const {
+    return _lastHorizonTopologyChange;
+}
+
 OpTime ReplicationCoordinatorImpl::getCurrentCommittedSnapshotOpTime() const {
     stdx::lock_guard<Latch> lk(_mutex);
     return _getCurrentCommittedSnapshotOpTime_inlock();
@@ -2286,7 +2311,9 @@ long long ReplicationCoordinatorImpl::_calculateRemainingQuiesceTimeMillis() con
 }
 
 std::shared_ptr<HelloResponse> ReplicationCoordinatorImpl::_makeHelloResponse(
-    boost::optional<StringData> horizonString, WithLock lock, const bool hasValidConfig) const {
+    const boost::optional<std::string>& horizonString,
+    WithLock lock,
+    const bool hasValidConfig) const {
 
     uassert(ShutdownInProgressQuiesceInfo(_calculateRemainingQuiesceTimeMillis()),
             kQuiesceModeShutdownMessage,
@@ -2334,7 +2361,7 @@ SharedSemiFuture<ReplicationCoordinatorImpl::SharedHelloResponse>
 ReplicationCoordinatorImpl::_getHelloResponseFuture(
     WithLock lk,
     const SplitHorizon::Parameters& horizonParams,
-    boost::optional<StringData> horizonString,
+    const boost::optional<std::string>& horizonString,
     boost::optional<TopologyVersion> clientTopologyVersion) {
 
     uassert(ShutdownInProgressQuiesceInfo(_calculateRemainingQuiesceTimeMillis()),
@@ -2366,6 +2393,10 @@ ReplicationCoordinatorImpl::_getHelloResponseFuture(
             prevCounter <= topologyVersionCounter);
 
     if (prevCounter < topologyVersionCounter) {
+        uassert(ErrorCodes::SplitHorizonChange,
+                "Stale horizon detected, we have since received a reconfig that changed the "
+                "horizon mappings.",
+                prevCounter >= _lastHorizonTopologyChange);
         // The received hello command contains a stale topology version so we respond
         // immediately with a more current topology version.
         return SharedSemiFuture<SharedHelloResponse>(
@@ -2398,11 +2429,11 @@ ReplicationCoordinatorImpl::getHelloResponseFuture(
     return _getHelloResponseFuture(lk, horizonParams, horizonString, clientTopologyVersion);
 }
 
-boost::optional<StringData> ReplicationCoordinatorImpl::_getHorizonString(
+boost::optional<std::string> ReplicationCoordinatorImpl::_getHorizonString(
     WithLock, const SplitHorizon::Parameters& horizonParams) const {
     const auto myState = _topCoord->getMemberState();
     const bool hasValidConfig = _rsConfig.isInitialized() && !myState.removed();
-    boost::optional<StringData> horizonString;
+    boost::optional<std::string> horizonString;
     if (hasValidConfig) {
         const auto& self = _rsConfig.getMemberAt(_selfIndex);
         horizonString = self.determineHorizon(horizonParams);
@@ -2643,19 +2674,10 @@ ReplicationCoordinatorImpl::AutoGetRstlForStepUpStepDown::AutoGetRstlForStepUpSt
         auto lockerInfo = opCtx->lockState()->getLockerInfo(CurOp::get(opCtx)->getLockStatsBase());
         BSONObjBuilder lockRep;
         lockerInfo->stats.report(&lockRep);
-
-        LOGV2_FATAL_CONTINUE(
-            5675600,
-            "Time out exceeded waiting for RSTL, stepUp/stepDown is not possible thus "
-            "calling abort() to allow cluster to progress",
-            "lockRep"_attr = lockRep.obj());
-
-#if defined(MONGO_STACKTRACE_CAN_DUMP_ALL_THREADS)
-        // Dump the stack of each thread.
-        printAllThreadStacksBlocking();
-#endif
-
-        fassertFailed(7152000);
+        LOGV2_FATAL(5675600,
+                    "Time out exceeded waiting for RSTL, stepUp/stepDown is not possible thus "
+                    "calling abort() to allow cluster to progress",
+                    "lockRep"_attr = lockRep.obj());
     });
 };
 
@@ -4442,6 +4464,8 @@ void ReplicationCoordinatorImpl::_errorOnPromisesIfHorizonChanged(WithLock lk,
             promise->setError({ErrorCodes::SplitHorizonChange,
                                "Received a reconfig that changed the horizon mappings."});
         }
+        _topCoord->incrementTopologyVersion();
+        _lastHorizonTopologyChange = _topCoord->getTopologyVersion().getCounter();
         _sniToValidConfigPromiseMap.clear();
         HelloMetrics::get(opCtx)->resetNumAwaitingTopologyChanges();
     }
@@ -4456,6 +4480,10 @@ void ReplicationCoordinatorImpl::_errorOnPromisesIfHorizonChanged(WithLock lk,
                 promise->setError({ErrorCodes::SplitHorizonChange,
                                    "Received a reconfig that changed the horizon mappings."});
             }
+            // Increment topology version to mark a horizon change, since a reconfig doesn't
+            // increment the topology version until the end.
+            _topCoord->incrementTopologyVersion();
+            _lastHorizonTopologyChange = _topCoord->getTopologyVersion().getCounter();
             _createHorizonTopologyChangePromiseMapping(lk);
             HelloMetrics::get(opCtx)->resetNumAwaitingTopologyChanges();
         }
@@ -4476,7 +4504,7 @@ void ReplicationCoordinatorImpl::_fulfillTopologyChangePromise(WithLock lock) {
                 Status(ShutdownInProgressQuiesceInfo(_calculateRemainingQuiesceTimeMillis()),
                        kQuiesceModeShutdownMessage));
         } else {
-            StringData horizonString = iter->first;
+            boost::optional<std::string> horizonString = iter->first;
             auto response = _makeHelloResponse(horizonString, lock, hasValidConfig);
             // Fulfill the promise and replace with a new one for future waiters.
             iter->second->emplaceValue(response);
@@ -4496,7 +4524,8 @@ void ReplicationCoordinatorImpl::_fulfillTopologyChangePromise(WithLock lock) {
                                    "The original request horizon parameter does not exist in the "
                                    "current replica set config"});
             } else {
-                const auto horizon = sni.empty() ? SplitHorizon::kDefaultHorizon : iter->second;
+                const boost::optional<std::string> horizon =
+                    sni.empty() ? SplitHorizon::kDefaultHorizon.toString() : iter->second;
                 const auto response = _makeHelloResponse(horizon, lock, hasValidConfig);
                 promise->emplaceValue(response);
             }
diff --git a/src/mongo/db/repl/replication_coordinator_impl.h b/src/mongo/db/repl/replication_coordinator_impl.h
index 8c207c70958..4da218cec58 100644
--- a/src/mongo/db/repl/replication_coordinator_impl.h
+++ b/src/mongo/db/repl/replication_coordinator_impl.h
@@ -468,6 +468,11 @@ public:
     executor::TaskExecutor::CallbackHandle getCatchupTakeoverCbh_forTest() const;
 
     /**
+     * Returns the cached horizon topology version from most recent SplitHorizonChange.
+     */
+    int64_t getLastHorizonChange_forTest() const;
+
+    /**
      * Simple wrappers around _setLastOptime to make it easier to test.
      */
     Status setLastAppliedOptime_forTest(long long cfgVer,
@@ -764,6 +769,9 @@ private:
 
     class WaiterList {
     public:
+        WaiterList() = delete;
+        WaiterList(Atomic64Metric& waiterCountMetric);
+
         // Adds waiter into the list.
         void add_inlock(const OpTime& opTime, SharedWaiterHandle waiter);
         // Adds a waiter into the list and returns the future of the waiter's promise.
@@ -781,8 +789,13 @@ private:
         void setErrorAll_inlock(Status status);
 
     private:
+        void _updateMetric_inlock();
+
         // Waiters sorted by OpTime.
         std::multimap<OpTime, SharedWaiterHandle> _waiters;
+        // We keep a separate count outside _waiters.size() in order to avoid having to
+        // take a lock to read the metric.
+        Atomic64Metric& _waiterCountMetric;
     };
 
     enum class HeartbeatState { kScheduled = 0, kSent = 1 };
@@ -1380,9 +1393,8 @@ private:
      * Fills a HelloResponse with the appropriate replication related fields. horizonString
      * should be passed in if hasValidConfig is true.
      */
-    std::shared_ptr<HelloResponse> _makeHelloResponse(boost::optional<StringData> horizonString,
-                                                      WithLock,
-                                                      bool hasValidConfig) const;
+    std::shared_ptr<HelloResponse> _makeHelloResponse(
+        const boost::optional<std::string>& horizonString, WithLock, bool hasValidConfig) const;
 
     /**
      * Creates a semi-future for HelloResponse. horizonString should be passed in if and only if
@@ -1391,14 +1403,14 @@ private:
     virtual SharedSemiFuture<SharedHelloResponse> _getHelloResponseFuture(
         WithLock,
         const SplitHorizon::Parameters& horizonParams,
-        boost::optional<StringData> horizonString,
+        const boost::optional<std::string>& horizonString,
         boost::optional<TopologyVersion> clientTopologyVersion);
 
     /**
      * Returns the horizon string by parsing horizonParams if the node is a valid member of the
      * replica set. Otherwise, return boost::none.
      */
-    boost::optional<StringData> _getHorizonString(
+    boost::optional<std::string> _getHorizonString(
         WithLock, const SplitHorizon::Parameters& horizonParams) const;
 
     /**
@@ -1836,6 +1848,9 @@ private:
     // The cached value of the 'counter' field in the server's TopologyVersion.
     AtomicWord<int64_t> _cachedTopologyVersionCounter;  // (S)
 
+    // The cached value of the topology from the most recent SplitHorizonChange.
+    int64_t _lastHorizonTopologyChange{-1};  // (M)
+
     // This should be set during sharding initialization.
     boost::optional<bool> _wasCWWCSetOnConfigServerOnStartup;
 
diff --git a/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp b/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp
index 98a6f722a73..ce5c77a1e7d 100644
--- a/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl_heartbeat.cpp
@@ -77,6 +77,7 @@ MONGO_FAIL_POINT_DEFINE(blockHeartbeatStepdown);
 MONGO_FAIL_POINT_DEFINE(blockHeartbeatReconfigFinish);
 MONGO_FAIL_POINT_DEFINE(hangAfterTrackingNewHandleInHandleHeartbeatResponseForTest);
 MONGO_FAIL_POINT_DEFINE(waitForPostActionCompleteInHbReconfig);
+MONGO_FAIL_POINT_DEFINE(pauseInHandleHeartbeatResponse);
 
 }  // namespace
 
@@ -184,6 +185,12 @@ void ReplicationCoordinatorImpl::handleHeartbeatResponse_forTest(BSONObj respons
 
 void ReplicationCoordinatorImpl::_handleHeartbeatResponse(
     const executor::TaskExecutor::RemoteCommandCallbackArgs& cbData, const std::string& setName) {
+    pauseInHandleHeartbeatResponse.executeIf(
+        [](const BSONObj& data) { pauseInHandleHeartbeatResponse.pauseWhileSet(); },
+        [&cbData](const BSONObj& data) -> bool {
+            StringData dtarget = data["target"].valueStringDataSafe();
+            return dtarget == cbData.request.target.toString();
+        });
     stdx::unique_lock<Latch> lk(_mutex);
 
     // remove handle from queued heartbeats
@@ -194,7 +201,15 @@ void ReplicationCoordinatorImpl::_handleHeartbeatResponse(
     Status responseStatus = cbData.response.status;
     const HostAndPort& target = cbData.request.target;
 
-    if (responseStatus == ErrorCodes::CallbackCanceled) {
+    // It is possible that the callback was canceled after handleHeartbeatResponse was called but
+    // before it got the lock above.
+    //
+    // In this case, the responseStatus will be OK and we can process the heartbeat.  However, if
+    // we do so, cancelling heartbeats no longer establishes a barrier after which all heartbeats
+    // processed are "new" (sent subsequent to the cancel), which is something we care about for
+    // catchup takeover.  So if we detect this situation (by checking if the handle was canceled)
+    // we will NOT process the 'stale' heartbeat.
+    if (responseStatus == ErrorCodes::CallbackCanceled || cbData.myHandle.isCanceled()) {
         LOGV2_FOR_HEARTBEATS(4615619,
                              2,
                              "Received response to heartbeat (requestId: {requestId}) from "
diff --git a/src/mongo/db/repl/replication_coordinator_impl_test.cpp b/src/mongo/db/repl/replication_coordinator_impl_test.cpp
index 9f0f0531710..f0ece048ca3 100644
--- a/src/mongo/db/repl/replication_coordinator_impl_test.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl_test.cpp
@@ -85,6 +85,9 @@
 
 namespace mongo {
 namespace repl {
+extern Atomic64Metric replicationWaiterListMetric;
+extern Atomic64Metric opTimeWaiterListMetric;
+
 namespace {
 
 using executor::NetworkInterfaceMock;
@@ -4074,6 +4077,131 @@ TEST_F(ReplCoordTest, AwaitHelloResponseReturnsErrorOnHorizonChange) {
     getHelloThread.join();
 }
 
+TEST_F(ReplCoordTest, ServerUassertAfterStaleHorizonTopology) {
+    init();
+    assertStartSuccess(BSON("_id"
+                            << "mySet"
+                            << "version" << 2 << "members"
+                            << BSON_ARRAY(BSON("host"
+                                               << "node1:12345"
+                                               << "_id" << 0)
+                                          << BSON("host"
+                                                  << "node2:12345"
+                                                  << "_id" << 1))),
+                       HostAndPort("node1", 12345));
+
+    // Become primary.
+    ASSERT_OK(getReplCoord()->setFollowerMode(MemberState::RS_SECONDARY));
+    replCoordSetMyLastAppliedAndDurableOpTime(OpTimeWithTermOne(100, 1), Date_t() + Seconds(100));
+    simulateSuccessfulV1Election();
+    ASSERT(getReplCoord()->getMemberState().primary());
+
+    auto maxAwaitTime = Milliseconds(5000);
+    auto deadline = getNet()->now() + maxAwaitTime;
+    auto opCtx = makeOperationContext();
+
+    auto topologyVersionBeforeReconfig = getTopoCoord().getTopologyVersion();
+    // awaitHelloResponse blocks and waits on a future when the request TopologyVersion equals
+    // the current TopologyVersion of the server.
+    stdx::thread getHelloThread([&] {
+        ASSERT_THROWS_CODE(
+            awaitHelloWithNewOpCtx(getReplCoord(), topologyVersionBeforeReconfig, {}, deadline),
+            AssertionException,
+            ErrorCodes::SplitHorizonChange);
+    });
+
+    auto lastHorizonBeforeReconfig = getReplCoord()->getLastHorizonChange_forTest();
+    ASSERT_EQUALS(lastHorizonBeforeReconfig, -1);
+
+    BSONObjBuilder garbage;
+    ReplSetReconfigArgs args;
+    // Use force to bypass the oplog commitment check, which we're not worried about testing here.
+    args.force = true;
+    // Do a reconfig that changes the SplitHorizon and also adds a third node. This should respond
+    // to all waiting hello requests with an error.
+    args.newConfigObj = BSON("_id"
+                             << "mySet"
+                             << "version" << 3 << "protocolVersion" << 1 << "members"
+                             << BSON_ARRAY(BSON("_id" << 0 << "host"
+                                                      << "node1:12345"
+                                                      << "priority" << 3 << "horizons"
+                                                      << BSON("testhorizon"
+                                                              << "test.monkey.example.com:24"))
+                                           << BSON("_id" << 1 << "host"
+                                                         << "node2:12345"
+                                                         << "horizons"
+                                                         << BSON("testhorizon"
+                                                                 << "test.giraffe.example.com:25"))
+                                           << BSON("_id"
+                                                   << 2 << "host"
+                                                   << "node3:12345"
+                                                   << "horizons"
+                                                   << BSON("testhorizon"
+                                                           << "test.elephant.example.com:26"))));
+    stdx::thread reconfigThread([&] {
+        Status status(ErrorCodes::InternalError, "Not Set");
+        status = getReplCoord()->processReplSetReconfig(opCtx.get(), args, &garbage);
+        ASSERT_OK(status);
+    });
+    replyToReceivedHeartbeatV1();
+    reconfigThread.join();
+    getHelloThread.join();
+
+    // After reconfig, the last horizon change topology counter should come out greater than the
+    // input topology counter , but less than the output topology counter.
+    ASSERT_GREATER_THAN(getReplCoord()->getLastHorizonChange_forTest(),
+                        topologyVersionBeforeReconfig.getCounter());
+    ASSERT_GREATER_THAN(getTopoCoord().getTopologyVersion().getCounter(),
+                        getReplCoord()->getLastHorizonChange_forTest());
+    ASSERT_GREATER_THAN(getReplCoord()->getLastHorizonChange_forTest(), lastHorizonBeforeReconfig);
+
+    // Send hello with a TopologyVersion older than the TopologyVersion of the last horizon change.
+    auto requestTopologyVersion =
+        TopologyVersion(getTopoCoord().getTopologyVersion().getProcessId(),
+                        getReplCoord()->getLastHorizonChange_forTest() - 1);
+
+    // AwaitHelloResponse should throw uassert with SplitHorizonChange if topology version
+    // corresponds to a stale horizon.
+    ASSERT_THROWS_CODE(awaitHelloWithNewOpCtx(getReplCoord(), requestTopologyVersion, {}, deadline),
+                       DBException,
+                       ErrorCodes::SplitHorizonChange);
+
+    // Send hello with a TopologyVersion version equal to the TopologyVersion of the last horizon
+    // change.
+    auto expectedTopologyVersion = getTopoCoord().getTopologyVersion();
+    requestTopologyVersion = TopologyVersion(expectedTopologyVersion.getProcessId(),
+                                             getReplCoord()->getLastHorizonChange_forTest());
+
+    ASSERT_GREATER_THAN(expectedTopologyVersion.getCounter(), requestTopologyVersion.getCounter());
+    // AwaitHelloResponse should return with a helloResponse that matches expectedTopologyVersion.
+    // Since expectedTopologyVersion > requestTopologyVersion, the call is non-blocking and will
+    // return immediately.
+    auto response = awaitHelloWithNewOpCtx(getReplCoord(), requestTopologyVersion, {}, deadline);
+    auto responseTopologyVersion = response->getTopologyVersion();
+    ASSERT_EQUALS(responseTopologyVersion->getCounter(), expectedTopologyVersion.getCounter());
+    ASSERT_EQUALS(responseTopologyVersion->getProcessId(), expectedTopologyVersion.getProcessId());
+
+    // Setup instance where lastHorizonChange topology counter < request topology counter < server
+    // topology counter. For server topology counter to be greater than both request's and horizon,
+    // we must increment server topology.
+    getTopoCoord().incrementTopologyVersion();
+    expectedTopologyVersion = getTopoCoord().getTopologyVersion();
+    // Send hello with a TopologyVersion version greater than the TopologyVersion of the last
+    // horizon change.
+    requestTopologyVersion = TopologyVersion(getTopoCoord().getTopologyVersion().getProcessId(),
+                                             getReplCoord()->getLastHorizonChange_forTest() + 1);
+
+    ASSERT_GREATER_THAN(expectedTopologyVersion.getCounter(), requestTopologyVersion.getCounter());
+    ASSERT_GREATER_THAN(requestTopologyVersion.getCounter(),
+                        getReplCoord()->getLastHorizonChange_forTest());
+
+    // AwaitHelloResponse should return with a helloResponse that matches expectedTopologyVersion.
+    response = awaitHelloWithNewOpCtx(getReplCoord(), requestTopologyVersion, {}, deadline);
+    responseTopologyVersion = response->getTopologyVersion();
+    ASSERT_EQUALS(responseTopologyVersion->getCounter(), expectedTopologyVersion.getCounter());
+    ASSERT_EQUALS(responseTopologyVersion->getProcessId(), expectedTopologyVersion.getProcessId());
+}
+
 TEST_F(ReplCoordTest, NonAwaitableHelloReturnsNoConfigsOnNodeWithUninitializedConfig) {
     start();
     auto opCtx = makeOperationContext();
@@ -4719,6 +4847,9 @@ TEST_F(ReplCoordTest, AwaitHelloRespondsCorrectlyWhenNodeRemovedAndReadded) {
     });
     waitForHelloFailPoint->waitForTimesEntered(timesEnteredFailPoint + 2);
 
+    auto lastHorizonBeforeReconfig = getReplCoord()->getLastHorizonChange_forTest();
+    ASSERT_EQUALS(lastHorizonBeforeReconfig, -1);
+
     const auto newHorizonNodeOne = "newhorizon.com:100";
     const auto newHorizonNodeTwo = "newhorizon.com:200";
 
@@ -4743,10 +4874,22 @@ TEST_F(ReplCoordTest, AwaitHelloRespondsCorrectlyWhenNodeRemovedAndReadded) {
     });
     replyToReceivedHeartbeatV1();
     reconfigThread.join();
+
     ASSERT_OK(
         getReplCoord()->waitForMemberState(opCtx.get(), MemberState::RS_SECONDARY, Seconds(1)));
     getHelloThread.join();
 
+    ASSERT_GREATER_THAN(getReplCoord()->getLastHorizonChange_forTest(), lastHorizonBeforeReconfig);
+    // Send hello with a TopologyVersion older than the TopologyVersion of the last horizon change.
+    auto requestTopologyVersion =
+        TopologyVersion(getTopoCoord().getTopologyVersion().getProcessId(),
+                        getReplCoord()->getLastHorizonChange_forTest() - 1);
+    // AwaitHelloResponse should throw uassert with SplitHorizonChange if topology version
+    // corresponds to a stale horizon.
+    ASSERT_THROWS_CODE(awaitHelloWithNewOpCtx(getReplCoord(), requestTopologyVersion, {}, deadline),
+                       DBException,
+                       ErrorCodes::SplitHorizonChange);
+
     stdx::thread getHelloThreadNewHorizon([&] {
         const auto expectedTopologyVersion = getTopoCoord().getTopologyVersion();
         // Sending a hello on the rejoined node should return the appropriate horizon view.
@@ -5758,6 +5901,140 @@ TEST_F(ReplCoordTest,
     awaiter.reset();
 }
 
+
+// We need to wait for replication to start waiting before the waiter metric increases.  We
+// return the metric value from the function for the convenience of the assert macro below,
+// which allows us to get a nice assert message without repeating the value.
+template <typename T, typename U>
+U expectMetricIncreaseTo(T& metric, U value) {
+    // If this doesn't go in 10 seconds, something's seriously wrong; even if just a slow machine,
+    // the test will likely fail anyway.
+    constexpr auto timeout = Seconds(10);
+    const auto deadline = Date_t::now() + timeout;
+    U lastValue = metric.get();
+    U curValue = metric.get();
+    while (curValue < value && Date_t::now() <= deadline) {
+        sleepFor(Milliseconds{10});
+        curValue = metric.get();
+        lastValue = curValue;
+    }
+    return curValue;
+}
+
+#define ASSERT_METRIC_INCREASE_TO(metric, value) \
+    ASSERT_EQ(expectMetricIncreaseTo(metric, value), value)
+
+TEST_F(ReplCoordTest, ReplicationWaiterMetrics) {
+    assertStartSuccess(BSON("_id"
+                            << "mySet"
+                            << "version" << 2 << "members"
+                            << BSON_ARRAY(BSON("host"
+                                               << "node1:12345"
+                                               << "_id" << 0)
+                                          << BSON("host"
+                                                  << "node2:12345"
+                                                  << "_id" << 1)
+                                          << BSON("host"
+                                                  << "node3:12345"
+                                                  << "_id" << 2))),
+                       HostAndPort("node1", 12345));
+    ASSERT_OK(getReplCoord()->setFollowerMode(MemberState::RS_SECONDARY));
+    replCoordSetMyLastAppliedOpTime(OpTimeWithTermOne(100, 1), Date_t() + Seconds(100));
+    replCoordSetMyLastDurableOpTime(OpTimeWithTermOne(100, 1), Date_t() + Seconds(100));
+    simulateSuccessfulV1Election();
+
+    ReplicationAwaiter awaiter1(getReplCoord(), getServiceContext());
+    ReplicationAwaiter awaiter2(getReplCoord(), getServiceContext());
+
+    OpTimeWithTermOne time1(100, 1);
+    OpTimeWithTermOne time2(100, 2);
+    OpTimeWithTermOne time3(100, 3);
+
+    WriteConcernOptions writeConcern;
+    writeConcern.wTimeout = WriteConcernOptions::kNoTimeout;
+    writeConcern.w = 2;
+
+    WriteConcernOptions writeConcernLocal;
+    writeConcernLocal.wTimeout = WriteConcernOptions::kNoTimeout;
+    writeConcernLocal.w = 1;
+    writeConcernLocal.syncMode = WriteConcernOptions::SyncMode::UNSET;
+
+    // 2 waiters waiting for 2 nodes to reach time1.
+    awaiter1.setOpTime(time1);
+    awaiter1.setWriteConcern(writeConcern);
+    awaiter1.start();
+    ASSERT_METRIC_INCREASE_TO(replicationWaiterListMetric, 1);
+    ASSERT_EQ(opTimeWaiterListMetric.get(), 0);
+
+    awaiter2.setOpTime(time1);
+    awaiter2.setWriteConcern(writeConcern);
+    awaiter2.start();
+    ASSERT_METRIC_INCREASE_TO(replicationWaiterListMetric, 2);
+    ASSERT_EQ(opTimeWaiterListMetric.get(), 0);
+
+    replCoordSetMyLastAppliedOpTime(time1, Date_t() + Seconds(100));
+    replCoordSetMyLastDurableOpTime(time1, Date_t() + Seconds(100));
+    ASSERT_OK(getReplCoord()->setLastAppliedOptime_forTest(2, 1, time1));
+    ReplicationCoordinator::StatusAndDuration statusAndDur = awaiter1.getResult();
+    ASSERT_OK(statusAndDur.status);
+    statusAndDur = awaiter2.getResult();
+    ASSERT_OK(statusAndDur.status);
+    ASSERT_EQ(replicationWaiterListMetric.get(), 0);
+    ASSERT_EQ(opTimeWaiterListMetric.get(), 0);
+    awaiter1.reset();
+    awaiter2.reset();
+
+    // 2 nodes waiting for time2, but only locally.
+    awaiter1.setOpTime(time2);
+    awaiter1.setWriteConcern(writeConcernLocal);
+    awaiter1.start();
+    ASSERT_METRIC_INCREASE_TO(opTimeWaiterListMetric, 1);
+    ASSERT_EQ(replicationWaiterListMetric.get(), 0);
+
+    awaiter2.setOpTime(time2);
+    awaiter2.setWriteConcern(writeConcernLocal);
+    awaiter2.start();
+    ASSERT_METRIC_INCREASE_TO(opTimeWaiterListMetric, 2);
+    ASSERT_EQ(replicationWaiterListMetric.get(), 0);
+
+    replCoordSetMyLastAppliedOpTime(time2, Date_t() + Seconds(100));
+    replCoordSetMyLastDurableOpTime(time2, Date_t() + Seconds(100));
+    ASSERT_OK(getReplCoord()->setLastAppliedOptime_forTest(2, 1, time2));
+    statusAndDur = awaiter1.getResult();
+    ASSERT_OK(statusAndDur.status);
+    statusAndDur = awaiter2.getResult();
+    ASSERT_OK(statusAndDur.status);
+    awaiter1.reset();
+    awaiter2.reset();
+    ASSERT_EQ(replicationWaiterListMetric.get(), 0);
+    ASSERT_EQ(opTimeWaiterListMetric.get(), 0);
+
+    // 2 nodes waiting for time3, one local one not, but we're going to step down first.
+    awaiter1.setOpTime(time3);
+    awaiter1.setWriteConcern(writeConcernLocal);
+    awaiter1.start();
+    ASSERT_METRIC_INCREASE_TO(opTimeWaiterListMetric, 1);
+    ASSERT_EQ(replicationWaiterListMetric.get(), 0);
+
+    awaiter2.setOpTime(time3);
+    awaiter2.setWriteConcern(writeConcern);
+    awaiter2.start();
+    ASSERT_METRIC_INCREASE_TO(replicationWaiterListMetric, 1);
+    ASSERT_EQ(opTimeWaiterListMetric.get(), 1);
+
+    const auto opCtx = makeOperationContext();
+    getReplCoord()->stepDown(opCtx.get(), true, Milliseconds(0), Milliseconds(1000));
+
+    statusAndDur = awaiter1.getResult();
+    ASSERT_EQUALS(ErrorCodes::PrimarySteppedDown, statusAndDur.status);
+    statusAndDur = awaiter2.getResult();
+    ASSERT_EQUALS(ErrorCodes::PrimarySteppedDown, statusAndDur.status);
+    awaiter1.reset();
+    awaiter2.reset();
+    ASSERT_EQ(replicationWaiterListMetric.get(), 0);
+    ASSERT_EQ(opTimeWaiterListMetric.get(), 0);
+}
+
 TEST_F(ReplCoordTest,
        NodeReturnsFromMajorityWriteConcernOnlyOnceTheWriteAppearsInACommittedSnapShot) {
     // Test that we can satisfy majority write concern can only be
diff --git a/src/mongo/db/repl/split_horizon.cpp b/src/mongo/db/repl/split_horizon.cpp
index f588445398b..6201d5f9ba5 100644
--- a/src/mongo/db/repl/split_horizon.cpp
+++ b/src/mongo/db/repl/split_horizon.cpp
@@ -181,7 +181,8 @@ auto SplitHorizon::getParameters(const Client* const client) -> Parameters {
     return getSplitHorizonParameters(*client);
 }
 
-StringData SplitHorizon::determineHorizon(const SplitHorizon::Parameters& horizonParameters) const {
+std::string SplitHorizon::determineHorizon(
+    const SplitHorizon::Parameters& horizonParameters) const {
     if (horizonParameters.sniName) {
         const auto sniName = *horizonParameters.sniName;
         const auto found = _reverseHostMapping.find(sniName);
@@ -189,7 +190,7 @@ StringData SplitHorizon::determineHorizon(const SplitHorizon::Parameters& horizo
             return found->second;
         }
     }
-    return kDefaultHorizon;
+    return kDefaultHorizon.toString();
 }
 
 void SplitHorizon::toBSON(BSONObjBuilder& configBuilder) const {
diff --git a/src/mongo/db/repl/split_horizon.h b/src/mongo/db/repl/split_horizon.h
index b597cc7dec3..e9e11f1bcbd 100644
--- a/src/mongo/db/repl/split_horizon.h
+++ b/src/mongo/db/repl/split_horizon.h
@@ -91,7 +91,7 @@ public:
      * Gets the horizon name for which the parameters (captured during the first `isMaster`)
      * correspond.
      */
-    StringData determineHorizon(const Parameters& horizonParameters) const;
+    std::string determineHorizon(const Parameters& horizonParameters) const;
 
     const HostAndPort& getHostAndPort(StringData horizon) const {
         invariant(!_forwardMapping.empty());
diff --git a/src/mongo/db/repl/split_horizon_test.cpp b/src/mongo/db/repl/split_horizon_test.cpp
index 486c43f2ade..9f728012614 100644
--- a/src/mongo/db/repl/split_horizon_test.cpp
+++ b/src/mongo/db/repl/split_horizon_test.cpp
@@ -110,7 +110,7 @@ TEST(SplitHorizonTesting, determineHorizon) {
         const auto& input = test.input;
 
         const std::string witness =
-            SplitHorizon(input.forwardMapping).determineHorizon(input.horizonParameters).toString();
+            SplitHorizon(input.forwardMapping).determineHorizon(input.horizonParameters);
         ASSERT_EQUALS(witness, expected);
     }
 
diff --git a/src/mongo/db/repl/storage_timestamp_test.cpp b/src/mongo/db/repl/storage_timestamp_test.cpp
index acb142f339e..6431be37aed 100644
--- a/src/mongo/db/repl/storage_timestamp_test.cpp
+++ b/src/mongo/db/repl/storage_timestamp_test.cpp
@@ -90,7 +90,6 @@
 #include "mongo/db/transaction_participant_gen.h"
 #include "mongo/db/vector_clock_mutable.h"
 #include "mongo/dbtests/dbtests.h"
-#include "mongo/idl/server_parameter_test_util.h"
 #include "mongo/logv2/log.h"
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/stdx/future.h"
diff --git a/src/mongo/db/repl/topology_coordinator.cpp b/src/mongo/db/repl/topology_coordinator.cpp
index 630fefa541c..77405c9a84f 100644
--- a/src/mongo/db/repl/topology_coordinator.cpp
+++ b/src/mongo/db/repl/topology_coordinator.cpp
@@ -339,6 +339,12 @@ HostAndPort TopologyCoordinator::_chooseNearbySyncSource(Date_t now,
     //
     // This loop attempts to set 'closestIndex', to select a viable candidate.
     for (int attempts = 0; attempts < 2; ++attempts) {
+        if (attempts == 1) {
+            LOGV2_INFO(
+                8423402,
+                "Failed to select a sync source on the first attempt. Starting second attempt");
+        }
+
         for (size_t candidateIndex = 0; candidateIndex < _memberData.size(); candidateIndex++) {
             if (!_isEligibleSyncSource(candidateIndex,
                                        now,
@@ -359,18 +365,21 @@ HostAndPort TopologyCoordinator::_chooseNearbySyncSource(Date_t now,
             const auto syncSourceCandidate = _rsConfig.getMemberAt(candidateIndex).getHostAndPort();
             const auto closestNode = _rsConfig.getMemberAt(closestIndex).getHostAndPort();
 
+            LOGV2_INFO(8423401,
+                       "Sync source candidate is eligible",
+                       "syncSourceCandidate"_attr = syncSourceCandidate);
+
             // Do not update 'closestIndex' if the candidate is not the closest node we've seen.
             auto syncSourceCandidatePing = _getPing(syncSourceCandidate);
             auto closestPing = _getPing(closestNode);
             if (syncSourceCandidatePing > closestPing) {
-                LOGV2_DEBUG(3873114,
-                            2,
-                            "Cannot select sync source with higher latency than the best "
-                            "candidate",
-                            "syncSourceCandidate"_attr = syncSourceCandidate,
-                            "syncSourceCandidatePing"_attr = syncSourceCandidatePing,
-                            "closestNode"_attr = closestNode,
-                            "closestPing"_attr = closestPing);
+                LOGV2_INFO(3873114,
+                           "Cannot select sync source with higher latency than the best "
+                           "candidate",
+                           "syncSourceCandidate"_attr = syncSourceCandidate,
+                           "syncSourceCandidatePing"_attr = syncSourceCandidatePing,
+                           "closestNode"_attr = closestNode,
+                           "closestPing"_attr = closestPing);
                 continue;
             }
             closestIndex = candidateIndex;
@@ -435,18 +444,16 @@ bool TopologyCoordinator::_isEligibleSyncSource(int candidateIndex,
 
     // Candidate must be up to be considered.
     if (!memberData.up()) {
-        LOGV2_DEBUG(3873106,
-                    2,
-                    "Cannot select sync source because it is not up",
-                    "syncSourceCandidate"_attr = syncSourceCandidate);
+        LOGV2_INFO(3873106,
+                   "Cannot select sync source because it is not up",
+                   "syncSourceCandidate"_attr = syncSourceCandidate);
         return false;
     }
     // Candidate must be PRIMARY or SECONDARY state to be considered.
     if (!memberData.getState().readable()) {
-        LOGV2_DEBUG(3873107,
-                    2,
-                    "Cannot select sync source because it is not readable",
-                    "syncSourceCandidate"_attr = syncSourceCandidate);
+        LOGV2_INFO(3873107,
+                   "Cannot select sync source because it is not readable",
+                   "syncSourceCandidate"_attr = syncSourceCandidate);
         return false;
     }
 
@@ -454,11 +461,10 @@ bool TopologyCoordinator::_isEligibleSyncSource(int candidateIndex,
     if (readPreference == ReadPreference::SecondaryOnly ||
         (readPreference == ReadPreference::SecondaryPreferred && firstAttempt)) {
         if (memberData.getState().primary()) {
-            LOGV2_DEBUG(3873101,
-                        2,
-                        "Cannot select sync source because it is a primary and we are "
-                        "looking for a secondary",
-                        "syncSourceCandidate"_attr = syncSourceCandidate);
+            LOGV2_INFO(3873101,
+                       "Cannot select sync source because it is a primary and we are "
+                       "looking for a secondary",
+                       "syncSourceCandidate"_attr = syncSourceCandidate);
             return false;
         }
     }
@@ -467,73 +473,66 @@ bool TopologyCoordinator::_isEligibleSyncSource(int candidateIndex,
     if (firstAttempt) {
         // Candidate must be a voter if we are a voter.
         if (_selfConfig().isVoter() && !memberConfig.isVoter()) {
-            LOGV2_DEBUG(3873108,
-                        2,
-                        "Cannot select sync source because we are a voter and it is not",
-                        "syncSourceCandidate"_attr = syncSourceCandidate);
+            LOGV2_INFO(3873108,
+                       "Cannot select sync source because we are a voter and it is not",
+                       "syncSourceCandidate"_attr = syncSourceCandidate);
             return false;
         }
         // Candidates must not be hidden.
         if (memberConfig.isHidden()) {
-            LOGV2_DEBUG(3873109,
-                        2,
-                        "Cannot select sync source because it is hidden",
-                        "syncSourceCandidate"_attr = syncSourceCandidate);
+            LOGV2_INFO(3873109,
+                       "Cannot select sync source because it is hidden",
+                       "syncSourceCandidate"_attr = syncSourceCandidate);
             return false;
         }
         // Candidates cannot be excessively behind, if we are checking for staleness.
         if (shouldCheckStaleness) {
             const auto oldestSyncOpTime = _getOldestSyncOpTime();
             if (memberData.getHeartbeatAppliedOpTime() < oldestSyncOpTime) {
-                LOGV2_DEBUG(3873110,
-                            2,
-                            "Cannot select sync source because it is too far behind",
-                            "syncSourceCandidate"_attr = syncSourceCandidate,
-                            "syncSourceCandidateOpTime"_attr =
-                                memberData.getHeartbeatAppliedOpTime(),
-                            "oldestAcceptableOpTime"_attr = oldestSyncOpTime);
+                LOGV2_INFO(3873110,
+                           "Cannot select sync source because it is too far behind",
+                           "syncSourceCandidate"_attr = syncSourceCandidate,
+                           "syncSourceCandidateOpTime"_attr =
+                               memberData.getHeartbeatAppliedOpTime(),
+                           "oldestAcceptableOpTime"_attr = oldestSyncOpTime);
                 return false;
             }
         }
         // Candidate must not have a configured delay larger than ours.
         if (_selfConfig().getSecondaryDelay() < memberConfig.getSecondaryDelay()) {
-            LOGV2_DEBUG(3873111,
-                        2,
-                        "Cannot select sync source with larger secondaryDelaySecs than ours",
-                        "syncSourceCandidate"_attr = syncSourceCandidate,
-                        "syncSourceCandidateSecondaryDelaySecs"_attr =
-                            memberConfig.getSecondaryDelay(),
-                        "secondaryDelaySecs"_attr = _selfConfig().getSecondaryDelay());
+            LOGV2_INFO(3873111,
+                       "Cannot select sync source with larger secondaryDelaySecs than ours",
+                       "syncSourceCandidate"_attr = syncSourceCandidate,
+                       "syncSourceCandidateSecondaryDelaySecs"_attr =
+                           memberConfig.getSecondaryDelay(),
+                       "secondaryDelaySecs"_attr = _selfConfig().getSecondaryDelay());
             return false;
         }
     }
     // Candidate must build indexes if we build indexes, to be considered.
     if (_selfConfig().shouldBuildIndexes()) {
         if (!memberConfig.shouldBuildIndexes()) {
-            LOGV2_DEBUG(3873112,
-                        2,
-                        "Cannot select sync source which does not build indexes when we do",
-                        "syncSourceCandidate"_attr = syncSourceCandidate);
+            LOGV2_INFO(3873112,
+                       "Cannot select sync source which does not build indexes when we do",
+                       "syncSourceCandidate"_attr = syncSourceCandidate);
             return false;
         }
     }
     // Only select a candidate that is ahead of me, if we are checking for staleness.
     if (shouldCheckStaleness && memberData.getHeartbeatAppliedOpTime() <= lastOpTimeFetched) {
-        LOGV2_DEBUG(3873113,
-                    1,
-                    "Cannot select sync source which is not ahead of me",
-                    "syncSourceCandidate"_attr = syncSourceCandidate,
-                    "syncSourceCandidateLastAppliedOpTime"_attr =
-                        memberData.getHeartbeatAppliedOpTime().toBSON(),
-                    "lastOpTimeFetched"_attr = lastOpTimeFetched.toBSON());
+        LOGV2_INFO(3873113,
+                   "Cannot select sync source which is not ahead of me",
+                   "syncSourceCandidate"_attr = syncSourceCandidate,
+                   "syncSourceCandidateLastAppliedOpTime"_attr =
+                       memberData.getHeartbeatAppliedOpTime().toBSON(),
+                   "lastOpTimeFetched"_attr = lastOpTimeFetched.toBSON());
         return false;
     }
     // Candidate cannot be denylisted.
     if (_memberIsDenylisted(memberConfig, now)) {
-        LOGV2_DEBUG(3873115,
-                    1,
-                    "Cannot select sync source which is denylisted",
-                    "syncSourceCandidate"_attr = syncSourceCandidate);
+        LOGV2_INFO(3873115,
+                   "Cannot select sync source which is denylisted",
+                   "syncSourceCandidate"_attr = syncSourceCandidate);
         return false;
     }
     // This candidate has passed all tests.
diff --git a/src/mongo/db/repl/topology_version_observer_test.cpp b/src/mongo/db/repl/topology_version_observer_test.cpp
index 6d7d54d57b8..d7ee56b4778 100644
--- a/src/mongo/db/repl/topology_version_observer_test.cpp
+++ b/src/mongo/db/repl/topology_version_observer_test.cpp
@@ -42,7 +42,9 @@
 #include "mongo/db/repl/replication_coordinator_impl.h"
 #include "mongo/db/repl/replication_coordinator_test_fixture.h"
 #include "mongo/db/repl/topology_version_observer.h"
+#include "mongo/logv2/log.h"
 #include "mongo/unittest/barrier.h"
+#include "mongo/unittest/log_test.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/clock_source.h"
@@ -118,6 +120,9 @@ protected:
     const Milliseconds sleepTime = Milliseconds(100);
 
     std::unique_ptr<TopologyVersionObserver> observer;
+
+    unittest::MinimumLoggedSeverityGuard severityGuard{logv2::LogComponent::kDefault,
+                                                       logv2::LogSeverity::Debug(4)};
 };
 
 
@@ -140,11 +145,15 @@ TEST_F(TopologyVersionObserverTest, UpdateCache) {
     auto electionTimeoutWhen = getReplCoord()->getElectionTimeout_forTest();
     simulateSuccessfulV1ElectionWithoutExitingDrainMode(electionTimeoutWhen, opCtx.get());
 
+    auto sleepCounter = 0;
     // Wait for the observer to update its cache
     while (observer->getCached()->getTopologyVersion()->getCounter() ==
            cachedResponse->getTopologyVersion()->getCounter()) {
         sleepFor(sleepTime);
+        // Make sure the test doesn't wait here for longer than 15 seconds.
+        ASSERT_LTE(sleepCounter++, 150);
     }
+    LOGV2(9326401, "Observer topology incremented after successful election");
 
     auto newResponse = observer->getCached();
     ASSERT(newResponse && newResponse->getTopologyVersion());
diff --git a/src/mongo/db/repl/transaction_oplog_application.cpp b/src/mongo/db/repl/transaction_oplog_application.cpp
index a71f8470212..20d8459a18b 100644
--- a/src/mongo/db/repl/transaction_oplog_application.cpp
+++ b/src/mongo/db/repl/transaction_oplog_application.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/db/repl/transaction_oplog_application.h"
 
+#include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/commands/txn_cmds_gen.h"
 #include "mongo/db/concurrency/exception_util.h"
@@ -538,6 +539,8 @@ void _reconstructPreparedTransaction(OperationContext* opCtx,
                                      const OplogEntry& prepareEntry,
                                      repl::OplogApplication::Mode mode) {
     repl::UnreplicatedWritesBlock uwb(opCtx);
+    // The transaction may have been prepared originally with document validation bypassed.
+    DisableDocumentValidation validationDisabler(opCtx);
 
     // Snapshot transaction can never conflict with the PBWM lock.
     opCtx->lockState()->setShouldConflictWithSecondaryBatchApplication(false);