SERVER-40785 Create WaitForMajorityService to allow waiting for write concern asynchronously

1 files changed, 45 insertions, 39 deletions

diff --git a/src/mongo/db/catalog/drop_database_test.cpp b/src/mongo/db/catalog/drop_database_test.cpp
index 4fe1d2bc098..2eb37e80a36 100644
--- a/src/mongo/db/catalog/drop_database_test.cpp
+++ b/src/mongo/db/catalog/drop_database_test.cpp
@@ -268,7 +268,7 @@ TEST_F(DropDatabaseTest, DropDatabaseWaitsForDropPendingCollectionOpTimeIfNoColl
 
     // Update ReplicationCoordinatorMock so that we record the optime passed to awaitReplication().
     _replCoord->setAwaitReplicationReturnValueFunction(
-        [&clientLastOpTime, this](const repl::OpTime& opTime) {
+        [&clientLastOpTime, this](OperationContext*, const repl::OpTime& opTime) {
             clientLastOpTime = opTime;
             ASSERT_GREATER_THAN(clientLastOpTime, repl::OpTime());
             return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
@@ -283,11 +283,12 @@ TEST_F(DropDatabaseTest, DropDatabaseWaitsForDropPendingCollectionOpTimeIfNoColl
 
 TEST_F(DropDatabaseTest, DropDatabasePassedThroughAwaitReplicationErrorForDropPendingCollection) {
     // Update ReplicationCoordinatorMock so that we record the optime passed to awaitReplication().
-    _replCoord->setAwaitReplicationReturnValueFunction([this](const repl::OpTime& opTime) {
-        ASSERT_GREATER_THAN(opTime, repl::OpTime());
-        return repl::ReplicationCoordinator::StatusAndDuration(
-            Status(ErrorCodes::WriteConcernFailed, ""), Milliseconds(0));
-    });
+    _replCoord->setAwaitReplicationReturnValueFunction(
+        [this](OperationContext*, const repl::OpTime& opTime) {
+            ASSERT_GREATER_THAN(opTime, repl::OpTime());
+            return repl::ReplicationCoordinator::StatusAndDuration(
+                Status(ErrorCodes::WriteConcernFailed, ""), Milliseconds(0));
+        });
 
     repl::OpTime dropOpTime(Timestamp(Seconds(100), 0), 1LL);
     auto dpns = _nss.makeDropPendingNamespace(dropOpTime);
@@ -342,7 +343,7 @@ void _testDropDatabaseResetsDropPendingStateIfAwaitReplicationFails(OperationCon
 TEST_F(DropDatabaseTest,
        DropDatabaseResetsDropPendingStateIfAwaitReplicationFailsAndDatabaseIsPresent) {
     // Update ReplicationCoordinatorMock so that awaitReplication() fails.
-    _replCoord->setAwaitReplicationReturnValueFunction([](const repl::OpTime&) {
+    _replCoord->setAwaitReplicationReturnValueFunction([](OperationContext*, const repl::OpTime&) {
         return repl::ReplicationCoordinator::StatusAndDuration(
             Status(ErrorCodes::WriteConcernFailed, ""), Milliseconds(0));
     });
@@ -353,11 +354,12 @@ TEST_F(DropDatabaseTest,
 TEST_F(DropDatabaseTest,
        DropDatabaseResetsDropPendingStateIfAwaitReplicationFailsAndDatabaseIsMissing) {
     // Update ReplicationCoordinatorMock so that awaitReplication() fails.
-    _replCoord->setAwaitReplicationReturnValueFunction([this](const repl::OpTime&) {
-        _removeDatabaseFromCatalog(_opCtx.get(), _nss.db());
-        return repl::ReplicationCoordinator::StatusAndDuration(
-            Status(ErrorCodes::WriteConcernFailed, ""), Milliseconds(0));
-    });
+    _replCoord->setAwaitReplicationReturnValueFunction(
+        [this](OperationContext*, const repl::OpTime&) {
+            _removeDatabaseFromCatalog(_opCtx.get(), _nss.db());
+            return repl::ReplicationCoordinator::StatusAndDuration(
+                Status(ErrorCodes::WriteConcernFailed, ""), Milliseconds(0));
+        });
 
     _testDropDatabaseResetsDropPendingStateIfAwaitReplicationFails(_opCtx.get(), _nss, false);
 }
@@ -368,15 +370,16 @@ TEST_F(DropDatabaseTest,
     // dropDatabase() should detect this and release the global lock temporarily if it needs to call
     // ReplicationCoordinator::awaitReplication().
     bool isAwaitReplicationCalled = false;
-    _replCoord->setAwaitReplicationReturnValueFunction([&, this](const repl::OpTime& opTime) {
-        isAwaitReplicationCalled = true;
-        // This test does not set the client's last optime.
-        ASSERT_EQUALS(opTime, repl::OpTime());
-        ASSERT_FALSE(_opCtx->lockState()->isW());
-        ASSERT_FALSE(_opCtx->lockState()->isDbLockedForMode(_nss.db(), MODE_X));
-        ASSERT_FALSE(_opCtx->lockState()->isLocked());
-        return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
-    });
+    _replCoord->setAwaitReplicationReturnValueFunction(
+        [&, this](OperationContext*, const repl::OpTime& opTime) {
+            isAwaitReplicationCalled = true;
+            // This test does not set the client's last optime.
+            ASSERT_EQUALS(opTime, repl::OpTime());
+            ASSERT_FALSE(_opCtx->lockState()->isW());
+            ASSERT_FALSE(_opCtx->lockState()->isDbLockedForMode(_nss.db(), MODE_X));
+            ASSERT_FALSE(_opCtx->lockState()->isLocked());
+            return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
+        });
 
     {
         Lock::GlobalWrite lk(_opCtx.get());
@@ -392,14 +395,15 @@ TEST_F(DropDatabaseTest,
     // dropDatabase() should detect this and release the global lock temporarily if it needs to call
     // ReplicationCoordinator::awaitReplication().
     bool isAwaitReplicationCalled = false;
-    _replCoord->setAwaitReplicationReturnValueFunction([&, this](const repl::OpTime& opTime) {
-        isAwaitReplicationCalled = true;
-        ASSERT_GREATER_THAN(opTime, repl::OpTime());
-        ASSERT_FALSE(_opCtx->lockState()->isW());
-        ASSERT_FALSE(_opCtx->lockState()->isDbLockedForMode(_nss.db(), MODE_X));
-        ASSERT_FALSE(_opCtx->lockState()->isLocked());
-        return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
-    });
+    _replCoord->setAwaitReplicationReturnValueFunction(
+        [&, this](OperationContext*, const repl::OpTime& opTime) {
+            isAwaitReplicationCalled = true;
+            ASSERT_GREATER_THAN(opTime, repl::OpTime());
+            ASSERT_FALSE(_opCtx->lockState()->isW());
+            ASSERT_FALSE(_opCtx->lockState()->isDbLockedForMode(_nss.db(), MODE_X));
+            ASSERT_FALSE(_opCtx->lockState()->isLocked());
+            return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
+        });
 
     repl::OpTime dropOpTime(Timestamp(Seconds(100), 0), 1LL);
     auto dpns = _nss.makeDropPendingNamespace(dropOpTime);
@@ -416,10 +420,11 @@ TEST_F(DropDatabaseTest,
 TEST_F(DropDatabaseTest,
        DropDatabaseReturnsNamespaceNotFoundIfDatabaseIsRemovedAfterCollectionsDropsAreReplicated) {
     // Update ReplicationCoordinatorMock so that awaitReplication() fails.
-    _replCoord->setAwaitReplicationReturnValueFunction([this](const repl::OpTime&) {
-        _removeDatabaseFromCatalog(_opCtx.get(), _nss.db());
-        return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
-    });
+    _replCoord->setAwaitReplicationReturnValueFunction(
+        [this](OperationContext*, const repl::OpTime&) {
+            _removeDatabaseFromCatalog(_opCtx.get(), _nss.db());
+            return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
+        });
 
     _createCollection(_opCtx.get(), _nss);
 
@@ -438,12 +443,13 @@ TEST_F(DropDatabaseTest,
 TEST_F(DropDatabaseTest,
        DropDatabaseReturnsNotMasterIfNotPrimaryAfterCollectionsDropsAreReplicated) {
     // Transition from PRIMARY to SECONDARY while awaiting replication of collection drops.
-    _replCoord->setAwaitReplicationReturnValueFunction([this](const repl::OpTime&) {
-        ASSERT_OK(_replCoord->setFollowerMode(repl::MemberState::RS_SECONDARY));
-        ASSERT_TRUE(_opCtx->writesAreReplicated());
-        ASSERT_FALSE(_replCoord->canAcceptWritesForDatabase(_opCtx.get(), _nss.db()));
-        return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
-    });
+    _replCoord->setAwaitReplicationReturnValueFunction(
+        [this](OperationContext*, const repl::OpTime&) {
+            ASSERT_OK(_replCoord->setFollowerMode(repl::MemberState::RS_SECONDARY));
+            ASSERT_TRUE(_opCtx->writesAreReplicated());
+            ASSERT_FALSE(_replCoord->canAcceptWritesForDatabase(_opCtx.get(), _nss.db()));
+            return repl::ReplicationCoordinator::StatusAndDuration(Status::OK(), Milliseconds(0));
+        });
 
     _createCollection(_opCtx.get(), _nss);