SERVER-47885 Added lookupCollectionByXXXForRead interface to the Collection catalog that returns collection as shared_ptr<const Collection>

AutoGetCollectionForRead and AutoGetCollectionForReadCommand now uses this and holds the shared_ptr. They return the collection as const.
Const correct various places to make this possible.

Moved some logic from Collection destructors to deregister from the catalog as they may now be destroyed at a later point.

93 files changed, 472 insertions, 358 deletions

diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index f3c722afee0..c8010e21b6b 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -346,7 +346,7 @@ Status _collModInternal(OperationContext* opCtx,
                         const BSONObj& cmdObj,
                         BSONObjBuilder* result) {
     StringData dbName = nss.db();
-    AutoGetCollection autoColl(opCtx, nss, MODE_X, AutoGetCollection::ViewMode::kViewsPermitted);
+    AutoGetCollection autoColl(opCtx, nss, MODE_X, AutoGetCollectionViewMode::kViewsPermitted);
     Lock::CollectionLock systemViewsLock(
         opCtx, NamespaceString(dbName, NamespaceString::kSystemDotViewsCollectionName), MODE_X);
 
diff --git a/src/mongo/db/catalog/collection.cpp b/src/mongo/db/catalog/collection.cpp
index 1969d159859..e02689c7622 100644
--- a/src/mongo/db/catalog/collection.cpp
+++ b/src/mongo/db/catalog/collection.cpp
@@ -41,7 +41,7 @@ namespace mongo {
 // CappedInsertNotifier
 //
 
-void CappedInsertNotifier::notifyAll() {
+void CappedInsertNotifier::notifyAll() const {
     stdx::lock_guard<Latch> lk(_mutex);
     ++_version;
     _notifier.notify_all();
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index 337e474e1dd..9b0d5f5b223 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -109,7 +109,7 @@ public:
     /**
      * Wakes up all threads waiting.
      */
-    void notifyAll();
+    void notifyAll() const;
 
     /**
      * Waits until 'deadline', or until notifyAll() is called to indicate that new
@@ -147,7 +147,7 @@ private:
     //
     // The condition which '_cappedNewDataNotifier' is being notified of is an increment of this
     // counter. Access to this counter is synchronized with '_mutex'.
-    uint64_t _version = 0;
+    mutable uint64_t _version = 0;
 
     // True once the notifier is dead.
     bool _dead = false;
@@ -190,7 +190,7 @@ public:
          * Constructs a Collection object. This does not persist any state to the storage engine,
          * only constructs an in-memory representation of what already exists on disk.
          */
-        virtual std::unique_ptr<Collection> make(OperationContext* opCtx,
+        virtual std::shared_ptr<Collection> make(OperationContext* opCtx,
                                                  const NamespaceString& nss,
                                                  RecordId catalogId,
                                                  CollectionUUID uuid,
@@ -490,6 +490,7 @@ public:
      * The storage engine interacts with capped collections through a CappedCallback interface.
      */
     virtual CappedCallback* getCappedCallback() = 0;
+    virtual const CappedCallback* getCappedCallback() const = 0;
 
     /**
      * Get a pointer to a capped insert notifier object. The caller can wait on this object
@@ -519,7 +520,7 @@ public:
      * If return value is not boost::none, reads with majority read concern using an older snapshot
      * must error.
      */
-    virtual boost::optional<Timestamp> getMinimumVisibleSnapshot() = 0;
+    virtual boost::optional<Timestamp> getMinimumVisibleSnapshot() const = 0;
 
     virtual void setMinimumVisibleSnapshot(const Timestamp name) = 0;
 
@@ -547,7 +548,7 @@ public:
     virtual std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> makePlanExecutor(
         OperationContext* opCtx,
         PlanYieldPolicy::YieldPolicy yieldPolicy,
-        ScanDirection scanDirection) = 0;
+        ScanDirection scanDirection) const = 0;
 
     virtual void indexBuildSuccess(OperationContext* opCtx, IndexCatalogEntry* index) = 0;
 
@@ -559,6 +560,11 @@ public:
      */
     virtual void establishOplogCollectionForLogging(OperationContext* opCtx) = 0;
 
+    /**
+     * Called when this Collection is deregistered from the catalog
+     */
+    virtual void onDeregisterFromCatalog() = 0;
+
     friend auto logAttrs(const Collection& col) {
         return logv2::multipleAttrs(col.ns(), col.uuid());
     }
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index f20653dc474..3ffb309d55a 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -50,7 +50,7 @@ const ServiceContext::Decoration<CollectionCatalog> getCatalog =
 class FinishDropCollectionChange : public RecoveryUnit::Change {
 public:
     FinishDropCollectionChange(CollectionCatalog* catalog,
-                               std::unique_ptr<Collection> coll,
+                               std::shared_ptr<Collection> coll,
                                CollectionUUID uuid)
         : _catalog(catalog), _coll(std::move(coll)), _uuid(uuid) {}
 
@@ -59,12 +59,12 @@ public:
     }
 
     void rollback() override {
-        _catalog->registerCollection(_uuid, std::move(&_coll));
+        _catalog->registerCollection(_uuid, std::move(_coll));
     }
 
 private:
     CollectionCatalog* _catalog;
-    std::unique_ptr<Collection> _coll;
+    std::shared_ptr<Collection> _coll;
     CollectionUUID _uuid;
 };
 
@@ -89,8 +89,8 @@ CollectionCatalog::iterator::iterator(StringData dbName,
     }
 }
 
-CollectionCatalog::iterator::iterator(
-    std::map<std::pair<std::string, CollectionUUID>, Collection*>::const_iterator mapIter)
+CollectionCatalog::iterator::iterator(std::map<std::pair<std::string, CollectionUUID>,
+                                               std::shared_ptr<Collection>>::const_iterator mapIter)
     : _mapIter(mapIter) {}
 
 CollectionCatalog::iterator::value_type CollectionCatalog::iterator::operator*() {
@@ -100,7 +100,7 @@ CollectionCatalog::iterator::value_type CollectionCatalog::iterator::operator*()
         return _nullCollection;
     }
 
-    return _mapIter->second;
+    return _mapIter->second.get();
 }
 
 boost::optional<CollectionUUID> CollectionCatalog::iterator::uuid() {
@@ -257,8 +257,8 @@ uint64_t CollectionCatalog::getEpoch() const {
     return _epoch;
 }
 
-Collection* CollectionCatalog::lookupCollectionByUUID(OperationContext* opCtx,
-                                                      CollectionUUID uuid) const {
+std::shared_ptr<const Collection> CollectionCatalog::lookupCollectionByUUIDForRead(
+    OperationContext* opCtx, CollectionUUID uuid) const {
     if (auto coll = UncommittedCollections::getForTxn(opCtx, uuid)) {
         return coll;
     }
@@ -268,6 +268,17 @@ Collection* CollectionCatalog::lookupCollectionByUUID(OperationContext* opCtx,
     return (coll && coll->isCommitted()) ? coll : nullptr;
 }
 
+Collection* CollectionCatalog::lookupCollectionByUUID(OperationContext* opCtx,
+                                                      CollectionUUID uuid) const {
+    if (auto coll = UncommittedCollections::getForTxn(opCtx, uuid)) {
+        return coll.get();
+    }
+
+    stdx::lock_guard<Latch> lock(_catalogLock);
+    auto coll = _lookupCollectionByUUID(lock, uuid);
+    return (coll && coll->isCommitted()) ? coll.get() : nullptr;
+}
+
 void CollectionCatalog::makeCollectionVisible(CollectionUUID uuid) {
     stdx::lock_guard<Latch> lock(_catalogLock);
     auto coll = _lookupCollectionByUUID(lock, uuid);
@@ -280,13 +291,14 @@ bool CollectionCatalog::isCollectionAwaitingVisibility(CollectionUUID uuid) cons
     return coll && !coll->isCommitted();
 }
 
-Collection* CollectionCatalog::_lookupCollectionByUUID(WithLock, CollectionUUID uuid) const {
+std::shared_ptr<Collection> CollectionCatalog::_lookupCollectionByUUID(WithLock,
+                                                                       CollectionUUID uuid) const {
     auto foundIt = _catalog.find(uuid);
-    return foundIt == _catalog.end() ? nullptr : foundIt->second.get();
+    return foundIt == _catalog.end() ? nullptr : foundIt->second;
 }
 
-Collection* CollectionCatalog::lookupCollectionByNamespace(OperationContext* opCtx,
-                                                           const NamespaceString& nss) const {
+std::shared_ptr<const Collection> CollectionCatalog::lookupCollectionByNamespaceForRead(
+    OperationContext* opCtx, const NamespaceString& nss) const {
     if (auto coll = UncommittedCollections::getForTxn(opCtx, nss)) {
         return coll;
     }
@@ -297,6 +309,18 @@ Collection* CollectionCatalog::lookupCollectionByNamespace(OperationContext* opC
     return (coll && coll->isCommitted()) ? coll : nullptr;
 }
 
+Collection* CollectionCatalog::lookupCollectionByNamespace(OperationContext* opCtx,
+                                                           const NamespaceString& nss) const {
+    if (auto coll = UncommittedCollections::getForTxn(opCtx, nss)) {
+        return coll.get();
+    }
+
+    stdx::lock_guard<Latch> lock(_catalogLock);
+    auto it = _collections.find(nss);
+    auto coll = (it == _collections.end() ? nullptr : it->second);
+    return (coll && coll->isCommitted()) ? coll.get() : nullptr;
+}
+
 boost::optional<NamespaceString> CollectionCatalog::lookupNSSByUUID(OperationContext* opCtx,
                                                                     CollectionUUID uuid) const {
     if (auto coll = UncommittedCollections::getForTxn(opCtx, uuid)) {
@@ -371,7 +395,7 @@ bool CollectionCatalog::checkIfCollectionSatisfiable(CollectionUUID uuid,
         return false;
     }
 
-    return predicate(collection);
+    return predicate(collection.get());
 }
 
 std::vector<CollectionUUID> CollectionCatalog::getAllCollectionUUIDsFromDb(
@@ -449,15 +473,14 @@ void CollectionCatalog::clearDatabaseProfileLevel(StringData dbName) {
     _databaseProfileLevels.erase(dbName);
 }
 
-void CollectionCatalog::registerCollection(CollectionUUID uuid, std::unique_ptr<Collection>* coll) {
-    auto ns = (*coll)->ns();
+void CollectionCatalog::registerCollection(CollectionUUID uuid, std::shared_ptr<Collection> coll) {
+    auto ns = coll->ns();
     stdx::lock_guard<Latch> lock(_catalogLock);
     if (_collections.find(ns) != _collections.end()) {
         LOGV2(20279,
               "Conflicted creating a collection. ns: {coll_ns} ({coll_uuid}).",
               "Conflicted creating a collection",
-              "namespace"_attr = (*coll)->ns(),
-              "uuid"_attr = (*coll)->uuid());
+              logAttrs(*coll));
         throw WriteConflictException();
     }
 
@@ -475,9 +498,9 @@ void CollectionCatalog::registerCollection(CollectionUUID uuid, std::unique_ptr<
     invariant(_catalog.find(uuid) == _catalog.end());
     invariant(_orderedCollections.find(dbIdPair) == _orderedCollections.end());
 
-    _catalog[uuid] = std::move(*coll);
-    _collections[ns] = _catalog[uuid].get();
-    _orderedCollections[dbIdPair] = _catalog[uuid].get();
+    _catalog[uuid] = coll;
+    _collections[ns] = coll;
+    _orderedCollections[dbIdPair] = coll;
 
     auto dbRid = ResourceId(RESOURCE_DATABASE, dbName);
     addResource(dbRid, dbName);
@@ -486,7 +509,7 @@ void CollectionCatalog::registerCollection(CollectionUUID uuid, std::unique_ptr<
     addResource(collRid, ns.ns());
 }
 
-std::unique_ptr<Collection> CollectionCatalog::deregisterCollection(CollectionUUID uuid) {
+std::shared_ptr<Collection> CollectionCatalog::deregisterCollection(CollectionUUID uuid) {
     stdx::lock_guard<Latch> lock(_catalogLock);
 
     invariant(_catalog.find(uuid) != _catalog.end());
@@ -506,6 +529,8 @@ std::unique_ptr<Collection> CollectionCatalog::deregisterCollection(CollectionUU
     _collections.erase(ns);
     _catalog.erase(uuid);
 
+    coll->onDeregisterFromCatalog();
+
     auto collRid = ResourceId(RESOURCE_COLLECTION, ns.ns());
     removeResource(collRid, ns.ns());
 
@@ -517,7 +542,7 @@ std::unique_ptr<Collection> CollectionCatalog::deregisterCollection(CollectionUU
 }
 
 std::unique_ptr<RecoveryUnit::Change> CollectionCatalog::makeFinishDropCollectionChange(
-    std::unique_ptr<Collection> coll, CollectionUUID uuid) {
+    std::shared_ptr<Collection> coll, CollectionUUID uuid) {
     return std::make_unique<FinishDropCollectionChange>(this, std::move(coll), uuid);
 }
 
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index 2fd90339a29..5ee50d0b745 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -61,8 +61,8 @@ public:
         using value_type = Collection*;
 
         iterator(StringData dbName, uint64_t genNum, const CollectionCatalog& catalog);
-        iterator(
-            std::map<std::pair<std::string, CollectionUUID>, Collection*>::const_iterator mapIter);
+        iterator(std::map<std::pair<std::string, CollectionUUID>,
+                          std::shared_ptr<Collection>>::const_iterator mapIter);
         value_type operator*();
         iterator operator++();
         iterator operator++(int);
@@ -90,7 +90,8 @@ public:
         std::string _dbName;
         boost::optional<CollectionUUID> _uuid;
         uint64_t _genNum;
-        std::map<std::pair<std::string, CollectionUUID>, Collection*>::const_iterator _mapIter;
+        std::map<std::pair<std::string, CollectionUUID>,
+                 std::shared_ptr<Collection>>::const_iterator _mapIter;
         const CollectionCatalog* _catalog;
         static constexpr Collection* _nullCollection = nullptr;
     };
@@ -116,18 +117,18 @@ public:
     /**
      * Register the collection with `uuid`.
      */
-    void registerCollection(CollectionUUID uuid, std::unique_ptr<Collection>* collection);
+    void registerCollection(CollectionUUID uuid, std::shared_ptr<Collection> collection);
 
     /**
      * Deregister the collection.
      */
-    std::unique_ptr<Collection> deregisterCollection(CollectionUUID uuid);
+    std::shared_ptr<Collection> deregisterCollection(CollectionUUID uuid);
 
     /**
      * Returns the RecoveryUnit's Change for dropping the collection
      */
     std::unique_ptr<RecoveryUnit::Change> makeFinishDropCollectionChange(
-        std::unique_ptr<Collection>, CollectionUUID uuid);
+        std::shared_ptr<Collection>, CollectionUUID uuid);
 
     /**
      * Deregister all the collection objects.
@@ -142,6 +143,9 @@ public:
      * Returns nullptr if the 'uuid' is not known.
      */
     Collection* lookupCollectionByUUID(OperationContext* opCtx, CollectionUUID uuid) const;
+    std::shared_ptr<const Collection> lookupCollectionByUUIDForRead(OperationContext* opCtx,
+                                                                    CollectionUUID uuid) const;
+
 
     void makeCollectionVisible(CollectionUUID uuid);
 
@@ -160,6 +164,8 @@ public:
      */
     Collection* lookupCollectionByNamespace(OperationContext* opCtx,
                                             const NamespaceString& nss) const;
+    std::shared_ptr<const Collection> lookupCollectionByNamespaceForRead(
+        OperationContext* opCtx, const NamespaceString& nss) const;
 
     /**
      * This function gets the NamespaceString from the collection catalog entry that
@@ -286,7 +292,7 @@ public:
 private:
     friend class CollectionCatalog::iterator;
 
-    Collection* _lookupCollectionByUUID(WithLock, CollectionUUID uuid) const;
+    std::shared_ptr<Collection> _lookupCollectionByUUID(WithLock, CollectionUUID uuid) const;
 
     const std::vector<CollectionUUID>& _getOrdering_inlock(const StringData& db,
                                                            const stdx::lock_guard<Latch>&);
@@ -301,9 +307,11 @@ private:
         _shadowCatalog;
 
     using CollectionCatalogMap =
-        stdx::unordered_map<CollectionUUID, std::unique_ptr<Collection>, CollectionUUID::Hash>;
-    using OrderedCollectionMap = std::map<std::pair<std::string, CollectionUUID>, Collection*>;
-    using NamespaceCollectionMap = stdx::unordered_map<NamespaceString, Collection*>;
+        stdx::unordered_map<CollectionUUID, std::shared_ptr<Collection>, CollectionUUID::Hash>;
+    using OrderedCollectionMap =
+        std::map<std::pair<std::string, CollectionUUID>, std::shared_ptr<Collection>>;
+    using NamespaceCollectionMap =
+        stdx::unordered_map<NamespaceString, std::shared_ptr<Collection>>;
     using DatabaseProfileLevelMap = StringMap<int>;
 
     CollectionCatalogMap _catalog;
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index 4a8e5785d4a..9781eb640df 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -68,10 +68,10 @@ public:
     void setUp() override {
         ServiceContextMongoDTest::setUp();
 
-        std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
+        std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
         col = collection.get();
         // Register dummy collection in catalog.
-        catalog.registerCollection(colUUID, &collection);
+        catalog.registerCollection(colUUID, std::move(collection));
     }
 
 protected:
@@ -92,15 +92,15 @@ public:
             NamespaceString barNss("bar", "coll" + std::to_string(counter));
 
             auto fooUuid = CollectionUUID::gen();
-            std::unique_ptr<Collection> fooColl = std::make_unique<CollectionMock>(fooNss);
+            std::shared_ptr<Collection> fooColl = std::make_shared<CollectionMock>(fooNss);
 
             auto barUuid = CollectionUUID::gen();
-            std::unique_ptr<Collection> barColl = std::make_unique<CollectionMock>(barNss);
+            std::shared_ptr<Collection> barColl = std::make_shared<CollectionMock>(barNss);
 
             dbMap["foo"].insert(std::make_pair(fooUuid, fooColl.get()));
             dbMap["bar"].insert(std::make_pair(barUuid, barColl.get()));
-            catalog.registerCollection(fooUuid, &fooColl);
-            catalog.registerCollection(barUuid, &barColl);
+            catalog.registerCollection(fooUuid, fooColl);
+            catalog.registerCollection(barUuid, barColl);
         }
     }
 
@@ -274,10 +274,10 @@ public:
     void setUp() {
         for (int i = 0; i < 5; i++) {
             NamespaceString nss("resourceDb", "coll" + std::to_string(i));
-            std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
+            std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
             auto uuid = collection->uuid();
 
-            catalog.registerCollection(uuid, &collection);
+            catalog.registerCollection(uuid, std::move(collection));
         }
 
         int numEntries = 0;
@@ -496,13 +496,13 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUID) {
 TEST_F(CollectionCatalogTest, InsertAfterLookup) {
     auto newUUID = CollectionUUID::gen();
     NamespaceString newNss(nss.db(), "newcol");
-    std::unique_ptr<Collection> newCollUnique = std::make_unique<CollectionMock>(newNss);
-    auto newCol = newCollUnique.get();
+    std::shared_ptr<Collection> newCollShared = std::make_shared<CollectionMock>(newNss);
+    auto newCol = newCollShared.get();
 
     // Ensure that looking up non-existing UUIDs doesn't affect later registration of those UUIDs.
     ASSERT(catalog.lookupCollectionByUUID(&opCtx, newUUID) == nullptr);
     ASSERT_EQUALS(catalog.lookupNSSByUUID(&opCtx, newUUID), boost::none);
-    catalog.registerCollection(newUUID, &newCollUnique);
+    catalog.registerCollection(newUUID, std::move(newCollShared));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(&opCtx, newUUID), newCol);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(&opCtx, colUUID), nss);
 }
@@ -516,9 +516,9 @@ TEST_F(CollectionCatalogTest, OnDropCollection) {
 TEST_F(CollectionCatalogTest, RenameCollection) {
     auto uuid = CollectionUUID::gen();
     NamespaceString oldNss(nss.db(), "oldcol");
-    std::unique_ptr<Collection> collUnique = std::make_unique<CollectionMock>(oldNss);
-    auto collection = collUnique.get();
-    catalog.registerCollection(uuid, &collUnique);
+    std::shared_ptr<Collection> collShared = std::make_shared<CollectionMock>(oldNss);
+    auto collection = collShared.get();
+    catalog.registerCollection(uuid, std::move(collShared));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(&opCtx, uuid), collection);
 
     NamespaceString newNss(nss.db(), "newcol");
@@ -539,14 +539,14 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsOldNSSIfDrop
 TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsNewlyCreatedNSS) {
     auto newUUID = CollectionUUID::gen();
     NamespaceString newNss(nss.db(), "newcol");
-    std::unique_ptr<Collection> newCollUnique = std::make_unique<CollectionMock>(newNss);
-    auto newCol = newCollUnique.get();
+    std::shared_ptr<Collection> newCollShared = std::make_shared<CollectionMock>(newNss);
+    auto newCol = newCollShared.get();
 
     // Ensure that looking up non-existing UUIDs doesn't affect later registration of those UUIDs.
     catalog.onCloseCatalog(&opCtx);
     ASSERT(catalog.lookupCollectionByUUID(&opCtx, newUUID) == nullptr);
     ASSERT_EQUALS(catalog.lookupNSSByUUID(&opCtx, newUUID), boost::none);
-    catalog.registerCollection(newUUID, &newCollUnique);
+    catalog.registerCollection(newUUID, std::move(newCollShared));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(&opCtx, newUUID), newCol);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(&opCtx, colUUID), nss);
 
@@ -558,14 +558,14 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsNewlyCreated
 
 TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsFreshestNSS) {
     NamespaceString newNss(nss.db(), "newcol");
-    std::unique_ptr<Collection> newCollUnique = std::make_unique<CollectionMock>(newNss);
-    auto newCol = newCollUnique.get();
+    std::shared_ptr<Collection> newCollShared = std::make_shared<CollectionMock>(newNss);
+    auto newCol = newCollShared.get();
 
     catalog.onCloseCatalog(&opCtx);
     catalog.deregisterCollection(colUUID);
     ASSERT(catalog.lookupCollectionByUUID(&opCtx, colUUID) == nullptr);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(&opCtx, colUUID), nss);
-    catalog.registerCollection(colUUID, &newCollUnique);
+    catalog.registerCollection(colUUID, std::move(newCollShared));
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(&opCtx, colUUID), newCol);
     ASSERT_EQUALS(*catalog.lookupNSSByUUID(&opCtx, colUUID), newNss);
 
@@ -600,9 +600,9 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNames) {
 
     std::vector<NamespaceString> nsss = {aColl, b1Coll, b2Coll, cColl, d1Coll, d2Coll, d3Coll};
     for (auto& nss : nsss) {
-        std::unique_ptr<Collection> newColl = std::make_unique<CollectionMock>(nss);
+        std::shared_ptr<Collection> newColl = std::make_shared<CollectionMock>(nss);
         auto uuid = CollectionUUID::gen();
-        catalog.registerCollection(uuid, &newColl);
+        catalog.registerCollection(uuid, std::move(newColl));
     }
 
     std::vector<NamespaceString> dCollList = {d1Coll, d2Coll, d3Coll};
@@ -646,9 +646,9 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNamesWithUncommitt
 
     std::vector<NamespaceString> nsss = {aColl, b1Coll, b2Coll, cColl, d1Coll, d2Coll, d3Coll};
     for (auto& nss : nsss) {
-        std::unique_ptr<Collection> newColl = std::make_unique<CollectionMock>(nss);
+        std::shared_ptr<Collection> newColl = std::make_shared<CollectionMock>(nss);
         auto uuid = CollectionUUID::gen();
-        catalog.registerCollection(uuid, &newColl);
+        catalog.registerCollection(uuid, std::move(newColl));
     }
 
     // One dbName with only an invisible collection does not appear in dbNames.
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 9d808376785..2c6dbeb335b 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -259,19 +259,21 @@ CollectionImpl::~CollectionImpl() {
         _recordStore->setCappedCallback(nullptr);
         _cappedNotifier->kill();
     }
+}
 
+void CollectionImpl::onDeregisterFromCatalog() {
     if (ns().isOplog()) {
         repl::clearLocalOplogPtr();
     }
 }
 
-std::unique_ptr<Collection> CollectionImpl::FactoryImpl::make(
+std::shared_ptr<Collection> CollectionImpl::FactoryImpl::make(
     OperationContext* opCtx,
     const NamespaceString& nss,
     RecordId catalogId,
     CollectionUUID uuid,
     std::unique_ptr<RecordStore> rs) const {
-    return std::make_unique<CollectionImpl>(opCtx, nss, catalogId, uuid, std::move(rs));
+    return std::make_shared<CollectionImpl>(opCtx, nss, catalogId, uuid, std::move(rs));
 }
 
 SharedCollectionDecorations* CollectionImpl::getSharedDecorations() const {
@@ -665,13 +667,13 @@ void CollectionImpl::setMinimumVisibleSnapshot(Timestamp newMinimumVisibleSnapsh
     }
 }
 
-bool CollectionImpl::haveCappedWaiters() {
+bool CollectionImpl::haveCappedWaiters() const {
     // Waiters keep a shared_ptr to '_cappedNotifier', so there are waiters if this CollectionImpl's
     // shared_ptr is not unique (use_count > 1).
     return _cappedNotifier.use_count() > 1;
 }
 
-void CollectionImpl::notifyCappedWaitersIfNeeded() {
+void CollectionImpl::notifyCappedWaitersIfNeeded() const {
     // If there is a notifier object and another thread is waiting on it, then we notify
     // waiters of this document insert.
     if (haveCappedWaiters())
@@ -881,6 +883,10 @@ CappedCallback* CollectionImpl::getCappedCallback() {
     return this;
 }
 
+const CappedCallback* CollectionImpl::getCappedCallback() const {
+    return this;
+}
+
 std::shared_ptr<CappedInsertNotifier> CollectionImpl::getCappedInsertNotifier() const {
     invariant(isCapped());
     return _cappedNotifier;
@@ -1189,7 +1195,7 @@ StatusWith<std::vector<BSONObj>> CollectionImpl::addCollationDefaultsToIndexSpec
 std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> CollectionImpl::makePlanExecutor(
     OperationContext* opCtx,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
-    ScanDirection scanDirection) {
+    ScanDirection scanDirection) const {
     auto isForward = scanDirection == ScanDirection::kForward;
     auto direction = isForward ? InternalPlanner::FORWARD : InternalPlanner::BACKWARD;
     return InternalPlanner::collectionScan(opCtx, _ns.ns(), this, yieldPolicy, direction);
diff --git a/src/mongo/db/catalog/collection_impl.h b/src/mongo/db/catalog/collection_impl.h
index de264eea471..573cda25612 100644
--- a/src/mongo/db/catalog/collection_impl.h
+++ b/src/mongo/db/catalog/collection_impl.h
@@ -52,7 +52,7 @@ public:
 
     class FactoryImpl : public Factory {
     public:
-        std::unique_ptr<Collection> make(OperationContext* opCtx,
+        std::shared_ptr<Collection> make(OperationContext* opCtx,
                                          const NamespaceString& nss,
                                          RecordId catalogId,
                                          CollectionUUID uuid,
@@ -284,6 +284,7 @@ public:
     bool isCapped() const final;
 
     CappedCallback* getCappedCallback() final;
+    const CappedCallback* getCappedCallback() const final;
 
     /**
      * Get a pointer to a capped insert notifier object. The caller can wait on this object
@@ -323,7 +324,7 @@ public:
      * If return value is not boost::none, reads with majority read concern using an older snapshot
      * must error.
      */
-    boost::optional<Timestamp> getMinimumVisibleSnapshot() final {
+    boost::optional<Timestamp> getMinimumVisibleSnapshot() const final {
         return _minVisibleSnapshot;
     }
 
@@ -333,12 +334,12 @@ public:
      */
     void setMinimumVisibleSnapshot(Timestamp newMinimumVisibleSnapshot) final;
 
-    bool haveCappedWaiters() final;
+    bool haveCappedWaiters() const final;
 
     /**
      * Notify (capped collection) waiters of data changes, like an insert.
      */
-    void notifyCappedWaitersIfNeeded() final;
+    void notifyCappedWaitersIfNeeded() const final;
 
     /**
      * Get a pointer to the collection's default collator. The pointer must not be used after this
@@ -352,11 +353,12 @@ public:
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> makePlanExecutor(
         OperationContext* opCtx,
         PlanYieldPolicy::YieldPolicy yieldPolicy,
-        ScanDirection scanDirection) final;
+        ScanDirection scanDirection) const final;
 
     void indexBuildSuccess(OperationContext* opCtx, IndexCatalogEntry* index) final;
 
     void establishOplogCollectionForLogging(OperationContext* opCtx) final;
+    void onDeregisterFromCatalog() final;
 
 private:
     /**
diff --git a/src/mongo/db/catalog/collection_mock.h b/src/mongo/db/catalog/collection_mock.h
index 21332edb9e8..e07a110f8ef 100644
--- a/src/mongo/db/catalog/collection_mock.h
+++ b/src/mongo/db/catalog/collection_mock.h
@@ -38,7 +38,7 @@ namespace mongo {
 /**
  * This class comprises a mock Collection for use by CollectionCatalog unit tests.
  */
-class CollectionMock : public Collection {
+class CollectionMock final : public Collection {
 public:
     CollectionMock(const NamespaceString& ns)
         : CollectionMock(ns, std::unique_ptr<IndexCatalog>()) {}
@@ -226,6 +226,9 @@ public:
     CappedCallback* getCappedCallback() {
         std::abort();
     }
+    const CappedCallback* getCappedCallback() const {
+        std::abort();
+    }
 
     std::shared_ptr<CappedInsertNotifier> getCappedInsertNotifier() const {
         std::abort();
@@ -251,7 +254,7 @@ public:
         std::abort();
     }
 
-    boost::optional<Timestamp> getMinimumVisibleSnapshot() {
+    boost::optional<Timestamp> getMinimumVisibleSnapshot() const {
         std::abort();
     }
 
@@ -271,7 +274,7 @@ public:
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> makePlanExecutor(
         OperationContext* opCtx,
         PlanYieldPolicy::YieldPolicy yieldPolicy,
-        ScanDirection scanDirection) {
+        ScanDirection scanDirection) const {
         std::abort();
     }
 
@@ -279,6 +282,8 @@ public:
         std::abort();
     }
 
+    void onDeregisterFromCatalog() {}
+
     UUID uuid() const {
         return _uuid;
     }
diff --git a/src/mongo/db/catalog/collection_test.cpp b/src/mongo/db/catalog/collection_test.cpp
index d831e8001a2..b973cf5a949 100644
--- a/src/mongo/db/catalog/collection_test.cpp
+++ b/src/mongo/db/catalog/collection_test.cpp
@@ -62,7 +62,7 @@ TEST_F(CollectionTest, CappedNotifierKillAndIsDead) {
     makeCapped(nss);
 
     AutoGetCollectionForRead acfr(operationContext(), nss);
-    Collection* col = acfr.getCollection();
+    const Collection* col = acfr.getCollection();
     auto notifier = col->getCappedInsertNotifier();
     ASSERT_FALSE(notifier->isDead());
     notifier->kill();
@@ -74,7 +74,7 @@ TEST_F(CollectionTest, CappedNotifierTimeouts) {
     makeCapped(nss);
 
     AutoGetCollectionForRead acfr(operationContext(), nss);
-    Collection* col = acfr.getCollection();
+    const Collection* col = acfr.getCollection();
     auto notifier = col->getCappedInsertNotifier();
     ASSERT_EQ(notifier->getVersion(), 0u);
 
@@ -90,7 +90,7 @@ TEST_F(CollectionTest, CappedNotifierWaitAfterNotifyIsImmediate) {
     makeCapped(nss);
 
     AutoGetCollectionForRead acfr(operationContext(), nss);
-    Collection* col = acfr.getCollection();
+    const Collection* col = acfr.getCollection();
     auto notifier = col->getCappedInsertNotifier();
 
     auto prevVersion = notifier->getVersion();
@@ -109,7 +109,7 @@ TEST_F(CollectionTest, CappedNotifierWaitUntilAsynchronousNotifyAll) {
     makeCapped(nss);
 
     AutoGetCollectionForRead acfr(operationContext(), nss);
-    Collection* col = acfr.getCollection();
+    const Collection* col = acfr.getCollection();
     auto notifier = col->getCappedInsertNotifier();
     auto prevVersion = notifier->getVersion();
     auto thisVersion = prevVersion + 1;
@@ -134,7 +134,7 @@ TEST_F(CollectionTest, CappedNotifierWaitUntilAsynchronousKill) {
     makeCapped(nss);
 
     AutoGetCollectionForRead acfr(operationContext(), nss);
-    Collection* col = acfr.getCollection();
+    const Collection* col = acfr.getCollection();
     auto notifier = col->getCappedInsertNotifier();
     auto prevVersion = notifier->getVersion();
 
@@ -158,7 +158,7 @@ TEST_F(CollectionTest, HaveCappedWaiters) {
     makeCapped(nss);
 
     AutoGetCollectionForRead acfr(operationContext(), nss);
-    Collection* col = acfr.getCollection();
+    const Collection* col = acfr.getCollection();
     ASSERT_FALSE(col->getCappedCallback()->haveCappedWaiters());
     {
         auto notifier = col->getCappedInsertNotifier();
@@ -172,7 +172,7 @@ TEST_F(CollectionTest, NotifyCappedWaitersIfNeeded) {
     makeCapped(nss);
 
     AutoGetCollectionForRead acfr(operationContext(), nss);
-    Collection* col = acfr.getCollection();
+    const Collection* col = acfr.getCollection();
     col->getCappedCallback()->notifyCappedWaitersIfNeeded();
     {
         auto notifier = col->getCappedInsertNotifier();
@@ -187,7 +187,7 @@ TEST_F(CollectionTest, AsynchronouslyNotifyCappedWaitersIfNeeded) {
     makeCapped(nss);
 
     AutoGetCollectionForRead acfr(operationContext(), nss);
-    Collection* col = acfr.getCollection();
+    const Collection* col = acfr.getCollection();
     auto notifier = col->getCappedInsertNotifier();
     auto prevVersion = notifier->getVersion();
     auto thisVersion = prevVersion + 1;
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 25a5eea8819..f8347739529 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -678,7 +678,7 @@ Collection* DatabaseImpl::createCollection(OperationContext* opCtx,
         uassertStatusOK(storageEngine->getCatalog()->createCollection(
             opCtx, nss, optionsWithUUID, true /*allocateDefaultSpace*/));
     auto catalogId = catalogIdRecordStorePair.first;
-    std::unique_ptr<Collection> ownedCollection =
+    std::shared_ptr<Collection> ownedCollection =
         Collection::Factory::get(opCtx)->make(opCtx,
                                               nss,
                                               catalogId,
diff --git a/src/mongo/db/catalog/database_test.cpp b/src/mongo/db/catalog/database_test.cpp
index 70bb59651a3..0d871c358bb 100644
--- a/src/mongo/db/catalog/database_test.cpp
+++ b/src/mongo/db/catalog/database_test.cpp
@@ -503,7 +503,7 @@ TEST_F(DatabaseTest, AutoGetCollectionForReadCommandSucceedsWithDeadlineNow) {
     ASSERT(_opCtx.get()->lockState()->isCollectionLockedForMode(nss, MODE_X));
     try {
         AutoGetCollectionForReadCommand db(
-            _opCtx.get(), nss, AutoGetCollection::kViewsForbidden, Date_t::now());
+            _opCtx.get(), nss, AutoGetCollectionViewMode::kViewsForbidden, Date_t::now());
     } catch (const ExceptionFor<ErrorCodes::LockTimeout>&) {
         FAIL("Should get the db within the timeout");
     }
@@ -517,7 +517,7 @@ TEST_F(DatabaseTest, AutoGetCollectionForReadCommandSucceedsWithDeadlineMin) {
     ASSERT(_opCtx.get()->lockState()->isCollectionLockedForMode(nss, MODE_X));
     try {
         AutoGetCollectionForReadCommand db(
-            _opCtx.get(), nss, AutoGetCollection::kViewsForbidden, Date_t());
+            _opCtx.get(), nss, AutoGetCollectionViewMode::kViewsForbidden, Date_t());
     } catch (const ExceptionFor<ErrorCodes::LockTimeout>&) {
         FAIL("Should get the db within the timeout");
     }
diff --git a/src/mongo/db/catalog/list_indexes.cpp b/src/mongo/db/catalog/list_indexes.cpp
index ff395e72802..937598f30fd 100644
--- a/src/mongo/db/catalog/list_indexes.cpp
+++ b/src/mongo/db/catalog/list_indexes.cpp
@@ -53,7 +53,7 @@ StatusWith<std::list<BSONObj>> listIndexes(OperationContext* opCtx,
                                            const NamespaceStringOrUUID& ns,
                                            bool includeBuildUUIDs) {
     AutoGetCollectionForReadCommand ctx(opCtx, ns);
-    Collection* collection = ctx.getCollection();
+    const Collection* collection = ctx.getCollection();
     auto nss = ctx.getNss();
     if (!collection) {
         return StatusWith<std::list<BSONObj>>(ErrorCodes::NamespaceNotFound,
@@ -65,7 +65,7 @@ StatusWith<std::list<BSONObj>> listIndexes(OperationContext* opCtx,
 }
 
 std::list<BSONObj> listIndexesInLock(OperationContext* opCtx,
-                                     Collection* collection,
+                                     const Collection* collection,
                                      const NamespaceString& nss,
                                      bool includeBuildUUIDs) {
     invariant(opCtx->lockState()->isCollectionLockedForMode(nss, MODE_IS));
diff --git a/src/mongo/db/catalog/list_indexes.h b/src/mongo/db/catalog/list_indexes.h
index 62296826e9a..8ceeb2d019e 100644
--- a/src/mongo/db/catalog/list_indexes.h
+++ b/src/mongo/db/catalog/list_indexes.h
@@ -44,7 +44,7 @@ StatusWith<std::list<BSONObj>> listIndexes(OperationContext* opCtx,
                                            const NamespaceStringOrUUID& ns,
                                            bool includeBuildUUIDs);
 std::list<BSONObj> listIndexesInLock(OperationContext* opCtx,
-                                     Collection* collection,
+                                     const Collection* collection,
                                      const NamespaceString& nss,
                                      bool includeBuildUUIDs);
 std::list<BSONObj> listIndexesEmptyListIfMissing(OperationContext* opCtx,
diff --git a/src/mongo/db/catalog/rename_collection.cpp b/src/mongo/db/catalog/rename_collection.cpp
index bb57194d5f5..c0d13ed746c 100644
--- a/src/mongo/db/catalog/rename_collection.cpp
+++ b/src/mongo/db/catalog/rename_collection.cpp
@@ -897,7 +897,7 @@ Status renameCollectionForApplyOps(OperationContext* opCtx,
     }
 
     const Collection* const sourceColl =
-        AutoGetCollectionForRead(opCtx, sourceNss, AutoGetCollection::ViewMode::kViewsPermitted)
+        AutoGetCollectionForRead(opCtx, sourceNss, AutoGetCollectionViewMode::kViewsPermitted)
             .getCollection();
 
     if (sourceNss.isDropPendingNamespace() || sourceColl == nullptr) {
diff --git a/src/mongo/db/catalog/rename_collection_test.cpp b/src/mongo/db/catalog/rename_collection_test.cpp
index 3acd605af57..fc8beb3d713 100644
--- a/src/mongo/db/catalog/rename_collection_test.cpp
+++ b/src/mongo/db/catalog/rename_collection_test.cpp
@@ -1203,7 +1203,8 @@ TEST_F(RenameCollectionTest, CollectionPointerRemainsValidThroughRename) {
 
 TEST_F(RenameCollectionTest, CatalogPointersRenameValidThroughRenameForApplyOps) {
     _createCollection(_opCtx.get(), _sourceNss);
-    Collection* sourceColl = AutoGetCollectionForRead(_opCtx.get(), _sourceNss).getCollection();
+    const Collection* sourceColl =
+        AutoGetCollectionForRead(_opCtx.get(), _sourceNss).getCollection();
     ASSERT(sourceColl);
 
     auto uuid = UUID::gen();
@@ -1211,7 +1212,8 @@ TEST_F(RenameCollectionTest, CatalogPointersRenameValidThroughRenameForApplyOps)
     ASSERT_OK(renameCollectionForApplyOps(_opCtx.get(), _sourceNss.db().toString(), uuid, cmd, {}));
     ASSERT_FALSE(_collectionExists(_opCtx.get(), _sourceNss));
 
-    Collection* targetColl = AutoGetCollectionForRead(_opCtx.get(), _targetNss).getCollection();
+    const Collection* targetColl =
+        AutoGetCollectionForRead(_opCtx.get(), _targetNss).getCollection();
     ASSERT(targetColl);
     ASSERT_EQ(targetColl, sourceColl);
     ASSERT_EQ(targetColl->ns(), _targetNss);
diff --git a/src/mongo/db/catalog/uncommitted_collections.cpp b/src/mongo/db/catalog/uncommitted_collections.cpp
index ad16e34c6f3..4ab118f2fd9 100644
--- a/src/mongo/db/catalog/uncommitted_collections.cpp
+++ b/src/mongo/db/catalog/uncommitted_collections.cpp
@@ -46,7 +46,7 @@ UncommittedCollections& UncommittedCollections::get(OperationContext* opCtx) {
     return getUncommittedCollections(opCtx);
 }
 
-void UncommittedCollections::addToTxn(OperationContext* opCtx, std::unique_ptr<Collection> coll) {
+void UncommittedCollections::addToTxn(OperationContext* opCtx, std::shared_ptr<Collection> coll) {
     auto collList = getUncommittedCollections(opCtx).getResources().lock();
     auto existingColl = collList->_collections.find(coll->uuid());
     uassert(31370,
@@ -87,8 +87,8 @@ void UncommittedCollections::addToTxn(OperationContext* opCtx, std::unique_ptr<C
         });
 }
 
-Collection* UncommittedCollections::getForTxn(OperationContext* opCtx,
-                                              const NamespaceStringOrUUID& id) {
+std::shared_ptr<Collection> UncommittedCollections::getForTxn(OperationContext* opCtx,
+                                                              const NamespaceStringOrUUID& id) {
     if (id.nss()) {
         return getForTxn(opCtx, id.nss().get());
     } else {
@@ -96,24 +96,26 @@ Collection* UncommittedCollections::getForTxn(OperationContext* opCtx,
     }
 }
 
-Collection* UncommittedCollections::getForTxn(OperationContext* opCtx, const NamespaceString& nss) {
+std::shared_ptr<Collection> UncommittedCollections::getForTxn(OperationContext* opCtx,
+                                                              const NamespaceString& nss) {
     auto collList = getUncommittedCollections(opCtx).getResources().lock();
     auto it = collList->_nssIndex.find(nss);
     if (it == collList->_nssIndex.end()) {
         return nullptr;
     }
 
-    return collList->_collections[it->second].get();
+    return collList->_collections[it->second];
 }
 
-Collection* UncommittedCollections::getForTxn(OperationContext* opCtx, const UUID& uuid) {
+std::shared_ptr<Collection> UncommittedCollections::getForTxn(OperationContext* opCtx,
+                                                              const UUID& uuid) {
     auto collList = getUncommittedCollections(opCtx).getResources().lock();
     auto it = collList->_collections.find(uuid);
     if (it == collList->_collections.end()) {
         return nullptr;
     }
 
-    return it->second.get();
+    return it->second;
 }
 
 void UncommittedCollections::erase(UUID uuid, NamespaceString nss, UncommittedCollectionsMap* map) {
@@ -142,7 +144,7 @@ void UncommittedCollections::commit(OperationContext* opCtx,
     auto collPtr = it->second.get();
 
     auto nss = it->second->ns();
-    CollectionCatalog::get(opCtx).registerCollection(uuid, &(it->second));
+    CollectionCatalog::get(opCtx).registerCollection(uuid, it->second);
     map->_collections.erase(it);
     map->_nssIndex.erase(nss);
     auto svcCtx = opCtx->getServiceContext();
diff --git a/src/mongo/db/catalog/uncommitted_collections.h b/src/mongo/db/catalog/uncommitted_collections.h
index 3d4c2978063..5e860cc3911 100644
--- a/src/mongo/db/catalog/uncommitted_collections.h
+++ b/src/mongo/db/catalog/uncommitted_collections.h
@@ -56,7 +56,7 @@ public:
             _nssIndex.erase(nss);
         }
 
-        std::map<UUID, std::unique_ptr<Collection>> _collections;
+        std::map<UUID, std::shared_ptr<Collection>> _collections;
         std::map<NamespaceString, UUID> _nssIndex;
     };
 
@@ -80,11 +80,13 @@ public:
 
     static UncommittedCollections& get(OperationContext* opCtx);
 
-    static void addToTxn(OperationContext* opCtx, std::unique_ptr<Collection> coll);
+    static void addToTxn(OperationContext* opCtx, std::shared_ptr<Collection> coll);
 
-    static Collection* getForTxn(OperationContext* opCtx, const NamespaceStringOrUUID& nss);
-    static Collection* getForTxn(OperationContext* opCtx, const NamespaceString& nss);
-    static Collection* getForTxn(OperationContext* opCtx, const UUID& uuid);
+    static std::shared_ptr<Collection> getForTxn(OperationContext* opCtx,
+                                                 const NamespaceStringOrUUID& nss);
+    static std::shared_ptr<Collection> getForTxn(OperationContext* opCtx,
+                                                 const NamespaceString& nss);
+    static std::shared_ptr<Collection> getForTxn(OperationContext* opCtx, const UUID& uuid);
 
     /**
      * Registers any uncommitted collections with the CollectionCatalog. If registering a collection
diff --git a/src/mongo/db/catalog_raii.cpp b/src/mongo/db/catalog_raii.cpp
index dd8591814ff..867296068e0 100644
--- a/src/mongo/db/catalog_raii.cpp
+++ b/src/mongo/db/catalog_raii.cpp
@@ -69,11 +69,13 @@ Database* AutoGetDb::ensureDbExists() {
     return _db;
 }
 
-AutoGetCollection::AutoGetCollection(OperationContext* opCtx,
-                                     const NamespaceStringOrUUID& nsOrUUID,
-                                     LockMode modeColl,
-                                     ViewMode viewMode,
-                                     Date_t deadline)
+template <typename CatalogCollectionLookupT>
+AutoGetCollectionBase<CatalogCollectionLookupT>::AutoGetCollectionBase(
+    OperationContext* opCtx,
+    const NamespaceStringOrUUID& nsOrUUID,
+    LockMode modeColl,
+    AutoGetCollectionViewMode viewMode,
+    Date_t deadline)
     : _autoDb(opCtx,
               !nsOrUUID.dbname().empty() ? nsOrUUID.dbname() : nsOrUUID.nss()->db(),
               isSharedLockMode(modeColl) ? MODE_IS : MODE_IX,
@@ -110,7 +112,7 @@ AutoGetCollection::AutoGetCollection(OperationContext* opCtx,
     if (!db)
         return;
 
-    _coll = CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, _resolvedNss);
+    _coll = CatalogCollectionLookupT::lookupCollection(opCtx, _resolvedNss);
     invariant(!nsOrUUID.uuid() || _coll,
               str::stream() << "Collection for " << _resolvedNss.ns()
                             << " disappeared after successufully resolving "
@@ -144,7 +146,17 @@ AutoGetCollection::AutoGetCollection(OperationContext* opCtx,
     _view = ViewCatalog::get(db)->lookup(opCtx, _resolvedNss.ns());
     uassert(ErrorCodes::CommandNotSupportedOnView,
             str::stream() << "Namespace " << _resolvedNss.ns() << " is a view, not a collection",
-            !_view || viewMode == kViewsPermitted);
+            !_view || viewMode == AutoGetCollectionViewMode::kViewsPermitted);
+}
+
+CatalogCollectionLookup::CollectionStorage CatalogCollectionLookup::lookupCollection(
+    OperationContext* opCtx, const NamespaceString& nss) {
+    return CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, nss);
+}
+
+CatalogCollectionLookupForRead::CollectionStorage CatalogCollectionLookupForRead::lookupCollection(
+    OperationContext* opCtx, const NamespaceString& nss) {
+    return CollectionCatalog::get(opCtx).lookupCollectionByNamespaceForRead(opCtx, nss);
 }
 
 LockMode fixLockModeForSystemDotViewsChanges(const NamespaceString& nss, LockMode mode) {
@@ -205,4 +217,7 @@ AutoGetOplog::AutoGetOplog(OperationContext* opCtx, OplogAccessMode mode, Date_t
     _oplog = _oplogInfo->getCollection();
 }
 
+template class AutoGetCollectionBase<CatalogCollectionLookup>;
+template class AutoGetCollectionBase<CatalogCollectionLookupForRead>;
+
 }  // namespace mongo
diff --git a/src/mongo/db/catalog_raii.h b/src/mongo/db/catalog_raii.h
index 62f46f0ac1e..c8d5f017a3b 100644
--- a/src/mongo/db/catalog_raii.h
+++ b/src/mongo/db/catalog_raii.h
@@ -97,18 +97,24 @@ private:
  * Any acquired locks may be released when this object goes out of scope, therefore the database
  * and the collection references returned by this class should not be retained.
  */
-class AutoGetCollection {
-    AutoGetCollection(const AutoGetCollection&) = delete;
-    AutoGetCollection& operator=(const AutoGetCollection&) = delete;
 
-public:
-    enum ViewMode { kViewsPermitted, kViewsForbidden };
+enum class AutoGetCollectionViewMode { kViewsPermitted, kViewsForbidden };
 
-    AutoGetCollection(OperationContext* opCtx,
-                      const NamespaceStringOrUUID& nsOrUUID,
-                      LockMode modeColl,
-                      ViewMode viewMode = kViewsForbidden,
-                      Date_t deadline = Date_t::max());
+template <typename CatalogCollectionLookupT>
+class AutoGetCollectionBase {
+    AutoGetCollectionBase(const AutoGetCollectionBase&) = delete;
+    AutoGetCollectionBase& operator=(const AutoGetCollectionBase&) = delete;
+
+    using CollectionStorage = typename CatalogCollectionLookupT::CollectionStorage;
+    using CollectionPtr = typename CatalogCollectionLookupT::CollectionPtr;
+
+public:
+    AutoGetCollectionBase(
+        OperationContext* opCtx,
+        const NamespaceStringOrUUID& nsOrUUID,
+        LockMode modeColl,
+        AutoGetCollectionViewMode viewMode = AutoGetCollectionViewMode::kViewsForbidden,
+        Date_t deadline = Date_t::max());
 
     /**
      * Returns the database, or nullptr if it didn't exist.
@@ -127,8 +133,8 @@ public:
     /**
      * Returns nullptr if the collection didn't exist.
      */
-    Collection* getCollection() const {
-        return _coll;
+    CollectionPtr getCollection() const {
+        return CatalogCollectionLookupT::toCollectionPtr(_coll);
     }
 
     /**
@@ -156,10 +162,34 @@ private:
     // might need to be relocked for the correct namespace
     boost::optional<Lock::CollectionLock> _collLock;
 
-    Collection* _coll = nullptr;
+    CollectionStorage _coll = nullptr;
     std::shared_ptr<ViewDefinition> _view;
 };
 
+struct CatalogCollectionLookup {
+    using CollectionStorage = Collection*;
+    using CollectionPtr = Collection*;
+
+    static CollectionStorage lookupCollection(OperationContext* opCtx, const NamespaceString& nss);
+    static CollectionPtr toCollectionPtr(CollectionStorage collection) {
+        return collection;
+    }
+};
+struct CatalogCollectionLookupForRead {
+    using CollectionStorage = std::shared_ptr<const Collection>;
+    using CollectionPtr = const Collection*;
+
+    static CollectionStorage lookupCollection(OperationContext* opCtx, const NamespaceString& nss);
+    static CollectionPtr toCollectionPtr(const CollectionStorage& collection) {
+        return collection.get();
+    }
+};
+
+class AutoGetCollection : public AutoGetCollectionBase<CatalogCollectionLookup> {
+public:
+    using AutoGetCollectionBase::AutoGetCollectionBase;
+};
+
 /**
  * Writes to system.views need to use a stronger lock to prevent inconsistencies like view cycles.
  */
diff --git a/src/mongo/db/catalog_raii_test.cpp b/src/mongo/db/catalog_raii_test.cpp
index 733ed190250..cc222301ca0 100644
--- a/src/mongo/db/catalog_raii_test.cpp
+++ b/src/mongo/db/catalog_raii_test.cpp
@@ -138,7 +138,7 @@ TEST_F(CatalogRAIITestFixture, AutoGetCollectionCollLockDeadline) {
             AutoGetCollection coll(client2.second.get(),
                                    nss,
                                    MODE_X,
-                                   AutoGetCollection::ViewMode::kViewsForbidden,
+                                   AutoGetCollectionViewMode::kViewsForbidden,
                                    Date_t::now() + timeoutMs);
         },
         timeoutMs);
@@ -152,7 +152,7 @@ TEST_F(CatalogRAIITestFixture, AutoGetCollectionDBLockDeadline) {
             AutoGetCollection coll(client2.second.get(),
                                    nss,
                                    MODE_X,
-                                   AutoGetCollection::ViewMode::kViewsForbidden,
+                                   AutoGetCollectionViewMode::kViewsForbidden,
                                    Date_t::now() + timeoutMs);
         },
         timeoutMs);
@@ -166,7 +166,7 @@ TEST_F(CatalogRAIITestFixture, AutoGetCollectionGlobalLockDeadline) {
             AutoGetCollection coll(client2.second.get(),
                                    nss,
                                    MODE_X,
-                                   AutoGetCollection::ViewMode::kViewsForbidden,
+                                   AutoGetCollectionViewMode::kViewsForbidden,
                                    Date_t::now() + timeoutMs);
         },
         timeoutMs);
@@ -183,7 +183,7 @@ TEST_F(CatalogRAIITestFixture, AutoGetCollectionDeadlineNow) {
             AutoGetCollection coll(client2.second.get(),
                                    nss,
                                    MODE_X,
-                                   AutoGetCollection::ViewMode::kViewsForbidden,
+                                   AutoGetCollectionViewMode::kViewsForbidden,
                                    Date_t::now());
         },
         Milliseconds(0));
@@ -200,7 +200,7 @@ TEST_F(CatalogRAIITestFixture, AutoGetCollectionDeadlineMin) {
             AutoGetCollection coll(client2.second.get(),
                                    nss,
                                    MODE_X,
-                                   AutoGetCollection::ViewMode::kViewsForbidden,
+                                   AutoGetCollectionViewMode::kViewsForbidden,
                                    Date_t());
         },
         Milliseconds(0));
diff --git a/src/mongo/db/commands/count_cmd.cpp b/src/mongo/db/commands/count_cmd.cpp
index fe8c4af8198..45b51b5913e 100644
--- a/src/mongo/db/commands/count_cmd.cpp
+++ b/src/mongo/db/commands/count_cmd.cpp
@@ -138,7 +138,7 @@ public:
         boost::optional<AutoGetCollectionForReadCommand> ctx;
         ctx.emplace(opCtx,
                     CommandHelpers::parseNsCollectionRequired(dbname, cmdObj),
-                    AutoGetCollection::ViewMode::kViewsPermitted);
+                    AutoGetCollectionViewMode::kViewsPermitted);
         const auto nss = ctx->getNss();
 
         CountCommand request(NamespaceStringOrUUID(NamespaceString{}));
@@ -173,7 +173,7 @@ public:
                                 result);
         }
 
-        Collection* const collection = ctx->getCollection();
+        const Collection* const collection = ctx->getCollection();
 
         // Prevent chunks from being cleaned up during yields - this allows us to only check the
         // version on initial entry into count.
@@ -208,7 +208,7 @@ public:
         boost::optional<AutoGetCollectionForReadCommand> ctx;
         ctx.emplace(opCtx,
                     CommandHelpers::parseNsOrUUID(dbname, cmdObj),
-                    AutoGetCollection::ViewMode::kViewsPermitted);
+                    AutoGetCollectionViewMode::kViewsPermitted);
         const auto& nss = ctx->getNss();
 
         CurOpFailpointHelpers::waitWhileFailPointEnabled(
@@ -236,7 +236,7 @@ public:
             return true;
         }
 
-        Collection* const collection = ctx->getCollection();
+        const Collection* const collection = ctx->getCollection();
 
         // Prevent chunks from being cleaned up during yields - this allows us to only check the
         // version on initial entry into count.
diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index d1addd57762..e6d860afe38 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -468,7 +468,7 @@ public:
 
         const NamespaceString nss(ns);
         AutoGetCollectionForReadCommand ctx(opCtx, nss);
-        Collection* collection = ctx.getCollection();
+        const Collection* collection = ctx.getCollection();
 
         const auto collDesc =
             CollectionShardingState::get(opCtx, nss)->getCollectionDescription(opCtx);
diff --git a/src/mongo/db/commands/dbcommands_d.cpp b/src/mongo/db/commands/dbcommands_d.cpp
index de2bf3ed410..37512fdb9ac 100644
--- a/src/mongo/db/commands/dbcommands_d.cpp
+++ b/src/mongo/db/commands/dbcommands_d.cpp
@@ -276,7 +276,7 @@ public:
             // We drop and re-acquire these locks every document because md5'ing is expensive
             unique_ptr<AutoGetCollectionForReadCommand> ctx(
                 new AutoGetCollectionForReadCommand(opCtx, nss));
-            Collection* coll = ctx->getCollection();
+            const Collection* coll = ctx->getCollection();
 
             auto exec = uassertStatusOK(getExecutor(opCtx,
                                                     coll,
diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp
index fc8977182e8..4d3f78259e6 100644
--- a/src/mongo/db/commands/distinct.cpp
+++ b/src/mongo/db/commands/distinct.cpp
@@ -136,7 +136,7 @@ public:
         boost::optional<AutoGetCollectionForReadCommand> ctx;
         ctx.emplace(opCtx,
                     CommandHelpers::parseNsCollectionRequired(dbname, cmdObj),
-                    AutoGetCollection::ViewMode::kViewsPermitted);
+                    AutoGetCollectionViewMode::kViewsPermitted);
         const auto nss = ctx->getNss();
 
         const ExtensionsCallbackReal extensionsCallback(opCtx, &nss);
@@ -170,7 +170,7 @@ public:
                                 result);
         }
 
-        Collection* const collection = ctx->getCollection();
+        const Collection* const collection = ctx->getCollection();
 
         auto executor = uassertStatusOK(
             getExecutorDistinct(collection, QueryPlannerParams::DEFAULT, &parsedDistinct));
@@ -190,7 +190,7 @@ public:
         boost::optional<AutoGetCollectionForReadCommand> ctx;
         ctx.emplace(opCtx,
                     CommandHelpers::parseNsOrUUID(dbname, cmdObj),
-                    AutoGetCollection::ViewMode::kViewsPermitted);
+                    AutoGetCollectionViewMode::kViewsPermitted);
         const auto& nss = ctx->getNss();
 
         if (!ctx->getView()) {
@@ -231,7 +231,7 @@ public:
             return true;
         }
 
-        Collection* const collection = ctx->getCollection();
+        const Collection* const collection = ctx->getCollection();
 
         auto executor =
             getExecutorDistinct(collection, QueryPlannerParams::DEFAULT, &parsedDistinct);
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index cba4f1d224e..3f4d7d840df 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -236,7 +236,7 @@ public:
             boost::optional<AutoGetCollectionForReadCommand> ctx;
             ctx.emplace(opCtx,
                         CommandHelpers::parseNsCollectionRequired(_dbName, _request.body),
-                        AutoGetCollection::ViewMode::kViewsPermitted);
+                        AutoGetCollectionViewMode::kViewsPermitted);
             const auto nss = ctx->getNss();
 
             // Parse the command BSON to a QueryRequest.
@@ -285,7 +285,7 @@ public:
 
             // The collection may be NULL. If so, getExecutor() should handle it by returning an
             // execution tree with an EOFStage.
-            Collection* const collection = ctx->getCollection();
+            const Collection* const collection = ctx->getCollection();
 
             // Get the execution plan for the query.
             bool permitYield = true;
@@ -359,7 +359,7 @@ public:
             boost::optional<AutoGetCollectionForReadCommand> ctx;
             ctx.emplace(opCtx,
                         CommandHelpers::parseNsOrUUID(_dbName, _request.body),
-                        AutoGetCollection::ViewMode::kViewsPermitted);
+                        AutoGetCollectionViewMode::kViewsPermitted);
             const auto& nss = ctx->getNss();
 
             qr->refreshNSS(opCtx);
@@ -408,7 +408,7 @@ public:
                 return;
             }
 
-            Collection* const collection = ctx->getCollection();
+            const Collection* const collection = ctx->getCollection();
 
             if (cq->getQueryRequest().isReadOnce()) {
                 // The readOnce option causes any storage-layer cursors created during plan
diff --git a/src/mongo/db/commands/haystack.cpp b/src/mongo/db/commands/haystack.cpp
index e4af2078f3a..bd991dffa1e 100644
--- a/src/mongo/db/commands/haystack.cpp
+++ b/src/mongo/db/commands/haystack.cpp
@@ -110,7 +110,7 @@ public:
 
     void searchHaystack(const HaystackAccessMethod* ham,
                         OperationContext* opCtx,
-                        Collection* collection,
+                        const Collection* collection,
                         const BSONObj& nearObj,
                         double maxDistance,
                         const BSONObj& search,
@@ -233,7 +233,7 @@ public:
         uassertStatusOK(replCoord->checkCanServeReadsFor(
             opCtx, nss, ReadPreferenceSetting::get(opCtx).canRunOnSecondary()));
 
-        Collection* collection = ctx.getCollection();
+        const Collection* collection = ctx.getCollection();
         if (!collection) {
             errmsg = "can't find ns";
             return false;
diff --git a/src/mongo/db/commands/index_filter_commands.cpp b/src/mongo/db/commands/index_filter_commands.cpp
index 5bc77e30947..a7d6239eaab 100644
--- a/src/mongo/db/commands/index_filter_commands.cpp
+++ b/src/mongo/db/commands/index_filter_commands.cpp
@@ -65,7 +65,7 @@ using namespace mongo;
  * Retrieves a collection's query settings and plan cache from the database.
  */
 static Status getQuerySettingsAndPlanCache(OperationContext* opCtx,
-                                           Collection* collection,
+                                           const Collection* collection,
                                            const string& ns,
                                            QuerySettings** querySettingsOut,
                                            PlanCache** planCacheOut) {
diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp
index 6b3a51ad12e..e793ddf2f6f 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -155,7 +155,7 @@ public:
         {
             AutoGetCollectionForReadCommand ctx(opCtx,
                                                 CommandHelpers::parseNsOrUUID(dbname, cmdObj));
-            Collection* collection = ctx.getCollection();
+            const Collection* collection = ctx.getCollection();
             uassert(ErrorCodes::NamespaceNotFound,
                     str::stream() << "ns does not exist: " << ctx.getNss().ns(),
                     collection);
diff --git a/src/mongo/db/commands/map_reduce_agg.cpp b/src/mongo/db/commands/map_reduce_agg.cpp
index 4cbbb87dc41..2f5b656262e 100644
--- a/src/mongo/db/commands/map_reduce_agg.cpp
+++ b/src/mongo/db/commands/map_reduce_agg.cpp
@@ -62,7 +62,7 @@ auto makeExpressionContext(OperationContext* opCtx,
     // AutoGetCollectionForReadCommand will throw if the sharding version for this connection is
     // out of date.
     AutoGetCollectionForReadCommand ctx(
-        opCtx, parsedMr.getNamespace(), AutoGetCollection::ViewMode::kViewsPermitted);
+        opCtx, parsedMr.getNamespace(), AutoGetCollectionViewMode::kViewsPermitted);
     uassert(ErrorCodes::CommandNotSupportedOnView,
             "mapReduce on a view is not supported",
             !ctx.getView());
diff --git a/src/mongo/db/commands/plan_cache_clear_command.cpp b/src/mongo/db/commands/plan_cache_clear_command.cpp
index 490af16b497..61660438f04 100644
--- a/src/mongo/db/commands/plan_cache_clear_command.cpp
+++ b/src/mongo/db/commands/plan_cache_clear_command.cpp
@@ -47,7 +47,7 @@
 namespace mongo {
 namespace {
 
-PlanCache* getPlanCache(OperationContext* opCtx, Collection* collection) {
+PlanCache* getPlanCache(OperationContext* opCtx, const Collection* collection) {
     invariant(collection);
     PlanCache* planCache = CollectionQueryInfo::get(collection).getPlanCache();
     invariant(planCache);
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 3083dd36d4c..74c118f5b8f 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -533,7 +533,7 @@ Status runAggregate(OperationContext* opCtx,
                 PipelineD::resolveCollator(opCtx, request.getCollation(), nullptr));
 
             // Obtain collection locks on the execution namespace; that is, the oplog.
-            ctx.emplace(opCtx, nss, AutoGetCollection::ViewMode::kViewsForbidden);
+            ctx.emplace(opCtx, nss, AutoGetCollectionViewMode::kViewsForbidden);
         } else if (nss.isCollectionlessAggregateNS() && pipelineInvolvedNamespaces.empty()) {
             // If this is a collectionless agg with no foreign namespaces, don't acquire any locks.
             statsTracker.emplace(opCtx,
@@ -545,7 +545,7 @@ Status runAggregate(OperationContext* opCtx,
                 PipelineD::resolveCollator(opCtx, request.getCollation(), nullptr));
         } else {
             // This is a regular aggregation. Lock the collection or view.
-            ctx.emplace(opCtx, nss, AutoGetCollection::ViewMode::kViewsPermitted);
+            ctx.emplace(opCtx, nss, AutoGetCollectionViewMode::kViewsPermitted);
             collatorToUse.emplace(
                 PipelineD::resolveCollator(opCtx, request.getCollation(), ctx->getCollection()));
             if (ctx->getCollection()) {
@@ -553,7 +553,7 @@ Status runAggregate(OperationContext* opCtx,
             }
         }
 
-        Collection* collection = ctx ? ctx->getCollection() : nullptr;
+        const Collection* collection = ctx ? ctx->getCollection() : nullptr;
 
         // If this is a view, resolve it by finding the underlying collection and stitching view
         // pipelines and this request's pipeline together. We then release our locks before
@@ -759,7 +759,7 @@ Status runAggregate(OperationContext* opCtx,
         // For an optimized away pipeline, signal the cache that a query operation has completed.
         // For normal pipelines this is done in DocumentSourceCursor.
         if (ctx && ctx->getCollection()) {
-            Collection* coll = ctx->getCollection();
+            const Collection* coll = ctx->getCollection();
             CollectionQueryInfo::get(coll).notifyOfQuery(opCtx, coll, stats);
         }
     }
diff --git a/src/mongo/db/db_raii.cpp b/src/mongo/db/db_raii.cpp
index 6bc2f77d183..a8329f4641d 100644
--- a/src/mongo/db/db_raii.cpp
+++ b/src/mongo/db/db_raii.cpp
@@ -88,7 +88,7 @@ AutoStatsTracker::~AutoStatsTracker() {
 
 AutoGetCollectionForRead::AutoGetCollectionForRead(OperationContext* opCtx,
                                                    const NamespaceStringOrUUID& nsOrUUID,
-                                                   AutoGetCollection::ViewMode viewMode,
+                                                   AutoGetCollectionViewMode viewMode,
                                                    Date_t deadline) {
     // Don't take the ParallelBatchWriterMode lock when the server parameter is set and our
     // storage engine supports snapshot reads.
@@ -240,7 +240,7 @@ AutoGetCollectionForRead::AutoGetCollectionForRead(OperationContext* opCtx,
 AutoGetCollectionForReadCommand::AutoGetCollectionForReadCommand(
     OperationContext* opCtx,
     const NamespaceStringOrUUID& nsOrUUID,
-    AutoGetCollection::ViewMode viewMode,
+    AutoGetCollectionViewMode viewMode,
     Date_t deadline,
     AutoStatsTracker::LogMode logMode)
     : _autoCollForRead(opCtx, nsOrUUID, viewMode, deadline),
diff --git a/src/mongo/db/db_raii.h b/src/mongo/db/db_raii.h
index d1e9e85cb4e..e8a1157b9ff 100644
--- a/src/mongo/db/db_raii.h
+++ b/src/mongo/db/db_raii.h
@@ -104,14 +104,14 @@ public:
     AutoGetCollectionForRead(
         OperationContext* opCtx,
         const NamespaceStringOrUUID& nsOrUUID,
-        AutoGetCollection::ViewMode viewMode = AutoGetCollection::ViewMode::kViewsForbidden,
+        AutoGetCollectionViewMode viewMode = AutoGetCollectionViewMode::kViewsForbidden,
         Date_t deadline = Date_t::max());
 
     Database* getDb() const {
         return _autoColl->getDb();
     }
 
-    Collection* getCollection() const {
+    const Collection* getCollection() const {
         return _autoColl->getCollection();
     }
 
@@ -132,7 +132,7 @@ private:
 
     // This field is optional, because the code to wait for majority committed snapshot needs to
     // release locks in order to block waiting
-    boost::optional<AutoGetCollection> _autoColl;
+    boost::optional<AutoGetCollectionBase<CatalogCollectionLookupForRead>> _autoColl;
 };
 
 /**
@@ -147,7 +147,7 @@ public:
     AutoGetCollectionForReadCommand(
         OperationContext* opCtx,
         const NamespaceStringOrUUID& nsOrUUID,
-        AutoGetCollection::ViewMode viewMode = AutoGetCollection::ViewMode::kViewsForbidden,
+        AutoGetCollectionViewMode viewMode = AutoGetCollectionViewMode::kViewsForbidden,
         Date_t deadline = Date_t::max(),
         AutoStatsTracker::LogMode logMode = AutoStatsTracker::LogMode::kUpdateTopAndCurOp);
 
@@ -155,7 +155,7 @@ public:
         return _autoCollForRead.getDb();
     }
 
-    Collection* getCollection() const {
+    const Collection* getCollection() const {
         return _autoCollForRead.getCollection();
     }
 
diff --git a/src/mongo/db/db_raii_test.cpp b/src/mongo/db/db_raii_test.cpp
index 77874604c86..b101ce91961 100644
--- a/src/mongo/db/db_raii_test.cpp
+++ b/src/mongo/db/db_raii_test.cpp
@@ -67,8 +67,8 @@ public:
     const ClientAndCtx client2 = makeClientWithLocker("client2");
 };
 
-std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> makeTailableQueryPlan(OperationContext* opCtx,
-                                                                           Collection* collection) {
+std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> makeTailableQueryPlan(
+    OperationContext* opCtx, const Collection* collection) {
     auto qr = std::make_unique<QueryRequest>(collection->ns());
     qr->setTailableMode(TailableModeEnum::kTailableAndAwaitData);
 
@@ -114,7 +114,7 @@ TEST_F(DBRAIITestFixture, AutoGetCollectionForReadCollLockDeadline) {
         [&] {
             AutoGetCollectionForRead acfr(client2.second.get(),
                                           nss,
-                                          AutoGetCollection::ViewMode::kViewsForbidden,
+                                          AutoGetCollectionViewMode::kViewsForbidden,
                                           Date_t::now() + timeoutMs);
         },
         timeoutMs);
@@ -127,7 +127,7 @@ TEST_F(DBRAIITestFixture, AutoGetCollectionForReadDBLockDeadline) {
         [&] {
             AutoGetCollectionForRead coll(client2.second.get(),
                                           nss,
-                                          AutoGetCollection::ViewMode::kViewsForbidden,
+                                          AutoGetCollectionViewMode::kViewsForbidden,
                                           Date_t::now() + timeoutMs);
         },
         timeoutMs);
@@ -140,7 +140,7 @@ TEST_F(DBRAIITestFixture, AutoGetCollectionForReadGlobalLockDeadline) {
         [&] {
             AutoGetCollectionForRead coll(client2.second.get(),
                                           nss,
-                                          AutoGetCollection::ViewMode::kViewsForbidden,
+                                          AutoGetCollectionViewMode::kViewsForbidden,
                                           Date_t::now() + timeoutMs);
         },
         timeoutMs);
@@ -156,7 +156,7 @@ TEST_F(DBRAIITestFixture, AutoGetCollectionForReadDeadlineNow) {
         [&] {
             AutoGetCollectionForRead coll(client2.second.get(),
                                           nss,
-                                          AutoGetCollection::ViewMode::kViewsForbidden,
+                                          AutoGetCollectionViewMode::kViewsForbidden,
                                           Date_t::now());
         },
         Milliseconds(0));
@@ -171,7 +171,7 @@ TEST_F(DBRAIITestFixture, AutoGetCollectionForReadDeadlineMin) {
     failsWithLockTimeout(
         [&] {
             AutoGetCollectionForRead coll(
-                client2.second.get(), nss, AutoGetCollection::ViewMode::kViewsForbidden, Date_t());
+                client2.second.get(), nss, AutoGetCollectionViewMode::kViewsForbidden, Date_t());
         },
         Milliseconds(0));
 }
diff --git a/src/mongo/db/dbhelpers.cpp b/src/mongo/db/dbhelpers.cpp
index f6c1c27b4d6..bf19f4bd8bf 100644
--- a/src/mongo/db/dbhelpers.cpp
+++ b/src/mongo/db/dbhelpers.cpp
@@ -61,7 +61,7 @@ using std::unique_ptr;
    set your db SavedContext first
 */
 bool Helpers::findOne(OperationContext* opCtx,
-                      Collection* collection,
+                      const Collection* collection,
                       const BSONObj& query,
                       BSONObj& result,
                       bool requireIndex) {
@@ -76,7 +76,7 @@ bool Helpers::findOne(OperationContext* opCtx,
    set your db SavedContext first
 */
 RecordId Helpers::findOne(OperationContext* opCtx,
-                          Collection* collection,
+                          const Collection* collection,
                           const BSONObj& query,
                           bool requireIndex) {
     if (!collection)
@@ -88,7 +88,7 @@ RecordId Helpers::findOne(OperationContext* opCtx,
 }
 
 RecordId Helpers::findOne(OperationContext* opCtx,
-                          Collection* collection,
+                          const Collection* collection,
                           std::unique_ptr<QueryRequest> qr,
                           bool requireIndex) {
     if (!collection)
@@ -129,6 +129,7 @@ bool Helpers::findById(OperationContext* opCtx,
                        bool* indexFound) {
     invariant(database);
 
+    // TODO ForRead?
     Collection* collection =
         CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, NamespaceString(ns));
     if (!collection) {
@@ -155,10 +156,10 @@ bool Helpers::findById(OperationContext* opCtx,
 }
 
 RecordId Helpers::findById(OperationContext* opCtx,
-                           Collection* collection,
+                           const Collection* collection,
                            const BSONObj& idquery) {
     verify(collection);
-    IndexCatalog* catalog = collection->getIndexCatalog();
+    const IndexCatalog* catalog = collection->getIndexCatalog();
     const IndexDescriptor* desc = catalog->findIdIndex(opCtx);
     uassert(13430, "no _id index", desc);
     return catalog->getEntry(desc)->accessMethod()->findSingle(opCtx, idquery["_id"].wrap());
@@ -166,10 +167,10 @@ RecordId Helpers::findById(OperationContext* opCtx,
 
 // Acquires necessary locks to read the collection with the given namespace. If this is an oplog
 // read, use AutoGetOplog for simplified locking.
-Collection* getCollectionForRead(OperationContext* opCtx,
-                                 const NamespaceString& ns,
-                                 boost::optional<AutoGetCollectionForReadCommand>& autoColl,
-                                 boost::optional<AutoGetOplog>& autoOplog) {
+const Collection* getCollectionForRead(OperationContext* opCtx,
+                                       const NamespaceString& ns,
+                                       boost::optional<AutoGetCollectionForReadCommand>& autoColl,
+                                       boost::optional<AutoGetOplog>& autoOplog) {
     if (ns.isOplog()) {
         // Simplify locking rules for oplog collection.
         autoOplog.emplace(opCtx, OplogAccessMode::kRead);
diff --git a/src/mongo/db/dbhelpers.h b/src/mongo/db/dbhelpers.h
index 7b606768d02..514a4eda352 100644
--- a/src/mongo/db/dbhelpers.h
+++ b/src/mongo/db/dbhelpers.h
@@ -59,18 +59,18 @@ struct Helpers {
        @return true if object found
     */
     static bool findOne(OperationContext* opCtx,
-                        Collection* collection,
+                        const Collection* collection,
                         const BSONObj& query,
                         BSONObj& result,
                         bool requireIndex = false);
 
     static RecordId findOne(OperationContext* opCtx,
-                            Collection* collection,
+                            const Collection* collection,
                             const BSONObj& query,
                             bool requireIndex);
 
     static RecordId findOne(OperationContext* opCtx,
-                            Collection* collection,
+                            const Collection* collection,
                             std::unique_ptr<QueryRequest> qr,
                             bool requireIndex);
 
@@ -89,7 +89,9 @@ struct Helpers {
     /* TODO: should this move into Collection?
      * uasserts if no _id index.
      * @return null loc if not found */
-    static RecordId findById(OperationContext* opCtx, Collection* collection, const BSONObj& query);
+    static RecordId findById(OperationContext* opCtx,
+                             const Collection* collection,
+                             const BSONObj& query);
 
     /**
      * Get the first object generated from a forward natural-order scan on "ns".  Callers do not
diff --git a/src/mongo/db/exec/cached_plan.cpp b/src/mongo/db/exec/cached_plan.cpp
index 3933f1aa938..a281fb744e5 100644
--- a/src/mongo/db/exec/cached_plan.cpp
+++ b/src/mongo/db/exec/cached_plan.cpp
@@ -59,7 +59,7 @@ namespace mongo {
 const char* CachedPlanStage::kStageType = "CACHED_PLAN";
 
 CachedPlanStage::CachedPlanStage(ExpressionContext* expCtx,
-                                 Collection* collection,
+                                 const Collection* collection,
                                  WorkingSet* ws,
                                  CanonicalQuery* cq,
                                  const QueryPlannerParams& params,
diff --git a/src/mongo/db/exec/cached_plan.h b/src/mongo/db/exec/cached_plan.h
index a1f4ec6deb4..6719b965225 100644
--- a/src/mongo/db/exec/cached_plan.h
+++ b/src/mongo/db/exec/cached_plan.h
@@ -57,7 +57,7 @@ class PlanYieldPolicy;
 class CachedPlanStage final : public RequiresAllIndicesStage {
 public:
     CachedPlanStage(ExpressionContext* expCtx,
-                    Collection* collection,
+                    const Collection* collection,
                     WorkingSet* ws,
                     CanonicalQuery* cq,
                     const QueryPlannerParams& params,
diff --git a/src/mongo/db/exec/count.cpp b/src/mongo/db/exec/count.cpp
index 07158d20b08..e5f3604efc1 100644
--- a/src/mongo/db/exec/count.cpp
+++ b/src/mongo/db/exec/count.cpp
@@ -46,7 +46,7 @@ using std::vector;
 const char* CountStage::kStageType = "COUNT";
 
 CountStage::CountStage(ExpressionContext* expCtx,
-                       Collection* collection,
+                       const Collection* collection,
                        long long limit,
                        long long skip,
                        WorkingSet* ws,
diff --git a/src/mongo/db/exec/count.h b/src/mongo/db/exec/count.h
index 2eeaf5266e5..8c21f2e1cff 100644
--- a/src/mongo/db/exec/count.h
+++ b/src/mongo/db/exec/count.h
@@ -47,7 +47,7 @@ namespace mongo {
 class CountStage final : public PlanStage {
 public:
     CountStage(ExpressionContext* expCtx,
-               Collection* collection,
+               const Collection* collection,
                long long limit,
                long long skip,
                WorkingSet* ws,
diff --git a/src/mongo/db/exec/multi_iterator.cpp b/src/mongo/db/exec/multi_iterator.cpp
index 8a33f98386e..d9aee1eb09f 100644
--- a/src/mongo/db/exec/multi_iterator.cpp
+++ b/src/mongo/db/exec/multi_iterator.cpp
@@ -45,7 +45,7 @@ const char* MultiIteratorStage::kStageType = "MULTI_ITERATOR";
 
 MultiIteratorStage::MultiIteratorStage(ExpressionContext* expCtx,
                                        WorkingSet* ws,
-                                       Collection* collection)
+                                       const Collection* collection)
     : RequiresCollectionStage(kStageType, expCtx, collection), _ws(ws) {}
 
 void MultiIteratorStage::addIterator(unique_ptr<RecordCursor> it) {
diff --git a/src/mongo/db/exec/multi_iterator.h b/src/mongo/db/exec/multi_iterator.h
index accc0e4b1a6..038d480b7f3 100644
--- a/src/mongo/db/exec/multi_iterator.h
+++ b/src/mongo/db/exec/multi_iterator.h
@@ -47,7 +47,7 @@ namespace mongo {
  */
 class MultiIteratorStage final : public RequiresCollectionStage {
 public:
-    MultiIteratorStage(ExpressionContext* expCtx, WorkingSet* ws, Collection* collection);
+    MultiIteratorStage(ExpressionContext* expCtx, WorkingSet* ws, const Collection* collection);
 
     void addIterator(std::unique_ptr<RecordCursor> it);
 
diff --git a/src/mongo/db/exec/record_store_fast_count.cpp b/src/mongo/db/exec/record_store_fast_count.cpp
index e7142458cb3..c454a1af19f 100644
--- a/src/mongo/db/exec/record_store_fast_count.cpp
+++ b/src/mongo/db/exec/record_store_fast_count.cpp
@@ -36,7 +36,7 @@ namespace mongo {
 const char* RecordStoreFastCountStage::kStageType = "RECORD_STORE_FAST_COUNT";
 
 RecordStoreFastCountStage::RecordStoreFastCountStage(ExpressionContext* expCtx,
-                                                     Collection* collection,
+                                                     const Collection* collection,
                                                      long long skip,
                                                      long long limit)
     : RequiresCollectionStage(kStageType, expCtx, collection), _skip(skip), _limit(limit) {
diff --git a/src/mongo/db/exec/record_store_fast_count.h b/src/mongo/db/exec/record_store_fast_count.h
index 1986b333bb1..1a73829afa2 100644
--- a/src/mongo/db/exec/record_store_fast_count.h
+++ b/src/mongo/db/exec/record_store_fast_count.h
@@ -43,7 +43,7 @@ public:
     static const char* kStageType;
 
     RecordStoreFastCountStage(ExpressionContext* expCtx,
-                              Collection* collection,
+                              const Collection* collection,
                               long long skip,
                               long long limit);
 
diff --git a/src/mongo/db/index_build_entry_helpers.cpp b/src/mongo/db/index_build_entry_helpers.cpp
index 95cfcb7c6c5..f537d445003 100644
--- a/src/mongo/db/index_build_entry_helpers.cpp
+++ b/src/mongo/db/index_build_entry_helpers.cpp
@@ -256,7 +256,7 @@ StatusWith<IndexBuildEntry> getIndexBuildEntry(OperationContext* opCtx, UUID ind
     // Read the most up to date data.
     ReadSourceScope readSourceScope(opCtx, RecoveryUnit::ReadSource::kNoTimestamp);
     AutoGetCollectionForRead autoCollection(opCtx, NamespaceString::kIndexBuildEntryNamespace);
-    Collection* collection = autoCollection.getCollection();
+    const Collection* collection = autoCollection.getCollection();
 
     // Must not be interruptible. This fail point is used to test the scenario where the index
     // build's OperationContext is interrupted by an abort, which will subsequently remove index
diff --git a/src/mongo/db/index_builds_coordinator_mongod.cpp b/src/mongo/db/index_builds_coordinator_mongod.cpp
index c66627251b5..e6d7c2544e0 100644
--- a/src/mongo/db/index_builds_coordinator_mongod.cpp
+++ b/src/mongo/db/index_builds_coordinator_mongod.cpp
@@ -781,7 +781,7 @@ Status IndexBuildsCoordinatorMongod::setCommitQuorum(OperationContext* opCtx,
     }
 
     AutoGetCollectionForRead autoColl(opCtx, nss);
-    Collection* collection = autoColl.getCollection();
+    const Collection* collection = autoColl.getCollection();
     if (!collection) {
         return Status(ErrorCodes::NamespaceNotFound,
                       str::stream() << "Collection '" << nss << "' was not found.");
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 4f78a51f3b8..12c155b91b7 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_test.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream_test.cpp
@@ -451,8 +451,8 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAtOperationTimeAndResumeAfter
     auto expCtx = getExpCtx();
 
     // Need to put the collection in the collection catalog so the resume token is valid.
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
-    CollectionCatalog::get(expCtx->opCtx).registerCollection(testUuid(), &collection);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+    CollectionCatalog::get(expCtx->opCtx).registerCollection(testUuid(), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -471,9 +471,9 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAfterAndResumeAfterOptions) {
     auto opCtx = expCtx->opCtx;
 
     // Need to put the collection in the collection catalog so the resume token is validcollection
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
     auto& catalog = CollectionCatalog::get(opCtx);
-    catalog.registerCollection(testUuid(), &collection);
+    catalog.registerCollection(testUuid(), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -493,9 +493,9 @@ TEST_F(ChangeStreamStageTest, ShouldRejectBothStartAtOperationTimeAndStartAfterO
     auto opCtx = expCtx->opCtx;
 
     // Need to put the collection in the collection catalog so the resume token is valid.
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
     auto& catalog = CollectionCatalog::get(opCtx);
-    catalog.registerCollection(testUuid(), &collection);
+    catalog.registerCollection(testUuid(), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -514,9 +514,9 @@ TEST_F(ChangeStreamStageTest, ShouldRejectResumeAfterWithResumeTokenMissingUUID)
     auto opCtx = expCtx->opCtx;
 
     // Need to put the collection in the collection catalog so the resume token is valid.
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
     auto& catalog = CollectionCatalog::get(opCtx);
-    catalog.registerCollection(testUuid(), &collection);
+    catalog.registerCollection(testUuid(), std::move(collection));
 
     ASSERT_THROWS_CODE(
         DSChangeStream::createFromBson(
@@ -1548,8 +1548,8 @@ TEST_F(ChangeStreamStageTest, DocumentKeyShouldIncludeShardKeyFromResumeToken) {
     const auto opTime = repl::OpTime(ts, term);
     const auto uuid = testUuid();
 
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, &collection);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -1593,8 +1593,8 @@ TEST_F(ChangeStreamStageTest, DocumentKeyShouldNotIncludeShardKeyFieldsIfNotPres
     const auto opTime = repl::OpTime(ts, term);
     const auto uuid = testUuid();
 
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, &collection);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -1635,8 +1635,8 @@ TEST_F(ChangeStreamStageTest, ResumeAfterFailsIfResumeTokenDoesNotContainUUID) {
     const Timestamp ts(3, 45);
     const auto uuid = testUuid();
 
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, &collection);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     // Create a resume token from only the timestamp.
     auto resumeToken = makeResumeToken(ts);
@@ -1688,8 +1688,8 @@ TEST_F(ChangeStreamStageTest, ResumeAfterWithTokenFromInvalidateShouldFail) {
     auto expCtx = getExpCtx();
 
     // Need to put the collection in the collection catalog so the resume token is valid.
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
-    CollectionCatalog::get(expCtx->opCtx).registerCollection(testUuid(), &collection);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+    CollectionCatalog::get(expCtx->opCtx).registerCollection(testUuid(), std::move(collection));
 
     const auto resumeTokenInvalidate =
         makeResumeToken(kDefaultTs,
@@ -2348,8 +2348,8 @@ TEST_F(ChangeStreamStageDBTest, DocumentKeyShouldIncludeShardKeyFromResumeToken)
     const auto opTime = repl::OpTime(ts, term);
     const auto uuid = testUuid();
 
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, &collection);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -2384,8 +2384,8 @@ TEST_F(ChangeStreamStageDBTest, DocumentKeyShouldNotIncludeShardKeyFieldsIfNotPr
     const auto opTime = repl::OpTime(ts, term);
     const auto uuid = testUuid();
 
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, &collection);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     BSONObj o2 = BSON("_id" << 1 << "shardKey" << 2);
     auto resumeToken = makeResumeToken(ts, uuid, o2);
@@ -2421,8 +2421,8 @@ TEST_F(ChangeStreamStageDBTest, DocumentKeyShouldNotIncludeShardKeyIfResumeToken
     const auto opTime = repl::OpTime(ts, term);
     const auto uuid = testUuid();
 
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, &collection);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     // Create a resume token from only the timestamp.
     auto resumeToken = makeResumeToken(ts);
@@ -2457,8 +2457,8 @@ TEST_F(ChangeStreamStageDBTest, ResumeAfterWithTokenFromInvalidateShouldFail) {
     auto expCtx = getExpCtx();
 
     // Need to put the collection in the collection catalog so the resume token is valid.
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
-    CollectionCatalog::get(expCtx->opCtx).registerCollection(testUuid(), &collection);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+    CollectionCatalog::get(expCtx->opCtx).registerCollection(testUuid(), std::move(collection));
 
     const auto resumeTokenInvalidate =
         makeResumeToken(kDefaultTs,
@@ -2478,8 +2478,8 @@ TEST_F(ChangeStreamStageDBTest, ResumeAfterWithTokenFromInvalidateShouldFail) {
 TEST_F(ChangeStreamStageDBTest, ResumeAfterWithTokenFromDropDatabase) {
     const auto uuid = testUuid();
 
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, &collection);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, std::move(collection));
 
     // Create a resume token from only the timestamp, similar to a 'dropDatabase' entry.
     auto resumeToken = makeResumeToken(
@@ -2507,8 +2507,8 @@ TEST_F(ChangeStreamStageDBTest, ResumeAfterWithTokenFromDropDatabase) {
 TEST_F(ChangeStreamStageDBTest, StartAfterSucceedsEvenIfResumeTokenDoesNotContainUUID) {
     const auto uuid = testUuid();
 
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
-    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, &collection);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
+    CollectionCatalog::get(getExpCtx()->opCtx).registerCollection(uuid, 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/pipeline/document_source_cursor.cpp b/src/mongo/db/pipeline/document_source_cursor.cpp
index 7c530ef0663..ba6151d341a 100644
--- a/src/mongo/db/pipeline/document_source_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_cursor.cpp
@@ -285,7 +285,7 @@ DocumentSourceCursor::~DocumentSourceCursor() {
 }
 
 DocumentSourceCursor::DocumentSourceCursor(
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
     const intrusive_ptr<ExpressionContext>& pCtx,
     CursorType cursorType,
@@ -316,7 +316,7 @@ DocumentSourceCursor::DocumentSourceCursor(
 }
 
 intrusive_ptr<DocumentSourceCursor> DocumentSourceCursor::create(
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
     const intrusive_ptr<ExpressionContext>& pExpCtx,
     CursorType cursorType,
diff --git a/src/mongo/db/pipeline/document_source_cursor.h b/src/mongo/db/pipeline/document_source_cursor.h
index 0418f60b877..a7794df2e2c 100644
--- a/src/mongo/db/pipeline/document_source_cursor.h
+++ b/src/mongo/db/pipeline/document_source_cursor.h
@@ -63,7 +63,7 @@ public:
      * $cursor stage can return a sequence of empty documents for the caller to count.
      */
     static boost::intrusive_ptr<DocumentSourceCursor> create(
-        Collection* collection,
+        const Collection* collection,
         std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
         const boost::intrusive_ptr<ExpressionContext>& pExpCtx,
         CursorType cursorType,
@@ -112,7 +112,7 @@ public:
     }
 
 protected:
-    DocumentSourceCursor(Collection* collection,
+    DocumentSourceCursor(const Collection* collection,
                          std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
                          const boost::intrusive_ptr<ExpressionContext>& pExpCtx,
                          CursorType cursorType,
diff --git a/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp b/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp
index 1e5907a00b8..4277bd26424 100644
--- a/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp
@@ -51,7 +51,7 @@
 namespace mongo {
 
 boost::intrusive_ptr<DocumentSourceGeoNearCursor> DocumentSourceGeoNearCursor::create(
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     FieldPath distanceField,
@@ -66,7 +66,7 @@ boost::intrusive_ptr<DocumentSourceGeoNearCursor> DocumentSourceGeoNearCursor::c
 }
 
 DocumentSourceGeoNearCursor::DocumentSourceGeoNearCursor(
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     FieldPath distanceField,
diff --git a/src/mongo/db/pipeline/document_source_geo_near_cursor.h b/src/mongo/db/pipeline/document_source_geo_near_cursor.h
index 084e8b76bbe..f8d3b483914 100644
--- a/src/mongo/db/pipeline/document_source_geo_near_cursor.h
+++ b/src/mongo/db/pipeline/document_source_geo_near_cursor.h
@@ -60,7 +60,7 @@ public:
      * nonnegative.
      */
     static boost::intrusive_ptr<DocumentSourceGeoNearCursor> create(
-        Collection*,
+        const Collection*,
         std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>,
         const boost::intrusive_ptr<ExpressionContext>&,
         FieldPath distanceField,
@@ -70,7 +70,7 @@ public:
     const char* getSourceName() const final;
 
 private:
-    DocumentSourceGeoNearCursor(Collection*,
+    DocumentSourceGeoNearCursor(const Collection*,
                                 std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>,
                                 const boost::intrusive_ptr<ExpressionContext>&,
                                 FieldPath distanceField,
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index e436fc29334..adec324007d 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -105,7 +105,7 @@ namespace {
  * percentage of the collection.
  */
 StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createRandomCursorExecutor(
-    Collection* coll,
+    const Collection* coll,
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     long long sampleSize,
     long long numRecords) {
@@ -176,7 +176,7 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createRandomCursorEx
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> attemptToGetExecutor(
     const intrusive_ptr<ExpressionContext>& expCtx,
-    Collection* collection,
+    const Collection* collection,
     const NamespaceString& nss,
     BSONObj queryObj,
     BSONObj projectionObj,
@@ -262,7 +262,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> attemptToGetExe
  *
  * The 'collection' is required to exist. Throws if no usable 2d or 2dsphere index could be found.
  */
-StringData extractGeoNearFieldFromIndexes(OperationContext* opCtx, Collection* collection) {
+StringData extractGeoNearFieldFromIndexes(OperationContext* opCtx, const Collection* collection) {
     invariant(collection);
 
     std::vector<const IndexDescriptor*> idxs;
@@ -302,7 +302,7 @@ StringData extractGeoNearFieldFromIndexes(OperationContext* opCtx, Collection* c
 }  // namespace
 
 std::pair<PipelineD::AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-PipelineD::buildInnerQueryExecutor(Collection* collection,
+PipelineD::buildInnerQueryExecutor(const Collection* collection,
                                    const NamespaceString& nss,
                                    const AggregationRequest* aggRequest,
                                    Pipeline* pipeline) {
@@ -352,7 +352,7 @@ PipelineD::buildInnerQueryExecutor(Collection* collection,
                     ? DocumentSourceCursor::CursorType::kEmptyDocuments
                     : DocumentSourceCursor::CursorType::kRegular;
                 auto attachExecutorCallback =
-                    [cursorType](Collection* collection,
+                    [cursorType](const Collection* collection,
                                  std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
                                  Pipeline* pipeline) {
                         auto cursor = DocumentSourceCursor::create(
@@ -376,7 +376,7 @@ PipelineD::buildInnerQueryExecutor(Collection* collection,
 }
 
 void PipelineD::attachInnerQueryExecutorToPipeline(
-    Collection* collection,
+    const Collection* collection,
     PipelineD::AttachExecutorCallback attachExecutorCallback,
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
     Pipeline* pipeline) {
@@ -388,7 +388,7 @@ void PipelineD::attachInnerQueryExecutorToPipeline(
     }
 }
 
-void PipelineD::buildAndAttachInnerQueryExecutorToPipeline(Collection* collection,
+void PipelineD::buildAndAttachInnerQueryExecutorToPipeline(const Collection* collection,
                                                            const NamespaceString& nss,
                                                            const AggregationRequest* aggRequest,
                                                            Pipeline* pipeline) {
@@ -487,7 +487,7 @@ auto buildProjectionForPushdown(const DepsTracker& deps, Pipeline* pipeline) {
 }  // namespace
 
 std::pair<PipelineD::AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-PipelineD::buildInnerQueryExecutorGeneric(Collection* collection,
+PipelineD::buildInnerQueryExecutorGeneric(const Collection* collection,
                                           const NamespaceString& nss,
                                           const AggregationRequest* aggRequest,
                                           Pipeline* pipeline) {
@@ -565,7 +565,7 @@ PipelineD::buildInnerQueryExecutorGeneric(Collection* collection,
         (pipeline->peekFront() && pipeline->peekFront()->constraints().isChangeStreamStage());
 
     auto attachExecutorCallback =
-        [cursorType, trackOplogTS](Collection* collection,
+        [cursorType, trackOplogTS](const Collection* collection,
                                    std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
                                    Pipeline* pipeline) {
             auto cursor = DocumentSourceCursor::create(
@@ -576,7 +576,7 @@ PipelineD::buildInnerQueryExecutorGeneric(Collection* collection,
 }
 
 std::pair<PipelineD::AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-PipelineD::buildInnerQueryExecutorGeoNear(Collection* collection,
+PipelineD::buildInnerQueryExecutorGeoNear(const Collection* collection,
                                           const NamespaceString& nss,
                                           const AggregationRequest* aggRequest,
                                           Pipeline* pipeline) {
@@ -620,7 +620,7 @@ PipelineD::buildInnerQueryExecutorGeoNear(Collection* collection,
                                    locationField = geoNearStage->getLocationField(),
                                    distanceMultiplier =
                                        geoNearStage->getDistanceMultiplier().value_or(1.0)](
-                                      Collection* collection,
+                                      const Collection* collection,
                                       std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
                                       Pipeline* pipeline) {
         auto cursor = DocumentSourceGeoNearCursor::create(collection,
@@ -638,7 +638,7 @@ PipelineD::buildInnerQueryExecutorGeoNear(Collection* collection,
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prepareExecutor(
     const intrusive_ptr<ExpressionContext>& expCtx,
-    Collection* collection,
+    const Collection* collection,
     const NamespaceString& nss,
     Pipeline* pipeline,
     const boost::intrusive_ptr<DocumentSourceSort>& sortStage,
diff --git a/src/mongo/db/pipeline/pipeline_d.h b/src/mongo/db/pipeline/pipeline_d.h
index 78f3fedb1bd..ec1baacf6fd 100644
--- a/src/mongo/db/pipeline/pipeline_d.h
+++ b/src/mongo/db/pipeline/pipeline_d.h
@@ -67,7 +67,7 @@ public:
      * the new stage to the pipeline.
      */
     using AttachExecutorCallback = std::function<void(
-        Collection*, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>, Pipeline*)>;
+        const Collection*, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>, Pipeline*)>;
 
     /**
      * This method looks for early pipeline stages that can be folded into the underlying
@@ -88,7 +88,7 @@ public:
      * 'nullptr'.
      */
     static std::pair<AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-    buildInnerQueryExecutor(Collection* collection,
+    buildInnerQueryExecutor(const Collection* collection,
                             const NamespaceString& nss,
                             const AggregationRequest* aggRequest,
                             Pipeline* pipeline);
@@ -101,7 +101,7 @@ public:
      * 'nullptr'.
      */
     static void attachInnerQueryExecutorToPipeline(
-        Collection* collection,
+        const Collection* collection,
         AttachExecutorCallback attachExecutorCallback,
         std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
         Pipeline* pipeline);
@@ -112,7 +112,7 @@ public:
      * used when the executor attachment phase doesn't need to be deferred and the $cursor stage
      * can be created right after buiding the executor.
      */
-    static void buildAndAttachInnerQueryExecutorToPipeline(Collection* collection,
+    static void buildAndAttachInnerQueryExecutorToPipeline(const Collection* collection,
                                                            const NamespaceString& nss,
                                                            const AggregationRequest* aggRequest,
                                                            Pipeline* pipeline);
@@ -149,7 +149,7 @@ private:
      * the 'pipeline'.
      */
     static std::pair<AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-    buildInnerQueryExecutorGeneric(Collection* collection,
+    buildInnerQueryExecutorGeneric(const Collection* collection,
                                    const NamespaceString& nss,
                                    const AggregationRequest* aggRequest,
                                    Pipeline* pipeline);
@@ -160,7 +160,7 @@ private:
      * not exist, as the $geoNearCursor requires a 2d or 2dsphere index.
      */
     static std::pair<AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-    buildInnerQueryExecutorGeoNear(Collection* collection,
+    buildInnerQueryExecutorGeoNear(const Collection* collection,
                                    const NamespaceString& nss,
                                    const AggregationRequest* aggRequest,
                                    Pipeline* pipeline);
@@ -179,7 +179,7 @@ private:
      */
     static StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> prepareExecutor(
         const boost::intrusive_ptr<ExpressionContext>& expCtx,
-        Collection* collection,
+        const Collection* collection,
         const NamespaceString& nss,
         Pipeline* pipeline,
         const boost::intrusive_ptr<DocumentSourceSort>& sortStage,
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 6c69dce204f..f2d9082f7f2 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
@@ -157,7 +157,7 @@ std::vector<Document> CommonMongodProcessInterface::getIndexStats(OperationConte
                                                                   bool addShardName) {
     AutoGetCollectionForReadCommand autoColl(opCtx, ns);
 
-    Collection* collection = autoColl.getCollection();
+    const Collection* collection = autoColl.getCollection();
     std::vector<Document> indexStats;
     if (!collection) {
         LOGV2_DEBUG(23881,
@@ -233,7 +233,7 @@ Status CommonMongodProcessInterface::appendQueryExecStats(OperationContext* opCt
                 str::stream() << "Database [" << nss.db().toString() << "] not found."};
     }
 
-    Collection* collection = autoColl.getCollection();
+    const Collection* collection = autoColl.getCollection();
 
     if (!collection) {
         return {ErrorCodes::NamespaceNotFound,
@@ -265,7 +265,7 @@ BSONObj CommonMongodProcessInterface::getCollectionOptions(OperationContext* opC
     if (!autoColl.getDb()) {
         return collectionOptions;
     }
-    Collection* collection = autoColl.getCollection();
+    const Collection* collection = autoColl.getCollection();
     if (!collection) {
         return collectionOptions;
     }
@@ -291,7 +291,7 @@ CommonMongodProcessInterface::attachCursorSourceToPipelineForLocalRead(Pipeline*
         : expCtx->ns;
     autoColl.emplace(expCtx->opCtx,
                      nsOrUUID,
-                     AutoGetCollection::ViewMode::kViewsForbidden,
+                     AutoGetCollectionViewMode::kViewsForbidden,
                      Date_t::max(),
                      AutoStatsTracker::LogMode::kUpdateTop);
 
diff --git a/src/mongo/db/query/collection_query_info.cpp b/src/mongo/db/query/collection_query_info.cpp
index 4440359ce35..43d67831ecf 100644
--- a/src/mongo/db/query/collection_query_info.cpp
+++ b/src/mongo/db/query/collection_query_info.cpp
@@ -160,8 +160,8 @@ void CollectionQueryInfo::computeIndexKeys(OperationContext* opCtx, Collection*
 }
 
 void CollectionQueryInfo::notifyOfQuery(OperationContext* opCtx,
-                                        Collection* coll,
-                                        const PlanSummaryStats& summaryStats) {
+                                        const Collection* coll,
+                                        const PlanSummaryStats& summaryStats) const {
     auto& collectionIndexUsageTracker =
         CollectionIndexUsageTrackerDecoration::get(coll->getSharedDecorations());
 
diff --git a/src/mongo/db/query/collection_query_info.h b/src/mongo/db/query/collection_query_info.h
index 1a99c1b6fdf..ac3a0aa2bee 100644
--- a/src/mongo/db/query/collection_query_info.h
+++ b/src/mongo/db/query/collection_query_info.h
@@ -87,8 +87,8 @@ public:
     void clearQueryCache(const Collection* coll);
 
     void notifyOfQuery(OperationContext* opCtx,
-                       Collection* coll,
-                       const PlanSummaryStats& summaryStats);
+                       const Collection* coll,
+                       const PlanSummaryStats& summaryStats) const;
 
 private:
     void computeIndexKeys(OperationContext* opCtx, Collection* coll);
diff --git a/src/mongo/db/query/find.cpp b/src/mongo/db/query/find.cpp
index cc317eadf3a..90eddfe4cd1 100644
--- a/src/mongo/db/query/find.cpp
+++ b/src/mongo/db/query/find.cpp
@@ -120,7 +120,7 @@ void beginQueryOp(OperationContext* opCtx,
 }
 
 void endQueryOp(OperationContext* opCtx,
-                Collection* collection,
+                const Collection* collection,
                 const PlanExecutor& exec,
                 long long numResults,
                 CursorId cursorId) {
@@ -613,8 +613,8 @@ bool runQuery(OperationContext* opCtx,
                 "query"_attr = redact(cq->toStringShort()));
 
     // Parse, canonicalize, plan, transcribe, and get a plan executor.
-    AutoGetCollectionForReadCommand ctx(opCtx, nss, AutoGetCollection::ViewMode::kViewsForbidden);
-    Collection* const collection = ctx.getCollection();
+    AutoGetCollectionForReadCommand ctx(opCtx, nss, AutoGetCollectionViewMode::kViewsForbidden);
+    const Collection* const collection = ctx.getCollection();
     const QueryRequest& qr = cq->getQueryRequest();
 
     opCtx->setExhaust(qr.isExhaust());
diff --git a/src/mongo/db/query/find.h b/src/mongo/db/query/find.h
index 0349bf60931..805fd18884d 100644
--- a/src/mongo/db/query/find.h
+++ b/src/mongo/db/query/find.h
@@ -79,7 +79,7 @@ void beginQueryOp(OperationContext* opCtx,
  * Uses explain functionality to extract stats from 'exec'.
  */
 void endQueryOp(OperationContext* opCtx,
-                Collection* collection,
+                const Collection* collection,
                 const PlanExecutor& exec,
                 long long numResults,
                 CursorId cursorId);
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index 3f14abe3645..3ff9dd31e55 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -145,7 +145,7 @@ bool turnIxscanIntoCount(QuerySolution* soln);
 /**
  * Returns 'true' if 'query' on the given 'collection' can be answered using a special IDHACK plan.
  */
-bool isIdHackEligibleQuery(Collection* collection, const CanonicalQuery& query) {
+bool isIdHackEligibleQuery(const Collection* collection, const CanonicalQuery& query) {
     return !query.getQueryRequest().showRecordId() && query.getQueryRequest().getHint().isEmpty() &&
         query.getQueryRequest().getMin().isEmpty() && query.getQueryRequest().getMax().isEmpty() &&
         !query.getQueryRequest().getSkip() &&
@@ -249,7 +249,7 @@ IndexEntry indexEntryFromIndexCatalogEntry(OperationContext* opCtx,
  * If query supports index filters, filter params.indices according to any index filters that have
  * been configured. In addition, sets that there were indeed index filters applied.
  */
-void applyIndexFilters(Collection* collection,
+void applyIndexFilters(const Collection* collection,
                        const CanonicalQuery& canonicalQuery,
                        QueryPlannerParams* plannerParams) {
     if (!isIdHackEligibleQuery(collection, canonicalQuery)) {
@@ -268,7 +268,7 @@ void applyIndexFilters(Collection* collection,
 }
 
 void fillOutPlannerParams(OperationContext* opCtx,
-                          Collection* collection,
+                          const Collection* collection,
                           CanonicalQuery* canonicalQuery,
                           QueryPlannerParams* plannerParams) {
     invariant(canonicalQuery);
@@ -512,7 +512,7 @@ template <typename PlanStageType, typename ResultType>
 class PrepareExecutionHelper {
 public:
     PrepareExecutionHelper(OperationContext* opCtx,
-                           Collection* collection,
+                           const Collection* collection,
                            CanonicalQuery* cq,
                            PlanYieldPolicy* yieldPolicy,
                            size_t plannerOptions)
@@ -725,7 +725,7 @@ protected:
         const QueryPlannerParams& plannerParams) = 0;
 
     OperationContext* _opCtx;
-    Collection* _collection;
+    const Collection* _collection;
     CanonicalQuery* _cq;
     PlanYieldPolicy* _yieldPolicy;
     const size_t _plannerOptions;
@@ -738,7 +738,7 @@ class ClassicPrepareExecutionHelper final
     : public PrepareExecutionHelper<std::unique_ptr<PlanStage>, ClassicPrepareExecutionResult> {
 public:
     ClassicPrepareExecutionHelper(OperationContext* opCtx,
-                                  Collection* collection,
+                                  const Collection* collection,
                                   WorkingSet* ws,
                                   CanonicalQuery* cq,
                                   PlanYieldPolicy* yieldPolicy,
@@ -957,7 +957,7 @@ private:
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getClassicExecutor(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     size_t plannerOptions) {
@@ -989,7 +989,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getClassicExecu
  */
 std::unique_ptr<sbe::RuntimePlanner> makeRuntimePlannerIfNeeded(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     CanonicalQuery* canonicalQuery,
     size_t numSolutions,
     boost::optional<size_t> decisionWorks,
@@ -1039,7 +1039,7 @@ std::unique_ptr<sbe::RuntimePlanner> makeRuntimePlannerIfNeeded(
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExecutor(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<CanonicalQuery> cq,
     PlanYieldPolicy::YieldPolicy requestedYieldPolicy,
     size_t plannerOptions) {
@@ -1086,7 +1086,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExe
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     size_t plannerOptions) {
@@ -1105,7 +1105,7 @@ namespace {
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> _getExecutorFind(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     size_t plannerOptions) {
@@ -1120,7 +1120,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> _getExecutorFin
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     bool permitYield,
     size_t plannerOptions) {
@@ -1133,7 +1133,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorLegacyFind(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery) {
     return _getExecutorFind(opCtx,
                             collection,
@@ -1658,7 +1658,7 @@ bool getDistinctNodeIndex(const std::vector<IndexEntry>& indices,
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorCount(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    Collection* collection,
+    const Collection* collection,
     const CountCommand& request,
     bool explain,
     const NamespaceString& nss) {
@@ -1925,7 +1925,7 @@ namespace {
 
 // Get the list of indexes that include the "distinct" field.
 QueryPlannerParams fillOutPlannerParamsForDistinct(OperationContext* opCtx,
-                                                   Collection* collection,
+                                                   const Collection* collection,
                                                    size_t plannerOptions,
                                                    const ParsedDistinct& parsedDistinct) {
     QueryPlannerParams plannerParams;
@@ -2009,7 +2009,7 @@ QueryPlannerParams fillOutPlannerParamsForDistinct(OperationContext* opCtx,
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorForSimpleDistinct(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     const QueryPlannerParams& plannerParams,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     ParsedDistinct* parsedDistinct) {
@@ -2083,7 +2083,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorForS
 // 'strictDistinctOnly' parameter.
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
 getExecutorDistinctFromIndexSolutions(OperationContext* opCtx,
-                                      Collection* collection,
+                                      const Collection* collection,
                                       std::vector<std::unique_ptr<QuerySolution>> solutions,
                                       PlanYieldPolicy::YieldPolicy yieldPolicy,
                                       ParsedDistinct* parsedDistinct,
@@ -2124,7 +2124,7 @@ getExecutorDistinctFromIndexSolutions(OperationContext* opCtx,
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorWithoutProjection(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     const CanonicalQuery* cq,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     size_t plannerOptions) {
@@ -2146,7 +2146,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorWith
 }  // namespace
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDistinct(
-    Collection* collection, size_t plannerOptions, ParsedDistinct* parsedDistinct) {
+    const Collection* collection, size_t plannerOptions, ParsedDistinct* parsedDistinct) {
     auto expCtx = parsedDistinct->getQuery()->getExpCtx();
     OperationContext* opCtx = expCtx->opCtx;
     const auto yieldPolicy = opCtx->inMultiDocumentTransaction()
diff --git a/src/mongo/db/query/get_executor.h b/src/mongo/db/query/get_executor.h
index 7f39d4436c2..c4f584a3c82 100644
--- a/src/mongo/db/query/get_executor.h
+++ b/src/mongo/db/query/get_executor.h
@@ -73,7 +73,7 @@ void filterAllowedIndexEntries(const AllowedIndicesFilter& allowedIndicesFilter,
  * 'collection'.  Exposed for testing.
  */
 void fillOutPlannerParams(OperationContext* opCtx,
-                          Collection* collection,
+                          const Collection* collection,
                           CanonicalQuery* canonicalQuery,
                           QueryPlannerParams* plannerParams);
 
@@ -119,7 +119,7 @@ bool shouldWaitForOplogVisibility(OperationContext* opCtx,
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     size_t plannerOptions = 0);
@@ -136,7 +136,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     bool permitYield = false,
     size_t plannerOptions = QueryPlannerParams::DEFAULT);
@@ -146,7 +146,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorLegacyFind(
     OperationContext* opCtx,
-    Collection* collection,
+    const Collection* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery);
 
 /**
@@ -203,7 +203,7 @@ bool turnIxscanIntoDistinctIxscan(QuerySolution* soln,
  * distinct.
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDistinct(
-    Collection* collection, size_t plannerOptions, ParsedDistinct* parsedDistinct);
+    const Collection* collection, size_t plannerOptions, ParsedDistinct* parsedDistinct);
 
 /*
  * Get a PlanExecutor for a query executing as part of a count command.
@@ -214,7 +214,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDist
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorCount(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    Collection* collection,
+    const Collection* collection,
     const CountCommand& request,
     bool explain,
     const NamespaceString& nss);
diff --git a/src/mongo/db/query/internal_plans.cpp b/src/mongo/db/query/internal_plans.cpp
index 7d932c8efcd..21a76b11120 100644
--- a/src/mongo/db/query/internal_plans.cpp
+++ b/src/mongo/db/query/internal_plans.cpp
@@ -51,7 +51,7 @@ namespace mongo {
 std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::collectionScan(
     OperationContext* opCtx,
     StringData ns,
-    Collection* collection,
+    const Collection* collection,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     const Direction direction) {
     std::unique_ptr<WorkingSet> ws = std::make_unique<WorkingSet>();
diff --git a/src/mongo/db/query/internal_plans.h b/src/mongo/db/query/internal_plans.h
index a7d91590554..c6bfc9b5dc4 100644
--- a/src/mongo/db/query/internal_plans.h
+++ b/src/mongo/db/query/internal_plans.h
@@ -72,7 +72,7 @@ public:
     static std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> collectionScan(
         OperationContext* opCtx,
         StringData ns,
-        Collection* collection,
+        const Collection* collection,
         PlanYieldPolicy::YieldPolicy yieldPolicy,
         const Direction direction = FORWARD);
 
diff --git a/src/mongo/db/query/query_request_test.cpp b/src/mongo/db/query/query_request_test.cpp
index 21ea341ee64..99b1f27ecc5 100644
--- a/src/mongo/db/query/query_request_test.cpp
+++ b/src/mongo/db/query/query_request_test.cpp
@@ -1644,9 +1644,9 @@ TEST_F(QueryRequestTest, ParseFromUUID) {
     // Register a UUID/Collection pair in the CollectionCatalog.
     const CollectionUUID uuid = UUID::gen();
     const NamespaceString nss("test.testns");
-    std::unique_ptr<Collection> collection = std::make_unique<CollectionMock>(nss);
+    std::shared_ptr<Collection> collection = std::make_shared<CollectionMock>(nss);
     CollectionCatalog& catalog = CollectionCatalog::get(opCtx.get());
-    catalog.registerCollection(uuid, &collection);
+    catalog.registerCollection(uuid, std::move(collection));
     QueryRequest qr(NamespaceStringOrUUID("test", uuid));
     // Ensure a call to refreshNSS succeeds.
     qr.refreshNSS(opCtx.get());
diff --git a/src/mongo/db/query/sbe_cached_solution_planner.h b/src/mongo/db/query/sbe_cached_solution_planner.h
index d14c44e2d72..264ddfbbbfa 100644
--- a/src/mongo/db/query/sbe_cached_solution_planner.h
+++ b/src/mongo/db/query/sbe_cached_solution_planner.h
@@ -43,7 +43,7 @@ namespace mongo::sbe {
 class CachedSolutionPlanner final : public BaseRuntimePlanner {
 public:
     CachedSolutionPlanner(OperationContext* opCtx,
-                          Collection* collection,
+                          const Collection* collection,
                           const CanonicalQuery& cq,
                           const QueryPlannerParams& queryParams,
                           size_t decisionReads,
diff --git a/src/mongo/db/query/sbe_sub_planner.h b/src/mongo/db/query/sbe_sub_planner.h
index ef6e6ae3e35..1bb465df0f4 100644
--- a/src/mongo/db/query/sbe_sub_planner.h
+++ b/src/mongo/db/query/sbe_sub_planner.h
@@ -43,7 +43,7 @@ namespace mongo::sbe {
 class SubPlanner final : public BaseRuntimePlanner {
 public:
     SubPlanner(OperationContext* opCtx,
-               Collection* collection,
+               const Collection* collection,
                const CanonicalQuery& cq,
                const QueryPlannerParams& queryParams,
                PlanYieldPolicySBE* yieldPolicy)
diff --git a/src/mongo/db/repl/idempotency_test_fixture.cpp b/src/mongo/db/repl/idempotency_test_fixture.cpp
index a8218696795..ca996de2692 100644
--- a/src/mongo/db/repl/idempotency_test_fixture.cpp
+++ b/src/mongo/db/repl/idempotency_test_fixture.cpp
@@ -324,7 +324,7 @@ OplogEntry IdempotencyTest::partialTxn(LogicalSessionId lsid,
                           prevOpTime);
 }
 
-std::string IdempotencyTest::computeDataHash(Collection* collection) {
+std::string IdempotencyTest::computeDataHash(const Collection* collection) {
     auto desc = collection->getIndexCatalog()->findIdIndex(_opCtx.get());
     ASSERT_TRUE(desc);
     auto exec = InternalPlanner::indexScan(_opCtx.get(),
diff --git a/src/mongo/db/repl/idempotency_test_fixture.h b/src/mongo/db/repl/idempotency_test_fixture.h
index 8ff6d0fdfc7..e871517e9f5 100644
--- a/src/mongo/db/repl/idempotency_test_fixture.h
+++ b/src/mongo/db/repl/idempotency_test_fixture.h
@@ -151,7 +151,7 @@ protected:
         return obj;
     };
 
-    std::string computeDataHash(Collection* collection);
+    std::string computeDataHash(const Collection* collection);
     virtual std::string getStatesString(const std::vector<CollectionState>& state1,
                                         const std::vector<CollectionState>& state2,
                                         const std::vector<OplogEntry>& ops);
diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index 0f9c3aa4969..4127ad6b410 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -303,9 +303,10 @@ namespace {
  * Returns NamespaceNotFound if the database or collection does not exist.
  */
 template <typename AutoGetCollectionType>
-StatusWith<Collection*> getCollection(const AutoGetCollectionType& autoGetCollection,
-                                      const NamespaceStringOrUUID& nsOrUUID,
-                                      const std::string& message) {
+StatusWith<decltype(std::declval<AutoGetCollectionType>().getCollection())> getCollection(
+    const AutoGetCollectionType& autoGetCollection,
+    const NamespaceStringOrUUID& nsOrUUID,
+    const std::string& message) {
     if (!autoGetCollection.getDb()) {
         StringData dbName = nsOrUUID.nss() ? nsOrUUID.nss()->db() : nsOrUUID.dbname();
         return {ErrorCodes::NamespaceNotFound,
diff --git a/src/mongo/db/repl/storage_interface_impl_test.cpp b/src/mongo/db/repl/storage_interface_impl_test.cpp
index 9a121099170..54a2661b70c 100644
--- a/src/mongo/db/repl/storage_interface_impl_test.cpp
+++ b/src/mongo/db/repl/storage_interface_impl_test.cpp
@@ -155,7 +155,9 @@ TimestampedBSONObj makeOplogEntry(OpTime opTime) {
 /**
  * Counts the number of keys in an index using an IndexAccessMethod::validate call.
  */
-int64_t getIndexKeyCount(OperationContext* opCtx, IndexCatalog* cat, const IndexDescriptor* desc) {
+int64_t getIndexKeyCount(OperationContext* opCtx,
+                         const IndexCatalog* cat,
+                         const IndexDescriptor* desc) {
     auto idx = cat->getEntry(desc)->accessMethod();
     int64_t numKeys;
     ValidateResults fullRes;
diff --git a/src/mongo/db/s/check_sharding_index_command.cpp b/src/mongo/db/s/check_sharding_index_command.cpp
index b8c39db6577..64f07ebcccf 100644
--- a/src/mongo/db/s/check_sharding_index_command.cpp
+++ b/src/mongo/db/s/check_sharding_index_command.cpp
@@ -88,7 +88,7 @@ public:
         }
 
         AutoGetCollectionForReadCommand autoColl(opCtx, nss);
-        Collection* const collection = autoColl.getCollection();
+        auto collection = autoColl.getCollection();
         if (!collection) {
             errmsg = "ns not found";
             return false;
diff --git a/src/mongo/db/s/collection_sharding_runtime.cpp b/src/mongo/db/s/collection_sharding_runtime.cpp
index 5a8204144b9..f03c817029f 100644
--- a/src/mongo/db/s/collection_sharding_runtime.cpp
+++ b/src/mongo/db/s/collection_sharding_runtime.cpp
@@ -408,7 +408,7 @@ CollectionCriticalSection::CollectionCriticalSection(OperationContext* opCtx, Na
     AutoGetCollection autoColl(_opCtx,
                                _nss,
                                MODE_S,
-                               AutoGetCollection::ViewMode::kViewsForbidden,
+                               AutoGetCollectionViewMode::kViewsForbidden,
                                _opCtx->getServiceContext()->getPreciseClockSource()->now() +
                                    Milliseconds(migrationLockAcquisitionMaxWaitMS.load()));
     auto* const csr = CollectionShardingRuntime::get(_opCtx, _nss);
@@ -428,7 +428,7 @@ void CollectionCriticalSection::enterCommitPhase() {
     AutoGetCollection autoColl(_opCtx,
                                _nss,
                                MODE_X,
-                               AutoGetCollection::ViewMode::kViewsForbidden,
+                               AutoGetCollectionViewMode::kViewsForbidden,
                                _opCtx->getServiceContext()->getPreciseClockSource()->now() +
                                    Milliseconds(migrationLockAcquisitionMaxWaitMS.load()));
     auto* const csr = CollectionShardingRuntime::get(_opCtx, _nss);
diff --git a/src/mongo/db/s/get_shard_version_command.cpp b/src/mongo/db/s/get_shard_version_command.cpp
index 42f3b200ac4..6d938f5fa9c 100644
--- a/src/mongo/db/s/get_shard_version_command.cpp
+++ b/src/mongo/db/s/get_shard_version_command.cpp
@@ -97,8 +97,7 @@ public:
         result.appendBool("inShardedMode", false);
         result.appendTimestamp("mine", 0);
 
-        AutoGetCollection autoColl(
-            opCtx, nss, MODE_IS, AutoGetCollection::ViewMode::kViewsPermitted);
+        AutoGetCollection autoColl(opCtx, nss, MODE_IS, AutoGetCollectionViewMode::kViewsPermitted);
         auto* const csr = CollectionShardingRuntime::get(opCtx, nss);
 
         const auto optMetadata = csr->getCurrentMetadataIfKnown();
diff --git a/src/mongo/db/s/migration_source_manager.cpp b/src/mongo/db/s/migration_source_manager.cpp
index 839ece87e56..5e59b07115f 100644
--- a/src/mongo/db/s/migration_source_manager.cpp
+++ b/src/mongo/db/s/migration_source_manager.cpp
@@ -246,7 +246,7 @@ Status MigrationSourceManager::startClone() {
         AutoGetCollection autoColl(_opCtx,
                                    getNss(),
                                    replEnabled ? MODE_IX : MODE_X,
-                                   AutoGetCollection::ViewMode::kViewsForbidden,
+                                   AutoGetCollectionViewMode::kViewsForbidden,
                                    _opCtx->getServiceContext()->getPreciseClockSource()->now() +
                                        Milliseconds(migrationLockAcquisitionMaxWaitMS.load()));
 
diff --git a/src/mongo/db/s/shard_filtering_metadata_refresh.cpp b/src/mongo/db/s/shard_filtering_metadata_refresh.cpp
index b7354a600e0..5475c70838e 100644
--- a/src/mongo/db/s/shard_filtering_metadata_refresh.cpp
+++ b/src/mongo/db/s/shard_filtering_metadata_refresh.cpp
@@ -103,7 +103,7 @@ SharedSemiFuture<void> recoverRefreshShardVersion(ServiceContext* serviceContext
             ON_BLOCK_EXIT([&] {
                 UninterruptibleLockGuard noInterrupt(opCtx->lockState());
                 AutoGetCollection autoColl(
-                    opCtx.get(), nss, MODE_IX, AutoGetCollection::ViewMode::kViewsForbidden);
+                    opCtx.get(), nss, MODE_IX, AutoGetCollectionViewMode::kViewsForbidden);
 
                 auto* const csr = CollectionShardingRuntime::get(opCtx.get(), nss);
                 auto csrLock = CollectionShardingRuntime::CSRLock::lockExclusive(opCtx.get(), csr);
@@ -161,7 +161,7 @@ void onShardVersionMismatch(OperationContext* opCtx,
 
         {
             AutoGetCollection autoColl(
-                opCtx, nss, MODE_IS, AutoGetCollection::ViewMode::kViewsForbidden);
+                opCtx, nss, MODE_IS, AutoGetCollectionViewMode::kViewsForbidden);
 
             auto* const csr = CollectionShardingRuntime::get(opCtx, nss);
 
@@ -235,7 +235,7 @@ ScopedShardVersionCriticalSection::ScopedShardVersionCriticalSection(OperationCo
             AutoGetCollection autoColl(_opCtx,
                                        _nss,
                                        MODE_S,
-                                       AutoGetCollection::ViewMode::kViewsForbidden,
+                                       AutoGetCollectionViewMode::kViewsForbidden,
                                        _opCtx->getServiceContext()->getPreciseClockSource()->now() +
                                            Milliseconds(migrationLockAcquisitionMaxWaitMS.load()));
 
@@ -285,7 +285,7 @@ void ScopedShardVersionCriticalSection::enterCommitPhase() {
     AutoGetCollection autoColl(_opCtx,
                                _nss,
                                MODE_IS,
-                               AutoGetCollection::ViewMode::kViewsForbidden,
+                               AutoGetCollectionViewMode::kViewsForbidden,
                                _opCtx->getServiceContext()->getPreciseClockSource()->now() +
                                    Milliseconds(migrationLockAcquisitionMaxWaitMS.load()));
     auto* const csr = CollectionShardingRuntime::get(_opCtx, _nss);
diff --git a/src/mongo/db/s/shardsvr_shard_collection.cpp b/src/mongo/db/s/shardsvr_shard_collection.cpp
index 95fb4a22143..d62d86baf44 100644
--- a/src/mongo/db/s/shardsvr_shard_collection.cpp
+++ b/src/mongo/db/s/shardsvr_shard_collection.cpp
@@ -190,7 +190,7 @@ void checkCollation(OperationContext* opCtx, const ShardsvrShardCollectionReques
     AutoGetCollection autoColl(opCtx,
                                *request.get_shardsvrShardCollection(),
                                MODE_IS,
-                               AutoGetCollection::ViewMode::kViewsForbidden);
+                               AutoGetCollectionViewMode::kViewsForbidden);
 
     const auto actualCollator = [&]() -> const CollatorInterface* {
         const auto* const coll = autoColl.getCollection();
diff --git a/src/mongo/db/stats/storage_stats.cpp b/src/mongo/db/stats/storage_stats.cpp
index 4d4591a84f6..429fe7d216e 100644
--- a/src/mongo/db/stats/storage_stats.cpp
+++ b/src/mongo/db/stats/storage_stats.cpp
@@ -63,14 +63,14 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
     try {
         autoColl.emplace(opCtx,
                          nss,
-                         AutoGetCollection::ViewMode::kViewsForbidden,
+                         AutoGetCollectionViewMode::kViewsForbidden,
                          waitForLock ? Date_t::max() : Date_t::now());
     } catch (const ExceptionForCat<ErrorCategory::Interruption>&) {
         LOGV2_DEBUG(3088801, 2, "Failed to retrieve storage statistics", logAttrs(nss));
         return Status::OK();
     }
 
-    Collection* collection = autoColl->getCollection();  // Will be set if present
+    const Collection* collection = autoColl->getCollection();  // Will be set if present
     if (!autoColl->getDb() || !collection) {
         result->appendNumber("size", 0);
         result->appendNumber("count", 0);
@@ -95,7 +95,7 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
     if (numRecords)
         result->append("avgObjSize", collection->averageObjectSize(opCtx));
 
-    RecordStore* recordStore = collection->getRecordStore();
+    const RecordStore* recordStore = collection->getRecordStore();
     auto storageSize =
         static_cast<long long>(recordStore->storageSize(opCtx, result, verbose ? 1 : 0));
     result->appendNumber("storageSize", storageSize / scale);
@@ -107,7 +107,7 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
 
     recordStore->appendCustomStats(opCtx, result, scale);
 
-    IndexCatalog* indexCatalog = collection->getIndexCatalog();
+    const IndexCatalog* indexCatalog = collection->getIndexCatalog();
     result->append("nindexes", indexCatalog->numIndexesTotal(opCtx));
 
     BSONObjBuilder indexDetails;
@@ -154,7 +154,7 @@ Status appendCollectionRecordCount(OperationContext* opCtx,
                 str::stream() << "Database [" << nss.db().toString() << "] not found."};
     }
 
-    Collection* collection = ctx.getCollection();
+    const Collection* collection = ctx.getCollection();
     if (!collection) {
         return {ErrorCodes::BadValue,
                 str::stream() << "Collection [" << nss.toString() << "] not found."};
diff --git a/src/mongo/db/storage/capped_callback.h b/src/mongo/db/storage/capped_callback.h
index cced6f61f5d..96a6e33aa29 100644
--- a/src/mongo/db/storage/capped_callback.h
+++ b/src/mongo/db/storage/capped_callback.h
@@ -56,11 +56,11 @@ public:
     /**
      * Returns true if there may be waiters.
      */
-    virtual bool haveCappedWaiters() = 0;
+    virtual bool haveCappedWaiters() const = 0;
 
     /**
      * Used to notify any waiters when new documents may be visible in the capped collection.
      */
-    virtual void notifyCappedWaitersIfNeeded() = 0;
+    virtual void notifyCappedWaitersIfNeeded() const = 0;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/storage/kv/durable_catalog_test.cpp b/src/mongo/db/storage/kv/durable_catalog_test.cpp
index 31a58b404dc..bb2026f7bdf 100644
--- a/src/mongo/db/storage/kv/durable_catalog_test.cpp
+++ b/src/mongo/db/storage/kv/durable_catalog_test.cpp
@@ -93,9 +93,10 @@ public:
             ASSERT_OK(swColl.getStatus());
             std::pair<RecordId, std::unique_ptr<RecordStore>> coll = std::move(swColl.getValue());
             _catalogId = coll.first;
-            std::unique_ptr<Collection> collection =
-                std::make_unique<CollectionMock>(_nss, _catalogId);
-            CollectionCatalog::get(opCtx.get()).registerCollection(options.uuid.get(), &collection);
+            std::shared_ptr<Collection> collection =
+                std::make_shared<CollectionMock>(_nss, _catalogId);
+            CollectionCatalog::get(opCtx.get())
+                .registerCollection(options.uuid.get(), std::move(collection));
             wuow.commit();
         }
     }
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index 93742937d27..002f20ca43b 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -274,7 +274,7 @@ void StorageEngineImpl::_initCollection(OperationContext* opCtx,
     auto collection = collectionFactory->make(opCtx, nss, catalogId, uuid, std::move(rs));
 
     auto& collectionCatalog = CollectionCatalog::get(getGlobalServiceContext());
-    collectionCatalog.registerCollection(uuid, &collection);
+    collectionCatalog.registerCollection(uuid, std::move(collection));
 }
 
 void StorageEngineImpl::closeCatalog(OperationContext* opCtx) {
diff --git a/src/mongo/db/storage/storage_engine_test_fixture.h b/src/mongo/db/storage/storage_engine_test_fixture.h
index c2f10dd198e..074904803ad 100644
--- a/src/mongo/db/storage/storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/storage_engine_test_fixture.h
@@ -69,8 +69,8 @@ public:
                 _storageEngine->getCatalog()->createCollection(opCtx, ns, options, true));
             wuow.commit();
         }
-        std::unique_ptr<Collection> coll = std::make_unique<CollectionMock>(ns, catalogId);
-        CollectionCatalog::get(opCtx).registerCollection(options.uuid.get(), &coll);
+        std::shared_ptr<Collection> coll = std::make_shared<CollectionMock>(ns, catalogId);
+        CollectionCatalog::get(opCtx).registerCollection(options.uuid.get(), std::move(coll));
 
         return {{_storageEngine->getCatalog()->getEntry(catalogId)}};
     }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
index 2aaec1fc7b9..3d99e182038 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
@@ -96,6 +96,7 @@
 #include "mongo/util/processinfo.h"
 #include "mongo/util/quick_exit.h"
 #include "mongo/util/scopeguard.h"
+#include "mongo/util/stacktrace.h"
 #include "mongo/util/testing_proctor.h"
 #include "mongo/util/time_support.h"
 
@@ -2388,17 +2389,21 @@ bool WiredTigerKVEngine::supportsOplogStones() const {
 void WiredTigerKVEngine::startOplogManager(OperationContext* opCtx,
                                            WiredTigerRecordStore* oplogRecordStore) {
     stdx::lock_guard<Latch> lock(_oplogManagerMutex);
-    if (_oplogManagerCount == 0)
-        _oplogManager->startVisibilityThread(opCtx, oplogRecordStore);
-    _oplogManagerCount++;
+    // Halt visibility thread if running on previous record store
+    if (_oplogRecordStore) {
+        _oplogManager->haltVisibilityThread();
+    }
+
+    _oplogManager->startVisibilityThread(opCtx, oplogRecordStore);
+    _oplogRecordStore = oplogRecordStore;
 }
 
-void WiredTigerKVEngine::haltOplogManager() {
+void WiredTigerKVEngine::haltOplogManager(WiredTigerRecordStore* oplogRecordStore) {
     stdx::unique_lock<Latch> lock(_oplogManagerMutex);
-    invariant(_oplogManagerCount > 0);
-    _oplogManagerCount--;
-    if (_oplogManagerCount == 0) {
+    // Halt visibility thread if the request match current
+    if (_oplogRecordStore == oplogRecordStore) {
         _oplogManager->haltVisibilityThread();
+        _oplogRecordStore = nullptr;
     }
 }
 
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
index eadcbc6a188..80dc2c370c7 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
@@ -304,7 +304,7 @@ public:
      * be started and stopped multiple times as tests create and destroy the oplog record store.
      */
     void startOplogManager(OperationContext* opCtx, WiredTigerRecordStore* oplogRecordStore);
-    void haltOplogManager();
+    void haltOplogManager(WiredTigerRecordStore* oplogRecordStore);
 
     /*
      * Always returns a non-nil pointer. However, the WiredTigerOplogManager may not have been
@@ -430,9 +430,9 @@ private:
     std::unique_ptr<WiredTigerSessionCache> _sessionCache;
     ClockSource* const _clockSource;
 
-    // Mutex to protect use of _oplogManagerCount by this instance of KV engine.
+    // Mutex to protect use of _oplogRecordStore by this instance of KV engine.
     mutable Mutex _oplogManagerMutex = MONGO_MAKE_LATCH("::_oplogManagerMutex");
-    std::size_t _oplogManagerCount = 0;
+    const WiredTigerRecordStore* _oplogRecordStore = nullptr;
     std::unique_ptr<WiredTigerOplogManager> _oplogManager;
 
     std::string _canonicalName;
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
index a83bd10d82b..c02c888296e 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
@@ -884,7 +884,7 @@ WiredTigerRecordStore::~WiredTigerRecordStore() {
 
     if (_isOplog) {
         // Delete oplog visibility manager on KV engine.
-        _kvEngine->haltOplogManager();
+        _kvEngine->haltOplogManager(this);
     }
 }
 
diff --git a/src/mongo/dbtests/plan_ranking.cpp b/src/mongo/dbtests/plan_ranking.cpp
index 6f7991e5977..0457a67d829 100644
--- a/src/mongo/dbtests/plan_ranking.cpp
+++ b/src/mongo/dbtests/plan_ranking.cpp
@@ -115,7 +115,7 @@ public:
      */
     const QuerySolution* pickBestPlan(CanonicalQuery* cq) {
         AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-        Collection* collection = ctx.getCollection();
+        const Collection* collection = ctx.getCollection();
 
         QueryPlannerParams plannerParams;
         fillOutPlannerParams(&_opCtx, collection, cq, &plannerParams);
@@ -246,7 +246,7 @@ public:
                                                  soln->root.get()));
 
         AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-        Collection* collection = ctx.getCollection();
+        const Collection* collection = ctx.getCollection();
 
         StatusWith<std::unique_ptr<PlanCacheEntry>> planCacheEntryWithStatus =
             CollectionQueryInfo::get(collection).getPlanCache()->getEntry(*(cq.get()));
diff --git a/src/mongo/dbtests/query_stage_cached_plan.cpp b/src/mongo/dbtests/query_stage_cached_plan.cpp
index ba32d9235d6..cf38ceadda4 100644
--- a/src/mongo/dbtests/query_stage_cached_plan.cpp
+++ b/src/mongo/dbtests/query_stage_cached_plan.cpp
@@ -142,7 +142,7 @@ public:
         return numResults;
     }
 
-    void forceReplanning(Collection* collection, CanonicalQuery* cq) {
+    void forceReplanning(const Collection* collection, CanonicalQuery* cq) {
         // Get planner params.
         QueryPlannerParams plannerParams;
         fillOutPlannerParams(&_opCtx, collection, cq, &plannerParams);
@@ -184,7 +184,7 @@ protected:
  */
 TEST_F(QueryStageCachedPlan, QueryStageCachedPlanFailureMemoryLimitExceeded) {
     AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-    Collection* collection = ctx.getCollection();
+    const Collection* collection = ctx.getCollection();
     ASSERT(collection);
 
     // Query can be answered by either index on "a" or index on "b".
@@ -235,7 +235,7 @@ TEST_F(QueryStageCachedPlan, QueryStageCachedPlanFailureMemoryLimitExceeded) {
  */
 TEST_F(QueryStageCachedPlan, QueryStageCachedPlanHitMaxWorks) {
     AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-    Collection* collection = ctx.getCollection();
+    const Collection* collection = ctx.getCollection();
     ASSERT(collection);
 
     // Query can be answered by either index on "a" or index on "b".
@@ -288,7 +288,7 @@ TEST_F(QueryStageCachedPlan, QueryStageCachedPlanHitMaxWorks) {
  */
 TEST_F(QueryStageCachedPlan, QueryStageCachedPlanAddsActiveCacheEntries) {
     AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-    Collection* collection = ctx.getCollection();
+    const Collection* collection = ctx.getCollection();
     ASSERT(collection);
 
     // Never run - just used as a key for the cache's get() functions, since all of the other
@@ -347,7 +347,7 @@ TEST_F(QueryStageCachedPlan, QueryStageCachedPlanAddsActiveCacheEntries) {
 
 TEST_F(QueryStageCachedPlan, DeactivatesEntriesOnReplan) {
     AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-    Collection* collection = ctx.getCollection();
+    const Collection* collection = ctx.getCollection();
     ASSERT(collection);
 
     // Never run - just used as a key for the cache's get() functions, since all of the other
@@ -406,7 +406,7 @@ TEST_F(QueryStageCachedPlan, EntriesAreNotDeactivatedWhenInactiveEntriesDisabled
     ON_BLOCK_EXIT([] { internalQueryCacheDisableInactiveEntries.store(false); });
 
     AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-    Collection* collection = ctx.getCollection();
+    const Collection* collection = ctx.getCollection();
     ASSERT(collection);
 
     // Never run - just used as a key for the cache's get() functions, since all of the other
@@ -449,7 +449,7 @@ TEST_F(QueryStageCachedPlan, ThrowsOnYieldRecoveryWhenIndexIsDroppedBeforePlanSe
 
     boost::optional<AutoGetCollectionForReadCommand> readLock;
     readLock.emplace(&_opCtx, nss);
-    Collection* collection = readLock->getCollection();
+    const Collection* collection = readLock->getCollection();
     ASSERT(collection);
 
     // Query can be answered by either index on "a" or index on "b".
@@ -491,7 +491,7 @@ TEST_F(QueryStageCachedPlan, DoesNotThrowOnYieldRecoveryWhenIndexIsDroppedAferPl
 
     boost::optional<AutoGetCollectionForReadCommand> readLock;
     readLock.emplace(&_opCtx, nss);
-    Collection* collection = readLock->getCollection();
+    const Collection* collection = readLock->getCollection();
     ASSERT(collection);
 
     // Query can be answered by either index on "a" or index on "b".
diff --git a/src/mongo/dbtests/query_stage_collscan.cpp b/src/mongo/dbtests/query_stage_collscan.cpp
index 04bfe282571..275f13f5158 100644
--- a/src/mongo/dbtests/query_stage_collscan.cpp
+++ b/src/mongo/dbtests/query_stage_collscan.cpp
@@ -123,7 +123,7 @@ public:
         return count;
     }
 
-    void getRecordIds(Collection* collection,
+    void getRecordIds(const Collection* collection,
                       CollectionScanParams::Direction direction,
                       vector<RecordId>* out) {
         WorkingSet ws;
diff --git a/src/mongo/dbtests/query_stage_distinct.cpp b/src/mongo/dbtests/query_stage_distinct.cpp
index 92a7bd2644e..5c6f026cbbb 100644
--- a/src/mongo/dbtests/query_stage_distinct.cpp
+++ b/src/mongo/dbtests/query_stage_distinct.cpp
@@ -126,7 +126,7 @@ public:
         addIndex(BSON("a" << 1));
 
         AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-        Collection* coll = ctx.getCollection();
+        const Collection* coll = ctx.getCollection();
 
         // Set up the distinct stage.
         std::vector<const IndexDescriptor*> indexes;
@@ -192,7 +192,7 @@ public:
         addIndex(BSON("a" << 1));
 
         AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-        Collection* coll = ctx.getCollection();
+        const Collection* coll = ctx.getCollection();
 
         // Set up the distinct stage.
         std::vector<const IndexDescriptor*> indexes;
@@ -259,7 +259,7 @@ public:
         addIndex(BSON("a" << 1 << "b" << 1));
 
         AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-        Collection* coll = ctx.getCollection();
+        const Collection* coll = ctx.getCollection();
 
         std::vector<const IndexDescriptor*> indices;
         coll->getIndexCatalog()->findIndexesByKeyPattern(
diff --git a/src/mongo/dbtests/query_stage_multiplan.cpp b/src/mongo/dbtests/query_stage_multiplan.cpp
index 83337ca69ec..0f946280824 100644
--- a/src/mongo/dbtests/query_stage_multiplan.cpp
+++ b/src/mongo/dbtests/query_stage_multiplan.cpp
@@ -370,7 +370,7 @@ TEST_F(QueryStageMultiPlanTest, MPSBackupPlan) {
     addIndex(BSON("b" << 1));
 
     AutoGetCollectionForReadCommand ctx(_opCtx.get(), nss);
-    Collection* collection = ctx.getCollection();
+    const Collection* collection = ctx.getCollection();
 
     // Query for both 'a' and 'b' and sort on 'b'.
     auto qr = std::make_unique<QueryRequest>(nss);
@@ -543,7 +543,7 @@ TEST_F(QueryStageMultiPlanTest, MPSSummaryStats) {
     addIndex(BSON("foo" << -1 << "bar" << 1));
 
     AutoGetCollectionForReadCommand ctx(_opCtx.get(), nss);
-    Collection* coll = ctx.getCollection();
+    const Collection* coll = ctx.getCollection();
 
     // Create the executor (Matching all documents).
     auto qr = std::make_unique<QueryRequest>(nss);
diff --git a/src/mongo/dbtests/query_stage_tests.cpp b/src/mongo/dbtests/query_stage_tests.cpp
index 1b6701a9de7..bb79ac22a1c 100644
--- a/src/mongo/dbtests/query_stage_tests.cpp
+++ b/src/mongo/dbtests/query_stage_tests.cpp
@@ -124,7 +124,7 @@ public:
 
     const IndexDescriptor* getIndex(const BSONObj& obj) {
         AutoGetCollectionForReadCommand ctx(&_opCtx, NamespaceString(ns()));
-        Collection* collection = ctx.getCollection();
+        const Collection* collection = ctx.getCollection();
         std::vector<const IndexDescriptor*> indexes;
         collection->getIndexCatalog()->findIndexesByKeyPattern(&_opCtx, obj, false, &indexes);
         return indexes.empty() ? nullptr : indexes[0];
diff --git a/src/mongo/dbtests/query_stage_update.cpp b/src/mongo/dbtests/query_stage_update.cpp
index c135f1d916d..5479714d190 100644
--- a/src/mongo/dbtests/query_stage_update.cpp
+++ b/src/mongo/dbtests/query_stage_update.cpp
@@ -125,7 +125,7 @@ public:
      * Uses a forward collection scan stage to get the docs, and populates 'out' with
      * the results.
      */
-    void getCollContents(Collection* collection, vector<BSONObj>* out) {
+    void getCollContents(const Collection* collection, vector<BSONObj>* out) {
         WorkingSet ws;
 
         CollectionScanParams params;
@@ -145,7 +145,7 @@ public:
         }
     }
 
-    void getRecordIds(Collection* collection,
+    void getRecordIds(const Collection* collection,
                       CollectionScanParams::Direction direction,
                       vector<RecordId>* out) {
         WorkingSet ws;
@@ -244,7 +244,7 @@ public:
         // Verify the contents of the resulting collection.
         {
             AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-            Collection* collection = ctx.getCollection();
+            const Collection* collection = ctx.getCollection();
 
             vector<BSONObj> objs;
             getCollContents(collection, &objs);
@@ -351,7 +351,7 @@ public:
         // Check the contents of the collection.
         {
             AutoGetCollectionForReadCommand ctx(&_opCtx, nss);
-            Collection* collection = ctx.getCollection();
+            const Collection* collection = ctx.getCollection();
 
             vector<BSONObj> objs;
             getCollContents(collection, &objs);
diff --git a/src/mongo/dbtests/storage_timestamp_tests.cpp b/src/mongo/dbtests/storage_timestamp_tests.cpp
index 45a3a5f5400..cc9bc42645d 100644
--- a/src/mongo/dbtests/storage_timestamp_tests.cpp
+++ b/src/mongo/dbtests/storage_timestamp_tests.cpp
@@ -303,7 +303,7 @@ public:
         abortOnExit.dismiss();
     }
 
-    std::int32_t itCount(Collection* coll) {
+    std::int32_t itCount(const Collection* coll) {
         std::uint64_t ret = 0;
         auto cursor = coll->getRecordStore()->getCursor(_opCtx);
         while (cursor->next() != boost::none) {
@@ -313,7 +313,7 @@ public:
         return ret;
     }
 
-    BSONObj findOne(Collection* coll) {
+    BSONObj findOne(const Collection* coll) {
         auto optRecord = coll->getRecordStore()->getCursor(_opCtx)->next();
         if (optRecord == boost::none) {
             // Print a stack trace to help disambiguate which `findOne` failed.
@@ -402,7 +402,7 @@ public:
             << ". Expected: " << expectedDoc.toBSON() << ". Found: " << doc.toBSON();
     }
 
-    void assertDocumentAtTimestamp(Collection* coll,
+    void assertDocumentAtTimestamp(const Collection* coll,
                                    const Timestamp& ts,
                                    const BSONObj& expectedDoc) {
         OneOffRead oor(_opCtx, ts);
@@ -418,7 +418,7 @@ public:
         }
     }
 
-    void assertFilteredDocumentAtTimestamp(Collection* coll,
+    void assertFilteredDocumentAtTimestamp(const Collection* coll,
                                            const BSONObj& query,
                                            const Timestamp& ts,
                                            boost::optional<const BSONObj&> expectedDoc) {
diff --git a/src/mongo/dbtests/wildcard_multikey_persistence_test.cpp b/src/mongo/dbtests/wildcard_multikey_persistence_test.cpp
index 69e45464355..f611a733d6c 100644
--- a/src/mongo/dbtests/wildcard_multikey_persistence_test.cpp
+++ b/src/mongo/dbtests/wildcard_multikey_persistence_test.cpp
@@ -233,13 +233,13 @@ protected:
         return collection->getIndexCatalog()->findIndexByName(opCtx(), indexName);
     }
 
-    const IndexAccessMethod* getIndex(Collection* collection, const StringData indexName) {
+    const IndexAccessMethod* getIndex(const Collection* collection, const StringData indexName) {
         return collection->getIndexCatalog()
             ->getEntry(getIndexDesc(collection, indexName))
             ->accessMethod();
     }
 
-    std::unique_ptr<SortedDataInterface::Cursor> getIndexCursor(Collection* collection,
+    std::unique_ptr<SortedDataInterface::Cursor> getIndexCursor(const Collection* collection,
                                                                 const StringData indexName) {
         return getIndex(collection, indexName)->newCursor(opCtx());
     }