SERVER-46995 move index completion logic from IndexBuildsCoordinator into ReplIndexBuildState

5 files changed, 133 insertions, 108 deletions

diff --git a/src/mongo/db/index_builds_coordinator.cpp b/src/mongo/db/index_builds_coordinator.cpp
index 76bf934e79e..f220b3a18a3 100644
--- a/src/mongo/db/index_builds_coordinator.cpp
+++ b/src/mongo/db/index_builds_coordinator.cpp
@@ -159,21 +159,6 @@ bool shouldBuildIndexesOnEmptyCollectionSinglePhased(OperationContext* opCtx,
     return collection->isEmpty(opCtx);
 }
 
-/*
- * Determines whether to skip the index build state transition check.
- * Index builder not using ReplIndexBuildState::waitForNextAction to signal primary and secondaries
- * to commit or abort signal will violate index build state transition. So, we should skip state
- * transition verification. Otherwise, we would invariant.
- */
-bool shouldSkipIndexBuildStateTransitionCheck(OperationContext* opCtx,
-                                              IndexBuildProtocol protocol) {
-    const auto replCoord = repl::ReplicationCoordinator::get(opCtx);
-    if (replCoord->getSettings().usingReplSets() && protocol == IndexBuildProtocol::kTwoPhase) {
-        return false;
-    }
-    return true;
-}
-
 /**
  * Removes the index build from the config.system.indexBuilds collection after the primary has
  * written the commitIndexBuild or abortIndexBuild oplog entry.
@@ -214,11 +199,8 @@ void onCommitIndexBuild(OperationContext* opCtx,
                         std::shared_ptr<ReplIndexBuildState> replState) {
     const auto& buildUUID = replState->buildUUID;
 
-    auto skipCheck = shouldSkipIndexBuildStateTransitionCheck(opCtx, replState->protocol);
-    opCtx->recoveryUnit()->onCommit([replState, skipCheck](boost::optional<Timestamp> commitTime) {
-        stdx::unique_lock<Latch> lk(replState->mutex);
-        replState->indexBuildState.setState(IndexBuildState::kCommitted, skipCheck);
-    });
+    replState->commit(opCtx);
+
     if (IndexBuildProtocol::kSinglePhase == replState->protocol) {
         return;
     }
@@ -747,27 +729,6 @@ Status IndexBuildsCoordinator::_setUpResumeIndexBuild(OperationContext* opCtx,
     return status;
 }
 
-std::string IndexBuildsCoordinator::_indexBuildActionToString(IndexBuildAction action) {
-    if (action == IndexBuildAction::kNoAction) {
-        return "No action";
-    } else if (action == IndexBuildAction::kOplogCommit) {
-        return "Oplog commit";
-    } else if (action == IndexBuildAction::kOplogAbort) {
-        return "Oplog abort";
-    } else if (action == IndexBuildAction::kInitialSyncAbort) {
-        return "Initial sync abort";
-    } else if (action == IndexBuildAction::kRollbackAbort) {
-        return "Rollback abort";
-    } else if (action == IndexBuildAction::kPrimaryAbort) {
-        return "Primary abort";
-    } else if (action == IndexBuildAction::kSinglePhaseCommit) {
-        return "Single-phase commit";
-    } else if (action == IndexBuildAction::kCommitQuorumSatisfied) {
-        return "Commit quorum Satisfied";
-    }
-    MONGO_UNREACHABLE;
-}
-
 void IndexBuildsCoordinator::waitForAllIndexBuildsToStopForShutdown(OperationContext* opCtx) {
     stdx::unique_lock<Latch> lk(_mutex);
 
@@ -1300,24 +1261,7 @@ void IndexBuildsCoordinator::_completeAbort(OperationContext* opCtx,
         // Deletes the index from the durable catalog.
         case IndexBuildAction::kOplogAbort: {
             invariant(IndexBuildProtocol::kTwoPhase == replState->protocol);
-            // This signal can be received during primary (drain phase), secondary,
-            // startup (startup recovery) and startup2 (initial sync).
-            bool isMaster = replCoord->canAcceptWritesFor(opCtx, nss);
-            invariant(!isMaster, str::stream() << "Index build: " << replState->buildUUID);
-            invariant(replState->indexBuildState.isAborted(),
-                      str::stream()
-                          << "Index build: " << replState->buildUUID
-                          << ",  index build state: " << replState->indexBuildState.toString());
-            invariant(replState->indexBuildState.getTimestamp() &&
-                          replState->indexBuildState.getAbortReason(),
-                      replState->buildUUID.toString());
-            LOGV2(3856206,
-                  "Aborting index build from oplog entry",
-                  "buildUUID"_attr = replState->buildUUID,
-                  "abortTimestamp"_attr = replState->indexBuildState.getTimestamp().get(),
-                  "abortReason"_attr = replState->indexBuildState.getAbortReason().get(),
-                  "collectionUUID"_attr = replState->collectionUUID);
-
+            replState->onOplogAbort(opCtx, nss);
             _indexBuildsManager.abortIndexBuild(
                 opCtx, coll, replState->buildUUID, MultiIndexBlock::kNoopOnCleanUpFn);
             break;
@@ -1342,11 +1286,7 @@ void IndexBuildsCoordinator::_completeSelfAbort(OperationContext* opCtx,
                                                 std::shared_ptr<ReplIndexBuildState> replState,
                                                 Status reason) {
     _completeAbort(opCtx, replState, IndexBuildAction::kPrimaryAbort, reason);
-    {
-        auto skipCheck = shouldSkipIndexBuildStateTransitionCheck(opCtx, replState->protocol);
-        stdx::unique_lock<Latch> lk(replState->mutex);
-        replState->indexBuildState.setState(IndexBuildState::kAborted, skipCheck);
-    }
+    replState->abortSelf(opCtx);
     {
         stdx::unique_lock<Latch> lk(_mutex);
         _unregisterIndexBuild(lk, replState);
@@ -1361,15 +1301,8 @@ void IndexBuildsCoordinator::_completeAbortForShutdown(
     _indexBuildsManager.abortIndexBuildWithoutCleanup(
         opCtx, collection, replState->buildUUID, replState->isResumable());
 
-    {
-        // Promise should be set at least once before it's getting destroyed.
-        stdx::unique_lock<Latch> lk(replState->mutex);
-        if (!replState->waitForNextAction->getFuture().isReady()) {
-            replState->waitForNextAction->emplaceValue(IndexBuildAction::kNoAction);
-        }
-        auto skipCheck = shouldSkipIndexBuildStateTransitionCheck(opCtx, replState->protocol);
-        replState->indexBuildState.setState(IndexBuildState::kAborted, skipCheck);
-    }
+    replState->abortForShutdown(opCtx);
+
     {
         // This allows the builder thread to exit.
         stdx::unique_lock<Latch> lk(_mutex);
@@ -2688,12 +2621,7 @@ IndexBuildsCoordinator::CommitResult IndexBuildsCoordinator::_insertKeysFromSide
     }
 
     if (IndexBuildAction::kOplogCommit == action) {
-        // This signal can be received during primary (drain phase), secondary, startup (startup
-        // recovery) and startup2 (initial sync).
-        invariant(!isMaster && replState->indexBuildState.isCommitPrepared(),
-                  str::stream() << "Index build: " << replState->buildUUID
-                                << ",  index build state: "
-                                << replState->indexBuildState.toString());
+        replState->onOplogCommit(isMaster);
     }
 
     // The collection object should always exist while an index build is registered.
diff --git a/src/mongo/db/index_builds_coordinator.h b/src/mongo/db/index_builds_coordinator.h
index d4156ab81d0..2c6d340b0c8 100644
--- a/src/mongo/db/index_builds_coordinator.h
+++ b/src/mongo/db/index_builds_coordinator.h
@@ -710,9 +710,6 @@ protected:
         std::shared_ptr<ReplIndexBuildState> replState,
         const IndexBuildOptions& indexBuildOptions) = 0;
 
-    std::string _indexBuildActionToString(IndexBuildAction action);
-
-
     /**
      * Third phase is catching up on all the writes that occurred during the first two phases.
      * Accepts a commit timestamp for the index, which could be null. See
diff --git a/src/mongo/db/index_builds_coordinator_mongod.cpp b/src/mongo/db/index_builds_coordinator_mongod.cpp
index 913234d665a..002bacc131a 100644
--- a/src/mongo/db/index_builds_coordinator_mongod.cpp
+++ b/src/mongo/db/index_builds_coordinator_mongod.cpp
@@ -450,22 +450,7 @@ bool IndexBuildsCoordinatorMongod::_signalIfCommitQuorumNotEnabled(
     auto replCoord = repl::ReplicationCoordinator::get(opCtx);
 
     if (IndexBuildProtocol::kSinglePhase == replState->protocol) {
-        // Single-phase builds don't support commit quorum, but they must go through the process of
-        // updating their state to synchronize with concurrent abort operations.
-        stdx::unique_lock<Latch> lk(replState->mutex);
-        if (replState->waitForNextAction->getFuture().isReady()) {
-            // If the signal action has been set, it should only be because a concurrent operation
-            // already aborted the index build.
-            auto action = replState->waitForNextAction->getFuture().get(opCtx);
-            invariant(action == IndexBuildAction::kPrimaryAbort,
-                      str::stream() << "action: " << _indexBuildActionToString(action)
-                                    << ", buildUUID: " << replState->buildUUID);
-            LOGV2(4639700,
-                  "Not committing single-phase build because it has already been aborted",
-                  "buildUUID"_attr = replState->buildUUID);
-            return true;
-        }
-        replState->waitForNextAction->emplaceValue(IndexBuildAction::kSinglePhaseCommit);
+        replState->setSinglePhaseCommit(opCtx);
         return true;
     }
 
@@ -666,7 +651,7 @@ void IndexBuildsCoordinatorMongod::_waitForNextIndexBuildActionAndCommit(
         LOGV2(3856204,
               "Index build: received signal",
               "buildUUID"_attr = replState->buildUUID,
-              "action"_attr = _indexBuildActionToString(nextAction));
+              "action"_attr = indexBuildActionToString(nextAction));
 
         // If the index build was aborted, this serves as a final interruption point. Since the
         // index builder thread is interrupted before the action is set, this must fail if the build
@@ -675,24 +660,29 @@ void IndexBuildsCoordinatorMongod::_waitForNextIndexBuildActionAndCommit(
 
         bool needsToRetryWait = false;
 
+        auto commitTimestamp = replState->getCommitTimestamp();
+
         switch (nextAction) {
             case IndexBuildAction::kOplogCommit: {
                 invariant(replState->protocol == IndexBuildProtocol::kTwoPhase);
-                invariant(replState->indexBuildState.getTimestamp(),
-                          replState->buildUUID.toString());
+                invariant(!commitTimestamp.isNull(), replState->buildUUID.toString());
                 LOGV2(3856205,
                       "Index build: committing from oplog entry",
                       "buildUUID"_attr = replState->buildUUID,
-                      "commitTimestamp"_attr = replState->indexBuildState.getTimestamp().get(),
+                      "commitTimestamp"_attr = commitTimestamp,
                       "collectionUUID"_attr = replState->collectionUUID);
                 break;
             }
             case IndexBuildAction::kCommitQuorumSatisfied: {
-                invariant(!replState->indexBuildState.getTimestamp());
+                invariant(commitTimestamp.isNull(),
+                          str::stream() << "commit ts: " << commitTimestamp.toString()
+                                        << "; index build: " << replState->buildUUID.toString());
                 break;
             }
             case IndexBuildAction::kSinglePhaseCommit:
-                invariant(replState->protocol == IndexBuildProtocol::kSinglePhase);
+                invariant(replState->protocol == IndexBuildProtocol::kSinglePhase,
+                          str::stream() << "commit ts: " << commitTimestamp.toString()
+                                        << "; index build: " << replState->buildUUID.toString());
                 break;
             case IndexBuildAction::kOplogAbort:
             case IndexBuildAction::kRollbackAbort:
@@ -704,10 +694,6 @@ void IndexBuildsCoordinatorMongod::_waitForNextIndexBuildActionAndCommit(
                 return;
         }
 
-        Timestamp commitTimestamp = replState->indexBuildState.getTimestamp()
-            ? replState->indexBuildState.getTimestamp().get()
-            : Timestamp();
-
         auto result = _insertKeysFromSideTablesAndCommit(
             opCtx, replState, nextAction, indexBuildOptions, commitTimestamp);
         switch (result) {
diff --git a/src/mongo/db/repl_index_build_state.cpp b/src/mongo/db/repl_index_build_state.cpp
index c1fda23d4dc..f297758121b 100644
--- a/src/mongo/db/repl_index_build_state.cpp
+++ b/src/mongo/db/repl_index_build_state.cpp
@@ -134,6 +134,61 @@ void ReplIndexBuildState::start(OperationContext* opCtx) {
     indexBuildState.setState(IndexBuildState::kInProgress, false /* skipCheck */);
 }
 
