SERVER-42925 Refactor idempotency tests oplog application and fix oplog visiblity issues

6 files changed, 48 insertions, 88 deletions

diff --git a/src/mongo/db/repl/idempotency_test.cpp b/src/mongo/db/repl/idempotency_test.cpp
index e76599bc86a..c7a12441d68 100644
--- a/src/mongo/db/repl/idempotency_test.cpp
+++ b/src/mongo/db/repl/idempotency_test.cpp
@@ -60,7 +60,7 @@ protected:
 
     std::string getStatesString(const std::vector<CollectionState>& state1,
                                 const std::vector<CollectionState>& state2,
-                                const MultiApplier::OperationPtrs& opPtrs) override;
+                                const std::vector<OplogEntry>& ops) override;
 
     Status resetState() override;
 
@@ -113,28 +113,28 @@ std::vector<OplogEntry> RandomizedIdempotencyTest::createUpdateSequence(
 
 std::string RandomizedIdempotencyTest::getStatesString(const std::vector<CollectionState>& state1,
                                                        const std::vector<CollectionState>& state2,
-                                                       const MultiApplier::OperationPtrs& opPtrs) {
-    unittest::log() << IdempotencyTest::getStatesString(state1, state2, opPtrs);
+                                                       const std::vector<OplogEntry>& ops) {
+    unittest::log() << IdempotencyTest::getStatesString(state1, state2, ops);
     StringBuilder sb;
     sb << "Ran update ops: ";
     sb << "[ ";
     bool firstIter = true;
-    for (auto op : opPtrs) {
+    for (const auto& op : ops) {
         if (!firstIter) {
             sb << ", ";
         } else {
             firstIter = false;
         }
-        sb << op->toString();
+        sb << op.toString();
     }
     sb << " ]\n";
 
     ASSERT_OK(resetState());
 
     sb << "Start: " << getDoc() << "\n";
-    for (auto op : opPtrs) {
-        ASSERT_OK(runOpInitialSync(*op));
-        sb << "Apply: " << op->getObject() << "\n  ==> " << getDoc() << "\n";
+    for (const auto& op : ops) {
+        ASSERT_OK(runOpInitialSync(op));
+        sb << "Apply: " << op.getObject() << "\n  ==> " << getDoc() << "\n";
     }
 
     sb << "Found from the seed: " << this->seed;
diff --git a/src/mongo/db/repl/idempotency_test_fixture.cpp b/src/mongo/db/repl/idempotency_test_fixture.cpp
index b6cf0d36764..bb8ba836f38 100644
--- a/src/mongo/db/repl/idempotency_test_fixture.cpp
+++ b/src/mongo/db/repl/idempotency_test_fixture.cpp
@@ -371,64 +371,34 @@ Status IdempotencyTest::resetState() {
 void IdempotencyTest::testOpsAreIdempotent(std::vector<OplogEntry> ops, SequenceType sequenceType) {
     ASSERT_OK(resetState());
 
-    // Write oplog entries to oplog collection.
-    for (auto&& entry : ops) {
-        ASSERT_OK(getStorageInterface()->insertDocument(
-            _opCtx.get(),
-            NamespaceString::kRsOplogNamespace,
-            {entry.toBSON(), entry.getOpTime().getTimestamp()},
-            entry.getOpTime().getTerm()));
-    }
-
-    SyncTail syncTail(nullptr,  // observer
-                      nullptr,  // consistency markers
-                      nullptr,  // storage interface
-                      SyncTail::MultiSyncApplyFunc(),
-                      nullptr,  // writer pool
-                      SyncTailTest::makeInitialSyncOptions());
-    std::vector<MultiApplier::OperationPtrs> writerVectors(1);
-    std::vector<MultiApplier::Operations> derivedOps;
-
-    // Keeps all operations in scope for the lifetime of this function.
-    std::vector<MultiApplier::Operations> singleOpVectors;
-    for (auto&& entry : ops) {
-        // Derive ops for transactions if necessary.
-        std::vector<OplogEntry> op;
-        op.push_back(entry);
-        singleOpVectors.emplace_back(op);
-        syncTail.fillWriterVectors(
-            _opCtx.get(), &singleOpVectors.back(), &writerVectors, &derivedOps);
-    }
+    ASSERT_OK(runOpsInitialSync(ops));
 
-    const auto& opPtrs = writerVectors[0];
-    ASSERT_OK(runOpPtrsInitialSync(opPtrs));
     auto state1 = validateAllCollections();
     auto iterations = sequenceType == SequenceType::kEntireSequence ? 1 : ops.size();
-
     for (std::size_t i = 0; i < iterations; i++) {
         // Since the end state after each iteration is expected to be the same as the start state,
         // we don't drop and re-create the collections. Dropping and re-creating the collections
         // won't work either because we don't have ways to wait until second-phase drop to
         // completely finish.
-        MultiApplier::OperationPtrs fullSequence;
+        std::vector<OplogEntry> fullSequence;
 
         if (sequenceType == SequenceType::kEntireSequence) {
-            ASSERT_OK(runOpPtrsInitialSync(opPtrs));
-            fullSequence.insert(fullSequence.end(), opPtrs.begin(), opPtrs.end());
+            ASSERT_OK(runOpsInitialSync(ops));
+            fullSequence.insert(fullSequence.end(), ops.begin(), ops.end());
         } else if (sequenceType == SequenceType::kAnyPrefix ||
                    sequenceType == SequenceType::kAnyPrefixOrSuffix) {
-            MultiApplier::OperationPtrs prefix(opPtrs.begin(), opPtrs.begin() + i + 1);
-            ASSERT_OK(runOpPtrsInitialSync(prefix));
+            std::vector<OplogEntry> prefix(ops.begin(), ops.begin() + i + 1);
+            ASSERT_OK(runOpsInitialSync(prefix));
             fullSequence.insert(fullSequence.end(), prefix.begin(), prefix.end());
         }
 
-        ASSERT_OK(runOpPtrsInitialSync(opPtrs));
-        fullSequence.insert(fullSequence.end(), opPtrs.begin(), opPtrs.end());
+        ASSERT_OK(runOpsInitialSync(ops));
+        fullSequence.insert(fullSequence.end(), ops.begin(), ops.end());
 
         if (sequenceType == SequenceType::kAnySuffix ||
             sequenceType == SequenceType::kAnyPrefixOrSuffix) {
-            MultiApplier::OperationPtrs suffix(opPtrs.begin() + i, opPtrs.end());
-            ASSERT_OK(runOpPtrsInitialSync(suffix));
+            std::vector<OplogEntry> suffix(ops.begin() + i, ops.end());
+            ASSERT_OK(runOpsInitialSync(suffix));
             fullSequence.insert(fullSequence.end(), suffix.begin(), suffix.end());
         }
 
@@ -645,7 +615,7 @@ CollectionState IdempotencyTest::validate(const NamespaceString& nss) {
 
 std::string IdempotencyTest::getStatesString(const std::vector<CollectionState>& state1,
                                              const std::vector<CollectionState>& state2,
-                                             const MultiApplier::OperationPtrs& opPtrs) {
+                                             const std::vector<OplogEntry>& ops) {
     StringBuilder sb;
     sb << "The states:\n";
     for (const auto& s : state1) {
@@ -657,8 +627,8 @@ std::string IdempotencyTest::getStatesString(const std::vector<CollectionState>&
     }
     sb << "found after applying the operations a second time, therefore breaking idempotency.\n";
     sb << "Applied ops:\n";
-    for (auto op : opPtrs) {
-        sb << op->toString() << "\n";
+    for (const auto& op : ops) {
+        sb << op.toString() << "\n";
     }
     return sb.str();
 }
diff --git a/src/mongo/db/repl/idempotency_test_fixture.h b/src/mongo/db/repl/idempotency_test_fixture.h
index 5443e79ce4d..d32ee1af713 100644
--- a/src/mongo/db/repl/idempotency_test_fixture.h
+++ b/src/mongo/db/repl/idempotency_test_fixture.h
@@ -142,7 +142,7 @@ protected:
     std::string computeDataHash(Collection* collection);
     virtual std::string getStatesString(const std::vector<CollectionState>& state1,
                                         const std::vector<CollectionState>& state2,
-                                        const MultiApplier::OperationPtrs& opPtrs);
+                                        const std::vector<OplogEntry>& ops);
     /**
      * Validate data and indexes. Return the MD5 hash of the documents ordered by _id.
      */
diff --git a/src/mongo/db/repl/sync_tail_test.cpp b/src/mongo/db/repl/sync_tail_test.cpp
index 22024e7d35c..5c92d9739ab 100644
--- a/src/mongo/db/repl/sync_tail_test.cpp
+++ b/src/mongo/db/repl/sync_tail_test.cpp
@@ -2101,10 +2101,6 @@ TEST_F(IdempotencyTest, CreateCollectionWithCollation) {
     CollectionUUID uuid = UUID::gen();
 
     auto runOpsAndValidate = [this, uuid]() {
-        auto insertOp1 = insert(fromjson("{ _id: 'foo' }"));
-        auto insertOp2 = insert(fromjson("{ _id: 'Foo', x: 1 }"));
-        auto updateOp = update("foo", BSON("$set" << BSON("x" << 2)));
-        auto dropColl = makeCommandOplogEntry(nextOpTime(), nss, BSON("drop" << nss.coll()));
         auto options = BSON("collation"
                             << BSON("locale"
                                     << "en"
@@ -2118,6 +2114,9 @@ TEST_F(IdempotencyTest, CreateCollectionWithCollation) {
                                     << "57.1")
                             << "uuid" << uuid);
         auto createColl = makeCreateCollectionOplogEntry(nextOpTime(), nss, options);
+        auto insertOp1 = insert(fromjson("{ _id: 'foo' }"));
+        auto insertOp2 = insert(fromjson("{ _id: 'Foo', x: 1 }"));
+        auto updateOp = update("foo", BSON("$set" << BSON("x" << 2)));
 
         // We don't drop and re-create the collection since we don't have ways
         // to wait until second-phase drop to completely finish.
diff --git a/src/mongo/db/repl/sync_tail_test_fixture.cpp b/src/mongo/db/repl/sync_tail_test_fixture.cpp
index 4d80f946a53..f828b69ccb4 100644
--- a/src/mongo/db/repl/sync_tail_test_fixture.cpp
+++ b/src/mongo/db/repl/sync_tail_test_fixture.cpp
@@ -230,45 +230,37 @@ Status SyncTailTest::runOpInitialSync(const OplogEntry& op) {
 }
 
 Status SyncTailTest::runOpsInitialSync(std::vector<OplogEntry> ops) {
+    auto writerPool = OplogApplier::makeWriterPool();
+    auto storageInterface = getStorageInterface();
     auto options = makeInitialSyncOptions();
     SyncTail syncTail(nullptr,
                       getConsistencyMarkers(),
-                      getStorageInterface(),
-                      SyncTail::MultiSyncApplyFunc(),
-                      nullptr,
-                      options);
-    // Apply each operation in a batch of one because 'ops' may contain a mix of commands and CRUD
-    // operations provided by idempotency tests.
-    for (auto& op : ops) {
-        MultiApplier::OperationPtrs opsPtrs;
-        opsPtrs.push_back(&op);
-        WorkerMultikeyPathInfo pathInfo;
-        auto status = multiSyncApply(_opCtx.get(), &opsPtrs, &syncTail, &pathInfo);
-        if (!status.isOK()) {
-            return status;
-        }
-    }
-    return Status::OK();
-}
-
-Status SyncTailTest::runOpPtrsInitialSync(MultiApplier::OperationPtrs ops) {
-    auto options = makeInitialSyncOptions();
-    SyncTail syncTail(nullptr,
-                      getConsistencyMarkers(),
-                      getStorageInterface(),
-                      SyncTail::MultiSyncApplyFunc(),
-                      nullptr,
+                      storageInterface,
+                      multiSyncApply,
+                      writerPool.get(),
                       options);
+    // Idempotency tests apply the same batch of oplog entries multiple times in a loop, which would
+    // result in out-of-order oplog inserts. So we truncate the oplog collection first before
+    // calling multiApply.
+    ASSERT_OK(
+        storageInterface->truncateCollection(_opCtx.get(), NamespaceString::kRsOplogNamespace));
     // Apply each operation in a batch of one because 'ops' may contain a mix of commands and CRUD
-    // operations provided by idempotency tests.
+    // operations provided by idempotency tests. Applying operations in a batch of one is also
+    // necessary to work around oplog visibility issues. For example, idempotency tests may contain
+    // a prepare and a commit that we don't apply both in the same batch in production oplog
+    // application because the commit needs to read the prepare entry. So we apply each operation in
+    // its own batch and update oplog visibility after each batch to make sure all previously
+    // applied entries are visible to subsequent batches.
     for (auto& op : ops) {
-        MultiApplier::OperationPtrs opsPtrs;
-        opsPtrs.push_back(op);
-        WorkerMultikeyPathInfo pathInfo;
-        auto status = multiSyncApply(_opCtx.get(), &opsPtrs, &syncTail, &pathInfo);
+        auto status = syncTail.multiApply(_opCtx.get(), {op});
         if (!status.isOK()) {
-            return status;
+            return status.getStatus();
         }
+        auto lastApplied = status.getValue();
+        const bool orderedCommit = true;
+        // Update oplog visibility by notifying the storage engine of the new oplog entries.
+        storageInterface->oplogDiskLocRegister(
+            _opCtx.get(), lastApplied.getTimestamp(), orderedCommit);
     }
     return Status::OK();
 }
diff --git a/src/mongo/db/repl/sync_tail_test_fixture.h b/src/mongo/db/repl/sync_tail_test_fixture.h
index 9b1adaea94f..c69c3575220 100644
--- a/src/mongo/db/repl/sync_tail_test_fixture.h
+++ b/src/mongo/db/repl/sync_tail_test_fixture.h
@@ -146,7 +146,6 @@ protected:
     Status runOpsSteadyState(std::vector<OplogEntry> ops);
     Status runOpInitialSync(const OplogEntry& entry);
     Status runOpsInitialSync(std::vector<OplogEntry> ops);
-    Status runOpPtrsInitialSync(MultiApplier::OperationPtrs ops);
 
     UUID kUuid{UUID::gen()};
 };