SERVER-29254 Moved MinValid into ReplicationProcess

38 files changed, 1208 insertions, 627 deletions

diff --git a/src/mongo/db/db.cpp b/src/mongo/db/db.cpp
index 52d09a884a6..3803808efab 100644
--- a/src/mongo/db/db.cpp
+++ b/src/mongo/db/db.cpp
@@ -84,6 +84,7 @@
 #include "mongo/db/repl/drop_pending_collection_reaper.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/repl_settings.h"
+#include "mongo/db/repl/replication_consistency_markers_impl.h"
 #include "mongo/db/repl/replication_coordinator_external_state_impl.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/repl/replication_coordinator_impl.h"
@@ -895,8 +896,11 @@ MONGO_INITIALIZER_WITH_PREREQUISITES(CreateReplicationManager,
     repl::StorageInterface::set(serviceContext, stdx::make_unique<repl::StorageInterfaceImpl>());
     auto storageInterface = repl::StorageInterface::get(serviceContext);
 
-    repl::ReplicationProcess::set(serviceContext,
-                                  stdx::make_unique<repl::ReplicationProcess>(storageInterface));
+    repl::ReplicationProcess::set(
+        serviceContext,
+        stdx::make_unique<repl::ReplicationProcess>(
+            storageInterface,
+            stdx::make_unique<repl::ReplicationConsistencyMarkersImpl>(storageInterface)));
     auto replicationProcess = repl::ReplicationProcess::get(serviceContext);
 
     repl::DropPendingCollectionReaper::set(
@@ -914,7 +918,7 @@ MONGO_INITIALIZER_WITH_PREREQUISITES(CreateReplicationManager,
         serviceContext,
         getGlobalReplSettings(),
         stdx::make_unique<repl::ReplicationCoordinatorExternalStateImpl>(
-            serviceContext, dropPendingCollectionReaper, storageInterface),
+            serviceContext, dropPendingCollectionReaper, storageInterface, replicationProcess),
         makeReplicationExecutor(serviceContext),
         stdx::make_unique<repl::TopologyCoordinatorImpl>(topoCoordOptions),
         replicationProcess,
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index 3809c5e62d7..c63f661058e 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -185,8 +185,33 @@ env.CppUnitTest(
 )
 
 env.Library(
+    target='replication_consistency_markers_impl',
+    source=[
+        'replication_consistency_markers_impl.cpp',
+    ],
+    LIBDEPS=[
+        'optime',
+        'repl_coordinator_impl',
+    ],
+)
+
+env.CppUnitTest(
+    target='replication_consistency_markers_impl_test',
+    source=[
+        'replication_consistency_markers_impl_test.cpp',
+    ],
+    LIBDEPS=[
+        'replication_consistency_markers_impl',
+        'replmocks',
+        'storage_interface_impl',
+        '$BUILD_DIR/mongo/db/service_context_d_test_fixture',
+    ],
+)
+
+env.Library(
     target='replication_process',
     source=[
+        'replication_consistency_markers.cpp',
         'replication_process.cpp',
     ],
     LIBDEPS=[
@@ -204,6 +229,7 @@ env.CppUnitTest(
         'replication_process_test.cpp',
     ],
     LIBDEPS=[
+        'replication_consistency_markers_impl',
         'replication_process',
         'replmocks',
         'storage_interface_impl',
@@ -392,6 +418,7 @@ env.Library(
     ],
     LIBDEPS=[
         'optime',
+        'replication_process',
         'replmocks',
         #'serveronly', # CYCLE
         '$BUILD_DIR/mongo/db/service_context',
@@ -502,6 +529,7 @@ env.CppUnitTest(
     ],
     LIBDEPS=[
         'oplog_interface_local',
+        'replication_process',
         'replmocks',
         'sync_tail',
         '$BUILD_DIR/mongo/db/service_context_d_test_fixture',
@@ -739,6 +767,7 @@ env.Library(
     ],
     LIBDEPS=[
         'optime',
+        'replication_process',
         '$BUILD_DIR/mongo/db/service_context',
         '$BUILD_DIR/mongo/util/net/hostandport',
         '$BUILD_DIR/mongo/util/decorable',
@@ -752,6 +781,7 @@ env.Library('repl_coordinator_global',
 env.Library(
     target='replmocks',
     source=[
+        'replication_consistency_markers_mock.cpp',
         'replication_coordinator_external_state_mock.cpp',
         'replication_coordinator_mock.cpp',
         'service_context_repl_mock.cpp',
@@ -1271,6 +1301,7 @@ env.CppUnitTest(
         'base_cloner_test_fixture',
         'initial_syncer',
         'data_replicator_external_state_mock',
+        'replication_process',
         'replmocks',
         'service_context_repl_mock_init',
         'sync_source_selector_mock',
@@ -1383,6 +1414,7 @@ env.Library(
         'repl_coordinator_interface',
         'repl_coordinator_impl',
         'repl_settings',
+        'replication_consistency_markers_impl',
         'replication_process',
         'sync_tail',
         '$BUILD_DIR/mongo/base',
diff --git a/src/mongo/db/repl/bgsync.cpp b/src/mongo/db/repl/bgsync.cpp
index 8bd3c0e8ab0..be735cb0b6d 100644
--- a/src/mongo/db/repl/bgsync.cpp
+++ b/src/mongo/db/repl/bgsync.cpp
@@ -45,6 +45,7 @@
 #include "mongo/db/repl/oplog_interface_local.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/repl/replication_coordinator_impl.h"
+#include "mongo/db/repl/replication_process.h"
 #include "mongo/db/repl/rollback_source_impl.h"
 #include "mongo/db/repl/rs_rollback.h"
 #include "mongo/db/repl/rs_sync.h"
@@ -135,10 +136,12 @@ static ServerStatusMetricField<Counter64> displayBufferMaxSize("repl.buffer.maxS
 
 BackgroundSync::BackgroundSync(
     ReplicationCoordinatorExternalState* replicationCoordinatorExternalState,
+    ReplicationProcess* replicationProcess,
     std::unique_ptr<OplogBuffer> oplogBuffer)
     : _oplogBuffer(std::move(oplogBuffer)),
       _replCoord(getGlobalReplicationCoordinator()),
-      _replicationCoordinatorExternalState(replicationCoordinatorExternalState) {
+      _replicationCoordinatorExternalState(replicationCoordinatorExternalState),
+      _replicationProcess(replicationProcess) {
     // Update "repl.buffer.maxSizeBytes" server status metric to reflect the current oplog buffer's
     // max size.
     bufferMaxSizeGauge.increment(_oplogBuffer->getMaxSize() - bufferMaxSizeGauge.get());
@@ -274,14 +277,14 @@ void BackgroundSync::_produce(OperationContext* opCtx) {
         }
     }
 
-    auto storageInterface = StorageInterface::get(opCtx);
     // find a target to sync from the last optime fetched
     OpTime lastOpTimeFetched;
     HostAndPort source;
     HostAndPort oldSource = _syncSourceHost;
     SyncSourceResolverResponse syncSourceResp;
     {
-        const OpTime minValidSaved = storageInterface->getMinValid(opCtx);
+        const OpTime minValidSaved =
+            _replicationProcess->getConsistencyMarkers()->getMinValid(opCtx);
 
         stdx::lock_guard<stdx::mutex> lock(_mutex);
         if (_state != ProducerState::Running) {
@@ -405,8 +408,9 @@ void BackgroundSync::_produce(OperationContext* opCtx) {
     // Set the applied point if unset. This is most likely the first time we've established a sync
     // source since stepping down or otherwise clearing the applied point. We need to set this here,
     // before the OplogWriter gets a chance to append to the oplog.
-    if (storageInterface->getAppliedThrough(opCtx).isNull()) {
-        storageInterface->setAppliedThrough(opCtx, _replCoord->getMyLastAppliedOpTime());
+    if (_replicationProcess->getConsistencyMarkers()->getAppliedThrough(opCtx).isNull()) {
+        _replicationProcess->getConsistencyMarkers()->setAppliedThrough(
+            opCtx, _replCoord->getMyLastAppliedOpTime());
     }
 
     // "lastFetched" not used. Already set in _enqueueDocuments.
@@ -477,6 +481,7 @@ void BackgroundSync::_produce(OperationContext* opCtx) {
         // if it can't return a matching oplog start from the last fetch oplog ts field.
         return;
     } else if (fetcherReturnStatus.code() == ErrorCodes::OplogStartMissing) {
+        auto storageInterface = StorageInterface::get(opCtx);
         _runRollback(opCtx, fetcherReturnStatus, source, syncSourceResp.rbid, storageInterface);
     } else if (fetcherReturnStatus == ErrorCodes::InvalidBSON) {
         Seconds blacklistDuration(60);
@@ -618,7 +623,7 @@ void BackgroundSync::_runRollback(OperationContext* opCtx,
     OplogInterfaceLocal localOplog(opCtx, rsOplogName);
     if (use3dot4Rollback) {
         log() << "Rollback falling back on 3.4 algorithm due to startup server parameter";
-        _fallBackOn3dot4Rollback(opCtx, source, requiredRBID, &localOplog, storageInterface);
+        _fallBackOn3dot4Rollback(opCtx, source, requiredRBID, &localOplog);
     } else {
         AbstractAsyncComponent* rollback;
         StatusWith<OpTime> onRollbackShutdownResult =
@@ -661,8 +666,7 @@ void BackgroundSync::_runRollback(OperationContext* opCtx,
                       << onRollbackShutdownResult.getValue();
             } else if (ErrorCodes::IncompatibleRollbackAlgorithm == status) {
                 log() << "Rollback falling back on 3.4 algorithm due to " << status;
-                _fallBackOn3dot4Rollback(
-                    opCtx, source, requiredRBID, &localOplog, storageInterface);
+                _fallBackOn3dot4Rollback(opCtx, source, requiredRBID, &localOplog);
             } else {
                 warning() << "Rollback failed with error: " << status;
             }
@@ -677,8 +681,7 @@ void BackgroundSync::_runRollback(OperationContext* opCtx,
 void BackgroundSync::_fallBackOn3dot4Rollback(OperationContext* opCtx,
                                               const HostAndPort& source,
                                               int requiredRBID,
-                                              OplogInterface* localOplog,
-                                              StorageInterface* storageInterface) {
+                                              OplogInterface* localOplog) {
     const int messagingPortTags = 0;
     ConnectionPool connectionPool(messagingPortTags);
     std::unique_ptr<ConnectionPool::ConnectionPtr> connection;
@@ -691,7 +694,7 @@ void BackgroundSync::_fallBackOn3dot4Rollback(OperationContext* opCtx,
     };
 
     RollbackSourceImpl rollbackSource(getConnection, source, rsOplogName);
-    rollback(opCtx, *localOplog, rollbackSource, requiredRBID, _replCoord, storageInterface);
+    rollback(opCtx, *localOplog, rollbackSource, requiredRBID, _replCoord, _replicationProcess);
 }
 
 HostAndPort BackgroundSync::getSyncTarget() const {
diff --git a/src/mongo/db/repl/bgsync.h b/src/mongo/db/repl/bgsync.h
index e76bcc0dd83..d208261309e 100644
--- a/src/mongo/db/repl/bgsync.h
+++ b/src/mongo/db/repl/bgsync.h
@@ -55,6 +55,7 @@ namespace repl {
 class OplogInterface;
 class ReplicationCoordinator;
 class ReplicationCoordinatorExternalState;
+class ReplicationProcess;
 class StorageInterface;
 
 class BackgroundSync {
@@ -76,6 +77,7 @@ public:
     enum class ProducerState { Starting, Running, Stopped };
 
     BackgroundSync(ReplicationCoordinatorExternalState* replicationCoordinatorExternalState,
+                   ReplicationProcess* replicationProcess,
                    std::unique_ptr<OplogBuffer> oplogBuffer);
 
     // stop syncing (when this node becomes a primary, e.g.)
@@ -187,8 +189,7 @@ private:
     void _fallBackOn3dot4Rollback(OperationContext* opCtx,
                                   const HostAndPort& source,
                                   int requiredRBID,
-                                  OplogInterface* localOplog,
-                                  StorageInterface* storageInterface);
+                                  OplogInterface* localOplog);
 
     // restart syncing
     void start(OperationContext* opCtx);
@@ -204,6 +205,9 @@ private:
     // A pointer to the replication coordinator external state.
     ReplicationCoordinatorExternalState* _replicationCoordinatorExternalState;
 
+    // A pointer to the replication process.
+    ReplicationProcess* _replicationProcess;
+
     /**
       * All member variables are labeled with one of the following codes indicating the
       * synchronization rules for accessing them:
diff --git a/src/mongo/db/repl/initial_syncer.cpp b/src/mongo/db/repl/initial_syncer.cpp
index b036ed08f7e..3923d3402b7 100644
--- a/src/mongo/db/repl/initial_syncer.cpp
+++ b/src/mongo/db/repl/initial_syncer.cpp
@@ -51,6 +51,7 @@
 #include "mongo/db/repl/oplog_buffer.h"
 #include "mongo/db/repl/oplog_fetcher.h"
 #include "mongo/db/repl/optime.h"
+#include "mongo/db/repl/replication_process.h"
 #include "mongo/db/repl/storage_interface.h"
 #include "mongo/db/repl/sync_source_selector.h"
 #include "mongo/db/server_parameters.h"
@@ -203,15 +204,18 @@ InitialSyncer::InitialSyncer(
     InitialSyncerOptions opts,
     std::unique_ptr<DataReplicatorExternalState> dataReplicatorExternalState,
     StorageInterface* storage,
+    ReplicationProcess* replicationProcess,
     const OnCompletionFn& onCompletion)
     : _fetchCount(0),
       _opts(opts),
       _dataReplicatorExternalState(std::move(dataReplicatorExternalState)),
       _exec(_dataReplicatorExternalState->getTaskExecutor()),
       _storage(storage),
+      _replicationProcess(replicationProcess),
       _onCompletion(onCompletion) {
     uassert(ErrorCodes::BadValue, "task executor cannot be null", _exec);
     uassert(ErrorCodes::BadValue, "invalid storage interface", _storage);
+    uassert(ErrorCodes::BadValue, "invalid replication process", _replicationProcess);
     uassert(ErrorCodes::BadValue, "invalid getMyLastOptime function", _opts.getMyLastOptime);
     uassert(ErrorCodes::BadValue, "invalid setMyLastOptime function", _opts.setMyLastOptime);
     uassert(ErrorCodes::BadValue, "invalid getSlaveDelay function", _opts.getSlaveDelay);
@@ -381,9 +385,8 @@ void InitialSyncer::setScheduleDbWorkFn_forTest(const CollectionCloner::Schedule
 }
 
 void InitialSyncer::_setUp_inlock(OperationContext* opCtx, std::uint32_t initialSyncMaxAttempts) {
-    // This will call through to the storageInterfaceImpl to ReplicationCoordinatorImpl.
     // 'opCtx' is passed through from startup().
-    _storage->setInitialSyncFlag(opCtx);
+    _replicationProcess->getConsistencyMarkers()->setInitialSyncFlag(opCtx);
 
     LOG(1) << "Creating oplogBuffer.";
     _oplogBuffer = _dataReplicatorExternalState->makeInitialSyncOplogBuffer(opCtx);
@@ -405,7 +408,7 @@ void InitialSyncer::_tearDown_inlock(OperationContext* opCtx,
     if (!lastApplied.isOK()) {
         return;
     }
-    _storage->clearInitialSyncFlag(opCtx);
+    _replicationProcess->getConsistencyMarkers()->clearInitialSyncFlag(opCtx);
     _opts.setMyLastOptime(lastApplied.getValue().opTime);
     log() << "initial sync done; took "
           << duration_cast<Seconds>(_stats.initialSyncEnd - _stats.initialSyncStart) << ".";
diff --git a/src/mongo/db/repl/initial_syncer.h b/src/mongo/db/repl/initial_syncer.h
index 81e42c7956e..b438a0c97e1 100644
--- a/src/mongo/db/repl/initial_syncer.h
+++ b/src/mongo/db/repl/initial_syncer.h
@@ -71,9 +71,9 @@ MONGO_FP_FORWARD_DECLARE(rsSyncApplyStop);
 
 struct InitialSyncState;
 struct MemberState;
+class ReplicationProcess;
 class StorageInterface;
 
-
 struct InitialSyncerOptions {
     /** Function to return optime of last operation applied on this node */
     using GetMyLastOptimeFn = stdx::function<OpTime()>;
@@ -174,6 +174,7 @@ public:
     InitialSyncer(InitialSyncerOptions opts,
                   std::unique_ptr<DataReplicatorExternalState> dataReplicatorExternalState,
                   StorageInterface* storage,
+                  ReplicationProcess* replicationProcess,
                   const OnCompletionFn& onCompletion);
 
     virtual ~InitialSyncer();
@@ -558,6 +559,7 @@ private:
     std::unique_ptr<DataReplicatorExternalState> _dataReplicatorExternalState;  // (R)
     executor::TaskExecutor* _exec;                                              // (R)
     StorageInterface* _storage;                                                 // (R)
+    ReplicationProcess* _replicationProcess;                                    // (S)
 
     // This is invoked with the final status of the initial sync. If startup() fails, this callback
     // is never invoked. The caller gets the last applied optime with hash when the initial sync
diff --git a/src/mongo/db/repl/initial_syncer_test.cpp b/src/mongo/db/repl/initial_syncer_test.cpp
index 1821e3f491d..6dd7bfd5f66 100644
--- a/src/mongo/db/repl/initial_syncer_test.cpp
+++ b/src/mongo/db/repl/initial_syncer_test.cpp
@@ -43,6 +43,8 @@
 #include "mongo/db/repl/oplog_entry.h"
 #include "mongo/db/repl/oplog_fetcher.h"
 #include "mongo/db/repl/optime.h"
+#include "mongo/db/repl/replication_consistency_markers_mock.h"
+#include "mongo/db/repl/replication_process.h"
 #include "mongo/db/repl/reporter.h"
 #include "mongo/db/repl/storage_interface.h"
 #include "mongo/db/repl/storage_interface_mock.h"
@@ -280,6 +282,9 @@ protected:
 
         launchExecutorThread();
 
+        _replicationProcess = stdx::make_unique<ReplicationProcess>(
+            _storageInterface.get(), stdx::make_unique<ReplicationConsistencyMarkersMock>());
+
         _executorProxy = stdx::make_unique<TaskExecutorMock>(&getExecutor());
 
         _myLastOpTime = OpTime({3, 0}, 1);
@@ -336,6 +341,7 @@ protected:
                 options,
                 std::move(dataReplicatorExternalState),
                 _storageInterface.get(),
+                _replicationProcess.get(),
                 [this](const StatusWith<OpTimeWithHash>& lastApplied) {
                     _onCompletion(lastApplied);
                 });
@@ -363,6 +369,7 @@ protected:
         _initialSyncer.reset();
         _dbWorkThreadPool->join();
         _dbWorkThreadPool.reset();
+        _replicationProcess.reset();
         _storageInterface.reset();
 
         // tearDown() destroys the task executor which was referenced by the initial syncer.
@@ -393,6 +400,7 @@ protected:
     OpTime _myLastOpTime;
     std::unique_ptr<SyncSourceSelectorMock> _syncSourceSelector;
     std::unique_ptr<StorageInterfaceMock> _storageInterface;
+    std::unique_ptr<ReplicationProcess> _replicationProcess;
     std::unique_ptr<OldThreadPool> _dbWorkThreadPool;
     std::map<NamespaceString, CollectionMockStats> _collectionStats;
     std::map<NamespaceString, CollectionCloneInfo> _collections;
@@ -525,12 +533,14 @@ TEST_F(InitialSyncerTest, InvalidConstruction) {
     // Null task executor in external state.
     {
         auto dataReplicatorExternalState = stdx::make_unique<DataReplicatorExternalStateMock>();
-        ASSERT_THROWS_CODE_AND_WHAT(
-            InitialSyncer(
-                options, std::move(dataReplicatorExternalState), _storageInterface.get(), callback),
-            UserException,
-            ErrorCodes::BadValue,
-            "task executor cannot be null");
+        ASSERT_THROWS_CODE_AND_WHAT(InitialSyncer(options,
+                                                  std::move(dataReplicatorExternalState),
+                                                  _storageInterface.get(),
+                                                  _replicationProcess.get(),
+                                                  callback),
+                                    UserException,
+                                    ErrorCodes::BadValue,
+                                    "task executor cannot be null");
     }
 
     // Null callback function.
@@ -540,6 +550,7 @@ TEST_F(InitialSyncerTest, InvalidConstruction) {
         ASSERT_THROWS_CODE_AND_WHAT(InitialSyncer(options,
                                                   std::move(dataReplicatorExternalState),
                                                   _storageInterface.get(),
+                                                  _replicationProcess.get(),
                                                   InitialSyncer::OnCompletionFn()),
                                     UserException,
                                     ErrorCodes::BadValue,
@@ -598,13 +609,13 @@ TEST_F(InitialSyncerTest, StartupSetsInitialSyncFlagOnSuccess) {
     auto opCtx = makeOpCtx();
 
     // Initial sync flag should not be set before starting.
-    ASSERT_FALSE(getStorage().getInitialSyncFlag(opCtx.get()));
+    ASSERT_FALSE(_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx.get()));
 
     ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
     ASSERT_TRUE(initialSyncer->isActive());
 
     // Initial sync flag should be set.
-    ASSERT_TRUE(getStorage().getInitialSyncFlag(opCtx.get()));
+    ASSERT_TRUE(_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx.get()));
 }
 
 TEST_F(InitialSyncerTest, InitialSyncerReturnsCallbackCanceledIfShutdownImmediatelyAfterStartup) {
@@ -832,6 +843,7 @@ TEST_F(InitialSyncerTest, InitialSyncerResetsOnCompletionCallbackFunctionPointer
         _options,
         std::move(dataReplicatorExternalState),
         _storageInterface.get(),
+        _replicationProcess.get(),
         [&lastApplied, sharedCallbackData](const StatusWith<OpTimeWithHash>& result) {
             lastApplied = result;
         });
@@ -2385,7 +2397,7 @@ TEST_F(InitialSyncerTest, LastOpTimeShouldBeSetEvenIfNoOperationsAreAppliedAfter
     _syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
     ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
 
-    ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
+    ASSERT_TRUE(_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx.get()));
 
     auto oplogEntry = makeOplogEntry(1);
     auto net = getNet();
@@ -2464,7 +2476,7 @@ TEST_F(InitialSyncerTest, LastOpTimeShouldBeSetEvenIfNoOperationsAreAppliedAfter
     initialSyncer->join();
     ASSERT_EQUALS(oplogEntry.getOpTime(), unittest::assertGet(_lastApplied).opTime);
     ASSERT_EQUALS(oplogEntry.getHash(), unittest::assertGet(_lastApplied).value);
-    ASSERT_FALSE(_storageInterface->getInitialSyncFlag(opCtx.get()));
+    ASSERT_FALSE(_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx.get()));
 }
 
 TEST_F(InitialSyncerTest, InitialSyncerPassesThroughGetNextApplierBatchScheduleError) {
@@ -2474,7 +2486,7 @@ TEST_F(InitialSyncerTest, InitialSyncerPassesThroughGetNextApplierBatchScheduleE
     _syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
     ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
 
-    ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
+    ASSERT_TRUE(_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx.get()));
 
     auto net = getNet();
     int baseRollbackId = 1;
@@ -2528,7 +2540,7 @@ TEST_F(InitialSyncerTest, InitialSyncerPassesThroughSecondGetNextApplierBatchSch
     _syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
     ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
 
-    ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
+    ASSERT_TRUE(_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx.get()));
 
     auto net = getNet();
     int baseRollbackId = 1;
@@ -2582,7 +2594,7 @@ TEST_F(InitialSyncerTest, InitialSyncerCancelsGetNextApplierBatchOnShutdown) {
     _syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
     ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
 
-    ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
+    ASSERT_TRUE(_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx.get()));
 
     auto net = getNet();
     int baseRollbackId = 1;
@@ -2632,7 +2644,7 @@ TEST_F(InitialSyncerTest, InitialSyncerPassesThroughGetNextApplierBatchInLockErr
     _syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
     ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
 
-    ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
+    ASSERT_TRUE(_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx.get()));
 
     // _getNextApplierBatch_inlock() returns BadValue when it gets an oplog entry with an unexpected
     // version (not OplogEntry::kOplogVersion).
@@ -2690,7 +2702,7 @@ TEST_F(
     _syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
     ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
 
-    ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
+    ASSERT_TRUE(_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx.get()));
 
     // _getNextApplierBatch_inlock() returns BadValue when it gets an oplog entry with an unexpected
     // version (not OplogEntry::kOplogVersion).
@@ -2757,7 +2769,7 @@ TEST_F(InitialSyncerTest, InitialSyncerPassesThroughMultiApplierScheduleError) {
     _syncSourceSelector->setChooseNewSyncSourceResult_forTest(HostAndPort("localhost", 12345));
     ASSERT_OK(initialSyncer->startup(opCtx.get(), maxAttempts));
 
-    ASSERT_TRUE(_storageInterface->getInitialSyncFlag(opCtx.get()));
+    ASSERT_TRUE(_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx.get()));
 
     auto net = getNet();
     int baseRollbackId = 1;
diff --git a/src/mongo/db/repl/repl_set_commands.cpp b/src/mongo/db/repl/repl_set_commands.cpp
index 44d75d15c51..f34b91f4af2 100644
--- a/src/mongo/db/repl/repl_set_commands.cpp
+++ b/src/mongo/db/repl/repl_set_commands.cpp
@@ -357,7 +357,8 @@ public:
             ReplicationCoordinatorExternalStateImpl externalState(
                 opCtx->getServiceContext(),
                 DropPendingCollectionReaper::get(opCtx),
-                StorageInterface::get(opCtx));
+                StorageInterface::get(opCtx),
+                ReplicationProcess::get(opCtx));
             std::string name;
             std::vector<HostAndPort> seeds;
             parseReplSetSeedList(&externalState, replSetString, &name, &seeds);  // may throw...
diff --git a/src/mongo/db/repl/replication_consistency_markers.cpp b/src/mongo/db/repl/replication_consistency_markers.cpp
new file mode 100644
index 00000000000..4e7fadb9e84
--- /dev/null
+++ b/src/mongo/db/repl/replication_consistency_markers.cpp
@@ -0,0 +1,40 @@
+/**
+ *    Copyright 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/repl/replication_consistency_markers.h"
+
+namespace mongo {
+namespace repl {
+
+ReplicationConsistencyMarkers::ReplicationConsistencyMarkers() {}
+ReplicationConsistencyMarkers::~ReplicationConsistencyMarkers() {}
+
+}  // namespace repl
+}  // namespace mongo
diff --git a/src/mongo/db/repl/replication_consistency_markers.h b/src/mongo/db/repl/replication_consistency_markers.h
new file mode 100644
index 00000000000..d57f54960c6
--- /dev/null
+++ b/src/mongo/db/repl/replication_consistency_markers.h
@@ -0,0 +1,164 @@
+/**
+*    Copyright (C) 2017 MongoDB Inc.
+*
+*    This program is free software: you can redistribute it and/or  modify
+*    it under the terms of the GNU Affero General Public License, version 3,
+*    as published by the Free Software Foundation.
+*
+*    This program is distributed in the hope that it will be useful,
+*    but WITHOUT ANY WARRANTY; without even the implied warranty of
+*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+*    GNU Affero General Public License for more details.
+*
+*    You should have received a copy of the GNU Affero General Public License
+*    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*
+*    As a special exception, the copyright holders give permission to link the
+*    code of portions of this program with the OpenSSL library under certain
+*    conditions as described in each individual source file and distribute
+*    linked combinations including the program with the OpenSSL library. You
+*    must comply with the GNU Affero General Public License in all respects for
+*    all of the code used other than as permitted herein. If you modify file(s)
+*    with this exception, you may extend this exception to your version of the
+*    file(s), but you are not obligated to do so. If you do not wish to do so,
+*    delete this exception statement from your version. If you delete this
+*    exception statement from all source files in the program, then also delete
+*    it in the license file.
+*/
+
+#pragma once
+
+#include "mongo/base/disallow_copying.h"
+#include "mongo/db/namespace_string.h"
+
+namespace mongo {
+
+class BSONObj;
+class OperationContext;
+class Timestamp;
+
+namespace repl {
+
+class OpTime;
+class StorageInterface;
+
+/**
+ * This interface provides helper functions for maintaining the documents used for
+ * maintaining data consistency.
+ *
+ * The minValid document, in 'local.replset.minvalid', is used for indicating whether or not the
+ * data on disk is consistent and for getting to a consistent point after unclean shutdown.
+ *
+ * Example of all fields:
+ * {
+ *      _id: <ObjectId>,                    // not used, but auto-generated
+ *      ts: <Timestamp>,
+ *      t: <long long>,                     // timestamp and term of minValid OpTime
+ *      doingInitialSync: <bool>,
+ *      begin: {
+ *                  ts: <Timestamp>,
+ *                  t: <long long>
+ *             },                           // field for 'appliedThrough'
+ *      oplogDeleteFromPoint: <Timestamp>
+ * }
+ *
+ * See below for explanations of each field.
+ */
+class ReplicationConsistencyMarkers {
+    MONGO_DISALLOW_COPYING(ReplicationConsistencyMarkers);
+
+public:
+    // Constructor and Destructor.
+    ReplicationConsistencyMarkers();
+    virtual ~ReplicationConsistencyMarkers();
+
+    // -------- Initial Sync Flag ----------
+
+    /**
+     * Returns true if initial sync was started but has not completed. If we start up and this is
+     * set to true, we know that we must do a resync.
+     */
+    virtual bool getInitialSyncFlag(OperationContext* opCtx) const = 0;
+
+    /**
+     * Sets the initial sync flag to record that initial sync has not completed.
+     *
+     * This operation is durable and waits for durable writes (which will block on
+     * journaling/checkpointing).
+     */
+    virtual void setInitialSyncFlag(OperationContext* opCtx) = 0;
+
+    /**
+     * Clears the initial sync flag to record that initial sync has completed.
+     *
+     * This operation is durable and waits for durable writes (which will block on
+     * journaling/checkpointing).
+     */
+    virtual void clearInitialSyncFlag(OperationContext* opCtx) = 0;
+
+    // -------- MinValid ----------
+
+    /**
+     * The minValid value is the earliest (minimum) OpTime that must be applied in order to
+     * consider the dataset consistent.
+     *   - This is set to the end of a batch before we begin applying a batch of oplog entries
+     *     since the oplog entries can be applied out of order.
+     *   - This is also set during rollback so we do not exit RECOVERING until we are consistent.
+     * If we crash while applying a batch, we apply from appliedThrough to minValid in order
+     * to be consistent. We may re-apply operations, but this is safe.
+     *
+     * Returns the minValid OpTime.
+     */
+    virtual OpTime getMinValid(OperationContext* opCtx) const = 0;
+
+    /**
+     * Sets the minValid OpTime to 'minValid'. This can set minValid backwards, which is necessary
+     * in rollback when the OpTimes in the oplog may move backwards.
+     */
+    virtual void setMinValid(OperationContext* opCtx, const OpTime& minValid) = 0;
+
+    /**
+     * Sets minValid only if it is not already higher than endOpTime.
+     *
+     * Warning, this compares the term and timestamp independently. Do not use if the current
+     * minValid could be from the other fork of a rollback.
+     */
+    virtual void setMinValidToAtLeast(OperationContext* opCtx, const OpTime& minValid) = 0;
+
+    // -------- Oplog Delete From Point ----------
+
+    /**
+     * The oplog delete from point is set to the beginning of a batch of oplog entries before
+     * the oplog entries are written into the oplog, and reset before we begin applying the batch.
+     * On startup all oplog entries with a value >= the oplog delete from point should be deleted.
+     * We write operations to the oplog in parallel so if we crash mid-batch there could be holes
+     * in the oplog. Deleting them at startup keeps us consistent.
+     *
+     * If null, no documents should be deleted.
+     */
+    virtual void setOplogDeleteFromPoint(OperationContext* opCtx, const Timestamp& timestamp) = 0;
+    virtual Timestamp getOplogDeleteFromPoint(OperationContext* opCtx) const = 0;
+
+    // -------- Applied Through ----------
+
+    /**
+     * The applied through point is a persistent record of which oplog entries we've applied.
+     * If we crash while applying a batch of oplog entries, this OpTime tells us where to start
+     * applying operations on startup.
+     *
+     * If null, the applied through point is the top of the oplog.
+     */
+    virtual void setAppliedThrough(OperationContext* opCtx, const OpTime& optime) = 0;
+
+    /**
+     * You should probably be calling ReplicationCoordinator::getLastAppliedOpTime() instead.
+     *
+     * This reads the value from storage which isn't always updated when the ReplicationCoordinator
+     * is. This is safe because it will only ever be stale and reapplying oplog operations is
+     * always safe.
+     */
+    virtual OpTime getAppliedThrough(OperationContext* opCtx) const = 0;
+};
+
+}  // namespace repl
+}  // namespace mongo
diff --git a/src/mongo/db/repl/replication_consistency_markers_impl.cpp b/src/mongo/db/repl/replication_consistency_markers_impl.cpp
new file mode 100644
index 00000000000..ec157379a74
--- /dev/null
+++ b/src/mongo/db/repl/replication_consistency_markers_impl.cpp
@@ -0,0 +1,207 @@
+
+/**
+ *    Copyright (C) 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kReplication
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/repl/replication_consistency_markers_impl.h"
+
+#include "mongo/db/catalog/collection_options.h"
+#include "mongo/db/concurrency/d_concurrency.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/repl/optime.h"
+#include "mongo/db/repl/replication_coordinator.h"
+#include "mongo/db/repl/storage_interface.h"
+#include "mongo/util/log.h"
+
+namespace mongo {
+namespace repl {
+
+constexpr StringData ReplicationConsistencyMarkersImpl::kDefaultMinValidNamespace;
+constexpr StringData ReplicationConsistencyMarkersImpl::kInitialSyncFlagFieldName;
+constexpr StringData ReplicationConsistencyMarkersImpl::kBeginFieldName;
+constexpr StringData ReplicationConsistencyMarkersImpl::kOplogDeleteFromPointFieldName;
+
+namespace {
+const BSONObj kInitialSyncFlag(BSON(ReplicationConsistencyMarkersImpl::kInitialSyncFlagFieldName
+                                    << true));
+}  // namespace
+
+ReplicationConsistencyMarkersImpl::ReplicationConsistencyMarkersImpl(
+    StorageInterface* storageInterface)
+    : ReplicationConsistencyMarkersImpl(
+          storageInterface,
+          NamespaceString(ReplicationConsistencyMarkersImpl::kDefaultMinValidNamespace)) {}
+
+ReplicationConsistencyMarkersImpl::ReplicationConsistencyMarkersImpl(
+    StorageInterface* storageInterface, NamespaceString minValidNss)
+    : _storageInterface(storageInterface), _minValidNss(minValidNss) {}
+
+BSONObj ReplicationConsistencyMarkersImpl::_getMinValidDocument(OperationContext* opCtx) const {
+    auto result = _storageInterface->findSingleton(opCtx, _minValidNss);
+    if (!result.isOK()) {
+        if (result.getStatus() == ErrorCodes::NamespaceNotFound ||
+            result.getStatus() == ErrorCodes::CollectionIsEmpty) {
+            return BSONObj();
+        }
+        // Fail if there is an error other than the collection being missing or being empty.
+        fassertFailedWithStatus(40466, result.getStatus());
+    }
+    return result.getValue();
+}
+
+void ReplicationConsistencyMarkersImpl::_updateMinValidDocument(OperationContext* opCtx,
+                                                                const BSONObj& updateSpec) {
+    Status status = _storageInterface->putSingleton(opCtx, _minValidNss, updateSpec);
+
+    // If the collection doesn't exist, create it and try again.
+    if (status == ErrorCodes::NamespaceNotFound) {
+        status = _storageInterface->createCollection(opCtx, _minValidNss, CollectionOptions());
+        if (status.isOK()) {
+            status = _storageInterface->putSingleton(opCtx, _minValidNss, updateSpec);
+        }
+    }
+
+    fassertStatusOK(40467, status);
+}
+
+bool ReplicationConsistencyMarkersImpl::getInitialSyncFlag(OperationContext* opCtx) const {
+    const BSONObj doc = _getMinValidDocument(opCtx);
+    const auto flag = doc[kInitialSyncFlagFieldName].trueValue();
+    LOG(3) << "returning initial sync flag value of " << flag;
+    return flag;
+}
+
+void ReplicationConsistencyMarkersImpl::setInitialSyncFlag(OperationContext* opCtx) {
+    LOG(3) << "setting initial sync flag";
+    _updateMinValidDocument(opCtx, BSON("$set" << kInitialSyncFlag));
+    opCtx->recoveryUnit()->waitUntilDurable();
+}
+
+void ReplicationConsistencyMarkersImpl::clearInitialSyncFlag(OperationContext* opCtx) {
+    LOG(3) << "clearing initial sync flag";
+
+    auto replCoord = repl::ReplicationCoordinator::get(opCtx);
+    OpTime time = replCoord->getMyLastAppliedOpTime();
+    _updateMinValidDocument(
+        opCtx,
+        BSON("$unset" << kInitialSyncFlag << "$set"
+                      << BSON("ts" << time.getTimestamp() << "t" << time.getTerm()
+                                   << kBeginFieldName
+                                   << time)));
+
+    if (getGlobalServiceContext()->getGlobalStorageEngine()->isDurable()) {
+        opCtx->recoveryUnit()->waitUntilDurable();
+        replCoord->setMyLastDurableOpTime(time);
+    }
+}
+
+OpTime ReplicationConsistencyMarkersImpl::getMinValid(OperationContext* opCtx) const {
+    const BSONObj doc = _getMinValidDocument(opCtx);
+    const auto opTimeStatus = OpTime::parseFromOplogEntry(doc);
+    // If any of the keys (fields) are missing from the minvalid document, we return
+    // a null OpTime.
+    if (opTimeStatus == ErrorCodes::NoSuchKey) {
+        return {};
+    }
+
+    if (!opTimeStatus.isOK()) {
+        severe() << "Error parsing minvalid entry: " << redact(doc)
+                 << ", with status:" << opTimeStatus.getStatus();
+        fassertFailedNoTrace(40052);
+    }
+
+    OpTime minValid = opTimeStatus.getValue();
+    LOG(3) << "returning minvalid: " << minValid.toString() << "(" << minValid.toBSON() << ")";
+
+    return minValid;
+}
+
+void ReplicationConsistencyMarkersImpl::setMinValid(OperationContext* opCtx,
+                                                    const OpTime& minValid) {
+    LOG(3) << "setting minvalid to exactly: " << minValid.toString() << "(" << minValid.toBSON()
+           << ")";
+    _updateMinValidDocument(
+        opCtx, BSON("$set" << BSON("ts" << minValid.getTimestamp() << "t" << minValid.getTerm())));
+}
+
+void ReplicationConsistencyMarkersImpl::setMinValidToAtLeast(OperationContext* opCtx,
+                                                             const OpTime& minValid) {
+    LOG(3) << "setting minvalid to at least: " << minValid.toString() << "(" << minValid.toBSON()
+           << ")";
+    _updateMinValidDocument(
+        opCtx, BSON("$max" << BSON("ts" << minValid.getTimestamp() << "t" << minValid.getTerm())));
+}
+
+void ReplicationConsistencyMarkersImpl::setOplogDeleteFromPoint(OperationContext* opCtx,
+                                                                const Timestamp& timestamp) {
+    LOG(3) << "setting oplog delete from point to: " << timestamp.toStringPretty();
+    _updateMinValidDocument(opCtx,
+                            BSON("$set" << BSON(kOplogDeleteFromPointFieldName << timestamp)));
+}
+
+Timestamp ReplicationConsistencyMarkersImpl::getOplogDeleteFromPoint(
+    OperationContext* opCtx) const {
+    const BSONObj doc = _getMinValidDocument(opCtx);
+    Timestamp out = {};
+    if (auto field = doc[kOplogDeleteFromPointFieldName]) {
+        out = field.timestamp();
+    }
+
+    LOG(3) << "returning oplog delete from point: " << out;
+    return out;
+}
+
+void ReplicationConsistencyMarkersImpl::setAppliedThrough(OperationContext* opCtx,
+                                                          const OpTime& optime) {
+    LOG(3) << "setting appliedThrough to: " << optime.toString() << "(" << optime.toBSON() << ")";
+    if (optime.isNull()) {
+        _updateMinValidDocument(opCtx, BSON("$unset" << BSON(kBeginFieldName << 1)));
+    } else {
+        _updateMinValidDocument(opCtx, BSON("$set" << BSON(kBeginFieldName << optime)));
+    }
+}
+
+OpTime ReplicationConsistencyMarkersImpl::getAppliedThrough(OperationContext* opCtx) const {
+    const BSONObj doc = _getMinValidDocument(opCtx);
+    const auto opTimeStatus = OpTime::parseFromOplogEntry(doc.getObjectField(kBeginFieldName));
+    if (!opTimeStatus.isOK()) {
+        // Return null OpTime on any parse failure, including if "begin" is missing.
+        return {};
+    }
+
+    OpTime appliedThrough = opTimeStatus.getValue();
+    LOG(3) << "returning appliedThrough: " << appliedThrough.toString() << "("
+           << appliedThrough.toBSON() << ")";
+
+    return appliedThrough;
+}
+
+}  // namespace repl
+}  // namespace mongo
diff --git a/src/mongo/db/repl/replication_consistency_markers_impl.h b/src/mongo/db/repl/replication_consistency_markers_impl.h
new file mode 100644
index 00000000000..fd8a6dc9d26
--- /dev/null
+++ b/src/mongo/db/repl/replication_consistency_markers_impl.h
@@ -0,0 +1,93 @@
+/**
+*    Copyright (C) 2017 MongoDB Inc.
+*
+*    This program is free software: you can redistribute it and/or  modify
+*    it under the terms of the GNU Affero General Public License, version 3,
+*    as published by the Free Software Foundation.
+*
+*    This program is distributed in the hope that it will be useful,
+*    but WITHOUT ANY WARRANTY; without even the implied warranty of
+*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+*    GNU Affero General Public License for more details.
+*
+*    You should have received a copy of the GNU Affero General Public License
+*    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*
+*    As a special exception, the copyright holders give permission to link the
+*    code of portions of this program with the OpenSSL library under certain
+*    conditions as described in each individual source file and distribute
+*    linked combinations including the program with the OpenSSL library. You
+*    must comply with the GNU Affero General Public License in all respects for
+*    all of the code used other than as permitted herein. If you modify file(s)
+*    with this exception, you may extend this exception to your version of the
+*    file(s), but you are not obligated to do so. If you do not wish to do so,
+*    delete this exception statement from your version. If you delete this
+*    exception statement from all source files in the program, then also delete
+*    it in the license file.
+*/
+
+#pragma once
+
+#include "mongo/base/disallow_copying.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/repl/replication_consistency_markers.h"
+
+namespace mongo {
+
+class BSONObj;
+class OperationContext;
+class Timestamp;
+
+namespace repl {
+
+class OpTime;
+class StorageInterface;
+
+class ReplicationConsistencyMarkersImpl : public ReplicationConsistencyMarkers {
+    MONGO_DISALLOW_COPYING(ReplicationConsistencyMarkersImpl);
+
+public:
+    static constexpr StringData kDefaultMinValidNamespace = "local.replset.minvalid"_sd;
+    static constexpr StringData kInitialSyncFlagFieldName = "doingInitialSync"_sd;
+    static constexpr StringData kBeginFieldName = "begin"_sd;
+    static constexpr StringData kOplogDeleteFromPointFieldName = "oplogDeleteFromPoint"_sd;
+
+    explicit ReplicationConsistencyMarkersImpl(StorageInterface* storageInterface);
+    ReplicationConsistencyMarkersImpl(StorageInterface* storageInterface,
+                                      NamespaceString minValidNss);
+
+    bool getInitialSyncFlag(OperationContext* opCtx) const override;
+    void setInitialSyncFlag(OperationContext* opCtx) override;
+    void clearInitialSyncFlag(OperationContext* opCtx) override;
+
+    OpTime getMinValid(OperationContext* opCtx) const override;
+    void setMinValid(OperationContext* opCtx, const OpTime& minValid) override;
+    void setMinValidToAtLeast(OperationContext* opCtx, const OpTime& minValid) override;
+
+    void setOplogDeleteFromPoint(OperationContext* opCtx, const Timestamp& timestamp) override;
+    Timestamp getOplogDeleteFromPoint(OperationContext* opCtx) const override;
+
+    void setAppliedThrough(OperationContext* opCtx, const OpTime& optime) override;
+    OpTime getAppliedThrough(OperationContext* opCtx) const override;
+
+private:
+    /**
+     * Reads the MinValid document from disk.
+     * Returns an empty document if not present.
+     */
+    BSONObj _getMinValidDocument(OperationContext* opCtx) const;
+
+    /**
+     * Updates the MinValid document according to the provided update spec. If the collection does
+     * not exist, it is created. If the document does not exist, it is upserted.
+     *
+     * This fasserts on failure.
+     */
+    void _updateMinValidDocument(OperationContext* opCtx, const BSONObj& updateSpec);
+
+    StorageInterface* _storageInterface;
+    const NamespaceString _minValidNss;
+};
+
+}  // namespace repl
+}  // namespace mongo
diff --git a/src/mongo/db/repl/replication_consistency_markers_impl_test.cpp b/src/mongo/db/repl/replication_consistency_markers_impl_test.cpp
new file mode 100644
index 00000000000..4762913ed41
--- /dev/null
+++ b/src/mongo/db/repl/replication_consistency_markers_impl_test.cpp
@@ -0,0 +1,227 @@
+/**
+ *    Copyright 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/repl/replication_consistency_markers_impl.h"
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/client.h"
+#include "mongo/db/concurrency/d_concurrency.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/db_raii.h"
+#include "mongo/db/dbhelpers.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/repl/replication_coordinator.h"
+#include "mongo/db/repl/replication_coordinator_mock.h"
+#include "mongo/db/repl/storage_interface_impl.h"
+#include "mongo/db/service_context_d_test_fixture.h"
+#include "mongo/db/storage/recovery_unit_noop.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/assert_util.h"
+#include "mongo/util/mongoutils/str.h"
+
+namespace {
+
+using namespace mongo;
+using namespace mongo::repl;
+
+/**
+ * Generates a unique namespace from the test registration agent.
+ */
+template <typename T>
+NamespaceString makeNamespace(const T& t, const std::string& suffix = "") {
+    return NamespaceString(std::string("local." + t.getSuiteName() + "_" + t.getTestName())
+                               .substr(0, NamespaceString::MaxNsCollectionLen - suffix.length()) +
+                           suffix);
+}
+
+/**
+ * Returns min valid document.
+ */
+BSONObj getMinValidDocument(OperationContext* opCtx, const NamespaceString& minValidNss) {
+    MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
+        Lock::DBLock dblk(opCtx, minValidNss.db(), MODE_IS);
+        Lock::CollectionLock lk(opCtx->lockState(), minValidNss.ns(), MODE_IS);
+        BSONObj mv;
+        if (Helpers::getSingleton(opCtx, minValidNss.ns().c_str(), mv)) {
+            return mv;
+        }
+    }
+    MONGO_WRITE_CONFLICT_RETRY_LOOP_END(opCtx, "getMinValidDocument", minValidNss.ns());
+    return BSONObj();
+}
+
+class ReplicationConsistencyMarkersTest : public ServiceContextMongoDTest {
+protected:
+    OperationContext* getOperationContext() {
+        return _opCtx.get();
+    }
+
+    StorageInterface* getStorageInterface() {
+        return _storageInterface.get();
+    }
+
+private:
+    void setUp() override {
+        ServiceContextMongoDTest::setUp();
+        _createOpCtx();
+        auto replCoord = stdx::make_unique<ReplicationCoordinatorMock>(getServiceContext());
+        ReplicationCoordinator::set(getServiceContext(), std::move(replCoord));
+        _storageInterface = stdx::make_unique<StorageInterfaceImpl>();
+    }
+
+    void tearDown() override {
+        _opCtx.reset(nullptr);
+        _storageInterface.reset();
+        ServiceContextMongoDTest::tearDown();
+    }
+
+    void _createOpCtx() {
+        _opCtx = cc().makeOperationContext();
+    }
+
+    ServiceContext::UniqueOperationContext _opCtx;
+    std::unique_ptr<StorageInterfaceImpl> _storageInterface;
+};
+
+/**
+ * Recovery unit that tracks if waitUntilDurable() is called.
+ */
+class RecoveryUnitWithDurabilityTracking : public RecoveryUnitNoop {
+public:
+    bool waitUntilDurable() override;
+    bool waitUntilDurableCalled = false;
+};
+
+bool RecoveryUnitWithDurabilityTracking::waitUntilDurable() {
+    waitUntilDurableCalled = true;
+    return RecoveryUnitNoop::waitUntilDurable();
+}
+
+TEST_F(ReplicationConsistencyMarkersTest, InitialSyncFlag) {
+    auto nss = makeNamespace(_agent);
+
+    ReplicationConsistencyMarkersImpl minValid(getStorageInterface(), nss);
+    auto opCtx = getOperationContext();
+
+    // Initial sync flag should be unset after initializing a new storage engine.
+    ASSERT_FALSE(minValid.getInitialSyncFlag(opCtx));
+
+    // Setting initial sync flag should affect getInitialSyncFlag() result.
+    minValid.setInitialSyncFlag(opCtx);
+    ASSERT_TRUE(minValid.getInitialSyncFlag(opCtx));
+
+    // Check min valid document using storage engine interface.
+    auto minValidDocument = getMinValidDocument(opCtx, nss);
+    ASSERT_TRUE(
+        minValidDocument.hasField(ReplicationConsistencyMarkersImpl::kInitialSyncFlagFieldName));
+    ASSERT_TRUE(minValidDocument.getBoolField(
+        ReplicationConsistencyMarkersImpl::kInitialSyncFlagFieldName));
+
+    // Clearing initial sync flag should affect getInitialSyncFlag() result.
+    minValid.clearInitialSyncFlag(opCtx);
+    ASSERT_FALSE(minValid.getInitialSyncFlag(opCtx));
+}
+
+TEST_F(ReplicationConsistencyMarkersTest, GetMinValidAfterSettingInitialSyncFlagWorks) {
+    auto nss = makeNamespace(_agent);
+
+    ReplicationConsistencyMarkersImpl minValid(getStorageInterface(), nss);
+    auto opCtx = getOperationContext();
+
+    // Initial sync flag should be unset after initializing a new storage engine.
+    ASSERT_FALSE(minValid.getInitialSyncFlag(opCtx));
+
+    // Setting initial sync flag should affect getInitialSyncFlag() result.
+    minValid.setInitialSyncFlag(opCtx);
+    ASSERT_TRUE(minValid.getInitialSyncFlag(opCtx));
+
+    ASSERT(minValid.getMinValid(opCtx).isNull());
+    ASSERT(minValid.getAppliedThrough(opCtx).isNull());
+    ASSERT(minValid.getOplogDeleteFromPoint(opCtx).isNull());
+}
+
+TEST_F(ReplicationConsistencyMarkersTest, ReplicationConsistencyMarkers) {
+    auto nss = makeNamespace(_agent);
+
+    ReplicationConsistencyMarkersImpl minValid(getStorageInterface(), nss);
+    auto opCtx = getOperationContext();
+
+    // MinValid boundaries should all be null after initializing a new storage engine.
+    ASSERT(minValid.getMinValid(opCtx).isNull());
+    ASSERT(minValid.getAppliedThrough(opCtx).isNull());
+    ASSERT(minValid.getOplogDeleteFromPoint(opCtx).isNull());
+
+    // Setting min valid boundaries should affect getMinValid() result.
+    OpTime startOpTime({Seconds(123), 0}, 1LL);
+    OpTime endOpTime({Seconds(456), 0}, 1LL);
+    minValid.setAppliedThrough(opCtx, startOpTime);
+    minValid.setMinValid(opCtx, endOpTime);
+    minValid.setOplogDeleteFromPoint(opCtx, endOpTime.getTimestamp());
+
+    ASSERT_EQ(minValid.getAppliedThrough(opCtx), startOpTime);
+    ASSERT_EQ(minValid.getMinValid(opCtx), endOpTime);
+    ASSERT_EQ(minValid.getOplogDeleteFromPoint(opCtx), endOpTime.getTimestamp());
+
+    // setMinValid always changes minValid, but setMinValidToAtLeast only does if higher.
+    minValid.setMinValid(opCtx, startOpTime);  // Forcibly lower it.
+    ASSERT_EQ(minValid.getMinValid(opCtx), startOpTime);
+    minValid.setMinValidToAtLeast(opCtx, endOpTime);  // Higher than current (sets it).
+    ASSERT_EQ(minValid.getMinValid(opCtx), endOpTime);
+    minValid.setMinValidToAtLeast(opCtx, startOpTime);  // Lower than current (no-op).
+    ASSERT_EQ(minValid.getMinValid(opCtx), endOpTime);
+
+    // Check min valid document using storage engine interface.
+    auto minValidDocument = getMinValidDocument(opCtx, nss);
+    ASSERT_TRUE(minValidDocument.hasField(ReplicationConsistencyMarkersImpl::kBeginFieldName));
+    ASSERT_TRUE(minValidDocument[ReplicationConsistencyMarkersImpl::kBeginFieldName].isABSONObj());
+    ASSERT_EQUALS(startOpTime,
+                  unittest::assertGet(OpTime::parseFromOplogEntry(
+                      minValidDocument[ReplicationConsistencyMarkersImpl::kBeginFieldName].Obj())));
+    ASSERT_EQUALS(endOpTime, unittest::assertGet(OpTime::parseFromOplogEntry(minValidDocument)));
+    ASSERT_EQUALS(
+        endOpTime.getTimestamp(),
+        minValidDocument[ReplicationConsistencyMarkersImpl::kOplogDeleteFromPointFieldName]
+            .timestamp());
+
+    // Recovery unit will be owned by "opCtx".
+    RecoveryUnitWithDurabilityTracking* recoveryUnit = new RecoveryUnitWithDurabilityTracking();
+    opCtx->setRecoveryUnit(recoveryUnit, OperationContext::kNotInUnitOfWork);
+
+    // Set min valid without waiting for the changes to be durable.
+    OpTime endOpTime2({Seconds(789), 0}, 1LL);
+    minValid.setMinValid(opCtx, endOpTime2);
+    minValid.setAppliedThrough(opCtx, {});
+    ASSERT_EQUALS(minValid.getAppliedThrough(opCtx), OpTime());
+    ASSERT_EQUALS(minValid.getMinValid(opCtx), endOpTime2);
+    ASSERT_FALSE(recoveryUnit->waitUntilDurableCalled);
+}
+
+}  // namespace
diff --git a/src/mongo/db/repl/replication_consistency_markers_mock.cpp b/src/mongo/db/repl/replication_consistency_markers_mock.cpp
new file mode 100644
index 00000000000..5fdd402d33b
--- /dev/null
+++ b/src/mongo/db/repl/replication_consistency_markers_mock.cpp
@@ -0,0 +1,92 @@
+/**
+ *    Copyright (C) 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/repl/replication_consistency_markers_mock.h"
+
+namespace mongo {
+namespace repl {
+
+bool ReplicationConsistencyMarkersMock::getInitialSyncFlag(OperationContext* opCtx) const {
+    stdx::lock_guard<stdx::mutex> lock(_initialSyncFlagMutex);
+    return _initialSyncFlag;
+}
+
+void ReplicationConsistencyMarkersMock::setInitialSyncFlag(OperationContext* opCtx) {
+    stdx::lock_guard<stdx::mutex> lock(_initialSyncFlagMutex);
+    _initialSyncFlag = true;
+}
+
+void ReplicationConsistencyMarkersMock::clearInitialSyncFlag(OperationContext* opCtx) {
+    stdx::lock_guard<stdx::mutex> lock(_initialSyncFlagMutex);
+    _initialSyncFlag = false;
+}
+
+OpTime ReplicationConsistencyMarkersMock::getMinValid(OperationContext* opCtx) const {
+    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
+    return _minValid;
+}
+
+void ReplicationConsistencyMarkersMock::setMinValid(OperationContext* opCtx,
+                                                    const OpTime& minValid) {
+    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
+    _minValid = minValid;
+}
+
+void ReplicationConsistencyMarkersMock::setMinValidToAtLeast(OperationContext* opCtx,
+                                                             const OpTime& minValid) {
+    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
+    _minValid = std::max(_minValid, minValid);
+}
+
+void ReplicationConsistencyMarkersMock::setOplogDeleteFromPoint(OperationContext* opCtx,
+                                                                const Timestamp& timestamp) {
+    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
+    _oplogDeleteFromPoint = timestamp;
+}
+
+Timestamp ReplicationConsistencyMarkersMock::getOplogDeleteFromPoint(
+    OperationContext* opCtx) const {
+    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
+    return _oplogDeleteFromPoint;
+}
+
+void ReplicationConsistencyMarkersMock::setAppliedThrough(OperationContext* opCtx,
+                                                          const OpTime& optime) {
+    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
+    _appliedThrough = optime;
+}
+
+OpTime ReplicationConsistencyMarkersMock::getAppliedThrough(OperationContext* opCtx) const {
+    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
+    return _appliedThrough;
+}
+
+}  // namespace repl
+}  // namespace mongo
diff --git a/src/mongo/db/repl/replication_consistency_markers_mock.h b/src/mongo/db/repl/replication_consistency_markers_mock.h
new file mode 100644
index 00000000000..82269f6a60b
--- /dev/null
+++ b/src/mongo/db/repl/replication_consistency_markers_mock.h
@@ -0,0 +1,78 @@
+/**
+*    Copyright (C) 2017 MongoDB Inc.
+*
+*    This program is free software: you can redistribute it and/or  modify
+*    it under the terms of the GNU Affero General Public License, version 3,
+*    as published by the Free Software Foundation.
+*
+*    This program is distributed in the hope that it will be useful,
+*    but WITHOUT ANY WARRANTY; without even the implied warranty of
+*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+*    GNU Affero General Public License for more details.
+*
+*    You should have received a copy of the GNU Affero General Public License
+*    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*
+*    As a special exception, the copyright holders give permission to link the
+*    code of portions of this program with the OpenSSL library under certain
+*    conditions as described in each individual source file and distribute
+*    linked combinations including the program with the OpenSSL library. You
+*    must comply with the GNU Affero General Public License in all respects for
+*    all of the code used other than as permitted herein. If you modify file(s)
+*    with this exception, you may extend this exception to your version of the
+*    file(s), but you are not obligated to do so. If you do not wish to do so,
+*    delete this exception statement from your version. If you delete this
+*    exception statement from all source files in the program, then also delete
+*    it in the license file.
+*/
+
+#pragma once
+
+#include "mongo/base/disallow_copying.h"
+#include "mongo/db/repl/optime.h"
+#include "mongo/db/repl/replication_consistency_markers.h"
+#include "mongo/stdx/mutex.h"
+
+namespace mongo {
+
+class BSONObj;
+class OperationContext;
+class Timestamp;
+
+namespace repl {
+
+/**
+ * A mock ReplicationConsistencyMarkers implementation that stores everything in memory.
+ */
+class ReplicationConsistencyMarkersMock : public ReplicationConsistencyMarkers {
+    MONGO_DISALLOW_COPYING(ReplicationConsistencyMarkersMock);
+
+public:
+    ReplicationConsistencyMarkersMock() = default;
+
+    bool getInitialSyncFlag(OperationContext* opCtx) const override;
+    void setInitialSyncFlag(OperationContext* opCtx) override;
+    void clearInitialSyncFlag(OperationContext* opCtx) override;
+
+    OpTime getMinValid(OperationContext* opCtx) const override;
+    void setMinValid(OperationContext* opCtx, const OpTime& minValid) override;
+    void setMinValidToAtLeast(OperationContext* opCtx, const OpTime& minValid) override;
+
+    void setOplogDeleteFromPoint(OperationContext* opCtx, const Timestamp& timestamp) override;
+    Timestamp getOplogDeleteFromPoint(OperationContext* opCtx) const override;
+
+    void setAppliedThrough(OperationContext* opCtx, const OpTime& optime) override;
+    OpTime getAppliedThrough(OperationContext* opCtx) const override;
+
+private:
+    mutable stdx::mutex _initialSyncFlagMutex;
+    bool _initialSyncFlag = false;
+
+    mutable stdx::mutex _minValidBoundariesMutex;
+    OpTime _appliedThrough;
+    OpTime _minValid;
+    Timestamp _oplogDeleteFromPoint;
+};
+
+}  // namespace repl
+}  // namespace mongo
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 a70df25f4e0..961926a1f35 100644
--- a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
@@ -64,6 +64,7 @@
 #include "mongo/db/repl/oplog_buffer_proxy.h"
 #include "mongo/db/repl/repl_settings.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
+#include "mongo/db/repl/replication_process.h"
 #include "mongo/db/repl/rs_sync.h"
 #include "mongo/db/repl/snapshot_thread.h"
 #include "mongo/db/repl/storage_interface.h"
@@ -203,16 +204,18 @@ void scheduleWork(executor::TaskExecutor* executor,
 ReplicationCoordinatorExternalStateImpl::ReplicationCoordinatorExternalStateImpl(
     ServiceContext* service,
     DropPendingCollectionReaper* dropPendingCollectionReaper,
-    StorageInterface* storageInterface)
+    StorageInterface* storageInterface,
+    ReplicationProcess* replicationProcess)
     : _service(service),
       _dropPendingCollectionReaper(dropPendingCollectionReaper),
-      _storageInterface(storageInterface) {
+      _storageInterface(storageInterface),
+      _replicationProcess(replicationProcess) {
     uassert(ErrorCodes::BadValue, "A StorageInterface is required.", _storageInterface);
 }
 ReplicationCoordinatorExternalStateImpl::~ReplicationCoordinatorExternalStateImpl() {}
 
 bool ReplicationCoordinatorExternalStateImpl::isInitialSyncFlagSet(OperationContext* opCtx) {
-    return _storageInterface->getInitialSyncFlag(opCtx);
+    return _replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx);
 }
 
 void ReplicationCoordinatorExternalStateImpl::startSteadyStateReplication(
@@ -222,7 +225,8 @@ void ReplicationCoordinatorExternalStateImpl::startSteadyStateReplication(
     invariant(replCoord);
     invariant(!_bgSync);
     log() << "Starting replication fetcher thread";
-    _bgSync = stdx::make_unique<BackgroundSync>(this, makeSteadyStateOplogBuffer(opCtx));
+    _bgSync = stdx::make_unique<BackgroundSync>(
+        this, _replicationProcess, makeSteadyStateOplogBuffer(opCtx));
     _bgSync->startup(opCtx);
 
     log() << "Starting replication applier thread";
@@ -332,11 +336,12 @@ void ReplicationCoordinatorExternalStateImpl::shutdown(OperationContext* opCtx)
 
     // Perform additional shutdown steps below that must be done outside _threadMutex.
 
-    if (_storageInterface->getOplogDeleteFromPoint(opCtx).isNull() &&
-        loadLastOpTime(opCtx) == _storageInterface->getAppliedThrough(opCtx)) {
+    if (_replicationProcess->getConsistencyMarkers()->getOplogDeleteFromPoint(opCtx).isNull() &&
+        loadLastOpTime(opCtx) ==
+            _replicationProcess->getConsistencyMarkers()->getAppliedThrough(opCtx)) {
         // Clear the appliedThrough marker to indicate we are consistent with the top of the
         // oplog.
-        _storageInterface->setAppliedThrough(opCtx, {});
+        _replicationProcess->getConsistencyMarkers()->setAppliedThrough(opCtx, {});
     }
 }
 
@@ -385,11 +390,6 @@ Status ReplicationCoordinatorExternalStateImpl::initializeReplSetStorage(Operati
         }
         MONGO_WRITE_CONFLICT_RETRY_LOOP_END(opCtx, "initiate oplog entry", "local.oplog.rs");
 
-        // This initializes the minvalid document with a null "ts" because older versions (<=3.2)
-        // get angry if the minValid document is present but doesn't have a "ts" field.
-        // Consider removing this once we no longer need to support downgrading to 3.2.
-        _storageInterface->setMinValidToAtLeast(opCtx, {});
-
         FeatureCompatibilityVersion::setIfCleanStartup(opCtx, _storageInterface);
     } catch (const DBException& ex) {
         return ex.toStatus();
@@ -414,8 +414,9 @@ OpTime ReplicationCoordinatorExternalStateImpl::onTransitionToPrimary(OperationC
 
     // Clear the appliedThrough marker so on startup we'll use the top of the oplog. This must be
     // done before we add anything to our oplog.
-    invariant(_storageInterface->getOplogDeleteFromPoint(opCtx).isNull());
-    _storageInterface->setAppliedThrough(opCtx, {});
+    invariant(
+        _replicationProcess->getConsistencyMarkers()->getOplogDeleteFromPoint(opCtx).isNull());
+    _replicationProcess->getConsistencyMarkers()->setAppliedThrough(opCtx, {});
 
     if (isV1ElectionProtocol) {
         MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
@@ -564,17 +565,14 @@ void ReplicationCoordinatorExternalStateImpl::setGlobalTimestamp(ServiceContext*
 }
 
 void ReplicationCoordinatorExternalStateImpl::cleanUpLastApplyBatch(OperationContext* opCtx) {
-    if (_storageInterface->getInitialSyncFlag(opCtx)) {
+    if (_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx)) {
         return;  // Initial Sync will take over so no cleanup is needed.
     }
 
-    // This initializes the minvalid document with a null "ts" because older versions (<=3.2)
-    // get angry if the minValid document is present but doesn't have a "ts" field.
-    // Consider removing this once we no longer need to support downgrading to 3.2.
-    _storageInterface->setMinValidToAtLeast(opCtx, {});
-
-    const auto deleteFromPoint = _storageInterface->getOplogDeleteFromPoint(opCtx);
-    const auto appliedThrough = _storageInterface->getAppliedThrough(opCtx);
+    const auto deleteFromPoint =
+        _replicationProcess->getConsistencyMarkers()->getOplogDeleteFromPoint(opCtx);
+    const auto appliedThrough =
+        _replicationProcess->getConsistencyMarkers()->getAppliedThrough(opCtx);
 
     const bool needToDeleteEndOfOplog = !deleteFromPoint.isNull() &&
         // This version should never have a non-null deleteFromPoint with a null appliedThrough.
@@ -591,7 +589,8 @@ void ReplicationCoordinatorExternalStateImpl::cleanUpLastApplyBatch(OperationCon
         log() << "Removing unapplied entries starting at: " << deleteFromPoint;
         truncateOplogTo(opCtx, deleteFromPoint);
     }
-    _storageInterface->setOplogDeleteFromPoint(opCtx, {});  // clear the deleteFromPoint
+    _replicationProcess->getConsistencyMarkers()->setOplogDeleteFromPoint(
+        opCtx, {});  // clear the deleteFromPoint
 
     if (appliedThrough.isNull()) {
         // No follow-up work to do.
@@ -643,7 +642,7 @@ void ReplicationCoordinatorExternalStateImpl::cleanUpLastApplyBatch(OperationCon
     while (cursor->more()) {
         auto entry = cursor->nextSafe();
         fassertStatusOK(40294, SyncTail::syncApply(opCtx, entry, true));
-        _storageInterface->setAppliedThrough(
+        _replicationProcess->getConsistencyMarkers()->setAppliedThrough(
             opCtx, fassertStatusOK(40295, OpTime::parseFromOplogEntry(entry)));
     }
 }
@@ -653,7 +652,7 @@ StatusWith<OpTime> ReplicationCoordinatorExternalStateImpl::loadLastOpTime(
     // TODO: handle WriteConflictExceptions below
     try {
         // If we are doing an initial sync do not read from the oplog.
-        if (_storageInterface->getInitialSyncFlag(opCtx)) {
+        if (_replicationProcess->getConsistencyMarkers()->getInitialSyncFlag(opCtx)) {
             return {ErrorCodes::InitialSyncFailure, "In the middle of an initial sync."};
         }
 
diff --git a/src/mongo/db/repl/replication_coordinator_external_state_impl.h b/src/mongo/db/repl/replication_coordinator_external_state_impl.h
index b6b4b535a7d..68318f01455 100644
--- a/src/mongo/db/repl/replication_coordinator_external_state_impl.h
+++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.h
@@ -52,6 +52,7 @@ using UniqueLock = stdx::unique_lock<stdx::mutex>;
 }  // namespace
 
 class DropPendingCollectionReaper;
+class ReplicationProcess;
 class SnapshotThread;
 class StorageInterface;
 class NoopWriter;
@@ -64,7 +65,8 @@ public:
     ReplicationCoordinatorExternalStateImpl(
         ServiceContext* service,
         DropPendingCollectionReaper* dropPendingCollectionReaper,
-        StorageInterface* storageInterface);
+        StorageInterface* storageInterface,
+        ReplicationProcess* replicationProcess);
     virtual ~ReplicationCoordinatorExternalStateImpl();
     virtual void startThreads(const ReplSettings& settings) override;
     virtual void startSteadyStateReplication(OperationContext* opCtx,
@@ -163,6 +165,9 @@ private:
     DropPendingCollectionReaper* _dropPendingCollectionReaper;
 
     StorageInterface* _storageInterface;
+
+    ReplicationProcess* _replicationProcess;
+
     // True when the threads have been started
     bool _startedThreads = false;
 
diff --git a/src/mongo/db/repl/replication_coordinator_impl.cpp b/src/mongo/db/repl/replication_coordinator_impl.cpp
index 2b738bc824c..222d8b1c507 100644
--- a/src/mongo/db/repl/replication_coordinator_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_impl.cpp
@@ -630,6 +630,7 @@ void ReplicationCoordinatorImpl::_startDataReplication(OperationContext* opCtx,
                 stdx::make_unique<DataReplicatorExternalStateInitialSync>(this,
                                                                           _externalState.get()),
                 _storage,
+                _replicationProcess,
                 onCompletion);
             _initialSyncer = initialSyncerCopy;
         }
diff --git a/src/mongo/db/repl/replication_coordinator_test_fixture.cpp b/src/mongo/db/repl/replication_coordinator_test_fixture.cpp
index 291d6a7652f..b1a93ec2b24 100644
--- a/src/mongo/db/repl/replication_coordinator_test_fixture.cpp
+++ b/src/mongo/db/repl/replication_coordinator_test_fixture.cpp
@@ -38,6 +38,7 @@
 #include "mongo/db/repl/repl_set_heartbeat_args.h"
 #include "mongo/db/repl/repl_set_heartbeat_args_v1.h"
 #include "mongo/db/repl/repl_settings.h"
+#include "mongo/db/repl/replication_consistency_markers_mock.h"
 #include "mongo/db/repl/replication_coordinator_external_state_mock.h"
 #include "mongo/db/repl/replication_coordinator_impl.h"
 #include "mongo/db/repl/replication_process.h"
@@ -124,7 +125,10 @@ void ReplCoordTest::init() {
     StorageInterface::set(service, std::unique_ptr<StorageInterface>(storageInterface));
     ASSERT_TRUE(storageInterface == StorageInterface::get(service));
 
-    ReplicationProcess::set(service, stdx::make_unique<ReplicationProcess>(storageInterface));
+    ReplicationProcess::set(
+        service,
+        stdx::make_unique<ReplicationProcess>(
+            storageInterface, stdx::make_unique<ReplicationConsistencyMarkersMock>()));
     auto replicationProcess = ReplicationProcess::get(service);
 
     // PRNG seed for tests.
diff --git a/src/mongo/db/repl/replication_process.cpp b/src/mongo/db/repl/replication_process.cpp
index 62eba5e679a..7e7dc077b11 100644
--- a/src/mongo/db/repl/replication_process.cpp
+++ b/src/mongo/db/repl/replication_process.cpp
@@ -33,6 +33,7 @@
 #include "mongo/db/client.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/operation_context.h"
+#include "mongo/db/repl/replication_consistency_markers.h"
 #include "mongo/db/repl/replication_process.h"
 #include "mongo/db/repl/rollback_gen.h"
 #include "mongo/db/repl/storage_interface.h"
@@ -74,13 +75,17 @@ ReplicationProcess* ReplicationProcess::get(OperationContext* opCtx) {
 }
 
 
-void ReplicationProcess::set(ServiceContext* service, std::unique_ptr<ReplicationProcess> storage) {
-    auto& storageInterface = getReplicationProcess(service);
-    storageInterface = std::move(storage);
+void ReplicationProcess::set(ServiceContext* service, std::unique_ptr<ReplicationProcess> process) {
+    auto& replicationProcess = getReplicationProcess(service);
+    replicationProcess = std::move(process);
 }
 
-ReplicationProcess::ReplicationProcess(StorageInterface* storageInterface)
-    : _storageInterface(storageInterface), _rbid(kUninitializedRollbackId) {}
+ReplicationProcess::ReplicationProcess(
+    StorageInterface* storageInterface,
+    std::unique_ptr<ReplicationConsistencyMarkers> consistencyMarkers)
+    : _storageInterface(storageInterface),
+      _consistencyMarkers(std::move(consistencyMarkers)),
+      _rbid(kUninitializedRollbackId) {}
 
 StatusWith<int> ReplicationProcess::getRollbackID(OperationContext* opCtx) {
     stdx::lock_guard<stdx::mutex> lock(_mutex);
@@ -173,5 +178,9 @@ Status ReplicationProcess::clearRollbackProgress(OperationContext* opCtx) {
     return Status::OK();
 }
 
+ReplicationConsistencyMarkers* ReplicationProcess::getConsistencyMarkers() {
+    return _consistencyMarkers.get();
+}
+
 }  // namespace repl
 }  // namespace mongo
diff --git a/src/mongo/db/repl/replication_process.h b/src/mongo/db/repl/replication_process.h
index 7e6a81f362c..da4a2886c6a 100644
--- a/src/mongo/db/repl/replication_process.h
+++ b/src/mongo/db/repl/replication_process.h
@@ -36,6 +36,7 @@
 #include "mongo/base/status_with.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/repl/optime.h"
+#include "mongo/db/repl/replication_consistency_markers.h"
 #include "mongo/stdx/mutex.h"
 
 namespace mongo {
@@ -77,10 +78,10 @@ public:
     static ReplicationProcess* get(ServiceContext* service);
     static ReplicationProcess* get(ServiceContext& service);
     static ReplicationProcess* get(OperationContext* opCtx);
-    static void set(ServiceContext* service, std::unique_ptr<ReplicationProcess> storageInterface);
+    static void set(ServiceContext* service, std::unique_ptr<ReplicationProcess> process);
 
-    // Constructor and Destructor.
-    explicit ReplicationProcess(StorageInterface* storageInterface);
+    ReplicationProcess(StorageInterface* storageInterface,
+                       std::unique_ptr<ReplicationConsistencyMarkers> consistencyMarkers);
     virtual ~ReplicationProcess() = default;
 
     /**
@@ -128,6 +129,11 @@ public:
      */
     Status clearRollbackProgress(OperationContext* opCtx);
 
+    /**
+     * Returns an object used for operating on the documents that maintain replication consistency.
+     */
+    ReplicationConsistencyMarkers* getConsistencyMarkers();
+
 private:
     // All member variables are labeled with one of the following codes indicating the
     // synchronization rules for accessing them.
@@ -142,6 +148,9 @@ private:
     // Used to access the storage layer.
     StorageInterface* const _storageInterface;  // (R)
 
+    // Used for operations on documents that maintain replication consistency.
+    std::unique_ptr<ReplicationConsistencyMarkers> _consistencyMarkers;  // (S)
+
     // Rollback ID. This is a cached copy of the persisted value in the local.system.rollback.id
     // collection.
     int _rbid;  // (M)
diff --git a/src/mongo/db/repl/replication_process_test.cpp b/src/mongo/db/repl/replication_process_test.cpp
index 4e42ad6e085..7228913507c 100644
--- a/src/mongo/db/repl/replication_process_test.cpp
+++ b/src/mongo/db/repl/replication_process_test.cpp
@@ -33,6 +33,7 @@
 #include "mongo/db/client.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/repl/optime.h"
+#include "mongo/db/repl/replication_consistency_markers_impl.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/repl/replication_process.h"
@@ -75,7 +76,9 @@ ServiceContext::UniqueOperationContext makeOpCtx() {
 TEST_F(ReplicationProcessTest, ServiceContextDecorator) {
     auto serviceContext = getServiceContext();
     ASSERT_FALSE(ReplicationProcess::get(serviceContext));
-    ReplicationProcess* replicationProcess = new ReplicationProcess(_storageInterface.get());
+    ReplicationProcess* replicationProcess = new ReplicationProcess(
+        _storageInterface.get(),
+        stdx::make_unique<ReplicationConsistencyMarkersImpl>(_storageInterface.get()));
     ReplicationProcess::set(serviceContext,
                             std::unique_ptr<ReplicationProcess>(replicationProcess));
     ASSERT_TRUE(replicationProcess == ReplicationProcess::get(serviceContext));
@@ -85,7 +88,9 @@ TEST_F(ReplicationProcessTest, ServiceContextDecorator) {
 
 TEST_F(ReplicationProcessTest,
        GetRollbackProgressReturnsNoSuchKeyIfDocumentWithIdProgressIsNotFound) {
-    ReplicationProcess replicationProcess(_storageInterface.get());
+    ReplicationProcess replicationProcess(
+        _storageInterface.get(),
+        stdx::make_unique<ReplicationConsistencyMarkersImpl>(_storageInterface.get()));
 
     // Collection is not found.
     auto opCtx = makeOpCtx();
@@ -119,7 +124,9 @@ TEST_F(ReplicationProcessTest, GetRollbackProgressReturnsBadStatusIfApplyUntilFi
     ASSERT_OK(_storageInterface->insertDocument(
         opCtx.get(), ReplicationProcess::kRollbackProgressNamespace, doc));
 
-    ReplicationProcess replicationProcess(_storageInterface.get());
+    ReplicationProcess replicationProcess(
+        _storageInterface.get(),
+        stdx::make_unique<ReplicationConsistencyMarkersImpl>(_storageInterface.get()));
     ASSERT_EQUALS(mongo::AssertionException::convertExceptionCode(40410),
                   replicationProcess.getRollbackProgress(opCtx.get()));
 }
@@ -139,7 +146,9 @@ TEST_F(ReplicationProcessTest,
     ASSERT_OK(_storageInterface->insertDocument(
         opCtx.get(), ReplicationProcess::kRollbackProgressNamespace, doc));
 
-    ReplicationProcess replicationProcess(_storageInterface.get());
+    ReplicationProcess replicationProcess(
+        _storageInterface.get(),
+        stdx::make_unique<ReplicationConsistencyMarkersImpl>(_storageInterface.get()));
     ASSERT_EQUALS(ErrorCodes::TypeMismatch, replicationProcess.getRollbackProgress(opCtx.get()));
 }
 
@@ -156,7 +165,9 @@ TEST_F(ReplicationProcessTest,
     ASSERT_OK(_storageInterface->insertDocument(
         opCtx.get(), ReplicationProcess::kRollbackProgressNamespace, doc));
 
-    ReplicationProcess replicationProcess(_storageInterface.get());
+    ReplicationProcess replicationProcess(
+        _storageInterface.get(),
+        stdx::make_unique<ReplicationConsistencyMarkersImpl>(_storageInterface.get()));
     ASSERT_EQUALS(applyUntil,
                   unittest::assertGet(replicationProcess.getRollbackProgress(opCtx.get())));
 
@@ -169,7 +180,9 @@ TEST_F(ReplicationProcessTest,
        SetRollbackProgressCreatesCollectionBeforeInsertingDocumentIfCollectionDoesNotExist) {
     OpTime applyUntil({Seconds(123), 0}, 1LL);
     auto opCtx = makeOpCtx();
-    ReplicationProcess replicationProcess(_storageInterface.get());
+    ReplicationProcess replicationProcess(
+        _storageInterface.get(),
+        stdx::make_unique<ReplicationConsistencyMarkersImpl>(_storageInterface.get()));
     ASSERT_OK(replicationProcess.setRollbackProgress(opCtx.get(), applyUntil));
     ASSERT_EQUALS(1U,
                   unittest::assertGet(_storageInterface->getCollectionCount(
@@ -185,14 +198,18 @@ TEST_F(ReplicationProcessTest,
 
     OpTime applyUntil({Seconds(123), 0}, 1LL);
     auto opCtx = makeOpCtx();
-    ReplicationProcess replicationProcess(_storageInterface.get());
+    ReplicationProcess replicationProcess(
+        _storageInterface.get(),
+        stdx::make_unique<ReplicationConsistencyMarkersImpl>(_storageInterface.get()));
     ASSERT_EQUALS(ErrorCodes::IllegalOperation,
                   replicationProcess.setRollbackProgress(opCtx.get(), applyUntil));
 }
 
 TEST_F(ReplicationProcessTest, ClearRollbackProgressReturnsSuccessIfCollectionDoesNotExist) {
     auto opCtx = makeOpCtx();
-    ReplicationProcess replicationProcess(_storageInterface.get());
+    ReplicationProcess replicationProcess(
+        _storageInterface.get(),
+        stdx::make_unique<ReplicationConsistencyMarkersImpl>(_storageInterface.get()));
     ASSERT_OK(replicationProcess.clearRollbackProgress(opCtx.get()));
 }
 
@@ -202,7 +219,9 @@ TEST_F(ReplicationProcessTest,
     _storageInterface = stdx::make_unique<StorageInterfaceMock>();
 
     auto opCtx = makeOpCtx();
-    ReplicationProcess replicationProcess(_storageInterface.get());
+    ReplicationProcess replicationProcess(
+        _storageInterface.get(),
+        stdx::make_unique<ReplicationConsistencyMarkersImpl>(_storageInterface.get()));
     ASSERT_EQUALS(ErrorCodes::IllegalOperation,
                   replicationProcess.clearRollbackProgress(opCtx.get()));
 }
diff --git a/src/mongo/db/repl/rollback_test_fixture.cpp b/src/mongo/db/repl/rollback_test_fixture.cpp
index 23866621c80..791cbf0e2e7 100644
--- a/src/mongo/db/repl/rollback_test_fixture.cpp
+++ b/src/mongo/db/repl/rollback_test_fixture.cpp
@@ -34,6 +34,7 @@
 
 #include "mongo/db/client.h"
 #include "mongo/db/repl/oplog.h"
+#include "mongo/db/repl/replication_consistency_markers_mock.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/repl/replication_process.h"
@@ -70,17 +71,18 @@ RollbackTest::RollbackTest() : _threadPoolExecutorTest(createThreadPoolOptions()
 void RollbackTest::setUp() {
     _serviceContextMongoDTest.setUp();
     _threadPoolExecutorTest.setUp();
-    _opCtx = cc().makeOperationContext();
     auto serviceContext = _serviceContextMongoDTest.getServiceContext();
-    ReplicationProcess::set(serviceContext,
-                            stdx::make_unique<ReplicationProcess>(&_storageInterface));
+    _replicationProcess = stdx::make_unique<ReplicationProcess>(
+        &_storageInterface, stdx::make_unique<ReplicationConsistencyMarkersMock>());
     _coordinator = new ReplicationCoordinatorRollbackMock(serviceContext);
     ReplicationCoordinator::set(serviceContext,
                                 std::unique_ptr<ReplicationCoordinator>(_coordinator));
     setOplogCollectionName();
-    _storageInterface.setAppliedThrough(_opCtx.get(), OpTime{});
-    _storageInterface.setMinValid(_opCtx.get(), OpTime{});
-    ReplicationProcess::get(_opCtx.get())->initializeRollbackID(_opCtx.get());
+
+    _opCtx = cc().makeOperationContext();
+    _replicationProcess->getConsistencyMarkers()->setAppliedThrough(_opCtx.get(), OpTime{});
+    _replicationProcess->getConsistencyMarkers()->setMinValid(_opCtx.get(), OpTime{});
+    _replicationProcess->initializeRollbackID(_opCtx.get());
 
     _threadPoolExecutorTest.launchExecutorThread();
 }
@@ -98,6 +100,7 @@ void RollbackTest::tearDown() {
     // ServiceContextMongoD::tearDown() does not destroy service context so it is okay
     // to access the service context after tearDown().
     auto serviceContext = _serviceContextMongoDTest.getServiceContext();
+    _replicationProcess.reset();
     ReplicationCoordinator::set(serviceContext, {});
 }
 
diff --git a/src/mongo/db/repl/rollback_test_fixture.h b/src/mongo/db/repl/rollback_test_fixture.h
index ece8cd682d2..d3e959ddc2f 100644
--- a/src/mongo/db/repl/rollback_test_fixture.h
+++ b/src/mongo/db/repl/rollback_test_fixture.h
@@ -29,6 +29,7 @@
 #pragma once
 
 #include "mongo/db/repl/replication_coordinator_mock.h"
+#include "mongo/db/repl/replication_process.h"
 #include "mongo/db/repl/storage_interface_mock.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/service_context_d_test_fixture.h"
@@ -86,8 +87,10 @@ protected:
     class ReplicationCoordinatorRollbackMock;
     ReplicationCoordinatorRollbackMock* _coordinator = nullptr;
 
-    // StorageInterface used to access minValid.
     StorageInterfaceMock _storageInterface;
+
+    // ReplicationProcess used to access consistency markers.
+    std::unique_ptr<ReplicationProcess> _replicationProcess;
 };
 
 /**
diff --git a/src/mongo/db/repl/rs_rollback.cpp b/src/mongo/db/repl/rs_rollback.cpp
index 85cdd0a4141..4dfd9dd5c68 100644
--- a/src/mongo/db/repl/rs_rollback.cpp
+++ b/src/mongo/db/repl/rs_rollback.cpp
@@ -63,7 +63,6 @@
 #include "mongo/db/repl/roll_back_local_operations.h"
 #include "mongo/db/repl/rollback_source.h"
 #include "mongo/db/repl/rslog.h"
-#include "mongo/db/repl/storage_interface.h"
 #include "mongo/db/s/shard_identity_rollback_notifier.h"
 #include "mongo/util/exit.h"
 #include "mongo/util/fail_point_service.h"
@@ -340,7 +339,7 @@ namespace {
 void checkRbidAndUpdateMinValid(OperationContext* opCtx,
                                 const int rbid,
                                 const RollbackSource& rollbackSource,
-                                StorageInterface* storageInterface) {
+                                ReplicationProcess* replicationProcess) {
     // It is important that the steps are performed in order to avoid racing with upstream rollbacks
     //
     // 1) Get the last doc in their oplog.
@@ -360,8 +359,8 @@ void checkRbidAndUpdateMinValid(OperationContext* opCtx,
     // online until we get to that point in freshness.
     OpTime minValid = fassertStatusOK(28774, OpTime::parseFromOplogEntry(newMinValidDoc));
     log() << "Setting minvalid to " << minValid;
-    storageInterface->setAppliedThrough(opCtx, {});  // Use top of oplog.
-    storageInterface->setMinValid(opCtx, minValid);
+    replicationProcess->getConsistencyMarkers()->setAppliedThrough(opCtx, {});  // Use top of oplog.
+    replicationProcess->getConsistencyMarkers()->setMinValid(opCtx, minValid);
 
     if (MONGO_FAIL_POINT(rollbackHangThenFailAfterWritingMinValid)) {
         // This log output is used in js tests so please leave it.
@@ -380,7 +379,7 @@ void syncFixUp(OperationContext* opCtx,
                const FixUpInfo& fixUpInfo,
                const RollbackSource& rollbackSource,
                ReplicationCoordinator* replCoord,
-               StorageInterface* storageInterface) {
+               ReplicationProcess* replicationProcess) {
     // fetch all first so we needn't handle interruption in a fancy way
 
     unsigned long long totalSize = 0;
@@ -419,7 +418,7 @@ void syncFixUp(OperationContext* opCtx,
     }
 
     log() << "rollback 3.5";
-    checkRbidAndUpdateMinValid(opCtx, fixUpInfo.rbid, rollbackSource, storageInterface);
+    checkRbidAndUpdateMinValid(opCtx, fixUpInfo.rbid, rollbackSource, replicationProcess);
 
     // update them
     log() << "rollback 4 n:" << goodVersions.size();
@@ -521,7 +520,7 @@ void syncFixUp(OperationContext* opCtx,
         // we did more reading from primary, so check it again for a rollback (which would mess
         // us up), and make minValid newer.
         log() << "rollback 4.2";
-        checkRbidAndUpdateMinValid(opCtx, fixUpInfo.rbid, rollbackSource, storageInterface);
+        checkRbidAndUpdateMinValid(opCtx, fixUpInfo.rbid, rollbackSource, replicationProcess);
     }
 
     log() << "rollback 4.6";
@@ -785,7 +784,7 @@ Status _syncRollback(OperationContext* opCtx,
                      const RollbackSource& rollbackSource,
                      int requiredRBID,
                      ReplicationCoordinator* replCoord,
-                     StorageInterface* storageInterface) {
+                     ReplicationProcess* replicationProcess) {
     invariant(!opCtx->lockState()->isLocked());
 
     FixUpInfo how;
@@ -828,10 +827,10 @@ Status _syncRollback(OperationContext* opCtx,
     log() << "rollback 3 fixup";
     try {
         ON_BLOCK_EXIT([&] {
-            auto status = ReplicationProcess::get(opCtx)->incrementRollbackID(opCtx);
+            auto status = replicationProcess->incrementRollbackID(opCtx);
             fassertStatusOK(40425, status);
         });
-        syncFixUp(opCtx, how, rollbackSource, replCoord, storageInterface);
+        syncFixUp(opCtx, how, rollbackSource, replCoord, replicationProcess);
     } catch (const RSFatalException& e) {
         return Status(ErrorCodes::UnrecoverableRollbackError, e.what(), 18753);
     }
@@ -856,7 +855,7 @@ Status syncRollback(OperationContext* opCtx,
                     const RollbackSource& rollbackSource,
                     int requiredRBID,
                     ReplicationCoordinator* replCoord,
-                    StorageInterface* storageInterface) {
+                    ReplicationProcess* replicationProcess) {
     invariant(opCtx);
     invariant(replCoord);
 
@@ -864,8 +863,8 @@ Status syncRollback(OperationContext* opCtx,
 
     DisableDocumentValidation validationDisabler(opCtx);
     UnreplicatedWritesBlock replicationDisabler(opCtx);
-    Status status =
-        _syncRollback(opCtx, localOplog, rollbackSource, requiredRBID, replCoord, storageInterface);
+    Status status = _syncRollback(
+        opCtx, localOplog, rollbackSource, requiredRBID, replCoord, replicationProcess);
 
     log() << "rollback finished" << rsLog;
     return status;
@@ -876,7 +875,7 @@ void rollback(OperationContext* opCtx,
               const RollbackSource& rollbackSource,
               int requiredRBID,
               ReplicationCoordinator* replCoord,
-              StorageInterface* storageInterface,
+              ReplicationProcess* replicationProcess,
               stdx::function<void(int)> sleepSecsFn) {
     // Set state to ROLLBACK while we are in this function. This prevents serving reads, even from
     // the oplog. This can fail if we are elected PRIMARY, in which case we better not do any
@@ -901,7 +900,7 @@ void rollback(OperationContext* opCtx,
 
     try {
         auto status = syncRollback(
-            opCtx, localOplog, rollbackSource, requiredRBID, replCoord, storageInterface);
+            opCtx, localOplog, rollbackSource, requiredRBID, replCoord, replicationProcess);
 
         // Abort only when syncRollback detects we are in a unrecoverable state.
         // WARNING: these statuses sometimes have location codes which are lost with uassertStatusOK
@@ -920,8 +919,8 @@ void rollback(OperationContext* opCtx,
         invariant(ex.getCode() != ErrorCodes::UnrecoverableRollbackError);
 
         warning() << "rollback cannot complete at this time (retrying later): " << redact(ex)
-                  << " appliedThrough=" << replCoord->getMyLastAppliedOpTime()
-                  << " minvalid=" << storageInterface->getMinValid(opCtx);
+                  << " appliedThrough=" << replCoord->getMyLastAppliedOpTime() << " minvalid="
+                  << replicationProcess->getConsistencyMarkers()->getMinValid(opCtx);
 
         // Sleep a bit to allow upstream node to coalesce, if that was the cause of the failure. If
         // we failed in a way that will keep failing, but wasn't flagged as a fatal failure, this
diff --git a/src/mongo/db/repl/rs_rollback.h b/src/mongo/db/repl/rs_rollback.h
index e305c076e44..db1b174c3dd 100644
--- a/src/mongo/db/repl/rs_rollback.h
+++ b/src/mongo/db/repl/rs_rollback.h
@@ -45,8 +45,8 @@ namespace repl {
 
 class OplogInterface;
 class ReplicationCoordinator;
+class ReplicationProcess;
 class RollbackSource;
-class StorageInterface;
 
 /**
  * Entry point to rollback process.
@@ -63,7 +63,7 @@ void rollback(OperationContext* opCtx,
               const RollbackSource& rollbackSource,
               int requiredRBID,
               ReplicationCoordinator* replCoord,
-              StorageInterface* storageInterface,
+              ReplicationProcess* replicationProcess,
               stdx::function<void(int)> sleepSecsFn = [](int secs) { sleepsecs(secs); });
 
 /**
@@ -87,7 +87,7 @@ void rollback(OperationContext* opCtx,
  *            supports fetching documents and copying collections.
  * @param requiredRBID Rollback ID we are required to have throughout rollback.
  * @param replCoord Used to track the rollback ID and to change the follower state
- * @param storageInterface Used to update minValid.
+ * @param replicationProcess Used to update minValid.
  *
  * If requiredRBID is supplied, we error if the upstream node has a different RBID (ie it rolled
  * back) after fetching any information from it.
@@ -101,7 +101,7 @@ Status syncRollback(OperationContext* opCtx,
                     const RollbackSource& rollbackSource,
                     int requiredRBID,
                     ReplicationCoordinator* replCoord,
-                    StorageInterface* storageInterface);
+                    ReplicationProcess* replicationProcess);
 
 /**
  * This namespace contains internal details of the rollback system. It is only exposed in a header
diff --git a/src/mongo/db/repl/rs_rollback_test.cpp b/src/mongo/db/repl/rs_rollback_test.cpp
index d16bbaf229e..6e3615ceee0 100644
--- a/src/mongo/db/repl/rs_rollback_test.cpp
+++ b/src/mongo/db/repl/rs_rollback_test.cpp
@@ -127,14 +127,16 @@ OplogInterfaceMock::Operation makeNoopOplogEntryAndRecordId(Seconds seconds) {
 }
 
 TEST_F(RSRollbackTest, InconsistentMinValid) {
-    _storageInterface.setAppliedThrough(_opCtx.get(), OpTime(Timestamp(Seconds(0), 0), 0));
-    _storageInterface.setMinValid(_opCtx.get(), OpTime(Timestamp(Seconds(1), 0), 0));
+    _replicationProcess->getConsistencyMarkers()->setAppliedThrough(
+        _opCtx.get(), OpTime(Timestamp(Seconds(0), 0), 0));
+    _replicationProcess->getConsistencyMarkers()->setMinValid(_opCtx.get(),
+                                                              OpTime(Timestamp(Seconds(1), 0), 0));
     auto status = syncRollback(_opCtx.get(),
                                OplogInterfaceMock(),
                                RollbackSourceMock(stdx::make_unique<OplogInterfaceMock>()),
                                {},
                                _coordinator,
-                               &_storageInterface);
+                               _replicationProcess.get());
     ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code());
     ASSERT_EQUALS(18752, status.location());
 }
@@ -151,7 +153,7 @@ TEST_F(RSRollbackTest, OplogStartMissing) {
                                RollbackSourceMock(std::move(remoteOplog)),
                                {},
                                _coordinator,
-                               &_storageInterface)
+                               _replicationProcess.get())
                       .code());
 }
 
@@ -164,7 +166,7 @@ TEST_F(RSRollbackTest, NoRemoteOpLog) {
                                RollbackSourceMock(stdx::make_unique<OplogInterfaceMock>()),
                                {},
                                _coordinator,
-                               &_storageInterface);
+                               _replicationProcess.get());
     ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code());
     ASSERT_EQUALS(18752, status.location());
 }
@@ -186,7 +188,7 @@ TEST_F(RSRollbackTest, RemoteGetRollbackIdThrows) {
                                     RollbackSourceLocal(stdx::make_unique<OplogInterfaceMock>()),
                                     {},
                                     _coordinator,
-                                    &_storageInterface),
+                                    _replicationProcess.get()),
                        UserException,
                        ErrorCodes::UnknownError);
 }
@@ -209,7 +211,7 @@ TEST_F(RSRollbackTest, RemoteGetRollbackIdDiffersFromRequiredRBID) {
                                     RollbackSourceLocal(stdx::make_unique<OplogInterfaceMock>()),
                                     1,
                                     _coordinator,
-                                    &_storageInterface),
+                                    _replicationProcess.get()),
                        UserException,
                        ErrorCodes::Error(40362));
 }
@@ -227,7 +229,7 @@ TEST_F(RSRollbackTest, BothOplogsAtCommonPoint) {
                      }))),
                      {},
                      _coordinator,
-                     &_storageInterface));
+                     _replicationProcess.get()));
 }
 
 /**
@@ -261,7 +263,7 @@ Collection* _createCollection(OperationContext* opCtx,
  */
 int _testRollbackDelete(OperationContext* opCtx,
                         ReplicationCoordinator* coordinator,
-                        StorageInterface* storageInterface,
+                        ReplicationProcess* replicationProcess,
                         const BSONObj& documentAtSource) {
     auto commonOperation =
         std::make_pair(BSON("ts" << Timestamp(Seconds(1), 0) << "h" << 1LL), RecordId(1));
@@ -297,7 +299,7 @@ int _testRollbackDelete(OperationContext* opCtx,
                            rollbackSource,
                            {},
                            coordinator,
-                           storageInterface));
+                           replicationProcess));
     ASSERT_TRUE(rollbackSource.called);
 
     Lock::DBLock dbLock(opCtx, "test", MODE_S);
@@ -313,16 +315,16 @@ int _testRollbackDelete(OperationContext* opCtx,
 
 TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionDoesNotExist) {
     createOplog(_opCtx.get());
-    ASSERT_EQUALS(-1,
-                  _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, BSONObj()));
+    ASSERT_EQUALS(
+        -1, _testRollbackDelete(_opCtx.get(), _coordinator, _replicationProcess.get(), BSONObj()));
 }
 
 TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionExistsNonCapped) {
     createOplog(_opCtx.get());
     _createCollection(_opCtx.get(), "test.t", CollectionOptions());
-    _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, BSONObj());
-    ASSERT_EQUALS(0,
-                  _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, BSONObj()));
+    _testRollbackDelete(_opCtx.get(), _coordinator, _replicationProcess.get(), BSONObj());
+    ASSERT_EQUALS(
+        0, _testRollbackDelete(_opCtx.get(), _coordinator, _replicationProcess.get(), BSONObj()));
 }
 
 TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionExistsCapped) {
@@ -330,16 +332,17 @@ TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionExistsCapped) {
     CollectionOptions options;
     options.capped = true;
     _createCollection(_opCtx.get(), "test.t", options);
-    ASSERT_EQUALS(0,
-                  _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, BSONObj()));
+    ASSERT_EQUALS(
+        0, _testRollbackDelete(_opCtx.get(), _coordinator, _replicationProcess.get(), BSONObj()));
 }
 
 TEST_F(RSRollbackTest, RollbackDeleteRestoreDocument) {
     createOplog(_opCtx.get());
     _createCollection(_opCtx.get(), "test.t", CollectionOptions());
     BSONObj doc = BSON("_id" << 0 << "a" << 1);
-    _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, doc);
-    ASSERT_EQUALS(1, _testRollbackDelete(_opCtx.get(), _coordinator, &_storageInterface, doc));
+    _testRollbackDelete(_opCtx.get(), _coordinator, _replicationProcess.get(), doc);
+    ASSERT_EQUALS(1,
+                  _testRollbackDelete(_opCtx.get(), _coordinator, _replicationProcess.get(), doc));
 }
 
 TEST_F(RSRollbackTest, RollbackInsertDocumentWithNoId) {
@@ -376,7 +379,7 @@ TEST_F(RSRollbackTest, RollbackInsertDocumentWithNoId) {
                                rollbackSource,
                                {},
                                _coordinator,
-                               &_storageInterface);
+                               _replicationProcess.get());
     stopCapturingLogMessages();
     ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code());
     ASSERT_EQUALS(18752, status.location());
@@ -441,7 +444,7 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommand) {
         rollbackSource,
         {},
         _coordinator,
-        &_storageInterface));
+        _replicationProcess.get()));
     stopCapturingLogMessages();
     ASSERT_EQUALS(1,
                   countLogLinesContaining("rollback drop index: collection: test.t. index: a_1"));
@@ -502,7 +505,7 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommandIndexNotInCatalog) {
                            rollbackSource,
                            {},
                            _coordinator,
-                           &_storageInterface));
+                           _replicationProcess.get()));
     stopCapturingLogMessages();
     ASSERT_EQUALS(1,
                   countLogLinesContaining("rollback drop index: collection: test.t. index: a_1"));
@@ -551,7 +554,7 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommandMissingNamespace) {
                                rollbackSource,
                                {},
                                _coordinator,
-                               &_storageInterface);
+                               _replicationProcess.get());
     stopCapturingLogMessages();
     ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code());
     ASSERT_EQUALS(18752, status.location());
@@ -614,7 +617,7 @@ TEST_F(RSRollbackTest, RollbackDropIndexCommandWithOneIndex) {
                            rollbackSource,
                            {},
                            _coordinator,
-                           &_storageInterface));
+                           _replicationProcess.get()));
     ASSERT(rollbackSource.called);
 }
 
@@ -694,7 +697,7 @@ TEST_F(RSRollbackTest, RollbackDropIndexCommandWithMultipleIndexes) {
         rollbackSource,
         {},
         _coordinator,
-        &_storageInterface));
+        _replicationProcess.get()));
     ASSERT(rollbackSource.called);
 }
 
@@ -737,7 +740,7 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommandInvalidNamespace) {
                                rollbackSource,
                                {},
                                _coordinator,
-                               &_storageInterface);
+                               _replicationProcess.get());
     stopCapturingLogMessages();
     ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code());
     ASSERT_EQUALS(18752, status.location());
@@ -783,7 +786,7 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommandMissingIndexName) {
                                rollbackSource,
                                {},
                                _coordinator,
-                               &_storageInterface);
+                               _replicationProcess.get());
     stopCapturingLogMessages();
     ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code());
     ASSERT_EQUALS(18752, status.location());
@@ -820,7 +823,7 @@ TEST_F(RSRollbackTest, RollbackUnknownCommand) {
                      }))),
                      {},
                      _coordinator,
-                     &_storageInterface);
+                     _replicationProcess.get());
     ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code());
     ASSERT_EQUALS(18751, status.location());
 }
