SERVER-39068 Replicate startIndexBuild and commitIndexBuild oplog entrires

21 files changed, 336 insertions, 155 deletions

diff --git a/src/mongo/db/catalog/collection_compact.cpp b/src/mongo/db/catalog/collection_compact.cpp
index cb5e854c11a..257c7a1f0b8 100644
--- a/src/mongo/db/catalog/collection_compact.cpp
+++ b/src/mongo/db/catalog/collection_compact.cpp
@@ -120,7 +120,7 @@ StatusWith<CompactStats> compactCollection(OperationContext* opCtx,
     MultiIndexBlock indexer(opCtx, collection);
     indexer.ignoreUniqueConstraint();  // in compact we should be doing no checking
 
-    Status status = indexer.init(indexSpecs).getStatus();
+    Status status = indexer.init(indexSpecs, MultiIndexBlock::kNoopOnInitFn).getStatus();
     if (!status.isOK())
         return StatusWith<CompactStats>(status);
 
@@ -135,7 +135,8 @@ StatusWith<CompactStats> compactCollection(OperationContext* opCtx,
 
     {
         WriteUnitOfWork wunit(opCtx);
-        status = indexer.commit();
+        status =
+            indexer.commit(MultiIndexBlock::kNoopOnCreateEachFn, MultiIndexBlock::kNoopOnCommitFn);
         if (!status.isOK()) {
             return StatusWith<CompactStats>(status);
         }
diff --git a/src/mongo/db/catalog/index_builds_manager.cpp b/src/mongo/db/catalog/index_builds_manager.cpp
index 4ecb47683e2..e14a03577cd 100644
--- a/src/mongo/db/catalog/index_builds_manager.cpp
+++ b/src/mongo/db/catalog/index_builds_manager.cpp
@@ -74,7 +74,8 @@ IndexBuildsManager::~IndexBuildsManager() {
 Status IndexBuildsManager::setUpIndexBuild(OperationContext* opCtx,
                                            Collection* collection,
                                            const std::vector<BSONObj>& specs,
-                                           const UUID& buildUUID) {
+                                           const UUID& buildUUID,
+                                           OnInitFn onInit) {
     _registerIndexBuild(opCtx, collection, buildUUID);
 
     const auto& nss = collection->ns();
@@ -85,10 +86,10 @@ Status IndexBuildsManager::setUpIndexBuild(OperationContext* opCtx,
 
     auto builder = _getBuilder(buildUUID);
 
-    auto initResult = writeConflictRetry(opCtx,
-                                         "IndexBuildsManager::setUpIndexBuild",
-                                         nss.ns(),
-                                         [builder, &specs] { return builder->init(specs); });
+    auto initResult = writeConflictRetry(
+        opCtx, "IndexBuildsManager::setUpIndexBuild", nss.ns(), [opCtx, builder, &onInit, &specs] {
+            return builder->init(specs, onInit);
+        });
 
     if (!initResult.isOK()) {
         return initResult.getStatus();
@@ -138,22 +139,23 @@ Status IndexBuildsManager::checkIndexConstraintViolations(const UUID& buildUUID)
 Status IndexBuildsManager::commitIndexBuild(OperationContext* opCtx,
                                             const NamespaceString& nss,
                                             const UUID& buildUUID,
-                                            OnCommitFn onCommitFn) {
+                                            MultiIndexBlock::OnCreateEachFn onCreateEachFn,
+                                            MultiIndexBlock::OnCommitFn onCommitFn) {
     auto builder = _getBuilder(buildUUID);
 
-    return writeConflictRetry(
-        opCtx, "IndexBuildsManager::commitIndexBuild", nss.ns(), [builder, opCtx, &onCommitFn] {
-            WriteUnitOfWork wunit(opCtx);
-
-            auto status = builder->commit(onCommitFn);
-            if (!status.isOK()) {
-                return status;
-            }
-
-            wunit.commit();
-
-            return Status::OK();
-        });
+    return writeConflictRetry(opCtx,
+                              "IndexBuildsManager::commitIndexBuild",
+                              nss.ns(),
+                              [builder, opCtx, &onCreateEachFn, &onCommitFn] {
+                                  WriteUnitOfWork wunit(opCtx);
+                                  auto status = builder->commit(onCreateEachFn, onCommitFn);
+                                  if (!status.isOK()) {
+                                      return status;
+                                  }
+
+                                  wunit.commit();
+                                  return Status::OK();
+                              });
 }
 
 bool IndexBuildsManager::abortIndexBuild(const UUID& buildUUID, const std::string& reason) {
diff --git a/src/mongo/db/catalog/index_builds_manager.h b/src/mongo/db/catalog/index_builds_manager.h
index 84db234fbe9..1aa735b371c 100644
--- a/src/mongo/db/catalog/index_builds_manager.h
+++ b/src/mongo/db/catalog/index_builds_manager.h
@@ -61,14 +61,13 @@ public:
 
     /**
      * Sets up the index build state and registers it in the manager.
-     *
-     * TODO: Not yet implemented. Only instantiates and registers a builder in the manager. Does not
-     * set up index build state.
      */
+    using OnInitFn = MultiIndexBlock::OnInitFn;
     Status setUpIndexBuild(OperationContext* opCtx,
                            Collection* collection,
                            const std::vector<BSONObj>& specs,
-                           const UUID& buildUUID);
+                           const UUID& buildUUID,
+                           OnInitFn onInit);
 
     /**
      * Recovers the index build from its persisted state and sets it up to run again.
@@ -113,13 +112,13 @@ public:
     /**
      * Persists information in the index catalog entry that the index is ready for use, as well as
      * updating the in-memory index catalog entry for this index to ready.
-     *
-     * TODO: Not yet implemented.
      */
-    using OnCommitFn = stdx::function<void(const BSONObj& spec)>;
+    using OnCreateEachFn = MultiIndexBlock::OnCreateEachFn;
+    using OnCommitFn = MultiIndexBlock::OnCommitFn;
     Status commitIndexBuild(OperationContext* opCtx,
                             const NamespaceString& nss,
                             const UUID& buildUUID,
+                            OnCreateEachFn onCreateEachFn,
                             OnCommitFn onCommitFn);
 
     /**
diff --git a/src/mongo/db/catalog/index_builds_manager_test.cpp b/src/mongo/db/catalog/index_builds_manager_test.cpp
index 10326c77dd3..7b547ca2d6b 100644
--- a/src/mongo/db/catalog/index_builds_manager_test.cpp
+++ b/src/mongo/db/catalog/index_builds_manager_test.cpp
@@ -85,8 +85,12 @@ std::vector<BSONObj> makeSpecs(const NamespaceString& nss, std::vector<std::stri
 TEST_F(IndexBuildsManagerTest, IndexBuildsManagerSetUpAndTearDown) {
     AutoGetCollection autoColl(operationContext(), _nss, MODE_X);
 
-    ASSERT_OK(_indexBuildsManager.setUpIndexBuild(
-        operationContext(), autoColl.getCollection(), makeSpecs(_nss, {"a", "b"}), _buildUUID));
+    auto specs = makeSpecs(_nss, {"a", "b"});
+    ASSERT_OK(_indexBuildsManager.setUpIndexBuild(operationContext(),
+                                                  autoColl.getCollection(),
+                                                  specs,
+                                                  _buildUUID,
+                                                  MultiIndexBlock::kNoopOnInitFn));
 
     _indexBuildsManager.tearDownIndexBuild(_buildUUID);
 }
diff --git a/src/mongo/db/catalog/multi_index_block.cpp b/src/mongo/db/catalog/multi_index_block.cpp
index 1b4d277ebd9..0e3582aa86e 100644
--- a/src/mongo/db/catalog/multi_index_block.cpp
+++ b/src/mongo/db/catalog/multi_index_block.cpp
@@ -160,12 +160,31 @@ void MultiIndexBlock::ignoreUniqueConstraint() {
     _ignoreUnique = true;
 }
 
-StatusWith<std::vector<BSONObj>> MultiIndexBlock::init(const BSONObj& spec) {
+MultiIndexBlock::OnInitFn MultiIndexBlock::kNoopOnInitFn = [] {};
+
+MultiIndexBlock::OnInitFn MultiIndexBlock::makeTimestampedIndexOnInitFn(OperationContext* opCtx,
+                                                                        const Collection* coll) {
+    return [ opCtx, ns = coll->ns() ]() {
+        auto replCoord = repl::ReplicationCoordinator::get(opCtx);
+        if (opCtx->recoveryUnit()->getCommitTimestamp().isNull() &&
+            replCoord->canAcceptWritesForDatabase(opCtx, "admin")) {
+            // Only primaries must timestamp this write. Secondaries run this from within a
+            // `TimestampBlock`. Primaries performing an index build via `applyOps` may have a
+            // wrapping commit timestamp that will be used instead.
+            opCtx->getServiceContext()->getOpObserver()->onOpMessage(
+                opCtx,
+                BSON("msg" << std::string(str::stream() << "Creating indexes. Coll: " << ns)));
+        }
+    };
+}
+
+StatusWith<std::vector<BSONObj>> MultiIndexBlock::init(const BSONObj& spec, OnInitFn onInit) {
     const auto indexes = std::vector<BSONObj>(1, spec);
-    return init(indexes);
+    return init(indexes, onInit);
 }
 
-StatusWith<std::vector<BSONObj>> MultiIndexBlock::init(const std::vector<BSONObj>& indexSpecs) {
+StatusWith<std::vector<BSONObj>> MultiIndexBlock::init(const std::vector<BSONObj>& indexSpecs,
+                                                       OnInitFn onInit) {
     if (State::kAborted == _getState()) {
         return {ErrorCodes::IndexBuildAborted,
                 str::stream() << "Index build aborted: " << _abortReason
@@ -290,15 +309,7 @@ StatusWith<std::vector<BSONObj>> MultiIndexBlock::init(const std::vector<BSONObj
     if (isBackgroundBuilding())
         _backgroundOperation.reset(new BackgroundOperation(ns));
 
-    auto replCoord = repl::ReplicationCoordinator::get(_opCtx);
-    if (_opCtx->recoveryUnit()->getCommitTimestamp().isNull() &&
-        replCoord->canAcceptWritesForDatabase(_opCtx, "admin")) {
-        // Only primaries must timestamp this write. Secondaries run this from within a
-        // `TimestampBlock`. Primaries performing an index build via `applyOps` may have a
-        // wrapping commit timestamp that will be used instead.
-        _opCtx->getServiceContext()->getOpObserver()->onOpMessage(
-            _opCtx, BSON("msg" << std::string(str::stream() << "Creating indexes. Coll: " << ns)));
-    }
+    onInit();
 
     wunit.commit();
 
@@ -637,11 +648,10 @@ void MultiIndexBlock::abortWithoutCleanup() {
     _needToCleanup = false;
 }
 
-Status MultiIndexBlock::commit() {
-    return commit({});
-}
+MultiIndexBlock::OnCreateEachFn MultiIndexBlock::kNoopOnCreateEachFn = [](const BSONObj& spec) {};
+MultiIndexBlock::OnCommitFn MultiIndexBlock::kNoopOnCommitFn = []() {};
 
-Status MultiIndexBlock::commit(stdx::function<void(const BSONObj& spec)> onCreateFn) {
+Status MultiIndexBlock::commit(OnCreateEachFn onCreateEach, OnCommitFn onCommit) {
     if (State::kAborted == _getState()) {
         return {ErrorCodes::IndexBuildAborted,
                 str::stream() << "Index build aborted: " << _abortReason
@@ -661,9 +671,7 @@ Status MultiIndexBlock::commit(stdx::function<void(const BSONObj& spec)> onCreat
     MultikeyPathTracker::get(_opCtx).stopTrackingMultikeyPathInfo();
 
     for (size_t i = 0; i < _indexes.size(); i++) {
-        if (onCreateFn) {
-            onCreateFn(_indexes[i].block->getSpec());
-        }
+        onCreateEach(_indexes[i].block->getSpec());
 
         // Do this before calling success(), which unsets the interceptor pointer on the index
         // catalog entry.
@@ -695,6 +703,8 @@ Status MultiIndexBlock::commit(stdx::function<void(const BSONObj& spec)> onCreat
         }
     }
 
+    onCommit();
+
     // The state of this index build is set to Committed only when the WUOW commits.
     // It is possible for abort() to be called after the check at the beginning of this function and
     // before the WUOW is committed. If the WUOW commits, the final state of this index builder will
diff --git a/src/mongo/db/catalog/multi_index_block.h b/src/mongo/db/catalog/multi_index_block.h
index 227ad8cc722..3a062d18bb6 100644
--- a/src/mongo/db/catalog/multi_index_block.h
+++ b/src/mongo/db/catalog/multi_index_block.h
@@ -87,13 +87,31 @@ public:
      * Prepares the index(es) for building and returns the canonicalized form of the requested index
      * specifications.
      *
+     * Calls 'onInitFn' in the same WriteUnitOfWork as the 'ready: false' write to the index after
+     * all indexes have been initialized. For callers that timestamp this write, use
+     * 'makeTimestampedIndexOnInitFn', otherwise use 'kNoopOnInitFn'.
+     *
      * Does not need to be called inside of a WriteUnitOfWork (but can be due to nesting).
      *
      * Requires holding an exclusive database lock.
      */
-    StatusWith<std::vector<BSONObj>> init(const std::vector<BSONObj>& specs);
+    using OnInitFn = stdx::function<void()>;
+    StatusWith<std::vector<BSONObj>> init(const std::vector<BSONObj>& specs, OnInitFn onInit);
+    StatusWith<std::vector<BSONObj>> init(const BSONObj& spec, OnInitFn onInit);
+
+    /**
+     * Not all index initializations need an OnInitFn, in particular index builds that do not need
+     * to timestamp catalog writes. This is a no-op.
+     */
+    static OnInitFn kNoopOnInitFn;
+
+    /**
+     * Returns an OnInit function for initialization when this index build should be timestamped.
+     * When called on primaries, this generates a new optime, writes a no-op oplog entry, and
+     * timestamps the first catalog write. Does nothing on secondaries.
+     */
+    static OnInitFn makeTimestampedIndexOnInitFn(OperationContext* opCtx, const Collection* coll);
 
-    StatusWith<std::vector<BSONObj>> init(const BSONObj& spec);
 
     /**
      * Inserts all documents in the Collection into the indexes and logs with timing info.
@@ -165,12 +183,21 @@ public:
      * Should be called inside of a WriteUnitOfWork. If the index building is to be logOp'd,
      * logOp() should be called from the same unit of work as commit().
      *
-     * `onCreateFn` will be called on each index before writes that mark the index as "ready".
+     * `onCreateEach` will be called after each index has been marked as "ready".
+     * `onCommit` will be called after all indexes have been marked "ready".
      *
      * Requires holding an exclusive database lock.
      */
-    Status commit();
-    Status commit(stdx::function<void(const BSONObj& spec)> onCreateFn);
+    using OnCommitFn = stdx::function<void()>;
+    using OnCreateEachFn = stdx::function<void(const BSONObj& spec)>;
+    Status commit(OnCreateEachFn onCreateEach, OnCommitFn onCommit);
+
+    /**
+     * Not all index commits need these functions, in particular index builds that do not need
+     * to timestamp catalog writes. These are no-ops.
+     */
+    static OnCreateEachFn kNoopOnCreateEachFn;
+    static OnCommitFn kNoopOnCommitFn;
 
     /**
      * Returns true if this index builder was added to the index catalog successfully.
diff --git a/src/mongo/db/catalog/multi_index_block_test.cpp b/src/mongo/db/catalog/multi_index_block_test.cpp
index 0a7c46a5bb3..bc4ebb99253 100644
--- a/src/mongo/db/catalog/multi_index_block_test.cpp
+++ b/src/mongo/db/catalog/multi_index_block_test.cpp
@@ -111,7 +111,8 @@ TEST_F(MultiIndexBlockTest, CommitWithoutInsertingDocuments) {
     auto indexer = getIndexer();
     ASSERT_EQUALS(MultiIndexBlock::State::kUninitialized, indexer->getState_forTest());
 
-    auto specs = unittest::assertGet(indexer->init(std::vector<BSONObj>()));
+    auto specs =
+        unittest::assertGet(indexer->init(std::vector<BSONObj>(), MultiIndexBlock::kNoopOnInitFn));
     ASSERT_EQUALS(0U, specs.size());
     ASSERT_EQUALS(MultiIndexBlock::State::kRunning, indexer->getState_forTest());
 
@@ -120,7 +121,8 @@ TEST_F(MultiIndexBlockTest, CommitWithoutInsertingDocuments) {
     ASSERT_FALSE(indexer->isCommitted());
     {
         WriteUnitOfWork wunit(getOpCtx());
-        ASSERT_OK(indexer->commit());
+        ASSERT_OK(indexer->commit(MultiIndexBlock::kNoopOnCreateEachFn,
+                                  MultiIndexBlock::kNoopOnCommitFn));
         wunit.commit();
     }
     ASSERT(indexer->isCommitted());
@@ -131,7 +133,8 @@ TEST_F(MultiIndexBlockTest, CommitAfterInsertingSingleDocument) {
     auto indexer = getIndexer();
     ASSERT_EQUALS(MultiIndexBlock::State::kUninitialized, indexer->getState_forTest());
 
-    auto specs = unittest::assertGet(indexer->init(std::vector<BSONObj>()));
+    auto specs =
+        unittest::assertGet(indexer->init(std::vector<BSONObj>(), MultiIndexBlock::kNoopOnInitFn));
     ASSERT_EQUALS(0U, specs.size());
     ASSERT_EQUALS(MultiIndexBlock::State::kRunning, indexer->getState_forTest());
 
@@ -141,7 +144,8 @@ TEST_F(MultiIndexBlockTest, CommitAfterInsertingSingleDocument) {
     ASSERT_FALSE(indexer->isCommitted());
     {
         WriteUnitOfWork wunit(getOpCtx());
-        ASSERT_OK(indexer->commit());
+        ASSERT_OK(indexer->commit(MultiIndexBlock::kNoopOnCreateEachFn,
+                                  MultiIndexBlock::kNoopOnCommitFn));
         wunit.commit();
     }
     ASSERT(indexer->isCommitted());
@@ -153,7 +157,8 @@ TEST_F(MultiIndexBlockTest, CommitAfterInsertingSingleDocument) {
 
 TEST_F(MultiIndexBlockTest, AbortWithoutCleanupAfterInsertingSingleDocument) {
     auto indexer = getIndexer();
-    auto specs = unittest::assertGet(indexer->init(std::vector<BSONObj>()));
+    auto specs =
+        unittest::assertGet(indexer->init(std::vector<BSONObj>(), MultiIndexBlock::kNoopOnInitFn));
     ASSERT_EQUALS(0U, specs.size());
     ASSERT_OK(indexer->insert({}, {}));
     indexer->abortWithoutCleanup();
@@ -169,7 +174,9 @@ TEST_F(MultiIndexBlockTest, InitFailsAfterAbort) {
     indexer->abort("test"_sd);
     ASSERT_EQUALS(MultiIndexBlock::State::kAborted, indexer->getState_forTest());
 
-    ASSERT_EQUALS(ErrorCodes::IndexBuildAborted, indexer->init(std::vector<BSONObj>()).getStatus());
+    ASSERT_EQUALS(
+        ErrorCodes::IndexBuildAborted,
+        indexer->init(std::vector<BSONObj>(), MultiIndexBlock::kNoopOnInitFn).getStatus());
     ASSERT_EQUALS(MultiIndexBlock::State::kAborted, indexer->getState_forTest());
 
     ASSERT_FALSE(indexer->isCommitted());
@@ -179,7 +186,8 @@ TEST_F(MultiIndexBlockTest, InsertingSingleDocumentFailsAfterAbort) {
     auto indexer = getIndexer();
     ASSERT_EQUALS(MultiIndexBlock::State::kUninitialized, indexer->getState_forTest());
 
-    auto specs = unittest::assertGet(indexer->init(std::vector<BSONObj>()));
+    auto specs =
+        unittest::assertGet(indexer->init(std::vector<BSONObj>(), MultiIndexBlock::kNoopOnInitFn));
     ASSERT_EQUALS(0U, specs.size());
     ASSERT_EQUALS(MultiIndexBlock::State::kRunning, indexer->getState_forTest());
 
@@ -197,7 +205,8 @@ TEST_F(MultiIndexBlockTest, DumpInsertsFromBulkFailsAfterAbort) {
     auto indexer = getIndexer();
     ASSERT_EQUALS(MultiIndexBlock::State::kUninitialized, indexer->getState_forTest());
 
-    auto specs = unittest::assertGet(indexer->init(std::vector<BSONObj>()));
+    auto specs =
+        unittest::assertGet(indexer->init(std::vector<BSONObj>(), MultiIndexBlock::kNoopOnInitFn));
     ASSERT_EQUALS(0U, specs.size());
     ASSERT_EQUALS(MultiIndexBlock::State::kRunning, indexer->getState_forTest());
 
@@ -217,7 +226,8 @@ TEST_F(MultiIndexBlockTest, CommitFailsAfterAbort) {
     auto indexer = getIndexer();
     ASSERT_EQUALS(MultiIndexBlock::State::kUninitialized, indexer->getState_forTest());
 
-    auto specs = unittest::assertGet(indexer->init(std::vector<BSONObj>()));
+    auto specs =
+        unittest::assertGet(indexer->init(std::vector<BSONObj>(), MultiIndexBlock::kNoopOnInitFn));
     ASSERT_EQUALS(0U, specs.size());
     ASSERT_EQUALS(MultiIndexBlock::State::kRunning, indexer->getState_forTest());
 
@@ -229,7 +239,9 @@ TEST_F(MultiIndexBlockTest, CommitFailsAfterAbort) {
     indexer->abort("test"_sd);
     ASSERT_EQUALS(MultiIndexBlock::State::kAborted, indexer->getState_forTest());
 
-    ASSERT_EQUALS(ErrorCodes::IndexBuildAborted, indexer->commit());
+    ASSERT_EQUALS(
+        ErrorCodes::IndexBuildAborted,
+        indexer->commit(MultiIndexBlock::kNoopOnCreateEachFn, MultiIndexBlock::kNoopOnCommitFn));
     ASSERT_EQUALS(MultiIndexBlock::State::kAborted, indexer->getState_forTest());
 
     ASSERT_FALSE(indexer->isCommitted());
diff --git a/src/mongo/db/cloner.cpp b/src/mongo/db/cloner.cpp
index 3d1ebdec71d..aecdf1806b2 100644
--- a/src/mongo/db/cloner.cpp
+++ b/src/mongo/db/cloner.cpp
@@ -409,11 +409,13 @@ void Cloner::copyIndexes(OperationContext* opCtx,
         return;
     }
 
-    auto indexInfoObjs = uassertStatusOK(indexer.init(prunedIndexesToBuild));
+    auto indexInfoObjs =
+        uassertStatusOK(indexer.init(prunedIndexesToBuild, MultiIndexBlock::kNoopOnInitFn));
     uassertStatusOK(indexer.insertAllDocumentsInCollection());
 
     WriteUnitOfWork wunit(opCtx);
-    uassertStatusOK(indexer.commit());
+    uassertStatusOK(
+        indexer.commit(MultiIndexBlock::kNoopOnCreateEachFn, MultiIndexBlock::kNoopOnCommitFn));
     if (opCtx->writesAreReplicated()) {
         for (auto&& infoObj : indexInfoObjs) {
             getGlobalServiceContext()->getOpObserver()->onCreateIndex(
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index 754e8cce6be..a0e5ce4cc3c 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -289,7 +289,6 @@ bool runCreateIndexes(OperationContext* opCtx,
                          AutoStatsTracker::LogMode::kUpdateTopAndCurop,
                          dbProfilingLevel);
 
-
     MultiIndexBlock indexer(opCtx, collection);
 
     const size_t origSpecsSize = specs.size();
@@ -314,8 +313,9 @@ bool runCreateIndexes(OperationContext* opCtx,
     }
 
     std::vector<BSONObj> indexInfoObjs =
-        writeConflictRetry(opCtx, kCommandName, ns.ns(), [&indexer, &specs] {
-            return uassertStatusOK(indexer.init(specs));
+        writeConflictRetry(opCtx, kCommandName, ns.ns(), [opCtx, collection, &indexer, &specs] {
+            return uassertStatusOK(indexer.init(
+                specs, MultiIndexBlock::makeTimestampedIndexOnInitFn(opCtx, collection)));
         });
 
     // If we're a background index, replace exclusive db lock with an intent lock, so that
@@ -408,10 +408,12 @@ bool runCreateIndexes(OperationContext* opCtx,
     writeConflictRetry(opCtx, kCommandName, ns.ns(), [&] {
         WriteUnitOfWork wunit(opCtx);
 
-        uassertStatusOK(indexer.commit([opCtx, &ns, collection](const BSONObj& spec) {
-            opCtx->getServiceContext()->getOpObserver()->onCreateIndex(
-                opCtx, ns, *(collection->uuid()), spec, false);
-        }));
+        uassertStatusOK(indexer.commit(
+            [opCtx, &ns, collection](const BSONObj& spec) {
+                opCtx->getServiceContext()->getOpObserver()->onCreateIndex(
+                    opCtx, ns, *(collection->uuid()), spec, false);
+            },
+            MultiIndexBlock::kNoopOnCommitFn));
 
         wunit.commit();
     });
@@ -532,11 +534,13 @@ bool runCreateIndexesWithCoordinator(OperationContext* opCtx,
 
     auto indexBuildsCoord = IndexBuildsCoordinator::get(opCtx);
     auto buildUUID = UUID::gen();
+    auto protocol =
+        (runTwoPhaseBuild) ? IndexBuildProtocol::kTwoPhase : IndexBuildProtocol::kSinglePhase;
     log() << "Registering index build: " << buildUUID;
     ReplIndexBuildState::IndexCatalogStats stats;
     try {
         auto buildIndexFuture = uassertStatusOK(
-            indexBuildsCoord->startIndexBuild(opCtx, *collectionUUID, specs, buildUUID));
+            indexBuildsCoord->startIndexBuild(opCtx, *collectionUUID, specs, buildUUID, protocol));
 
         auto deadline = opCtx->getDeadline();
         // Date_t::max() means no deadline.
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index 8d01d7f5004..0081a133cea 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -204,7 +204,7 @@ public:
 
             indexer = std::make_unique<MultiIndexBlock>(opCtx, collection);
 
-            swIndexesToRebuild = indexer->init(all);
+            swIndexesToRebuild = indexer->init(all, MultiIndexBlock::kNoopOnInitFn);
             uassertStatusOK(swIndexesToRebuild.getStatus());
             wunit.commit();
         }
@@ -214,7 +214,8 @@ public:
 
         {
             WriteUnitOfWork wunit(opCtx);
-            uassertStatusOK(indexer->commit());
+            uassertStatusOK(indexer->commit(MultiIndexBlock::kNoopOnCreateEachFn,
+                                            MultiIndexBlock::kNoopOnCommitFn));
             wunit.commit();
         }
 
diff --git a/src/mongo/db/index_builder.cpp b/src/mongo/db/index_builder.cpp
index 754848ae5cd..d73a31ab4e2 100644
--- a/src/mongo/db/index_builder.cpp
+++ b/src/mongo/db/index_builder.cpp
@@ -224,10 +224,13 @@ Status IndexBuilder::_build(OperationContext* opCtx,
     }
 
     Status status = Status::OK();
+
     {
         TimestampBlock tsBlock(opCtx, _initIndexTs);
-        status = writeConflictRetry(
-            opCtx, "Init index build", ns.ns(), [&] { return indexer.init(_index).getStatus(); });
+        status = writeConflictRetry(opCtx, "Init index build", ns.ns(), [&] {
+            return indexer.init(_index, MultiIndexBlock::makeTimestampedIndexOnInitFn(opCtx, coll))
+                .getStatus();
+        });
     }
 
     if (status == ErrorCodes::IndexAlreadyExists ||
@@ -309,10 +312,12 @@ Status IndexBuilder::_build(OperationContext* opCtx,
 
     status = writeConflictRetry(opCtx, "Commit index build", ns.ns(), [opCtx, coll, &indexer, &ns] {
         WriteUnitOfWork wunit(opCtx);
-        auto status = indexer.commit([opCtx, coll, &ns](const BSONObj& indexSpec) {
-            opCtx->getServiceContext()->getOpObserver()->onCreateIndex(
-                opCtx, ns, *(coll->uuid()), indexSpec, false);
-        });
+        auto status = indexer.commit(
+            [opCtx, coll, &ns](const BSONObj& indexSpec) {
+                opCtx->getServiceContext()->getOpObserver()->onCreateIndex(
+                    opCtx, ns, *(coll->uuid()), indexSpec, false);
+            },
+            MultiIndexBlock::kNoopOnCommitFn);
         if (!status.isOK()) {
             return status;
         }
diff --git a/src/mongo/db/index_builds_coordinator.cpp b/src/mongo/db/index_builds_coordinator.cpp
index e77d4b1e046..8c9a6dd7636 100644
--- a/src/mongo/db/index_builds_coordinator.cpp
+++ b/src/mongo/db/index_builds_coordinator.cpp
@@ -436,13 +436,6 @@ void IndexBuildsCoordinator::_runIndexBuild(OperationContext* opCtx,
         // not allow locks or re-locks to be interrupted.
         UninterruptibleLockGuard noInterrupt(opCtx->lockState());
 
-        if (!repl::ReplicationCoordinator::get(opCtx)->canAcceptWritesFor(opCtx, nss)) {
-            uasserted(ErrorCodes::NotMaster,
-                      str::stream() << "Not primary while creating indexes in " << nss.ns() << " ("
-                                    << collectionUUID
-                                    << ")");
-        }
-
         auto collection = uuidCatalog.lookupCollectionByUUID(collectionUUID);
         uassert(ErrorCodes::NamespaceNotFound,
                 str::stream() << "Collection not found for index build: " << buildUUID << ": "
@@ -480,8 +473,20 @@ void IndexBuildsCoordinator::_runIndexBuild(OperationContext* opCtx,
         }
 
         mustTearDown = true;
-        uassertStatusOK(
-            _indexBuildsManager.setUpIndexBuild(opCtx, collection, specsToBuild, buildUUID));
+
+        MultiIndexBlock::OnInitFn onInitFn;
+        // Two-phase index builds write a different oplog entry than the default behavior which
+        // writes a no-op just to generate an optime.
+        if (IndexBuildProtocol::kTwoPhase == replState->protocol) {
+            onInitFn = [&] {
+                opCtx->getServiceContext()->getOpObserver()->onStartIndexBuild(
+                    opCtx, nss, collectionUUID, buildUUID, specsToBuild, false /* fromMigrate */);
+            };
+        } else {
+            onInitFn = MultiIndexBlock::makeTimestampedIndexOnInitFn(opCtx, collection);
+        }
+        uassertStatusOK(_indexBuildsManager.setUpIndexBuild(
+            opCtx, collection, specsToBuild, buildUUID, onInitFn));
 
         // If we're a background index, replace exclusive db lock with an intent lock, so that
         // other readers and writers can proceed during this phase.
@@ -564,12 +569,25 @@ void IndexBuildsCoordinator::_runIndexBuild(OperationContext* opCtx,
         // Index constraint checking phase.
         uassertStatusOK(_indexBuildsManager.checkIndexConstraintViolations(buildUUID));
 
+        auto onCommitFn = MultiIndexBlock::kNoopOnCommitFn;
+        auto onCreateEachFn = MultiIndexBlock::kNoopOnCreateEachFn;
+        if (IndexBuildProtocol::kTwoPhase == replState->protocol) {
+            // Two-phase index builds write one oplog entry for all indexes that are completed.
+            onCommitFn = [&] {
+                opCtx->getServiceContext()->getOpObserver()->onCommitIndexBuild(
+                    opCtx, nss, collectionUUID, buildUUID, specsToBuild, false /* fromMigrate */);
+            };
+        } else {
+            // Single-phase index builds write an oplog entry per index being built.
+            onCreateEachFn = [opCtx, &nss, &collectionUUID](const BSONObj& spec) {
+                opCtx->getServiceContext()->getOpObserver()->onCreateIndex(
+                    opCtx, nss, collectionUUID, spec, false);
+            };
+        }
+
         // Commit index build.
-        auto onCommitFn = [opCtx, &nss, &collectionUUID](const BSONObj& spec) {
-            opCtx->getServiceContext()->getOpObserver()->onCreateIndex(
-                opCtx, nss, collectionUUID, spec, false);
-        };
-        uassertStatusOK(_indexBuildsManager.commitIndexBuild(opCtx, nss, buildUUID, onCommitFn));
+        uassertStatusOK(_indexBuildsManager.commitIndexBuild(
+            opCtx, nss, buildUUID, onCreateEachFn, onCommitFn));
 
         indexCatalogStats.numIndexesAfter = getNumIndexesTotal(opCtx, collection);
         log() << "Index builds manager completed successfully: " << buildUUID << ": " << nss
@@ -628,7 +646,6 @@ void IndexBuildsCoordinator::_runIndexBuild(OperationContext* opCtx,
     } else {
         replState->sharedPromise.setError(status);
     }
-
     return;
 }
 
diff --git a/src/mongo/db/index_builds_coordinator.h b/src/mongo/db/index_builds_coordinator.h
index be2266b6186..ec8eee35344 100644
--- a/src/mongo/db/index_builds_coordinator.h
+++ b/src/mongo/db/index_builds_coordinator.h
@@ -102,7 +102,8 @@ public:
         OperationContext* opCtx,
         CollectionUUID collectionUUID,
         const std::vector<BSONObj>& specs,
-        const UUID& buildUUID) = 0;
+        const UUID& buildUUID,
+        IndexBuildProtocol protocol) = 0;
 
     /**
      * TODO: not yet implemented.
diff --git a/src/mongo/db/index_builds_coordinator_mongod.cpp b/src/mongo/db/index_builds_coordinator_mongod.cpp
index 70a45354458..d4591665eb6 100644
--- a/src/mongo/db/index_builds_coordinator_mongod.cpp
+++ b/src/mongo/db/index_builds_coordinator_mongod.cpp
@@ -84,7 +84,8 @@ StatusWith<SharedSemiFuture<ReplIndexBuildState::IndexCatalogStats>>
 IndexBuildsCoordinatorMongod::startIndexBuild(OperationContext* opCtx,
                                               CollectionUUID collectionUUID,
                                               const std::vector<BSONObj>& specs,
-                                              const UUID& buildUUID) {
+                                              const UUID& buildUUID,
+                                              IndexBuildProtocol protocol) {
     std::vector<std::string> indexNames;
     for (auto& spec : specs) {
         std::string name = spec.getStringField(IndexDescriptor::kIndexNameFieldName);
@@ -99,8 +100,8 @@ IndexBuildsCoordinatorMongod::startIndexBuild(OperationContext* opCtx,
 
     auto nss = UUIDCatalog::get(opCtx).lookupNSSByUUID(collectionUUID);
     auto dbName = nss.db().toString();
-    auto replIndexBuildState =
-        std::make_shared<ReplIndexBuildState>(buildUUID, collectionUUID, dbName, indexNames, specs);
+    auto replIndexBuildState = std::make_shared<ReplIndexBuildState>(
+        buildUUID, collectionUUID, dbName, indexNames, specs, protocol);
 
     Status status = _registerIndexBuild(opCtx, replIndexBuildState);
     if (!status.isOK()) {
@@ -119,9 +120,23 @@ IndexBuildsCoordinatorMongod::startIndexBuild(OperationContext* opCtx,
         return replIndexBuildState->sharedPromise.getFuture();
     }
 
+    // Copy over all necessary OperationContext state.
+
     // Task in thread pool should retain the caller's deadline.
-    auto deadline = opCtx->getDeadline();
-    auto timeoutError = opCtx->getTimeoutError();
+    const auto deadline = opCtx->getDeadline();
+    const auto timeoutError = opCtx->getTimeoutError();
+
+    // TODO: SERVER-39484 Because both 'writesAreReplicated' and
+    // 'shouldNotConflictWithSecondaryBatchApplication' depend on the current replication state,
+    // just passing the state here is not resilient to member state changes like stepup/stepdown.
+
+    // If the calling thread is replicating oplog writes (primary), this state should be passed to
+    // the builder.
+    const bool writesAreReplicated = opCtx->writesAreReplicated();
+    // Index builds on secondaries can't hold the PBWM lock because it would conflict with
+    // replication.
+    const bool shouldNotConflictWithSecondaryBatchApplication =
+        !opCtx->lockState()->shouldConflictWithSecondaryBatchApplication();
 
     // Task in thread pool should have similar CurOp representation to the caller so that it can be
     // identified as a createIndexes operation.
@@ -132,11 +147,30 @@ IndexBuildsCoordinatorMongod::startIndexBuild(OperationContext* opCtx,
         opDesc = curOp->opDescription().getOwned();
     }
 
-    status = _threadPool.schedule([ this, buildUUID, deadline, timeoutError, opDesc ]() noexcept {
+    status = _threadPool.schedule([
+        this,
+        buildUUID,
+        deadline,
+        timeoutError,
+        writesAreReplicated,
+        shouldNotConflictWithSecondaryBatchApplication,
+        opDesc
+    ]() noexcept {
         auto opCtx = Client::getCurrent()->makeOperationContext();
 
         opCtx->setDeadlineByDate(deadline, timeoutError);
 
+        boost::optional<repl::UnreplicatedWritesBlock> unreplicatedWrites;
+        if (!writesAreReplicated) {
+            unreplicatedWrites.emplace(opCtx.get());
+        }
+
+        // If the calling thread should not take the PBWM lock, neither should this thread.
+        boost::optional<ShouldNotConflictWithSecondaryBatchApplicationBlock> shouldNotConflictBlock;
+        if (shouldNotConflictWithSecondaryBatchApplication) {
+            shouldNotConflictBlock.emplace(opCtx->lockState());
+        }
+
         {
             stdx::unique_lock<Client> lk(*opCtx->getClient());
             auto curOp = CurOp::get(opCtx.get());
diff --git a/src/mongo/db/index_builds_coordinator_mongod.h b/src/mongo/db/index_builds_coordinator_mongod.h
index fe62eb70323..aef08abda8e 100644
--- a/src/mongo/db/index_builds_coordinator_mongod.h
+++ b/src/mongo/db/index_builds_coordinator_mongod.h
@@ -73,7 +73,8 @@ public:
         OperationContext* opCtx,
         CollectionUUID collectionUUID,
         const std::vector<BSONObj>& specs,
-        const UUID& buildUUID) override;
+        const UUID& buildUUID,
+        IndexBuildProtocol protocol) override;
 
     /**
      * TODO: not yet implemented.
diff --git a/src/mongo/db/index_builds_coordinator_mongod_test.cpp b/src/mongo/db/index_builds_coordinator_mongod_test.cpp
index 7f22c59bf5b..c74b1afb34d 100644
--- a/src/mongo/db/index_builds_coordinator_mongod_test.cpp
+++ b/src/mongo/db/index_builds_coordinator_mongod_test.cpp
@@ -103,15 +103,22 @@ TEST_F(IndexBuildsCoordinatorMongodTest, CannotBuildIndexWithSameIndexName) {
     _indexBuildsCoord->sleepIndexBuilds_forTestOnly(true);
 
     // Register an index build on _testFooNss.
-    auto testFoo1Future = assertGet(_indexBuildsCoord->startIndexBuild(
-        operationContext(), _testFooUUID, makeSpecs(_testFooNss, {"a", "b"}), UUID::gen()));
+    auto testFoo1Future =
+        assertGet(_indexBuildsCoord->startIndexBuild(operationContext(),
+                                                     _testFooUUID,
+                                                     makeSpecs(_testFooNss, {"a", "b"}),
+                                                     UUID::gen(),
+                                                     IndexBuildProtocol::kTwoPhase));
 
     // Attempt and fail to register an index build on _testFooNss with the same index name, while
     // the prior build is still running.
     ASSERT_EQ(ErrorCodes::IndexKeySpecsConflict,
               _indexBuildsCoord
-                  ->startIndexBuild(
-                      operationContext(), _testFooUUID, makeSpecs(_testFooNss, {"b"}), UUID::gen())
+                  ->startIndexBuild(operationContext(),
+                                    _testFooUUID,
+                                    makeSpecs(_testFooNss, {"b"}),
+                                    UUID::gen(),
+                                    IndexBuildProtocol::kTwoPhase)
                   .getStatus());
 
     _indexBuildsCoord->sleepIndexBuilds_forTestOnly(false);
@@ -126,8 +133,12 @@ TEST_F(IndexBuildsCoordinatorMongodTest, Registration) {
     _indexBuildsCoord->sleepIndexBuilds_forTestOnly(true);
 
     // Register an index build on _testFooNss.
-    auto testFoo1Future = assertGet(_indexBuildsCoord->startIndexBuild(
-        operationContext(), _testFooUUID, makeSpecs(_testFooNss, {"a", "b"}), UUID::gen()));
+    auto testFoo1Future =
+        assertGet(_indexBuildsCoord->startIndexBuild(operationContext(),
+                                                     _testFooUUID,
+                                                     makeSpecs(_testFooNss, {"a", "b"}),
+                                                     UUID::gen(),
+                                                     IndexBuildProtocol::kTwoPhase));
 
     ASSERT_EQ(_indexBuildsCoord->numInProgForDb(_testFooNss.db()), 1);
     ASSERT(_indexBuildsCoord->inProgForCollection(_testFooUUID));
@@ -140,8 +151,12 @@ TEST_F(IndexBuildsCoordinatorMongodTest, Registration) {
                        ErrorCodes::BackgroundOperationInProgressForDatabase);
 
     // Register a second index build on _testFooNss.
-    auto testFoo2Future = assertGet(_indexBuildsCoord->startIndexBuild(
-        operationContext(), _testFooUUID, makeSpecs(_testFooNss, {"c", "d"}), UUID::gen()));
+    auto testFoo2Future =
+        assertGet(_indexBuildsCoord->startIndexBuild(operationContext(),
+                                                     _testFooUUID,
+                                                     makeSpecs(_testFooNss, {"c", "d"}),
+                                                     UUID::gen(),
+                                                     IndexBuildProtocol::kTwoPhase));
 
     ASSERT_EQ(_indexBuildsCoord->numInProgForDb(_testFooNss.db()), 2);
     ASSERT(_indexBuildsCoord->inProgForCollection(_testFooUUID));
@@ -154,8 +169,12 @@ TEST_F(IndexBuildsCoordinatorMongodTest, Registration) {
                        ErrorCodes::BackgroundOperationInProgressForDatabase);
 
     // Register an index build on a different collection _testBarNss.
-    auto testBarFuture = assertGet(_indexBuildsCoord->startIndexBuild(
-        operationContext(), _testBarUUID, makeSpecs(_testBarNss, {"x", "y"}), UUID::gen()));
+    auto testBarFuture =
+        assertGet(_indexBuildsCoord->startIndexBuild(operationContext(),
+                                                     _testBarUUID,
+                                                     makeSpecs(_testBarNss, {"x", "y"}),
+                                                     UUID::gen(),
+                                                     IndexBuildProtocol::kTwoPhase));
 
     ASSERT_EQ(_indexBuildsCoord->numInProgForDb(_testBarNss.db()), 3);
     ASSERT(_indexBuildsCoord->inProgForCollection(_testBarUUID));
@@ -172,7 +191,8 @@ TEST_F(IndexBuildsCoordinatorMongodTest, Registration) {
         assertGet(_indexBuildsCoord->startIndexBuild(operationContext(),
                                                      _othertestFooUUID,
                                                      makeSpecs(_othertestFooNss, {"r", "s"}),
-                                                     UUID::gen()));
+                                                     UUID::gen(),
+                                                     IndexBuildProtocol::kTwoPhase));
 
     ASSERT_EQ(_indexBuildsCoord->numInProgForDb(_othertestFooNss.db()), 1);
     ASSERT(_indexBuildsCoord->inProgForCollection(_othertestFooUUID));
@@ -232,17 +252,23 @@ TEST_F(IndexBuildsCoordinatorMongodTest, DisallowNewBuildsOnNamespace) {
                       ->startIndexBuild(operationContext(),
                                         _testFooUUID,
                                         makeSpecs(_testFooNss, {"a", "b"}),
-                                        UUID::gen())
+                                        UUID::gen(),
+                                        IndexBuildProtocol::kTwoPhase)
                       .getStatus());
 
         // Registering index builds on other collections and databases should still succeed.
-        auto testBarFuture = assertGet(_indexBuildsCoord->startIndexBuild(
-            operationContext(), _testBarUUID, makeSpecs(_testBarNss, {"c", "d"}), UUID::gen()));
+        auto testBarFuture =
+            assertGet(_indexBuildsCoord->startIndexBuild(operationContext(),
+                                                         _testBarUUID,
+                                                         makeSpecs(_testBarNss, {"c", "d"}),
+                                                         UUID::gen(),
+                                                         IndexBuildProtocol::kTwoPhase));
         auto othertestFooFuture =
             assertGet(_indexBuildsCoord->startIndexBuild(operationContext(),
                                                          _othertestFooUUID,
                                                          makeSpecs(_othertestFooNss, {"e", "f"}),
-                                                         UUID::gen()));
+                                                         UUID::gen(),
+                                                         IndexBuildProtocol::kTwoPhase));
 
         _indexBuildsCoord->sleepIndexBuilds_forTestOnly(false);
 
@@ -256,8 +282,12 @@ TEST_F(IndexBuildsCoordinatorMongodTest, DisallowNewBuildsOnNamespace) {
 
     {
         // Check that the scoped object correctly cleared.
-        auto testFooFuture = assertGet(_indexBuildsCoord->startIndexBuild(
-            operationContext(), _testFooUUID, makeSpecs(_testFooNss, {"a", "b"}), UUID::gen()));
+        auto testFooFuture =
+            assertGet(_indexBuildsCoord->startIndexBuild(operationContext(),
+                                                         _testFooUUID,
+                                                         makeSpecs(_testFooNss, {"a", "b"}),
+                                                         UUID::gen(),
+                                                         IndexBuildProtocol::kTwoPhase));
         auto indexCatalogStats = unittest::assertGet(testFooFuture.getNoThrow());
         ASSERT_EQ(1, indexCatalogStats.numIndexesBefore);
         ASSERT_EQ(3, indexCatalogStats.numIndexesAfter);
@@ -275,14 +305,16 @@ TEST_F(IndexBuildsCoordinatorMongodTest, DisallowNewBuildsOnNamespace) {
                       ->startIndexBuild(operationContext(),
                                         _testFooUUID,
                                         makeSpecs(_testFooNss, {"a", "b"}),
-                                        UUID::gen())
+                                        UUID::gen(),
+                                        IndexBuildProtocol::kTwoPhase)
                       .getStatus());
         ASSERT_EQ(ErrorCodes::CannotCreateIndex,
                   _indexBuildsCoord
                       ->startIndexBuild(operationContext(),
                                         _testBarUUID,
                                         makeSpecs(_testBarNss, {"c", "d"}),
-                                        UUID::gen())
+                                        UUID::gen(),
+                                        IndexBuildProtocol::kTwoPhase)
                       .getStatus());
 
         // Registering index builds on another database should still succeed.
@@ -290,7 +322,8 @@ TEST_F(IndexBuildsCoordinatorMongodTest, DisallowNewBuildsOnNamespace) {
             assertGet(_indexBuildsCoord->startIndexBuild(operationContext(),
                                                          _othertestFooUUID,
                                                          makeSpecs(_othertestFooNss, {"g", "h"}),
-                                                         UUID::gen()));
+                                                         UUID::gen(),
+                                                         IndexBuildProtocol::kTwoPhase));
 
         _indexBuildsCoord->sleepIndexBuilds_forTestOnly(false);
 
@@ -301,8 +334,12 @@ TEST_F(IndexBuildsCoordinatorMongodTest, DisallowNewBuildsOnNamespace) {
 
     {
         // Check that the scoped object correctly cleared.
-        auto testFooFuture = assertGet(_indexBuildsCoord->startIndexBuild(
-            operationContext(), _testFooUUID, makeSpecs(_testFooNss, {"c", "d"}), UUID::gen()));
+        auto testFooFuture =
+            assertGet(_indexBuildsCoord->startIndexBuild(operationContext(),
+                                                         _testFooUUID,
+                                                         makeSpecs(_testFooNss, {"c", "d"}),
+                                                         UUID::gen(),
+                                                         IndexBuildProtocol::kTwoPhase));
         auto indexCatalogStats = unittest::assertGet(testFooFuture.getNoThrow());
         ASSERT_EQ(3, indexCatalogStats.numIndexesBefore);
         ASSERT_EQ(5, indexCatalogStats.numIndexesAfter);
@@ -320,7 +357,8 @@ TEST_F(IndexBuildsCoordinatorMongodTest, DisallowNewBuildsOnNamespace) {
                           ->startIndexBuild(operationContext(),
                                             _testFooUUID,
                                             makeSpecs(_testFooNss, {"a", "b"}),
-                                            UUID::gen())
+                                            UUID::gen(),
+                                            IndexBuildProtocol::kTwoPhase)
                           .getStatus());
         }
         ASSERT_EQ(ErrorCodes::CannotCreateIndex,
@@ -328,14 +366,19 @@ TEST_F(IndexBuildsCoordinatorMongodTest, DisallowNewBuildsOnNamespace) {
                       ->startIndexBuild(operationContext(),
                                         _testFooUUID,
                                         makeSpecs(_testFooNss, {"a", "b"}),
-                                        UUID::gen())
+                                        UUID::gen(),
+                                        IndexBuildProtocol::kTwoPhase)
                       .getStatus());
     }
 
     {
         // Check that the scoped object correctly cleared.
-        auto testFooFuture = assertGet(_indexBuildsCoord->startIndexBuild(
-            operationContext(), _testFooUUID, makeSpecs(_testFooNss, {"e", "f"}), UUID::gen()));
+        auto testFooFuture =
+            assertGet(_indexBuildsCoord->startIndexBuild(operationContext(),
+                                                         _testFooUUID,
+                                                         makeSpecs(_testFooNss, {"e", "f"}),
+                                                         UUID::gen(),
+                                                         IndexBuildProtocol::kTwoPhase));
         auto indexCatalogStats = unittest::assertGet(testFooFuture.getNoThrow());
         ASSERT_EQ(5, indexCatalogStats.numIndexesBefore);
         ASSERT_EQ(7, indexCatalogStats.numIndexesAfter);
diff --git a/src/mongo/db/repair_database.cpp b/src/mongo/db/repair_database.cpp
index b45e3ad1298..22215db1fa4 100644
--- a/src/mongo/db/repair_database.cpp
+++ b/src/mongo/db/repair_database.cpp
@@ -151,7 +151,7 @@ Status rebuildIndexesOnCollection(OperationContext* opCtx,
         collection = databaseHolder->makeCollection(opCtx, ns, uuid, cce, rs, dbce);
 
         indexer = std::make_unique<MultiIndexBlock>(opCtx, collection.get());
-        Status status = indexer->init(indexSpecs).getStatus();
+        Status status = indexer->init(indexSpecs, MultiIndexBlock::kNoopOnInitFn).getStatus();
         if (!status.isOK()) {
             // The WUOW will handle cleanup, so the indexer shouldn't do its own.
             indexer->abortWithoutCleanup();
@@ -220,7 +220,8 @@ Status rebuildIndexesOnCollection(OperationContext* opCtx,
 
     {
         WriteUnitOfWork wunit(opCtx);
-        status = indexer->commit();
+        status =
+            indexer->commit(MultiIndexBlock::kNoopOnCreateEachFn, MultiIndexBlock::kNoopOnCommitFn);
         if (!status.isOK()) {
             return status;
         }
diff --git a/src/mongo/db/repl/collection_bulk_loader_impl.cpp b/src/mongo/db/repl/collection_bulk_loader_impl.cpp
index f29692a1f0a..d8453e28a8d 100644
--- a/src/mongo/db/repl/collection_bulk_loader_impl.cpp
+++ b/src/mongo/db/repl/collection_bulk_loader_impl.cpp
@@ -86,7 +86,8 @@ Status CollectionBulkLoaderImpl::init(const std::vector<BSONObj>& secondaryIndex
             auto specs = indexCatalog->removeExistingIndexes(_opCtx.get(), secondaryIndexSpecs);
             if (specs.size()) {
                 _secondaryIndexesBlock->ignoreUniqueConstraint();
-                auto status = _secondaryIndexesBlock->init(specs).getStatus();
+                auto status =
+                    _secondaryIndexesBlock->init(specs, MultiIndexBlock::kNoopOnInitFn).getStatus();
                 if (!status.isOK()) {
                     return status;
                 }
@@ -94,7 +95,8 @@ Status CollectionBulkLoaderImpl::init(const std::vector<BSONObj>& secondaryIndex
                 _secondaryIndexesBlock.reset();
             }
             if (!_idIndexSpec.isEmpty()) {
-                auto status = _idIndexBlock->init(_idIndexSpec).getStatus();
+                auto status =
+                    _idIndexBlock->init(_idIndexSpec, MultiIndexBlock::kNoopOnInitFn).getStatus();
                 if (!status.isOK()) {
                     return status;
                 }
@@ -176,7 +178,8 @@ Status CollectionBulkLoaderImpl::commit() {
             status = writeConflictRetry(
                 _opCtx.get(), "CollectionBulkLoaderImpl::commit", _nss.ns(), [this] {
                     WriteUnitOfWork wunit(_opCtx.get());
-                    auto status = _secondaryIndexesBlock->commit();
+                    auto status = _secondaryIndexesBlock->commit(
+                        MultiIndexBlock::kNoopOnCreateEachFn, MultiIndexBlock::kNoopOnCommitFn);
                     if (!status.isOK()) {
                         return status;
                     }
@@ -202,7 +205,8 @@ Status CollectionBulkLoaderImpl::commit() {
             status = writeConflictRetry(
                 _opCtx.get(), "CollectionBulkLoaderImpl::commit", _nss.ns(), [this] {
                     WriteUnitOfWork wunit(_opCtx.get());
-                    auto status = _idIndexBlock->commit();
+                    auto status = _idIndexBlock->commit(MultiIndexBlock::kNoopOnCreateEachFn,
+                                                        MultiIndexBlock::kNoopOnCommitFn);
                     if (!status.isOK()) {
                         return status;
                     }
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 12df8c58003..e7cb394d9c5 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -260,42 +260,50 @@ void setOplogCollectionName(ServiceContext* service) {
 /**
  * Parse the given BSON array of BSON into a vector of BSON.
  */
-StatusWith<std::vector<BSONObj>> parseBSONArrayIntoVector(const BSONElement& bsonArrayElem) {
+StatusWith<std::vector<BSONObj>> parseBSONSpecsIntoVector(const BSONElement& bsonArrayElem,
+                                                          const NamespaceString& nss) {
     invariant(bsonArrayElem.type() == Array);
     std::vector<BSONObj> vec;
-    for (auto& bsonElem : bsonArrayElem.Obj()) {
+    for (auto& bsonElem : bsonArrayElem.Array()) {
         if (bsonElem.type() != BSONType::Object) {
             return {ErrorCodes::TypeMismatch,
                     str::stream() << "The elements of '" << bsonArrayElem.fieldName()
                                   << "' array must be objects, but found "
                                   << typeName(bsonElem.type())};
         }
-        BSONObjBuilder builder;
-        builder.append(bsonElem);
+        BSONObjBuilder builder(bsonElem.Obj());
+        builder.append("ns", nss.toString());
         vec.emplace_back(builder.obj());
     }
     return vec;
 }
 
 Status startIndexBuild(OperationContext* opCtx,
+                       const NamespaceString& nss,
                        const UUID& collUUID,
                        const UUID& indexBuildUUID,
                        const BSONElement& indexesElem,
                        OplogApplication::Mode mode) {
-    auto statusWithIndexes = parseBSONArrayIntoVector(indexesElem);
+    auto statusWithIndexes = parseBSONSpecsIntoVector(indexesElem, nss);
     if (!statusWithIndexes.isOK()) {
         return statusWithIndexes.getStatus();
     }
     return IndexBuildsCoordinator::get(opCtx)
-        ->startIndexBuild(opCtx, collUUID, statusWithIndexes.getValue(), indexBuildUUID)
+        ->startIndexBuild(opCtx,
+                          collUUID,
+                          statusWithIndexes.getValue(),
+                          indexBuildUUID,
+                          /* This oplog entry is only replicated for two-phase index builds */
+                          IndexBuildProtocol::kTwoPhase)
         .getStatus();
 }
 
 Status commitIndexBuild(OperationContext* opCtx,
+                        const NamespaceString& nss,
                         const UUID& indexBuildUUID,
                         const BSONElement& indexesElem,
                         OplogApplication::Mode mode) {
-    auto statusWithIndexes = parseBSONArrayIntoVector(indexesElem);
+    auto statusWithIndexes = parseBSONSpecsIntoVector(indexesElem, nss);
     if (!statusWithIndexes.isOK()) {
         return statusWithIndexes.getStatus();
     }
@@ -986,20 +994,20 @@ std::map<std::string, ApplyOpMetadata> opsMap = {
          uassert(ErrorCodes::BadValue,
                  "Error parsing 'startIndexBuild' oplog entry, missing required field "
                  "'indexBuildUUID'.",
-                 buildUUIDElem.eoo());
+                 !buildUUIDElem.eoo());
          UUID indexBuildUUID = uassertStatusOK(UUID::parse(buildUUIDElem));
 
          auto indexesElem = cmd.getField("indexes");
          uassert(ErrorCodes::BadValue,
                  "Error parsing 'startIndexBuild' oplog entry, missing required field 'indexes'.",
-                 indexesElem.eoo());
+                 !indexesElem.eoo());
          uassert(ErrorCodes::BadValue,
                  "Error parsing 'startIndexBuild' oplog entry, field 'indexes' must be an array.",
                  indexesElem.type() == Array);
 
          auto collUUID = uassertStatusOK(UUID::parse(ui));
 
-         return startIndexBuild(opCtx, collUUID, indexBuildUUID, indexesElem, mode);
+         return startIndexBuild(opCtx, nss, collUUID, indexBuildUUID, indexesElem, mode);
      }}},
     {"commitIndexBuild",
      {[](OperationContext* opCtx,
@@ -1042,22 +1050,24 @@ std::map<std::string, ApplyOpMetadata> opsMap = {
                  "commitIndexBuild value must be a string",
                  first.type() == mongo::String);
 
+         const NamespaceString nss(parseUUIDorNs(opCtx, ns, ui, cmd));
+
          auto buildUUIDElem = cmd.getField("indexBuildUUID");
          uassert(ErrorCodes::BadValue,
                  "Error parsing 'commitIndexBuild' oplog entry, missing required field "
                  "'indexBuildUUID'.",
-                 buildUUIDElem.eoo());
+                 !buildUUIDElem.eoo());
          UUID indexBuildUUID = uassertStatusOK(UUID::parse(buildUUIDElem));
 
          auto indexesElem = cmd.getField("indexes");
          uassert(ErrorCodes::BadValue,
                  "Error parsing 'commitIndexBuild' oplog entry, missing required field 'indexes'.",
-                 indexesElem.eoo());
+                 !indexesElem.eoo());
          uassert(ErrorCodes::BadValue,
                  "Error parsing 'commitIndexBuild' oplog entry, field 'indexes' must be an array.",
                  indexesElem.type() == Array);
 
-         return commitIndexBuild(opCtx, indexBuildUUID, indexesElem, mode);
+         return commitIndexBuild(opCtx, nss, indexBuildUUID, indexesElem, mode);
      }}},
     {"abortIndexBuild",
      {[](OperationContext* opCtx,
diff --git a/src/mongo/db/repl_index_build_state.h b/src/mongo/db/repl_index_build_state.h
index d21c2916a07..7183a0c7c74 100644
--- a/src/mongo/db/repl_index_build_state.h
+++ b/src/mongo/db/repl_index_build_state.h
@@ -35,6 +35,7 @@
 #include <vector>
 
 #include "mongo/bson/bsonobj.h"
+#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/write_concern_options.h"
@@ -57,12 +58,14 @@ struct ReplIndexBuildState {
                         const UUID& collUUID,
                         const std::string& dbName,
                         const std::vector<std::string> names,
-                        const std::vector<BSONObj>& specs)
+                        const std::vector<BSONObj>& specs,
+                        IndexBuildProtocol protocol)
         : buildUUID(indexBuildUUID),
           collectionUUID(collUUID),
           dbName(dbName),
           indexNames(names),
-          indexSpecs(specs) {
+          indexSpecs(specs),
+          protocol(protocol) {
         // Verify that the given index names and index specs match.
         invariant(names.size() == specs.size());
         for (auto& spec : specs) {
@@ -91,7 +94,7 @@ struct ReplIndexBuildState {
 
     // Whether to do a two phase index build or a single phase index build like in v4.0. The FCV
     // at the start of the index build will determine this setting.
-    bool twoPhaseIndexBuild = false;
+    IndexBuildProtocol protocol;
 
     // Protects the state below.
     mutable stdx::mutex mutex;
diff --git a/src/mongo/db/system_index.cpp b/src/mongo/db/system_index.cpp
index b55f9ff203c..9fd964ea291 100644
--- a/src/mongo/db/system_index.cpp
+++ b/src/mongo/db/system_index.cpp
@@ -123,8 +123,8 @@ SharedSemiFuture<ReplIndexBuildState::IndexCatalogStats> generateSystemIndexForE
 
         UUID buildUUID = UUID::gen();
         IndexBuildsCoordinator* indexBuildsCoord = IndexBuildsCoordinator::get(opCtx);
-        auto indexBuildFuture = uassertStatusOK(
-            indexBuildsCoord->startIndexBuild(opCtx, collectionUUID, {indexSpec}, buildUUID));
+        auto indexBuildFuture = uassertStatusOK(indexBuildsCoord->startIndexBuild(
+            opCtx, collectionUUID, {indexSpec}, buildUUID, IndexBuildProtocol::kSinglePhase));
         return indexBuildFuture;
     } catch (const DBException& e) {
         severe() << "Failed to regenerate index for " << ns << ". Exception: " << e.what();