SERVER-32206 timestamp catalog change to declare index multikey

16 files changed, 790 insertions, 121 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index bf72c48cc2d..442ae9ff8a6 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -451,6 +451,7 @@ env.Library(
     target='service_context',
     source=[
         'client.cpp',
+        'multi_key_path_tracker.cpp',
         'operation_context.cpp',
         'service_context.cpp',
         'service_context_noop.cpp',
diff --git a/src/mongo/db/catalog/index_catalog_entry.h b/src/mongo/db/catalog/index_catalog_entry.h
index 13edf2b4c21..fecfd197bb1 100644
--- a/src/mongo/db/catalog/index_catalog_entry.h
+++ b/src/mongo/db/catalog/index_catalog_entry.h
@@ -214,6 +214,11 @@ public:
      * If this index supports path-level multikey tracking, then 'multikeyPaths' must be a vector
      * with size equal to the number of elements in the index key pattern. Additionally, at least
      * one path component of the indexed fields must cause this index to be multikey.
+     *
+     * If isTrackingMultikeyPathInfo() is set on the OperationContext's MultikeyPathTracker,
+     * then after we confirm that we actually need to set the index as multikey, we will save the
+     * namespace, index name, and multikey paths on the OperationContext rather than set the index
+     * as multikey here.
      */
     void setMultikey(OperationContext* const opCtx, const MultikeyPaths& multikeyPaths) {
         return this->_impl().setMultikey(opCtx, multikeyPaths);
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.cpp b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
index fdaee922b71..cbea2f00a00 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
@@ -44,6 +44,7 @@
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/matcher/expression.h"
 #include "mongo/db/matcher/expression_parser.h"
+#include "mongo/db/multi_key_path_tracker.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/service_context.h"
@@ -203,38 +204,6 @@ void IndexCatalogEntryImpl::setHead(OperationContext* opCtx, RecordId newHead) {
     _head = newHead;
 }
 
-
-/**
- * RAII class, which associates a new RecoveryUnit with an OperationContext for the purposes
- * of simulating a side-transaction. Takes ownership of the new recovery unit and frees it at
- * destruction time.
- */
-class RecoveryUnitSwap {
-public:
-    RecoveryUnitSwap(OperationContext* opCtx, RecoveryUnit* newRecoveryUnit)
-        : _opCtx(opCtx),
-          _oldRecoveryUnit(_opCtx->releaseRecoveryUnit()),
-          _oldRecoveryUnitState(
-              _opCtx->setRecoveryUnit(newRecoveryUnit, OperationContext::kNotInUnitOfWork)),
-          _newRecoveryUnit(newRecoveryUnit) {}
-
-    ~RecoveryUnitSwap() {
-        _opCtx->releaseRecoveryUnit();
-        _opCtx->setRecoveryUnit(_oldRecoveryUnit, _oldRecoveryUnitState);
-    }
-
-private:
-    // Not owned
-    OperationContext* const _opCtx;
-
-    // Owned, but life-time is not controlled
-    RecoveryUnit* const _oldRecoveryUnit;
-    OperationContext::RecoveryUnitState const _oldRecoveryUnitState;
-
-    // Owned and life-time is controlled
-    const std::unique_ptr<RecoveryUnit> _newRecoveryUnit;
-};
-
 void IndexCatalogEntryImpl::setMultikey(OperationContext* opCtx,
                                         const MultikeyPaths& multikeyPaths) {
     if (!_indexTracksPathLevelMultikeyInfo && isMultikey()) {
@@ -267,56 +236,51 @@ void IndexCatalogEntryImpl::setMultikey(OperationContext* opCtx,
         }
     }
 
-    {
-        // Only one thread should set the multi-key value per collection, because the metadata for a
-        // collection is one large document.
-        Lock::ResourceLock collMDLock(
-            opCtx->lockState(), ResourceId(RESOURCE_METADATA, _ns), MODE_X);
-
-        if (!_indexTracksPathLevelMultikeyInfo && isMultikey()) {
-            // It's possible that we raced with another thread when acquiring the MD lock. If the
-            // index is already set as multikey and we don't have any path-level information to
-            // update, then there's nothing more for us to do.
-            return;
-        }
-
-        // This effectively emulates a side-transaction off the main transaction, which invoked
-        // setMultikey. The reason we need is to avoid artificial WriteConflicts, which happen with
-        // snapshot isolation.
-        {
-            StorageEngine* storageEngine = getGlobalServiceContext()->getGlobalStorageEngine();
-
-            // This ensures that the recovery unit is not swapped for engines that do not support
-            // database level locking.
-            std::unique_ptr<RecoveryUnitSwap> ruSwap;
-            if (storageEngine->supportsDBLocking()) {
-                ruSwap =
-                    stdx::make_unique<RecoveryUnitSwap>(opCtx, storageEngine->newRecoveryUnit());
-            }
-
-            WriteUnitOfWork wuow(opCtx);
-
-            // It's possible that the index type (e.g. ascending/descending index) supports tracking
-            // path-level multikey information, but this particular index doesn't.
-            // CollectionCatalogEntry::setIndexIsMultikey() requires that we discard the path-level
-            // multikey information in order to avoid unintentionally setting path-level multikey
-            // information on an index created before 3.4.
-            if (_collection->setIndexIsMultikey(
-                    opCtx,
-                    _descriptor->indexName(),
-                    _indexTracksPathLevelMultikeyInfo ? multikeyPaths : MultikeyPaths{})) {
-                if (_infoCache) {
-                    LOG(1) << _ns << ": clearing plan cache - index " << _descriptor->keyPattern()
-                           << " set to multi key.";
-                    _infoCache->clearQueryCache();
-                }
-            }
+    MultikeyPaths paths = _indexTracksPathLevelMultikeyInfo ? multikeyPaths : MultikeyPaths{};
+
+    // On a primary, we can simply assign this write the same timestamp as the index creation,
+    // insert, or update that caused this index to become multikey. This is because if two
+    // operations concurrently try to change the index to be multikey, they will conflict and the
+    // loser will simply get a higher timestamp and go into the oplog second with a later optime.
+    //
+    // On a secondary, writes must get the timestamp of their oplog entry, and the multikey change
+    // must occur before the timestamp of the earliest write that makes the index multikey.
+    // Secondaries only serialize writes by document, not by collection. If two inserts that both
+    // make an index multikey are applied out of order, changing the index to multikey at the
+    // insert timestamps would change the index to multikey at the later timestamp, which would be
+    // wrong. To prevent this, rather than setting the index to be multikey here, we add the
+    // necessary information to the OperationContext and do the write at the timestamp of the
+    // beginning of the batch.
+    //
+    // One exception to this rule is for background indexes. Background indexes are built using
+    // a different OperationContext and thus this information would be ignored. Background index
+    // builds happen concurrently though and thus the multikey write can safely occur at the
+    // current clock time. Once a background index is committed, if a future write makes
+    // it multikey, that write will be marked as "isTrackingMultikeyPathInfo" on the applier's
+    // OperationContext and we can safely defer that write to the end of the batch.
+    if (MultikeyPathTracker::get(opCtx).isTrackingMultikeyPathInfo()) {
+        MultikeyPathInfo info;
+        info.nss = _collection->ns();
+        info.indexName = _descriptor->indexName();
+        info.multikeyPaths = paths;
+        MultikeyPathTracker::get(opCtx).addMultikeyPathInfo(info);
+        return;
+    }
 
-            wuow.commit();
+    // It's possible that the index type (e.g. ascending/descending index) supports tracking
+    // path-level multikey information, but this particular index doesn't.
+    // CollectionCatalogEntry::setIndexIsMultikey() requires that we discard the path-level
+    // multikey information in order to avoid unintentionally setting path-level multikey
+    // information on an index created before 3.4.
+    if (_collection->setIndexIsMultikey(opCtx, _descriptor->indexName(), paths)) {
+        if (_infoCache) {
+            LOG(1) << _ns << ": clearing plan cache - index " << _descriptor->keyPattern()
+                   << " set to multi key.";
+            _infoCache->clearQueryCache();
         }
     }
 
-    _isMultikey.store(true);
+    opCtx->recoveryUnit()->onCommit([this] { _isMultikey.store(true); });
 
     if (_indexTracksPathLevelMultikeyInfo) {
         stdx::lock_guard<stdx::mutex> lk(_indexMultikeyPathsMutex);
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.h b/src/mongo/db/catalog/index_catalog_entry_impl.h
index 2a03cd363d6..4885702dc90 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.h
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.h
@@ -138,6 +138,11 @@ public:
      * If this index supports path-level multikey tracking, then 'multikeyPaths' must be a vector
      * with size equal to the number of elements in the index key pattern. Additionally, at least
      * one path component of the indexed fields must cause this index to be multikey.
+     *
+     * If isTrackingMultikeyPathInfo() is set on the OperationContext's MultikeyPathTracker,
+     * then after we confirm that we actually need to set the index as multikey, we will save the
+     * namespace, index name, and multikey paths on the OperationContext rather than set the index
+     * as multikey here.
      */
     void setMultikey(OperationContext* opCtx, const MultikeyPaths& multikeyPaths) final;
 
diff --git a/src/mongo/db/index/index_access_method.cpp b/src/mongo/db/index/index_access_method.cpp
index 7abcca26aad..7bc0c321b62 100644
--- a/src/mongo/db/index/index_access_method.cpp
+++ b/src/mongo/db/index/index_access_method.cpp
@@ -46,12 +46,14 @@
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/keypattern.h"
+#include "mongo/db/multi_key_path_tracker.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/server_parameters.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/util/log.h"
 #include "mongo/util/progress_meter.h"
+#include "mongo/util/scopeguard.h"
 
 namespace mongo {
 
@@ -471,6 +473,13 @@ Status IndexAccessMethod::commitBulk(OperationContext* opCtx,
                                      bool mayInterrupt,
                                      bool dupsAllowed,
                                      set<RecordId>* dupsToDrop) {
+    // Do not track multikey path info for index builds.
+    ScopeGuard restartTracker =
+        MakeGuard([opCtx] { MultikeyPathTracker::get(opCtx).startTrackingMultikeyPathInfo(); });
+    if (!MultikeyPathTracker::get(opCtx).isTrackingMultikeyPathInfo()) {
+        restartTracker.Dismiss();
+    }
+    MultikeyPathTracker::get(opCtx).stopTrackingMultikeyPathInfo();
     Timer timer;
 
     std::unique_ptr<BulkBuilder::Sorter::Iterator> i(bulk->_sorter->done());
@@ -551,6 +560,10 @@ Status IndexAccessMethod::commitBulk(OperationContext* opCtx,
     return Status::OK();
 }
 
+void IndexAccessMethod::setIndexIsMultikey(OperationContext* opCtx, MultikeyPaths paths) {
+    _btreeState->setMultikey(opCtx, paths);
+}
+
 void IndexAccessMethod::getKeys(const BSONObj& obj,
                                 GetKeysMode mode,
                                 BSONObjSet* keys,
diff --git a/src/mongo/db/index/index_access_method.h b/src/mongo/db/index/index_access_method.h
index b5d8340b28e..32b821379c8 100644
--- a/src/mongo/db/index/index_access_method.h
+++ b/src/mongo/db/index/index_access_method.h
@@ -190,6 +190,11 @@ public:
      */
     Status compact(OperationContext* opCtx);
 
+    /**
+     * Sets this index as multikey with the provided paths.
+     */
+    void setIndexIsMultikey(OperationContext* opCtx, MultikeyPaths paths);
+
     //
     // Bulk operations support
     //
diff --git a/src/mongo/db/multi_key_path_tracker.cpp b/src/mongo/db/multi_key_path_tracker.cpp
new file mode 100644
index 00000000000..ecb1348d92a
--- /dev/null
+++ b/src/mongo/db/multi_key_path_tracker.cpp
@@ -0,0 +1,59 @@
+/**
+ *    Copyright (C) 2018 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/multi_key_path_tracker.h"
+
+namespace mongo {
+
+const OperationContext::Decoration<MultikeyPathTracker> MultikeyPathTracker::get =
+    OperationContext::declareDecoration<MultikeyPathTracker>();
+
+void MultikeyPathTracker::addMultikeyPathInfo(MultikeyPathInfo info) {
+    invariant(_trackMultikeyPathInfo);
+    _multikeyPathInfo.emplace_back(info);
+}
+
+const std::vector<MultikeyPathInfo>& MultikeyPathTracker::getMultikeyPathInfo() const {
+    return _multikeyPathInfo;
+}
+
+void MultikeyPathTracker::startTrackingMultikeyPathInfo() {
+    _trackMultikeyPathInfo = true;
+}
+
+void MultikeyPathTracker::stopTrackingMultikeyPathInfo() {
+    _trackMultikeyPathInfo = false;
+}
+
+bool MultikeyPathTracker::isTrackingMultikeyPathInfo() const {
+    return _trackMultikeyPathInfo;
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/multi_key_path_tracker.h b/src/mongo/db/multi_key_path_tracker.h
new file mode 100644
index 00000000000..781ec9924cc
--- /dev/null
+++ b/src/mongo/db/multi_key_path_tracker.h
@@ -0,0 +1,94 @@
+/**
+ *    Copyright (C) 2018 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include <string>
+
+#include "mongo/db/index/multikey_paths.h"
+#include "mongo/db/operation_context.h"
+
+namespace mongo {
+
+struct MultikeyPathInfo {
+    NamespaceString nss;
+    std::string indexName;
+    MultikeyPaths multikeyPaths;
+};
+
+/**
+ * An OperationContext decoration that tracks which indexes should be made multikey. This is used
+ * by IndexCatalogEntryImpl::setMultikey() to track what indexes should be set as multikey during
+ * secondary oplog application. This both marks if the multikey path information should be tracked
+ * instead of set immediately and saves the multikey path information for later if needed.
+ */
+class MultikeyPathTracker {
+public:
+    static const OperationContext::Decoration<MultikeyPathTracker> get;
+
+    // Decoration requires a default constructor.
+    MultikeyPathTracker() = default;
+
+    /**
+     * Appends the provided multikey path information to the list of indexes to set as multikey
+     * after the current replication batch finishes.
+     * Must call startTrackingMultikeyPathInfo() first.
+     */
+    void addMultikeyPathInfo(MultikeyPathInfo info);
+
+    /**
+     * Returns the multikey path information that has been saved.
+     */
+    const std::vector<MultikeyPathInfo>& getMultikeyPathInfo() const;
+
+    /**
+     * Specifies that we should track multikey path information on this MultikeyPathTracker. This is
+     * only expected to be called during oplog application on secondaries. We cannot simply check
+     * 'canAcceptWritesFor' because background index builds use their own OperationContext and
+     * cannot store their multikey path info here.
+     */
+    void startTrackingMultikeyPathInfo();
+
+    /**
+     * Specifies to stop tracking multikey path information.
+     */
+    void stopTrackingMultikeyPathInfo();
+
+    /**
+     * Returns if we've called startTrackingMultikeyPathInfo() and not yet called
+     * stopTrackingMultikeyPathInfo().
+     */
+    bool isTrackingMultikeyPathInfo() const;
+
+
+private:
+    std::vector<MultikeyPathInfo> _multikeyPathInfo;
+    bool _trackMultikeyPathInfo = false;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/db/repl/storage_interface.h b/src/mongo/db/repl/storage_interface.h
index 26812b5ea84..bade022996a 100644
--- a/src/mongo/db/repl/storage_interface.h
+++ b/src/mongo/db/repl/storage_interface.h
@@ -179,6 +179,15 @@ public:
                                     bool stayTemp) = 0;
 
     /**
+     * Sets the given index on the given namespace as multikey with the given paths. Does the
+     * write at the provided timestamp.
+     */
+    virtual Status setIndexIsMultikey(OperationContext* opCtx,
+                                      const NamespaceString& nss,
+                                      const std::string& indexName,
+                                      const MultikeyPaths& paths,
+                                      Timestamp ts) = 0;
+    /**
      * Drops all databases except "local".
      */
     virtual Status dropReplicatedDatabases(OperationContext* opCtx) = 0;
diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index 7657dae752b..0c9f7a399a0 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -499,6 +499,46 @@ Status StorageInterfaceImpl::renameCollection(OperationContext* opCtx,
     });
 }
 
+Status StorageInterfaceImpl::setIndexIsMultikey(OperationContext* opCtx,
+                                                const NamespaceString& nss,
+                                                const std::string& indexName,
+                                                const MultikeyPaths& paths,
+                                                Timestamp ts) {
+    if (ts.isNull()) {
+        return Status(ErrorCodes::InvalidOptions,
+                      str::stream() << "Cannot set index " << indexName << " on " << nss.ns()
+                                    << " as multikey at null timestamp");
+    }
+
+    return writeConflictRetry(opCtx, "StorageInterfaceImpl::setIndexIsMultikey", nss.ns(), [&] {
+        AutoGetCollection autoColl(opCtx, nss, MODE_X);
+        auto collectionResult = getCollection(
+            autoColl, nss, "The collection must exist before setting an index to multikey.");
+        if (!collectionResult.isOK()) {
+            return collectionResult.getStatus();
+        }
+        auto collection = collectionResult.getValue();
+
+        WriteUnitOfWork wunit(opCtx);
+        auto tsResult = opCtx->recoveryUnit()->setTimestamp(ts);
+        if (!tsResult.isOK()) {
+            return tsResult;
+        }
+
+        auto idx = collection->getIndexCatalog()->findIndexByName(
+            opCtx, indexName, true /* includeUnfinishedIndexes */);
+        if (!idx) {
+            return Status(ErrorCodes::IndexNotFound,
+                          str::stream() << "Could not find index " << indexName << " in "
+                                        << nss.ns()
+                                        << " to set to multikey.");
+        }
+        collection->getIndexCatalog()->getIndex(idx)->setIndexIsMultikey(opCtx, paths);
+        wunit.commit();
+        return Status::OK();
+    });
+}
+
 namespace {
 
 /**
diff --git a/src/mongo/db/repl/storage_interface_impl.h b/src/mongo/db/repl/storage_interface_impl.h
index 9b7bf9f6690..d5545e5270f 100644
--- a/src/mongo/db/repl/storage_interface_impl.h
+++ b/src/mongo/db/repl/storage_interface_impl.h
@@ -93,6 +93,12 @@ public:
                             const NamespaceString& toNS,
                             bool stayTemp) override;
 
+    Status setIndexIsMultikey(OperationContext* opCtx,
+                              const NamespaceString& nss,
+                              const std::string& indexName,
+                              const MultikeyPaths& paths,
+                              Timestamp ts) override;
+
     StatusWith<std::vector<BSONObj>> findDocuments(OperationContext* opCtx,
                                                    const NamespaceString& nss,
                                                    boost::optional<StringData> indexName,
diff --git a/src/mongo/db/repl/storage_interface_impl_test.cpp b/src/mongo/db/repl/storage_interface_impl_test.cpp
index 8a7fbd9e0d4..fc6c159c51d 100644
--- a/src/mongo/db/repl/storage_interface_impl_test.cpp
+++ b/src/mongo/db/repl/storage_interface_impl_test.cpp
@@ -126,6 +126,27 @@ void createCollection(OperationContext* opCtx,
 }
 
 /**
+ * Create an index on the given collection. Returns the number of indexes that exist on the
+ * collection after the given index is created.
+ */
+int createIndexForColl(OperationContext* opCtx, NamespaceString nss, BSONObj indexSpec) {
+    Lock::DBLock dbLock(opCtx, nss.db(), MODE_X);
+    AutoGetCollection autoColl(opCtx, nss, MODE_X);
+    auto coll = autoColl.getCollection();
+
+    MultiIndexBlock indexer(opCtx, coll);
+    ASSERT_OK(indexer.init(indexSpec).getStatus());
+
+    WriteUnitOfWork wunit(opCtx);
+    indexer.commit();
+    wunit.commit();
+
+    auto indexCatalog = coll->getIndexCatalog();
+    ASSERT(indexCatalog);
+    return indexCatalog->numIndexesReady(opCtx);
+}
+
+/**
  * Creates an oplog entry with given optime.
  */
 TimestampedBSONObj makeOplogEntry(OpTime opTime) {
@@ -2413,4 +2434,62 @@ TEST_F(StorageInterfaceImplTest, GetCollectionSizeReturnsCollectionSize) {
     ASSERT_NOT_EQUALS(0UL, size);
 }
 
+TEST_F(StorageInterfaceImplTest, SetIndexIsMultikeyReturnsNamespaceNotFoundForMissingDatabase) {
+    auto opCtx = getOperationContext();
+    StorageInterfaceImpl storage;
+    auto nss = makeNamespace(_agent);
+    ASSERT_EQUALS(ErrorCodes::NamespaceNotFound,
+                  storage.setIndexIsMultikey(opCtx, nss, "foo", {}, Timestamp(3, 3)));
+}
+
+TEST_F(StorageInterfaceImplTest, SetIndexIsMultikeyReturnsNamespaceNotFoundForMissingCollection) {
+    auto opCtx = getOperationContext();
+    StorageInterfaceImpl storage;
+    auto nss = makeNamespace(_agent);
+    NamespaceString wrongColl(nss.db(), "wrongColl"_sd);
+    ASSERT_OK(storage.createCollection(opCtx, nss, CollectionOptions()));
+    ASSERT_EQUALS(ErrorCodes::NamespaceNotFound,
+                  storage.setIndexIsMultikey(opCtx, wrongColl, "foo", {}, Timestamp(3, 3)));
+}
+
+TEST_F(StorageInterfaceImplTest, SetIndexIsMultikeyReturnsIndexNotFoundForMissingIndex) {
+    auto opCtx = getOperationContext();
+    StorageInterfaceImpl storage;
+    auto nss = makeNamespace(_agent);
+    ASSERT_OK(storage.createCollection(opCtx, nss, CollectionOptions()));
+    ASSERT_EQUALS(ErrorCodes::IndexNotFound,
+                  storage.setIndexIsMultikey(opCtx, nss, "foo", {}, Timestamp(3, 3)));
+}
+
+TEST_F(StorageInterfaceImplTest, SetIndexIsMultikeyReturnsInvalidOptionsForNullTimestamp) {
+    auto opCtx = getOperationContext();
+    StorageInterfaceImpl storage;
+    auto nss = makeNamespace(_agent);
+    ASSERT_OK(storage.createCollection(opCtx, nss, CollectionOptions()));
+    ASSERT_EQUALS(ErrorCodes::InvalidOptions,
+                  storage.setIndexIsMultikey(opCtx, nss, "foo", {}, Timestamp()));
+}
+
+TEST_F(StorageInterfaceImplTest, SetIndexIsMultikeySucceeds) {
+    auto opCtx = getOperationContext();
+    StorageInterfaceImpl storage;
+    auto nss = makeNamespace(_agent);
+    ASSERT_OK(storage.createCollection(opCtx, nss, CollectionOptions()));
+
+    auto indexName = "a_b_1";
+    auto indexSpec =
+        BSON("name" << indexName << "ns" << nss.ns() << "key" << BSON("a.b" << 1) << "v"
+                    << static_cast<int>(kIndexVersion));
+    ASSERT_EQUALS(createIndexForColl(opCtx, nss, indexSpec), 2);
+
+    MultikeyPaths paths = {{1}};
+    ASSERT_OK(storage.setIndexIsMultikey(opCtx, nss, indexName, paths, Timestamp(3, 3)));
+    AutoGetCollectionForReadCommand autoColl(opCtx, nss);
+    ASSERT_TRUE(autoColl.getCollection());
+    auto indexCatalog = autoColl.getCollection()->getIndexCatalog();
+    ASSERT(indexCatalog->isMultikey(opCtx, indexCatalog->findIndexByName(opCtx, indexName)));
+    ASSERT(paths ==
+           indexCatalog->getMultikeyPaths(opCtx, indexCatalog->findIndexByName(opCtx, indexName)));
+}
+
 }  // namespace
diff --git a/src/mongo/db/repl/storage_interface_mock.h b/src/mongo/db/repl/storage_interface_mock.h
index a3c9735220d..3eff258e2c7 100644
--- a/src/mongo/db/repl/storage_interface_mock.h
+++ b/src/mongo/db/repl/storage_interface_mock.h
@@ -189,6 +189,15 @@ public:
         return Status{ErrorCodes::IllegalOperation, "renameCollection not implemented."};
     }
 
+    Status setIndexIsMultikey(OperationContext* opCtx,
+                              const NamespaceString& nss,
+                              const std::string& indexName,
+                              const MultikeyPaths& paths,
+                              Timestamp ts) override {
+
+        return Status{ErrorCodes::IllegalOperation, "setIndexIsMultikey not implemented."};
+    }
+
     StatusWith<std::vector<BSONObj>> findDocuments(OperationContext* opCtx,
                                                    const NamespaceString& nss,
                                                    boost::optional<StringData> indexName,
diff --git a/src/mongo/db/repl/sync_tail.cpp b/src/mongo/db/repl/sync_tail.cpp
index fd6b08e9845..1ca318f3ad3 100644
--- a/src/mongo/db/repl/sync_tail.cpp
+++ b/src/mongo/db/repl/sync_tail.cpp
@@ -54,6 +54,7 @@
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/logical_session_id.h"
+#include "mongo/db/multi_key_path_tracker.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/prefetch.h"
 #include "mongo/db/query/query_knobs.h"
@@ -706,6 +707,9 @@ SessionRecordMap getLatestSessionRecords(const MultiApplier::Operations& ops) {
 
 }  // namespace
 
+OpTime SyncTail::multiApply_forTest(OperationContext* opCtx, MultiApplier::Operations ops) {
+    return multiApply(opCtx, ops);
+}
 /**
  * Applies a batch of oplog entries by writing the oplog entries to the local oplog and then using
  * a set of threads to apply the operations. If the batch application is successful, returns the
@@ -721,8 +725,26 @@ OpTime SyncTail::multiApply(OperationContext* opCtx, MultiApplier::Operations op
         // _applyFunc() will throw or abort on error, so we return OK here.
         return Status::OK();
     };
-    return fassertStatusOK(
+    Timestamp firstTimeInBatch = ops.front().getTimestamp();
+
+    OpTime finalOpTime = fassertStatusOK(
         34437, repl::multiApply(opCtx, _writerPool.get(), std::move(ops), applyOperation));
+
+    invariant(!MultikeyPathTracker::get(opCtx).isTrackingMultikeyPathInfo());
+    // Set any indexes to multikey that this batch ignored.
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+    for (MultikeyPathInfo info : _multikeyPathInfo) {
+        // We timestamp every multikey write with the first timestamp in the batch. It is always
+        // safe to set an index as multikey too early, just not too late. We conservatively pick
+        // the first timestamp in the batch since we do not have enough information to find out
+        // the timestamp of the first write that set the given multikey path.
+        fassertStatusOK(50686,
+                        StorageInterface::get(opCtx)->setIndexIsMultikey(
+                            opCtx, info.nss, info.indexName, info.multikeyPaths, firstTimeInBatch));
+    }
+    _multikeyPathInfo.clear();
+
+    return finalOpTime;
 }
 
 namespace {
@@ -1225,8 +1247,16 @@ bool SyncTail::fetchAndInsertMissingDocument(OperationContext* opCtx,
     });
 }
 
+void SyncTail::addMultikeyPathInfo(std::vector<MultikeyPathInfo> infoList) {
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+    _multikeyPathInfo.reserve(_multikeyPathInfo.size() + infoList.size());
+    for (MultikeyPathInfo info : infoList) {
+        _multikeyPathInfo.emplace_back(info);
+    }
+}
+
 // This free function is used by the writer threads to apply each op
-void multiSyncApply(MultiApplier::OperationPtrs* ops, SyncTail*) {
+void multiSyncApply(MultiApplier::OperationPtrs* ops, SyncTail* st) {
     initializeWriterThread();
     auto opCtx = cc().makeOperationContext();
     auto syncApply = [](
@@ -1234,7 +1264,14 @@ void multiSyncApply(MultiApplier::OperationPtrs* ops, SyncTail*) {
         return SyncTail::syncApply(opCtx, op, oplogApplicationMode);
     };
 
+    ON_BLOCK_EXIT(
+        [&opCtx] { MultikeyPathTracker::get(opCtx.get()).stopTrackingMultikeyPathInfo(); });
+    MultikeyPathTracker::get(opCtx.get()).startTrackingMultikeyPathInfo();
     fassertNoTrace(16359, multiSyncApply_noAbort(opCtx.get(), ops, syncApply));
+
+    if (!MultikeyPathTracker::get(opCtx.get()).getMultikeyPathInfo().empty()) {
+        st->addMultikeyPathInfo(MultikeyPathTracker::get(opCtx.get()).getMultikeyPathInfo());
+    }
 }
 
 Status multiSyncApply_noAbort(OperationContext* opCtx,
@@ -1414,14 +1451,6 @@ Status multiSyncApply_noAbort(OperationContext* opCtx,
     return Status::OK();
 }
 
-// This free function is used by the initial sync writer threads to apply each op
-void multiInitialSyncApply_abortOnFailure(MultiApplier::OperationPtrs* ops, SyncTail* st) {
-    initializeWriterThread();
-    auto opCtx = cc().makeOperationContext();
-    AtomicUInt32 fetchCount(0);
-    fassertNoTrace(15915, multiInitialSyncApply_noAbort(opCtx.get(), ops, st, &fetchCount));
-}
-
 Status multiInitialSyncApply(MultiApplier::OperationPtrs* ops,
                              SyncTail* st,
                              AtomicUInt32* fetchCount) {
@@ -1436,43 +1465,59 @@ Status multiInitialSyncApply_noAbort(OperationContext* opCtx,
                                      AtomicUInt32* fetchCount) {
     UnreplicatedWritesBlock uwb(opCtx);
     DisableDocumentValidation validationDisabler(opCtx);
+    {  // Ensure that the MultikeyPathTracker stops tracking paths.
+        ON_BLOCK_EXIT([opCtx] { MultikeyPathTracker::get(opCtx).stopTrackingMultikeyPathInfo(); });
+        MultikeyPathTracker::get(opCtx).startTrackingMultikeyPathInfo();
+
+        // allow us to get through the magic barrier
+        opCtx->lockState()->setShouldConflictWithSecondaryBatchApplication(false);
+
+        for (auto it = ops->begin(); it != ops->end(); ++it) {
+            auto& entry = **it;
+            try {
+                const Status s =
+                    SyncTail::syncApply(opCtx, entry.raw, OplogApplication::Mode::kInitialSync);
+                if (!s.isOK()) {
+                    // In initial sync, update operations can cause documents to be missed during
+                    // collection cloning. As a result, it is possible that a document that we need
+                    // to update is not present locally. In that case we fetch the document from the
+                    // sync source.
+                    if (s != ErrorCodes::UpdateOperationFailed) {
+                        error() << "Error applying operation: " << redact(s) << " ("
+                                << redact(entry.toBSON()) << ")";
+                        return s;
+                    }
 
-    // allow us to get through the magic barrier
-    opCtx->lockState()->setShouldConflictWithSecondaryBatchApplication(false);
-
-    for (auto it = ops->begin(); it != ops->end(); ++it) {
-        auto& entry = **it;
-        try {
-            const Status s =
-                SyncTail::syncApply(opCtx, entry.raw, OplogApplication::Mode::kInitialSync);
-            if (!s.isOK()) {
-                // In initial sync, update operations can cause documents to be missed during
-                // collection cloning. As a result, it is possible that a document that we need to
-                // update is not present locally. In that case we fetch the document from the
-                // sync source.
-                if (s != ErrorCodes::UpdateOperationFailed) {
-                    error() << "Error applying operation: " << redact(s) << " ("
-                            << redact(entry.toBSON()) << ")";
-                    return s;
+                    // We might need to fetch the missing docs from the sync source.
+                    fetchCount->fetchAndAdd(1);
+                    st->fetchAndInsertMissingDocument(opCtx, entry);
+                }
+            } catch (const DBException& e) {
+                // SERVER-24927 If we have a NamespaceNotFound exception, then this document will be
+                // dropped before initial sync ends anyways and we should ignore it.
+                if (e.code() == ErrorCodes::NamespaceNotFound && entry.isCrudOpType()) {
+                    continue;
                 }
 
-                // We might need to fetch the missing docs from the sync source.
-                fetchCount->fetchAndAdd(1);
-                st->fetchAndInsertMissingDocument(opCtx, entry);
+                severe() << "writer worker caught exception: " << causedBy(redact(e))
+                         << " on: " << redact(entry.toBSON());
+                return e.toStatus();
             }
-        } catch (const DBException& e) {
-            // SERVER-24927 If we have a NamespaceNotFound exception, then this document will be
-            // dropped before initial sync ends anyways and we should ignore it.
-            if (e.code() == ErrorCodes::NamespaceNotFound && entry.isCrudOpType()) {
-                continue;
-            }
-
-            severe() << "writer worker caught exception: " << causedBy(redact(e))
-                     << " on: " << redact(entry.toBSON());
-            return e.toStatus();
         }
     }
 
+    invariant(!MultikeyPathTracker::get(opCtx).isTrackingMultikeyPathInfo());
+    // Set any indexes to multikey that this batch ignored.
+    Timestamp firstTimeInBatch = ops->front()->getTimestamp();
+    for (MultikeyPathInfo info : MultikeyPathTracker::get(opCtx).getMultikeyPathInfo()) {
+        // We timestamp every multikey write with the first timestamp in the batch. It is always
+        // safe to set an index as multikey too early, just not too late. We conservatively pick
+        // the first timestamp in the batch since we do not have enough information to find out
+        // the timestamp of the first write that set the given multikey path.
+        fassertStatusOK(50685,
+                        StorageInterface::get(opCtx)->setIndexIsMultikey(
+                            opCtx, info.nss, info.indexName, info.multikeyPaths, firstTimeInBatch));
+    }
     return Status::OK();
 }
 
diff --git a/src/mongo/db/repl/sync_tail.h b/src/mongo/db/repl/sync_tail.h
index c5c4d83002b..c4436bb4b1c 100644
--- a/src/mongo/db/repl/sync_tail.h
+++ b/src/mongo/db/repl/sync_tail.h
@@ -43,6 +43,7 @@ namespace mongo {
 
 class Database;
 class OperationContext;
+struct MultikeyPathInfo;
 
 namespace repl {
 class BackgroundSync;
@@ -224,6 +225,17 @@ public:
 
     static AtomicInt32 replBatchLimitOperations;
 
+    /**
+     * Adds the given multikey path information to the list of indexes to make multikey at the
+     * end of the current batch.
+     */
+    void addMultikeyPathInfo(std::vector<MultikeyPathInfo> infoList);
+
+    /**
+     * Passthrough function to test multiApply.
+     */
+    OpTime multiApply_forTest(OperationContext* opCtx, MultiApplier::Operations ops);
+
 protected:
     static const unsigned int replBatchLimitBytes = 100 * 1024 * 1024;
     static const int replBatchLimitSeconds = 1;
@@ -244,6 +256,12 @@ private:
 
     // persistent pool of worker threads for writing ops to the databases
     std::unique_ptr<OldThreadPool> _writerPool;
+
+    // Protects member variables below.
+    mutable stdx::mutex _mutex;
+
+    // Maintains the information for all indexes that must be set as multikey in the current batch.
+    std::vector<MultikeyPathInfo> _multikeyPathInfo;
 };
 
 /**
@@ -266,9 +284,6 @@ StatusWith<OpTime> multiApply(OperationContext* opCtx,
 // operations because the OperationPtrs container contains const pointers.
 void multiSyncApply(MultiApplier::OperationPtrs* ops, SyncTail* st);
 
-// Used by 3.2 initial sync.
-void multiInitialSyncApply_abortOnFailure(MultiApplier::OperationPtrs* ops, SyncTail* st);
-
 // Used by 3.4 initial sync.
 Status multiInitialSyncApply(MultiApplier::OperationPtrs* ops,
                              SyncTail* st,
diff --git a/src/mongo/dbtests/storage_timestamp_tests.cpp b/src/mongo/dbtests/storage_timestamp_tests.cpp
index d30157e4d4e..9d4c766f465 100644
--- a/src/mongo/dbtests/storage_timestamp_tests.cpp
+++ b/src/mongo/dbtests/storage_timestamp_tests.cpp
@@ -45,26 +45,35 @@
 #include "mongo/db/dbhelpers.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/logical_clock.h"
+#include "mongo/db/multi_key_path_tracker.h"
 #include "mongo/db/op_observer_impl.h"
 #include "mongo/db/op_observer_registry.h"
 #include "mongo/db/repl/apply_ops.h"
 #include "mongo/db/repl/drop_pending_collection_reaper.h"
 #include "mongo/db/repl/oplog.h"
 #include "mongo/db/repl/oplog_entry.h"
+#include "mongo/db/repl/oplog_entry.h"
 #include "mongo/db/repl/optime.h"
 #include "mongo/db/repl/repl_client_info.h"
 #include "mongo/db/repl/replication_consistency_markers_impl.h"
+#include "mongo/db/repl/replication_consistency_markers_mock.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
+#include "mongo/db/repl/replication_process.h"
+#include "mongo/db/repl/replication_recovery_mock.h"
 #include "mongo/db/repl/storage_interface_impl.h"
+#include "mongo/db/repl/sync_tail.h"
 #include "mongo/db/repl/timestamp_block.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/dbtests/dbtests.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/stacktrace.h"
 
 namespace mongo {
 
+const auto kIndexVersion = IndexDescriptor::IndexVersion::kV2;
+
 class StorageTimestampTest {
 public:
     ServiceContext::UniqueOperationContext _opCtxRaii = cc().makeOperationContext();
@@ -80,6 +89,7 @@ public:
     const Timestamp futureTs = futureLt.asTimestamp();
     const Timestamp nullTs = Timestamp();
     const int presentTerm = 1;
+    repl::ReplicationCoordinatorMock* _coordinatorMock;
 
     StorageTimestampTest() {
         if (mongo::storageGlobalParams.engine != "wiredTiger") {
@@ -91,11 +101,20 @@ public:
         replSettings.setReplSetString("rs0");
         auto coordinatorMock =
             new repl::ReplicationCoordinatorMock(_opCtx->getServiceContext(), replSettings);
+        _coordinatorMock = coordinatorMock;
         coordinatorMock->alwaysAllowWrites(true);
         setGlobalReplicationCoordinator(coordinatorMock);
         repl::StorageInterface::set(_opCtx->getServiceContext(),
                                     stdx::make_unique<repl::StorageInterfaceImpl>());
 
+        auto replicationProcess = new repl::ReplicationProcess(
+            repl::StorageInterface::get(_opCtx->getServiceContext()),
+            stdx::make_unique<repl::ReplicationConsistencyMarkersMock>(),
+            stdx::make_unique<repl::ReplicationRecoveryMock>());
+        repl::ReplicationProcess::set(
+            cc().getServiceContext(),
+            std::unique_ptr<repl::ReplicationProcess>(replicationProcess));
+
         // Since the Client object persists across tests, even though the global
         // ReplicationCoordinator does not, we need to clear the last op associated with the client
         // to avoid the invariant in ReplClientInfo::setLastOp that the optime only goes forward.
@@ -351,6 +370,50 @@ public:
         ASSERT(std::find(allIdents.begin(), allIdents.end(), collIdent) == allIdents.end());
         ASSERT(std::find(allIdents.begin(), allIdents.end(), indexIdent) == allIdents.end());
     }
+
+    std::string dumpMultikeyPaths(const MultikeyPaths& multikeyPaths) {
+        std::stringstream ss;
+
+        ss << "[ ";
+        for (const auto multikeyComponents : multikeyPaths) {
+            ss << "[ ";
+            for (const auto multikeyComponent : multikeyComponents) {
+                ss << multikeyComponent << " ";
+            }
+            ss << "] ";
+        }
+        ss << "]";
+
+        return ss.str();
+    }
+
+    void assertMultikeyPaths(OperationContext* opCtx,
+                             Collection* collection,
+                             StringData indexName,
+                             Timestamp ts,
+                             bool shouldBeMultikey,
+                             const MultikeyPaths& expectedMultikeyPaths) {
+        auto catalog = collection->getCatalogEntry();
+
+        auto recoveryUnit = _opCtx->recoveryUnit();
+        recoveryUnit->abandonSnapshot();
+        ASSERT_OK(recoveryUnit->selectSnapshot(ts));
+
+        MultikeyPaths actualMultikeyPaths;
+        if (!shouldBeMultikey) {
+            ASSERT_FALSE(catalog->isIndexMultikey(opCtx, indexName, &actualMultikeyPaths));
+        } else {
+            ASSERT(catalog->isIndexMultikey(opCtx, indexName, &actualMultikeyPaths));
+        }
+
+        const bool match = (expectedMultikeyPaths == actualMultikeyPaths);
+        if (!match) {
+            FAIL(str::stream() << "Expected: " << dumpMultikeyPaths(expectedMultikeyPaths)
+                               << ", Actual: "
+                               << dumpMultikeyPaths(actualMultikeyPaths));
+        }
+        ASSERT_TRUE(match);
+    }
 };
 
 class SecondaryInsertTimes : public StorageTimestampTest {
@@ -1085,6 +1148,259 @@ public:
     }
 };
 
+class SecondarySetIndexMultikeyOnInsert : public StorageTimestampTest {
+
+public:
+    void run() {
+        // Only run on 'wiredTiger'. No other storage engines to-date support timestamp writes.
+        if (mongo::storageGlobalParams.engine != "wiredTiger") {
+            return;
+        }
+
+        // Pretend to be a secondary.
+        repl::UnreplicatedWritesBlock uwb(_opCtx);
+
+        NamespaceString nss("unittests.SecondarySetIndexMultikeyOnInsert");
+        reset(nss);
+        UUID uuid = UUID::gen();
+        {
+            AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X, LockMode::MODE_IX);
+            uuid = autoColl.getCollection()->uuid().get();
+        }
+        auto indexName = "a_1";
+        auto indexSpec =
+            BSON("name" << indexName << "ns" << nss.ns() << "key" << BSON("a" << 1) << "v"
+                        << static_cast<int>(kIndexVersion));
+        ASSERT_OK(dbtests::createIndexFromSpec(_opCtx, nss.ns(), indexSpec));
+
+        _coordinatorMock->alwaysAllowWrites(false);
+
+        const LogicalTime pastTime = _clock->reserveTicks(1);
+        const LogicalTime insertTime0 = _clock->reserveTicks(1);
+        const LogicalTime insertTime1 = _clock->reserveTicks(1);
+        const LogicalTime insertTime2 = _clock->reserveTicks(1);
+
+        BSONObj doc0 = BSON("_id" << 0 << "a" << 3);
+        BSONObj doc1 = BSON("_id" << 1 << "a" << BSON_ARRAY(1 << 2));
+        BSONObj doc2 = BSON("_id" << 2 << "a" << BSON_ARRAY(1 << 2));
+        auto op0 = repl::OplogEntry(
+            BSON("ts" << insertTime0.asTimestamp() << "t" << 1LL << "h" << 0xBEEFBEEFLL << "v" << 2
+                      << "op"
+                      << "i"
+                      << "ns"
+                      << nss.ns()
+                      << "ui"
+                      << uuid
+                      << "o"
+                      << doc0));
+        auto op1 = repl::OplogEntry(
+            BSON("ts" << insertTime1.asTimestamp() << "t" << 1LL << "h" << 0xBEEFBEEFLL << "v" << 2
+                      << "op"
+                      << "i"
+                      << "ns"
+                      << nss.ns()
+                      << "ui"
+                      << uuid
+                      << "o"
+                      << doc1));
+        auto op2 = repl::OplogEntry(
+            BSON("ts" << insertTime2.asTimestamp() << "t" << 1LL << "h" << 0xBEEFBEEFLL << "v" << 2
+                      << "op"
+                      << "i"
+                      << "ns"
+                      << nss.ns()
+                      << "ui"
+                      << uuid
+                      << "o"
+                      << doc2));
+        std::vector<repl::OplogEntry> ops = {op0, op1, op2};
+
+        repl::SyncTail(nullptr, repl::multiSyncApply).multiApply_forTest(_opCtx, ops);
+
+        AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X, LockMode::MODE_IX);
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, pastTime.asTimestamp(), false, {{}});
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, insertTime0.asTimestamp(), true, {{0}});
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, insertTime1.asTimestamp(), true, {{0}});
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, insertTime2.asTimestamp(), true, {{0}});
+    }
+};
+
+class InitialSyncSetIndexMultikeyOnInsert : public StorageTimestampTest {
+
+public:
+    void run() {
+        // Only run on 'wiredTiger'. No other storage engines to-date support timestamp writes.
+        if (mongo::storageGlobalParams.engine != "wiredTiger") {
+            return;
+        }
+
+        // Pretend to be a secondary.
+        repl::UnreplicatedWritesBlock uwb(_opCtx);
+
+        NamespaceString nss("unittests.InitialSyncSetIndexMultikeyOnInsert");
+        reset(nss);
+        UUID uuid = UUID::gen();
+        {
+            AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X, LockMode::MODE_IX);
+            uuid = autoColl.getCollection()->uuid().get();
+        }
+        auto indexName = "a_1";
+        auto indexSpec =
+            BSON("name" << indexName << "ns" << nss.ns() << "key" << BSON("a" << 1) << "v"
+                        << static_cast<int>(kIndexVersion));
+        ASSERT_OK(dbtests::createIndexFromSpec(_opCtx, nss.ns(), indexSpec));
+
+        _coordinatorMock->alwaysAllowWrites(false);
+
+        const LogicalTime pastTime = _clock->reserveTicks(1);
+        const LogicalTime insertTime0 = _clock->reserveTicks(1);
+        const LogicalTime indexBuildTime = _clock->reserveTicks(1);
+        const LogicalTime insertTime1 = _clock->reserveTicks(1);
+        const LogicalTime insertTime2 = _clock->reserveTicks(1);
+
+        BSONObj doc0 = BSON("_id" << 0 << "a" << 3);
+        BSONObj doc1 = BSON("_id" << 1 << "a" << BSON_ARRAY(1 << 2));
+        BSONObj doc2 = BSON("_id" << 2 << "a" << BSON_ARRAY(1 << 2));
+        auto op0 = repl::OplogEntry(
+            BSON("ts" << insertTime0.asTimestamp() << "t" << 1LL << "h" << 0xBEEFBEEFLL << "v" << 2
+                      << "op"
+                      << "i"
+                      << "ns"
+                      << nss.ns()
+                      << "ui"
+                      << uuid
+                      << "o"
+                      << doc0));
+        auto op1 = repl::OplogEntry(
+            BSON("ts" << insertTime1.asTimestamp() << "t" << 1LL << "h" << 0xBEEFBEEFLL << "v" << 2
+                      << "op"
+                      << "i"
+                      << "ns"
+                      << nss.ns()
+                      << "ui"
+                      << uuid
+                      << "o"
+                      << doc1));
+        auto op2 = repl::OplogEntry(
+            BSON("ts" << insertTime2.asTimestamp() << "t" << 1LL << "h" << 0xBEEFBEEFLL << "v" << 2
+                      << "op"
+                      << "i"
+                      << "ns"
+                      << nss.ns()
+                      << "ui"
+                      << uuid
+                      << "o"
+                      << doc2));
+        auto indexSpec2 = BSON("createIndexes" << nss.coll() << "ns" << nss.ns() << "v"
+                                               << static_cast<int>(kIndexVersion)
+                                               << "key"
+                                               << BSON("b" << 1)
+                                               << "name"
+                                               << "b_1");
+        auto createIndexOp = repl::OplogEntry(BSON(
+            "ts" << indexBuildTime.asTimestamp() << "t" << 1LL << "h" << 0xBEEFBEEFLL << "v" << 2
+                 << "op"
+                 << "c"
+                 << "ns"
+                 << nss.getCommandNS().ns()
+                 << "ui"
+                 << uuid
+                 << "o"
+                 << indexSpec2));
+
+        // We add in an index creation op to test that we restart tracking multikey path info
+        // after bulk index builds.
+        std::vector<const repl::OplogEntry*> ops = {&op0, &createIndexOp, &op1, &op2};
+
+        repl::SyncTail syncTail(nullptr, repl::SyncTail::MultiSyncApplyFunc(), nullptr);
+        AtomicUInt32 fetchCount(0);
+        ASSERT_OK(repl::multiInitialSyncApply_noAbort(_opCtx, &ops, &syncTail, &fetchCount));
+
+        AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X, LockMode::MODE_IX);
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, pastTime.asTimestamp(), false, {{}});
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, insertTime0.asTimestamp(), true, {{0}});
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, insertTime1.asTimestamp(), true, {{0}});
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, insertTime2.asTimestamp(), true, {{0}});
+    }
+};
+
+class PrimarySetIndexMultikeyOnInsert : public StorageTimestampTest {
+
+public:
+    void run() {
+        // Only run on 'wiredTiger'. No other storage engines to-date support timestamp writes.
+        if (mongo::storageGlobalParams.engine != "wiredTiger") {
+            return;
+        }
+
+        NamespaceString nss("unittests.PrimarySetIndexMultikeyOnInsert");
+        reset(nss);
+
+        AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X, LockMode::MODE_IX);
+        auto indexName = "a_1";
+        auto indexSpec =
+            BSON("name" << indexName << "ns" << nss.ns() << "key" << BSON("a" << 1) << "v"
+                        << static_cast<int>(kIndexVersion));
+        ASSERT_OK(dbtests::createIndexFromSpec(_opCtx, nss.ns(), indexSpec));
+
+        const LogicalTime pastTime = _clock->reserveTicks(1);
+        const LogicalTime insertTime = pastTime.addTicks(1);
+
+        BSONObj doc = BSON("_id" << 1 << "a" << BSON_ARRAY(1 << 2));
+        WriteUnitOfWork wunit(_opCtx);
+        insertDocument(autoColl.getCollection(), InsertStatement(doc));
+        wunit.commit();
+
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, pastTime.asTimestamp(), false, {{}});
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, insertTime.asTimestamp(), true, {{0}});
+    }
+};
+
+class PrimarySetIndexMultikeyOnInsertUnreplicated : public StorageTimestampTest {
+
+public:
+    void run() {
+        // Only run on 'wiredTiger'. No other storage engines to-date support timestamp writes.
+        if (mongo::storageGlobalParams.engine != "wiredTiger") {
+            return;
+        }
+
+        // Use an unreplicated collection.
+        NamespaceString nss("unittests.system.profile");
+        reset(nss);
+
+        AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_X, LockMode::MODE_IX);
+        auto indexName = "a_1";
+        auto indexSpec =
+            BSON("name" << indexName << "ns" << nss.ns() << "key" << BSON("a" << 1) << "v"
+                        << static_cast<int>(kIndexVersion));
+        ASSERT_OK(dbtests::createIndexFromSpec(_opCtx, nss.ns(), indexSpec));
+
+        const LogicalTime pastTime = _clock->reserveTicks(1);
+        const LogicalTime insertTime = pastTime.addTicks(1);
+
+        BSONObj doc = BSON("_id" << 1 << "a" << BSON_ARRAY(1 << 2));
+        WriteUnitOfWork wunit(_opCtx);
+        insertDocument(autoColl.getCollection(), InsertStatement(doc));
+        wunit.commit();
+
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, pastTime.asTimestamp(), true, {{0}});
+        assertMultikeyPaths(
+            _opCtx, autoColl.getCollection(), indexName, insertTime.asTimestamp(), true, {{0}});
+    }
+};
+
 class InitializeMinValid : public StorageTimestampTest {
 public:
     void run() {
@@ -1474,6 +1790,10 @@ public:
         add<SecondaryCreateTwoCollections>();
         add<SecondaryCreateCollectionBetweenInserts>();
         add<PrimaryCreateCollectionInApplyOps>();
+        add<SecondarySetIndexMultikeyOnInsert>();
+        add<InitialSyncSetIndexMultikeyOnInsert>();
+        add<PrimarySetIndexMultikeyOnInsert>();
+        add<PrimarySetIndexMultikeyOnInsertUnreplicated>();
         add<InitializeMinValid>();
         add<SetMinValidInitialSyncFlag>();
         add<SetMinValidToAtLeast>();