Revert SERVER-42996 Move ApplierState to OplogApplier

Author: mtandrei

src/mongo/db/repl/SConscript

@@ -722,7 +722,6 @@ 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,16 +287,6 @@ 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,65 +190,6 @@ 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:
     /**
@@ -287,8 +228,6 @@ class OplogApplier {
 
     // Configures this OplogApplier.
     const Options _options;
-
-    ApplierState _applierState = ApplierState::Running;
 };
 
 /**

src/mongo/db/repl/oplog_applier_impl.cpp

@@ -557,7 +557,10 @@ StatusWith<OpTime> OplogApplierImpl::_multiApply(OperationContext* opCtx,
     // entries from the oplog until we finish writing.
     Lock::ParallelBatchWriterMode pbwm(opCtx->lockState());
 
-    if (getApplierState() == ApplierState::Stopped) {
+    // 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) {
         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 =
-                _oplogApplier->getApplierState() == OplogApplier::ApplierState::Draining;
+                replCoord->getApplierState() == ReplicationCoordinator::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,6 +483,64 @@ 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,7 +35,6 @@
 
 #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"
@@ -301,16 +300,6 @@ 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,15 +910,6 @@ 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,8 +108,6 @@ 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,31 +46,7 @@
 
 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"),
@@ -81,9 +57,7 @@ ReplicationCoordinatorExternalStateMock::ReplicationCoordinatorExternalStateMock
       _storeLocalLastVoteDocumentStatus(Status::OK()),
       _storeLocalLastVoteDocumentShouldHang(false),
       _connectionsClosed(false),
-      _threadsStarted(false),
-      _oplogApplier(std::make_unique<OplogApplierMock>(
-          repl::OplogApplier::Options(repl::OplogApplication::Mode::kSecondary))) {}
+      _threadsStarted(false) {}
 
 ReplicationCoordinatorExternalStateMock::~ReplicationCoordinatorExternalStateMock() {}
 
@@ -319,14 +293,6 @@ 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) {