+void ReplIndexBuildState::commit(OperationContext* opCtx) {
+    auto skipCheck = _shouldSkipIndexBuildStateTransitionCheck(opCtx);
+    opCtx->recoveryUnit()->onCommit([this, skipCheck](boost::optional<Timestamp> commitTime) {
+        stdx::unique_lock<Latch> lk(mutex);
+        indexBuildState.setState(IndexBuildState::kCommitted, skipCheck);
+    });
+}
+
+Timestamp ReplIndexBuildState::getCommitTimestamp() const {
+    stdx::unique_lock<Latch> lk(mutex);
+    return indexBuildState.getTimestamp().value_or(Timestamp());
+}
+
+void ReplIndexBuildState::onOplogCommit(bool isPrimary) const {
+    stdx::unique_lock<Latch> lk(mutex);
+    invariant(!isPrimary && indexBuildState.isCommitPrepared(),
+              str::stream() << "Index build: " << buildUUID
+                            << ",  index build state: " << indexBuildState.toString());
+}
+
+void ReplIndexBuildState::abortSelf(OperationContext* opCtx) {
+    auto skipCheck = _shouldSkipIndexBuildStateTransitionCheck(opCtx);
+    stdx::unique_lock<Latch> lk(mutex);
+    indexBuildState.setState(IndexBuildState::kAborted, skipCheck);
+}
+
+void ReplIndexBuildState::abortForShutdown(OperationContext* opCtx) {
+    // Promise should be set at least once before it's getting destroyed.
+    stdx::unique_lock<Latch> lk(mutex);
+    if (!waitForNextAction->getFuture().isReady()) {
+        waitForNextAction->emplaceValue(IndexBuildAction::kNoAction);
+    }
+    auto skipCheck = _shouldSkipIndexBuildStateTransitionCheck(opCtx);
+    indexBuildState.setState(IndexBuildState::kAborted, skipCheck);
+}
+
+void ReplIndexBuildState::onOplogAbort(OperationContext* opCtx, const NamespaceString& nss) const {
+    auto replCoord = repl::ReplicationCoordinator::get(opCtx);
+    bool isPrimary = replCoord->canAcceptWritesFor(opCtx, nss);
+    invariant(!isPrimary, str::stream() << "Index build: " << buildUUID);
+
+    stdx::unique_lock<Latch> lk(mutex);
+    invariant(indexBuildState.isAborted(),
+              str::stream() << "Index build: " << buildUUID
+                            << ",  index build state: " << indexBuildState.toString());
+    invariant(indexBuildState.getTimestamp() && indexBuildState.getAbortReason(),
+              buildUUID.toString());
+    LOGV2(3856206,
+          "Aborting index build from oplog entry",
+          "buildUUID"_attr = buildUUID,
+          "abortTimestamp"_attr = indexBuildState.getTimestamp().get(),
+          "abortReason"_attr = indexBuildState.getAbortReason().get(),
+          "collectionUUID"_attr = collectionUUID);
+}
+
 bool ReplIndexBuildState::isAborted() const {
     stdx::unique_lock<Latch> lk(mutex);
     return indexBuildState.isAborted();
@@ -172,6 +227,23 @@ void ReplIndexBuildState::setCommitQuorumSatisfied(OperationContext* opCtx) {
     }
 }
 
