SERVER-26965 Use RAII type for turning off replicated writes

8 files changed, 18 insertions, 13 deletions

diff --git a/src/mongo/db/repl/collection_bulk_loader_impl.cpp b/src/mongo/db/repl/collection_bulk_loader_impl.cpp
index 9b1c96ec95b..2732376875a 100644
--- a/src/mongo/db/repl/collection_bulk_loader_impl.cpp
+++ b/src/mongo/db/repl/collection_bulk_loader_impl.cpp
@@ -93,6 +93,9 @@ Status CollectionBulkLoaderImpl::init(Collection* coll,
             invariant(opCtx);
             invariant(coll);
             invariant(opCtx->getClient() == &cc());
+            // All writes in CollectionBulkLoaderImpl should be unreplicated.
+            // The opCtx is accessed indirectly through _secondaryIndexesBlock.
+            UnreplicatedWritesBlock uwb(opCtx);
             std::vector<BSONObj> specs(secondaryIndexSpecs);
             // This enforces the buildIndexes setting in the replica set configuration.
             _secondaryIndexesBlock->removeExistingIndexes(&specs);
@@ -124,6 +127,7 @@ Status CollectionBulkLoaderImpl::insertDocuments(const std::vector<BSONObj>::con
     return _runTaskReleaseResourcesOnFailure(
         [begin, end, &count, this](OperationContext* opCtx) -> Status {
             invariant(opCtx);
+            UnreplicatedWritesBlock uwb(opCtx);
 
             for (auto iter = begin; iter != end; ++iter) {
                 std::vector<MultiIndexBlock*> indexers;
@@ -157,6 +161,7 @@ Status CollectionBulkLoaderImpl::commit() {
             LOG(2) << "Creating indexes for ns: " << _nss.ns();
             invariant(opCtx->getClient() == &cc());
             invariant(opCtx == _opCtx);
+            UnreplicatedWritesBlock uwb(opCtx);
 
             // Commit before deleting dups, so the dups will be removed from secondary indexes when
             // deleted.
diff --git a/src/mongo/db/repl/collection_cloner.cpp b/src/mongo/db/repl/collection_cloner.cpp
index 85c9f39169d..4a47981da2d 100644
--- a/src/mongo/db/repl/collection_cloner.cpp
+++ b/src/mongo/db/repl/collection_cloner.cpp
@@ -338,7 +338,7 @@ void CollectionCloner::_listIndexesCallback(const Fetcher::QueryResponseStatus&
                     return;
                 }
                 auto opCtx = cbd.opCtx;
-                opCtx->setReplicatedWrites(false);
+                UnreplicatedWritesBlock uwb(opCtx);
                 auto&& createStatus =
                     _storageInterface->createCollection(opCtx, _destNss, _options);
                 _finishCallback(createStatus);
diff --git a/src/mongo/db/repl/master_slave.cpp b/src/mongo/db/repl/master_slave.cpp
index 21194aeff8d..f76856d2802 100644
--- a/src/mongo/db/repl/master_slave.cpp
+++ b/src/mongo/db/repl/master_slave.cpp
@@ -731,7 +731,7 @@ void ReplSource::_sync_pullOpLog_applyOperation(OperationContext* opCtx,
     // Push the CurOp stack for "opCtx" so each individual oplog entry application is separately
     // reported.
     CurOp individualOp(opCtx);
-    opCtx->setReplicatedWrites(false);
+    UnreplicatedWritesBlock uwb(opCtx);
     const ReplSettings& replSettings = getGlobalReplicationCoordinator()->getSettings();
     if (replSettings.getPretouch() &&
         !alreadyLocked /*doesn't make sense if in write lock already*/) {
diff --git a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
index 04a2fcf88f8..c2619816034 100644
--- a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
@@ -373,7 +373,7 @@ Status ReplicationCoordinatorExternalStateImpl::runRepairOnLocalDB(OperationCont
             return Status::OK();
         }
 
-        opCtx->setReplicatedWrites(false);
+        UnreplicatedWritesBlock uwb(opCtx);
         Status status = repairDatabase(opCtx, engine, localDbName, false, false);
 
         // Open database before returning
@@ -660,9 +660,7 @@ void ReplicationCoordinatorExternalStateImpl::cleanUpLastApplyBatch(OperationCon
     }
 
     // Apply remaining ops one at at time, but don't log them because they are already logged.
-    const bool wereWritesReplicated = opCtx->writesAreReplicated();
-    ON_BLOCK_EXIT([&] { opCtx->setReplicatedWrites(wereWritesReplicated); });
-    opCtx->setReplicatedWrites(false);
+    UnreplicatedWritesBlock uwb(opCtx);
 
     while (cursor->more()) {
         auto entry = cursor->nextSafe();
diff --git a/src/mongo/db/repl/rs_initialsync.cpp b/src/mongo/db/repl/rs_initialsync.cpp
index a6a5cf78baf..24463e23091 100644
--- a/src/mongo/db/repl/rs_initialsync.cpp
+++ b/src/mongo/db/repl/rs_initialsync.cpp
@@ -265,7 +265,7 @@ bool _initialSyncApplyOplog(OperationContext* opCtx,
 Status _initialSync(OperationContext* opCtx, BackgroundSync* bgsync) {
     log() << "initial sync pending";
 
-    opCtx->setReplicatedWrites(false);
+    UnreplicatedWritesBlock uwb(opCtx);
     DisableDocumentValidation validationDisabler(opCtx);
     ReplicationCoordinator* replCoord(getGlobalReplicationCoordinator());
 
diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index ff5aa7d4260..bece9cedeb6 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -257,7 +257,7 @@ StorageInterfaceImpl::createCollectionForBulkLoading(
     auto status = runner->runSynchronousTask([&](OperationContext* opCtx) -> Status {
         // We are not replicating nor validating writes under this OperationContext*.
         // The OperationContext* is used for all writes to the (newly) cloned collection.
-        opCtx->setReplicatedWrites(false);
+        UnreplicatedWritesBlock uwb(opCtx);
         documentValidationDisabled(opCtx) = true;
 
         // Retry if WCE.
diff --git a/src/mongo/db/repl/storage_interface_impl_test.cpp b/src/mongo/db/repl/storage_interface_impl_test.cpp
index cd8bdd98686..e94761e5d35 100644
--- a/src/mongo/db/repl/storage_interface_impl_test.cpp
+++ b/src/mongo/db/repl/storage_interface_impl_test.cpp
@@ -195,6 +195,7 @@ protected:
         setGlobalReplicationCoordinator(_coordinator);
     }
     void tearDown() override {
+        _uwb.reset(nullptr);
         _opCtx.reset(nullptr);
         ServiceContextMongoDTest::tearDown();
     }
@@ -203,7 +204,7 @@ protected:
     void createOptCtx() {
         _opCtx = cc().makeOperationContext();
         // We are not replicating nor validating these writes.
-        _opCtx->setReplicatedWrites(false);
+        _uwb = stdx::make_unique<UnreplicatedWritesBlock>(_opCtx.get());
         DisableDocumentValidation validationDisabler(_opCtx.get());
     }
 
@@ -213,6 +214,7 @@ protected:
 
 private:
     ServiceContext::UniqueOperationContext _opCtx;
+    std::unique_ptr<UnreplicatedWritesBlock> _uwb;
 
     // Owned by service context
     ReplicationCoordinator* _coordinator;
diff --git a/src/mongo/db/repl/sync_tail.cpp b/src/mongo/db/repl/sync_tail.cpp
index 8738b47d027..06c86f5823f 100644
--- a/src/mongo/db/repl/sync_tail.cpp
+++ b/src/mongo/db/repl/sync_tail.cpp
@@ -325,7 +325,7 @@ Status SyncTail::syncApply(OperationContext* opCtx,
     auto applyOp = [&](Database* db) {
         // For non-initial-sync, we convert updates to upserts
         // to suppress errors when replaying oplog entries.
-        opCtx->setReplicatedWrites(false);
+        UnreplicatedWritesBlock uwb(opCtx);
         DisableDocumentValidation validationDisabler(opCtx);
 
         Status status =
@@ -481,7 +481,7 @@ void scheduleWritesToOplog(OperationContext* opCtx,
             const auto txnHolder = cc().makeOperationContext();
             const auto opCtx = txnHolder.get();
             opCtx->lockState()->setShouldConflictWithSecondaryBatchApplication(false);
-            opCtx->setReplicatedWrites(false);
+            UnreplicatedWritesBlock uwb(opCtx);
 
             std::vector<BSONObj> docs;
             docs.reserve(end - begin);
@@ -1062,7 +1062,7 @@ void multiSyncApply(MultiApplier::OperationPtrs* ops, SyncTail*) {
 Status multiSyncApply_noAbort(OperationContext* opCtx,
                               MultiApplier::OperationPtrs* oplogEntryPointers,
                               SyncApplyFn syncApply) {
-    opCtx->setReplicatedWrites(false);
+    UnreplicatedWritesBlock uwb(opCtx);
     DisableDocumentValidation validationDisabler(opCtx);
 
     // allow us to get through the magic barrier
@@ -1182,7 +1182,7 @@ Status multiInitialSyncApply_noAbort(OperationContext* opCtx,
                                      MultiApplier::OperationPtrs* ops,
                                      SyncTail* st,
                                      AtomicUInt32* fetchCount) {
-    opCtx->setReplicatedWrites(false);
+    UnreplicatedWritesBlock uwb(opCtx);
     DisableDocumentValidation validationDisabler(opCtx);
 
     // allow us to get through the magic barrier