SERVER-42996 Move ApplierState to OplogApplier

Author: mtandrei

src/mongo/db/repl/SConscript

@@ -722,6 +722,7 @@ env.Library(
         'storage_interface_mock.cpp',
     ],
     LIBDEPS=[
+        'oplog_application_interface',
         'oplog_buffer_blocking_queue',
         'repl_coordinator_interface',
         'repl_settings',

src/mongo/db/repl/oplog_applier.cpp

@@ -287,6 +287,16 @@ void OplogApplier::_consume(OperationContext* opCtx, OplogBuffer* oplogBuffer) {
     invariant(oplogBuffer->tryPop(opCtx, &opToPopAndDiscard) || inShutdown());
 }
 
+OplogApplier::ApplierState OplogApplier::getApplierState() const {
+    stdx::lock_guard<Latch> lock(_mutex);
+    return _applierState;
+}
+
+void OplogApplier::setApplierState(ApplierState st) {
+    stdx::lock_guard<Latch> lock(_mutex);
+    _applierState = st;
+}
+
 std::unique_ptr<ThreadPool> makeReplWriterPool() {
     return makeReplWriterPool(replWriterThreadCount);
 }

src/mongo/db/repl/oplog_applier.h

@@ -190,6 +190,65 @@ class OplogApplier {
     StatusWith<OpTime> multiApply(OperationContext* opCtx, Operations ops);
 
     const Options& getOptions() const;
+    /**
+     * Step-up
+     * =======
+     * On stepup, repl coord enters catch-up mode. It's the same as the secondary mode from
+     * the perspective of producer and applier, so there's nothing to do with them.
+     * When a node enters drain mode, producer state = Stopped, applier state = Draining.
+     *
+     * If the applier state is Draining, it will signal repl coord when there's nothing to apply.
+     * The applier goes into Stopped state at the same time.
+     *
+     * The states go like the following:
+     * - secondary and during catchup mode
+     * (producer: Running, applier: Running)
+     *      |
+     *      | finish catch-up, enter drain mode
+     *      V
+     * - drain mode
+     * (producer: Stopped, applier: Draining)
+     *      |
+     *      | applier signals drain is complete
+     *      V
+     * - primary is in master mode
+     * (producer: Stopped, applier: Stopped)
+     *
+     *
+     * Step-down
+     * =========
+     * The state transitions become:
+     * - primary is in master mode
+     * (producer: Stopped, applier: Stopped)
+     *      |
+     *      | step down
+     *      V
+     * - secondary mode, starting bgsync
+     * (producer: Starting, applier: Running)
+     *      |
+     *      | bgsync runs start()
+     *      V
+     * - secondary mode, normal
+     * (producer: Running, applier: Running)
+     *
+     * When a node steps down during draining mode, it's OK to change from (producer: Stopped,
+     * applier: Draining) to (producer: Starting, applier: Running).
+     *
+     * When a node steps down during catchup mode, the states remain the same (producer: Running,
+     * applier: Running).
+     */
+    enum class ApplierState { Running, Draining, Stopped };
+
+    /**
+     * In normal cases: Running -> Draining -> Stopped -> Running.
+     * Draining -> Running is also possible if a node steps down during drain mode.
+     *
+     * Only the applier can make the transition from Draining to Stopped by calling
+     * signalDrainComplete().
+     */
+    virtual ApplierState getApplierState() const;
+
+    virtual void setApplierState(ApplierState st);
 
 private:
     /**
@@ -228,6 +287,8 @@ class OplogApplier {
 
     // Configures this OplogApplier.
     const Options _options;
+
+    ApplierState _applierState = ApplierState::Running;
 };
 
 /**

src/mongo/db/repl/oplog_applier_impl.cpp

@@ -557,10 +557,7 @@ StatusWith<OpTime> OplogApplierImpl::_multiApply(OperationContext* opCtx,
     // entries from the oplog until we finish writing.
     Lock::ParallelBatchWriterMode pbwm(opCtx->lockState());
 
-    // TODO (SERVER-42996): This is a temporary invariant to protect against segfaults. This will
-    // be removed once ApplierState is moved from ReplicationCoordinator to OplogApplier.
-    invariant(_replCoord);
-    if (_replCoord->getApplierState() == ReplicationCoordinator::ApplierState::Stopped) {
+    if (getApplierState() == ApplierState::Stopped) {
         severe() << "attempting to replicate ops while primary";
         return {ErrorCodes::CannotApplyOplogWhilePrimary,
                 "attempting to replicate ops while primary"};

src/mongo/db/repl/opqueue_batcher.cpp

@@ -154,7 +154,7 @@ void OpQueueBatcher::run() {
             // Draining state guarantees the producer has already been fully stopped and no more
             // operations will be pushed in to the oplog buffer until the applier state changes.
             auto isDraining =
-                replCoord->getApplierState() == ReplicationCoordinator::ApplierState::Draining;
+                _oplogApplier->getApplierState() == OplogApplier::ApplierState::Draining;
             // Check the oplog buffer after the applier state to ensure the producer is stopped.
             if (isDraining && _oplogBuffer->isEmpty()) {
                 ops.setTermWhenExhausted(termWhenBufferIsEmpty);

src/mongo/db/repl/replication_coordinator.h

@@ -483,64 +483,6 @@ class ReplicationCoordinator : public SyncSourceSelector {
      */
     virtual Status setFollowerModeStrict(OperationContext* opCtx, const MemberState& newState) = 0;
 
-    /**
-     * Step-up
-     * =======
-     * On stepup, repl coord enters catch-up mode. It's the same as the secondary mode from
-     * the perspective of producer and applier, so there's nothing to do with them.
-     * When a node enters drain mode, producer state = Stopped, applier state = Draining.
-     *
-     * If the applier state is Draining, it will signal repl coord when there's nothing to apply.
-     * The applier goes into Stopped state at the same time.
-     *
-     * The states go like the following:
-     * - secondary and during catchup mode
-     * (producer: Running, applier: Running)
-     *      |
-     *      | finish catch-up, enter drain mode
-     *      V
-     * - drain mode
-     * (producer: Stopped, applier: Draining)
-     *      |
-     *      | applier signals drain is complete
-     *      V
-     * - primary is in master mode
-     * (producer: Stopped, applier: Stopped)
-     *
-     *
-     * Step-down
-     * =========
-     * The state transitions become:
-     * - primary is in master mode
-     * (producer: Stopped, applier: Stopped)
-     *      |
-     *      | step down
-     *      V
-     * - secondary mode, starting bgsync
-     * (producer: Starting, applier: Running)
-     *      |
-     *      | bgsync runs start()
-     *      V
-     * - secondary mode, normal
-     * (producer: Running, applier: Running)
-     *
-     * When a node steps down during draining mode, it's OK to change from (producer: Stopped,
-     * applier: Draining) to (producer: Starting, applier: Running).
-     *
-     * When a node steps down during catchup mode, the states remain the same (producer: Running,
-     * applier: Running).
-     */
-    enum class ApplierState { Running, Draining, Stopped };
-
-    /**
-     * In normal cases: Running -> Draining -> Stopped -> Running.
-     * Draining -> Running is also possible if a node steps down during drain mode.
-     *
-     * Only the applier can make the transition from Draining to Stopped by calling
-     * signalDrainComplete().
-     */
-    virtual ApplierState getApplierState() = 0;
-
     /**
      * Signals that a previously requested pause and drain of the applier buffer
      * has completed.

src/mongo/db/repl/replication_coordinator_external_state.h

@@ -35,6 +35,7 @@
 
 #include "mongo/bson/timestamp.h"
 #include "mongo/db/repl/member_state.h"
+#include "mongo/db/repl/oplog_applier.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/executor/task_executor.h"
 #include "mongo/util/concurrency/thread_pool.h"
@@ -300,6 +301,16 @@ class ReplicationCoordinatorExternalState {
      */
     virtual std::size_t getOplogFetcherInitialSyncMaxFetcherRestarts() const = 0;
 
+    /*
+     * Returns the OplogApplier's current state.
+     */
+    virtual OplogApplier::ApplierState getApplierState() const = 0;
+
+    /*
+     * Updates the OplogApplier's current state.
+     */
+    virtual void setApplierState(const OplogApplier::ApplierState st) = 0;
+
     /*
      * Creates noop writer instance. Setting the _noopWriter member is not protected by a guard,
      * hence it must be called before multi-threaded operations start.

src/mongo/db/repl/replication_coordinator_external_state_impl.cpp

@@ -910,6 +910,15 @@ std::size_t ReplicationCoordinatorExternalStateImpl::getOplogFetcherInitialSyncM
     return oplogFetcherInitialSyncMaxFetcherRestarts.load();
 }
 
+
+OplogApplier::ApplierState ReplicationCoordinatorExternalStateImpl::getApplierState() const {
+    return _oplogApplier.get()->getApplierState();
+}
+
+void ReplicationCoordinatorExternalStateImpl::setApplierState(const OplogApplier::ApplierState st) {
+    _oplogApplier.get()->setApplierState(st);
+}
+
 JournalListener::Token ReplicationCoordinatorExternalStateImpl::getToken() {
     return repl::ReplicationCoordinator::get(_service)->getMyLastAppliedOpTimeAndWallTime();
 }

src/mongo/db/repl/replication_coordinator_external_state_impl.h

@@ -108,6 +108,8 @@ class ReplicationCoordinatorExternalStateImpl final : public ReplicationCoordina
     virtual bool isReadConcernSnapshotSupportedByStorageEngine(OperationContext* opCtx) const;
     virtual std::size_t getOplogFetcherSteadyStateMaxFetcherRestarts() const override;
     virtual std::size_t getOplogFetcherInitialSyncMaxFetcherRestarts() const override;
+    virtual OplogApplier::ApplierState getApplierState() const override;
+    virtual void setApplierState(const OplogApplier::ApplierState st) override;
 
     // Methods from JournalListener.
     virtual JournalListener::Token getToken();

src/mongo/db/repl/replication_coordinator_external_state_mock.cpp

@@ -46,7 +46,31 @@
 
 namespace mongo {
 namespace repl {
+namespace {
+/**
+ * Minimal implementation of OplogApplier for testing.
+ */
+class OplogApplierMock : public OplogApplier {
+    OplogApplierMock(const OplogApplierMock&) = delete;
+    OplogApplierMock& operator=(const OplogApplierMock&) = delete;
+
+public:
+    explicit OplogApplierMock(OplogApplier::Options options);
+
+    void _run(OplogBuffer* oplogBuffer) final;
+    StatusWith<OpTime> _multiApply(OperationContext* opCtx, Operations ops) final;
+};
+
+OplogApplierMock::OplogApplierMock(OplogApplier::Options options)
+    : OplogApplier(nullptr, nullptr, nullptr, options) {}
+
+void OplogApplierMock::_run(OplogBuffer* oplogBuffer) {}
 
+StatusWith<OpTime> OplogApplierMock::_multiApply(OperationContext* opCtx, Operations ops) {
+    return OpTime();
+}
+
+}  // namespace
 ReplicationCoordinatorExternalStateMock::ReplicationCoordinatorExternalStateMock()
     : _localRsConfigDocument(ErrorCodes::NoMatchingDocument, "No local config document"),
       _localRsLastVoteDocument(ErrorCodes::NoMatchingDocument, "No local lastVote document"),
@@ -57,7 +81,9 @@ ReplicationCoordinatorExternalStateMock::ReplicationCoordinatorExternalStateMock
       _storeLocalLastVoteDocumentStatus(Status::OK()),
       _storeLocalLastVoteDocumentShouldHang(false),
       _connectionsClosed(false),
-      _threadsStarted(false) {}
+      _threadsStarted(false),
+      _oplogApplier(std::make_unique<OplogApplierMock>(
+          repl::OplogApplier::Options(repl::OplogApplication::Mode::kSecondary))) {}
 
 ReplicationCoordinatorExternalStateMock::~ReplicationCoordinatorExternalStateMock() {}
 
@@ -293,6 +319,14 @@ void ReplicationCoordinatorExternalStateMock::setIsReadCommittedEnabled(bool val
     _isReadCommittedSupported = val;
 }
 
+void ReplicationCoordinatorExternalStateMock::setApplierState(const OplogApplier::ApplierState st) {
+    _oplogApplier->setApplierState(st);
+}
+
+OplogApplier::ApplierState ReplicationCoordinatorExternalStateMock::getApplierState() const {
+    return _oplogApplier->getApplierState();
+}
+
 void ReplicationCoordinatorExternalStateMock::onDrainComplete(OperationContext* opCtx) {}
 
 OpTime ReplicationCoordinatorExternalStateMock::onTransitionToPrimary(OperationContext* opCtx) {