SERVER-63865 Handle missing index idents during standalone startup recovery

29 files changed, 271 insertions, 163 deletions

diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index 0f88e0fc07a..0072c7a9f15 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -288,7 +288,8 @@ StatusWith<std::pair<ParsedCollModRequest, BSONObj>> parseCollModRequest(Operati
             }
         } else {
             std::vector<const IndexDescriptor*> indexes;
-            coll->getIndexCatalog()->findIndexesByKeyPattern(opCtx, keyPattern, false, &indexes);
+            coll->getIndexCatalog()->findIndexesByKeyPattern(
+                opCtx, keyPattern, IndexCatalog::InclusionPolicy::kReady, &indexes);
 
             if (indexes.size() > 1) {
                 return {ErrorCodes::AmbiguousIndexKeyPattern,
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index ecb86d77238..e57cda3743a 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -1726,7 +1726,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;
 
@@ -1748,8 +1749,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()) {
@@ -1773,8 +1774,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 610d7c39ca7..8fa78a9ba9f 100644
--- a/src/mongo/db/catalog/collection_validation.cpp
+++ b/src/mongo/db/catalog/collection_validation.cpp
@@ -80,8 +80,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.
@@ -412,7 +411,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/drop_indexes.cpp b/src/mongo/db/catalog/drop_indexes.cpp
index e003e0e56ef..493c8326146 100644
--- a/src/mongo/db/catalog/drop_indexes.cpp
+++ b/src/mongo/db/catalog/drop_indexes.cpp
@@ -109,9 +109,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);
@@ -340,7 +344,6 @@ void dropReadyIndexes(OperationContext* opCtx,
         return;
     }
 
-    bool includeUnfinished = true;
     for (const auto& indexName : indexNames) {
         if (collDescription.isSharded()) {
             uassert(
@@ -350,7 +353,11 @@ void dropReadyIndexes(OperationContext* opCtx,
                     opCtx, collection, indexCatalog, indexName, collDescription.getKeyPattern()));
         }
 
-        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 << "]");
@@ -509,7 +516,6 @@ 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(opCtx)->getIndexCatalog();
-            const bool includeUnfinished = false;
             for (const auto& indexName : indexNames) {
                 auto collDescription =
                     CollectionShardingState::get(opCtx, nss)->getCollectionDescription(opCtx);
@@ -524,7 +530,12 @@ DropIndexesReply dropIndexes(OperationContext* opCtx,
                                                  collDescription.getKeyPattern()));
                 }
 
-                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 fc5d39109d8..6c3c5df5ae3 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").toString();
     auto descriptor = collection->getIndexCatalog()->findIndexByName(
-        opCtx, _indexName, true /* includeUnfinishedIndexes */);
+        opCtx,
+        _indexName,
+        IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
 
     auto indexCatalogEntry = descriptor->getEntry();
 
@@ -284,14 +286,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 2754ac3e22f..58e48e08be5 100644
--- a/src/mongo/db/catalog/index_catalog.h
+++ b/src/mongo/db/catalog/index_catalog.h
@@ -197,6 +197,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;
 
@@ -237,9 +243,10 @@ public:
      *
      * @return null if cannot find
      */
-    virtual const IndexDescriptor* findIndexByName(OperationContext* opCtx,
-                                                   StringData name,
-                                                   bool includeUnfinishedIndexes = false) const = 0;
+    virtual const IndexDescriptor* findIndexByName(
+        OperationContext* opCtx,
+        StringData name,
+        InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const = 0;
 
     /**
      * Find index by matching key pattern and options. The key pattern, collation spec, and partial
@@ -251,7 +258,7 @@ public:
         OperationContext* opCtx,
         const BSONObj& key,
         const BSONObj& indexSpec,
-        bool includeUnfinishedIndexes = false) const = 0;
+        InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const = 0;
 
     /**
      * Find indexes with a matching key pattern, putting them into the vector 'matches'.  The key
@@ -261,12 +268,13 @@ public:
      */
     virtual void findIndexesByKeyPattern(OperationContext* opCtx,
                                          const BSONObj& key,
-                                         bool includeUnfinishedIndexes,
+                                         InclusionPolicy inclusionPolicy,
                                          std::vector<const IndexDescriptor*>* matches) const = 0;
-    virtual void findIndexByType(OperationContext* opCtx,
-                                 const std::string& type,
-                                 std::vector<const IndexDescriptor*>& matches,
-                                 bool includeUnfinishedIndexes = false) const = 0;
+    virtual void findIndexByType(
+        OperationContext* opCtx,
+        const std::string& type,
+        std::vector<const IndexDescriptor*>& matches,
+        InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const = 0;
 
     /**
      * Reload the index definition for 'oldDesc' from the CollectionCatalogEntry.  'oldDesc'
@@ -308,7 +316,7 @@ public:
      * Returns an iterator for the index descriptors in this IndexCatalog.
      */
     virtual std::unique_ptr<IndexIterator> getIndexIterator(
-        OperationContext* opCtx, bool includeUnfinishedIndexes) const = 0;
+        OperationContext* opCtx, InclusionPolicy inclusionPolicy) const = 0;
 
     // ---- index set modifiers ------
 
@@ -521,4 +529,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 fa77370bb96..9761f590c9d 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 b53aadbf696..cb47381ae39 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
@@ -124,7 +124,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);
 }
