SERVER-63865 Handle missing index idents during standalone startup recovery

(cherry picked from commit 0b5f72d6e251cb402aa3d8bb102cbba6b5734e8d)

48 files changed, 468 insertions, 268 deletions

diff --git a/jstests/noPassthrough/libs/missing_index_ident.js b/jstests/noPassthrough/libs/missing_index_ident.js
new file mode 100644
index 00000000000..1c7e7dc2782
--- /dev/null
+++ b/jstests/noPassthrough/libs/missing_index_ident.js
@@ -0,0 +1,93 @@
+load('jstests/libs/fail_point_util.js');
+load('jstests/noPassthrough/libs/index_build.js');
+
+const MissingIndexIdent = class {
+    /**
+     * Starts a one-node replica set and gets it into a state where the catalog entry for index
+     * 'a_1' references an ident which is missing. Returns with the node stopped.
+     */
+    static run() {
+        const replTest = new ReplSetTest({nodes: 1});
+        replTest.startSet();
+        replTest.initiate();
+
+        const primary = function() {
+            return replTest.getPrimary();
+        };
+        const testDB = function() {
+            return primary().getDB('test');
+        };
+        const coll = function() {
+            return testDB()[jsTestName()];
+        };
+
+        assert.commandWorked(coll().insert({a: 0}));
+
+        const fp = configureFailPoint(primary(), 'hangIndexBuildBeforeCommit');
+        const awaitCreateIndex =
+            IndexBuildTest.startIndexBuild(primary(), coll().getFullName(), {a: 1});
+        fp.wait();
+
+        // Get a timestamp before the commit timestamp of the index build.
+        const ts =
+            assert
+                .commandWorked(testDB().runCommand({insert: coll().getName(), documents: [{a: 1}]}))
+                .operationTime;
+
+        configureFailPoint(primary(), 'holdStableTimestampAtSpecificTimestamp', {timestamp: ts});
+        fp.off();
+        awaitCreateIndex();
+
+        const ident = assert.commandWorked(testDB().runCommand({collStats: coll().getName()}))
+                          .indexDetails.a_1.uri.substring('statistics:table:'.length);
+
+        replTest.restart(primary(), {
+            setParameter: {
+                // Set minSnapshotHistoryWindowInSeconds to 0 so that the the oldest timestamp can
+                // move forward, despite the stable timestamp being held steady.
+                minSnapshotHistoryWindowInSeconds: 0,
+                'failpoint.holdStableTimestampAtSpecificTimestamp':
+                    tojson({mode: 'alwaysOn', data: {timestamp: ts}})
+            }
+        });
+
+        MissingIndexIdent.checkRecoveryLogs(primary(), coll(), ts, ident);
+
+        const dbpath = primary().dbpath;
+
+        // Shut down uncleanly so that a checkpoint is not taken. This will cause the index catalog
+        // entry referencing the now-dropped ident to still be present.
+        replTest.stop(0, 9, {allowedExitCode: MongoRunner.EXIT_SIGKILL}, {forRestart: true});
+
+        return {replTest, dbpath, ts, ident};
+    }
+
+    /**
+     * Checks that the index was recovered.
+     */
+    static checkRecoveryLogs(conn, coll, ts, ident) {
+        // On startup, the node will recover from before the index commit timestamp.
+        checkLog.containsJson(conn, 23987, {
+            recoveryTimestamp: (recoveryTs) => {
+                return timestampCmp(
+                           Timestamp(recoveryTs['$timestamp']['t'], recoveryTs['$timestamp']['i']),
+                           ts) <= 0;
+            }
+        });
+
+        // Since the index build was not yet completed at the recovery timestamp, its ident will
+        // be dropped.
+        checkLog.containsJson(conn, 22206, {
+            index: 'a_1',
+            namespace: coll.getFullName(),
+            ident: ident,
+            commitTimestamp: {$timestamp: {t: 0, i: 0}},
+        });
+
+        // The oldest timestamp moving forward will cause the ident reaper to drop the ident.
+        checkLog.containsJson(conn, 22237, {
+            ident: ident,
+            dropTimestamp: {$timestamp: {t: 0, i: 0}},
+        });
+    }
+};
diff --git a/jstests/noPassthrough/missing_index_ident_standalone_drop.js b/jstests/noPassthrough/missing_index_ident_standalone_drop.js
new file mode 100644
index 00000000000..abb59d5bf64
--- /dev/null
+++ b/jstests/noPassthrough/missing_index_ident_standalone_drop.js
@@ -0,0 +1,32 @@
+/**
+ * Tests that if a node is in a state where the catalog entry for an index references a missing
+ * ident, the node can be started as a standalone and the index can be dropped.
+ *
+ * @tags: [
+ *     requires_persistence,
+ *     requires_replication,
+ * ]
+ */
+(function() {
+'use strict';
+
+load('jstests/noPassthrough/libs/missing_index_ident.js');
+
+const {replTest, dbpath} = MissingIndexIdent.run();
+
+const standalone = MongoRunner.runMongod({
+    dbpath: dbpath,
+    noCleanData: true,
+});
+const coll = standalone.getDB('test')[jsTestName()];
+
+IndexBuildTest.assertIndexes(coll, 2, ['_id_', 'a_1']);
+assert.commandWorked(standalone.getDB('test')[jsTestName()].dropIndex('a_1'));
+IndexBuildTest.assertIndexes(coll, 1, ['_id_']);
+
+MongoRunner.stopMongod(standalone);
+replTest.start(0, undefined, true /* restart */);
+IndexBuildTest.assertIndexes(replTest.getPrimary().getDB('test')[jsTestName()], 1, ['_id_']);
+
+replTest.stopSet();
+})();
diff --git a/jstests/noPassthrough/restart_and_unclean_shutdown_recovery_before_index_build_commit_timestamp.js b/jstests/noPassthrough/restart_and_unclean_shutdown_recovery_before_index_build_commit_timestamp.js
index d810f1cba84..cb3a5fae7f8 100644
--- a/jstests/noPassthrough/restart_and_unclean_shutdown_recovery_before_index_build_commit_timestamp.js
+++ b/jstests/noPassthrough/restart_and_unclean_shutdown_recovery_before_index_build_commit_timestamp.js
@@ -10,87 +10,15 @@
 (function() {
 'use strict';
 
-load('jstests/libs/fail_point_util.js');
-load('jstests/noPassthrough/libs/index_build.js');
+load('jstests/noPassthrough/libs/missing_index_ident.js');
 
-const replTest = new ReplSetTest({nodes: 1});
-replTest.startSet();
-replTest.initiate();
+const {replTest, ts, ident} = MissingIndexIdent.run();
 
-const primary = function() {
-    return replTest.getPrimary();
-};
-const testDB = function() {
-    return primary().getDB('test');
-};
-const coll = function() {
-    return testDB()[jsTestName()];
-};
-
-assert.commandWorked(coll().insert({a: 0}));
-
-const fp = configureFailPoint(primary(), 'hangIndexBuildBeforeCommit');
-const awaitCreateIndex = IndexBuildTest.startIndexBuild(primary(), coll().getFullName(), {a: 1});
-fp.wait();
-
-// Get a timestamp before the commit timestamp of the index build.
-const ts =
-    assert.commandWorked(testDB().runCommand({insert: coll().getName(), documents: [{a: 1}]}))
-        .operationTime;
-
-configureFailPoint(primary(), 'holdStableTimestampAtSpecificTimestamp', {timestamp: ts});
-fp.off();
-awaitCreateIndex();
-
-const ident = assert.commandWorked(testDB().runCommand({collStats: coll().getName()}))
-                  .indexDetails.a_1.uri.substring('statistics:table:'.length);
-
-replTest.restart(primary(), {
-    setParameter: {
-        // Set minSnapshotHistoryWindowInSeconds to 0 so that the the oldest timestamp can move
-        // forward, despite the stable timestamp being held steady.
-        minSnapshotHistoryWindowInSeconds: 0,
-        'failpoint.holdStableTimestampAtSpecificTimestamp':
-            tojson({mode: 'alwaysOn', data: {timestamp: ts}})
-    }
-});
-
-const checkLogs = function() {
-    // On startup, the node will recover from before the index commit timestamp.
-    checkLog.containsJson(primary(), 23987, {
-        recoveryTimestamp: (recoveryTs) => {
-            return timestampCmp(
-                       Timestamp(recoveryTs['$timestamp']['t'], recoveryTs['$timestamp']['i']),
-                       ts) <= 0;
-        }
-    });
-
-    // Since the index build was not yet completed at the recovery timestamp, its ident will be
-    // dropped.
-    checkLog.containsJson(primary(), 22206, {
-        index: 'a_1',
-        namespace: coll().getFullName(),
-        ident: ident,
-        commitTimestamp: {$timestamp: {t: 0, i: 0}},
-    });
-
-    // The oldest timestamp moving forward will cause the ident reaper to drop the ident.
-    checkLog.containsJson(primary(), 22237, {
-        ident: ident,
-        dropTimestamp: {$timestamp: {t: 0, i: 0}},
-    });
-};
-
-checkLogs();
-
-// Shut down uncleanly so that a checkpoint is not taken. This will cause the index catalog entry
-// referencing the now-dropped ident to still be present.
-replTest.stop(0, 9, {allowedExitCode: MongoRunner.EXIT_SIGKILL}, {forRestart: true});
 replTest.start(0, undefined, true /* restart */);
+const coll = replTest.getPrimary().getDB('test')[jsTestName()];
 
-checkLogs();
-
-IndexBuildTest.assertIndexes(coll(), 2, ['_id_', 'a_1']);
+MissingIndexIdent.checkRecoveryLogs(replTest.getPrimary(), coll, ts, ident);
+IndexBuildTest.assertIndexes(coll, 2, ['_id_', 'a_1']);
 
 replTest.stopSet();
 })();
diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index a28352c821c..be4bd25382c 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -199,7 +199,7 @@ StatusWith<CollModRequest> parseCollModRequest(OperationContext* opCtx,
             } else {
                 std::vector<const IndexDescriptor*> indexes;
                 coll->getIndexCatalog()->findIndexesByKeyPattern(
-                    opCtx, keyPattern, false, &indexes);
+                    opCtx, keyPattern, IndexCatalog::InclusionPolicy::kReady, &indexes);
 
                 if (indexes.size() > 1) {
                     return Status(ErrorCodes::AmbiguousIndexKeyPattern,
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 0ad4f0a0600..f6ddd829eb1 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -1536,7 +1536,8 @@ uint64_t CollectionImpl::getIndexSize(OperationContext* opCtx,
                                       int scale) const {
     const IndexCatalog* idxCatalog = getIndexCatalog();
 
-    std::unique_ptr<IndexCatalog::IndexIterator> ii = idxCatalog->getIndexIterator(opCtx, true);
+    auto ii = idxCatalog->getIndexIterator(
+        opCtx, IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
 
     uint64_t totalSize = 0;
 
@@ -1558,8 +1559,8 @@ uint64_t CollectionImpl::getIndexSize(OperationContext* opCtx,
 
 uint64_t CollectionImpl::getIndexFreeStorageBytes(OperationContext* const opCtx) const {
     const auto idxCatalog = getIndexCatalog();
-    const bool includeUnfinished = true;
-    auto indexIt = idxCatalog->getIndexIterator(opCtx, includeUnfinished);
+    auto indexIt = idxCatalog->getIndexIterator(
+        opCtx, IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
 
     uint64_t totalSize = 0;
     while (indexIt->more()) {
@@ -1583,8 +1584,7 @@ Status CollectionImpl::truncate(OperationContext* opCtx) {
     // 1) store index specs
     std::vector<BSONObj> indexSpecs;
     {
-        std::unique_ptr<IndexCatalog::IndexIterator> ii =
-            _indexCatalog->getIndexIterator(opCtx, false);
+        auto ii = _indexCatalog->getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
         while (ii->more()) {
             const IndexDescriptor* idx = ii->next()->descriptor();
             indexSpecs.push_back(idx->infoObj().getOwned());
diff --git a/src/mongo/db/catalog/collection_validation.cpp b/src/mongo/db/catalog/collection_validation.cpp
index 235928c5b98..a43405e8b89 100644
--- a/src/mongo/db/catalog/collection_validation.cpp
+++ b/src/mongo/db/catalog/collection_validation.cpp
@@ -75,8 +75,7 @@ void _validateIndexesInternalStructure(OperationContext* opCtx,
     // Need to use the IndexCatalog here because the 'validateState->indexes' object hasn't been
     // constructed yet. It must be initialized to ensure we're validating all indexes.
     const IndexCatalog* indexCatalog = validateState->getCollection()->getIndexCatalog();
-    const std::unique_ptr<IndexCatalog::IndexIterator> it =
-        indexCatalog->getIndexIterator(opCtx, false);
+    const auto it = indexCatalog->getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
 
     // Validate Indexes Internal Structure, checking if index files have been compromised or
     // corrupted.
@@ -455,7 +454,10 @@ void _validateCatalogEntry(OperationContext* opCtx,
     }
 
     const auto& indexCatalog = collection->getIndexCatalog();
-    auto indexIt = indexCatalog->getIndexIterator(opCtx, /*includeUnfinishedIndexes=*/true);
+    auto indexIt = indexCatalog->getIndexIterator(opCtx,
+                                                  IndexCatalog::InclusionPolicy::kReady |
+                                                      IndexCatalog::InclusionPolicy::kUnfinished |
+                                                      IndexCatalog::InclusionPolicy::kFrozen);
 
     while (indexIt->more()) {
         const IndexCatalogEntry* indexEntry = indexIt->next();
diff --git a/src/mongo/db/catalog/create_collection.cpp b/src/mongo/db/catalog/create_collection.cpp
index b89955abff0..84f4d247fe0 100644
--- a/src/mongo/db/catalog/create_collection.cpp
+++ b/src/mongo/db/catalog/create_collection.cpp
@@ -289,8 +289,11 @@ Status _createTimeseries(OperationContext* opCtx,
                     coll->getCollectionOptions().matchesStorageOptions(
                         bucketsOptions, CollatorFactoryInterface::get(opCtx->getServiceContext()));
                 if (expireAfterSeconds && !bucketsOptions.clusteredIndex) {
-                    auto indexDescriptor =
-                        coll->getIndexCatalog()->findIndexByName(opCtx, indexName, true);
+                    auto indexDescriptor = coll->getIndexCatalog()->findIndexByName(
+                        opCtx,
+                        indexName,
+                        IndexCatalog::InclusionPolicy::kReady |
+                            IndexCatalog::InclusionPolicy::kUnfinished);
                     existingBucketCollectionIsCompatible &=
                         indexDescriptor && indexDescriptor->infoObj().woCompare(indexSpec) == 0;
                 }
diff --git a/src/mongo/db/catalog/drop_indexes.cpp b/src/mongo/db/catalog/drop_indexes.cpp
index 4b55d9ae113..08fc6b8a73d 100644
--- a/src/mongo/db/catalog/drop_indexes.cpp
+++ b/src/mongo/db/catalog/drop_indexes.cpp
@@ -106,9 +106,13 @@ Status checkReplState(OperationContext* opCtx,
 StatusWith<const IndexDescriptor*> getDescriptorByKeyPattern(OperationContext* opCtx,
                                                              const IndexCatalog* indexCatalog,
                                                              const BSONObj& keyPattern) {
-    const bool includeUnfinished = true;
     std::vector<const IndexDescriptor*> indexes;
-    indexCatalog->findIndexesByKeyPattern(opCtx, keyPattern, includeUnfinished, &indexes);
+    indexCatalog->findIndexesByKeyPattern(opCtx,
+                                          keyPattern,
+                                          IndexCatalog::InclusionPolicy::kReady |
+                                              IndexCatalog::InclusionPolicy::kUnfinished |
+                                              IndexCatalog::InclusionPolicy::kFrozen,
+                                          &indexes);
     if (indexes.empty()) {
         return Status(ErrorCodes::IndexNotFound,
                       str::stream() << "can't find index with key: " << keyPattern);
@@ -267,9 +271,12 @@ void dropReadyIndexes(OperationContext* opCtx,
         return;
     }
 
-    bool includeUnfinished = true;
     for (const auto& indexName : indexNames) {
-        auto desc = indexCatalog->findIndexByName(opCtx, indexName, includeUnfinished);
+        auto desc = indexCatalog->findIndexByName(opCtx,
+                                                  indexName,
+                                                  IndexCatalog::InclusionPolicy::kReady |
+                                                      IndexCatalog::InclusionPolicy::kUnfinished |
+                                                      IndexCatalog::InclusionPolicy::kFrozen);
         if (!desc) {
             uasserted(ErrorCodes::IndexNotFound,
                       str::stream() << "index not found with name [" << indexName << "]");
@@ -419,9 +426,13 @@ DropIndexesReply dropIndexes(OperationContext* opCtx,
             // the index catalog. This would indicate that while we yielded our locks during the
             // abort phase, a new identical index was created.
             auto indexCatalog = collection->getWritableCollection()->getIndexCatalog();
-            const bool includeUnfinished = false;
             for (const auto& indexName : indexNames) {
-                auto desc = indexCatalog->findIndexByName(opCtx, indexName, includeUnfinished);
+                auto desc =
+                    indexCatalog->findIndexByName(opCtx,
+                                                  indexName,
+                                                  IndexCatalog::InclusionPolicy::kReady |
+                                                      IndexCatalog::InclusionPolicy::kUnfinished |
+                                                      IndexCatalog::InclusionPolicy::kFrozen);
                 if (!desc) {
                     // A similar index wasn't created while we yielded the locks during abort.
                     continue;
diff --git a/src/mongo/db/catalog/index_build_block.cpp b/src/mongo/db/catalog/index_build_block.cpp
index d5b410ee03b..8fbd644a71e 100644
--- a/src/mongo/db/catalog/index_build_block.cpp
+++ b/src/mongo/db/catalog/index_build_block.cpp
@@ -94,7 +94,9 @@ Status IndexBuildBlock::initForResume(OperationContext* opCtx,
 
     _indexName = _spec.getStringField("name");
     auto descriptor = collection->getIndexCatalog()->findIndexByName(
-        opCtx, _indexName, true /* includeUnfinishedIndexes */);
+        opCtx,
+        _indexName,
+        IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
 
     auto indexCatalogEntry = descriptor->getEntry();
 
@@ -279,14 +281,18 @@ void IndexBuildBlock::success(OperationContext* opCtx, Collection* collection) {
 const IndexCatalogEntry* IndexBuildBlock::getEntry(OperationContext* opCtx,
                                                    const CollectionPtr& collection) const {
     auto descriptor = collection->getIndexCatalog()->findIndexByName(
-        opCtx, _indexName, true /* includeUnfinishedIndexes */);
+        opCtx,
+        _indexName,
+        IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
 
     return descriptor->getEntry();
 }
 
 IndexCatalogEntry* IndexBuildBlock::getEntry(OperationContext* opCtx, Collection* collection) {
     auto descriptor = collection->getIndexCatalog()->findIndexByName(
-        opCtx, _indexName, true /* includeUnfinishedIndexes */);
+        opCtx,
+        _indexName,
+        IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
 
     return descriptor->getEntry();
 }
diff --git a/src/mongo/db/catalog/index_catalog.h b/src/mongo/db/catalog/index_catalog.h
index 9f5b431d990..36acb0d6580 100644
--- a/src/mongo/db/catalog/index_catalog.h
+++ b/src/mongo/db/catalog/index_catalog.h
@@ -175,6 +175,12 @@ public:
         std::unique_ptr<std::vector<IndexCatalogEntry*>> _ownedContainer;
     };
 
+    enum class InclusionPolicy {
+        kReady = 1 << 0,
+        kUnfinished = 1 << 1,
+        kFrozen = 1 << 2,
+    };
+
     IndexCatalog() = default;
     virtual ~IndexCatalog() = default;
 
@@ -218,7 +224,7 @@ public:
     virtual const IndexDescriptor* findIndexByName(
         OperationContext* const opCtx,
         const StringData name,
-        const bool includeUnfinishedIndexes = false) const = 0;
+        const InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const = 0;
 
     /**
      * Find index by matching key pattern and options. The key pattern, collation spec, and partial
@@ -230,7 +236,7 @@ public:
         OperationContext* const opCtx,
         const BSONObj& key,
         const BSONObj& indexSpec,
-        const bool includeUnfinishedIndexes = false) const = 0;
+        const InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const = 0;
 
     /**
      * Find indexes with a matching key pattern, putting them into the vector 'matches'.  The key
@@ -241,7 +247,7 @@ public:
     virtual void findIndexesByKeyPattern(
         OperationContext* const opCtx,
         const BSONObj& key,
-        const bool includeUnfinishedIndexes,
+        const InclusionPolicy inclusionPolicy,
         std::vector<const IndexDescriptor*>* const matches) const = 0;
 
     /**
@@ -261,10 +267,11 @@ public:
                                                              const BSONObj& shardKey,
                                                              const bool requireSingleKey) const = 0;
 
-    virtual void findIndexByType(OperationContext* const opCtx,
-                                 const std::string& type,
-                                 std::vector<const IndexDescriptor*>& matches,
-                                 const bool includeUnfinishedIndexes = false) const = 0;
+    virtual void findIndexByType(
+        OperationContext* const opCtx,
+        const std::string& type,
+        std::vector<const IndexDescriptor*>& matches,
+        const InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const = 0;
 
     /**
      * Reload the index definition for 'oldDesc' from the CollectionCatalogEntry.  'oldDesc'
@@ -305,7 +312,7 @@ public:
      * Returns an iterator for the index descriptors in this IndexCatalog.
      */
     virtual std::unique_ptr<IndexIterator> getIndexIterator(
-        OperationContext* const opCtx, const bool includeUnfinishedIndexes) const = 0;
+        OperationContext* const opCtx, const InclusionPolicy inclusionPolicy) const = 0;
 
     // ---- index set modifiers ------
 
@@ -510,4 +517,16 @@ public:
                                    Collection* coll,
                                    IndexCatalogEntry* index) = 0;
 };
+
+inline IndexCatalog::InclusionPolicy operator|(IndexCatalog::InclusionPolicy lhs,
+                                               IndexCatalog::InclusionPolicy rhs) {
+    return static_cast<IndexCatalog::InclusionPolicy>(
+        static_cast<std::underlying_type_t<IndexCatalog::InclusionPolicy>>(lhs) |
+        static_cast<std::underlying_type_t<IndexCatalog::InclusionPolicy>>(rhs));
+}
+
+inline bool operator&(IndexCatalog::InclusionPolicy lhs, IndexCatalog::InclusionPolicy rhs) {
+    return static_cast<std::underlying_type_t<IndexCatalog::InclusionPolicy>>(lhs) &
+        static_cast<std::underlying_type_t<IndexCatalog::InclusionPolicy>>(rhs);
+}
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_catalog_entry.h b/src/mongo/db/catalog/index_catalog_entry.h
index 5dd6acb6bd8..abec0c5e1d5 100644
--- a/src/mongo/db/catalog/index_catalog_entry.h
+++ b/src/mongo/db/catalog/index_catalog_entry.h
@@ -64,8 +64,6 @@ public:
     inline IndexCatalogEntry(IndexCatalogEntry&&) = delete;
     inline IndexCatalogEntry& operator=(IndexCatalogEntry&&) = delete;
 
-    virtual void init(std::unique_ptr<IndexAccessMethod> accessMethod) = 0;
-
     virtual const std::string& getIdent() const = 0;
     virtual std::shared_ptr<Ident> getSharedIdent() const = 0;
 
@@ -75,6 +73,8 @@ public:
 
     virtual IndexAccessMethod* accessMethod() const = 0;
 
+    virtual void setAccessMethod(std::unique_ptr<IndexAccessMethod> accessMethod) = 0;
+
     virtual bool isHybridBuilding() const = 0;
 
     virtual IndexBuildInterceptor* indexBuildInterceptor() const = 0;
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.cpp b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
index 2af03905b74..db312741ebd 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
@@ -119,7 +119,7 @@ IndexCatalogEntryImpl::IndexCatalogEntryImpl(OperationContext* const opCtx,
     }
 }
 
-void IndexCatalogEntryImpl::init(std::unique_ptr<IndexAccessMethod> accessMethod) {
+void IndexCatalogEntryImpl::setAccessMethod(std::unique_ptr<IndexAccessMethod> accessMethod) {
     invariant(!_accessMethod);
     _accessMethod = std::move(accessMethod);
 }
@@ -360,7 +360,9 @@ Status IndexCatalogEntryImpl::_setMultikeyInMultiDocumentTransaction(
 }
 
 std::shared_ptr<Ident> IndexCatalogEntryImpl::getSharedIdent() const {
-    return {shared_from_this(), _accessMethod->getSortedDataInterface()};  // aliasing constructor
+    return _accessMethod
+        ? std::shared_ptr<Ident>{shared_from_this(), _accessMethod->getSortedDataInterface()}
+        : nullptr;
 }
 
 // ----
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.h b/src/mongo/db/catalog/index_catalog_entry_impl.h
index 6a6f84ff3aa..8cdf5da9981 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.h
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.h
@@ -62,8 +62,6 @@ public:
                           std::unique_ptr<IndexDescriptor> descriptor,  // ownership passes to me
                           bool isFrozen);
 
-    void init(std::unique_ptr<IndexAccessMethod> accessMethod) final;
-
     const std::string& getIdent() const final {
         return _ident;
     }
@@ -81,6 +79,8 @@ public:
         return _accessMethod.get();
     }
 
+    void setAccessMethod(std::unique_ptr<IndexAccessMethod> accessMethod) final;
+
     bool isHybridBuilding() const final {
         return _indexBuildInterceptor != nullptr;
     }
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index 91ea476dff1..8f0ef69a5a8 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -197,8 +197,8 @@ Status IndexCatalogImpl::init(OperationContext* opCtx, Collection* collection) {
 }
 
 std::unique_ptr<IndexCatalog::IndexIterator> IndexCatalogImpl::getIndexIterator(
-    OperationContext* const opCtx, const bool includeUnfinishedIndexes) const {
-    if (!includeUnfinishedIndexes) {
+    OperationContext* const opCtx, InclusionPolicy inclusionPolicy) const {
+    if (inclusionPolicy == InclusionPolicy::kReady) {
         // If the caller only wants the ready indexes, we return an iterator over the catalog's
         // ready indexes vector. When the user advances this iterator, it will filter out any
         // indexes that were not ready at the OperationContext's read timestamp.
@@ -206,17 +206,28 @@ std::unique_ptr<IndexCatalog::IndexIterator> IndexCatalogImpl::getIndexIterator(
             opCtx, _readyIndexes.begin(), _readyIndexes.end());
     }
 
-    // If the caller wants all indexes, for simplicity of implementation, we copy the pointers to
-    // a new vector. The vector's ownership is passed to the iterator. The query code path from an
-    // external client is not expected to hit this case so the cost isn't paid by the important
-    // code path.
+    // If the caller doesn't only want the ready indexes, for simplicity of implementation, we copy
+    // the pointers to a new vector. The vector's ownership is passed to the iterator. The query
+    // code path from an external client is not expected to hit this case so the cost isn't paid by
+    // the important code path.
     auto allIndexes = std::make_unique<std::vector<IndexCatalogEntry*>>();
-    for (auto it = _readyIndexes.begin(); it != _readyIndexes.end(); ++it) {
-        allIndexes->push_back(it->get());
+
+    if (inclusionPolicy & InclusionPolicy::kReady) {
+        for (auto it = _readyIndexes.begin(); it != _readyIndexes.end(); ++it) {
+            allIndexes->push_back(it->get());
+        }
+    }
+
+    if (inclusionPolicy & InclusionPolicy::kUnfinished) {
+        for (auto it = _buildingIndexes.begin(); it != _buildingIndexes.end(); ++it) {
+            allIndexes->push_back(it->get());
+        }
     }
 
-    for (auto it = _buildingIndexes.begin(); it != _buildingIndexes.end(); ++it) {
-        allIndexes->push_back(it->get());
+    if (inclusionPolicy & InclusionPolicy::kFrozen) {
+        for (auto it = _frozenIndexes.begin(); it != _frozenIndexes.end(); ++it) {
+            allIndexes->push_back(it->get());
+        }
     }
 
     return std::make_unique<AllIndexesIterator>(opCtx, std::move(allIndexes));
@@ -291,6 +302,7 @@ void IndexCatalogImpl::_logInternalState(OperationContext* opCtx,
                 "numIndexesInCollectionCatalogEntry"_attr = numIndexesInCollectionCatalogEntry,
                 "readyIndexes_size"_attr = _readyIndexes.size(),
                 "buildingIndexes_size"_attr = _buildingIndexes.size(),
+                "frozenIndexes_size"_attr = _frozenIndexes.size(),
                 "indexNamesToDrop_size"_attr = indexNamesToDrop.size(),
                 "haveIdIndex"_attr = haveIdIndex);
 
@@ -371,7 +383,8 @@ StatusWith<BSONObj> IndexCatalogImpl::prepareSpecForCreate(
     }
 
     // First check against only the ready indexes for conflicts.
-    status = _doesSpecConflictWithExisting(opCtx, collection, validatedSpec, false);
+    status =
+        _doesSpecConflictWithExisting(opCtx, collection, validatedSpec, InclusionPolicy::kReady);
     if (!status.isOK()) {
         return status;
     }
@@ -387,7 +400,12 @@ StatusWith<BSONObj> IndexCatalogImpl::prepareSpecForCreate(
     // The index catalog cannot currently iterate over only in-progress indexes. So by previously
     // checking against only ready indexes without error, we know that any errors encountered
     // checking against all indexes occurred due to an in-progress index.
-    status = _doesSpecConflictWithExisting(opCtx, collection, validatedSpec, true);
+    status = _doesSpecConflictWithExisting(opCtx,
+                                           collection,
+                                           validatedSpec,
+                                           IndexCatalog::InclusionPolicy::kReady |
+                                               IndexCatalog::InclusionPolicy::kUnfinished |
+                                               IndexCatalog::InclusionPolicy::kFrozen);
     if (!status.isOK()) {
         if (ErrorCodes::IndexAlreadyExists == status.code()) {
             // Callers need to be able to distinguish conflicts against ready indexes versus
@@ -416,8 +434,11 @@ std::vector<BSONObj> IndexCatalogImpl::removeExistingIndexesNoChecks(
         // _doesSpecConflictWithExisting currently does more work than we require here: we are only
         // interested in the index already exists error.
         if (ErrorCodes::IndexAlreadyExists ==
-            _doesSpecConflictWithExisting(
-                opCtx, collection, spec, true /*includeUnfinishedIndexes*/)) {
+            _doesSpecConflictWithExisting(opCtx,
+                                          collection,
+                                          spec,
+                                          IndexCatalog::InclusionPolicy::kReady |
+                                              IndexCatalog::InclusionPolicy::kUnfinished)) {
             continue;
         }
 
@@ -469,20 +490,22 @@ IndexCatalogEntry* IndexCatalogImpl::createIndexEntry(OperationContext* opCtx,
     auto entry = std::make_shared<IndexCatalogEntryImpl>(
         opCtx, collection, ident, std::move(descriptor), frozen);
 
-    IndexDescriptor* desc = entry->descriptor();
-    const auto& collOptions = collection->getCollectionOptions();
-    std::unique_ptr<SortedDataInterface> sdi =
-        engine->getEngine()->getSortedDataInterface(opCtx, collOptions, ident, desc);
-
-    std::unique_ptr<IndexAccessMethod> accessMethod =
-        IndexAccessMethod::make(entry.get(), std::move(sdi));
-
-    entry->init(std::move(accessMethod));
+    if (!frozen) {
+        IndexDescriptor* desc = entry->descriptor();
+        const auto& collOptions = collection->getCollectionOptions();
+        std::unique_ptr<SortedDataInterface> sdi =
+            engine->getEngine()->getSortedDataInterface(opCtx, collOptions, ident, desc);
 
+        std::unique_ptr<IndexAccessMethod> accessMethod =
+            IndexAccessMethod::make(entry.get(), std::move(sdi));
+        entry->setAccessMethod(std::move(accessMethod));
+    }
 
     IndexCatalogEntry* save = entry.get();
     if (isReadyIndex) {
         _readyIndexes.add(std::move(entry));
+    } else if (frozen) {
+        _frozenIndexes.add(std::move(entry));
     } else {
         _buildingIndexes.add(std::move(entry));
     }
@@ -821,10 +844,11 @@ Status IndexCatalogImpl::_isSpecOk(OperationContext* opCtx,
     return Status::OK();
 }
 
-Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
-                                                       const CollectionPtr& collection,
-                                                       const BSONObj& spec,
-                                                       const bool includeUnfinishedIndexes) const {
+Status IndexCatalogImpl::_doesSpecConflictWithExisting(
+    OperationContext* opCtx,
+    const CollectionPtr& collection,
+    const BSONObj& spec,
+    const InclusionPolicy inclusionPolicy) const {
     const char* name = spec.getStringField(IndexDescriptor::kIndexNameFieldName);
     invariant(name[0]);
 
@@ -832,7 +856,7 @@ Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
 
     {
         // Check whether an index with the specified candidate name already exists in the catalog.
-        const IndexDescriptor* desc = findIndexByName(opCtx, name, includeUnfinishedIndexes);
+        const IndexDescriptor* desc = findIndexByName(opCtx, name, inclusionPolicy);
 
         if (desc) {
             // Index already exists with same name. Check whether the options are the same as well.
@@ -885,7 +909,7 @@ Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
     {
         // No index with the candidate name exists. Check for an index with conflicting options.
         const IndexDescriptor* desc =
-            findIndexByKeyPatternAndOptions(opCtx, key, spec, includeUnfinishedIndexes);
+            findIndexByKeyPatternAndOptions(opCtx, key, spec, inclusionPolicy);
 
         if (desc) {
             LOGV2_DEBUG(20353,
@@ -936,7 +960,7 @@ Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
     string pluginName = IndexNames::findPluginName(key);
     if (pluginName == IndexNames::TEXT) {
         vector<const IndexDescriptor*> textIndexes;
-        findIndexByType(opCtx, IndexNames::TEXT, textIndexes, includeUnfinishedIndexes);
+        findIndexByType(opCtx, IndexNames::TEXT, textIndexes, inclusionPolicy);
         if (textIndexes.size() > 0) {
             return Status(ErrorCodes::CannotCreateIndex,
                           str::stream() << "only one text index per collection allowed, "
@@ -977,7 +1001,10 @@ void IndexCatalogImpl::dropAllIndexes(OperationContext* opCtx,
     vector<string> indexNamesToDrop;
     {
         int seen = 0;
-        std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, true);
+        auto ii = getIndexIterator(opCtx,
+                                   IndexCatalog::InclusionPolicy::kReady |
+                                       IndexCatalog::InclusionPolicy::kUnfinished |
+                                       IndexCatalog::InclusionPolicy::kFrozen);
         while (ii->more()) {
             seen++;
             const IndexDescriptor* desc = ii->next()->descriptor();
@@ -992,7 +1019,11 @@ void IndexCatalogImpl::dropAllIndexes(OperationContext* opCtx,
 
     for (size_t i = 0; i < indexNamesToDrop.size(); i++) {
         string indexName = indexNamesToDrop[i];
-        const IndexDescriptor* desc = findIndexByName(opCtx, indexName, true);
+        const IndexDescriptor* desc = findIndexByName(
+            opCtx,
+            indexName,
+            IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished |
+                IndexCatalog::InclusionPolicy::kFrozen);
         invariant(desc);
         LOGV2_DEBUG(20355, 1, "\t dropAllIndexes dropping: {desc}", "desc"_attr = *desc);
         IndexCatalogEntry* entry = desc->getEntry();
@@ -1116,14 +1147,21 @@ Status IndexCatalogImpl::dropIndexEntry(OperationContext* opCtx,
                                                 collection->ns(),
                                                 std::move(released),
                                                 collection->getSharedDecorations()));
-    } else {
-        released = _buildingIndexes.release(entry->descriptor());
+    } else if ((released = _buildingIndexes.release(entry->descriptor()))) {
         invariant(released.get() == entry);
         opCtx->recoveryUnit()->registerChange(
             std::make_unique<IndexRemoveChange>(&_buildingIndexes,
                                                 collection->ns(),
                                                 std::move(released),
                                                 collection->getSharedDecorations()));
+    } else {
+        released = _frozenIndexes.release(entry->descriptor());
+        invariant(released.get() == entry);
+        opCtx->recoveryUnit()->registerChange(
+            std::make_unique<IndexRemoveChange>(&_frozenIndexes,
+                                                collection->ns(),
+                                                std::move(released),
+                                                collection->getSharedDecorations()));
     }
 
     CollectionQueryInfo::get(collection).rebuildIndexData(opCtx, collection);
@@ -1144,7 +1182,11 @@ void IndexCatalogImpl::_deleteIndexFromDisk(OperationContext* opCtx,
                                             Collection* collection,
                                             const string& indexName,
                                             std::shared_ptr<Ident> ident) {
-    invariant(!findIndexByName(opCtx, indexName, true /* includeUnfinishedIndexes*/));
+    invariant(!findIndexByName(opCtx,
+                               indexName,
+                               IndexCatalog::InclusionPolicy::kReady |
+                                   IndexCatalog::InclusionPolicy::kUnfinished |
+                                   IndexCatalog::InclusionPolicy::kFrozen));
     catalog::removeIndex(opCtx, indexName, collection, std::move(ident));
 }
 
@@ -1174,7 +1216,7 @@ int IndexCatalogImpl::numIndexesTotal(OperationContext* opCtx) const {
 
 int IndexCatalogImpl::numIndexesReady(OperationContext* opCtx) const {
     std::vector<const IndexDescriptor*> itIndexes;
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, /*includeUnfinished*/ false);
+    auto ii = getIndexIterator(opCtx, InclusionPolicy::kReady);
     while (ii->more()) {
         itIndexes.push_back(ii->next()->descriptor());
     }
@@ -1186,7 +1228,7 @@ bool IndexCatalogImpl::haveIdIndex(OperationContext* opCtx) const {
 }
 
 const IndexDescriptor* IndexCatalogImpl::findIdIndex(OperationContext* opCtx) const {
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, false);
+    auto ii = getIndexIterator(opCtx, InclusionPolicy::kReady);
     while (ii->more()) {
         const IndexDescriptor* desc = ii->next()->descriptor();
         if (desc->isIdIndex())
@@ -1197,8 +1239,8 @@ const IndexDescriptor* IndexCatalogImpl::findIdIndex(OperationContext* opCtx) co
 
 const IndexDescriptor* IndexCatalogImpl::findIndexByName(OperationContext* opCtx,
                                                          StringData name,
-                                                         bool includeUnfinishedIndexes) const {
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, includeUnfinishedIndexes);
+                                                         InclusionPolicy inclusionPolicy) const {
+    auto ii = getIndexIterator(opCtx, inclusionPolicy);
     while (ii->more()) {
         const IndexDescriptor* desc = ii->next()->descriptor();
         if (desc->indexName() == name)
@@ -1211,8 +1253,8 @@ const IndexDescriptor* IndexCatalogImpl::findIndexByKeyPatternAndOptions(
     OperationContext* opCtx,
     const BSONObj& key,
     const BSONObj& indexSpec,
-    bool includeUnfinishedIndexes) const {
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, includeUnfinishedIndexes);
+    InclusionPolicy inclusionPolicy) const {
+    auto ii = getIndexIterator(opCtx, inclusionPolicy);
     IndexDescriptor needle(_getAccessMethodName(key), indexSpec);
     while (ii->more()) {
         const auto* entry = ii->next();
@@ -1226,10 +1268,10 @@ const IndexDescriptor* IndexCatalogImpl::findIndexByKeyPatternAndOptions(
 
 void IndexCatalogImpl::findIndexesByKeyPattern(OperationContext* opCtx,
                                                const BSONObj& key,
-                                               bool includeUnfinishedIndexes,
+                                               InclusionPolicy inclusionPolicy,
                                                std::vector<const IndexDescriptor*>* matches) const {
     invariant(matches);
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, includeUnfinishedIndexes);
+    auto ii = getIndexIterator(opCtx, inclusionPolicy);
     while (ii->more()) {
         const IndexDescriptor* desc = ii->next()->descriptor();
         if (SimpleBSONObjComparator::kInstance.evaluate(desc->keyPattern() == key)) {
@@ -1243,7 +1285,8 @@ const IndexDescriptor* IndexCatalogImpl::findShardKeyPrefixedIndex(OperationCont
                                                                    bool requireSingleKey) const {
     const IndexDescriptor* best = nullptr;
 
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, false);
+    std::unique_ptr<IndexIterator> ii =
+        getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
     while (ii->more()) {
         const IndexCatalogEntry* entry = ii->next();
         const IndexDescriptor* desc = entry->descriptor();
@@ -1268,8 +1311,8 @@ const IndexDescriptor* IndexCatalogImpl::findShardKeyPrefixedIndex(OperationCont
 void IndexCatalogImpl::findIndexByType(OperationContext* opCtx,
                                        const string& type,
                                        vector<const IndexDescriptor*>& matches,
-                                       bool includeUnfinishedIndexes) const {
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, includeUnfinishedIndexes);
+                                       InclusionPolicy inclusionPolicy) const {
+    auto ii = getIndexIterator(opCtx, inclusionPolicy);
     while (ii->more()) {
         const IndexDescriptor* desc = ii->next()->descriptor();
         if (IndexNames::findPluginName(desc->keyPattern()) == type) {
@@ -1742,7 +1785,10 @@ Status IndexCatalogImpl::compactIndexes(OperationContext* opCtx) const {
 }
 
 std::string::size_type IndexCatalogImpl::getLongestIndexNameLength(OperationContext* opCtx) const {
-    std::unique_ptr<IndexIterator> it = getIndexIterator(opCtx, true);
+    auto it = getIndexIterator(opCtx,
+                               IndexCatalog::InclusionPolicy::kReady |
+                                   IndexCatalog::InclusionPolicy::kUnfinished |
+                                   IndexCatalog::InclusionPolicy::kFrozen);
     std::string::size_type longestIndexNameLength = 0;
     while (it->more()) {
         auto thisLength = it->next()->descriptor()->indexName().length();
diff --git a/src/mongo/db/catalog/index_catalog_impl.h b/src/mongo/db/catalog/index_catalog_impl.h
index 1e8799889e3..704de0190f3 100644
--- a/src/mongo/db/catalog/index_catalog_impl.h
+++ b/src/mongo/db/catalog/index_catalog_impl.h
@@ -96,9 +96,10 @@ public:
      *
      * @return null if cannot find
      */
-    const IndexDescriptor* findIndexByName(OperationContext* opCtx,
-                                           StringData name,
-                                           bool includeUnfinishedIndexes = false) const override;
+    const IndexDescriptor* findIndexByName(
+        OperationContext* opCtx,
+        StringData name,
+        const InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const override;
 
     /**
      * Find index by matching key pattern and options. The key pattern, collation spec, and partial
@@ -110,7 +111,7 @@ public:
         OperationContext* opCtx,
         const BSONObj& key,
         const BSONObj& indexSpec,
-        bool includeUnfinishedIndexes = false) const override;
+        const InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const override;
 
     /**
      * Find indexes with a matching key pattern, putting them into the vector 'matches'.  The key
@@ -120,7 +121,7 @@ public:
      */
     void findIndexesByKeyPattern(OperationContext* opCtx,
                                  const BSONObj& key,
-                                 bool includeUnfinishedIndexes,
+                                 const InclusionPolicy inclusionPolicy,
                                  std::vector<const IndexDescriptor*>* matches) const override;
 
     /**
@@ -140,10 +141,11 @@ public:
                                                      const BSONObj& shardKey,
                                                      bool requireSingleKey) const override;
 
-    void findIndexByType(OperationContext* opCtx,
-                         const std::string& type,
-                         std::vector<const IndexDescriptor*>& matches,
-                         bool includeUnfinishedIndexes = false) const override;
+    void findIndexByType(
+        OperationContext* opCtx,
+        const std::string& type,
+        std::vector<const IndexDescriptor*>& matches,
+        const InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const override;
 
 
     /**
@@ -170,7 +172,7 @@ public:
 
     using IndexIterator = IndexCatalog::IndexIterator;
     std::unique_ptr<IndexIterator> getIndexIterator(
-        OperationContext* const opCtx, const bool includeUnfinishedIndexes) const override;
+        OperationContext* const opCtx, const InclusionPolicy inclusionPolicy) const override;
 
     // ---- index set modifiers ------
 
@@ -401,8 +403,8 @@ private:
     /**
      * Checks whether there are any spec conflicts with existing ready indexes or in-progress index
      * builds. Also checks whether any limits set on this server would be exceeded by building the
-     * index. 'includeUnfinishedIndexes' dictates whether in-progress index builds are checked for
-     * conflicts, along with ready indexes.
+     * index. 'inclusionPolicy' dictates whether in-progress index builds are checked for conflicts,
+     * along with ready indexes.
      *
      * Returns IndexAlreadyExists for both ready and in-progress index builds. Can also return other
      * errors.
@@ -410,7 +412,7 @@ private:
     Status _doesSpecConflictWithExisting(OperationContext* opCtx,
                                          const CollectionPtr& collection,
                                          const BSONObj& spec,
-                                         const bool includeUnfinishedIndexes) const;
+                                         const InclusionPolicy inclusionPolicy) const;
 
     /**
      * Returns true if the replica set member's config has {buildIndexes:false} set, which means
@@ -427,5 +429,6 @@ private:
 
     IndexCatalogEntryContainer _readyIndexes;
     IndexCatalogEntryContainer _buildingIndexes;
+    IndexCatalogEntryContainer _frozenIndexes;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_catalog_noop.h b/src/mongo/db/catalog/index_catalog_noop.h
index 64d7b9c46b2..861a0427323 100644
--- a/src/mongo/db/catalog/index_catalog_noop.h
+++ b/src/mongo/db/catalog/index_catalog_noop.h
@@ -75,9 +75,10 @@ public:
         return nullptr;
     }
 
-    IndexDescriptor* findIndexByName(OperationContext* const opCtx,
-                                     const StringData name,
-                                     const bool includeUnfinishedIndexes = false) const override {
+    IndexDescriptor* findIndexByName(
+        OperationContext* const opCtx,
+        const StringData name,
+        const InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const override {
         return nullptr;
     }
 
@@ -85,14 +86,14 @@ public:
         OperationContext* const opCtx,
         const BSONObj& key,
         const BSONObj& indexSpec,
-        const bool includeUnfinishedIndexes = false) const override {
+        const InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const override {
         return nullptr;
     }
 
     void findIndexesByKeyPattern(
         OperationContext* const opCtx,
         const BSONObj& key,
-        const bool includeUnfinishedIndexes,
+        const InclusionPolicy inclusionPolicy,
         std::vector<const IndexDescriptor*>* const matches) const override {}
 
     IndexDescriptor* findShardKeyPrefixedIndex(OperationContext* const opCtx,
@@ -101,10 +102,11 @@ public:
         return nullptr;
     }
 
-    void findIndexByType(OperationContext* const opCtx,
-                         const std::string& type,
-                         std::vector<const IndexDescriptor*>& matches,
-                         const bool includeUnfinishedIndexes = false) const override {}
+    void findIndexByType(
+        OperationContext* const opCtx,
+        const std::string& type,
+        std::vector<const IndexDescriptor*>& matches,
+        const InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const override {}
 
     const IndexDescriptor* refreshEntry(OperationContext* const opCtx,
                                         const IndexDescriptor* const oldDesc) override {
@@ -125,7 +127,7 @@ public:
     }
 
     std::unique_ptr<IndexIterator> getIndexIterator(
-        OperationContext* const opCtx, const bool includeUnfinishedIndexes) const override {
+        OperationContext* const opCtx, const InclusionPolicy inclusionPolicy) const override {
         return {};
     }
 
diff --git a/src/mongo/db/catalog/rename_collection.cpp b/src/mongo/db/catalog/rename_collection.cpp
index b4c27775f12..080ffc86f09 100644
--- a/src/mongo/db/catalog/rename_collection.cpp
+++ b/src/mongo/db/catalog/rename_collection.cpp
@@ -598,7 +598,10 @@ Status renameBetweenDBs(OperationContext* opCtx,
 
     // Copy the index descriptions from the source collection.
     std::vector<BSONObj> indexesToCopy;
-    for (auto sourceIndIt = sourceColl->getIndexCatalog()->getIndexIterator(opCtx, true);
+    for (auto sourceIndIt = sourceColl->getIndexCatalog()->getIndexIterator(
+             opCtx,
+             IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished |
+                 IndexCatalog::InclusionPolicy::kFrozen);
          sourceIndIt->more();) {
         auto descriptor = sourceIndIt->next()->descriptor();
         if (descriptor->isIdIndex()) {
diff --git a/src/mongo/db/catalog/validate_state.cpp b/src/mongo/db/catalog/validate_state.cpp
index a299d774cc9..cb75f468200 100644
--- a/src/mongo/db/catalog/validate_state.cpp
+++ b/src/mongo/db/catalog/validate_state.cpp
@@ -257,8 +257,7 @@ void ValidateState::initializeCursors(OperationContext* opCtx) {
     const IndexCatalog* indexCatalog = _collection->getIndexCatalog();
     // The index iterator for ready indexes is timestamp-aware and will only return indexes that
     // are visible at our read time.
-    const std::unique_ptr<IndexCatalog::IndexIterator> it =
-        indexCatalog->getIndexIterator(opCtx, /*includeUnfinished*/ false);
+    const auto it = indexCatalog->getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
     while (it->more()) {
         const IndexCatalogEntry* entry = it->next();
         const IndexDescriptor* desc = entry->descriptor();
diff --git a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
index cff10caf7d8..c60b0b2d619 100644
--- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
+++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
@@ -696,7 +696,10 @@ private:
 
                 SimpleBSONObjUnorderedSet indexKeyPatterns;
                 auto indexIterator = coll->getIndexCatalog()->getIndexIterator(
-                    opCtx, /* includeUnfinishedIndexes  */ true);
+                    opCtx,
+                    IndexCatalog::InclusionPolicy::kReady |
+                        IndexCatalog::InclusionPolicy::kUnfinished |
+                        IndexCatalog::InclusionPolicy::kFrozen);
                 while (indexIterator->more()) {
                     auto indexKeyPattern = indexIterator->next()->descriptor()->keyPattern();
                     uassert(ErrorCodes::CannotDowngrade,
diff --git a/src/mongo/db/commands/validate_db_metadata_cmd.cpp b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
index 8ef1d91fd7a..f17b5d6d506 100644
--- a/src/mongo/db/commands/validate_db_metadata_cmd.cpp
+++ b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
@@ -211,8 +211,10 @@ public:
 
             // Ensure there are no unstable indexes.
             const auto* indexCatalog = collection->getIndexCatalog();
-            std::unique_ptr<IndexCatalog::IndexIterator> ii =
-                indexCatalog->getIndexIterator(opCtx, true /* includeUnfinishedIndexes */);
+            auto ii = indexCatalog->getIndexIterator(
+                opCtx,
+                IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished |
+                    IndexCatalog::InclusionPolicy::kFrozen);
             while (ii->more()) {
                 // Check if the index is allowed in API version 1.
                 const IndexDescriptor* desc = ii->next()->descriptor();
diff --git a/src/mongo/db/exec/stagedebug_cmd.cpp b/src/mongo/db/exec/stagedebug_cmd.cpp
index 7bc6c4f5423..1dac666e3a5 100644
--- a/src/mongo/db/exec/stagedebug_cmd.cpp
+++ b/src/mongo/db/exec/stagedebug_cmd.cpp
@@ -261,7 +261,7 @@ public:
                 BSONObj keyPatternObj = keyPatternElement.Obj();
                 std::vector<const IndexDescriptor*> indexes;
                 collection->getIndexCatalog()->findIndexesByKeyPattern(
-                    opCtx, keyPatternObj, false, &indexes);
+                    opCtx, keyPatternObj, IndexCatalog::InclusionPolicy::kReady, &indexes);
                 uassert(16890,
                         str::stream() << "Can't find index: " << keyPatternObj,
                         !indexes.empty());
diff --git a/src/mongo/db/index_builds_coordinator.cpp b/src/mongo/db/index_builds_coordinator.cpp
index 23dc532ccb5..4d7336081f7 100644
--- a/src/mongo/db/index_builds_coordinator.cpp
+++ b/src/mongo/db/index_builds_coordinator.cpp
@@ -542,9 +542,8 @@ Status IndexBuildsCoordinator::_startIndexBuildForRecovery(OperationContext* opC
 
 
         for (size_t i = 0; i < indexNames.size(); i++) {
-            bool includeUnfinished = false;
-            auto descriptor =
-                indexCatalog->findIndexByName(opCtx, indexNames[i], includeUnfinished);
+            auto descriptor = indexCatalog->findIndexByName(
+                opCtx, indexNames[i], IndexCatalog::InclusionPolicy::kReady);
             if (descriptor) {
                 Status s =
                     indexCatalog->dropIndex(opCtx, collection.getWritableCollection(), descriptor);
@@ -580,8 +579,11 @@ Status IndexBuildsCoordinator::_startIndexBuildForRecovery(OperationContext* opC
             // If the unfinished index is in the IndexCatalog, drop it through there, otherwise drop
             // it from the DurableCatalog. Rollback-via-refetch does not clear any in-memory state,
             // so we should do it manually here.
-            includeUnfinished = true;
-            descriptor = indexCatalog->findIndexByName(opCtx, indexNames[i], includeUnfinished);
+            descriptor = indexCatalog->findIndexByName(
+                opCtx,
+                indexNames[i],
+                IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished |
+                    IndexCatalog::InclusionPolicy::kFrozen);
             if (descriptor) {
                 Status s = indexCatalog->dropUnfinishedIndex(
                     opCtx, collection.getWritableCollection(), descriptor);
@@ -845,14 +847,14 @@ void IndexBuildsCoordinator::applyStartIndexBuild(OperationContext* opCtx,
 
             IndexCatalog* indexCatalog = coll.getWritableCollection()->getIndexCatalog();
 
-            const bool includeUnfinished = false;
             for (const auto& spec : oplogEntry.indexSpecs) {
                 std::string name = spec.getStringField(IndexDescriptor::kIndexNameFieldName);
                 uassert(ErrorCodes::BadValue,
                         str::stream() << "Index spec is missing the 'name' field " << spec,
                         !name.empty());
 
-                if (auto desc = indexCatalog->findIndexByName(opCtx, name, includeUnfinished)) {
+                if (auto desc = indexCatalog->findIndexByName(
+                        opCtx, name, IndexCatalog::InclusionPolicy::kReady)) {
                     uassertStatusOK(
                         indexCatalog->dropIndex(opCtx, coll.getWritableCollection(), desc));
                 }
@@ -1040,7 +1042,8 @@ void IndexBuildsCoordinator::applyAbortIndexBuild(OperationContext* opCtx,
             const IndexDescriptor* desc = indexCatalog->findIndexByName(
                 opCtx,
                 indexSpec.getStringField(IndexDescriptor::kIndexNameFieldName),
-                /*includeUnfinishedIndexes=*/true);
+                IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished |
+                    IndexCatalog::InclusionPolicy::kFrozen);
 
             LOGV2(6455400,
                   "Dropping unfinished index during oplog recovery as standalone",
diff --git a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
index 38d6222a843..23c1e73af81 100644
--- a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
@@ -197,7 +197,8 @@ std::vector<Document> CommonMongodProcessInterface::getIndexStats(OperationConte
         auto idxCatalog = collection->getIndexCatalog();
         auto idx = idxCatalog->findIndexByName(opCtx,
                                                indexName,
-                                               /* includeUnfinishedIndexes */ true);
+                                               IndexCatalog::InclusionPolicy::kReady |
+                                                   IndexCatalog::InclusionPolicy::kUnfinished);
         uassert(ErrorCodes::IndexNotFound,
                 "Could not find entry in IndexCatalog for index " + indexName,
                 idx);
@@ -468,7 +469,8 @@ bool CommonMongodProcessInterface::fieldsHaveSupportingUniqueIndex(
         return fieldPaths == std::set<FieldPath>{"_id"};
     }
 
-    auto indexIterator = collection->getIndexCatalog()->getIndexIterator(opCtx, false);
+    auto indexIterator = collection->getIndexCatalog()->getIndexIterator(
+        opCtx, IndexCatalog::InclusionPolicy::kReady);
     while (indexIterator->more()) {
         const IndexCatalogEntry* entry = indexIterator->next();
         if (supportsUniqueKey(expCtx, entry, fieldPaths)) {
diff --git a/src/mongo/db/query/collection_query_info.cpp b/src/mongo/db/query/collection_query_info.cpp
index 1a87201bae3..9af41919d1d 100644
--- a/src/mongo/db/query/collection_query_info.cpp
+++ b/src/mongo/db/query/collection_query_info.cpp
@@ -90,8 +90,8 @@ const UpdateIndexData& CollectionQueryInfo::getIndexKeys(OperationContext* opCtx
 void CollectionQueryInfo::computeIndexKeys(OperationContext* opCtx, const CollectionPtr& coll) {
     _indexedPaths.clear();
 
-    std::unique_ptr<IndexCatalog::IndexIterator> it =
-        coll->getIndexCatalog()->getIndexIterator(opCtx, true);
+    auto it = coll->getIndexCatalog()->getIndexIterator(
+        opCtx, IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
     while (it->more()) {
         const IndexCatalogEntry* entry = it->next();
         const IndexDescriptor* descriptor = entry->descriptor();
@@ -216,21 +216,21 @@ void CollectionQueryInfo::updatePlanCacheIndexEntries(OperationContext* opCtx,
 
     // TODO We shouldn't need to include unfinished indexes, but we must here because the index
     // catalog may be in an inconsistent state.  SERVER-18346.
-    const bool includeUnfinishedIndexes = true;
-    std::unique_ptr<IndexCatalog::IndexIterator> ii =
-        coll->getIndexCatalog()->getIndexIterator(opCtx, includeUnfinishedIndexes);
+    std::unique_ptr<IndexCatalog::IndexIterator> ii = coll->getIndexCatalog()->getIndexIterator(
+        opCtx, IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
     while (ii->more()) {
         const IndexCatalogEntry* ice = ii->next();
-        indexCores.emplace_back(indexInfoFromIndexCatalogEntry(*ice));
+        if (ice->accessMethod()) {
+            indexCores.emplace_back(indexInfoFromIndexCatalogEntry(*ice));
+        }
     }
 
     _planCache->notifyOfIndexUpdates(indexCores);
 }
 
 void CollectionQueryInfo::init(OperationContext* opCtx, const CollectionPtr& coll) {
-    const bool includeUnfinishedIndexes = false;
-    std::unique_ptr<IndexCatalog::IndexIterator> ii =
-        coll->getIndexCatalog()->getIndexIterator(opCtx, includeUnfinishedIndexes);
+    auto ii =
+        coll->getIndexCatalog()->getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
     while (ii->more()) {
         const IndexDescriptor* desc = ii->next()->descriptor();
         CollectionIndexUsageTrackerDecoration::get(coll->getSharedDecorations())
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index 28e3c96934b..0b1b9566ae3 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -275,7 +275,8 @@ void fillOutPlannerParams(OperationContext* opCtx,
     bool apiStrict = APIParameters::get(opCtx).getAPIStrict().value_or(false);
     // If it's not NULL, we may have indices.  Access the catalog and fill out IndexEntry(s)
     std::unique_ptr<IndexCatalog::IndexIterator> ii =
-        collection->getIndexCatalog()->getIndexIterator(opCtx, false);
+        collection->getIndexCatalog()->getIndexIterator(opCtx,
+                                                        IndexCatalog::InclusionPolicy::kReady);
     while (ii->more()) {
         const IndexCatalogEntry* ice = ii->next();
 
@@ -2176,8 +2177,8 @@ QueryPlannerParams fillOutPlannerParamsForDistinct(OperationContext* opCtx,
     // If the caller did not request a "strict" distinct scan then we may choose a plan which
     // unwinds arrays and treats each element in an array as its own key.
     const bool mayUnwindArrays = !(plannerOptions & QueryPlannerParams::STRICT_DISTINCT_ONLY);
-    std::unique_ptr<IndexCatalog::IndexIterator> ii =
-        collection->getIndexCatalog()->getIndexIterator(opCtx, false);
+    auto ii = collection->getIndexCatalog()->getIndexIterator(
+        opCtx, IndexCatalog::InclusionPolicy::kReady);
     auto query = parsedDistinct.getQuery()->getFindCommandRequest().getFilter();
     while (ii->more()) {
         const IndexCatalogEntry* ice = ii->next();
diff --git a/src/mongo/db/repl/rs_rollback.cpp b/src/mongo/db/repl/rs_rollback.cpp
index c0baecbb166..95cad0836a4 100644
--- a/src/mongo/db/repl/rs_rollback.cpp
+++ b/src/mongo/db/repl/rs_rollback.cpp
@@ -860,9 +860,10 @@ void dropIndex(OperationContext* opCtx,
                const string& indexName,
                NamespaceString& nss) {
     IndexCatalog* indexCatalog = collection->getIndexCatalog();
-    bool includeUnfinishedIndexes = true;
-    auto indexDescriptor =
-        indexCatalog->findIndexByName(opCtx, indexName, includeUnfinishedIndexes);
+    auto indexDescriptor = indexCatalog->findIndexByName(
+        opCtx,
+        indexName,
+        IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
     if (!indexDescriptor) {
         LOGV2_WARNING(21725,
                       "Rollback failed to drop index {indexName} in {namespace}: index not found.",
diff --git a/src/mongo/db/repl/rs_rollback_test.cpp b/src/mongo/db/repl/rs_rollback_test.cpp
index ab900376345..420e0e2e136 100644
--- a/src/mongo/db/repl/rs_rollback_test.cpp
+++ b/src/mongo/db/repl/rs_rollback_test.cpp
@@ -714,7 +714,8 @@ TEST_F(RSRollbackTest, RollingBackCreateAndDropOfSameIndexIgnoresBothCommands) {
                                 ->getIndexCatalog();
         ASSERT(indexCatalog);
         ASSERT_EQUALS(1, indexCatalog->numIndexesReady(_opCtx.get()));
-        auto indexDescriptor = indexCatalog->findIndexByName(_opCtx.get(), "a_1", false);
+        auto indexDescriptor = indexCatalog->findIndexByName(
+            _opCtx.get(), "a_1", IndexCatalog::InclusionPolicy::kReady);
         ASSERT(!indexDescriptor);
     }
 }
@@ -764,7 +765,8 @@ TEST_F(RSRollbackTest, RollingBackCreateIndexAndRenameWithLongName) {
         ASSERT_EQUALS(1, indexCatalog->numIndexesReady(_opCtx.get()));
 
         std::vector<const IndexDescriptor*> indexes;
-        indexCatalog->findIndexesByKeyPattern(_opCtx.get(), BSON("b" << 1), false, &indexes);
+        indexCatalog->findIndexesByKeyPattern(
+            _opCtx.get(), BSON("b" << 1), IndexCatalog::InclusionPolicy::kReady, &indexes);
         ASSERT(indexes.size() == 0);
     }
 }
@@ -813,12 +815,14 @@ TEST_F(RSRollbackTest, RollingBackDropAndCreateOfSameIndexNameWithDifferentSpecs
         ASSERT_EQUALS(1, countTextFormatLogLinesContaining("Dropped index in rollback"));
         ASSERT_EQUALS(1, countTextFormatLogLinesContaining("Created index in rollback"));
         std::vector<const IndexDescriptor*> indexes;
-        indexCatalog->findIndexesByKeyPattern(_opCtx.get(), BSON("a" << 1), false, &indexes);
+        indexCatalog->findIndexesByKeyPattern(
+            _opCtx.get(), BSON("a" << 1), IndexCatalog::InclusionPolicy::kReady, &indexes);
         ASSERT(indexes.size() == 1);
         ASSERT(indexes[0]->indexName() == "a_1");
 
         std::vector<const IndexDescriptor*> indexes2;
-        indexCatalog->findIndexesByKeyPattern(_opCtx.get(), BSON("b" << 1), false, &indexes2);
+        indexCatalog->findIndexesByKeyPattern(
+            _opCtx.get(), BSON("b" << 1), IndexCatalog::InclusionPolicy::kReady, &indexes2);
         ASSERT(indexes2.size() == 0);
     }
 }
diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index ffc1f1ad18e..3394c518edc 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -625,7 +625,9 @@ Status StorageInterfaceImpl::setIndexIsMultikey(OperationContext* opCtx,
         }
 
         auto idx = collection->getIndexCatalog()->findIndexByName(
-            opCtx, indexName, true /* includeUnfinishedIndexes */);
+            opCtx,
+            indexName,
+            IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
         if (!idx) {
             return Status(ErrorCodes::IndexNotFound,
                           str::stream()
@@ -754,9 +756,8 @@ StatusWith<std::vector<BSONObj>> _findOrDeleteDocuments(
                 // Use index scan.
                 auto indexCatalog = collection->getIndexCatalog();
                 invariant(indexCatalog);
-                bool includeUnfinishedIndexes = false;
-                const IndexDescriptor* indexDescriptor =
-                    indexCatalog->findIndexByName(opCtx, *indexName, includeUnfinishedIndexes);
+                const IndexDescriptor* indexDescriptor = indexCatalog->findIndexByName(
+                    opCtx, *indexName, IndexCatalog::InclusionPolicy::kReady);
                 if (!indexDescriptor) {
                     return Result(ErrorCodes::IndexNotFound,
                                   str::stream() << "Index not found, ns:" << nsOrUUID.toString()
diff --git a/src/mongo/db/stats/storage_stats.cpp b/src/mongo/db/stats/storage_stats.cpp
index 96067384f6a..3e99a529fb1 100644
--- a/src/mongo/db/stats/storage_stats.cpp
+++ b/src/mongo/db/stats/storage_stats.cpp
@@ -128,8 +128,8 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
     BSONObjBuilder indexDetails;
     std::vector<std::string> indexBuilds;
 
-    std::unique_ptr<IndexCatalog::IndexIterator> it =
-        indexCatalog->getIndexIterator(opCtx, /*includeUnfinishedIndexes=*/true);
+    auto it = indexCatalog->getIndexIterator(
+        opCtx, IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
     while (it->more()) {
         const IndexCatalogEntry* entry = it->next();
         const IndexDescriptor* descriptor = entry->descriptor();
diff --git a/src/mongo/db/storage/storage_debug_util.cpp b/src/mongo/db/storage/storage_debug_util.cpp
index 3b4fe5bc98a..cbd8a4139e1 100644
--- a/src/mongo/db/storage/storage_debug_util.cpp
+++ b/src/mongo/db/storage/storage_debug_util.cpp
@@ -94,7 +94,7 @@ void printCollectionAndIndexTableEntries(OperationContext* opCtx, const Namespac
 
     // Iterate and print each index's table of documents.
     const auto indexCatalog = coll->getIndexCatalog();
-    const auto it = indexCatalog->getIndexIterator(opCtx, /*includeUnfinished*/ false);
+    const auto it = indexCatalog->getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
     while (it->more()) {
         const auto indexCatalogEntry = it->next();
         const auto indexDescriptor = indexCatalogEntry->descriptor();
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index a5b0926d3c4..3bb168a039f 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -602,21 +602,19 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
                     "namespace"_attr = coll);
             }
 
-            // Two-phase index drop ensures that the underlying data table for an index in the
-            // catalog is not dropped until the index removal from the catalog has been majority
-            // committed and become part of the latest checkpoint. Therefore, there should almost
-            // never be a case where the index catalog entry remains but the index table (identified
-            // by ident) has been removed.
-            //
-            // There is an exception to this due to the fact that we drop the index ident without a
-            // timestamp when restarting an index build for startup recovery. Then, if we experience
-            // an unclean shutdown before a checkpoint is taken, the subsequent startup recovery can
-            // see the now-dropped ident referenced by the old index catalog entry.
-            invariant(engineIdents.find(indexIdent) != engineIdents.end() ||
-                          lastShutdownState == LastShutdownState::kUnclean,
-                      str::stream() << "Failed to find an index data table matching " << indexIdent
-                                    << " for durable index catalog entry " << indexMetaData.spec
-                                    << " in collection " << coll);
+            if (!engineIdents.count(indexIdent)) {
+                // There are cetain cases where the catalog entry may reference an index ident which
+                // is no longer present. One example of this is when an unclean shutdown occurs
+                // before a checkpoint is taken during startup recovery. Since we drop the index
+                // ident without a timestamp when restarting the index build for startup recovery,
+                // the subsequent startup recovery can see the now-dropped ident referenced by the
+                // old index catalog entry.
+                LOGV2(6386500,
+                      "Index catalog entry ident not found",
+                      "ident"_attr = indexIdent,
+                      "entry"_attr = indexMetaData.spec,
+                      "namespace"_attr = coll);
+            }
 
             // Any index build with a UUID is an unfinished two-phase build and must be restarted.
             // There are no special cases to handle on primaries or secondaries. An index build may
@@ -818,8 +816,10 @@ Status StorageEngineImpl::_dropCollectionsNoTimestamp(OperationContext* opCtx,
 
         // No need to remove the indexes from the IndexCatalog because eliminating the Collection
         // will have the same effect.
-        auto ii =
-            coll->getIndexCatalog()->getIndexIterator(opCtx, true /* includeUnfinishedIndexes */);
+        auto ii = coll->getIndexCatalog()->getIndexIterator(
+            opCtx,
+            IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished |
+                IndexCatalog::InclusionPolicy::kFrozen);
         while (ii->more()) {
             const IndexCatalogEntry* ice = ii->next();
 
@@ -1252,7 +1252,9 @@ int64_t StorageEngineImpl::sizeOnDiskForDb(OperationContext* opCtx, StringData d
     catalog::forEachCollectionFromDb(opCtx, dbName, MODE_IS, [&](const CollectionPtr& collection) {
         size += collection->getRecordStore()->storageSize(opCtx);
 
-        auto it = collection->getIndexCatalog()->getIndexIterator(opCtx, true);
+        auto it = collection->getIndexCatalog()->getIndexIterator(
+            opCtx,
+            IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
         while (it->more()) {
             size += _engine->getIdentSize(opCtx, it->next()->getIdent());
         }
diff --git a/src/mongo/db/storage/storage_engine_test_fixture.h b/src/mongo/db/storage/storage_engine_test_fixture.h
index 57b50a1d32c..8762a2934e3 100644
--- a/src/mongo/db/storage/storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/storage_engine_test_fixture.h
@@ -178,7 +178,10 @@ public:
         Collection* collection =
             CollectionCatalog::get(opCtx)->lookupCollectionByNamespaceForMetadataWrite(
                 opCtx, CollectionCatalog::LifetimeMode::kInplace, collNs);
-        auto descriptor = collection->getIndexCatalog()->findIndexByName(opCtx, key, true);
+        auto descriptor = collection->getIndexCatalog()->findIndexByName(
+            opCtx,
+            key,
+            IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
         collection->indexBuildSuccess(opCtx, descriptor->getEntry());
     }
 
diff --git a/src/mongo/db/system_index.cpp b/src/mongo/db/system_index.cpp
index 2e6e19b04ae..95a6d16bac8 100644
--- a/src/mongo/db/system_index.cpp
+++ b/src/mongo/db/system_index.cpp
@@ -149,7 +149,8 @@ Status verifySystemIndexes(OperationContext* opCtx) {
 
             // Make sure the old unique index from v2.4 on system.users doesn't exist.
             std::vector<const IndexDescriptor*> indexes;
-            indexCatalog->findIndexesByKeyPattern(opCtx, v1SystemUsersKeyPattern, false, &indexes);
+            indexCatalog->findIndexesByKeyPattern(
+                opCtx, v1SystemUsersKeyPattern, IndexCatalog::InclusionPolicy::kReady, &indexes);
 
             if (!indexes.empty()) {
                 fassert(ErrorCodes::AmbiguousIndexKeyPattern, indexes.size() == 1);
@@ -160,7 +161,8 @@ Status verifySystemIndexes(OperationContext* opCtx) {
             }
 
             // Ensure that system indexes exist for the user collection.
-            indexCatalog->findIndexesByKeyPattern(opCtx, v3SystemUsersKeyPattern, false, &indexes);
+            indexCatalog->findIndexesByKeyPattern(
+                opCtx, v3SystemUsersKeyPattern, IndexCatalog::InclusionPolicy::kReady, &indexes);
             if (indexes.empty()) {
                 try {
                     generateSystemIndexForExistingCollection(
@@ -182,7 +184,8 @@ Status verifySystemIndexes(OperationContext* opCtx) {
             invariant(indexCatalog);
 
             std::vector<const IndexDescriptor*> indexes;
-            indexCatalog->findIndexesByKeyPattern(opCtx, v3SystemRolesKeyPattern, false, &indexes);
+            indexCatalog->findIndexesByKeyPattern(
+                opCtx, v3SystemRolesKeyPattern, IndexCatalog::InclusionPolicy::kReady, &indexes);
             if (indexes.empty()) {
                 try {
                     generateSystemIndexForExistingCollection(
diff --git a/src/mongo/db/timeseries/timeseries_extended_range.cpp b/src/mongo/db/timeseries/timeseries_extended_range.cpp
index b6ceb6e8158..44a7ff060d9 100644
--- a/src/mongo/db/timeseries/timeseries_extended_range.cpp
+++ b/src/mongo/db/timeseries/timeseries_extended_range.cpp
@@ -99,7 +99,8 @@ bool collectionHasTimeIndex(OperationContext* opCtx, const Collection& collectio
     auto indexCatalog = collection.getIndexCatalog();
     // The IndexIterator is initialized lazily, so the first call to 'next' positions it to the
     // first entry.
-    for (auto it = indexCatalog->getIndexIterator(opCtx, false); it->more();) {
+    for (auto it = indexCatalog->getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
+         it->more();) {
         auto index = it->next();
         auto desc = index->descriptor();
         auto pattern = desc->keyPattern();
diff --git a/src/mongo/dbtests/indexcatalogtests.cpp b/src/mongo/dbtests/indexcatalogtests.cpp
index 0afe4634f4f..1403378adc8 100644
--- a/src/mongo/dbtests/indexcatalogtests.cpp
+++ b/src/mongo/dbtests/indexcatalogtests.cpp
@@ -81,7 +81,8 @@ public:
 
         ASSERT_TRUE(_catalog->numIndexesReady(&opCtx) == numFinishedIndexesStart + 2);
 
-        std::unique_ptr<IndexCatalog::IndexIterator> ii = _catalog->getIndexIterator(&opCtx, false);
+        std::unique_ptr<IndexCatalog::IndexIterator> ii =
+            _catalog->getIndexIterator(&opCtx, IndexCatalog::InclusionPolicy::kReady);
         int indexesIterated = 0;
         bool foundIndex = false;
         while (ii->more()) {
diff --git a/src/mongo/dbtests/indexupdatetests.cpp b/src/mongo/dbtests/indexupdatetests.cpp
index 497e630135f..4803aaea9ff 100644
--- a/src/mongo/dbtests/indexupdatetests.cpp
+++ b/src/mongo/dbtests/indexupdatetests.cpp
@@ -223,8 +223,10 @@ public:
                           .getStatus());
 
             auto& coll = collection();
-            auto desc =
-                coll->getIndexCatalog()->findIndexByName(_opCtx, "a", true /* includeUnfinished */);
+            auto desc = coll->getIndexCatalog()->findIndexByName(
+                _opCtx,
+                "a",
+                IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
             ASSERT(desc);
 
             // Hybrid index builds check duplicates explicitly.
diff --git a/src/mongo/dbtests/multikey_paths_test.cpp b/src/mongo/dbtests/multikey_paths_test.cpp
index 6443c2d15d5..8e62c1c1ed2 100644
--- a/src/mongo/dbtests/multikey_paths_test.cpp
+++ b/src/mongo/dbtests/multikey_paths_test.cpp
@@ -88,7 +88,8 @@ public:
                              const MultikeyPaths& expectedMultikeyPaths) {
         const IndexCatalog* indexCatalog = collection->getIndexCatalog();
         std::vector<const IndexDescriptor*> indexes;
-        indexCatalog->findIndexesByKeyPattern(_opCtx.get(), keyPattern, false, &indexes);
+        indexCatalog->findIndexesByKeyPattern(
+            _opCtx.get(), keyPattern, IndexCatalog::InclusionPolicy::kReady, &indexes);
         ASSERT_EQ(indexes.size(), 1U);
         auto desc = indexes[0];
         const IndexCatalogEntry* ice = indexCatalog->getEntry(desc);
diff --git a/src/mongo/dbtests/query_plan_executor.cpp b/src/mongo/dbtests/query_plan_executor.cpp
index 3a71e0be8e2..4a777f76fee 100644
--- a/src/mongo/dbtests/query_plan_executor.cpp
+++ b/src/mongo/dbtests/query_plan_executor.cpp
@@ -185,7 +185,8 @@ private:
         CollectionPtr collection =
             CollectionCatalog::get(&_opCtx)->lookupCollectionByNamespace(&_opCtx, nss);
         std::vector<const IndexDescriptor*> indexes;
-        collection->getIndexCatalog()->findIndexesByKeyPattern(&_opCtx, obj, false, &indexes);
+        collection->getIndexCatalog()->findIndexesByKeyPattern(
+            &_opCtx, obj, IndexCatalog::InclusionPolicy::kReady, &indexes);
         ASSERT_LTE(indexes.size(), 1U);
         return indexes.size() == 0 ? nullptr : indexes[0];
     }
diff --git a/src/mongo/dbtests/query_stage_and.cpp b/src/mongo/dbtests/query_stage_and.cpp
index 064f2a59236..f18abe0a523 100644
--- a/src/mongo/dbtests/query_stage_and.cpp
+++ b/src/mongo/dbtests/query_stage_and.cpp
@@ -75,7 +75,8 @@ public:
 
     const IndexDescriptor* getIndex(const BSONObj& obj, const CollectionPtr& coll) {
         std::vector<const IndexDescriptor*> indexes;
-        coll->getIndexCatalog()->findIndexesByKeyPattern(&_opCtx, obj, false, &indexes);
+        coll->getIndexCatalog()->findIndexesByKeyPattern(
+            &_opCtx, obj, IndexCatalog::InclusionPolicy::kReady, &indexes);
         if (indexes.empty()) {
             FAIL(str::stream() << "Unable to find index with key pattern " << obj);
         }
diff --git a/src/mongo/dbtests/query_stage_count.cpp b/src/mongo/dbtests/query_stage_count.cpp
index e6e32ddec35..b66773c5c1b 100644
--- a/src/mongo/dbtests/query_stage_count.cpp
+++ b/src/mongo/dbtests/query_stage_count.cpp
@@ -203,7 +203,8 @@ public:
     IndexScan* createIndexScan(MatchExpression* expr, WorkingSet* ws) {
         const IndexCatalog* catalog = _coll->getIndexCatalog();
         std::vector<const IndexDescriptor*> indexes;
-        catalog->findIndexesByKeyPattern(&_opCtx, BSON("x" << 1), false, &indexes);
+        catalog->findIndexesByKeyPattern(
+            &_opCtx, BSON("x" << 1), IndexCatalog::InclusionPolicy::kReady, &indexes);
         ASSERT_EQ(indexes.size(), 1U);
         auto descriptor = indexes[0];
 
diff --git a/src/mongo/dbtests/query_stage_count_scan.cpp b/src/mongo/dbtests/query_stage_count_scan.cpp
index dc9a544bede..8da097f42d0 100644
--- a/src/mongo/dbtests/query_stage_count_scan.cpp
+++ b/src/mongo/dbtests/query_stage_count_scan.cpp
@@ -95,7 +95,8 @@ public:
 
     const IndexDescriptor* getIndex(Database* db, const BSONObj& obj) {
         std::vector<const IndexDescriptor*> indexes;
-        getCollection()->getIndexCatalog()->findIndexesByKeyPattern(&_opCtx, obj, false, &indexes);
+        getCollection()->getIndexCatalog()->findIndexesByKeyPattern(
+            &_opCtx, obj, IndexCatalog::InclusionPolicy::kReady, &indexes);
         return indexes.empty() ? nullptr : indexes[0];
     }
 
diff --git a/src/mongo/dbtests/query_stage_distinct.cpp b/src/mongo/dbtests/query_stage_distinct.cpp
index e79b260493d..11198612d50 100644
--- a/src/mongo/dbtests/query_stage_distinct.cpp
+++ b/src/mongo/dbtests/query_stage_distinct.cpp
@@ -130,7 +130,8 @@ public:
 
         // Set up the distinct stage.
         std::vector<const IndexDescriptor*> indexes;
-        coll->getIndexCatalog()->findIndexesByKeyPattern(&_opCtx, BSON("a" << 1), false, &indexes);
+        coll->getIndexCatalog()->findIndexesByKeyPattern(
+            &_opCtx, BSON("a" << 1), IndexCatalog::InclusionPolicy::kReady, &indexes);
         ASSERT_EQ(indexes.size(), 1U);
 
         DistinctParams params{&_opCtx, indexes[0]};
@@ -196,7 +197,8 @@ public:
 
         // Set up the distinct stage.
         std::vector<const IndexDescriptor*> indexes;
-        coll->getIndexCatalog()->findIndexesByKeyPattern(&_opCtx, BSON("a" << 1), false, &indexes);
+        coll->getIndexCatalog()->findIndexesByKeyPattern(
+            &_opCtx, BSON("a" << 1), IndexCatalog::InclusionPolicy::kReady, &indexes);
         verify(indexes.size() == 1);
 
         DistinctParams params{&_opCtx, indexes[0]};
@@ -263,7 +265,7 @@ public:
 
         std::vector<const IndexDescriptor*> indices;
         coll->getIndexCatalog()->findIndexesByKeyPattern(
-            &_opCtx, BSON("a" << 1 << "b" << 1), false, &indices);
+            &_opCtx, BSON("a" << 1 << "b" << 1), IndexCatalog::InclusionPolicy::kReady, &indices);
         ASSERT_EQ(1U, indices.size());
 
         DistinctParams params{&_opCtx, indices[0]};
diff --git a/src/mongo/dbtests/query_stage_ixscan.cpp b/src/mongo/dbtests/query_stage_ixscan.cpp
index b574b51ec84..6c77f40bb68 100644
--- a/src/mongo/dbtests/query_stage_ixscan.cpp
+++ b/src/mongo/dbtests/query_stage_ixscan.cpp
@@ -97,7 +97,8 @@ public:
     IndexScan* createIndexScanSimpleRange(BSONObj startKey, BSONObj endKey) {
         IndexCatalog* catalog = _coll->getIndexCatalog();
         std::vector<const IndexDescriptor*> indexes;
-        catalog->findIndexesByKeyPattern(&_opCtx, BSON("x" << 1), false, &indexes);
+        catalog->findIndexesByKeyPattern(
+            &_opCtx, BSON("x" << 1), IndexCatalog::InclusionPolicy::kReady, &indexes);
         ASSERT_EQ(indexes.size(), 1U);
 
         // We are not testing indexing here so use maximal bounds
@@ -120,7 +121,8 @@ public:
                                int direction = 1) {
         IndexCatalog* catalog = _coll->getIndexCatalog();
         std::vector<const IndexDescriptor*> indexes;
-        catalog->findIndexesByKeyPattern(&_opCtx, BSON("x" << 1), false, &indexes);
+        catalog->findIndexesByKeyPattern(
+            &_opCtx, BSON("x" << 1), IndexCatalog::InclusionPolicy::kReady, &indexes);
         ASSERT_EQ(indexes.size(), 1U);
 
         IndexScanParams params(&_opCtx, indexes[0]);
diff --git a/src/mongo/dbtests/query_stage_merge_sort.cpp b/src/mongo/dbtests/query_stage_merge_sort.cpp
index a504c629f1e..45286e8237d 100644
--- a/src/mongo/dbtests/query_stage_merge_sort.cpp
+++ b/src/mongo/dbtests/query_stage_merge_sort.cpp
@@ -74,7 +74,8 @@ public:
 
     const IndexDescriptor* getIndex(const BSONObj& obj, const CollectionPtr& coll) {
         std::vector<const IndexDescriptor*> indexes;
-        coll->getIndexCatalog()->findIndexesByKeyPattern(&_opCtx, obj, false, &indexes);
+        coll->getIndexCatalog()->findIndexesByKeyPattern(
+            &_opCtx, obj, IndexCatalog::InclusionPolicy::kReady, &indexes);
         return indexes.empty() ? nullptr : indexes[0];
     }
 
diff --git a/src/mongo/dbtests/query_stage_multiplan.cpp b/src/mongo/dbtests/query_stage_multiplan.cpp
index 2093f8f4ad4..c81882ffc67 100644
--- a/src/mongo/dbtests/query_stage_multiplan.cpp
+++ b/src/mongo/dbtests/query_stage_multiplan.cpp
@@ -145,7 +145,7 @@ unique_ptr<PlanStage> getIxScanPlan(ExpressionContext* expCtx,
                                     int desiredFooValue) {
     std::vector<const IndexDescriptor*> indexes;
     coll->getIndexCatalog()->findIndexesByKeyPattern(
-        expCtx->opCtx, BSON("foo" << 1), false, &indexes);
+        expCtx->opCtx, BSON("foo" << 1), IndexCatalog::InclusionPolicy::kReady, &indexes);
     ASSERT_EQ(indexes.size(), 1U);
 
     IndexScanParams ixparams(expCtx->opCtx, indexes[0]);
diff --git a/src/mongo/dbtests/query_stage_tests.cpp b/src/mongo/dbtests/query_stage_tests.cpp
index 739b397ea94..8d6c4c933e8 100644
--- a/src/mongo/dbtests/query_stage_tests.cpp
+++ b/src/mongo/dbtests/query_stage_tests.cpp
@@ -126,7 +126,8 @@ public:
     const IndexDescriptor* getIndex(const BSONObj& obj) {
         AutoGetCollectionForReadCommand collection(&_opCtx, NamespaceString(ns()));
         std::vector<const IndexDescriptor*> indexes;
-        collection->getIndexCatalog()->findIndexesByKeyPattern(&_opCtx, obj, false, &indexes);
+        collection->getIndexCatalog()->findIndexesByKeyPattern(
+            &_opCtx, obj, IndexCatalog::InclusionPolicy::kReady, &indexes);
         return indexes.empty() ? nullptr : indexes[0];
     }
 
diff --git a/src/mongo/dbtests/rollbacktests.cpp b/src/mongo/dbtests/rollbacktests.cpp
index 95310d74bc1..e622619e951 100644
--- a/src/mongo/dbtests/rollbacktests.cpp
+++ b/src/mongo/dbtests/rollbacktests.cpp
@@ -110,11 +110,16 @@ void assertEmpty(OperationContext* opCtx, const NamespaceString& nss) {
 }
 bool indexExists(OperationContext* opCtx, const NamespaceString& nss, const string& idxName) {
     auto coll = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss);
-    return coll->getIndexCatalog()->findIndexByName(opCtx, idxName, true) != nullptr;
+    return coll->getIndexCatalog()->findIndexByName(
+               opCtx,
+               idxName,
+               IndexCatalog::InclusionPolicy::kReady |
+                   IndexCatalog::InclusionPolicy::kUnfinished) != nullptr;
 }
 bool indexReady(OperationContext* opCtx, const NamespaceString& nss, const string& idxName) {
     auto coll = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss);
-    return coll->getIndexCatalog()->findIndexByName(opCtx, idxName, false) != nullptr;
+    return coll->getIndexCatalog()->findIndexByName(
+               opCtx, idxName, IndexCatalog::InclusionPolicy::kReady) != nullptr;
 }
 size_t getNumIndexEntries(OperationContext* opCtx,
                           const NamespaceString& nss,
@@ -123,7 +128,7 @@ size_t getNumIndexEntries(OperationContext* opCtx,
 
     auto coll = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss);
     const IndexCatalog* catalog = coll->getIndexCatalog();
-    auto desc = catalog->findIndexByName(opCtx, idxName, false);
+    auto desc = catalog->findIndexByName(opCtx, idxName, IndexCatalog::InclusionPolicy::kReady);
 
     if (desc) {
         auto cursor = catalog->getEntry(desc)->accessMethod()->newCursor(opCtx);
diff --git a/src/mongo/dbtests/storage_timestamp_tests.cpp b/src/mongo/dbtests/storage_timestamp_tests.cpp
index f2cee34a605..06f3f931d8b 100644
--- a/src/mongo/dbtests/storage_timestamp_tests.cpp
+++ b/src/mongo/dbtests/storage_timestamp_tests.cpp
@@ -2277,7 +2277,10 @@ public:
 
         auto indexCatalog = autoColl.getCollection()->getIndexCatalog();
         const IndexCatalogEntry* buildingIndex = indexCatalog->getEntry(
-            indexCatalog->findIndexByName(_opCtx, "a_1", /* includeUnfinished */ true));
+            indexCatalog->findIndexByName(_opCtx,
+                                          "a_1",
+                                          IndexCatalog::InclusionPolicy::kReady |
+                                              IndexCatalog::InclusionPolicy::kUnfinished));
         ASSERT(buildingIndex);
 
         {
@@ -2992,7 +2995,10 @@ public:
 
             auto indexCatalog = collection->getIndexCatalog();
             buildingIndex = indexCatalog->getEntry(
-                indexCatalog->findIndexByName(_opCtx, "a_1_b_1", /* includeUnfinished */ true));
+                indexCatalog->findIndexByName(_opCtx,
+                                              "a_1_b_1",
+                                              IndexCatalog::InclusionPolicy::kReady |
+                                                  IndexCatalog::InclusionPolicy::kUnfinished));
             ASSERT(buildingIndex);
 
             ASSERT_OK(indexer.insertAllDocumentsInCollection(_opCtx, collection.get()));