@@ -857,7 +860,7 @@ TEST_F(RSRollbackTest, RollbackDropCollectionCommand) {
                            rollbackSource,
                            {},
                            _coordinator,
-                           &_storageInterface));
+                           _replicationProcess.get()));
     ASSERT_TRUE(rollbackSource.called);
 }
 
@@ -896,7 +899,7 @@ TEST_F(RSRollbackTest, RollbackDropCollectionCommandFailsIfRBIDChangesWhileSynci
                                     rollbackSource,
                                     0,
                                     _coordinator,
-                                    &_storageInterface),
+                                    _replicationProcess.get()),
                        DBException,
                        40365);
     ASSERT(rollbackSource.copyCollectionCalled);
@@ -1021,7 +1024,7 @@ TEST_F(RSRollbackTest, RollbackApplyOpsCommand) {
                            rollbackSource,
                            {},
                            _coordinator,
-                           &_storageInterface));
+                           _replicationProcess.get()));
     ASSERT_EQUALS(4U, rollbackSource.searchedIds.size());
     ASSERT_EQUALS(1U, rollbackSource.searchedIds.count(1));
     ASSERT_EQUALS(1U, rollbackSource.searchedIds.count(2));
@@ -1062,7 +1065,7 @@ TEST_F(RSRollbackTest, RollbackCreateCollectionCommand) {
                            rollbackSource,
                            {},
                            _coordinator,
-                           &_storageInterface));
+                           _replicationProcess.get()));
     {
         Lock::DBLock dbLock(_opCtx.get(), "test", MODE_S);
         auto db = dbHolder().get(_opCtx.get(), "test");
@@ -1106,7 +1109,7 @@ TEST_F(RSRollbackTest, RollbackCollectionModificationCommand) {
                            rollbackSource,
                            {},
                            _coordinator,
-                           &_storageInterface));
+                           _replicationProcess.get()));
     stopCapturingLogMessages();
     ASSERT_TRUE(rollbackSource.called);
     for (const auto& message : getCapturedLogMessages()) {
@@ -1148,7 +1151,7 @@ TEST_F(RSRollbackTest, RollbackCollectionModificationCommandInvalidCollectionOpt
                      rollbackSource,
                      {},
                      _coordinator,
-                     &_storageInterface);
+                     _replicationProcess.get());
     ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, status.code());
     ASSERT_EQUALS(18753, status.location());
 }