@@ -372,7 +372,8 @@ Status IndexCatalogEntryImpl::_setMultikeyInMultiDocumentTransaction(
 }
 
 std::shared_ptr<Ident> IndexCatalogEntryImpl::getSharedIdent() const {
-    return {shared_from_this(), _accessMethod->getIdentPtr()};  // aliasing constructor
+    return _accessMethod ? std::shared_ptr<Ident>{shared_from_this(), _accessMethod->getIdentPtr()}
+                         : nullptr;
 }
 
 // ----
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.h b/src/mongo/db/catalog/index_catalog_entry_impl.h
index 821176164e1..30ef5a80921 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.h
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.h
@@ -61,8 +61,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;
     }
@@ -80,6 +78,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 1c186a7e7ef..74049b34427 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -258,8 +258,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.
@@ -267,17 +267,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));
@@ -351,6 +362,7 @@ void IndexCatalogImpl::_logInternalState(OperationContext* opCtx,
                 "numIndexesInCollectionCatalogEntry"_attr = numIndexesInCollectionCatalogEntry,
                 "numReadyIndexes"_attr = _readyIndexes.size(),
                 "numBuildingIndexes"_attr = _buildingIndexes.size(),
+                "numFrozenIndexes"_attr = _frozenIndexes.size(),
                 "indexNamesToDrop"_attr = indexNamesToDrop);
 
     // Report the ready indexes.
@@ -426,7 +438,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;
     }
@@ -442,7 +455,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
@@ -471,8 +489,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;
         }
 
@@ -535,14 +556,16 @@ IndexCatalogEntry* IndexCatalogImpl::createIndexEntry(OperationContext* opCtx,
         engine->getEngine()->alterIdentMetadata(opCtx, ident, desc, isForceUpdateMetadata);
     }
 
-    std::unique_ptr<IndexAccessMethod> accessMethod = IndexAccessMethodFactory::get(opCtx)->make(
-        opCtx, collection->ns(), collection->getCollectionOptions(), entry.get(), ident);
-
-    entry->init(std::move(accessMethod));
+    if (!frozen) {
+        entry->setAccessMethod(IndexAccessMethodFactory::get(opCtx)->make(
+            opCtx, collection->ns(), collection->getCollectionOptions(), entry.get(), ident));
+    }
 
     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));
     }
