SERVER-40716 Refactor catalog initialization logic on startup

(cherry picked from commit ac88c6d4c085e72b6f0ae364b8d11dc604627efb)

4 files changed, 42 insertions, 26 deletions

diff --git a/src/mongo/db/storage/kv/kv_catalog.cpp b/src/mongo/db/storage/kv/kv_catalog.cpp
index d268f64394e..a102a8707d4 100644
--- a/src/mongo/db/storage/kv/kv_catalog.cpp
+++ b/src/mongo/db/storage/kv/kv_catalog.cpp
@@ -707,15 +707,13 @@ StatusWith<std::string> KVCatalog::newOrphanedIdent(OperationContext* opCtx, std
     return StatusWith<std::string>(std::move(ns));
 }
 
-void KVCatalog::initCollection(OperationContext* opCtx,
-                               const NamespaceString& nss,
-                               bool forRepair) {
+std::unique_ptr<CollectionCatalogEntry> KVCatalog::makeCollectionCatalogEntry(
+    OperationContext* opCtx, const NamespaceString& nss, bool forRepair) {
     BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, nss);
     uassert(ErrorCodes::MustDowngrade,
             str::stream() << "Collection does not have UUID in KVCatalog. Collection: " << nss,
             md.options.uuid);
 
-    auto uuid = md.options.uuid.get();
     auto ident = getCollectionIdent(nss);
 
     std::unique_ptr<RecordStore> rs;
@@ -729,16 +727,8 @@ void KVCatalog::initCollection(OperationContext* opCtx,
         invariant(rs);
     }
 
-    CollectionCatalog::get(getGlobalServiceContext())
-        .registerCatalogEntry(uuid,
-                              std::make_unique<KVCollectionCatalogEntry>(
-                                  _engine, this, nss.ns(), ident, std::move(rs)));
-}
-
-void KVCatalog::reinitCollectionAfterRepair(OperationContext* opCtx, const NamespaceString& nss) {
-    auto& catalog = CollectionCatalog::get(getGlobalServiceContext());
-    catalog.deregisterCatalogEntry(catalog.lookupUUIDByNSS(nss).get());
-    initCollection(opCtx, nss, false);
+    return std::make_unique<KVCollectionCatalogEntry>(
+        _engine, this, nss.ns(), ident, std::move(rs));
 }
 
 Status KVCatalog::createCollection(OperationContext* opCtx,
diff --git a/src/mongo/db/storage/kv/kv_catalog.h b/src/mongo/db/storage/kv/kv_catalog.h
index 23b5c3cb561..b22b6cf68ff 100644
--- a/src/mongo/db/storage/kv/kv_catalog.h
+++ b/src/mongo/db/storage/kv/kv_catalog.h
@@ -110,9 +110,9 @@ public:
      */
     std::string newInternalIdent();
 
-    void initCollection(OperationContext* opCtx, const NamespaceString& nss, bool forRepair);
-
-    void reinitCollectionAfterRepair(OperationContext* opCtx, const NamespaceString& nss);
+    std::unique_ptr<CollectionCatalogEntry> makeCollectionCatalogEntry(OperationContext* opCtx,
+                                                                       const NamespaceString& nss,
+                                                                       bool forRepair);
 
     Status createCollection(OperationContext* opCtx,
                             const NamespaceString& nss,
diff --git a/src/mongo/db/storage/kv/kv_storage_engine.cpp b/src/mongo/db/storage/kv/kv_storage_engine.cpp
index ae7a10f561c..d295631076b 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine.cpp
+++ b/src/mongo/db/storage/kv/kv_storage_engine.cpp
@@ -184,8 +184,7 @@ void KVStorageEngine::loadCatalog(OperationContext* opCtx) {
     }
 
     KVPrefix maxSeenPrefix = KVPrefix::kNotPrefixed;
-    for (const auto& coll : collectionsKnownToCatalog) {
-        NamespaceString nss(coll);
+    for (const auto& nss : collectionsKnownToCatalog) {
         std::string dbName = nss.db().toString();
 
         if (loadingFromUncleanShutdownOrRepair) {
@@ -193,7 +192,7 @@ void KVStorageEngine::loadCatalog(OperationContext* opCtx) {
             // possible that there are collections in the catalog that are unknown to the storage
             // engine. If we can't find a table in the list of storage engine idents, either
             // attempt to recover the ident or drop it.
-            const auto collectionIdent = _catalog->getCollectionIdent(coll);
+            const auto collectionIdent = _catalog->getCollectionIdent(nss);
             bool orphan = !std::binary_search(identsKnownToStorageEngine.begin(),
                                               identsKnownToStorageEngine.end(),
                                               collectionIdent);
@@ -203,14 +202,14 @@ void KVStorageEngine::loadCatalog(OperationContext* opCtx) {
             if (orphan) {
                 auto status = _recoverOrphanedCollection(opCtx, nss, collectionIdent);
                 if (!status.isOK()) {
-                    warning() << "Failed to recover orphaned data file for collection '" << coll
+                    warning() << "Failed to recover orphaned data file for collection '" << nss
                               << "': " << status;
                     WriteUnitOfWork wuow(opCtx);
-                    fassert(50716, _catalog->_removeEntry(opCtx, coll));
+                    fassert(50716, _catalog->_removeEntry(opCtx, nss));
 
                     if (_options.forRepair) {
                         StorageRepairObserver::get(getGlobalServiceContext())
-                            ->onModification(str::stream() << "Collection " << coll << " dropped: "
+                            ->onModification(str::stream() << "Collection " << nss << " dropped: "
                                                            << status.reason());
                     }
                     wuow.commit();
@@ -219,8 +218,8 @@ void KVStorageEngine::loadCatalog(OperationContext* opCtx) {
             }
         }
 
-        _catalog->initCollection(opCtx, coll, _options.forRepair);
-        auto maxPrefixForCollection = _catalog->getMetaData(opCtx, coll).getMaxPrefix();
+        _initCollection(opCtx, nss, _options.forRepair);
+        auto maxPrefixForCollection = _catalog->getMetaData(opCtx, nss).getMaxPrefix();
         maxSeenPrefix = std::max(maxSeenPrefix, maxPrefixForCollection);
 
         if (nss.isOrphanCollection()) {
@@ -236,6 +235,20 @@ void KVStorageEngine::loadCatalog(OperationContext* opCtx) {
     startingAfterUncleanShutdown(getGlobalServiceContext()) = false;
 }
 
+void KVStorageEngine::_initCollection(OperationContext* opCtx,
+                                      const NamespaceString& nss,
+                                      bool forRepair) {
+    auto catalogEntry = _catalog->makeCollectionCatalogEntry(opCtx, nss, forRepair);
+    auto uuid = catalogEntry->getCollectionOptions(opCtx).uuid.get();
+
+    auto collectionFactory = Collection::Factory::get(getGlobalServiceContext());
+    auto collection = collectionFactory->make(opCtx, catalogEntry.get());
+
+    auto& collectionCatalog = CollectionCatalog::get(getGlobalServiceContext());
+    collectionCatalog.registerCatalogEntry(uuid, std::move(catalogEntry));
+    collectionCatalog.registerCollectionObject(uuid, std::move(collection));
+}
+
 void KVStorageEngine::closeCatalog(OperationContext* opCtx) {
     dassert(opCtx->lockState()->isLocked());
     if (shouldLog(::mongo::logger::LogComponent::kStorageRecovery, kCatalogLogLevel)) {
@@ -651,8 +664,19 @@ Status KVStorageEngine::repairRecordStore(OperationContext* opCtx, const Namespa
         repairObserver->onModification(str::stream() << "Collection " << nss << ": "
                                                      << status.reason());
     }
-    _catalog->reinitCollectionAfterRepair(opCtx, nss);
 
+    auto& collectionCatalog = CollectionCatalog::get(getGlobalServiceContext());
+    auto uuid = collectionCatalog.lookupUUIDByNSS(nss).get();
+
+    // It's possible the Collection may not already have been removed if the no DatabaseHolder was
+    // opened for a database.
+    if (collectionCatalog.lookupCollectionByUUID(uuid)) {
+        collectionCatalog.deregisterCollectionObject(uuid);
+    }
+    collectionCatalog.deregisterCatalogEntry(uuid);
+
+    // After repairing, initialize the collection with a valid RecordStore.
+    _initCollection(opCtx, nss, false);
     return Status::OK();
 }
 
diff --git a/src/mongo/db/storage/kv/kv_storage_engine.h b/src/mongo/db/storage/kv/kv_storage_engine.h
index 70c4923b80a..263c5cdde9e 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine.h
+++ b/src/mongo/db/storage/kv/kv_storage_engine.h
@@ -348,6 +348,8 @@ public:
 private:
     using CollIter = std::list<std::string>::iterator;
 
+    void _initCollection(OperationContext* opCtx, const NamespaceString& nss, bool forRepair);
+
     Status _dropCollectionsNoTimestamp(OperationContext* opCtx,
                                        std::vector<NamespaceString>& toDrop);