SERVER-30081: Run storage recovery at startup.

9 files changed, 454 insertions, 42 deletions

diff --git a/src/mongo/db/db.cpp b/src/mongo/db/db.cpp
index 98e23a256ad..ef48122ce6f 100644
--- a/src/mongo/db/db.cpp
+++ b/src/mongo/db/db.cpp
@@ -326,12 +326,39 @@ void repairDatabasesAndCheckVersion(OperationContext* opCtx) {
     const repl::ReplSettings& replSettings = repl::getGlobalReplicationCoordinator()->getSettings();
 
     if (!storageGlobalParams.readOnly) {
-        // We open the "local" database before calling checkIfReplMissingFromCommandLine() to ensure
-        // the in-memory catalog entries for the 'kSystemReplSetCollection' collection have been
-        // populated if the collection exists. If the "local" database didn't exist at this point
-        // yet, then it will be created. If the mongod is running in a read-only mode, then it is
-        // fine to not open the "local" database and populate the catalog entries because we won't
-        // attempt to drop the temporary collections anyway.
+        StatusWith<std::vector<StorageEngine::CollectionIndexNamePair>> swIndexesToRebuild =
+            storageEngine->reconcileCatalogAndIdents(opCtx);
+        fassertStatusOK(40593, swIndexesToRebuild);
+        for (auto&& collIndexPair : swIndexesToRebuild.getValue()) {
+            const std::string& coll = collIndexPair.first;
+            const std::string& indexName = collIndexPair.second;
+            DatabaseCatalogEntry* dbce =
+                storageEngine->getDatabaseCatalogEntry(opCtx, NamespaceString(coll).db());
+            invariant(dbce);
+            CollectionCatalogEntry* cce = dbce->getCollectionCatalogEntry(coll);
+            invariant(cce);
+
+            StatusWith<IndexNameObjs> swIndexToRebuild(
+                getIndexNameObjs(opCtx, dbce, cce, [&indexName](const std::string& str) {
+                    return str == indexName;
+                }));
+            if (!swIndexToRebuild.isOK() || swIndexToRebuild.getValue().first.empty()) {
+                severe() << "Unable to get indexes for collection. Collection: " << coll;
+                fassertFailedNoTrace(40590);
+            }
+
+            invariant(swIndexToRebuild.getValue().first.size() == 1 &&
+                      swIndexToRebuild.getValue().second.size() == 1);
+            fassertStatusOK(
+                40592, rebuildIndexesOnCollection(opCtx, dbce, cce, swIndexToRebuild.getValue()));
+        }
+
+        // We open the "local" database before calling checkIfReplMissingFromCommandLine() to
+        // ensure the in-memory catalog entries for the 'kSystemReplSetCollection' collection have
+        // been populated if the collection exists. If the "local" database didn't exist at this
+        // point yet, then it will be created. If the mongod is running in a read-only mode, then
+        // it is fine to not open the "local" database and populate the catalog entries because we
+        // won't attempt to drop the temporary collections anyway.
         Lock::DBLock dbLock(opCtx, kSystemReplSetCollection.db(), MODE_X);
         dbHolder().openDb(opCtx, kSystemReplSetCollection.db());
     }
diff --git a/src/mongo/db/repair_database.cpp b/src/mongo/db/repair_database.cpp
index af2e4b1be3b..552f4577f04 100644
--- a/src/mongo/db/repair_database.cpp
+++ b/src/mongo/db/repair_database.cpp
@@ -30,11 +30,14 @@
 
 #include "mongo/platform/basic.h"
 
+#include <algorithm>
+
 #include "mongo/db/repair_database.h"
 
 #include "mongo/base/status.h"
 #include "mongo/base/string_data.h"
 #include "mongo/bson/bson_validate.h"
+#include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/background.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog_entry.h"
@@ -60,17 +63,22 @@ using std::string;
 
 using IndexVersion = IndexDescriptor::IndexVersion;
 
-namespace {
-Status rebuildIndexesOnCollection(OperationContext* opCtx,
-                                  DatabaseCatalogEntry* dbce,
-                                  const std::string& collectionName) {
-    CollectionCatalogEntry* cce = dbce->getCollectionCatalogEntry(collectionName);
-
-    std::vector<string> indexNames;
-    std::vector<BSONObj> indexSpecs;
+StatusWith<IndexNameObjs> getIndexNameObjs(OperationContext* opCtx,
+                                           DatabaseCatalogEntry* dbce,
+                                           CollectionCatalogEntry* cce,
+                                           stdx::function<bool(const std::string&)> filter) {
+    IndexNameObjs ret;
+    std::vector<string>& indexNames = ret.first;
+    std::vector<BSONObj>& indexSpecs = ret.second;
     {
         // Fetch all indexes
         cce->getAllIndexes(opCtx, &indexNames);
+        auto newEnd =
+            std::remove_if(indexNames.begin(),
+                           indexNames.end(),
+                           [&filter](const std::string& indexName) { return !filter(indexName); });
+        indexNames.erase(newEnd, indexNames.end());
+
         indexSpecs.reserve(indexNames.size());
 
         for (size_t i = 0; i < indexNames.size(); i++) {
@@ -118,6 +126,16 @@ Status rebuildIndexesOnCollection(OperationContext* opCtx,
         }
     }
 
+    return ret;
+}
+
+Status rebuildIndexesOnCollection(OperationContext* opCtx,
+                                  DatabaseCatalogEntry* dbce,
+                                  CollectionCatalogEntry* cce,
+                                  const IndexNameObjs& indexNameObjs) {
+    const std::vector<std::string>& indexNames = indexNameObjs.first;
+    const std::vector<BSONObj>& indexSpecs = indexNameObjs.second;
+
     // Skip the rest if there are no indexes to rebuild.
     if (indexSpecs.empty())
         return Status::OK();
@@ -211,7 +229,6 @@ Status rebuildIndexesOnCollection(OperationContext* opCtx,
 
     return Status::OK();
 }
-}  // namespace
 
 Status repairDatabase(OperationContext* opCtx,
                       StorageEngine* engine,
@@ -283,7 +300,12 @@ Status repairDatabase(OperationContext* opCtx,
         if (!status.isOK())
             return status;
 
-        status = rebuildIndexesOnCollection(opCtx, dbce, *it);
+        CollectionCatalogEntry* cce = dbce->getCollectionCatalogEntry(*it);
+        auto swIndexNameObjs = getIndexNameObjs(opCtx, dbce, cce);
+        if (!swIndexNameObjs.isOK())
+            return swIndexNameObjs.getStatus();
+
+        status = rebuildIndexesOnCollection(opCtx, dbce, cce, swIndexNameObjs.getValue());
         if (!status.isOK())
             return status;
 
diff --git a/src/mongo/db/repair_database.h b/src/mongo/db/repair_database.h
index b945b54c19f..1aa3d4bb911 100644
--- a/src/mongo/db/repair_database.h
+++ b/src/mongo/db/repair_database.h
@@ -30,12 +30,44 @@
 
 #include <string>
 
+#include "mongo/bson/bsonobj.h"
+#include "mongo/stdx/functional.h"
+
 namespace mongo {
+class CollectionCatalogEntry;
+class DatabaseCatalogEntry;
 class OperationContext;
 class Status;
 class StorageEngine;
 class StringData;
 
+typedef std::pair<std::vector<std::string>, std::vector<BSONObj>> IndexNameObjs;
+
+/**
+ * Returns a pair of parallel vectors. The first item is the index name. The second is the
+ * `BSONObj` "index spec" with an index name matching the `filter`.
+ *
+ * @param filter is a predicate that is passed in an index name, returning true if the index
+ *               should be included in the result.
+ */
+StatusWith<IndexNameObjs> getIndexNameObjs(OperationContext* opCtx,
+                                           DatabaseCatalogEntry* dbce,
+                                           CollectionCatalogEntry* cce,
+                                           stdx::function<bool(const std::string&)> filter =
+                                               [](const std::string& indexName) { return true; });
+
+/**
+ * Selectively rebuild some indexes on a collection. Indexes will be built in parallel with a
+ * `MultiIndexBlock`. One example usage is when a `dropIndex` command is rolled back. The dropped
+ * index must be remade.
+ *
+ * @param indexNameObjs is expected to be the result of a call to `getIndexNameObjs`.
+ */
+Status rebuildIndexesOnCollection(OperationContext* opCtx,
+                                  DatabaseCatalogEntry* dbce,
+                                  CollectionCatalogEntry* cce,
+                                  const IndexNameObjs& indexNameObjs);
+
 /**
  * Repairs a database using a storage engine-specific, best-effort process.
  * Some data may be lost or modified in the process but the output will
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.cpp
index f774aec7091..8b680fd6fc8 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.cpp
@@ -53,7 +53,9 @@ Status EphemeralForTestEngine::createRecordStore(OperationContext* opCtx,
                                                  StringData ns,
                                                  StringData ident,
                                                  const CollectionOptions& options) {
-    // All work done in getRecordStore
+    // Register the ident in the `_dataMap` (for `getAllIdents`). Remainder of work done in
+    // `getRecordStore`.
+    _dataMap[ident] = {};
     return Status::OK();
 }
 
@@ -75,7 +77,9 @@ std::unique_ptr<RecordStore> EphemeralForTestEngine::getRecordStore(
 Status EphemeralForTestEngine::createSortedDataInterface(OperationContext* opCtx,
                                                          StringData ident,
                                                          const IndexDescriptor* desc) {
-    // All work done in getSortedDataInterface
+    // Register the ident in `_dataMap` (for `getAllIdents`). Remainder of work done in
+    // `getSortedDataInterface`.
+    _dataMap[ident] = {};
     return Status::OK();
 }
 
diff --git a/src/mongo/db/storage/kv/SConscript b/src/mongo/db/storage/kv/SConscript
index d6f2a541df9..2d67b4b5bf3 100644
--- a/src/mongo/db/storage/kv/SConscript
+++ b/src/mongo/db/storage/kv/SConscript
@@ -126,3 +126,19 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store',
         ]
     )
+
+env.CppUnitTest(
+    target='kv_storage_engine_test',
+    source=[
+        'kv_storage_engine_test.cpp',
+    ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/db/catalog/collection_options',
+        '$BUILD_DIR/mongo/db/namespace_string',
+        '$BUILD_DIR/mongo/db/repl/replmocks',
+        '$BUILD_DIR/mongo/db/serveronly',
+        '$BUILD_DIR/mongo/db/service_context_d_test_fixture',
+        '$BUILD_DIR/mongo/db/storage/ephemeral_for_test/storage_ephemeral_for_test_core',
+        'kv_engine_mock',
+    ],
+)
diff --git a/src/mongo/db/storage/kv/kv_storage_engine.cpp b/src/mongo/db/storage/kv/kv_storage_engine.cpp
index 79bb56028a1..804acd9f101 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine.cpp
+++ b/src/mongo/db/storage/kv/kv_storage_engine.cpp
@@ -131,40 +131,102 @@ KVStorageEngine::KVStorageEngine(
 
     KVPrefix::setLargestPrefix(maxSeenPrefix);
     opCtx.recoveryUnit()->abandonSnapshot();
+}
 
-    // now clean up orphaned idents
-    // we don't do this in readOnly mode.
-    if (storageGlobalParams.readOnly) {
-        return;
+/**
+ * This method reconciles differences between idents the KVEngine is aware of and the
+ * KVCatalog. There are three differences to consider:
+ *
+ * First, a KVEngine may know of an ident that the KVCatalog does not. This method will drop
+ * the ident from the KVEngine.
+ *
+ * Second, a KVCatalog may have a collection ident that the KVEngine does not. This is an
+ * illegal state and this method fasserts.
+ *
+ * Third, a KVCatalog may have an index ident that the KVEngine does not. This method will
+ * rebuild the index.
+ */
+StatusWith<std::vector<StorageEngine::CollectionIndexNamePair>>
+KVStorageEngine::reconcileCatalogAndIdents(OperationContext* opCtx) {
+    // Gather all tables known to the storage engine and drop those that aren't cross-referenced
+    // in the _mdb_catalog. This can happen for two reasons.
+    //
+    // First, collection creation and deletion happen in two steps. First the storage engine
+    // creates/deletes the table, followed by the change to the _mdb_catalog. It's not assumed a
+    // storage engine can make these steps atomic.
+    //
+    // Second, a replica set node in 3.6+ on supported storage engines will only persist "stable"
+    // data to disk. That is data which replication guarantees won't be rolled back. The
+    // _mdb_catalog will reflect the "stable" set of collections/indexes. However, it's not
+    // expected for a storage engine's ability to persist stable data to extend to "stable
+    // tables".
+    std::set<std::string> engineIdents;
+    {
+        std::vector<std::string> vec = _engine->getAllIdents(opCtx);
+        engineIdents.insert(vec.begin(), vec.end());
+        engineIdents.erase(catalogInfo);
     }
+
+    std::set<std::string> catalogIdents;
     {
-        // get all idents
-        std::set<std::string> allIdents;
-        {
-            std::vector<std::string> v = _engine->getAllIdents(&opCtx);
-            allIdents.insert(v.begin(), v.end());
-            allIdents.erase(catalogInfo);
+        std::vector<std::string> vec = _catalog->getAllIdents(opCtx);
+        catalogIdents.insert(vec.begin(), vec.end());
+    }
+
+    // Drop all idents in the storage engine that are not known to the catalog. This can happen in
+    // the case of a collection or index creation being rolled back.
+    for (const auto& it : engineIdents) {
+        if (catalogIdents.find(it) != catalogIdents.end()) {
+            continue;
         }
 
-        // remove ones still in use
-        {
-            vector<string> idents = _catalog->getAllIdents(&opCtx);
-            for (size_t i = 0; i < idents.size(); i++) {
-                allIdents.erase(idents[i]);
-            }
+        if (!_catalog->isUserDataIdent(it)) {
+            continue;
+        }
+
+        const auto& toRemove = it;
+        log() << "Dropping unknown ident: " << toRemove;
+        WriteUnitOfWork wuow(opCtx);
+        fassertStatusOK(40591, _engine->dropIdent(opCtx, toRemove));
+        wuow.commit();
+    }
+
+    // Scan all collections in the catalog and make sure their ident is known to the storage
+    // engine. An omission here is fatal. A missing ident could mean a collection drop was rolled
+    // back. Note that startup already attempts to open tables; this should only catch errors in
+    // other contexts such as `recoverToStableTimestamp`.
+    std::vector<std::string> collections;
+    _catalog->getAllCollections(&collections);
+    for (const auto& coll : collections) {
+        const auto& identForColl = _catalog->getCollectionIdent(coll);
+        if (engineIdents.find(identForColl) == engineIdents.end()) {
+            return {ErrorCodes::UnrecoverableRollbackError,
+                    str::stream() << "Expected collection does not exist. NS: " << coll
+                                  << " Ident: "
+                                  << identForColl};
         }
+    }
 
-        for (std::set<std::string>::const_iterator it = allIdents.begin(); it != allIdents.end();
-             ++it) {
-            const std::string& toRemove = *it;
-            if (!_catalog->isUserDataIdent(toRemove))
+    // Scan all indexes and return those in the catalog where the storage engine does not have the
+    // corresponding ident. The caller is expected to rebuild these indexes.
+    std::vector<CollectionIndexNamePair> ret;
+    for (const auto& coll : collections) {
+        const BSONCollectionCatalogEntry::MetaData metaData = _catalog->getMetaData(opCtx, coll);
+        for (const auto& indexMetaData : metaData.indexes) {
+            const std::string& indexName = indexMetaData.name();
+            std::string indexIdent = _catalog->getIndexIdent(opCtx, coll, indexName);
+            if (engineIdents.find(indexIdent) != engineIdents.end()) {
                 continue;
-            log() << "dropping unused ident: " << toRemove;
-            WriteUnitOfWork wuow(&opCtx);
-            _engine->dropIdent(&opCtx, toRemove).transitional_ignore();
-            wuow.commit();
+            }
+
+            log() << "Expected index data is missing, rebuilding. NS: " << coll
+                  << " Index: " << indexName << " Ident: " << indexIdent;
+
+            ret.push_back(CollectionIndexNamePair(coll, indexName));
         }
     }
+
+    return ret;
 }
 
 void KVStorageEngine::cleanShutdown() {
diff --git a/src/mongo/db/storage/kv/kv_storage_engine.h b/src/mongo/db/storage/kv/kv_storage_engine.h
index 40a5908c186..af7c4f9fdaf 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine.h
+++ b/src/mongo/db/storage/kv/kv_storage_engine.h
@@ -31,6 +31,7 @@
 #include <map>
 #include <string>
 
+#include "mongo/base/status_with.h"
 #include "mongo/base/string_data.h"
 #include "mongo/db/storage/journal_listener.h"
 #include "mongo/db/storage/kv/kv_catalog.h"
@@ -133,6 +134,12 @@ public:
         return _catalog.get();
     }
 
+    /**
+     * Drop abandoned idents. Returns a parallel list of index name, index spec pairs to rebuild.
+     */
+    StatusWith<std::vector<StorageEngine::CollectionIndexNamePair>> reconcileCatalogAndIdents(
+        OperationContext* opCtx) override;
+
 private:
     class RemoveDBChange;
 
diff --git a/src/mongo/db/storage/kv/kv_storage_engine_test.cpp b/src/mongo/db/storage/kv/kv_storage_engine_test.cpp
new file mode 100644
index 00000000000..e3c814efcec
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_storage_engine_test.cpp
@@ -0,0 +1,230 @@
+/**
+ *    Copyright (C) 2017 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/base/status_with.h"
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/db/db_raii.h"
+#include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/index_names.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/operation_context_noop.h"
+#include "mongo/db/repair_database.h"
+#include "mongo/db/repl/repl_settings.h"
+#include "mongo/db/repl/replication_coordinator_global.h"
+#include "mongo/db/repl/replication_coordinator_mock.h"
+#include "mongo/db/service_context_d_test_fixture.h"
+#include "mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.h"
+#include "mongo/db/storage/kv/kv_database_catalog_entry.h"
+#include "mongo/db/storage/kv/kv_database_catalog_entry_mock.h"
+#include "mongo/db/storage/kv/kv_engine.h"
+#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+
+class KVStorageEngineTest : public unittest::Test {
+public:
+    KVStorageEngineTest()
+        : _storageEngine(stdx::make_unique<KVStorageEngine>(new EphemeralForTestEngine(),
+                                                            KVStorageEngineOptions())) {}
+
+    void setUp() final {
+        _serviceContext.setUp();
+    }
+
+    void tearDown() final {
+        _storageEngine->cleanShutdown();
+        _serviceContext.tearDown();
+    }
+
+    /**
+     * Create a collection in the catalog and in the KVEngine. Return the storage engine's `ident`.
+     */
+    StatusWith<std::string> createCollection(OperationContext* opCtx, NamespaceString ns) {
+        AutoGetDb db(opCtx, ns.db(), LockMode::MODE_X);
+        DatabaseCatalogEntry* dbce = _storageEngine->getDatabaseCatalogEntry(opCtx, ns.db());
+        auto ret = dbce->createCollection(opCtx, ns.ns(), CollectionOptions(), false);
+        if (!ret.isOK()) {
+            return ret;
+        }
+
+        return _storageEngine->getCatalog()->getCollectionIdent(ns.ns());
+    }
+
+
+    /**
+     * Create a collection table in the KVEngine not reflected in the KVCatalog.
+     */
+    Status createCollTable(OperationContext* opCtx, NamespaceString collName) {
+        const std::string identName = "collection-" + collName.ns();
+        return _storageEngine->getEngine()->createGroupedRecordStore(
+            opCtx, collName.ns(), identName, CollectionOptions(), KVPrefix::kNotPrefixed);
+    }
+
+    Status dropIndexTable(OperationContext* opCtx, NamespaceString nss, std::string indexName) {
+        std::string indexIdent =
+            _storageEngine->getCatalog()->getIndexIdent(opCtx, nss.ns(), indexName);
+        return dropIdent(opCtx, indexIdent);
+    }
+
+    Status dropIdent(OperationContext* opCtx, StringData ident) {
+        return _storageEngine->getEngine()->dropIdent(opCtx, ident);
+    }
+
+    StatusWith<std::vector<StorageEngine::CollectionIndexNamePair>> reconcile(
+        OperationContext* opCtx) {
+        return _storageEngine->reconcileCatalogAndIdents(opCtx);
+    }
+
+    std::vector<std::string> getAllKVEngineIdents(OperationContext* opCtx) {
+        return _storageEngine->getEngine()->getAllIdents(opCtx);
+    }
+
+    /**
+     * Create an index with a key of `{<key>: 1}` and a `name` of <key>.
+     */
+    Status createIndex(OperationContext* opCtx, NamespaceString collNs, std::string key) {
+        Collection* coll = nullptr;
+        BSONObjBuilder builder;
+        {
+            BSONObjBuilder keyObj;
+            builder.append("key", keyObj.append(key, 1).done());
+        }
+        BSONObj spec = builder.append("name", key).append("ns", collNs.ns()).append("v", 2).done();
+
+        auto descriptor =
+            stdx::make_unique<IndexDescriptor>(coll, IndexNames::findPluginName(spec), spec);
+
+        DatabaseCatalogEntry* dbce = _storageEngine->getDatabaseCatalogEntry(opCtx, collNs.db());
+        CollectionCatalogEntry* cce = dbce->getCollectionCatalogEntry(collNs.ns());
+        auto ret = cce->prepareForIndexBuild(opCtx, descriptor.get());
+        if (!ret.isOK()) {
+            return ret;
+        }
+
+        cce->indexBuildSuccess(opCtx, key);
+        return Status::OK();
+    }
+
+    ServiceContextMongoDTest _serviceContext;
+    std::unique_ptr<KVStorageEngine> _storageEngine;
+};
+
+TEST_F(KVStorageEngineTest, ReconcileIdentsTest) {
+    auto opCtx = cc().makeOperationContext();
+
+    // Add a collection, `db.coll1` to both the KVCatalog and KVEngine. The returned value is the
+    // `ident` name given to the collection.
+    auto swIdentName = createCollection(opCtx.get(), NamespaceString("db.coll1"));
+    ASSERT_OK(swIdentName);
+    // Create a table in the KVEngine not reflected in the KVCatalog. This should be dropped when
+    // reconciling.
+    ASSERT_OK(createCollTable(opCtx.get(), NamespaceString("db.coll2")));
+    ASSERT_OK(reconcile(opCtx.get()).getStatus());
+    auto identsVec = getAllKVEngineIdents(opCtx.get());
+    auto idents = std::set<std::string>(identsVec.begin(), identsVec.end());
+    // There are two idents. `_mdb_catalog` and the ident for `db.coll1`.
+    ASSERT_EQUALS(static_cast<const unsigned long>(2), idents.size());
+    ASSERT_TRUE(idents.find(swIdentName.getValue()) != idents.end());
+    ASSERT_TRUE(idents.find("_mdb_catalog") != idents.end());
+
+    // Create a catalog entry for the `_id` index. Drop the created the table.
+    ASSERT_OK(createIndex(opCtx.get(), NamespaceString("db.coll1"), "_id"));
+    ASSERT_OK(dropIndexTable(opCtx.get(), NamespaceString("db.coll1"), "_id"));
+    // The reconcile response should include this index as needing to be rebuilt.
+    auto reconcileStatus = reconcile(opCtx.get());
+    ASSERT_OK(reconcileStatus.getStatus());
+    ASSERT_EQUALS(static_cast<const unsigned long>(1), reconcileStatus.getValue().size());
+    StorageEngine::CollectionIndexNamePair& toRebuild = reconcileStatus.getValue()[0];
+    ASSERT_EQUALS("db.coll1", toRebuild.first);
+    ASSERT_EQUALS("_id", toRebuild.second);
+
+    // Now drop the `db.coll1` table, while leaving the KVCatalog entry.
+    ASSERT_OK(dropIdent(opCtx.get(), swIdentName.getValue()));
+    ASSERT_EQUALS(static_cast<const unsigned long>(1), getAllKVEngineIdents(opCtx.get()).size());
+
+    // Reconciling this should result in an error.
+    reconcileStatus = reconcile(opCtx.get());
+    ASSERT_NOT_OK(reconcileStatus.getStatus());
+    ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, reconcileStatus.getStatus());
+}
+
+TEST_F(KVStorageEngineTest, RecreateIndexes) {
+    repl::setGlobalReplicationCoordinator(
+        new repl::ReplicationCoordinatorMock(getGlobalServiceContext(), repl::ReplSettings()));
+
+    auto opCtx = cc().makeOperationContext();
+
+    // Create two indexes for `db.coll1` in the catalog named `foo` and `bar`. Verify the indexes
+    // appear as idents in the KVEngine.
+    ASSERT_OK(createCollection(opCtx.get(), NamespaceString("db.coll1")).getStatus());
+    ASSERT_OK(createIndex(opCtx.get(), NamespaceString("db.coll1"), "foo"));
+    ASSERT_OK(createIndex(opCtx.get(), NamespaceString("db.coll1"), "bar"));
+    auto kvIdents = getAllKVEngineIdents(opCtx.get());
+    ASSERT_EQUALS(2, std::count_if(kvIdents.begin(), kvIdents.end(), [](const std::string& str) {
+                      return str.find("index-") == 0;
+                  }));
+
+    // Use the `getIndexNameObjs` to find the `foo` index in the IndexCatalog.
+    DatabaseCatalogEntry* dbce = _storageEngine->getDatabaseCatalogEntry(opCtx.get(), "db");
+    CollectionCatalogEntry* cce = dbce->getCollectionCatalogEntry("db.coll1");
+    auto swIndexNameObjs = getIndexNameObjs(
+        opCtx.get(), dbce, cce, [](const std::string& indexName) { return indexName == "foo"; });
+    ASSERT_OK(swIndexNameObjs.getStatus());
+    auto& indexNameObjs = swIndexNameObjs.getValue();
+    // There's one index that matched the name `foo`.
+    ASSERT_EQUALS(static_cast<const unsigned long>(1), indexNameObjs.first.size());
+    // Assert the parallel vectors have matching sizes.
+    ASSERT_EQUALS(static_cast<const unsigned long>(1), indexNameObjs.second.size());
+    // The index that matched should be named `foo`.
+    ASSERT_EQUALS("foo", indexNameObjs.first[0]);
+    ASSERT_EQUALS("db.coll1"_sd, indexNameObjs.second[0].getStringField("ns"));
+    ASSERT_EQUALS("foo"_sd, indexNameObjs.second[0].getStringField("name"));
+    ASSERT_EQUALS(2, indexNameObjs.second[0].getIntField("v"));
+    ASSERT_EQUALS(1, indexNameObjs.second[0].getObjectField("key").getIntField("foo"));
+
+    // Drop the `foo` index table. Count one remaining index ident according to the KVEngine.
+    ASSERT_OK(dropIndexTable(opCtx.get(), NamespaceString("db.coll1"), "foo"));
+    kvIdents = getAllKVEngineIdents(opCtx.get());
+    ASSERT_EQUALS(1, std::count_if(kvIdents.begin(), kvIdents.end(), [](const std::string& str) {
+                      return str.find("index-") == 0;
+                  }));
+
+    AutoGetCollection coll(opCtx.get(), NamespaceString("db.coll1"), LockMode::MODE_X);
+    // Find the `foo` index in the catalog. Rebuild it. Count two indexes in the KVEngine.
+    ASSERT_OK(rebuildIndexesOnCollection(opCtx.get(), dbce, cce, indexNameObjs));
+    ASSERT_TRUE(cce->isIndexReady(opCtx.get(), "foo"));
+    kvIdents = getAllKVEngineIdents(opCtx.get());
+    ASSERT_EQUALS(2, std::count_if(kvIdents.begin(), kvIdents.end(), [](const std::string& str) {
+                      return str.find("index-") == 0;
+                  }));
+}
+}  // namespace mongo
diff --git a/src/mongo/db/storage/storage_engine.h b/src/mongo/db/storage/storage_engine.h
index 7e9d77641c3..cee183e9f8c 100644
--- a/src/mongo/db/storage/storage_engine.h
+++ b/src/mongo/db/storage/storage_engine.h
@@ -321,6 +321,18 @@ public:
      */
     virtual void setInitialDataTimestamp(SnapshotName snapshotName) {}
 
+    // (CollectionName, IndexName)
+    typedef std::pair<std::string, std::string> CollectionIndexNamePair;
+
+    /**
+     * Drop abandoned idents. In the successful case, returns a list of collection, index name
+     * pairs to rebuild.
+     */
+    virtual StatusWith<std::vector<CollectionIndexNamePair>> reconcileCatalogAndIdents(
+        OperationContext* opCtx) {
+        return std::vector<CollectionIndexNamePair>();
+    };
+
 protected:
     /**
      * The destructor will never be called. See cleanShutdown instead.