@@ -1000,7 +1023,7 @@ Status IndexCatalogImpl::_isSpecOk(OperationContext* opCtx,
 Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
                                                        const CollectionPtr& collection,
                                                        const BSONObj& spec,
-                                                       const bool includeUnfinishedIndexes) const {
+                                                       InclusionPolicy inclusionPolicy) const {
     StringData name = spec.getStringField(IndexDescriptor::kIndexNameFieldName);
     invariant(name[0]);
 
@@ -1014,7 +1037,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.
@@ -1067,7 +1090,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,
@@ -1118,7 +1141,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, "
@@ -1159,7 +1182,10 @@ void IndexCatalogImpl::dropIndexes(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();
@@ -1174,7 +1200,11 @@ void IndexCatalogImpl::dropIndexes(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();
@@ -1300,25 +1330,26 @@ Status IndexCatalogImpl::dropIndexEntry(OperationContext* opCtx,
 
     audit::logDropIndex(opCtx->getClient(), indexName, collection->ns());
 
-    auto released = _readyIndexes.release(entry->descriptor());
-    if (released) {
-        invariant(released.get() == entry);
-        opCtx->recoveryUnit()->registerChange(
-            std::make_unique<IndexRemoveChange>(opCtx,
-                                                collection->ns(),
-                                                collection->uuid(),
-                                                std::move(released),
-                                                collection->getSharedDecorations()));
-    } else {
-        released = _buildingIndexes.release(entry->descriptor());
-        invariant(released.get() == entry);
-        opCtx->recoveryUnit()->registerChange(
-            std::make_unique<IndexRemoveChange>(opCtx,
-                                                collection->ns(),
-                                                collection->uuid(),
-                                                std::move(released),
-                                                collection->getSharedDecorations()));
-    }
+    auto released = [&] {
+        if (auto released = _readyIndexes.release(entry->descriptor())) {
+            return released;
+        }
+        if (auto released = _buildingIndexes.release(entry->descriptor())) {
+            return released;
+        }
+        if (auto released = _frozenIndexes.release(entry->descriptor())) {
+            return released;
+        }
+        MONGO_UNREACHABLE;
+    }();
+
+    invariant(released.get() == entry);
+    opCtx->recoveryUnit()->registerChange(
+        std::make_unique<IndexRemoveChange>(opCtx,
+                                            collection->ns(),
+                                            collection->uuid(),
+                                            std::move(released),
+                                            collection->getSharedDecorations()));
 
     CollectionQueryInfo::get(collection).rebuildIndexData(opCtx, collection);
     CollectionIndexUsageTrackerDecoration::get(collection->getSharedDecorations())
@@ -1338,7 +1369,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));
 }
 
@@ -1368,7 +1403,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());
     }
