SERVER-32776 add ReplicationProcess::refreshRollbackID()

This forces the ReplicationProcess to update its cached rollback ID from storage.

4 files changed, 48 insertions, 5 deletions

diff --git a/src/mongo/db/repl/replication_coordinator_impl.cpp b/src/mongo/db/repl/replication_coordinator_impl.cpp
index 9ebff1e9394..e552cccae98 100644
--- a/src/mongo/db/repl/replication_coordinator_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl.cpp
@@ -466,14 +466,14 @@ bool ReplicationCoordinatorImpl::_startLoadLocalConfig(OperationContext* opCtx)
     }
 
     // Check that we have a local Rollback ID. If we do not have one, create one.
-    auto rbid = _replicationProcess->getRollbackID(opCtx);
-    if (!rbid.isOK()) {
-        if (rbid.getStatus() == ErrorCodes::NamespaceNotFound) {
+    auto status = _replicationProcess->refreshRollbackID(opCtx);
+    if (!status.isOK()) {
+        if (status == ErrorCodes::NamespaceNotFound) {
             log() << "Did not find local Rollback ID document at startup. Creating one.";
             auto initializingStatus = _replicationProcess->initializeRollbackID(opCtx);
             fassertStatusOK(40424, initializingStatus);
         } else {
-            severe() << "Error loading local Rollback ID document at startup; " << rbid.getStatus();
+            severe() << "Error loading local Rollback ID document at startup; " << status;
             fassertFailedNoTrace(40428);
         }
     }
@@ -485,7 +485,7 @@ bool ReplicationCoordinatorImpl::_startLoadLocalConfig(OperationContext* opCtx)
         return true;
     }
     ReplSetConfig localConfig;
-    Status status = localConfig.initialize(cfg.getValue());
+    status = localConfig.initialize(cfg.getValue());
     if (!status.isOK()) {
         error() << "Locally stored replica set configuration does not parse; See "
                    "http://www.mongodb.org/dochub/core/recover-replica-set-from-invalid-config "
diff --git a/src/mongo/db/repl/replication_process.cpp b/src/mongo/db/repl/replication_process.cpp
index 89e1a3e9b99..3e4b9cd7669 100644
--- a/src/mongo/db/repl/replication_process.cpp
+++ b/src/mongo/db/repl/replication_process.cpp
@@ -83,6 +83,24 @@ ReplicationProcess::ReplicationProcess(
       _recovery(std::move(recovery)),
       _rbid(kUninitializedRollbackId) {}
 
+Status ReplicationProcess::refreshRollbackID(OperationContext* opCtx) {
+    stdx::lock_guard<stdx::mutex> lock(_mutex);
+
+    auto rbidResult = _storageInterface->getRollbackID(opCtx);
+    if (!rbidResult.isOK()) {
+        return rbidResult.getStatus();
+    }
+
+    if (kUninitializedRollbackId == _rbid) {
+        log() << "Rollback ID is " << rbidResult.getValue();
+    } else {
+        log() << "Rollback ID is " << rbidResult.getValue() << " (previously " << _rbid << ")";
+    }
+    _rbid = rbidResult.getValue();
+
+    return Status::OK();
+}
+
 StatusWith<int> ReplicationProcess::getRollbackID(OperationContext* opCtx) {
     stdx::lock_guard<stdx::mutex> lock(_mutex);
 
diff --git a/src/mongo/db/repl/replication_process.h b/src/mongo/db/repl/replication_process.h
index 848f0aa52c4..ae155fe09d5 100644
--- a/src/mongo/db/repl/replication_process.h
+++ b/src/mongo/db/repl/replication_process.h
@@ -78,6 +78,7 @@ public:
     /**
      * Rollback ID is an increasing counter of how many rollbacks have occurred on this server.
      */
+    Status refreshRollbackID(OperationContext* opCtx);
     StatusWith<int> getRollbackID(OperationContext* opCtx);
     Status initializeRollbackID(OperationContext* opCtx);
     Status incrementRollbackID(OperationContext* opCtx);
diff --git a/src/mongo/db/repl/replication_process_test.cpp b/src/mongo/db/repl/replication_process_test.cpp
index aa8f75245c6..e031cd7551f 100644
--- a/src/mongo/db/repl/replication_process_test.cpp
+++ b/src/mongo/db/repl/replication_process_test.cpp
@@ -105,4 +105,28 @@ TEST_F(ReplicationProcessTest, RollbackIDIncrementsBy1) {
     ASSERT_EQ(rbid, initRBID + 1);
 }
 
+TEST_F(ReplicationProcessTest, RefreshRollbackIDResetsCachedValueFromStorage) {
+    auto opCtx = makeOpCtx();
+    ReplicationProcess replicationProcess(
+        _storageInterface.get(),
+        stdx::make_unique<ReplicationConsistencyMarkersImpl>(_storageInterface.get()),
+        stdx::make_unique<ReplicationRecoveryMock>());
+
+    // RefreshRollbackID returns NamespaceNotFound if there is no rollback.id collection.
+    ASSERT_EQUALS(ErrorCodes::NamespaceNotFound, replicationProcess.refreshRollbackID(opCtx.get()));
+
+    // We make no assumptions about the initial value of the rollback ID.
+    ASSERT_OK(replicationProcess.initializeRollbackID(opCtx.get()));
+    int initRBID = unittest::assertGet(replicationProcess.getRollbackID(opCtx.get()));
+
+    // Increment rollback ID on disk. Cached value should different from storage.
+    int storageRBID = unittest::assertGet(_storageInterface->incrementRollbackID(opCtx.get()));
+    ASSERT_EQUALS(storageRBID, initRBID + 1);
+    ASSERT_EQUALS(initRBID, unittest::assertGet(replicationProcess.getRollbackID(opCtx.get())));
+
+    // Refresh cached value and check cached value against storage again.
+    ASSERT_OK(replicationProcess.refreshRollbackID(opCtx.get()));
+    ASSERT_EQUALS(storageRBID, unittest::assertGet(replicationProcess.getRollbackID(opCtx.get())));
+}
+
 }  // namespace