SERVER-41583 Refactor the registration and de-registration of collection and catalog entry

(cherry picked from commit 5eda33f9fa40a1a17f9f63f904a8c147700d648c)

16 files changed, 146 insertions, 345 deletions

diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 3d491c268b3..d34b865f076 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -43,29 +43,36 @@ namespace mongo {
 namespace {
 const ServiceContext::Decoration<CollectionCatalog> getCatalog =
     ServiceContext::declareDecoration<CollectionCatalog>();
-}  // namespace
 
-class CollectionCatalog::FinishDropChange : public RecoveryUnit::Change {
+class FinishDropCollectionChange : public RecoveryUnit::Change {
 public:
-    FinishDropChange(CollectionCatalog& catalog,
-                     std::unique_ptr<Collection> coll,
-                     CollectionUUID uuid)
-        : _catalog(catalog), _coll(std::move(coll)), _uuid(uuid) {}
+    FinishDropCollectionChange(CollectionCatalog* catalog,
+                               std::unique_ptr<Collection> coll,
+                               std::unique_ptr<CollectionCatalogEntry> catalogEntry,
+                               CollectionUUID uuid)
+        : _catalog(catalog),
+          _coll(std::move(coll)),
+          _catalogEntry(std::move(catalogEntry)),
+          _uuid(uuid) {}
 
     void commit(boost::optional<Timestamp>) override {
         _coll.reset();
+        _catalogEntry.reset();
     }
 
     void rollback() override {
-        _catalog.registerCollectionObject(_uuid, std::move(_coll));
+        _catalog->registerCollection(_uuid, std::move(_catalogEntry), std::move(_coll));
     }
 
 private:
-    CollectionCatalog& _catalog;
+    CollectionCatalog* _catalog;
     std::unique_ptr<Collection> _coll;
+    std::unique_ptr<CollectionCatalogEntry> _catalogEntry;
     CollectionUUID _uuid;
 };
 
+}  // namespace
+
 CollectionCatalog::iterator::iterator(StringData dbName,
                                       uint64_t genNum,
                                       const CollectionCatalog& catalog)
@@ -196,18 +203,6 @@ CollectionCatalog& CollectionCatalog::get(OperationContext* opCtx) {
     return getCatalog(opCtx->getServiceContext());
 }
 
-void CollectionCatalog::onCreateCollection(OperationContext* opCtx,
-                                           std::unique_ptr<Collection> coll,
-                                           CollectionUUID uuid) {
-    registerCollectionObject(uuid, std::move(coll));
-    opCtx->recoveryUnit()->onRollback([this, uuid] { deregisterCollectionObject(uuid); });
-}
-
-void CollectionCatalog::onDropCollection(OperationContext* opCtx, CollectionUUID uuid) {
-    auto coll = deregisterCollectionObject(uuid);
-    opCtx->recoveryUnit()->registerChange(new FinishDropChange(*this, std::move(coll), uuid));
-}
-
 void CollectionCatalog::setCollectionNamespace(OperationContext* opCtx,
                                                Collection* coll,
                                                const NamespaceString& fromCollection,
@@ -255,10 +250,6 @@ void CollectionCatalog::setCollectionNamespace(OperationContext* opCtx,
 
 void CollectionCatalog::onCloseDatabase(OperationContext* opCtx, std::string dbName) {
     invariant(opCtx->lockState()->isW());
-    for (auto it = begin(dbName); it != end(); ++it) {
-        deregisterCollectionObject(it.uuid().get());
-    }
-
     auto rid = ResourceId(RESOURCE_DATABASE, dbName);
     removeResource(rid, dbName);
 }
@@ -413,11 +404,14 @@ std::vector<std::string> CollectionCatalog::getAllDbNames() const {
     return ret;
 }
 
-void CollectionCatalog::registerCatalogEntry(
-    CollectionUUID uuid, std::unique_ptr<CollectionCatalogEntry> collectionCatalogEntry) {
+void CollectionCatalog::registerCollection(
+    CollectionUUID uuid,
+    std::unique_ptr<CollectionCatalogEntry> collectionCatalogEntry,
+    std::unique_ptr<Collection> coll) {
     stdx::lock_guard<stdx::mutex> lock(_catalogLock);
 
     LOG(0) << "Registering catalog entry " << collectionCatalogEntry->ns() << " with UUID " << uuid;
+    LOG(0) << "Registering collection object " << coll->ns() << " with UUID " << uuid;
 
     auto ns = collectionCatalogEntry->ns();
     auto dbName = ns.db().toString();
@@ -428,41 +422,12 @@ void CollectionCatalog::registerCatalogEntry(
     invariant(_collections.find(ns) == _collections.end());
     invariant(_orderedCollections.find(dbIdPair) == _orderedCollections.end());
 
-    CollectionInfo collectionInfo = {nullptr, /* std::unique_ptr<Collection> */
-                                     nullptr,
-                                     std::move(collectionCatalogEntry)};
+    auto collPtr = coll.get();
+    CollectionInfo collectionInfo = {std::move(coll), collPtr, std::move(collectionCatalogEntry)};
 
     _catalog[uuid] = std::move(collectionInfo);
     _collections[ns] = &_catalog[uuid];
     _orderedCollections[dbIdPair] = &_catalog[uuid];
-}
-
-void CollectionCatalog::registerCollectionObject(CollectionUUID uuid,
-                                                 std::unique_ptr<Collection> coll) {
-    stdx::lock_guard<stdx::mutex> lock(_catalogLock);
-
-    LOG(0) << "Registering collection object " << coll->ns() << " with UUID " << uuid;
-
-    auto ns = coll->ns();
-    auto dbName = ns.db().toString();
-    auto dbIdPair = std::make_pair(dbName, uuid);
-
-    // Make sure catalog entry associated with this uuid already exists.
-    invariant(_catalog.find(uuid) != _catalog.end());
-    invariant(_collections.find(ns) != _collections.end());
-    invariant(_orderedCollections.find(dbIdPair) != _orderedCollections.end());
-    invariant(_catalog[uuid].collectionCatalogEntry);
-    invariant(_collections[ns]->collectionCatalogEntry);
-    invariant(_orderedCollections[dbIdPair]->collectionCatalogEntry);
-
-    // Make sure collection object does not exist.
-    invariant(_catalog[uuid].collection == nullptr);
-    invariant(_collections[ns]->collection == nullptr);
-    invariant(_orderedCollections[dbIdPair]->collection == nullptr);
-
-
-    _catalog[uuid].collection = std::move(coll);
-    _catalog[uuid].collectionPtr = _catalog[uuid].collection.get();
 
     auto dbRid = ResourceId(RESOURCE_DATABASE, dbName);
     addResource(dbRid, dbName);
@@ -471,28 +436,30 @@ void CollectionCatalog::registerCollectionObject(CollectionUUID uuid,
     addResource(collRid, ns.ns());
 }
 
-std::unique_ptr<Collection> CollectionCatalog::deregisterCollectionObject(CollectionUUID uuid) {
+std::tuple<std::unique_ptr<Collection>, std::unique_ptr<CollectionCatalogEntry>>
+CollectionCatalog::deregisterCollection(CollectionUUID uuid) {
     stdx::lock_guard<stdx::mutex> lock(_catalogLock);
 
     invariant(_catalog.find(uuid) != _catalog.end());
     invariant(_catalog[uuid].collection);
+    invariant(_catalog[uuid].collectionCatalogEntry);
 
     auto coll = std::move(_catalog[uuid].collection);
+    auto catalogEntry = std::move(_catalog[uuid].collectionCatalogEntry);
     auto ns = coll->ns();
     auto dbName = ns.db().toString();
     auto dbIdPair = std::make_pair(dbName, uuid);
 
     LOG(0) << "Deregistering collection object " << ns << " with UUID " << uuid;
+    LOG(0) << "Deregistering catalog entry " << ns << " with UUID " << uuid;
 
     // Make sure collection object exists.
     invariant(_collections.find(ns) != _collections.end());
     invariant(_orderedCollections.find(dbIdPair) != _orderedCollections.end());
 
-    _catalog[uuid].collection = nullptr;
-    _catalog[uuid].collectionPtr = nullptr;
-
-    // Make sure collection catalog entry still exists.
-    invariant(_catalog[uuid].collectionCatalogEntry);
+    _orderedCollections.erase(dbIdPair);
+    _collections.erase(ns);
+    _catalog.erase(uuid);
 
     auto collRid = ResourceId(RESOURCE_COLLECTION, ns.ns());
     removeResource(collRid, ns.ns());
@@ -501,40 +468,14 @@ std::unique_ptr<Collection> CollectionCatalog::deregisterCollectionObject(Collec
     // references to the erased element.
     _generationNumber++;
 
-    return coll;
+    return std::make_tuple(std::move(coll), std::move(catalogEntry));
 }
 
-std::unique_ptr<CollectionCatalogEntry> CollectionCatalog::deregisterCatalogEntry(
+RecoveryUnit::Change* CollectionCatalog::makeFinishDropCollectionChange(
+    std::unique_ptr<Collection> coll,
+    std::unique_ptr<CollectionCatalogEntry> catalogEntry,
     CollectionUUID uuid) {
-    stdx::lock_guard<stdx::mutex> lock(_catalogLock);
-
-    invariant(_catalog.find(uuid) != _catalog.end());
-    invariant(_catalog[uuid].collectionCatalogEntry);
-
-    auto catalogEntry = std::move(_catalog[uuid].collectionCatalogEntry);
-    auto ns = catalogEntry->ns();
-    auto dbName = ns.db().toString();
-    auto dbIdPair = std::make_pair(dbName, uuid);
-
-    LOG(0) << "Deregistering catalog entry " << ns << " with UUID " << uuid;
-
-    // Make sure collection object is already gone.
-    invariant(_catalog[uuid].collection == nullptr);
-    invariant(_catalog[uuid].collectionPtr == nullptr);
-
-    // Make sure catalog entry exist.
-    invariant(_collections.find(ns) != _collections.end());
-    invariant(_orderedCollections.find(dbIdPair) != _orderedCollections.end());
-
-    _orderedCollections.erase(dbIdPair);
-    _collections.erase(ns);
-    _catalog.erase(uuid);
-
-    // Removal from an ordered map will invalidate iterators and potentially references to the
-    // references to the erased element.
-    _generationNumber++;
-
-    return catalogEntry;
+    return new FinishDropCollectionChange(this, std::move(coll), std::move(catalogEntry), uuid);
 }
 
 void CollectionCatalog::deregisterAllCatalogEntriesAndCollectionObjects() {
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index 475d12241fe..6ec515f19c7 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -106,20 +106,6 @@ public:
     CollectionCatalog() = default;
 
     /**
-     * This function inserts the entry for uuid, coll into the collection catalog. It is called by
-     * the op observer when a collection is created.
-     */
-    void onCreateCollection(OperationContext* opCtx,
-                            std::unique_ptr<Collection> coll,
-                            CollectionUUID uuid);
-
-    /**
-     * This function removes the entry for uuid from the collection catalog. It is called by the op
-     * observer when a collection is dropped.
-     */
-    void onDropCollection(OperationContext* opCtx, CollectionUUID uuid);
-
-    /**
      * This function is responsible for safely setting the namespace string inside 'coll' to the
      * value of 'toCollection'. The caller need not hold locks on the collection.
      *
@@ -131,36 +117,27 @@ public:
                                 const NamespaceString& fromCollection,
                                 const NamespaceString& toCollection);
 
-    /**
-     * Implies onDropCollection for all collections in db, but is not transactional.
-     */
     void onCloseDatabase(OperationContext* opCtx, std::string dbName);
 
     /**
-     * Register the collection catalog entry with `uuid`. The collection object with `uuid` must not
-     * exist in the CollectionCatalog yet.
-     */
-    void registerCatalogEntry(CollectionUUID uuid,
-                              std::unique_ptr<CollectionCatalogEntry> collectionCatalogEntry);
-
-    /**
-     * Deregister the collection catalog entry. The collection object with `uuid` is already gone,
-     * so this function completely removes any info about uuid.
+     * Register the collection object and collection catalog entry with `uuid`.
      */
-    std::unique_ptr<CollectionCatalogEntry> deregisterCatalogEntry(CollectionUUID uuid);
+    void registerCollection(CollectionUUID uuid,
+                            std::unique_ptr<CollectionCatalogEntry> collectionCatalogEntry,
+                            std::unique_ptr<Collection> collection);
 
     /**
-     * Register the collection object with `uuid`. The collection catalog entry with `uuid` already
-     * exists in the CollectionCatalog.
+     * Deregister the collection object and collection catalog entry.
      */
-    void registerCollectionObject(CollectionUUID uuid, std::unique_ptr<Collection> coll);
+    std::tuple<std::unique_ptr<Collection>, std::unique_ptr<CollectionCatalogEntry>>
+    deregisterCollection(CollectionUUID uuid);
 
     /**
-     * Deregister the collection object. The collection catalog entry still exists and will be
-     * deregistered later.
+     * Returns the RecoveryUnit's Change for dropping the collection
      */
-    std::unique_ptr<Collection> deregisterCollectionObject(CollectionUUID uuid);
-
+    RecoveryUnit::Change* makeFinishDropCollectionChange(std::unique_ptr<Collection>,
+                                                         std::unique_ptr<CollectionCatalogEntry>,
+                                                         CollectionUUID uuid);
 
     /**
      * Deregister all the collection objects and catalog entries.
@@ -290,7 +267,6 @@ public:
     void addResource(const ResourceId& rid, const std::string& entry);
 
 private:
-    class FinishDropChange;
     friend class CollectionCatalog::iterator;
 
     Collection* _lookupCollectionByUUID(WithLock, CollectionUUID uuid) const;
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index ea931c9acc9..b4629639497 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -66,8 +66,7 @@ public:
         auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
         col = collection.get();
         // Register dummy collection in catalog.
-        catalog.registerCatalogEntry(colUUID, std::move(catalogEntry));
-        catalog.onCreateCollection(&opCtx, std::move(collection), colUUID);
+        catalog.registerCollection(colUUID, std::move(catalogEntry), std::move(collection));
     }
 
 protected:
@@ -97,17 +96,15 @@ public:
 
             dbMap["foo"].insert(std::make_pair(fooUuid, fooColl.get()));
             dbMap["bar"].insert(std::make_pair(barUuid, barColl.get()));
-            catalog.registerCatalogEntry(fooUuid, std::move(fooCatalogEntry));
-            catalog.registerCatalogEntry(barUuid, std::move(barCatalogEntry));
-            catalog.onCreateCollection(&opCtx, std::move(fooColl), fooUuid);
-            catalog.onCreateCollection(&opCtx, std::move(barColl), barUuid);
+            catalog.registerCollection(fooUuid, std::move(fooCatalogEntry), std::move(fooColl));
+            catalog.registerCollection(barUuid, std::move(barCatalogEntry), std::move(barColl));
         }
     }
 
     void tearDown() {
         for (auto& it : dbMap) {
             for (auto& kv : it.second) {
-                catalog.onDropCollection(&opCtx, kv.first);
+                catalog.deregisterCollection(kv.first);
             }
         }
     }
@@ -278,8 +275,7 @@ public:
             auto newCatalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
             auto uuid = coll->uuid();
 
-            catalog.registerCatalogEntry(uuid.get(), std::move(newCatalogEntry));
-            catalog.onCreateCollection(&opCtx, std::move(coll), uuid.get());
+            catalog.registerCollection(uuid.get(), std::move(newCatalogEntry), std::move(coll));
         }
 
         int numEntries = 0;
@@ -302,8 +298,7 @@ public:
                 break;
             }
 
-            catalog.deregisterCollectionObject(uuid);
-            catalog.deregisterCatalogEntry(uuid);
+            catalog.deregisterCollection(uuid);
         }
 
         int numEntries = 0;
@@ -367,8 +362,7 @@ TEST_F(CollectionCatalogResourceTest, LookupMissingCollectionResource) {
 TEST_F(CollectionCatalogResourceTest, RemoveCollection) {
     const std::string collNs = "resourceDb.coll1";
     auto coll = catalog.lookupCollectionByNamespace(NamespaceString(collNs));
-    auto uniqueColl = catalog.deregisterCollectionObject(coll->uuid().get());
-    catalog.deregisterCatalogEntry(uniqueColl->uuid().get());
+    catalog.deregisterCollection(coll->uuid().get());
     auto rid = ResourceId(RESOURCE_COLLECTION, collNs);
     ASSERT(!catalog.lookupResourceName(rid));
 }
@@ -393,13 +387,13 @@ TEST_F(CollectionCatalogIterationTest, InvalidateEntry) {
     // Invalidate bar.coll1.
     for (auto collsIt = collsIterator("bar"); collsIt != collsIteratorEnd("bar"); ++collsIt) {
         if (collsIt->second->ns().ns() == "bar.coll1") {
-            catalog.onDropCollection(&opCtx, collsIt->first);
+            catalog.deregisterCollection(collsIt->first);
             dropColl("bar", collsIt->first);
             break;
         }
     }
 
-    // Ensure bar.coll1 is not returned by the iterator.
+
     for (; it != catalog.end(); ++it) {
         auto coll = *it;
         ASSERT(coll && coll->ns().ns() != "bar.coll1");
@@ -411,7 +405,7 @@ TEST_F(CollectionCatalogIterationTest, InvalidateAndDereference) {
     auto it = catalog.begin("bar");
     auto collsIt = collsIterator("bar");
     auto uuid = collsIt->first;
-    catalog.onDropCollection(&opCtx, uuid);
+    catalog.deregisterCollection(uuid);
     ++collsIt;
 
     ASSERT(it != catalog.end());
@@ -441,7 +435,7 @@ TEST_F(CollectionCatalogIterationTest, InvalidateLastEntryAndDereference) {
         }
     }
 
-    catalog.onDropCollection(&opCtx, *uuid);
+    catalog.deregisterCollection(*uuid);
     dropColl("bar", *uuid);
     ASSERT(*it == nullptr);
 }
@@ -465,60 +459,16 @@ TEST_F(CollectionCatalogIterationTest, InvalidateLastEntryInMapAndDereference) {
         }
     }
 
-    catalog.onDropCollection(&opCtx, *uuid);
+    catalog.deregisterCollection(*uuid);
     dropColl("foo", *uuid);
     ASSERT(*it == nullptr);
 }
 
-TEST_F(CollectionCatalogIterationTest, BeginSkipsOverEmptyCollectionObject) {
-    NamespaceString a1("a", "coll1");
-    NamespaceString a2("a", "coll2");
-
-    auto a1Uuid = CollectionUUID::gen();
-    auto a1CatalogEntry = std::make_unique<CollectionCatalogEntryMock>(a1.ns());
-
-    auto a2Uuid = CollectionUUID::gen();
-    if (a2Uuid < a1Uuid)
-        std::swap(a1Uuid, a2Uuid);
-    auto a2Coll = std::make_unique<CollectionMock>(a2);
-    auto a2CatalogEntry = std::make_unique<CollectionCatalogEntryMock>(a2.ns());
-
-    catalog.registerCatalogEntry(a1Uuid, std::move(a1CatalogEntry));
-    catalog.registerCatalogEntry(a2Uuid, std::move(a2CatalogEntry));
-    catalog.onCreateCollection(&opCtx, std::move(a2Coll), a2Uuid);
-
-    auto it = catalog.begin("a");
-    ASSERT(it != catalog.end());
-    auto coll = *it;
-    // Skips a.coll1 due to empty collection object.
-    ASSERT(coll->ns().ns() == a2.ns());
-}
-
-TEST_F(CollectionCatalogIterationTest, BeginSkipsOverEmptyCollectionObjectButStopsAtDbBoundary) {
-    NamespaceString a("a", "coll1");
-    NamespaceString b("b", "coll1");
-
-    auto aUuid = CollectionUUID::gen();
-    auto aCatalogEntry = std::make_unique<CollectionCatalogEntryMock>(a.ns());
-
-    auto bUuid = CollectionUUID::gen();
-
-    auto bColl = std::make_unique<CollectionMock>(b);
-    auto bCatalogEntry = std::make_unique<CollectionCatalogEntryMock>(b.ns());
-
-    catalog.registerCatalogEntry(aUuid, std::move(aCatalogEntry));
-    catalog.registerCatalogEntry(bUuid, std::move(bCatalogEntry));
-    catalog.onCreateCollection(&opCtx, std::move(bColl), bUuid);
-
-    auto it = catalog.begin("a");
-    ASSERT(it == catalog.end());
-}
-
 TEST_F(CollectionCatalogIterationTest, GetUUIDWontRepositionEvenIfEntryIsDropped) {
     auto it = catalog.begin("bar");
     auto collsIt = collsIterator("bar");
     auto uuid = collsIt->first;
-    catalog.onDropCollection(&opCtx, uuid);
+    catalog.deregisterCollection(uuid);
     dropColl("bar", uuid);
 
     ASSERT_EQUALS(uuid, it.uuid());
@@ -553,14 +503,13 @@ TEST_F(CollectionCatalogTest, InsertAfterLookup) {
     // Ensure that looking up non-existing UUIDs doesn't affect later registration of those UUIDs.
     ASSERT(catalog.lookupCollectionByUUID(newUUID) == nullptr);
     ASSERT_EQUALS(catalog.lookupNSSByUUID(newUUID), boost::none);
-    catalog.registerCatalogEntry(newUUID, std::move(newCatalogEntry));
-    catalog.onCreateCollection(&opCtx, std::move(newCollUnique), newUUID);
+    catalog.registerCollection(newUUID, std::move(newCatalogEntry), std::move(newCollUnique));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(newUUID), newCol);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(colUUID), nss);
 }
 
 TEST_F(CollectionCatalogTest, OnDropCollection) {
-    catalog.onDropCollection(&opCtx, colUUID);
+    catalog.deregisterCollection(colUUID);
     // Ensure the lookup returns a null pointer upon removing the colUUID entry.
     ASSERT(catalog.lookupCollectionByUUID(colUUID) == nullptr);
 }
@@ -571,8 +520,7 @@ TEST_F(CollectionCatalogTest, RenameCollection) {
     auto collUnique = std::make_unique<CollectionMock>(oldNss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(oldNss.ns());
     auto collection = collUnique.get();
-    catalog.registerCatalogEntry(uuid, std::move(catalogEntry));
-    catalog.onCreateCollection(&opCtx, std::move(collUnique), uuid);
+    catalog.registerCollection(uuid, std::move(catalogEntry), std::move(collUnique));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(uuid), collection);
 
     NamespaceString newNss(nss.db(), "newcol");
@@ -583,8 +531,7 @@ TEST_F(CollectionCatalogTest, RenameCollection) {
 
 TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsOldNSSIfDropped) {
     catalog.onCloseCatalog(&opCtx);
-    catalog.onDropCollection(&opCtx, colUUID);
-    catalog.deregisterCatalogEntry(colUUID);
+    catalog.deregisterCollection(colUUID);
     ASSERT(catalog.lookupCollectionByUUID(colUUID) == nullptr);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(colUUID), nss);
     catalog.onOpenCatalog(&opCtx);
@@ -602,8 +549,7 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsNewlyCreated
     catalog.onCloseCatalog(&opCtx);
     ASSERT(catalog.lookupCollectionByUUID(newUUID) == nullptr);
     ASSERT_EQUALS(catalog.lookupNSSByUUID(newUUID), boost::none);
-    catalog.registerCatalogEntry(newUUID, std::move(newCatalogEntry));
-    catalog.onCreateCollection(&opCtx, std::move(newCollUnique), newUUID);
+    catalog.registerCollection(newUUID, std::move(newCatalogEntry), std::move(newCollUnique));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(newUUID), newCol);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(colUUID), nss);
 
@@ -620,12 +566,10 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsFreshestNSS)
     auto newCol = newCollUnique.get();
 
     catalog.onCloseCatalog(&opCtx);
-    catalog.onDropCollection(&opCtx, colUUID);
-    catalog.deregisterCatalogEntry(colUUID);
+    catalog.deregisterCollection(colUUID);
     ASSERT(catalog.lookupCollectionByUUID(colUUID) == nullptr);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(colUUID), nss);
-    catalog.registerCatalogEntry(colUUID, std::move(newCatalogEntry));
-    catalog.onCreateCollection(&opCtx, std::move(newCollUnique), colUUID);
+    catalog.registerCollection(colUUID, std::move(newCatalogEntry), std::move(newCollUnique));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(colUUID), newCol);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(colUUID), newNss);
 
@@ -654,8 +598,7 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNames) {
         auto newColl = std::make_unique<CollectionMock>(nss);
         auto newCatalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
         auto uuid = CollectionUUID::gen();
-        catalog.registerCatalogEntry(uuid, std::move(newCatalogEntry));
-        catalog.registerCollectionObject(uuid, std::move(newColl));
+        catalog.registerCollection(uuid, std::move(newCatalogEntry), std::move(newColl));
     }
 
     std::vector<NamespaceString> dCollList = {d1Coll, d2Coll, d3Coll};
diff --git a/src/mongo/db/catalog/database.h b/src/mongo/db/catalog/database.h
index 83402b8650e..ed38c42ba52 100644
--- a/src/mongo/db/catalog/database.h
+++ b/src/mongo/db/catalog/database.h
@@ -82,9 +82,6 @@ public:
      */
     virtual void init(OperationContext* opCtx) const = 0;
 
-    // closes files and other cleanup see below.
-    virtual void close(OperationContext* const opCtx) const = 0;
-
     virtual const std::string& name() const = 0;
 
     virtual void clearTmpCollections(OperationContext* const opCtx) const = 0;
diff --git a/src/mongo/db/catalog/database_holder_impl.cpp b/src/mongo/db/catalog/database_holder_impl.cpp
index b1dc0d6e2c4..7f7c570d6c0 100644
--- a/src/mongo/db/catalog/database_holder_impl.cpp
+++ b/src/mongo/db/catalog/database_holder_impl.cpp
@@ -40,7 +40,6 @@
 #include "mongo/db/catalog/database_impl.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/operation_context.h"
-#include "mongo/db/repl/oplog.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/top.h"
 #include "mongo/db/storage/storage_engine.h"
@@ -213,10 +212,8 @@ void DatabaseHolderImpl::close(OperationContext* opCtx, StringData ns) {
     }
 
     auto db = it->second;
-    repl::oplogCheckCloseDatabase(opCtx, db);
     CollectionCatalog::get(opCtx).onCloseDatabase(opCtx, dbName.toString());
 
-    db->close(opCtx);
     delete db;
     db = nullptr;
 
@@ -254,9 +251,7 @@ void DatabaseHolderImpl::closeAll(OperationContext* opCtx) {
         LOG(2) << "DatabaseHolder::closeAll name:" << name;
 
         Database* db = _dbs[name];
-        repl::oplogCheckCloseDatabase(opCtx, db);
         CollectionCatalog::get(opCtx).onCloseDatabase(opCtx, name);
-        db->close(opCtx);
         delete db;
 
         _dbs.erase(name);
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index fc3f00c1511..8c853465d80 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -105,13 +105,6 @@ void uassertNamespaceNotIndex(StringData ns, StringData caller) {
             NamespaceString::normal(ns));
 }
 
-void DatabaseImpl::close(OperationContext* opCtx) const {
-    invariant(opCtx->lockState()->isW());
-
-    // Clear cache of oplog Collection pointer.
-    repl::oplogCheckCloseDatabase(opCtx, this);
-}
-
 Status DatabaseImpl::validateDBName(StringData dbname) {
     if (dbname.size() <= 0)
         return Status(ErrorCodes::BadValue, "db name is empty");
@@ -482,12 +475,19 @@ Status DatabaseImpl::_finishDropCollection(OperationContext* opCtx,
     UUID uuid = *collection->uuid();
     log() << "Finishing collection drop for " << nss << " (" << uuid << ").";
 
-    CollectionCatalog& catalog = CollectionCatalog::get(opCtx);
-    catalog.onDropCollection(opCtx, uuid);
-
     auto storageEngine =
         checked_cast<KVStorageEngine*>(opCtx->getServiceContext()->getStorageEngine());
-    return storageEngine->getCatalog()->dropCollection(opCtx, nss);
+    auto status = storageEngine->getCatalog()->dropCollection(opCtx, nss);
+    if (!status.isOK())
+        return status;
+
+    auto[removedColl, removedCatalogEntry] =
+        CollectionCatalog::get(opCtx).deregisterCollection(uuid);
+    opCtx->recoveryUnit()->registerChange(
+        CollectionCatalog::get(opCtx).makeFinishDropCollectionChange(
+            std::move(removedColl), std::move(removedCatalogEntry), uuid));
+
+    return Status::OK();
 }
 
 Collection* DatabaseImpl::getCollection(OperationContext* opCtx, const NamespaceString& nss) const {
@@ -666,22 +666,32 @@ Collection* DatabaseImpl::createCollection(OperationContext* opCtx,
     // Create CollectionCatalogEntry
     auto storageEngine =
         checked_cast<KVStorageEngine*>(opCtx->getServiceContext()->getStorageEngine());
-    massertStatusOK(storageEngine->getCatalog()->createCollection(
-        opCtx, nss, optionsWithUUID, true /*allocateDefaultSpace*/));
+    auto statusWithCatalogEntry = storageEngine->getCatalog()->createCollection(
+        opCtx, nss, optionsWithUUID, true /*allocateDefaultSpace*/);
+    massertStatusOK(statusWithCatalogEntry.getStatus());
+    std::unique_ptr<CollectionCatalogEntry> ownedCatalogEntry =
+        std::move(statusWithCatalogEntry.getValue());
 
     // Create Collection object
-    auto& catalog = CollectionCatalog::get(opCtx);
-    auto catalogEntry = catalog.lookupCollectionCatalogEntryByUUID(optionsWithUUID.uuid.get());
-    auto ownedCollection = Collection::Factory::get(opCtx)->make(opCtx, catalogEntry);
+    std::unique_ptr<Collection> ownedCollection =
+        Collection::Factory::get(opCtx)->make(opCtx, ownedCatalogEntry.get());
+    auto collection = ownedCollection.get();
     ownedCollection->init(opCtx);
-    Collection* collection = ownedCollection.get();
-    catalog.onCreateCollection(opCtx, std::move(ownedCollection), *(collection->uuid()));
+
     opCtx->recoveryUnit()->onCommit([collection](auto commitTime) {
         // Ban reading from this collection on committed reads on snapshots before now.
         if (commitTime)
             collection->setMinimumVisibleSnapshot(commitTime.get());
     });
 
+    auto& catalog = CollectionCatalog::get(opCtx);
+    auto uuid = ownedCollection->uuid().get();
+    catalog.registerCollection(uuid, std::move(ownedCatalogEntry), std::move(ownedCollection));
+    opCtx->recoveryUnit()->onRollback([uuid, &catalog] {
+        auto[removedColl, removedCatalogEntry] = catalog.deregisterCollection(uuid);
+        removedColl.reset();
+        removedCatalogEntry.reset();
+    });
 
     BSONObj fullIdIndexSpec;
 
diff --git a/src/mongo/db/catalog/database_impl.h b/src/mongo/db/catalog/database_impl.h
index ef90e9cf08f..da4b567e91f 100644
--- a/src/mongo/db/catalog/database_impl.h
+++ b/src/mongo/db/catalog/database_impl.h
@@ -41,10 +41,6 @@ public:
 
     void init(OperationContext*) const final;
 
-
-    // closes files and other cleanup see below.
-    void close(OperationContext* opCtx) const final;
-
     const std::string& name() const final {
         return _name;
     }
diff --git a/src/mongo/db/pipeline/document_source_change_stream_test.cpp b/src/mongo/db/pipeline/document_source_change_stream_test.cpp
index 016d7132283..2bf1da5aed6 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_test.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream_test.cpp
@@ -418,9 +418,7 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAtOperationTimeAndResumeAfter
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCatalogEntry(testUuid(), std::move(catalogEntry));
-    CollectionCatalog::get(expCtx->opCtx)
-        .onCreateCollection(expCtx->opCtx, std::move(collection), testUuid());
+        .registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -443,8 +441,7 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAfterAndResumeAfterOptions) {
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     auto& catalog = CollectionCatalog::get(opCtx);
-    catalog.registerCatalogEntry(testUuid(), std::move(catalogEntry));
-    catalog.onCreateCollection(expCtx->opCtx, std::move(collection), testUuid());
+    catalog.registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -467,8 +464,7 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAtOperationTimeAndStartAfterO
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     auto& catalog = CollectionCatalog::get(opCtx);
-    catalog.registerCatalogEntry(testUuid(), std::move(catalogEntry));
-    catalog.onCreateCollection(expCtx->opCtx, std::move(collection), testUuid());
+    catalog.registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -491,8 +487,7 @@ TEST_F(ChangeStreamStageTest, ShouldRejectResumeAfterWithResumeTokenMissingUUID)
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     auto& catalog = CollectionCatalog::get(opCtx);
-    catalog.registerCatalogEntry(testUuid(), std::move(catalogEntry));
-    catalog.onCreateCollection(expCtx->opCtx, std::move(collection), testUuid());
+    catalog.registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -1520,9 +1515,8 @@ TEST_F(ChangeStreamStageTest, DocumentKeyShouldIncludeShardKeyFromResumeToken) {
 
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCatalogEntry(uuid, std::move(catalogEntry));
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .onCreateCollection(getExpCtx()->opCtx, std::move(collection), uuid);
+        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -1568,9 +1562,8 @@ TEST_F(ChangeStreamStageTest, DocumentKeyShouldNotIncludeShardKeyFieldsIfNotPres
 
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCatalogEntry(uuid, std::move(catalogEntry));
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .onCreateCollection(getExpCtx()->opCtx, std::move(collection), uuid);
+        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -1613,9 +1606,8 @@ TEST_F(ChangeStreamStageTest, ResumeAfterFailsIfResumeTokenDoesNotContainUUID) {
 
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCatalogEntry(uuid, std::move(catalogEntry));
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .onCreateCollection(getExpCtx()->opCtx, std::move(collection), uuid);
+        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
 
     // Create a resume token from only the timestamp.
     auto resumeToken = makeResumeToken(ts);
@@ -1670,9 +1662,7 @@ TEST_F(ChangeStreamStageTest, ResumeAfterWithTokenFromInvalidateShouldFail) {
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCatalogEntry(testUuid(), std::move(catalogEntry));
-    CollectionCatalog::get(expCtx->opCtx)
-        .onCreateCollection(expCtx->opCtx, std::move(collection), testUuid());
+        .registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
 
     const auto resumeTokenInvalidate =
         makeResumeToken(kDefaultTs,
@@ -2108,9 +2098,8 @@ TEST_F(ChangeStreamStageDBTest, DocumentKeyShouldIncludeShardKeyFromResumeToken)
 
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCatalogEntry(uuid, std::move(catalogEntry));
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .onCreateCollection(getExpCtx()->opCtx, std::move(collection), uuid);
+        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -2147,9 +2136,8 @@ TEST_F(ChangeStreamStageDBTest, DocumentKeyShouldNotIncludeShardKeyFieldsIfNotPr
 
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCatalogEntry(uuid, std::move(catalogEntry));
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .onCreateCollection(getExpCtx()->opCtx, std::move(collection), uuid);
+        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -2187,9 +2175,8 @@ TEST_F(ChangeStreamStageDBTest, DocumentKeyShouldNotIncludeShardKeyIfResumeToken
 
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCatalogEntry(uuid, std::move(catalogEntry));
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .onCreateCollection(getExpCtx()->opCtx, std::move(collection), uuid);
+        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
 
     // Create a resume token from only the timestamp.
     auto resumeToken = makeResumeToken(ts);
@@ -2227,9 +2214,7 @@ TEST_F(ChangeStreamStageDBTest, ResumeAfterWithTokenFromInvalidateShouldFail) {
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .registerCatalogEntry(testUuid(), std::move(catalogEntry));
-    CollectionCatalog::get(expCtx->opCtx)
-        .onCreateCollection(expCtx->opCtx, std::move(collection), testUuid());
+        .registerCollection(testUuid(), std::move(catalogEntry), std::move(collection));
 
     const auto resumeTokenInvalidate =
         makeResumeToken(kDefaultTs,
@@ -2251,9 +2236,8 @@ TEST_F(ChangeStreamStageDBTest, ResumeAfterWithTokenFromDropDatabase) {
 
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCatalogEntry(uuid, std::move(catalogEntry));
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .onCreateCollection(getExpCtx()->opCtx, std::move(collection), uuid);
+        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
 
     // Create a resume token from only the timestamp, similar to a 'dropDatabase' entry.
     auto resumeToken = makeResumeToken(
@@ -2283,9 +2267,8 @@ TEST_F(ChangeStreamStageDBTest, StartAfterSucceedsEvenIfResumeTokenDoesNotContai
 
     auto collection = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCatalogEntry(uuid, std::move(catalogEntry));
     CollectionCatalog::get(getExpCtx()->opCtx)
-        .onCreateCollection(getExpCtx()->opCtx, std::move(collection), uuid);
+        .registerCollection(uuid, std::move(catalogEntry), std::move(collection));
 
     // Create a resume token from only the timestamp, similar to a 'dropDatabase' entry.
     auto resumeToken = makeResumeToken(kDefaultTs);
diff --git a/src/mongo/db/query/query_request_test.cpp b/src/mongo/db/query/query_request_test.cpp
index 0870bf245da..1ad7544a371 100644
--- a/src/mongo/db/query/query_request_test.cpp
+++ b/src/mongo/db/query/query_request_test.cpp
@@ -1480,8 +1480,7 @@ TEST_F(QueryRequestTest, ParseFromUUID) {
     auto coll = std::make_unique<CollectionMock>(nss);
     auto catalogEntry = std::make_unique<CollectionCatalogEntryMock>(nss.ns());
     CollectionCatalog& catalog = CollectionCatalog::get(opCtx.get());
-    catalog.registerCatalogEntry(uuid, std::move(catalogEntry));
-    catalog.onCreateCollection(opCtx.get(), std::move(coll), uuid);
+    catalog.registerCollection(uuid, std::move(catalogEntry), std::move(coll));
     QueryRequest qr(NamespaceStringOrUUID("test", uuid));
     // Ensure a call to refreshNSS succeeds.
     qr.refreshNSS(opCtx.get());
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 658353d6cad..da7a63b1454 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -1990,13 +1990,6 @@ void initTimestampFromOplog(OperationContext* opCtx, const NamespaceString& oplo
     }
 }
 
-void oplogCheckCloseDatabase(OperationContext* opCtx, const Database* db) {
-    invariant(opCtx->lockState()->isW());
-    if (db->name() == "local") {
-        LocalOplogInfo::get(opCtx)->resetCollection();
-    }
-}
-
 void clearLocalOplogPtr() {
     LocalOplogInfo::get(getGlobalServiceContext())->resetCollection();
 }
diff --git a/src/mongo/db/repl/oplog.h b/src/mongo/db/repl/oplog.h
index b8729038179..155272bb417 100644
--- a/src/mongo/db/repl/oplog.h
+++ b/src/mongo/db/repl/oplog.h
@@ -145,8 +145,6 @@ OpTime logOp(OperationContext* opCtx,
              const OplogSlot& oplogSlot);
 
 // Flush out the cached pointer to the oplog.
-// Used by the closeDatabase command to ensure we don't cache closed things.
-void oplogCheckCloseDatabase(OperationContext* opCtx, const Database* db);
 void clearLocalOplogPtr();
 
 /**
diff --git a/src/mongo/db/storage/kv/kv_catalog.cpp b/src/mongo/db/storage/kv/kv_catalog.cpp
index a102a8707d4..bcdec978c0e 100644
--- a/src/mongo/db/storage/kv/kv_catalog.cpp
+++ b/src/mongo/db/storage/kv/kv_catalog.cpp
@@ -35,7 +35,6 @@
 
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/bson/util/builder.h"
-#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
@@ -731,10 +730,11 @@ std::unique_ptr<CollectionCatalogEntry> KVCatalog::makeCollectionCatalogEntry(
         _engine, this, nss.ns(), ident, std::move(rs));
 }
 
-Status KVCatalog::createCollection(OperationContext* opCtx,
-                                   const NamespaceString& nss,
-                                   const CollectionOptions& options,
-                                   bool allocateDefaultSpace) {
+StatusWith<std::unique_ptr<CollectionCatalogEntry>> KVCatalog::createCollection(
+    OperationContext* opCtx,
+    const NamespaceString& nss,
+    const CollectionOptions& options,
+    bool allocateDefaultSpace) {
     invariant(opCtx->lockState()->isDbLockedForMode(nss.db(), MODE_IX));
     invariant(nss.coll().size() > 0);
 
@@ -769,19 +769,13 @@ Status KVCatalog::createCollection(OperationContext* opCtx,
     opCtx->recoveryUnit()->onRollback([ opCtx, catalog = this, nss, ident, uuid ]() {
         // Intentionally ignoring failure
         catalog->_engine->getEngine()->dropIdent(opCtx, ident).ignore();
-
-        CollectionCatalog::get(opCtx).deregisterCatalogEntry(uuid);
     });
 
     auto rs = _engine->getEngine()->getGroupedRecordStore(opCtx, nss.ns(), ident, options, prefix);
     invariant(rs);
 
-    CollectionCatalog::get(getGlobalServiceContext())
-        .registerCatalogEntry(uuid,
-                              std::make_unique<KVCollectionCatalogEntry>(
-                                  _engine, this, nss.ns(), ident, std::move(rs)));
-
-    return Status::OK();
+    return std::make_unique<KVCollectionCatalogEntry>(
+        _engine, this, nss.ns(), ident, std::move(rs));
 }
 
 Status KVCatalog::renameCollection(OperationContext* opCtx,
@@ -808,29 +802,6 @@ Status KVCatalog::renameCollection(OperationContext* opCtx,
     return Status::OK();
 }
 
-class KVCatalog::FinishDropCatalogEntryChange : public RecoveryUnit::Change {
-public:
-    FinishDropCatalogEntryChange(CollectionCatalog& catalog,
-                                 std::unique_ptr<CollectionCatalogEntry> collectionCatalogEntry,
-                                 CollectionUUID uuid)
-        : _catalog(catalog),
-          _collectionCatalogEntry(std::move(collectionCatalogEntry)),
-          _uuid(uuid) {}
-
-    void commit(boost::optional<Timestamp>) override {
-        _collectionCatalogEntry.reset();
-    }
-
-    void rollback() override {
-        _catalog.registerCatalogEntry(_uuid, std::move(_collectionCatalogEntry));
-    }
-
-private:
-    CollectionCatalog& _catalog;
-    std::unique_ptr<CollectionCatalogEntry> _collectionCatalogEntry;
-    CollectionUUID _uuid;
-};
-
 Status KVCatalog::dropCollection(OperationContext* opCtx, const NamespaceString& nss) {
     invariant(opCtx->lockState()->isCollectionLockedForMode(nss, MODE_X));
 
@@ -863,13 +834,6 @@ Status KVCatalog::dropCollection(OperationContext* opCtx, const NamespaceString&
         return status;
     }
 
-    // Remove catalog entry
-    std::unique_ptr<CollectionCatalogEntry> removedCatalogEntry =
-        CollectionCatalog::get(opCtx).deregisterCatalogEntry(uuid.get());
-
-    opCtx->recoveryUnit()->registerChange(new FinishDropCatalogEntryChange(
-        CollectionCatalog::get(opCtx), std::move(removedCatalogEntry), uuid.get()));
-
     // This will lazily delete the KVCollectionCatalogEntry and notify the storageEngine to
     // drop the collection only on WUOW::commit().
     opCtx->recoveryUnit()->onCommit(
@@ -888,7 +852,6 @@ Status KVCatalog::dropCollection(OperationContext* opCtx, const NamespaceString&
             }
         });
 
-
     return Status::OK();
 }
 }
diff --git a/src/mongo/db/storage/kv/kv_catalog.h b/src/mongo/db/storage/kv/kv_catalog.h
index b22b6cf68ff..a173209b53d 100644
--- a/src/mongo/db/storage/kv/kv_catalog.h
+++ b/src/mongo/db/storage/kv/kv_catalog.h
@@ -34,6 +34,7 @@
 #include <string>
 
 #include "mongo/base/string_data.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/record_id.h"
 #include "mongo/db/storage/bson_collection_catalog_entry.h"
@@ -114,10 +115,11 @@ public:
                                                                        const NamespaceString& nss,
                                                                        bool forRepair);
 
-    Status createCollection(OperationContext* opCtx,
-                            const NamespaceString& nss,
-                            const CollectionOptions& options,
-                            bool allocateDefaultSpace);
+    StatusWith<std::unique_ptr<CollectionCatalogEntry>> createCollection(
+        OperationContext* opCtx,
+        const NamespaceString& nss,
+        const CollectionOptions& options,
+        bool allocateDefaultSpace);
 
     Status renameCollection(OperationContext* opCtx,
                             const NamespaceString& fromNss,
@@ -129,7 +131,6 @@ public:
 private:
     class AddIdentChange;
     class RemoveIdentChange;
-    class FinishDropCatalogEntryChange;
 
     friend class KVStorageEngine;
     friend class KVCatalogTest;
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp b/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp
index 7d4e035c222..4beb5d860fc 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp
@@ -84,8 +84,14 @@ public:
             const bool allocateDefaultSpace = true;
             CollectionOptions options;
             options.uuid = UUID::gen();
-            ASSERT_OK(_storageEngine.getCatalog()->createCollection(
-                opCtx.get(), _nss, options, allocateDefaultSpace));
+            auto statusWithCatalogEntry = _storageEngine.getCatalog()->createCollection(
+                opCtx.get(), _nss, options, allocateDefaultSpace);
+            ASSERT_OK(statusWithCatalogEntry.getStatus());
+            auto collection = std::make_unique<CollectionMock>(_nss);
+            CollectionCatalog::get(opCtx.get())
+                .registerCollection(options.uuid.get(),
+                                    std::move(statusWithCatalogEntry.getValue()),
+                                    std::move(collection));
             wuow.commit();
         }
     }
diff --git a/src/mongo/db/storage/kv/kv_storage_engine.cpp b/src/mongo/db/storage/kv/kv_storage_engine.cpp
index d295631076b..328fa7cd1c5 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine.cpp
+++ b/src/mongo/db/storage/kv/kv_storage_engine.cpp
@@ -245,8 +245,7 @@ void KVStorageEngine::_initCollection(OperationContext* opCtx,
     auto collection = collectionFactory->make(opCtx, catalogEntry.get());
 
     auto& collectionCatalog = CollectionCatalog::get(getGlobalServiceContext());
-    collectionCatalog.registerCatalogEntry(uuid, std::move(catalogEntry));
-    collectionCatalog.registerCollectionObject(uuid, std::move(collection));
+    collectionCatalog.registerCollection(uuid, std::move(catalogEntry), std::move(collection));
 }
 
 void KVStorageEngine::closeCatalog(OperationContext* opCtx) {
@@ -594,10 +593,18 @@ Status KVStorageEngine::_dropCollectionsNoTimestamp(OperationContext* opCtx,
     WriteUnitOfWork untimestampedDropWuow(opCtx);
     for (auto& nss : toDrop) {
         invariant(getCatalog());
+        auto uuid = CollectionCatalog::get(opCtx).lookupUUIDByNSS(nss).get();
         Status result = getCatalog()->dropCollection(opCtx, nss);
+
         if (!result.isOK() && firstError.isOK()) {
             firstError = result;
         }
+
+        auto[removedColl, removedCatalogEntry] =
+            CollectionCatalog::get(opCtx).deregisterCollection(uuid);
+        opCtx->recoveryUnit()->registerChange(
+            CollectionCatalog::get(opCtx).makeFinishDropCollectionChange(
+                std::move(removedColl), std::move(removedCatalogEntry), uuid));
     }
 
     untimestampedDropWuow.commit();
@@ -665,17 +672,10 @@ Status KVStorageEngine::repairRecordStore(OperationContext* opCtx, const Namespa
                                                      << status.reason());
     }
 
+    // After repairing, re-initialize the collection with a valid RecordStore.
     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.
+    collectionCatalog.deregisterCollection(uuid);
     _initCollection(opCtx, nss, false);
     return Status::OK();
 }
diff --git a/src/mongo/db/storage/kv/kv_storage_engine_test_fixture.h b/src/mongo/db/storage/kv/kv_storage_engine_test_fixture.h
index 29cf88e7a84..c9cfbafe911 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/kv/kv_storage_engine_test_fixture.h
@@ -55,10 +55,10 @@ public:
         AutoGetDb db(opCtx, ns.db(), LockMode::MODE_X);
         CollectionOptions options;
         options.uuid = UUID::gen();
-        auto ret = _storageEngine->getCatalog()->createCollection(opCtx, ns, options, true);
-        if (!ret.isOK()) {
-            return ret;
-        }
+        auto catalogEntry = unittest::assertGet(
+            _storageEngine->getCatalog()->createCollection(opCtx, ns, options, true));
+        CollectionCatalog::get(opCtx).registerCollection(
+            options.uuid.get(), std::move(catalogEntry), std::make_unique<CollectionMock>(ns));
 
         return _storageEngine->getCatalog()->getCollectionIdent(ns);
     }