+void ReplIndexBuildState::setSinglePhaseCommit(OperationContext* opCtx) {
+    stdx::unique_lock<Latch> lk(mutex);
+    if (waitForNextAction->getFuture().isReady()) {
+        // If the signal action has been set, it should only be because a concurrent operation
+        // already aborted the index build.
+        auto action = waitForNextAction->getFuture().get(opCtx);
+        invariant(action == IndexBuildAction::kPrimaryAbort,
+                  str::stream() << "action: " << indexBuildActionToString(action)
+                                << ", buildUUID: " << buildUUID);
+        LOGV2(4639700,
+              "Not committing single-phase build because it has already been aborted",
+              "buildUUID"_attr = buildUUID);
+        return;
+    }
+    waitForNextAction->emplaceValue(IndexBuildAction::kSinglePhaseCommit);
+}
+
 bool ReplIndexBuildState::tryCommit(OperationContext* opCtx) {
     stdx::unique_lock<Latch> lk(mutex);
     if (indexBuildState.isSettingUp()) {
diff --git a/src/mongo/db/repl_index_build_state.h b/src/mongo/db/repl_index_build_state.h
index acc1c28c377..66fe15d13f9 100644
--- a/src/mongo/db/repl_index_build_state.h
+++ b/src/mongo/db/repl_index_build_state.h
@@ -238,6 +238,41 @@ public:
     void start(OperationContext* opCtx);
 
     /**
+     * This index build has completed successfully and there is no further work to be done.
+     */
+    void commit(OperationContext* opCtx);
+
+    /**
+     * Returns timestamp for committing this index build.
+     * Returns null timestamp if not set.
+     */
+    Timestamp getCommitTimestamp() const;
+
+    /**
+     * Called when handling a commitIndexIndexBuild oplog entry.
+     * This signal can be received during primary (drain phase), secondary,
+     * startup (startup recovery) and startup2 (initial sync).
+     */
+    void onOplogCommit(bool isPrimary) const;
+
+    /**
+     * This index build has failed while running in the builder thread due to a non-shutdown reason.
+     */
+    void abortSelf(OperationContext* opCtx);
+
+    /**
+     * This index build was interrupted because the server is shutting down.
+     */
+    void abortForShutdown(OperationContext* opCtx);
+
+    /**
+     * Called when handling an abortIndexIndexBuild oplog entry.
+     * This signal can be received during primary (drain phase), secondary,
+     * startup (startup recovery) and startup2 (initial sync).
+     */
+    void onOplogAbort(OperationContext* opCtx, const NamespaceString& nss) const;
+
+    /**
      * Returns true if this index build has been aborted.
      */
     bool isAborted() const;
@@ -255,6 +290,13 @@ public:
     void setCommitQuorumSatisfied(OperationContext* opCtx);
 
     /**
+     * Called when we are about to complete a single-phased index build.
+     * Single-phase builds don't support commit quorum, but they must go through the process of
+     * updating their state to synchronize with concurrent abort operations
+     */
+    void setSinglePhaseCommit(OperationContext* opCtx);
+
+    /**
      * Attempt to signal the index build to commit and advance the index build to the kPrepareCommit
      * state.
      * Returns true if successful and false if the attempt was unnecessful and the caller should