@@ -1380,7 +1415,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())
@@ -1391,8 +1426,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)
@@ -1405,8 +1440,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();
@@ -1420,10 +1455,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)) {
@@ -1435,8 +1470,8 @@ void IndexCatalogImpl::findIndexesByKeyPattern(OperationContext* opCtx,
 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) {
@@ -1677,7 +1712,10 @@ Status IndexCatalogImpl::indexRecords(OperationContext* opCtx,
 
     for (const MultikeyPathInfo& newPath : newPaths) {
         invariant(newPath.nss == coll->ns());
-        auto idx = findIndexByName(opCtx, newPath.indexName, /*includeUnfinishedIndexes=*/true);
+        auto idx = findIndexByName(opCtx,
+                                   newPath.indexName,
+                                   IndexCatalog::InclusionPolicy::kReady |
+                                       IndexCatalog::InclusionPolicy::kUnfinished);
         if (!idx) {
             return Status(ErrorCodes::IndexNotFound,
                           str::stream()
@@ -1780,7 +1818,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 259732da61c..f9baf272f11 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,
+        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;
+        InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const override;
 
     /**
      * Find indexes with a matching key pattern, putting them into the vector 'matches'.  The key
@@ -120,12 +121,12 @@ public:
      */
     void findIndexesByKeyPattern(OperationContext* opCtx,
                                  const BSONObj& key,
-                                 bool includeUnfinishedIndexes,
+                                 InclusionPolicy inclusionPolicy,
                                  std::vector<const IndexDescriptor*>* matches) const override;
     void findIndexByType(OperationContext* opCtx,
                          const std::string& type,
                          std::vector<const IndexDescriptor*>& matches,
-                         bool includeUnfinishedIndexes = false) const override;
+                         InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const override;
 
 
     /**
@@ -153,7 +154,7 @@ public:
 
     using IndexIterator = IndexCatalog::IndexIterator;
     std::unique_ptr<IndexIterator> getIndexIterator(OperationContext* opCtx,
-                                                    bool includeUnfinishedIndexes) const override;
+                                                    InclusionPolicy inclusionPolicy) const override;
 
     // ---- index set modifiers ------
 
@@ -377,7 +378,7 @@ private:
     Status _doesSpecConflictWithExisting(OperationContext* opCtx,
                                          const CollectionPtr& collection,
                                          const BSONObj& spec,
-                                         bool includeUnfinishedIndexes) const;
+                                         InclusionPolicy inclusionPolicy) const;
 
     /**
      * Returns true if the replica set member's config has {buildIndexes:false} set, which means
@@ -393,5 +394,6 @@ private:
 
     IndexCatalogEntryContainer _readyIndexes;
     IndexCatalogEntryContainer _buildingIndexes;
+    IndexCatalogEntryContainer _frozenIndexes;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/rename_collection.cpp b/src/mongo/db/catalog/rename_collection.cpp
index e3de3e5ddd5..1cc630d7458 100644
--- a/src/mongo/db/catalog/rename_collection.cpp
+++ b/src/mongo/db/catalog/rename_collection.cpp
@@ -617,7 +617,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 09b597752cf..09c8ec34ba3 100644
--- a/src/mongo/db/catalog/validate_state.cpp
+++ b/src/mongo/db/catalog/validate_state.cpp
@@ -242,8 +242,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/cqf/cqf_aggregate.cpp b/src/mongo/db/commands/cqf/cqf_aggregate.cpp
index 89d2045fec9..aabfc99c3a5 100644
--- a/src/mongo/db/commands/cqf/cqf_aggregate.cpp
+++ b/src/mongo/db/commands/cqf/cqf_aggregate.cpp
@@ -81,7 +81,8 @@ static opt::unordered_map<std::string, optimizer::IndexDefinition> buildIndexSpe
 
     const IndexCatalog& indexCatalog = *collection->getIndexCatalog();
     opt::unordered_map<std::string, IndexDefinition> result;
-    auto indexIterator = indexCatalog.getIndexIterator(opCtx, false /*includeUnfinished*/);
+    auto indexIterator =
+        indexCatalog.getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
 
     while (indexIterator->more()) {
         const IndexCatalogEntry& catalogEntry = *indexIterator->next();
diff --git a/src/mongo/db/commands/validate_db_metadata_cmd.cpp b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
index 0d057e498f5..6b0a494c3f6 100644
--- a/src/mongo/db/commands/validate_db_metadata_cmd.cpp
+++ b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
@@ -217,8 +217,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 d83af04df07..a4548ce4a4c 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 6f0adb10684..6149ecd913a 100644
--- a/src/mongo/db/index_builds_coordinator.cpp
+++ b/src/mongo/db/index_builds_coordinator.cpp
@@ -554,9 +554,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);
@@ -592,8 +591,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);
@@ -924,7 +926,6 @@ void IndexBuildsCoordinator::applyStartIndexBuild(OperationContext* opCtx,
 
             IndexCatalog* indexCatalog = coll.getWritableCollection(opCtx)->getIndexCatalog();
 
-            const bool includeUnfinished = false;
             for (const auto& spec : oplogEntry.indexSpecs) {
                 std::string name =
                     spec.getStringField(IndexDescriptor::kIndexNameFieldName).toString();
@@ -932,7 +933,8 @@ void IndexBuildsCoordinator::applyStartIndexBuild(OperationContext* opCtx,
                         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(opCtx), desc));
                 }
@@ -1119,7 +1121,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 43c97f7e7d2..b1d0090aa56 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
@@ -202,7 +202,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);
@@ -605,7 +606,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 d346265ff86..a0b4fe3cd36 100644
--- a/src/mongo/db/query/collection_query_info.cpp
+++ b/src/mongo/db/query/collection_query_info.cpp
@@ -92,12 +92,13 @@ CollectionQueryInfo::PlanCacheState::PlanCacheState(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 =
-        collection->getIndexCatalog()->getIndexIterator(opCtx, includeUnfinishedIndexes);
+    auto ii = collection->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));
+        }
     }
 
     planCacheIndexabilityState.updateDiscriminators(indexCores);