@@ -1196,7 +1199,7 @@ TEST_F(RSRollbackTest, RollbackReturnsImmediatelyOnFailureToTransitionToRollback
              rollbackSourceWithInvalidOplog,
              {},
              _coordinator,
-             &_storageInterface);
+             _replicationProcess.get());
     stopCapturingLogMessages();
 
     ASSERT_EQUALS(1, countLogLinesContaining("Cannot transition from SECONDARY to ROLLBACK"));
@@ -1219,7 +1222,7 @@ DEATH_TEST_F(RSRollbackTest,
              rollbackSourceWithInvalidOplog,
              {},
              _coordinator,
-             &_storageInterface);
+             _replicationProcess.get());
 }
 
 TEST_F(RSRollbackTest, RollbackLogsRetryMessageAndReturnsOnNonUnrecoverableRollbackError) {
@@ -1238,7 +1241,7 @@ TEST_F(RSRollbackTest, RollbackLogsRetryMessageAndReturnsOnNonUnrecoverableRollb
              rollbackSourceWithValidOplog,
              {},
              _coordinator,
-             &_storageInterface,
+             _replicationProcess.get(),
              noopSleepSecsFn);
     stopCapturingLogMessages();
 
@@ -1261,7 +1264,7 @@ DEATH_TEST_F(RSRollbackTest,
     ASSERT_TRUE(ShardIdentityRollbackNotifier::get(_opCtx.get())->didRollbackHappen());
 
     createOplog(_opCtx.get());
-    rollback(_opCtx.get(), localOplog, rollbackSource, {}, _coordinator, &_storageInterface);
+    rollback(_opCtx.get(), localOplog, rollbackSource, {}, _coordinator, _replicationProcess.get());
 }
 
 DEATH_TEST_F(
@@ -1277,7 +1280,7 @@ DEATH_TEST_F(
     _coordinator->_failSetFollowerModeOnThisMemberState = MemberState::RS_RECOVERING;
 
     createOplog(_opCtx.get());
-    rollback(_opCtx.get(), localOplog, rollbackSource, {}, _coordinator, &_storageInterface);
+    rollback(_opCtx.get(), localOplog, rollbackSource, {}, _coordinator, _replicationProcess.get());
 }
 
 // The testcases used here are trying to detect off-by-one errors in
diff --git a/src/mongo/db/repl/storage_interface.h b/src/mongo/db/repl/storage_interface.h
index bf5a101b6a7..630d481660f 100644
--- a/src/mongo/db/repl/storage_interface.h
+++ b/src/mongo/db/repl/storage_interface.h
@@ -61,28 +61,6 @@ namespace repl {
  *      * Drop database and all user databases
  *      * Drop a collection
  *      * Insert documents into a collection
- *      * Manage minvalid boundaries and initial sync state
- *
- * ***** MINVALID *****
- * This interface provides helper functions for maintaining a single document in the
- * local.replset.minvalid collection.
- *
- * When a member reaches its minValid optime it is in a consistent state.  Thus, minValid is
- * set as the last step in initial sync.  At the beginning of initial sync, doingInitialSync
- * is appended onto minValid to indicate that initial sync was started but has not yet
- * completed.
- *
- * The document is also updated during "normal" sync. The optime of the last op in each batch is
- * used to set minValid, along with a "begin" field to demark the start and the fact that a batch
- * is active. When the batch is done the "begin" field is removed to indicate that we are in a
- * consistent state when the batch has been fully applied.
- *
- * Example of all fields:
- * { _id:...,
- *      doingInitialSync: true // initial sync is active
- *      ts:..., t:...   // end-OpTime
- *      begin: {ts:..., t:...} // a batch is currently being applied, and not consistent
- * }
  */
 class StorageInterface {
     MONGO_DISALLOW_COPYING(StorageInterface);
@@ -98,42 +76,6 @@ public:
     StorageInterface() = default;
     virtual ~StorageInterface() = default;
 
-    // MinValid and Initial Sync Flag.
-    /**
-     * Returns true if initial sync was started but has not not completed.
-     */
-    virtual bool getInitialSyncFlag(OperationContext* opCtx) const = 0;
-
-    /**
-     * Sets the the initial sync flag to record that initial sync has not completed.
-     *
-     * This operation is durable and waits for durable writes (which will block on
-     *journaling/checkpointing).
-     */
-    virtual void setInitialSyncFlag(OperationContext* opCtx) = 0;
-
-    /**
-     * Clears the the initial sync flag to record that initial sync has completed.
-     *
-     * This operation is durable and waits for durable writes (which will block on
-     *journaling/checkpointing).
-     */
-    virtual void clearInitialSyncFlag(OperationContext* opCtx) = 0;
-
-    /**
-     * The minValid value is the earliest (minimum) Timestamp that must be applied in order to
-     * consider the dataset consistent.
-     */
-    virtual void setMinValid(OperationContext* opCtx, const OpTime& minValid) = 0;
-    virtual OpTime getMinValid(OperationContext* opCtx) const = 0;
-
-    /**
-     * Sets minValid only if it is not already higher than endOpTime.
-     * Warning, this compares the term and timestamp independently. Do not use if the current
-     * minValid could be from the other fork of a rollback.
-     */
-    virtual void setMinValidToAtLeast(OperationContext* opCtx, const OpTime& endOpTime) = 0;
-
     /**
      * Rollback ID is an increasing counter of how many rollbacks have occurred on this server.
      */
@@ -141,27 +83,6 @@ public:
     virtual Status initializeRollbackID(OperationContext* opCtx) = 0;
     virtual Status incrementRollbackID(OperationContext* opCtx) = 0;
 
-    /**
-     * On startup all oplog entries with a value >= the oplog delete from point should be deleted.
-     * If null, no documents should be deleted.
-     */
-    virtual void setOplogDeleteFromPoint(OperationContext* opCtx, const Timestamp& timestamp) = 0;
-    virtual Timestamp getOplogDeleteFromPoint(OperationContext* opCtx) = 0;
-
-    /**
-     * The applied through point is a persistent record of where we've applied through. If null, the
-     * applied through point is the top of the oplog.
-     */
-    virtual void setAppliedThrough(OperationContext* opCtx, const OpTime& optime) = 0;
-
-    /**
-     * You should probably be calling ReplicationCoordinator::getLastAppliedOpTime() instead.
-     *
-     * This reads the value from storage which isn't always updated when the ReplicationCoordinator
-     * is.
-     */
-    virtual OpTime getAppliedThrough(OperationContext* opCtx) = 0;
-
 
     // Collection creation and population for initial sync.
     /**
diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index 4ab7100f2b3..ae01e86d259 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -80,10 +80,6 @@
 namespace mongo {
 namespace repl {
 
-const char StorageInterfaceImpl::kDefaultMinValidNamespace[] = "local.replset.minvalid";
-const char StorageInterfaceImpl::kInitialSyncFlagFieldName[] = "doingInitialSync";
-const char StorageInterfaceImpl::kBeginFieldName[] = "begin";
-const char StorageInterfaceImpl::kOplogDeleteFromPointFieldName[] = "oplogDeleteFromPoint";
 const char StorageInterfaceImpl::kDefaultRollbackIdNamespace[] = "local.system.rollback.id";
 const char StorageInterfaceImpl::kRollbackIdFieldName[] = "rollbackId";
 const char StorageInterfaceImpl::kRollbackIdDocumentId[] = "rollbackId";
@@ -91,48 +87,12 @@ const char StorageInterfaceImpl::kRollbackIdDocumentId[] = "rollbackId";
 namespace {
 using UniqueLock = stdx::unique_lock<stdx::mutex>;
 
-const BSONObj kInitialSyncFlag(BSON(StorageInterfaceImpl::kInitialSyncFlagFieldName << true));
-
 const auto kIdIndexName = "_id_"_sd;
 
 }  // namespace
 
 StorageInterfaceImpl::StorageInterfaceImpl()
-    : StorageInterfaceImpl(NamespaceString(StorageInterfaceImpl::kDefaultMinValidNamespace)) {}
-
-StorageInterfaceImpl::StorageInterfaceImpl(const NamespaceString& minValidNss)
-    : _minValidNss(minValidNss),
-      _rollbackIdNss(StorageInterfaceImpl::kDefaultRollbackIdNamespace) {}
-
-NamespaceString StorageInterfaceImpl::getMinValidNss() const {
-    return _minValidNss;
-}
-
-BSONObj StorageInterfaceImpl::getMinValidDocument(OperationContext* opCtx) const {
-    MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
-        Lock::DBLock dblk(opCtx, _minValidNss.db(), MODE_IS);
-        Lock::CollectionLock lk(opCtx->lockState(), _minValidNss.ns(), MODE_IS);
-        BSONObj doc;
-        bool found = Helpers::getSingleton(opCtx, _minValidNss.ns().c_str(), doc);
-        invariant(found || doc.isEmpty());
-        return doc;
-    }
-    MONGO_WRITE_CONFLICT_RETRY_LOOP_END(
-        opCtx, "StorageInterfaceImpl::getMinValidDocument", _minValidNss.ns());
-
-    MONGO_UNREACHABLE;
-}
-
-void StorageInterfaceImpl::updateMinValidDocument(OperationContext* opCtx,
-                                                  const BSONObj& updateSpec) {
-    MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
-        // For now this needs to be MODE_X because it sometimes creates the collection.
-        Lock::DBLock dblk(opCtx, _minValidNss.db(), MODE_X);
-        Helpers::putSingleton(opCtx, _minValidNss.ns().c_str(), updateSpec);
-    }
-    MONGO_WRITE_CONFLICT_RETRY_LOOP_END(
-        opCtx, "StorageInterfaceImpl::updateMinValidDocument", _minValidNss.ns());
-}
+    : _rollbackIdNss(StorageInterfaceImpl::kDefaultRollbackIdNamespace) {}
 
 StatusWith<int> StorageInterfaceImpl::getRollbackID(OperationContext* opCtx) {
     BSONObjBuilder bob;
@@ -201,114 +161,6 @@ Status StorageInterfaceImpl::incrementRollbackID(OperationContext* opCtx) {
     return status;
 }
 
-bool StorageInterfaceImpl::getInitialSyncFlag(OperationContext* opCtx) const {
-    const BSONObj doc = getMinValidDocument(opCtx);
-    const auto flag = doc[kInitialSyncFlagFieldName].trueValue();
-    LOG(3) << "returning initial sync flag value of " << flag;
-    return flag;
-}
-
-void StorageInterfaceImpl::setInitialSyncFlag(OperationContext* opCtx) {
-    LOG(3) << "setting initial sync flag";
-    updateMinValidDocument(opCtx, BSON("$set" << kInitialSyncFlag));
-    opCtx->recoveryUnit()->waitUntilDurable();
-}
-
-void StorageInterfaceImpl::clearInitialSyncFlag(OperationContext* opCtx) {
-    LOG(3) << "clearing initial sync flag";
-
-    auto replCoord = repl::ReplicationCoordinator::get(opCtx);
-    OpTime time = replCoord->getMyLastAppliedOpTime();
-    updateMinValidDocument(
-        opCtx,
-        BSON("$unset" << kInitialSyncFlag << "$set"
-                      << BSON("ts" << time.getTimestamp() << "t" << time.getTerm()
-                                   << kBeginFieldName
-                                   << time)));
-
-    if (getGlobalServiceContext()->getGlobalStorageEngine()->isDurable()) {
-        opCtx->recoveryUnit()->waitUntilDurable();
-        replCoord->setMyLastDurableOpTime(time);
-    }
-}
-
-OpTime StorageInterfaceImpl::getMinValid(OperationContext* opCtx) const {
-    const BSONObj doc = getMinValidDocument(opCtx);
-    const auto opTimeStatus = OpTime::parseFromOplogEntry(doc);
-    // If any of the keys (fields) are missing from the minvalid document, we return
-    // a null OpTime.
-    if (opTimeStatus == ErrorCodes::NoSuchKey) {
-        return {};
-    }
-
-    if (!opTimeStatus.isOK()) {
-        severe() << "Error parsing minvalid entry: " << redact(doc)
-                 << ", with status:" << opTimeStatus.getStatus();
-        fassertFailedNoTrace(40052);
-    }
-
-    OpTime minValid = opTimeStatus.getValue();
-    LOG(3) << "returning minvalid: " << minValid.toString() << "(" << minValid.toBSON() << ")";
-
-    return minValid;
-}
-
-void StorageInterfaceImpl::setMinValid(OperationContext* opCtx, const OpTime& minValid) {
-    LOG(3) << "setting minvalid to exactly: " << minValid.toString() << "(" << minValid.toBSON()
-           << ")";
-    updateMinValidDocument(
-        opCtx, BSON("$set" << BSON("ts" << minValid.getTimestamp() << "t" << minValid.getTerm())));
-}
-
-void StorageInterfaceImpl::setMinValidToAtLeast(OperationContext* opCtx, const OpTime& minValid) {
-    LOG(3) << "setting minvalid to at least: " << minValid.toString() << "(" << minValid.toBSON()
-           << ")";
-    updateMinValidDocument(
-        opCtx, BSON("$max" << BSON("ts" << minValid.getTimestamp() << "t" << minValid.getTerm())));
-}
-
-void StorageInterfaceImpl::setOplogDeleteFromPoint(OperationContext* opCtx,
-                                                   const Timestamp& timestamp) {
-    LOG(3) << "setting oplog delete from point to: " << timestamp.toStringPretty();
-    updateMinValidDocument(opCtx,
-                           BSON("$set" << BSON(kOplogDeleteFromPointFieldName << timestamp)));
-}
-
-Timestamp StorageInterfaceImpl::getOplogDeleteFromPoint(OperationContext* opCtx) {
-    const BSONObj doc = getMinValidDocument(opCtx);
-    Timestamp out = {};
-    if (auto field = doc[kOplogDeleteFromPointFieldName]) {
-        out = field.timestamp();
-    }
-
-    LOG(3) << "returning oplog delete from point: " << out;
-    return out;
-}
-
-void StorageInterfaceImpl::setAppliedThrough(OperationContext* opCtx, const OpTime& optime) {
-    LOG(3) << "setting appliedThrough to: " << optime.toString() << "(" << optime.toBSON() << ")";
-    if (optime.isNull()) {
-        updateMinValidDocument(opCtx, BSON("$unset" << BSON(kBeginFieldName << 1)));
-    } else {
-        updateMinValidDocument(opCtx, BSON("$set" << BSON(kBeginFieldName << optime)));
-    }
-}
-
-OpTime StorageInterfaceImpl::getAppliedThrough(OperationContext* opCtx) {
-    const BSONObj doc = getMinValidDocument(opCtx);
-    const auto opTimeStatus = OpTime::parseFromOplogEntry(doc.getObjectField(kBeginFieldName));
-    if (!opTimeStatus.isOK()) {
-        // Return null OpTime on any parse failure, including if "begin" is missing.
-        return {};
-    }
-
-    OpTime appliedThrough = opTimeStatus.getValue();
-    LOG(3) << "returning appliedThrough: " << appliedThrough.toString() << "("
-           << appliedThrough.toBSON() << ")";
-
-    return appliedThrough;
-}
-
 StatusWith<std::unique_ptr<CollectionBulkLoader>>
 StorageInterfaceImpl::createCollectionForBulkLoading(
     const NamespaceString& nss,
diff --git a/src/mongo/db/repl/storage_interface_impl.h b/src/mongo/db/repl/storage_interface_impl.h
index a41d0ae4584..634fc428d05 100644
--- a/src/mongo/db/repl/storage_interface_impl.h
+++ b/src/mongo/db/repl/storage_interface_impl.h
@@ -45,38 +45,15 @@ class StorageInterfaceImpl : public StorageInterface {
     MONGO_DISALLOW_COPYING(StorageInterfaceImpl);
 
 public:
-    static const char kDefaultMinValidNamespace[];
-    static const char kInitialSyncFlagFieldName[];
-    static const char kBeginFieldName[];
-    static const char kOplogDeleteFromPointFieldName[];
     static const char kDefaultRollbackIdNamespace[];
     static const char kRollbackIdFieldName[];
     static const char kRollbackIdDocumentId[];
 
     StorageInterfaceImpl();
-    explicit StorageInterfaceImpl(const NamespaceString& minValidNss);
 
-    /**
-     * Returns namespace of collection containing the minvalid boundaries and initial sync flag.
-     */
-    NamespaceString getMinValidNss() const;
-
-    bool getInitialSyncFlag(OperationContext* opCtx) const override;
-
-    void setInitialSyncFlag(OperationContext* opCtx) override;
-
-    void clearInitialSyncFlag(OperationContext* opCtx) override;
-
-    OpTime getMinValid(OperationContext* opCtx) const override;
-    void setMinValid(OperationContext* opCtx, const OpTime& minValid) override;
-    void setMinValidToAtLeast(OperationContext* opCtx, const OpTime& endOpTime) override;
     StatusWith<int> getRollbackID(OperationContext* opCtx) override;
     Status initializeRollbackID(OperationContext* opCtx) override;
     Status incrementRollbackID(OperationContext* opCtx) override;
-    void setOplogDeleteFromPoint(OperationContext* opCtx, const Timestamp& timestamp) override;
-    Timestamp getOplogDeleteFromPoint(OperationContext* opCtx) override;
-    void setAppliedThrough(OperationContext* opCtx, const OpTime& optime) override;
-    OpTime getAppliedThrough(OperationContext* opCtx) override;
 
     /**
      *  Allocates a new TaskRunner for use by the passed in collection.
@@ -158,11 +135,6 @@ public:
     Status isAdminDbValid(OperationContext* opCtx) override;
 
 private:
-    // Returns empty document if not present.
-    BSONObj getMinValidDocument(OperationContext* opCtx) const;
-    void updateMinValidDocument(OperationContext* opCtx, const BSONObj& updateSpec);
-
-    const NamespaceString _minValidNss;
     const NamespaceString _rollbackIdNss;
 };
 
diff --git a/src/mongo/db/repl/storage_interface_impl_test.cpp b/src/mongo/db/repl/storage_interface_impl_test.cpp
index e15f54014b2..2c6087aef83 100644
--- a/src/mongo/db/repl/storage_interface_impl_test.cpp
+++ b/src/mongo/db/repl/storage_interface_impl_test.cpp
@@ -39,19 +39,15 @@
 #include "mongo/db/client.h"
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
-#include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
-#include "mongo/db/dbhelpers.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/repl/oplog.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_impl.h"
 #include "mongo/db/service_context_d_test_fixture.h"
-#include "mongo/db/storage/recovery_unit_noop.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/stdx/thread.h"
 #include "mongo/unittest/unittest.h"
@@ -87,22 +83,6 @@ NamespaceString makeNamespace(const T& t, const std::string& suffix = "") {
 }
 
 /**
- * Returns min valid document.
- */
-BSONObj getMinValidDocument(OperationContext* opCtx, const NamespaceString& minValidNss) {
-    MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
-        Lock::DBLock dblk(opCtx, minValidNss.db(), MODE_IS);
-        Lock::CollectionLock lk(opCtx->lockState(), minValidNss.ns(), MODE_IS);
-        BSONObj mv;
-        if (Helpers::getSingleton(opCtx, minValidNss.ns().c_str(), mv)) {
-            return mv;
-        }
-    }
-    MONGO_WRITE_CONFLICT_RETRY_LOOP_END(opCtx, "getMinValidDocument", minValidNss.ns());
-    return BSONObj();
-}
-
-/**
  * Creates collection options suitable for oplog.
  */
 CollectionOptions createOplogCollectionOptions() {
@@ -194,27 +174,12 @@ private:
         _ddv = stdx::make_unique<DisableDocumentValidation>(_opCtx.get());
     }
 
-private:
     ServiceContext::UniqueOperationContext _opCtx;
     std::unique_ptr<UnreplicatedWritesBlock> _uwb;
     std::unique_ptr<DisableDocumentValidation> _ddv;
     ReplicationCoordinatorMock* _replicationCoordinatorMock = nullptr;
 };
 
-/**
- * Recovery unit that tracks if waitUntilDurable() is called.
- */
-class RecoveryUnitWithDurabilityTracking : public RecoveryUnitNoop {
-public:
-    bool waitUntilDurable() override;
-    bool waitUntilDurableCalled = false;
-};
-
-bool RecoveryUnitWithDurabilityTracking::waitUntilDurable() {
-    waitUntilDurableCalled = true;
-    return RecoveryUnitNoop::waitUntilDurable();
-}
-
 TEST_F(StorageInterfaceImplTest, ServiceContextDecorator) {
     auto serviceContext = getServiceContext();
     ASSERT_FALSE(StorageInterface::get(serviceContext));
@@ -225,108 +190,6 @@ TEST_F(StorageInterfaceImplTest, ServiceContextDecorator) {
     ASSERT_TRUE(storage == StorageInterface::get(getOperationContext()));
 }
 
-TEST_F(StorageInterfaceImplTest, DefaultMinValidNamespace) {
-    ASSERT_EQUALS(NamespaceString(StorageInterfaceImpl::kDefaultMinValidNamespace),
-                  StorageInterfaceImpl().getMinValidNss());
-}
-
-TEST_F(StorageInterfaceImplTest, InitialSyncFlag) {
-    auto nss = makeNamespace(_agent);
-
-    StorageInterfaceImpl storage(nss);
-    auto opCtx = getOperationContext();
-
-    // Initial sync flag should be unset after initializing a new storage engine.
-    ASSERT_FALSE(storage.getInitialSyncFlag(opCtx));
-
-    // Setting initial sync flag should affect getInitialSyncFlag() result.
-    storage.setInitialSyncFlag(opCtx);
-    ASSERT_TRUE(storage.getInitialSyncFlag(opCtx));
-
-    // Check min valid document using storage engine interface.
-    auto minValidDocument = getMinValidDocument(opCtx, nss);
-    ASSERT_TRUE(minValidDocument.hasField(StorageInterfaceImpl::kInitialSyncFlagFieldName));
-    ASSERT_TRUE(minValidDocument.getBoolField(StorageInterfaceImpl::kInitialSyncFlagFieldName));
-
-    // Clearing initial sync flag should affect getInitialSyncFlag() result.
-    storage.clearInitialSyncFlag(opCtx);
-    ASSERT_FALSE(storage.getInitialSyncFlag(opCtx));
-}
-
-TEST_F(StorageInterfaceImplTest, GetMinValidAfterSettingInitialSyncFlagWorks) {
-    auto nss = makeNamespace(_agent);
-
-    StorageInterfaceImpl storage(nss);
-    auto opCtx = getOperationContext();
-
-    // Initial sync flag should be unset after initializing a new storage engine.
-    ASSERT_FALSE(storage.getInitialSyncFlag(opCtx));
-
-    // Setting initial sync flag should affect getInitialSyncFlag() result.
-    storage.setInitialSyncFlag(opCtx);
-    ASSERT_TRUE(storage.getInitialSyncFlag(opCtx));
-
-    ASSERT(storage.getMinValid(opCtx).isNull());
-    ASSERT(storage.getAppliedThrough(opCtx).isNull());
-    ASSERT(storage.getOplogDeleteFromPoint(opCtx).isNull());
-}
-
-TEST_F(StorageInterfaceImplTest, MinValid) {
-    auto nss = makeNamespace(_agent);
-
-    StorageInterfaceImpl storage(nss);
-    auto opCtx = getOperationContext();
-
-    // MinValid boundaries should all be null after initializing a new storage engine.
-    ASSERT(storage.getMinValid(opCtx).isNull());
-    ASSERT(storage.getAppliedThrough(opCtx).isNull());
-    ASSERT(storage.getOplogDeleteFromPoint(opCtx).isNull());
-
-    // Setting min valid boundaries should affect getMinValid() result.
-    OpTime startOpTime({Seconds(123), 0}, 1LL);
-    OpTime endOpTime({Seconds(456), 0}, 1LL);
-    storage.setAppliedThrough(opCtx, startOpTime);
-    storage.setMinValid(opCtx, endOpTime);
-    storage.setOplogDeleteFromPoint(opCtx, endOpTime.getTimestamp());
-
-    ASSERT_EQ(storage.getAppliedThrough(opCtx), startOpTime);
-    ASSERT_EQ(storage.getMinValid(opCtx), endOpTime);
-    ASSERT_EQ(storage.getOplogDeleteFromPoint(opCtx), endOpTime.getTimestamp());
-
-
-    // setMinValid always changes minValid, but setMinValidToAtLeast only does if higher.
-    storage.setMinValid(opCtx, startOpTime);  // Forcibly lower it.
-    ASSERT_EQ(storage.getMinValid(opCtx), startOpTime);
-    storage.setMinValidToAtLeast(opCtx, endOpTime);  // Higher than current (sets it).
-    ASSERT_EQ(storage.getMinValid(opCtx), endOpTime);
-    storage.setMinValidToAtLeast(opCtx, startOpTime);  // Lower than current (no-op).
-    ASSERT_EQ(storage.getMinValid(opCtx), endOpTime);
-
-    // Check min valid document using storage engine interface.
-    auto minValidDocument = getMinValidDocument(opCtx, nss);
-    ASSERT_TRUE(minValidDocument.hasField(StorageInterfaceImpl::kBeginFieldName));
-    ASSERT_TRUE(minValidDocument[StorageInterfaceImpl::kBeginFieldName].isABSONObj());
-    ASSERT_EQUALS(startOpTime,
-                  unittest::assertGet(OpTime::parseFromOplogEntry(
-                      minValidDocument[StorageInterfaceImpl::kBeginFieldName].Obj())));
-    ASSERT_EQUALS(endOpTime, unittest::assertGet(OpTime::parseFromOplogEntry(minValidDocument)));
-    ASSERT_EQUALS(
-        endOpTime.getTimestamp(),
-        minValidDocument[StorageInterfaceImpl::kOplogDeleteFromPointFieldName].timestamp());
-
-    // Recovery unit will be owned by "opCtx".
-    RecoveryUnitWithDurabilityTracking* recoveryUnit = new RecoveryUnitWithDurabilityTracking();
-    opCtx->setRecoveryUnit(recoveryUnit, OperationContext::kNotInUnitOfWork);
-
-    // Set min valid without waiting for the changes to be durable.
-    OpTime endOpTime2({Seconds(789), 0}, 1LL);
-    storage.setMinValid(opCtx, endOpTime2);
-    storage.setAppliedThrough(opCtx, {});
-    ASSERT_EQUALS(storage.getAppliedThrough(opCtx), OpTime());
-    ASSERT_EQUALS(storage.getMinValid(opCtx), endOpTime2);
-    ASSERT_FALSE(recoveryUnit->waitUntilDurableCalled);
-}
-
 TEST_F(StorageInterfaceImplTest, GetRollbackIDReturnsNamespaceNotFoundOnMissingCollection) {
     StorageInterfaceImpl storage;
     auto opCtx = getOperationContext();
@@ -474,7 +337,7 @@ TEST_F(StorageInterfaceImplTest, InsertDocumentsReturnsOKWhenNoOperationsAreGive
     auto opCtx = getOperationContext();
     auto nss = makeNamespace(_agent);
     createCollection(opCtx, nss);
-    StorageInterfaceImpl storage(nss);
+    StorageInterfaceImpl storage;
     ASSERT_OK(storage.insertDocuments(opCtx, nss, {}));
 }
 
@@ -487,7 +350,7 @@ TEST_F(StorageInterfaceImplTest,
     createCollection(opCtx, nss);
 
     // Non-oplog collection will enforce mandatory _id field requirement on insertion.
-    StorageInterfaceImpl storage(nss);
+    StorageInterfaceImpl storage;
     auto op = makeOplogEntry({Timestamp(Seconds(1), 0), 1LL});
     auto status = storage.insertDocuments(opCtx, nss, {op});
     ASSERT_EQUALS(ErrorCodes::InternalError, status);
@@ -504,7 +367,7 @@ TEST_F(StorageInterfaceImplTest,
     options.cappedSize = 1024 * 1024;
     createCollection(opCtx, nss, options);
     // StorageInterfaceImpl::insertDocuments should fall back on inserting the batch one at a time.
-    StorageInterfaceImpl storage(nss);
+    StorageInterfaceImpl storage;
     auto doc1 = BSON("_id" << 1);
     auto doc2 = BSON("_id" << 2);
     std::vector<BSONObj> docs({doc1, doc2});
diff --git a/src/mongo/db/repl/storage_interface_mock.cpp b/src/mongo/db/repl/storage_interface_mock.cpp
index 531e4b17310..64ebecdc573 100644
--- a/src/mongo/db/repl/storage_interface_mock.cpp
+++ b/src/mongo/db/repl/storage_interface_mock.cpp
@@ -39,38 +39,8 @@
 namespace mongo {
 namespace repl {
 
-bool StorageInterfaceMock::getInitialSyncFlag(OperationContext* opCtx) const {
-    stdx::lock_guard<stdx::mutex> lock(_initialSyncFlagMutex);
-    return _initialSyncFlag;
-}
-
-void StorageInterfaceMock::setInitialSyncFlag(OperationContext* opCtx) {
-    stdx::lock_guard<stdx::mutex> lock(_initialSyncFlagMutex);
-    _initialSyncFlag = true;
-}
-
-void StorageInterfaceMock::clearInitialSyncFlag(OperationContext* opCtx) {
-    stdx::lock_guard<stdx::mutex> lock(_initialSyncFlagMutex);
-    _initialSyncFlag = false;
-}
-
-OpTime StorageInterfaceMock::getMinValid(OperationContext* opCtx) const {
-    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
-    return _minValid;
-}
-
-void StorageInterfaceMock::setMinValid(OperationContext* opCtx, const OpTime& minValid) {
-    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
-    _minValid = minValid;
-}
-
-void StorageInterfaceMock::setMinValidToAtLeast(OperationContext* opCtx, const OpTime& minValid) {
-    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
-    _minValid = std::max(_minValid, minValid);
-}
-
 StatusWith<int> StorageInterfaceMock::getRollbackID(OperationContext* opCtx) {
-    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
+    stdx::lock_guard<stdx::mutex> lock(_rbidMutex);
     if (!_rbidInitialized) {
         return Status(ErrorCodes::NamespaceNotFound, "Rollback ID not initialized");
     }
@@ -78,7 +48,7 @@ StatusWith<int> StorageInterfaceMock::getRollbackID(OperationContext* opCtx) {
 }
 
 Status StorageInterfaceMock::initializeRollbackID(OperationContext* opCtx) {
-    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
+    stdx::lock_guard<stdx::mutex> lock(_rbidMutex);
     if (_rbidInitialized) {
         return Status(ErrorCodes::NamespaceExists, "Rollback ID already initialized");
     }
@@ -90,35 +60,13 @@ Status StorageInterfaceMock::initializeRollbackID(OperationContext* opCtx) {
 }
 
 Status StorageInterfaceMock::incrementRollbackID(OperationContext* opCtx) {
-    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
+    stdx::lock_guard<stdx::mutex> lock(_rbidMutex);
     if (!_rbidInitialized) {
         return Status(ErrorCodes::NamespaceNotFound, "Rollback ID not initialized");
     }
     _rbid++;
     return Status::OK();
 }
-
-void StorageInterfaceMock::setOplogDeleteFromPoint(OperationContext* opCtx,
-                                                   const Timestamp& timestamp) {
-    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
-    _oplogDeleteFromPoint = timestamp;
-}
-
-Timestamp StorageInterfaceMock::getOplogDeleteFromPoint(OperationContext* opCtx) {
-    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
-    return _oplogDeleteFromPoint;
-}
-
-void StorageInterfaceMock::setAppliedThrough(OperationContext* opCtx, const OpTime& optime) {
-    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
-    _appliedThrough = optime;
-}
-
-OpTime StorageInterfaceMock::getAppliedThrough(OperationContext* opCtx) {
-    stdx::lock_guard<stdx::mutex> lock(_minValidBoundariesMutex);
-    return _appliedThrough;
-}
-
 Status CollectionBulkLoaderMock::init(Collection* coll,
                                       const std::vector<BSONObj>& secondaryIndexSpecs) {
     LOG(1) << "CollectionBulkLoaderMock::init called";
diff --git a/src/mongo/db/repl/storage_interface_mock.h b/src/mongo/db/repl/storage_interface_mock.h
index eaf888b036e..74a6bac4abc 100644
--- a/src/mongo/db/repl/storage_interface_mock.h
+++ b/src/mongo/db/repl/storage_interface_mock.h
@@ -122,20 +122,9 @@ public:
 
     StorageInterfaceMock() = default;
 
-    bool getInitialSyncFlag(OperationContext* opCtx) const override;
-    void setInitialSyncFlag(OperationContext* opCtx) override;
-    void clearInitialSyncFlag(OperationContext* opCtx) override;
-
-    OpTime getMinValid(OperationContext* opCtx) const override;
-    void setMinValid(OperationContext* opCtx, const OpTime& minValid) override;
-    void setMinValidToAtLeast(OperationContext* opCtx, const OpTime& minValid) override;
     StatusWith<int> getRollbackID(OperationContext* opCtx) override;
     Status initializeRollbackID(OperationContext* opCtx) override;
     Status incrementRollbackID(OperationContext* opCtx) override;
-    void setOplogDeleteFromPoint(OperationContext* opCtx, const Timestamp& timestamp) override;
-    Timestamp getOplogDeleteFromPoint(OperationContext* opCtx) override;
-    void setAppliedThrough(OperationContext* opCtx, const OpTime& optime) override;
-    OpTime getAppliedThrough(OperationContext* opCtx) override;
 
     StatusWith<std::unique_ptr<CollectionBulkLoader>> createCollectionForBulkLoading(
         const NamespaceString& nss,
@@ -306,15 +295,9 @@ public:
     };
 
 private:
-    mutable stdx::mutex _initialSyncFlagMutex;
-    bool _initialSyncFlag = false;
-
-    mutable stdx::mutex _minValidBoundariesMutex;
-    OpTime _appliedThrough;
-    OpTime _minValid;
+    mutable stdx::mutex _rbidMutex;
     int _rbid;
     bool _rbidInitialized = false;
-    Timestamp _oplogDeleteFromPoint;
 };
 
 }  // namespace repl
diff --git a/src/mongo/db/repl/sync_tail.cpp b/src/mongo/db/repl/sync_tail.cpp
index 68542125d4d..6dda8980def 100644
--- a/src/mongo/db/repl/sync_tail.cpp
+++ b/src/mongo/db/repl/sync_tail.cpp
@@ -62,6 +62,7 @@
 #include "mongo/db/repl/repl_client_info.h"
 #include "mongo/db/repl/repl_set_config.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
+#include "mongo/db/repl/replication_process.h"
 #include "mongo/db/repl/storage_interface.h"
 #include "mongo/db/server_parameters.h"
 #include "mongo/db/service_context.h"
@@ -656,7 +657,8 @@ void tryToGoLiveAsASecondary(OperationContext* opCtx, ReplicationCoordinator* re
     }
 
     // We can't go to SECONDARY until we reach minvalid.
-    if (replCoord->getMyLastAppliedOpTime() < StorageInterface::get(opCtx)->getMinValid(opCtx)) {
+    if (replCoord->getMyLastAppliedOpTime() <
+        ReplicationProcess::get(opCtx)->getConsistencyMarkers()->getMinValid(opCtx)) {
         return;
     }
 
@@ -817,8 +819,10 @@ void SyncTail::oplogApplication(ReplicationCoordinator* replCoord) {
         // Update various things that care about our last applied optime. Tests rely on 2 happening
         // before 3 even though it isn't strictly necessary. The order of 1 doesn't matter.
         setNewTimestamp(opCtx.getServiceContext(), lastOpTimeInBatch.getTimestamp());  // 1
-        StorageInterface::get(&opCtx)->setAppliedThrough(&opCtx, lastOpTimeInBatch);   // 2
-        finalizer->record(lastOpTimeInBatch);                                          // 3
+        ReplicationProcess::get(&opCtx)->getConsistencyMarkers()->setAppliedThrough(
+            &opCtx,
+            lastOpTimeInBatch);                // 2
+        finalizer->record(lastOpTimeInBatch);  // 3
     }
 }
 
@@ -1251,7 +1255,7 @@ StatusWith<OpTime> multiApply(OperationContext* opCtx,
         prefetchOps(ops, workerPool);
     }
 
-    auto storage = StorageInterface::get(opCtx);
+    auto consistencyMarkers = ReplicationProcess::get(opCtx)->getConsistencyMarkers();
 
     LOG(2) << "replication batch size is " << ops.size();
     // Stop all readers until we're done. This also prevents doc-locking engines from deleting old
@@ -1272,14 +1276,14 @@ StatusWith<OpTime> multiApply(OperationContext* opCtx,
         std::vector<MultiApplier::OperationPtrs> writerVectors(workerPool->getNumThreads());
         ON_BLOCK_EXIT([&] { workerPool->join(); });
 
-        storage->setOplogDeleteFromPoint(opCtx, ops.front().getTimestamp());
+        consistencyMarkers->setOplogDeleteFromPoint(opCtx, ops.front().getTimestamp());
         scheduleWritesToOplog(opCtx, workerPool, ops);
         fillWriterVectors(opCtx, &ops, &writerVectors);
 
         workerPool->join();
 
-        storage->setOplogDeleteFromPoint(opCtx, Timestamp());
-        storage->setMinValidToAtLeast(opCtx, ops.back().getOpTime());
+        consistencyMarkers->setOplogDeleteFromPoint(opCtx, Timestamp());
+        consistencyMarkers->setMinValidToAtLeast(opCtx, ops.back().getOpTime());
 
         applyOps(writerVectors, workerPool, applyOperation, &statusVector);
     }
diff --git a/src/mongo/db/repl/sync_tail_test.cpp b/src/mongo/db/repl/sync_tail_test.cpp
index 6b893f22e3e..3ecdd885777 100644
--- a/src/mongo/db/repl/sync_tail_test.cpp
+++ b/src/mongo/db/repl/sync_tail_test.cpp
@@ -48,8 +48,10 @@
 #include "mongo/db/repl/bgsync.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/oplog_interface_local.h"
+#include "mongo/db/repl/replication_consistency_markers_mock.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
+#include "mongo/db/repl/replication_process.h"
 #include "mongo/db/repl/storage_interface.h"
 #include "mongo/db/repl/storage_interface_mock.h"
 #include "mongo/db/repl/sync_tail.h"
@@ -75,6 +77,7 @@ protected:
     SyncTail::ApplyCommandInLockFn _applyCmd;
     SyncTail::IncrementOpsAppliedStatsFn _incOps;
     StorageInterfaceMock* _storageInterface = nullptr;
+    ReplicationProcess* _replicationProcess = nullptr;
 
     // Implements the MultiApplier::ApplyOperationFn interface and does nothing.
     static Status noopApplyOperationFn(MultiApplier::OperationPtrs*) {
@@ -118,6 +121,12 @@ void SyncTailTest::setUp() {
                                              const std::vector<BSONObj>&) { return Status::OK(); };
     StorageInterface::set(service, std::move(storageInterface));
 
+    _replicationProcess = new ReplicationProcess(
+        _storageInterface, stdx::make_unique<ReplicationConsistencyMarkersMock>());
+    ReplicationProcess::set(cc().getServiceContext(),
+                            std::unique_ptr<ReplicationProcess>(_replicationProcess));
+
+
     _opCtx = cc().makeOperationContext();
     _opsApplied = 0;
     _applyOp = [](OperationContext* opCtx,
@@ -133,6 +142,7 @@ void SyncTailTest::tearDown() {
     _opCtx.reset();
     ServiceContextMongoDTest::tearDown();
     _storageInterface = nullptr;
+    ReplicationProcess::set(getServiceContext(), {});
 }
 
 SyncTailWithLocalDocumentFetcher::SyncTailWithLocalDocumentFetcher(const BSONObj& document)
diff --git a/src/mongo/dbtests/SConscript b/src/mongo/dbtests/SConscript
index 64079e4c712..34a511adeba 100644
--- a/src/mongo/dbtests/SConscript
+++ b/src/mongo/dbtests/SConscript
@@ -123,6 +123,7 @@ dbtest = env.Program(
         "$BUILD_DIR/mongo/db/query/query_test_service_context",
         "$BUILD_DIR/mongo/db/repl/drop_pending_collection_reaper",
         "$BUILD_DIR/mongo/db/repl/repl_coordinator_global",
+        "$BUILD_DIR/mongo/db/repl/replication_consistency_markers_impl",
         "$BUILD_DIR/mongo/db/repl/replmocks",
         "$BUILD_DIR/mongo/db/serveronly",
         "$BUILD_DIR/mongo/db/logical_clock",
diff --git a/src/mongo/dbtests/replica_set_tests.cpp b/src/mongo/dbtests/replica_set_tests.cpp
index 7ef712e1b4b..91fa21f8800 100644
--- a/src/mongo/dbtests/replica_set_tests.cpp
+++ b/src/mongo/dbtests/replica_set_tests.cpp
@@ -32,7 +32,9 @@
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/repl/drop_pending_collection_reaper.h"
 #include "mongo/db/repl/last_vote.h"
+#include "mongo/db/repl/replication_consistency_markers_impl.h"
 #include "mongo/db/repl/replication_coordinator_external_state_impl.h"
+#include "mongo/db/repl/replication_process.h"
 #include "mongo/db/repl/storage_interface_mock.h"
 #include "mongo/db/service_context.h"
 #include "mongo/unittest/unittest.h"
@@ -51,10 +53,14 @@ protected:
         _storageInterface = stdx::make_unique<repl::StorageInterfaceMock>();
         _dropPendingCollectionReaper =
             stdx::make_unique<repl::DropPendingCollectionReaper>(_storageInterface.get());
+        _replicationProcess = stdx::make_unique<repl::ReplicationProcess>(
+            _storageInterface.get(),
+            stdx::make_unique<repl::ReplicationConsistencyMarkersImpl>(_storageInterface.get()));
         _replCoordExternalState = stdx::make_unique<repl::ReplicationCoordinatorExternalStateImpl>(
             opCtx->getServiceContext(),
             _dropPendingCollectionReaper.get(),
-            _storageInterface.get());
+            _storageInterface.get(),
+            _replicationProcess.get());
     }
 
     void tearDown() {
@@ -65,6 +71,7 @@ protected:
         _replCoordExternalState.reset();
         _dropPendingCollectionReaper.reset();
         _storageInterface.reset();
+        _replicationProcess.reset();
     }
 
     repl::ReplicationCoordinatorExternalStateImpl* getReplCoordExternalState() {
@@ -79,6 +86,7 @@ private:
     std::unique_ptr<repl::ReplicationCoordinatorExternalStateImpl> _replCoordExternalState;
     std::unique_ptr<repl::StorageInterface> _storageInterface;
     std::unique_ptr<repl::DropPendingCollectionReaper> _dropPendingCollectionReaper;
+    std::unique_ptr<repl::ReplicationProcess> _replicationProcess;
 };
 
 TEST_F(ReplicaSetTest, ReplCoordExternalStateStoresLastVoteWithNewTerm) {
diff --git a/src/mongo/s/sharding_mongod_test_fixture.cpp b/src/mongo/s/sharding_mongod_test_fixture.cpp
index 186d6fdd903..655ff7c1efb 100644
--- a/src/mongo/s/sharding_mongod_test_fixture.cpp
+++ b/src/mongo/s/sharding_mongod_test_fixture.cpp
@@ -46,6 +46,7 @@
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/read_concern_args.h"
 #include "mongo/db/repl/repl_settings.h"
+#include "mongo/db/repl/replication_consistency_markers_mock.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/repl/replication_process.h"
@@ -127,8 +128,10 @@ void ShardingMongodTestFixture::setUp() {
 
     auto storagePtr = stdx::make_unique<repl::StorageInterfaceMock>();
 
-    repl::ReplicationProcess::set(service,
-                                  stdx::make_unique<repl::ReplicationProcess>(storagePtr.get()));
+    repl::ReplicationProcess::set(
+        service,
+        stdx::make_unique<repl::ReplicationProcess>(
+            storagePtr.get(), stdx::make_unique<repl::ReplicationConsistencyMarkersMock>()));
     repl::ReplicationProcess::get(_opCtx.get())->initializeRollbackID(_opCtx.get());
 
     repl::StorageInterface::set(service, std::move(storagePtr));