SERVER-21700 Do not relax constraints during steady state replication.

10 files changed, 436 insertions, 52 deletions

diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index 5d0971b65db..6398e3bcb99 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -41,6 +41,7 @@ env.Library(
         'dbcheck',
         'local_oplog_info',
         'repl_coordinator_interface',
+        'repl_server_parameters',
         'repl_settings',
         'timestamp_block',
         '$BUILD_DIR/mongo/base',
@@ -563,6 +564,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/commands/mongod_fsync',
+        'repl_server_parameters',
         'replication_auth',
     ],
 )
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index ef6c546f2c3..d1fb11630dc 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -79,6 +79,7 @@
 #include "mongo/db/repl/local_oplog_info.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/repl/repl_client_info.h"
+#include "mongo/db/repl/repl_server_parameters_gen.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl/timestamp_block.h"
 #include "mongo/db/repl/transaction_oplog_application.h"
@@ -648,6 +649,9 @@ using OpApplyFn = std::function<Status(
 
 struct ApplyOpMetadata {
     OpApplyFn applyFunc;
+    // acceptableErrors are errors we accept for idempotency reasons.  Except for IndexNotFound,
+    // they are only valid in non-steady-state oplog application modes.  IndexNotFound is always
+    // allowed because index builds are not necessarily synchronized between secondary and primary.
     std::set<ErrorCodes::Error> acceptableErrors;
 
     ApplyOpMetadata(OpApplyFn fun) {
@@ -760,7 +764,15 @@ const StringMap<ApplyOpMetadata> kOpsMap = {
               return swOplogEntry.getStatus().withContext(
                   "Error parsing 'commitIndexBuild' oplog entry");
           }
-          IndexBuildsCoordinator::get(opCtx)->applyCommitIndexBuild(opCtx, swOplogEntry.getValue());
+          try {
+              IndexBuildsCoordinator::get(opCtx)->applyCommitIndexBuild(opCtx,
+                                                                        swOplogEntry.getValue());
+          } catch (ExceptionFor<ErrorCodes::IndexAlreadyExists>&) {
+              // TODO(SERVER-46656): We sometimes do two-phase builds of empty collections on
+              // the primary, but treat them as one-phase on the secondary.  This will result
+              // in an IndexAlreadyExists when we commit.  When SERVER-46656 is fixed we should
+              // no longer catch and ignore this error.
+          }
           return Status::OK();
       },
       {ErrorCodes::IndexAlreadyExists,
@@ -1112,13 +1124,14 @@ Status applyOperation_inlock(OperationContext* opCtx,
                 // 1. Insert if
                 //   a) we do not have a wrapping WriteUnitOfWork, which implies we are not part of
                 //      an "applyOps" command, OR
-                //   b) we are part of a multi-document transaction[1].
+                //   b) we are part of a multi-document transaction[1], OR
                 //
                 // 2. Upsert[2] if
                 //   a) we have a wrapping WriteUnitOfWork AND we are not part of a transaction,
                 //      which implies we are part of an "applyOps" command, OR
-                //   b) the previous insert failed with a DuplicateKey error AND we are not part of
-                //      a transaction.
+                //   b) the previous insert failed with a DuplicateKey error AND we are not part
+                //      a transaction AND either we are not in steady state replication mode OR
+                //      the oplogApplicationEnforcesSteadyStateConstraints parameter is false.
                 //
                 // [1] Transactions should not convert inserts to upserts because on secondaries
                 //     they will perform a lookup that never occurred on the primary. This may cause
@@ -1162,6 +1175,12 @@ Status applyOperation_inlock(OperationContext* opCtx,
                         if (inTxn) {
                             return status;
                         }
+                        if (mode == OplogApplication::Mode::kSecondary) {
+                            opCounters->gotInsertOnExistingDoc();
+                            if (oplogApplicationEnforcesSteadyStateConstraints) {
+                                return status;
+                            }
+                        }
                         // Continue to the next block to retry the operation as an upsert.
                         needToDoUpsert = true;
                     } else {
@@ -1227,7 +1246,8 @@ Status applyOperation_inlock(OperationContext* opCtx,
             // IDHACK.
             BSONObj updateCriteria = idField.wrap();
 
-            const bool upsert = alwaysUpsert || op.getUpsert().value_or(false);
+            const bool upsertOplogEntry = op.getUpsert().value_or(false);
+            const bool upsert = alwaysUpsert || upsertOplogEntry;
             UpdateRequest request(requestNss);
             request.setQuery(updateCriteria);
             request.setUpdateModification(o);
@@ -1248,7 +1268,17 @@ Status applyOperation_inlock(OperationContext* opCtx,
 
                 UpdateResult ur = update(opCtx, db, request);
                 if (ur.numMatched == 0 && ur.upserted.isEmpty()) {
-                    if (ur.modifiers) {
+                    if (collection && collection->isCapped() &&
+                        mode == OplogApplication::Mode::kSecondary) {
+                        // We can't assume there was a problem when the collection is capped,
+                        // because the item may have been deleted by the cappedDeleter.  This only
+                        // matters for steady-state mode, because all errors on missing updates are
+                        // ignored at a higher level for recovery and initial sync.
+                        LOGV2_DEBUG(2170003,
+                                    2,
+                                    "couldn't find doc in capped collection",
+                                    "op"_attr = redact(op.toBSON()));
+                    } else if (ur.modifiers) {
                         if (updateCriteria.nFields() == 1) {
                             // was a simple { _id : ... } update criteria
                             string msg = str::stream()
@@ -1287,6 +1317,18 @@ Status applyOperation_inlock(OperationContext* opCtx,
                             return Status(ErrorCodes::UpdateOperationFailed, msg);
                         }
                     }
+                } else if (mode == OplogApplication::Mode::kSecondary && !upsertOplogEntry &&
+                           !ur.upserted.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.toBSON()));
+                    opCounters->gotUpdateOnMissingDoc();
+
+                    // We shouldn't be doing upserts in secondary mode when enforcing steady state
+                    // constraints.
+                    invariant(!oplogApplicationEnforcesSteadyStateConstraints);
                 }
 
                 wuow.commit();
@@ -1331,7 +1373,24 @@ Status applyOperation_inlock(OperationContext* opCtx,
                 if (timestamp != Timestamp::min()) {
                     uassertStatusOK(opCtx->recoveryUnit()->setTimestamp(timestamp));
                 }
-                deleteObjects(opCtx, collection, requestNss, deleteCriteria, true /* justOne */);
+                auto nDeleted = deleteObjects(
+                    opCtx, collection, requestNss, deleteCriteria, true /* justOne */);
+                if (nDeleted == 0 && mode == OplogApplication::Mode::kSecondary) {
+                    LOGV2_WARNING(2170002,
+                                  "Applied a delete which did not delete anything in steady state "
+                                  "replication",
+                                  "op"_attr = redact(op.toBSON()));
+                    if (collection)
+                        opCounters->gotDeleteWasEmpty();
+                    else
+                        opCounters->gotDeleteFromMissingNamespace();
+                    // This error is fatal when we are enforcing steady state constraints.
+                    uassert(collection ? ErrorCodes::NoSuchKey : ErrorCodes::NamespaceNotFound,
+                            str::stream() << "Applied a delete which did not delete anything in "
+                                             "steady state replication : "
+                                          << redact(op.toBSON()),
+                            !oplogApplicationEnforcesSteadyStateConstraints);
+                }
                 wuow.commit();
             });
 
@@ -1523,7 +1582,17 @@ Status applyCommand_inlock(OperationContext* opCtx,
                 break;
             }
             default: {
-                if (!curOpToApply.acceptableErrors.count(status.code())) {
+                // Even when enforcing steady state constraints, we must allow IndexNotFound as
+                // an index may not have been built on a secondary when a command dropping it
+                // comes in.
+                //
+                // TODO(SERVER-46550): We should be able to enforce constraints on "dropDatabase"
+                // once we're no longer able to create databases on the primary without an oplog
+                // entry.
+                if ((mode == OplogApplication::Mode::kSecondary &&
+                     oplogApplicationEnforcesSteadyStateConstraints &&
+                     status.code() != ErrorCodes::IndexNotFound && op->first != "dropDatabase") ||
+                    !curOpToApply.acceptableErrors.count(status.code())) {
                     LOGV2_ERROR(21262,
                                 "Failed command {o} on {db} with status {status} during oplog "
                                 "application",
@@ -1533,13 +1602,22 @@ Status applyCommand_inlock(OperationContext* opCtx,
                     return status;
                 }
 
-                LOGV2_DEBUG(51776,
-                            1,
-                            "Acceptable error during oplog application on db '{db}' with status "
-                            "'{status}' from oplog entry {entry}",
-                            "db"_attr = nss.db(),
-                            "status"_attr = status,
-                            "entry"_attr = redact(entry.toBSON()));
+                if (mode == OplogApplication::Mode::kSecondary &&
+                    status.code() != ErrorCodes::IndexNotFound) {
+                    LOGV2_WARNING(2170000,
+                                  "Acceptable error during oplog application",
+                                  "db"_attr = nss.db(),
+                                  "status"_attr = status,
+                                  "oplogEntry"_attr = redact(entry.toBSON()));
+                    opCounters->gotAcceptableErrorInCommand();
+                } else {
+                    LOGV2_DEBUG(51776,
+                                1,
+                                "Acceptable error during oplog application",
+                                "db"_attr = nss.db(),
+                                "status"_attr = status,
+                                "oplogEntry"_attr = redact(entry.toBSON()));
+                }
             }
             // fallthrough
             case ErrorCodes::OK:
diff --git a/src/mongo/db/repl/oplog_applier_impl.cpp b/src/mongo/db/repl/oplog_applier_impl.cpp
index eb44ffda308..2bf49db1f3f 100644
--- a/src/mongo/db/repl/oplog_applier_impl.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl.cpp
@@ -44,7 +44,9 @@
 #include "mongo/db/logical_session_id.h"
 #include "mongo/db/repl/apply_ops.h"
 #include "mongo/db/repl/insert_group.h"
+#include "mongo/db/repl/repl_server_parameters_gen.h"
 #include "mongo/db/repl/transaction_oplog_application.h"
+#include "mongo/db/stats/counters.h"
 #include "mongo/db/stats/timer_stats.h"
 #include "mongo/logv2/log.h"
 #include "mongo/platform/basic.h"
@@ -948,17 +950,16 @@ Status applyOplogEntryOrGroupedInserts(OperationContext* opCtx,
                             db);
                     OldClientContext ctx(opCtx, autoColl.getNss().ns(), db);
 
-                    // We convert updates to upserts when not in initial sync because after rollback
-                    // and during startup we may replay an update after a delete and crash since we
-                    // do not ignore errors. In initial sync we simply ignore these update errors so
-                    // there is no reason to upsert.
+                    // We convert updates to upserts in secondary mode when the
+                    // oplogApplicationEnforcesSteadyStateConstraints parameter is false, to avoid
+                    // failing on the constraint that updates in steady state mode always update
+                    // an existing document.
                     //
-                    // TODO (SERVER-21700): Never upsert during oplog application unless an external
-                    // applyOps wants to. We should ignore these errors intelligently while in
-                    // RECOVERING and STARTUP mode (similar to initial sync) instead so we do not
-                    // accidentally ignore real errors.
-                    bool shouldAlwaysUpsert =
-                        (oplogApplicationMode != OplogApplication::Mode::kInitialSync);
+                    // In initial sync and recovery modes we always ignore errors about missing
+                    // documents on update, so there is no reason to convert the updates to upsert.
+
+                    bool shouldAlwaysUpsert = !oplogApplicationEnforcesSteadyStateConstraints &&
+                        oplogApplicationMode == OplogApplication::Mode::kSecondary;
                     Status status = applyOperation_inlock(opCtx,
                                                           db,
                                                           entryOrGroupedInserts,
@@ -970,12 +971,16 @@ Status applyOplogEntryOrGroupedInserts(OperationContext* opCtx,
                     }
                     return status;
                 } catch (ExceptionFor<ErrorCodes::NamespaceNotFound>& ex) {
-                    // Delete operations on non-existent namespaces can be treated as successful for
-                    // idempotency reasons.
-                    // During RECOVERING mode, we ignore NamespaceNotFound for all CRUD ops since
-                    // storage does not wait for drops to be checkpointed (SERVER-33161).
-                    if (opType == OpTypeEnum::kDelete ||
-                        oplogApplicationMode == OplogApplication::Mode::kRecovering) {
+                    // This can happen in initial sync or recovery modes (when a delete of the
+                    // namespace appears later in the oplog), but we will ignore it in the caller.
+                    //
+                    // When we're not enforcing steady-state constraints, the error is ignored
+                    // only for deletes, on the grounds that deleting from a non-existent collection
+                    // is a no-op.
+                    if (opType == OpTypeEnum::kDelete &&
+                        !oplogApplicationEnforcesSteadyStateConstraints &&
+                        oplogApplicationMode == OplogApplication::Mode::kSecondary) {
+                        replOpCounters.gotDeleteFromMissingNamespace();
                         return Status::OK();
                     }
 
@@ -1050,7 +1055,8 @@ Status OplogApplierImpl::applyOplogBatchPerWorker(OperationContext* opCtx,
                     // Tried to apply an update operation but the document is missing, there must be
                     // a delete operation for the document later in the oplog.
                     if (status == ErrorCodes::UpdateOperationFailed &&
-                        oplogApplicationMode == OplogApplication::Mode::kInitialSync) {
+                        (oplogApplicationMode == OplogApplication::Mode::kInitialSync ||
+                         oplogApplicationMode == OplogApplication::Mode::kRecovering)) {
                         continue;
                     }
 
diff --git a/src/mongo/db/repl/oplog_applier_impl_test.cpp b/src/mongo/db/repl/oplog_applier_impl_test.cpp
index e097ed0192f..7450c4ed352 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl_test.cpp
@@ -56,6 +56,7 @@
 #include "mongo/db/repl/idempotency_test_fixture.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/oplog_applier.h"
+#include "mongo/db/repl/repl_server_parameters_gen.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl/replication_process.h"
 #include "mongo/db/repl/storage_interface.h"
@@ -79,7 +80,11 @@ namespace {
 /**
  * Creates an OplogEntry with given parameters and preset defaults for this test suite.
  */
-OplogEntry makeOplogEntry(OpTypeEnum opType, NamespaceString nss, OptionalCollectionUUID uuid) {
+OplogEntry makeOplogEntry(OpTypeEnum opType,
+                          NamespaceString nss,
+                          OptionalCollectionUUID uuid,
+                          BSONObj o,
+                          boost::optional<BSONObj> o2) {
     return OplogEntry(OpTime(Timestamp(1, 1), 1),  // optime
                       boost::none,                 // hash
                       opType,                      // opType
@@ -87,8 +92,8 @@ OplogEntry makeOplogEntry(OpTypeEnum opType, NamespaceString nss, OptionalCollec
                       uuid,                        // uuid
                       boost::none,                 // fromMigrate
                       OplogEntry::kOplogVersion,   // version
-                      BSON("_id" << 0),            // o
-                      boost::none,                 // o2
+                      o,                           // o
+                      o2,                          // o2
                       {},                          // sessionInfo
                       boost::none,                 // upsert
                       Date_t(),                    // wall clock time
@@ -98,6 +103,9 @@ OplogEntry makeOplogEntry(OpTypeEnum opType, NamespaceString nss, OptionalCollec
                       boost::none);  // post-image optime
 }
 
+OplogEntry makeOplogEntry(OpTypeEnum opType, NamespaceString nss, OptionalCollectionUUID uuid) {
+    return makeOplogEntry(opType, nss, uuid, BSON("_id" << 0), boost::none);
+}
 /**
  * Creates collection options suitable for oplog.
  */
@@ -109,6 +117,15 @@ CollectionOptions createOplogCollectionOptions() {
     return options;
 }
 
+/*
+ * Creates collection options for recording pre-images for testing deletes
+ */
+CollectionOptions createRecordPreImageCollectionOptions() {
+    CollectionOptions options;
+    options.recordPreImages = true;
+    return options;
+}
+
 /**
  * Create test collection.
  * Returns collection.
@@ -172,6 +189,29 @@ auto parseFromOplogEntryArray(const BSONObj& obj, int elem) {
     return OpTime(tsArray.Array()[elem].timestamp(), termArray.Array()[elem].Long());
 };
 
+template <typename T, bool enable>
+class SetSteadyStateConstraints : public T {
+protected:
+    void setUp() override {
+        T::setUp();
+        _constraintsEnabled = oplogApplicationEnforcesSteadyStateConstraints;
+        oplogApplicationEnforcesSteadyStateConstraints = enable;
+    }
+
+    void tearDown() override {
+        oplogApplicationEnforcesSteadyStateConstraints = _constraintsEnabled;
+        T::tearDown();
+    }
+
+private:
+    bool _constraintsEnabled;
+};
+
+typedef SetSteadyStateConstraints<OplogApplierImplTest, false>
+    OplogApplierImplTestDisableSteadyStateConstraints;
+typedef SetSteadyStateConstraints<OplogApplierImplTest, true>
+    OplogApplierImplTestEnableSteadyStateConstraints;
+
 TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentDatabaseMissing) {
     NamespaceString nss("test.t");
     auto op = makeOplogEntry(OpTypeEnum::kInsert, nss, {});
@@ -180,10 +220,22 @@ TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentDataba
                   ExceptionFor<ErrorCodes::NamespaceNotFound>);
 }
 
-TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentDatabaseMissing) {
+TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsDeleteDocumentDatabaseMissing) {
     NamespaceString otherNss("test.othername");
     auto op = makeOplogEntry(OpTypeEnum::kDelete, otherNss, {});
+    int prevDeleteFromMissing = replOpCounters.getDeleteFromMissingNamespace()->load();
     _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, false);
+    ASSERT_EQ(1, replOpCounters.getDeleteFromMissingNamespace()->load() - prevDeleteFromMissing);
+}
+
+TEST_F(OplogApplierImplTestEnableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsDeleteDocumentDatabaseMissing) {
+    NamespaceString otherNss("test.othername");
+    auto op = makeOplogEntry(OpTypeEnum::kDelete, otherNss, {});
+    ASSERT_THROWS(_applyOplogEntryOrGroupedInsertsWrapper(
+                      _opCtx.get(), &op, OplogApplication::Mode::kSecondary),
+                  ExceptionFor<ErrorCodes::NamespaceNotFound>);
 }
 
 TEST_F(OplogApplierImplTest,
@@ -197,13 +249,26 @@ TEST_F(OplogApplierImplTest,
                   ExceptionFor<ErrorCodes::NamespaceNotFound>);
 }
 
-TEST_F(OplogApplierImplTest,
+TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
        applyOplogEntryOrGroupedInsertsDeleteDocumentCollectionLookupByUUIDFails) {
     const NamespaceString nss("test.t");
     createDatabase(_opCtx.get(), nss.db());
     NamespaceString otherNss(nss.getSisterNS("othername"));
     auto op = makeOplogEntry(OpTypeEnum::kDelete, otherNss, kUuid);
+    int prevDeleteFromMissing = replOpCounters.getDeleteFromMissingNamespace()->load();
     _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, false);
+    ASSERT_EQ(1, replOpCounters.getDeleteFromMissingNamespace()->load() - prevDeleteFromMissing);
+}
+
+TEST_F(OplogApplierImplTestEnableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsDeleteDocumentCollectionLookupByUUIDFails) {
+    const NamespaceString nss("test.t");
+    createDatabase(_opCtx.get(), nss.db());
+    NamespaceString otherNss(nss.getSisterNS("othername"));
+    auto op = makeOplogEntry(OpTypeEnum::kDelete, otherNss, kUuid);
+    ASSERT_THROWS(_applyOplogEntryOrGroupedInsertsWrapper(
+                      _opCtx.get(), &op, OplogApplication::Mode::kSecondary),
+                  ExceptionFor<ErrorCodes::NamespaceNotFound>);
 }
 
 TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentCollectionMissing) {
@@ -219,15 +284,30 @@ TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentCollec
     ASSERT_FALSE(collectionExists(_opCtx.get(), nss));
 }
 
-TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentCollectionMissing) {
+TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsDeleteDocumentCollectionMissing) {
     const NamespaceString nss("test.t");
     createDatabase(_opCtx.get(), nss.db());
     // Even though the collection doesn't exist, this is handled in the actual application function,
     // which in the case of this test just ignores such errors. This tests mostly that we don't
     // implicitly create the collection.
     auto op = makeOplogEntry(OpTypeEnum::kDelete, nss, {});
+    int prevDeleteFromMissing = replOpCounters.getDeleteFromMissingNamespace()->load();
     _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, false);
     ASSERT_FALSE(collectionExists(_opCtx.get(), nss));
+    ASSERT_EQ(1, replOpCounters.getDeleteFromMissingNamespace()->load() - prevDeleteFromMissing);
+}
+
+TEST_F(OplogApplierImplTestEnableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsDeleteDocumentCollectionMissing) {
+    const NamespaceString nss("test.t");
+    createDatabase(_opCtx.get(), nss.db());
+    // With steady state constraints enabled, attempting to delete from a missing collection is an
+    // error.
+    auto op = makeOplogEntry(OpTypeEnum::kDelete, nss, {});
+    ASSERT_THROWS(_applyOplogEntryOrGroupedInsertsWrapper(
+                      _opCtx.get(), &op, OplogApplication::Mode::kSecondary),
+                  ExceptionFor<ErrorCodes::NamespaceNotFound>);
 }
 
 TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentCollectionExists) {
@@ -237,11 +317,72 @@ TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentCollec
     _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, true);
 }
 
-TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentCollectionExists) {
+TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsDeleteDocumentDocMissing) {
     const NamespaceString nss("test.t");
     createCollection(_opCtx.get(), nss, {});
     auto op = makeOplogEntry(OpTypeEnum::kDelete, nss, {});
+    int prevDeleteWasEmpty = replOpCounters.getDeleteWasEmpty()->load();
     _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, false);
+    ASSERT_EQ(1, replOpCounters.getDeleteWasEmpty()->load() - prevDeleteWasEmpty);
+}
+
+TEST_F(OplogApplierImplTestEnableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsDeleteDocumentDocMissing) {
+    const NamespaceString nss("test.t");
+    createCollection(_opCtx.get(), nss, {});
+    auto op = makeOplogEntry(OpTypeEnum::kDelete, nss, {});
+    ASSERT_THROWS(_applyOplogEntryOrGroupedInsertsWrapper(
+                      _opCtx.get(), &op, OplogApplication::Mode::kSecondary),
+                  ExceptionFor<ErrorCodes::NoSuchKey>);
+}
+
+TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentCollectionAndDocExist) {
+    const NamespaceString nss("test.t");
+    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);
+}
+
+TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsInsertExistingDocument) {
+    const NamespaceString nss("test.t");
+    auto uuid = createCollectionWithUuid(_opCtx.get(), nss);
+    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);
+    ASSERT_EQ(1, replOpCounters.getInsertOnExistingDoc()->load() - prevInsertOnExistingDoc);
+}
+
+TEST_F(OplogApplierImplTestEnableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsInsertExistingDocument) {
+    const NamespaceString nss("test.t");
+    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);
+}
+
+TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsUpdateMissingDocument) {
+    const NamespaceString nss("test.t");
+    auto uuid = createCollectionWithUuid(_opCtx.get(), nss);
+    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);
+    ASSERT_EQ(1, replOpCounters.getUpdateOnMissingDoc()->load() - prevUpdateOnMissingDoc);
+}
+
+TEST_F(OplogApplierImplTestEnableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsUpdateMissingDocument) {
+    const NamespaceString nss("test.t");
+    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);
 }
 
 TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentCollectionLockedByUUID) {
@@ -253,7 +394,8 @@ TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentCollec
     _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, true);
 }
 
-TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentCollectionLockedByUUID) {
+TEST_F(OplogApplierImplTestDisableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsDeleteMissingDocCollectionLockedByUUID) {
     const NamespaceString nss("test.t");
     CollectionOptions options;
     options.uuid = kUuid;
@@ -262,7 +404,38 @@ 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);
+    int prevDeleteWasEmpty = replOpCounters.getDeleteWasEmpty()->load();
     _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, false);
+    ASSERT_EQ(1, replOpCounters.getDeleteWasEmpty()->load() - prevDeleteWasEmpty);
+}
+
+TEST_F(OplogApplierImplTestEnableSteadyStateConstraints,
+       applyOplogEntryOrGroupedInsertsDeleteMissingDocCollectionLockedByUUID) {
+    const NamespaceString nss("test.t");
+    CollectionOptions options;
+    options.uuid = kUuid;
+    createCollection(_opCtx.get(), nss, options);
+
+    NamespaceString otherNss(nss.getSisterNS("othername"));
+    auto op = makeOplogEntry(OpTypeEnum::kDelete, otherNss, options.uuid);
+    ASSERT_THROWS(_applyOplogEntryOrGroupedInsertsWrapper(
+                      _opCtx.get(), &op, OplogApplication::Mode::kSecondary),
+                  ExceptionFor<ErrorCodes::NoSuchKey>);
+}
+
+TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentCollectionLockedByUUID) {
+    const NamespaceString nss("test.t");
+    CollectionOptions options = createRecordPreImageCollectionOptions();
+    options.uuid = kUuid;
+    createCollection(_opCtx.get(), nss, options);
+
+    // Make sure the document to be deleted exists.
+    ASSERT_OK(getStorageInterface()->insertDocument(_opCtx.get(), nss, {BSON("_id" << 0)}, 0));
+
+    // 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);
 }
 
 TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsCommand) {
@@ -2952,6 +3125,37 @@ TEST_F(IdempotencyTest, ConvertToCappedNamespaceNotFound) {
     ASSERT_FALSE(autoColl.getDb());
 }
 
+typedef SetSteadyStateConstraints<IdempotencyTest, false>
+    IdempotencyTestDisableSteadyStateConstraints;
+typedef SetSteadyStateConstraints<IdempotencyTest, true>
+    IdempotencyTestEnableSteadyStateConstraints;
+
+TEST_F(IdempotencyTestDisableSteadyStateConstraints, AcceptableErrorsRecordedInSteadyStateMode) {
+    // Create a BSON "convertToCapped" command.
+    auto convertToCappedCmd = BSON("convertToCapped" << nss.coll());
+
+    // Create a "convertToCapped" oplog entry.
+    auto convertToCappedOp = makeCommandOplogEntry(nextOpTime(), nss, convertToCappedCmd);
+
+    // Ensure that NamespaceNotFound is "acceptable" but counted.
+    int prevAcceptableError = replOpCounters.getAcceptableErrorInCommand()->load();
+    ASSERT_OK(runOpSteadyState(convertToCappedOp));
+
+    ASSERT_EQ(1, replOpCounters.getAcceptableErrorInCommand()->load() - prevAcceptableError);
+}
+
+TEST_F(IdempotencyTestEnableSteadyStateConstraints,
+       AcceptableErrorsNotAcceptableInSteadyStateMode) {
+    // Create a BSON "convertToCapped" command.
+    auto convertToCappedCmd = BSON("convertToCapped" << nss.coll());
+
+    // Create a "convertToCapped" oplog entry.
+    auto convertToCappedOp = makeCommandOplogEntry(nextOpTime(), nss, convertToCappedCmd);
+
+    // Ensure that NamespaceNotFound is returned.
+    ASSERT_EQUALS(ErrorCodes::NamespaceNotFound, runOpSteadyState(convertToCappedOp));
+}
+
 class IdempotencyTestTxns : public IdempotencyTest {};
 
 // Document used by transaction idempotency tests.
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 7e3f8d89040..fdcd4911a4c 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
@@ -79,6 +79,14 @@ void OplogApplierImplOpObserver::onDelete(OperationContext* opCtx,
     onDeleteFn(opCtx, nss, uuid, stmtId, fromMigrate, deletedDoc);
 }
 
+void OplogApplierImplOpObserver::onUpdate(OperationContext* opCtx,
+                                          const OplogUpdateEntryArgs& args) {
+    if (!onUpdateFn) {
+        return;
+    }
+    onUpdateFn(opCtx, args);
+}
+
 void OplogApplierImplOpObserver::onCreateCollection(OperationContext* opCtx,
                                                     Collection* coll,
                                                     const NamespaceString& collectionName,
@@ -173,7 +181,10 @@ void OplogApplierImplTest::_testApplyOplogEntryOrGroupedInsertsCrudOperation(
             checkOpCtx(opCtx);
             ASSERT_EQUALS(NamespaceString("test.t"), nss);
             ASSERT_EQUALS(1U, docs.size());
-            ASSERT_BSONOBJ_EQ(op.getObject(), docs[0]);
+            // For upserts we don't know the intended value of the document.
+            if (op.getOpType() == repl::OpTypeEnum::kInsert) {
+                ASSERT_BSONOBJ_EQ(op.getObject(), docs[0]);
+            }
             return Status::OK();
         };
 
@@ -191,6 +202,13 @@ void OplogApplierImplTest::_testApplyOplogEntryOrGroupedInsertsCrudOperation(
         return Status::OK();
     };
 
+    _opObserver->onUpdateFn = [&](OperationContext* opCtx, const OplogUpdateEntryArgs& args) {
+        applyOpCalled = true;
+        checkOpCtx(opCtx);
+        ASSERT_EQUALS(NamespaceString("test.t"), args.nss);
+        return Status::OK();
+    };
+
     ASSERT_EQ(_applyOplogEntryOrGroupedInsertsWrapper(
                   _opCtx.get(), &op, OplogApplication::Mode::kSecondary),
               expectedError);
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 10aa03fea65..2f1fbd02296 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test_fixture.h
+++ b/src/mongo/db/repl/oplog_applier_impl_test_fixture.h
@@ -91,6 +91,11 @@ public:
                   const boost::optional<BSONObj>& deletedDoc) override;
 
     /**
+     * This function is called whenever OplogApplierImpl updates a document in a collection.
+     */
+    void onUpdate(OperationContext* opCtx, const OplogUpdateEntryArgs& args) override;
+
+    /**
      * Called when OplogApplierImpl creates a collection.
      */
     void onCreateCollection(OperationContext* opCtx,
@@ -113,6 +118,8 @@ public:
                        const boost::optional<BSONObj>&)>
         onDeleteFn;
 
+    std::function<void(OperationContext*, const OplogUpdateEntryArgs&)> onUpdateFn;
+
     std::function<void(OperationContext*,
                        Collection*,
                        const NamespaceString&,
diff --git a/src/mongo/db/repl/repl_server_parameters.idl b/src/mongo/db/repl/repl_server_parameters.idl
index a04ec8e2b42..eddddc2a331 100644
--- a/src/mongo/db/repl/repl_server_parameters.idl
+++ b/src/mongo/db/repl/repl_server_parameters.idl
@@ -312,3 +312,12 @@ server_parameters:
         cpp_vartype: bool
         cpp_varname: enableAutomaticReconfig
         default: false
+
+    oplogApplicationEnforcesSteadyStateConstraints:
+        description: >-
+            Whether or not secondary oplog application enforces (by fassert) consistency
+            constraints that apply if an oplog entry is to be applied exactly once and in order.
+        set_at: startup
+        cpp_vartype: bool
+        cpp_varname: oplogApplicationEnforcesSteadyStateConstraints
+        default: false
diff --git a/src/mongo/db/stats/counters.h b/src/mongo/db/stats/counters.h
index d2962ee26c1..9b10cb2a049 100644
--- a/src/mongo/db/stats/counters.h
+++ b/src/mongo/db/stats/counters.h
@@ -77,6 +77,24 @@ public:
 
     BSONObj getObj() const;
 
+    // These opcounters record operations that would fail if we were fully enforcing our consistency
+    // constraints in steady-state oplog application mode.
+    void gotInsertOnExistingDoc() {
+        _checkWrap(&OpCounters::_insertOnExistingDoc, 1);
+    }
+    void gotUpdateOnMissingDoc() {
+        _checkWrap(&OpCounters::_updateOnMissingDoc, 1);
+    }
+    void gotDeleteWasEmpty() {
+        _checkWrap(&OpCounters::_deleteWasEmpty, 1);
+    }
+    void gotDeleteFromMissingNamespace() {
+        _checkWrap(&OpCounters::_deleteFromMissingNamespace, 1);
+    }
+    void gotAcceptableErrorInCommand() {
+        _checkWrap(&OpCounters::_acceptableErrorInCommand, 1);
+    }
+
     // thse are used by snmp, and other things, do not remove
     const AtomicWord<long long>* getInsert() const {
         return &_insert;
@@ -96,6 +114,21 @@ public:
     const AtomicWord<long long>* getCommand() const {
         return &_command;
     }
+    const AtomicWord<long long>* getInsertOnExistingDoc() const {
+        return &_insertOnExistingDoc;
+    }
+    const AtomicWord<long long>* getUpdateOnMissingDoc() const {
+        return &_updateOnMissingDoc;
+    }
+    const AtomicWord<long long>* getDeleteWasEmpty() const {
+        return &_deleteWasEmpty;
+    }
+    const AtomicWord<long long>* getDeleteFromMissingNamespace() const {
+        return &_deleteFromMissingNamespace;
+    }
+    const AtomicWord<long long>* getAcceptableErrorInCommand() const {
+        return &_acceptableErrorInCommand;
+    }
 
 private:
     // Increment member `counter` by `n`, resetting all counters if it was > 2^60.
@@ -107,6 +140,12 @@ private:
     CacheAligned<AtomicWord<long long>> _delete;
     CacheAligned<AtomicWord<long long>> _getmore;
     CacheAligned<AtomicWord<long long>> _command;
+
+    CacheAligned<AtomicWord<long long>> _insertOnExistingDoc;
+    CacheAligned<AtomicWord<long long>> _updateOnMissingDoc;
+    CacheAligned<AtomicWord<long long>> _deleteWasEmpty;
+    CacheAligned<AtomicWord<long long>> _deleteFromMissingNamespace;
+    CacheAligned<AtomicWord<long long>> _acceptableErrorInCommand;
 };
 
 extern OpCounters globalOpCounters;
diff --git a/src/mongo/dbtests/repltests.cpp b/src/mongo/dbtests/repltests.cpp
index bdb122a7efb..b552398ef83 100644
--- a/src/mongo/dbtests/repltests.cpp
+++ b/src/mongo/dbtests/repltests.cpp
@@ -169,6 +169,9 @@ public:
     }
 
 protected:
+    virtual OplogApplication::Mode getOplogApplicationMode() {
+        return OplogApplication::Mode::kSecondary;
+    }
     static const char* ns() {
         return "unittests.repltests";
     }
@@ -250,7 +253,7 @@ protected:
                 repl::UnreplicatedWritesBlock uwb(&_opCtx);
                 auto entry = uassertStatusOK(OplogEntry::parse(*i));
                 uassertStatusOK(applyOperation_inlock(
-                    &_opCtx, ctx.db(), &entry, false, OplogApplication::Mode::kSecondary));
+                    &_opCtx, ctx.db(), &entry, false, getOplogApplicationMode()));
             }
         }
     }
@@ -364,7 +367,16 @@ protected:
     virtual void reset() const = 0;
 };
 
-class InsertTimestamp : public Base {
+// Some operations are only idempotent when in RECOVERING, not in SECONDARY.  This includes
+// duplicate inserts and deletes.
+class Recovering : public Base {
+protected:
+    virtual OplogApplication::Mode getOplogApplicationMode() {
+        return OplogApplication::Mode::kRecovering;
+    }
+};
+
+class InsertTimestamp : public Recovering {
 public:
     void doIt() const {
         BSONObjBuilder b;
@@ -386,7 +398,7 @@ private:
     mutable Date_t date_;
 };
 
-class InsertAutoId : public Base {
+class InsertAutoId : public Recovering {
 public:
     InsertAutoId() : o_(fromjson("{\"a\":\"b\"}")) {}
     void doIt() const {
@@ -414,7 +426,7 @@ public:
     }
 };
 
-class InsertTwo : public Base {
+class InsertTwo : public Recovering {
 public:
     InsertTwo() : o_(fromjson("{'_id':1,a:'b'}")), t_(fromjson("{'_id':2,c:'d'}")) {}
     void doIt() const {
@@ -437,7 +449,7 @@ private:
     BSONObj t_;
 };
 
-class InsertTwoIdentical : public Base {
+class InsertTwoIdentical : public Recovering {
 public:
     InsertTwoIdentical() : o_(fromjson("{\"a\":\"b\"}")) {}
     void doIt() const {
@@ -697,7 +709,7 @@ protected:
 };
 
 
-class UpsertInsertIdMod : public Base {
+class UpsertInsertIdMod : public Recovering {
 public:
     UpsertInsertIdMod()
         : q_(fromjson("{'_id':5,a:4}")),
@@ -718,7 +730,7 @@ protected:
     BSONObj q_, u_, ou_;
 };
 
-class UpsertInsertSet : public Base {
+class UpsertInsertSet : public Recovering {
 public:
     UpsertInsertSet()
         : q_(fromjson("{a:5}")), u_(fromjson("{$set:{a:7}}")), ou_(fromjson("{a:7}")) {}
@@ -738,7 +750,7 @@ protected:
     BSONObj o_, q_, u_, ou_;
 };
 
-class UpsertInsertInc : public Base {
+class UpsertInsertInc : public Recovering {
 public:
     UpsertInsertInc()
         : q_(fromjson("{a:5}")), u_(fromjson("{$inc:{a:3}}")), ou_(fromjson("{a:8}")) {}
@@ -757,7 +769,7 @@ protected:
     BSONObj o_, q_, u_, ou_;
 };
 
-class MultiInc : public Base {
+class MultiInc : public Recovering {
 public:
     string s() const {
         stringstream ss;
@@ -823,7 +835,7 @@ protected:
     BSONObj o_, u_, ot_;
 };
 
-class Remove : public Base {
+class Remove : public Recovering {
 public:
     Remove()
         : o1_(f("{\"_id\":\"010101010101010101010101\",\"a\":\"b\"}")),
@@ -854,7 +866,7 @@ class RemoveOne : public Remove {
     }
 };
 
-class FailingUpdate : public Base {
+class FailingUpdate : public Recovering {
 public:
     FailingUpdate() : o_(fromjson("{'_id':1,a:'b'}")), u_(fromjson("{'_id':1,c:'d'}")) {}
     void doIt() const {
diff --git a/src/mongo/shell/servers.js b/src/mongo/shell/servers.js
index 22ca5e87df3..1024e3d1316 100644
--- a/src/mongo/shell/servers.js
+++ b/src/mongo/shell/servers.js
@@ -1169,6 +1169,15 @@ function appendSetParameterArgs(argArray) {
                 }
             }
 
+            // New mongod-specific option in 4.3.x.
+            if (!programMajorMinorVersion || programMajorMinorVersion >= 430) {
+                if (!argArrayContainsSetParameterValue(
+                        'oplogApplicationEnforcesSteadyStateConstraints=')) {
+                    argArray.push(...['--setParameter',
+                                      'oplogApplicationEnforcesSteadyStateConstraints=true']);
+                }
+            }
+
             // New mongod-specific options in 4.0.x
             if (!programMajorMinorVersion || programMajorMinorVersion >= 400) {
                 if (jsTest.options().transactionLifetimeLimitSeconds !== undefined) {