@@ -119,8 +120,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();
@@ -242,9 +243,8 @@ void CollectionQueryInfo::updatePlanCacheIndexEntries(OperationContext* opCtx,
 }
 
 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 6f6db41882c..35c83e8c5a5 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -284,7 +284,8 @@ void fillOutIndexEntries(OperationContext* opCtx,
                          const CollectionPtr& collection,
                          std::vector<IndexEntry>& entries,
                          std::vector<ColumnIndexEntry>& columnEntries) {
-    auto ii = collection->getIndexCatalog()->getIndexIterator(opCtx, false);
+    auto ii = collection->getIndexCatalog()->getIndexIterator(
+        opCtx, IndexCatalog::InclusionPolicy::kReady);
     while (ii->more()) {
         const IndexCatalogEntry* ice = ii->next();
 
@@ -2393,8 +2394,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/query/plan_cache_key_factory.cpp b/src/mongo/db/query/plan_cache_key_factory.cpp
index d47f1768858..6b154b29105 100644
--- a/src/mongo/db/query/plan_cache_key_factory.cpp
+++ b/src/mongo/db/query/plan_cache_key_factory.cpp
@@ -111,8 +111,8 @@ boost::optional<Timestamp> computeNewestVisibleIndexTimestamp(OperationContext*
 
     Timestamp currentNewestVisible = Timestamp::min();
 
-    std::unique_ptr<IndexCatalog::IndexIterator> ii =
-        collection->getIndexCatalog()->getIndexIterator(opCtx, /*includeUnfinishedIndexes*/ true);
+    auto ii = collection->getIndexCatalog()->getIndexIterator(
+        opCtx, IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
     while (ii->more()) {
         const IndexCatalogEntry* ice = ii->next();
         auto minVisibleSnapshot = ice->getMinimumVisibleSnapshot();
diff --git a/src/mongo/db/repl/rs_rollback.cpp b/src/mongo/db/repl/rs_rollback.cpp
index 7d254bc723d..e527aa204eb 100644
--- a/src/mongo/db/repl/rs_rollback.cpp
+++ b/src/mongo/db/repl/rs_rollback.cpp
@@ -862,9 +862,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/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index ca84c39b6f5..1a90e3a57c8 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -640,7 +640,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()
@@ -788,9 +790,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/repl/storage_timestamp_test.cpp b/src/mongo/db/repl/storage_timestamp_test.cpp
index fbaa81a4251..467fa99b314 100644
--- a/src/mongo/db/repl/storage_timestamp_test.cpp
+++ b/src/mongo/db/repl/storage_timestamp_test.cpp
@@ -2708,8 +2708,10 @@ TEST_F(StorageTimestampTest, IndexBuildsResolveErrorsDuringStateChangeToPrimary)
         }
 
         auto indexCatalog = collection->getIndexCatalog();
-        buildingIndex = indexCatalog->getEntry(
-            indexCatalog->findIndexByName(_opCtx, "a_1_b_1", /* includeUnfinished */ true));
+        buildingIndex = indexCatalog->getEntry(indexCatalog->findIndexByName(
+            _opCtx,
+            "a_1_b_1",
+            IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished));
         ASSERT(buildingIndex);
 
         ASSERT_OK(indexer.insertAllDocumentsInCollection(_opCtx, collection.get()));
diff --git a/src/mongo/db/s/shard_key_index_util.cpp b/src/mongo/db/s/shard_key_index_util.cpp
index 4097c9f236f..9b3b6371a4a 100644
--- a/src/mongo/db/s/shard_key_index_util.cpp
+++ b/src/mongo/db/s/shard_key_index_util.cpp
@@ -57,7 +57,8 @@ boost::optional<ShardKeyIndex> _findShardKeyPrefixedIndex(
 
     const IndexDescriptor* best = nullptr;
 
-    auto indexIterator = indexCatalog->getIndexIterator(opCtx, false);
+    auto indexIterator =
+        indexCatalog->getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
     while (indexIterator->more()) {
         auto indexEntry = indexIterator->next();
         auto indexDescriptor = indexEntry->descriptor();
diff --git a/src/mongo/db/stats/storage_stats.cpp b/src/mongo/db/stats/storage_stats.cpp
index 8e3252e61b8..48668dd07a1 100644
--- a/src/mongo/db/stats/storage_stats.cpp
+++ b/src/mongo/db/stats/storage_stats.cpp
@@ -148,8 +148,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_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index f866314229b..3465e9addf4 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -667,21 +667,19 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
                     logAttrs(nss));
             }
 
-            // 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 " << nss.ns());
+            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,
+                      logAttrs(nss));
+            }
 
             // 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
@@ -879,8 +877,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();
 
@@ -1306,7 +1306,9 @@ int64_t StorageEngineImpl::sizeOnDiskForDb(OperationContext* opCtx, const Databa
     auto perCollectionWork = [&](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 96d3c7d0af8..eaedd287615 100644
--- a/src/mongo/db/storage/storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/storage_engine_test_fixture.h
@@ -183,7 +183,10 @@ public:
         Collection* collection =
             CollectionCatalog::get(opCtx)->lookupCollectionByNamespaceForMetadataWrite(opCtx,
                                                                                        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 d9daf28b194..02084801dda 100644
--- a/src/mongo/db/system_index.cpp
+++ b/src/mongo/db/system_index.cpp
@@ -143,7 +143,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);
@@ -154,7 +155,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(
@@ -176,7 +178,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(