SERVER-24273 added unit tests for repl::multiInitialSyncApply

4 files changed, 152 insertions, 16 deletions

diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index a405be517d8..8f5bd5759bc 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -271,6 +271,7 @@ env.CppUnitTest(
         'sync_tail_test.cpp',
     ],
     LIBDEPS=[
+        'oplog_interface_local',
         'replmocks',
         'sync_tail',
         '$BUILD_DIR/mongo/db/service_context_d_test_fixture',
diff --git a/src/mongo/db/repl/sync_tail.cpp b/src/mongo/db/repl/sync_tail.cpp
index efa68be10a4..6e9ceabc098 100644
--- a/src/mongo/db/repl/sync_tail.cpp
+++ b/src/mongo/db/repl/sync_tail.cpp
@@ -1123,27 +1123,30 @@ void multiSyncApply(const std::vector<OplogEntry>& ops, SyncTail*) {
 // This free function is used by the initial sync writer threads to apply each op
 void multiInitialSyncApply(const std::vector<OplogEntry>& ops, SyncTail* st) {
     initializeWriterThread();
+    auto txn = cc().makeOperationContext();
+    fassertNoTrace(15915, multiInitialSyncApply_noAbort(txn.get(), ops, st));
+}
 
-    const ServiceContext::UniqueOperationContext txnPtr = cc().makeOperationContext();
-    OperationContext& txn = *txnPtr;
-    txn.setReplicatedWrites(false);
-    DisableDocumentValidation validationDisabler(&txn);
+Status multiInitialSyncApply_noAbort(OperationContext* txn,
+                                     const std::vector<OplogEntry>& ops,
+                                     SyncTail* st) {
+    txn->setReplicatedWrites(false);
+    DisableDocumentValidation validationDisabler(txn);
 
     // allow us to get through the magic barrier
-    txn.lockState()->setIsBatchWriter(true);
+    txn->lockState()->setIsBatchWriter(true);
 
     bool convertUpdatesToUpserts = false;
 
     for (std::vector<OplogEntry>::const_iterator it = ops.begin(); it != ops.end(); ++it) {
         try {
-            const Status s = SyncTail::syncApply(&txn, it->raw, convertUpdatesToUpserts);
+            const Status s = SyncTail::syncApply(txn, it->raw, convertUpdatesToUpserts);
             if (!s.isOK()) {
-                if (st->shouldRetry(&txn, it->raw)) {
-                    const Status s2 = SyncTail::syncApply(&txn, it->raw, convertUpdatesToUpserts);
+                if (st->shouldRetry(txn, it->raw)) {
+                    const Status s2 = SyncTail::syncApply(txn, it->raw, convertUpdatesToUpserts);
                     if (!s2.isOK()) {
-                        severe() << "Error applying operation (" << it->raw.toString()
-                                 << "): " << s2;
-                        fassertFailedNoTrace(15915);
+                        severe() << "Error applying operation (" << it->raw << "): " << s2;
+                        return s2;
                     }
                 }
 
@@ -1152,16 +1155,17 @@ void multiInitialSyncApply(const std::vector<OplogEntry>& ops, SyncTail* st) {
                 // subsequently got deleted and no longer exists on the Sync Target at all
             }
         } catch (const DBException& e) {
-            severe() << "writer worker caught exception: " << causedBy(e)
-                     << " on: " << it->raw.toString();
+            severe() << "writer worker caught exception: " << causedBy(e) << " on: " << it->raw;
 
             if (inShutdown()) {
-                return;
+                return {ErrorCodes::InterruptedAtShutdown, e.toString()};
             }
 
-            fassertFailedNoTrace(16361);
+            return e.toStatus();
         }
     }
+
+    return Status::OK();
 }
 
 StatusWith<OpTime> multiApply(OperationContext* txn,
diff --git a/src/mongo/db/repl/sync_tail.h b/src/mongo/db/repl/sync_tail.h
index 745d5287a18..204fce1f037 100644
--- a/src/mongo/db/repl/sync_tail.h
+++ b/src/mongo/db/repl/sync_tail.h
@@ -202,5 +202,13 @@ StatusWith<OpTime> multiApply(OperationContext* txn,
 void multiSyncApply(const std::vector<OplogEntry>& ops, SyncTail* st);
 void multiInitialSyncApply(const std::vector<OplogEntry>& ops, SyncTail* st);
 
+/**
+ * Testing-only version of multiInitialSyncApply that accepts an external operation context and
+ * returns an error instead of aborting.
+ */
+Status multiInitialSyncApply_noAbort(OperationContext* txn,
+                                     const std::vector<OplogEntry>& ops,
+                                     SyncTail* st);
+
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/repl/sync_tail_test.cpp b/src/mongo/db/repl/sync_tail_test.cpp
index 5a02addfd22..ac3b6e1ddd2 100644
--- a/src/mongo/db/repl/sync_tail_test.cpp
+++ b/src/mongo/db/repl/sync_tail_test.cpp
@@ -43,6 +43,7 @@
 #include "mongo/db/db_raii.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/repl/bgsync.h"
+#include "mongo/db/repl/oplog_interface_local.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/repl/storage_interface.h"
@@ -75,6 +76,27 @@ private:
     void tearDown() override;
 };
 
+/**
+ * Testing-only SyncTail that returns user-provided "document" for getMissingDoc().
+ */
+class SyncTailWithLocalDocumentFetcher : public SyncTail {
+public:
+    SyncTailWithLocalDocumentFetcher(const BSONObj& document);
+    BSONObj getMissingDoc(OperationContext* txn, Database* db, const BSONObj& o) override;
+
+private:
+    BSONObj _document;
+};
+
+/**
+ * Testing-only SyncTail that checks the operation context in shouldRetry().
+ */
+class SyncTailWithOperationContextChecker : public SyncTail {
+public:
+    SyncTailWithOperationContextChecker();
+    bool shouldRetry(OperationContext* txn, const BSONObj& o) override;
+};
+
 void SyncTailTest::setUp() {
     ServiceContextMongoDTest::setUp();
     ReplSettings replSettings;
@@ -100,7 +122,7 @@ void SyncTailTest::setUp() {
 }
 
 void SyncTailTest::tearDown() {
-    {
+    if (_txn) {
         Lock::GlobalWrite globalLock(_txn->lockState());
         BSONObjBuilder unused;
         invariant(mongo::dbHolder().closeAll(_txn.get(), unused, false));
@@ -109,6 +131,25 @@ void SyncTailTest::tearDown() {
     _storageInterface = nullptr;
 }
 
+SyncTailWithLocalDocumentFetcher::SyncTailWithLocalDocumentFetcher(const BSONObj& document)
+    : SyncTail(nullptr, SyncTail::MultiSyncApplyFunc(), nullptr), _document(document) {}
+
+BSONObj SyncTailWithLocalDocumentFetcher::getMissingDoc(OperationContext*,
+                                                        Database*,
+                                                        const BSONObj&) {
+    return _document;
+}
+
+SyncTailWithOperationContextChecker::SyncTailWithOperationContextChecker()
+    : SyncTail(nullptr, SyncTail::MultiSyncApplyFunc(), nullptr) {}
+
+bool SyncTailWithOperationContextChecker::shouldRetry(OperationContext* txn, const BSONObj&) {
+    ASSERT_FALSE(txn->writesAreReplicated());
+    ASSERT_TRUE(txn->lockState()->isBatchWriter());
+    ASSERT_TRUE(documentValidationDisabled(txn));
+    return false;
+}
+
 /**
  * Creates collection options suitable for oplog.
  */
@@ -169,6 +210,23 @@ OplogEntry makeInsertDocumentOplogEntry(OpTime opTime,
     return OplogEntry(bob.obj());
 }
 
+/**
+ * Creates an update oplog entry with given optime and namespace.
+ */
+OplogEntry makeUpdateDocumentOplogEntry(OpTime opTime,
+                                        const NamespaceString& nss,
+                                        const BSONObj& documentToUpdate,
+                                        const BSONObj& updatedDocument) {
+    BSONObjBuilder bob;
+    bob.appendElements(opTime.toBSON());
+    bob.append("h", 1LL);
+    bob.append("op", "u");
+    bob.append("ns", nss.ns());
+    bob.append("o2", documentToUpdate);
+    bob.append("o", updatedDocument);
+    return OplogEntry(bob.obj());
+}
+
 TEST_F(SyncTailTest, SyncApplyNoNamespaceBadOp) {
     const BSONObj op = BSON("op"
                             << "x");
@@ -522,4 +580,69 @@ TEST_F(SyncTailTest, MultiApplyAssignsOperationsToWriterThreadsBasedOnNamespaceH
     ASSERT_EQUALS(op2, operationsWritternToOplog[1]);
 }
 
+TEST_F(SyncTailTest, MultiInitialSyncApplyDisablesDocumentValidationWhileApplyingOperations) {
+    SyncTailWithOperationContextChecker syncTail;
+    NamespaceString nss("local." + _agent.getSuiteName() + "_" + _agent.getTestName());
+    ASSERT_TRUE(_txn->writesAreReplicated());
+    ASSERT_FALSE(documentValidationDisabled(_txn.get()));
+    ASSERT_FALSE(_txn->lockState()->isBatchWriter());
+    auto op = makeUpdateDocumentOplogEntry(
+        {Timestamp(Seconds(1), 0), 1LL}, nss, BSON("_id" << 0), BSON("_id" << 0 << "x" << 2));
+    ASSERT_OK(multiInitialSyncApply_noAbort(_txn.get(), {op}, &syncTail));
+}
+
+TEST_F(SyncTailTest,
+       MultiInitialSyncApplyDoesNotRetryFailedUpdateIfDocumentIsMissingFromSyncSource) {
+    BSONObj emptyDoc;
+    SyncTailWithLocalDocumentFetcher syncTail(emptyDoc);
+    NamespaceString nss("local." + _agent.getSuiteName() + "_" + _agent.getTestName());
+    auto op = makeUpdateDocumentOplogEntry(
+        {Timestamp(Seconds(1), 0), 1LL}, nss, BSON("_id" << 0), BSON("_id" << 0 << "x" << 2));
+    ASSERT_OK(multiInitialSyncApply_noAbort(_txn.get(), {op}, &syncTail));
+
+    // Since the missing document is not found on the sync source, the collection referenced by
+    // the failed operation should not be automatically created.
+    ASSERT_FALSE(AutoGetCollectionForRead(_txn.get(), nss).getCollection());
+}
+
+TEST_F(SyncTailTest, MultiInitialSyncApplyRetriesFailedUpdateIfDocumentIsAvailableFromSyncSource) {
+    SyncTailWithLocalDocumentFetcher syncTail(BSON("_id" << 0 << "x" << 1));
+    NamespaceString nss("local." + _agent.getSuiteName() + "_" + _agent.getTestName());
+    auto updatedDocument = BSON("_id" << 0 << "x" << 2);
+    auto op = makeUpdateDocumentOplogEntry(
+        {Timestamp(Seconds(1), 0), 1LL}, nss, BSON("_id" << 0), updatedDocument);
+    ASSERT_OK(multiInitialSyncApply_noAbort(_txn.get(), {op}, &syncTail));
+
+    // The collection referenced by "ns" in the failed operation is automatically created to hold
+    // the missing document fetched from the sync source. We verify the contents of the collection
+    // with the OplogInterfaceLocal class.
+    OplogInterfaceLocal collectionReader(_txn.get(), nss.ns());
+    auto iter = collectionReader.makeIterator();
+    ASSERT_EQUALS(updatedDocument, unittest::assertGet(iter->next()).first);
+    ASSERT_EQUALS(ErrorCodes::NoSuchKey, iter->next().getStatus());
+}
+
+TEST_F(SyncTailTest, MultiInitialSyncApplyPassesThroughSyncApplyErrorAfterFailingToRetryBadOp) {
+    SyncTailWithLocalDocumentFetcher syncTail(BSON("_id" << 0 << "x" << 1));
+    NamespaceString nss("local." + _agent.getSuiteName() + "_" + _agent.getTestName());
+    auto updatedDocument = BSON("_id" << 0 << "x" << 2);
+    OplogEntry op(BSON("op"
+                       << "x"
+                       << "ns"
+                       << nss.ns()));
+    ASSERT_EQUALS(ErrorCodes::BadValue, multiInitialSyncApply_noAbort(_txn.get(), {op}, &syncTail));
+}
+
+TEST_F(SyncTailTest, MultiInitialSyncApplyPassesThroughShouldSyncTailRetryError) {
+    SyncTail syncTail(nullptr, SyncTail::MultiSyncApplyFunc(), nullptr);
+    NamespaceString nss("local." + _agent.getSuiteName() + "_" + _agent.getTestName());
+    auto updatedDocument = BSON("_id" << 0 << "x" << 2);
+    auto op = makeUpdateDocumentOplogEntry(
+        {Timestamp(Seconds(1), 0), 1LL}, nss, BSON("_id" << 0), BSON("_id" << 0 << "x" << 2));
+    ASSERT_THROWS_CODE(
+        syncTail.shouldRetry(_txn.get(), op.raw), mongo::UserException, ErrorCodes::FailedToParse);
+    ASSERT_EQUALS(ErrorCodes::FailedToParse,
+                  multiInitialSyncApply_noAbort(_txn.get(), {op}, &syncTail));
+}
+
 }  // namespace