diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index 64a893f59a..1346642401 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -156,9 +156,26 @@ class CappedInsertNotifier {
  * this is NOT safe through a yield right now.
  * not sure if it will be, or what yet.
  */
-class Collection final : CappedCallback, UpdateNotifier {
+class Collection final : CappedCallback,
+                         UpdateNotifier,
+                         public std::enable_shared_from_this<Collection> {
 public:
     using Uptr = std::unique_ptr<Collection>;
+    // Shared ownership is required for the Database collection caches: getCollection() rebuilds
+    // cached entries in place on catalog version changes while operations still hold raw pointers
+    // (pinned on their RecoveryUnit), so eviction must not destroy eagerly.
+    using Sptr = std::shared_ptr<Collection>;
+
+    /**
+     * Shared ownership of this object when it is owned by a shared collection cache
+     * (DatabaseImpl::_collections, UUIDCatalog); null for uniquely-owned instances. Used by
+     * holders that outlive the operation that resolved the collection -- e.g. globally-managed
+     * aggregation cursors, whose getMore path never re-resolves through getCollection() and so
+     * never takes a RecoveryUnit pin.
+     */
+    Sptr sharedFromThisIfShared() {
+        return weak_from_this().lock();
+    }
 
     enum ValidationAction { WARN, ERROR_V };
     enum ValidationLevel { OFF, MODERATE, STRICT_V };
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 6d9215ed1c..ed82cca69b 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -335,7 +335,23 @@ Status CollectionImpl::insertDocumentsForOplog(OperationContext* opCtx,
         return status;
 
     opCtx->recoveryUnit()->onCommit(
-        [this](boost::optional<Timestamp>) { notifyCappedWaitersIfNeeded(); });
+        [notifier = _cappedNotifier](boost::optional<Timestamp>) {
+            // Capture the notifier, not the Collection. Eloq refreshes the cached Collection
+            // while committing catalog metadata, before RecoveryUnit callbacks run, so a
+            // captured `this` can already be freed here -- the same hazard the eloq guard in
+            // IndexCatalogEntryImpl::setMultikey was added for. Holding a shared_ptr copy keeps
+            // the notifier alive independently of the Collection.
+            //
+            // The haveCappedWaiters() short-circuit is dropped along with `this`: its use_count
+            // test cannot be expressed once the Collection may be gone, and it was only an
+            // optimisation. notifyAll() on a notifier nobody waits on takes one uncontended
+            // mutex and bumps a version that no waiter reads; on a killed notifier it is
+            // likewise harmless, because waiters exit on the _dead flag. Non-capped collections
+            // have a null notifier and do nothing at all.
+            if (notifier) {
+                notifier->notifyAll();
+            }
+        });
 
     return status;
 }
@@ -388,7 +404,23 @@ Status CollectionImpl::insertDocuments(OperationContext* opCtx,
         opCtx, ns(), uuid(), begin, end, fromMigrate);
 
     opCtx->recoveryUnit()->onCommit(
-        [this](boost::optional<Timestamp>) { notifyCappedWaitersIfNeeded(); });
+        [notifier = _cappedNotifier](boost::optional<Timestamp>) {
+            // Capture the notifier, not the Collection. Eloq refreshes the cached Collection
+            // while committing catalog metadata, before RecoveryUnit callbacks run, so a
+            // captured `this` can already be freed here -- the same hazard the eloq guard in
+            // IndexCatalogEntryImpl::setMultikey was added for. Holding a shared_ptr copy keeps
+            // the notifier alive independently of the Collection.
+            //
+            // The haveCappedWaiters() short-circuit is dropped along with `this`: its use_count
+            // test cannot be expressed once the Collection may be gone, and it was only an
+            // optimisation. notifyAll() on a notifier nobody waits on takes one uncontended
+            // mutex and bumps a version that no waiter reads; on a killed notifier it is
+            // likewise harmless, because waiters exit on the _dead flag. Non-capped collections
+            // have a null notifier and do nothing at all.
+            if (notifier) {
+                notifier->notifyAll();
+            }
+        });
 
     MONGO_FAIL_POINT_BLOCK(hangAfterCollectionInserts, extraData) {
         const BSONObj& data = extraData.getData();
@@ -472,7 +504,23 @@ Status CollectionImpl::insertDocument(OperationContext* opCtx,
         opCtx, ns(), uuid(), inserts.begin(), inserts.end(), false);
 
     opCtx->recoveryUnit()->onCommit(
-        [this](boost::optional<Timestamp>) { notifyCappedWaitersIfNeeded(); });
+        [notifier = _cappedNotifier](boost::optional<Timestamp>) {
+            // Capture the notifier, not the Collection. Eloq refreshes the cached Collection
+            // while committing catalog metadata, before RecoveryUnit callbacks run, so a
+            // captured `this` can already be freed here -- the same hazard the eloq guard in
+            // IndexCatalogEntryImpl::setMultikey was added for. Holding a shared_ptr copy keeps
+            // the notifier alive independently of the Collection.
+            //
+            // The haveCappedWaiters() short-circuit is dropped along with `this`: its use_count
+            // test cannot be expressed once the Collection may be gone, and it was only an
+            // optimisation. notifyAll() on a notifier nobody waits on takes one uncontended
+            // mutex and bumps a version that no waiter reads; on a killed notifier it is
+            // likewise harmless, because waiters exit on the _dead flag. Non-capped collections
+            // have a null notifier and do nothing at all.
+            if (notifier) {
+                notifier->notifyAll();
+            }
+        });
 
     return loc.getStatus();
 }
diff --git a/src/mongo/db/catalog/database.h b/src/mongo/db/catalog/database.h
index 3a6eb6aaf1..6ddfa366cf 100644
--- a/src/mongo/db/catalog/database.h
+++ b/src/mongo/db/catalog/database.h
@@ -59,8 +59,8 @@ namespace mongo {
 class Database : public Decorable<Database> {
 public:
     // Used for range-based loop only
-    using CollectionMapView = std::map<std::string, Collection::Uptr>;
-    using CollectionMap = std::map<std::string, Collection::Uptr, std::less<void>>;
+    using CollectionMapView = std::map<std::string, Collection::Sptr>;
+    using CollectionMap = std::map<std::string, Collection::Sptr, std::less<void>>;
 
     class Impl {
     public:
@@ -130,7 +130,10 @@ class Database : public Decorable<Database> {
         virtual StatusWith<NamespaceString> makeUniqueCollectionNamespace(
             OperationContext* opCtx, StringData collectionNameModel) = 0;
 
-        virtual CollectionMapView& collections(OperationContext* opCtx) = 0;
+        // Returned by value: an operation-owned snapshot. A shared member map would be cleared
+        // and rebuilt while earlier callers iterate it (map-iterator invalidation that shared_ptr
+        // entries do not cure).
+        virtual CollectionMapView collections(OperationContext* opCtx) = 0;
         // virtual CollectionMap& collections() = 0;
         // virtual const CollectionMap& collections() const = 0;
         // virtual CollectionMap::const_iterator begin() const = 0;
@@ -219,7 +222,7 @@ class Database : public Decorable<Database> {
     inline Database(Database&&) = delete;
     inline Database& operator=(Database&&) = delete;
 
-    CollectionMapView& collections(OperationContext* opCtx) {
+    CollectionMapView collections(OperationContext* opCtx) {
         return this->_impl().collections(opCtx);
     }
     // inline iterator begin() const {
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 10f97cf7e2..bcf630eb61 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -158,9 +158,10 @@ void uassertNamespaceNotIndex(StringData ns, StringData caller) {
 // };
 
 DatabaseImpl::~DatabaseImpl() {
+    // Entries may be kept alive past this point by RecoveryUnit pins and cursor pins; clearing
+    // under the mutex only synchronises the map structure itself against background accessors.
+    stdx::lock_guard<stdx::mutex> lk(_collectionsMutex);
     _collections.clear();
-    // for (CollectionMap::const_iterator i = _collections.begin(); i != _collections.end(); ++i)
-    //     delete i->second;
 }
 
 void DatabaseImpl::close(OperationContext* opCtx, const std::string& reason) {
@@ -169,8 +170,17 @@ void DatabaseImpl::close(OperationContext* opCtx, const std::string& reason) {
     // Clear cache of oplog Collection pointer.
     repl::oplogCheckCloseDatabase(opCtx, this->_this);
 
-    for (const auto& [name, coll] : _collections) {
-        // auto coll = pair.second;
+    // Snapshot under the mutex, invalidate outside it; skip null entries (the map can hold them).
+    std::vector<Collection::Sptr> colls;
+    {
+        stdx::lock_guard<stdx::mutex> lk(_collectionsMutex);
+        for (const auto& [name, coll] : _collections) {
+            if (coll) {
+                colls.push_back(coll);
+            }
+        }
+    }
+    for (const auto& coll : colls) {
         coll->getCursorManager()->invalidateAll(opCtx, true, reason);
     }
 }
@@ -249,32 +259,53 @@ Collection* DatabaseImpl::_getOrCreateCollectionInstance(OperationContext* opCtx
     return coll;
 }
 
-Collection* DatabaseImpl::_createCollectionHandler(OperationContext* opCtx,
-                                                   const NamespaceString& nss,
-                                                   bool createIdIndex,
-                                                   const BSONObj& idIndexSpec,
-                                                   bool forView) {
-    MONGO_LOG(1) << "DatabaseImpl::_createCollectionHandler";
-    if (!forView) {
-        if (auto iter = _collections.find(nss.toString()); iter != _collections.end()) {
-            return iter->second.get();
-        }
-    }
+Collection::Sptr DatabaseImpl::_buildCollectionInstance(OperationContext* opCtx,
+                                                        const NamespaceString& nss,
+                                                        CollectionOptions* optionsOut) {
     auto cce = _dbEntry->getCollectionCatalogEntry(opCtx, nss.toStringData());
     if (!cce) {
         // The collection not exists in the Eloq
         return nullptr;
     }
-    CollectionCatalogEntry::MetaData metadata = cce->getMetaData(opCtx);
-    auto uuid = metadata.options.uuid;
+    CollectionCatalogEntry::MetaData metadata = cce->getMetaData(opCtx);  // yields
     auto rs = cce->getRecordStore();
-    auto collection =
-        std::make_unique<Collection>(opCtx, nss.toStringData(), uuid, cce, rs, _dbEntry);
+    auto collection = std::make_shared<Collection>(
+        opCtx, nss.toStringData(), metadata.options.uuid, cce, rs, _dbEntry);
+    if (optionsOut) {
+        *optionsOut = std::move(metadata.options);
+    }
+    return collection;
+}
 
-    if (forView) {
-        _collectionsView.try_emplace(nss.toString(), std::move(collection));
+Collection* DatabaseImpl::_createCollectionHandler(OperationContext* opCtx,
+                                                   const NamespaceString& nss,
+                                                   bool createIdIndex,
+                                                   const BSONObj& idIndexSpec) {
+    MONGO_LOG(1) << "DatabaseImpl::_createCollectionHandler";
+    {
+        bool hasEntry = false;
+        Collection::Sptr existing;
+        {
+            stdx::lock_guard<stdx::mutex> lk(_collectionsMutex);
+            if (auto iter = _collections.find(nss.toString()); iter != _collections.end()) {
+                hasEntry = true;
+                existing = iter->second;
+            }
+        }
+        if (hasEntry) {
+            // The map can hold null entries; return whatever the entry holds.
+            if (existing && opCtx->recoveryUnit()) {
+                opCtx->recoveryUnit()->pinResource(existing);
+            }
+            return existing.get();
+        }
+    }
+    CollectionOptions options;
+    Collection::Sptr collection = _buildCollectionInstance(opCtx, nss, &options);
+    if (!collection) {
         return nullptr;
     }
+    auto uuid = options.uuid;
 
     if (uuid) {
         // We are not in a WUOW only when we are called from Database::init(). There is no need
@@ -290,8 +321,8 @@ Collection* DatabaseImpl::_createCollectionHandler(OperationContext* opCtx,
     BSONObj fullIdIndexSpec;
     if (createIdIndex) {
         if (collection->requiresIdIndex()) {
-            if (metadata.options.autoIndexId == CollectionOptions::YES ||
-                metadata.options.autoIndexId == CollectionOptions::DEFAULT) {
+            if (options.autoIndexId == CollectionOptions::YES ||
+                options.autoIndexId == CollectionOptions::DEFAULT) {
                 // createCollection() may be called before the in-memory fCV parameter is
                 // initialized, so use the unsafe fCV getter here.
 
@@ -338,12 +369,23 @@ Collection* DatabaseImpl::_createCollectionHandler(OperationContext* opCtx,
         // createSystemIndexes(opCtx, collection.get());
     }
 
-    auto [iter, _] = _collections.try_emplace(nss.toString(), std::move(collection));
-
+    // The catalog read in _buildCollectionInstance can yield the coroutine, so another accessor
+    // may have rebuilt this entry meanwhile; try_emplace keeps the winner and our redundant
+    // instance is discarded.
+    {
+        stdx::lock_guard<stdx::mutex> lk(_collectionsMutex);
+        auto [iter, inserted] = _collections.try_emplace(nss.toString(), collection);
+        if (!inserted) {
+            collection = iter->second;
+        }
+    }
+    if (collection && opCtx->recoveryUnit()) {
+        opCtx->recoveryUnit()->pinResource(collection);
+    }
 
     MONGO_LOG(1) << "[opID]=" << opCtx->getOpID() << "DatabaseImpl::createCollection"
                  << ". create done and handler to collection is available";
-    return iter->second.get();
+    return collection.get();
 }
 
 DatabaseImpl::DatabaseImpl(Database* const this_,
@@ -769,19 +811,38 @@ Status DatabaseImpl::_finishDropCollection(OperationContext* opCtx,
 void DatabaseImpl::_clearCollectionCache(OperationContext* opCtx,
                                          StringData fullns,
                                          const std::string& reason,
-                                         bool collectionGoingAway) {
+                                         bool collectionGoingAway,
+                                         const Collection* onlyIfIs) {
     invariant(_name == nsToDatabaseSubstring(fullns));
-    auto it = _collections.find(fullns);
+    Collection::Sptr evicted;
+    {
+        stdx::lock_guard<stdx::mutex> lk(_collectionsMutex);
+        auto it = _collections.find(fullns);
+        if (it == _collections.end()) {
+            return;
+        }
+        if (onlyIfIs && it->second.get() != onlyIfIs) {
+            // A concurrent accessor already refreshed this entry; keep its rebuild.
+            return;
+        }
+        evicted = std::move(it->second);
+        _collections.erase(it);
+    }
 
-    if (it == _collections.end()) {
+    // The map can hold null entries -- getCollection() checks `it->second` before using it.
+    if (!evicted) {
         return;
     }
 
-    // Takes ownership of the collection
-    // opCtx->recoveryUnit()->registerChange(new RemoveCollectionChange(this, it->second));
+    evicted->getCursorManager()->invalidateAll(opCtx, collectionGoingAway, reason);
 
-    it->second->getCursorManager()->invalidateAll(opCtx, collectionGoingAway, reason);
-    _collections.erase(it);
+    // Destruction is deferred, not immediate: every operation that obtained this Collection via
+    // getCollection() holds its own pin, and pinning here also covers raw pointers this operation
+    // received down a call chain (e.g. _finishDropCollection's argument). The object dies when
+    // the last pinning operation's RecoveryUnit is reset.
+    if (auto* ru = opCtx->recoveryUnit()) {
+        ru->pinResource(std::move(evicted));
+    }
 }
 
 Collection* DatabaseImpl::getCollection(OperationContext* opCtx, StringData ns, bool isForWrite) {
@@ -804,25 +865,41 @@ Collection* DatabaseImpl::getCollection(OperationContext* opCtx,
                  << " exists: " << exists << ", isForWrite: " << isForWrite;
     uassertStatusOK(status);
     if (!exists) {
+        // The table is gone from the transactional catalog (dropped in this transaction or on
+        // another node). Evict any stale cached entry so a later re-create of the same namespace
+        // cannot be served the old object; destruction is deferred via pinResource.
+        _clearCollectionCache(opCtx, nss.ns(), "collection no longer exists", true);
         return nullptr;
     }
 
-    if (auto it = _collections.find(nss.ns()); it != _collections.end() && it->second) {
-        auto found = it->second.get();
-
+    Collection::Sptr found;
+    {
+        stdx::lock_guard<stdx::mutex> lk(_collectionsMutex);
+        if (auto it = _collections.find(nss.ns()); it != _collections.end() && it->second) {
+            found = it->second;
+        }
+    }
+    if (found) {
         if (found->catalogVersion() == version) {
             NamespaceUUIDCache& cache = NamespaceUUIDCache::get(opCtx);
             if (auto uuid = found->uuid()) {
                 cache.ensureNamespaceInCache(nss, uuid.get());
             }
-            return found;
+            // A null RecoveryUnit only occurs on early-boot OperationContexts that never reach
+            // collection code; a future RU-less path reaching here would return an unpinned raw
+            // pointer, losing the lifetime protection.
+            if (auto* ru = opCtx->recoveryUnit()) {
+                ru->pinResource(found);
+            }
+            return found.get();
         } else {
             MONGO_LOG(1) << "nss: " << nss.toStringData()
                          << " version changed. old: " << found->catalogVersion()
                          << ", new: " << version;
             auto& uuidCatalog = UUIDCatalog::get(opCtx);
             uuidCatalog.removeUUIDCatalogEntry(found->uuid().get());
-            _clearCollectionCache(opCtx, nss.ns(), "collection version changed", true);
+            _clearCollectionCache(
+                opCtx, nss.ns(), "collection version changed", true, found.get());
             return _createCollectionHandler(opCtx, nss, false);
         }
     } else {
@@ -1516,21 +1593,23 @@ StatusWith<NamespaceString> DatabaseImpl::makeUniqueCollectionNamespace(
                       << " attempts due to namespace conflicts with existing collections.");
 }
 
-DatabaseImpl::CollectionMapView& DatabaseImpl::collections(OperationContext* opCtx) {
+DatabaseImpl::CollectionMapView DatabaseImpl::collections(OperationContext* opCtx) {
     MONGO_LOG(1) << "DatabaseImpl::collections";
 
     std::vector<std::string> collectionInStorageEngine;
     _dbEntry->getCollectionNamespaces(collectionInStorageEngine);
 
-    _collectionsView.clear();
-
+    // Operation-owned snapshot; no shared state, so no lock. A shared member map here would be
+    // cleared and rebuilt under earlier callers' iterators.
+    CollectionMapView view;
     for (auto& collectionName : collectionInStorageEngine) {
         NamespaceString nss{std::move(collectionName)};
         MONGO_LOG(1) << "nss: " << nss;
-        _createCollectionHandler(opCtx, nss, false, BSONObj{}, true);
+        if (auto coll = _buildCollectionInstance(opCtx, nss, nullptr)) {
+            view.try_emplace(nss.toString(), std::move(coll));
+        }
     }
-
-    return _collectionsView;
+    return view;
 }
 
 
diff --git a/src/mongo/db/catalog/database_impl.h b/src/mongo/db/catalog/database_impl.h
index f1f8fac4cc..75110488b7 100644
--- a/src/mongo/db/catalog/database_impl.h
+++ b/src/mongo/db/catalog/database_impl.h
@@ -41,6 +41,7 @@
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/db/views/view.h"
 #include "mongo/db/views/view_catalog.h"
+#include "mongo/stdx/mutex.h"
 #include "mongo/util/mongoutils/str.h"
 #include "mongo/util/string_map.h"
 
@@ -233,7 +234,7 @@ class DatabaseImpl final : public Database::Impl {
     StatusWith<NamespaceString> makeUniqueCollectionNamespace(OperationContext* opCtx,
                                                               StringData collectionNameModel) final;
 
-    CollectionMapView& collections(OperationContext* opCtx) override;
+    CollectionMapView collections(OperationContext* opCtx) override;
 
     // const CollectionMap& collections() const override;
     // CollectionMap::const_iterator begin() const override;
@@ -261,8 +262,19 @@ class DatabaseImpl final : public Database::Impl {
     Collection* _createCollectionHandler(OperationContext* opCtx,
                                          const NamespaceString& nss,
                                          bool createIdIndex,
-                                         const BSONObj& idIndexSpec = BSONObj{},
-                                         bool forView = false);
+                                         const BSONObj& idIndexSpec = BSONObj{});
+
+    /**
+     * Builds a Collection instance for 'nss' from the storage catalog without touching the
+     * shared _collections cache. Returns null if the collection does not exist. Yields during
+     * the catalog read. 'optionsOut', when non-null, receives the collection options that were
+     * read. (Options, not the full CollectionCatalogEntry::MetaData: that type would pull
+     * kv-layer headers into this one and an undefined KVPrefix reference into every library
+     * that includes it.)
+     */
+    Collection::Sptr _buildCollectionInstance(OperationContext* opCtx,
+                                              const NamespaceString& nss,
+                                              CollectionOptions* optionsOut);
 
     /**
      * Throws if there is a reason 'ns' cannot be created as a user collection.
@@ -275,11 +287,15 @@ class DatabaseImpl final : public Database::Impl {
      * Deregisters and invalidates all cursors on collection 'fullns'.  Callers must specify
      * 'reason' for why the cache is being cleared. If 'collectionGoingAway' is false,
      * unpinned cursors will not be killed.
+     *
+     * When 'onlyIfIs' is non-null, the entry is only evicted if it still holds that object; a
+     * concurrent accessor may already have refreshed the entry, and its rebuild must stay.
      */
     void _clearCollectionCache(OperationContext* opCtx,
                                StringData fullns,
                                const std::string& reason,
-                               bool collectionGoingAway);
+                               bool collectionGoingAway,
+                               const Collection* onlyIfIs = nullptr);
 
     /**
      * Completes a collection drop by removing all the indexes and removing the collection itself
@@ -316,9 +332,22 @@ class DatabaseImpl final : public Database::Impl {
     // This variable may only be read/written while the database is locked in MODE_X.
     std::unique_ptr<PseudoRandom> _uniqueCollectionNamespacePseudoRandom;
 
-    CollectionMap _collections;  // owner
-    CollectionMapView _collectionsView;
-    // mutable std::mutex _collectionsMutex;
+    CollectionMap _collections;  // shared owner; entries evicted on catalog version change
+
+    // Collection lifetime: getCollection() refreshes the cached Collection whenever the catalog
+    // version moves, which happens constantly under concurrent DDL, and callers hold raw
+    // Collection* across coroutine yields and into RecoveryUnit onCommit callbacks
+    // (EloqLockerNoop provides none of the exclusion upstream's lock manager gave this code).
+    // Entries are therefore shared_ptr: every getCollection() pins the shared_ptr on the caller's
+    // RecoveryUnit until its operation fully ends, so eviction is a plain map erase and the old
+    // object is destroyed only after the last operation using it completes.
+    //
+    // _collectionsMutex serialises the pure in-memory map operations only -- request-thread
+    // coroutines interleave at yields, while background threads (e.g. the TTLMonitor) mutate the
+    // maps concurrently for real. It must never be held across a call that can yield the
+    // coroutine (readCatalog/getMetaData); holding a mutex across a yield can deadlock the
+    // shared worker thread.
+    mutable stdx::mutex _collectionsMutex;
 
     DurableViewCatalogImpl _durableViews;  // interface for system.views operations
     ViewCatalog _views;                    // in-memory representation of _durableViews
diff --git a/src/mongo/db/clientcursor.cpp b/src/mongo/db/clientcursor.cpp
index fca4761b28..d0bfece6d4 100644
--- a/src/mongo/db/clientcursor.cpp
+++ b/src/mongo/db/clientcursor.cpp
@@ -89,6 +89,7 @@ ClientCursor::ClientCursor(ClientCursorParams params,
       _cursorManager(cursorManager),
       _originatingCommand(params.originatingCommandObj),
       _queryOptions(params.queryOptions),
+      _collectionPin(std::move(params.collectionPin)),
       _exec(std::move(params.exec)),
       _operationUsingCursor(operationUsingCursor),
       _lastUseDate(now) {
diff --git a/src/mongo/db/clientcursor.h b/src/mongo/db/clientcursor.h
index a07e54e152..b2438a6dff 100644
--- a/src/mongo/db/clientcursor.h
+++ b/src/mongo/db/clientcursor.h
@@ -91,6 +91,15 @@ struct ClientCursorParams {
     const repl::ReadConcernLevel readConcernLevel;
     int queryOptions = 0;
     BSONObj originatingCommandObj;
+
+    // EloqDoc: shared ownership of the Collection the executor's stages reference, for cursors
+    // whose getMore path never re-resolves the collection (globally-managed aggregation cursors,
+    // getmore_cmd.cpp's lock-free branch). Per-operation RecoveryUnit pins only cover the
+    // operation that resolved the collection; a stashed executor outlives it, and a catalog
+    // version bump would otherwise destroy the object under the cursor. Null for cursors whose
+    // getMore re-resolves and re-pins (collection-managed cursors), where eviction kills the
+    // cursor before any dereference.
+    std::shared_ptr<const void> collectionPin;
 };
 
 /**
@@ -310,6 +319,11 @@ class ClientCursor {
     // Unused maxTime budget for this cursor.
     Microseconds _leftoverMaxTimeMicros = Microseconds::max();
 
+    // Keeps the Collection referenced by '_exec' alive for the cursor's whole life; see
+    // ClientCursorParams::collectionPin. Declared before '_exec' so the executor is destroyed
+    // first.
+    std::shared_ptr<const void> _collectionPin;
+
     // The underlying query execution machinery. Must be non-null.
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> _exec;
 
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index fdd88f11a8..14520ad5a7 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -325,6 +325,9 @@ Status runAggregate(OperationContext* opCtx,
     unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec;
     boost::intrusive_ptr<ExpressionContext> expCtx;
     Pipeline* unownedPipeline;
+    // EloqDoc: shared ownership of the collection the pipeline's input executor references,
+    // transferred to the globally-managed cursor below; see ClientCursorParams::collectionPin.
+    Collection::Sptr collectionPin;
     auto curOp = CurOp::get(opCtx);
     {
         const LiteParsedPipeline liteParsedPipeline(request);
@@ -391,6 +394,7 @@ Status runAggregate(OperationContext* opCtx,
         }
 
         Collection* collection = ctx ? ctx->getCollection() : nullptr;
+        collectionPin = collection ? collection->sharedFromThisIfShared() : nullptr;
 
         // For change streams, the UUID will already have been set for the original namespace.
         if (!liteParsedPipeline.hasChangeStream()) {
@@ -537,6 +541,7 @@ Status runAggregate(OperationContext* opCtx,
         AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(),
         repl::ReadConcernArgs::get(opCtx).getLevel(),
         cmdObj);
+    cursorParams.collectionPin = std::move(collectionPin);
     if (expCtx->tailableMode == TailableModeEnum::kTailableAndAwaitData) {
         cursorParams.setTailable(true);
         cursorParams.setAwaitData(true);
diff --git a/src/mongo/db/concurrency/write_conflict_exception.h b/src/mongo/db/concurrency/write_conflict_exception.h
index 193ebfb112..2230554ba2 100644
--- a/src/mongo/db/concurrency/write_conflict_exception.h
+++ b/src/mongo/db/concurrency/write_conflict_exception.h
@@ -91,9 +91,14 @@ auto writeConflictRetry(OperationContext* opCtx, StringData opStr, StringData ns
             return f();
         } catch (WriteConflictException const&) {
             CurOp::get(opCtx)->debug().additiveMetrics.incrementWriteConflicts(1);
+            // Abandon the snapshot before backing off, not after: under Eloq the snapshot is a
+            // transaction whose accumulated catalog/key intents are exactly what the conflict
+            // winner is waiting to drain. Sleeping first parks this loser on top of its locks
+            // for the whole backoff window, starving the winner it is about to retry against;
+            // abandoning first hands the winner the backoff window to finish.
+            opCtx->recoveryUnit()->abandonSnapshot();
             WriteConflictException::logAndBackoff(attempts, opStr, ns);
             ++attempts;
-            opCtx->recoveryUnit()->abandonSnapshot();
         }
     }
 }
diff --git a/src/mongo/db/modules/eloq/SConscript b/src/mongo/db/modules/eloq/SConscript
index 36f941cbc7..4f911642b3 100644
--- a/src/mongo/db/modules/eloq/SConscript
+++ b/src/mongo/db/modules/eloq/SConscript
@@ -388,6 +388,16 @@ env.Library(
     SYSLIBDEPS=eloq_dependencies,
 )
 
+env.CppUnitTest(
+    target="eloq_util_test",
+    source=[
+        "src/base/eloq_util_test.cpp",
+    ],
+    LIBDEPS=[
+        "storage_eloq_core",
+    ],
+)
+
 env.Library(
     target="storage_eloq",
     source=[
diff --git a/src/mongo/db/modules/eloq/data_substrate b/src/mongo/db/modules/eloq/data_substrate
index 63db0ecbe3..5a768968a5 160000
--- a/src/mongo/db/modules/eloq/data_substrate
+++ b/src/mongo/db/modules/eloq/data_substrate
@@ -1 +1 @@
-Subproject commit 63db0ecbe3cd04e23282aa61af3961491647bddf
+Subproject commit 5a768968a552f6958fb07c948f664bbdaa3895ea
diff --git a/src/mongo/db/modules/eloq/src/base/eloq_util.cpp b/src/mongo/db/modules/eloq/src/base/eloq_util.cpp
index 395313199e..e4d3a6390e 100644
--- a/src/mongo/db/modules/eloq/src/base/eloq_util.cpp
+++ b/src/mongo/db/modules/eloq/src/base/eloq_util.cpp
@@ -8,12 +8,31 @@
 
 
 namespace mongo {
+void ThrowIfWriteConflict(txservice::TxErrorCode txErr) {
+    switch (txErr) {
+        case txservice::TxErrorCode::WRITE_WRITE_CONFLICT:
+        case txservice::TxErrorCode::OCC_BREAK_REPEATABLE_READ:
+        case txservice::TxErrorCode::DEAD_LOCK_ABORT:
+        case txservice::TxErrorCode::GET_RANGE_ID_ERROR:
+        case txservice::TxErrorCode::SI_R4W_ERR_KEY_WAS_UPDATED:
+        case txservice::TxErrorCode::UPSERT_TABLE_ACQUIRE_WRITE_INTENT_FAIL:
+            // Like wtRCToStatus_slow.
+            throw WriteConflictException();
+        default:
+            return;
+    }
+}
+
 Status TxErrorCodeToMongoStatus(txservice::TxErrorCode txErr) {
     if (MONGO_likely(txErr == txservice::TxErrorCode::NO_ERROR))
         return Status::OK();
 
     log() << "Eloq engine error report: " << txservice::TxErrorMessage(txErr);
 
+    // Conflicts leave through an exception rather than a Status; see the header
+    // for why the distinction is load-bearing.
+    ThrowIfWriteConflict(txErr);
+
     ErrorCodes::Error err;
     switch (txErr) {
         case txservice::TxErrorCode::TX_INIT_FAIL:
@@ -32,15 +51,8 @@ Status TxErrorCodeToMongoStatus(txservice::TxErrorCode txErr) {
         case txservice::TxErrorCode::READ_WRITE_CONFLICT:
             err = ErrorCodes::SnapshotUnavailable;
             break;
-        case txservice::TxErrorCode::WRITE_WRITE_CONFLICT:
-        case txservice::TxErrorCode::OCC_BREAK_REPEATABLE_READ:
-        case txservice::TxErrorCode::DEAD_LOCK_ABORT:
-        case txservice::TxErrorCode::GET_RANGE_ID_ERROR:
-        case txservice::TxErrorCode::SI_R4W_ERR_KEY_WAS_UPDATED:
-        case txservice::TxErrorCode::UPSERT_TABLE_ACQUIRE_WRITE_INTENT_FAIL:
-            // Like wtRCToStatus_slow.
-            throw WriteConflictException();
-            break;
+        // The conflict group is handled by the ThrowIfWriteConflict() call
+        // above and never reaches this switch.
         case txservice::TxErrorCode::OUT_OF_MEMORY:
             err = ErrorCodes::ExceededMemoryLimit;
             break;
diff --git a/src/mongo/db/modules/eloq/src/base/eloq_util.h b/src/mongo/db/modules/eloq/src/base/eloq_util.h
index de82460b1c..f11ecbb98d 100644
--- a/src/mongo/db/modules/eloq/src/base/eloq_util.h
+++ b/src/mongo/db/modules/eloq/src/base/eloq_util.h
@@ -150,6 +150,30 @@ inline std::pair<std::string, std::set<txservice::TableName>> ExtractReadyIndexe
 
 
 namespace mongo {
+/**
+ * Throws mongo::WriteConflictException if txErr belongs to the group of tx
+ * service errors that mean "another transaction got there first"; returns
+ * normally for every other code, including NO_ERROR.
+ *
+ * The group is wider than its name suggests: besides WRITE_WRITE_CONFLICT it
+ * covers OCC_BREAK_REPEATABLE_READ, DEAD_LOCK_ABORT, GET_RANGE_ID_ERROR,
+ * SI_R4W_ERR_KEY_WAS_UPDATED and UPSERT_TABLE_ACQUIRE_WRITE_INTENT_FAIL. All of
+ * them are resolved the same way -- abort the transaction and let the command
+ * layer re-run it -- which is what WriteConflictException means to mongo.
+ *
+ * This exists for call sites that must surface a conflict immediately while
+ * keeping their own retry loop for transient errors, so they cannot simply run
+ * TxErrorCodeToMongoStatus through uassertStatusOK. TxErrorCodeToMongoStatus
+ * calls this first, so the two can never drift apart.
+ *
+ * The thrown type matters: writeConflictRetry catches only
+ * WriteConflictException, which is `final : public DBException` and has no
+ * inheritance relationship with ExceptionFor<ErrorCodes::WriteConflict>.
+ * Returning a Status carrying ErrorCodes::WriteConflict would sail straight
+ * through every writeConflictRetry boundary.
+ */
+void ThrowIfWriteConflict(txservice::TxErrorCode txErr);
+
 Status TxErrorCodeToMongoStatus(txservice::TxErrorCode txErr);
 
 inline constexpr std::string_view kMongoCatalogTableNameSV{"_mdb_catalog"};
diff --git a/src/mongo/db/modules/eloq/src/base/eloq_util_test.cpp b/src/mongo/db/modules/eloq/src/base/eloq_util_test.cpp
new file mode 100644
index 0000000000..e41e969404
--- /dev/null
+++ b/src/mongo/db/modules/eloq/src/base/eloq_util_test.cpp
@@ -0,0 +1,156 @@
+/**
+ *    Copyright (C) 2025 EloqData Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the license:
+ *    1. GNU Affero General Public License, version 3, as published by the Free
+ *    Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/base/error_codes.h"
+#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/assert_util.h"
+
+#include "mongo/db/modules/eloq/src/base/eloq_util.h"
+
+namespace mongo {
+namespace {
+
+using txservice::TxErrorCode;
+
+// The tx service errors that mean "another transaction got there first". Every
+// one of them must leave a catalog record store loop through an exception, or
+// the loop retries a conflict inside the losing transaction while it still
+// holds the catalog read lock the winner is waiting on.
+const TxErrorCode kConflictCodes[] = {
+    TxErrorCode::WRITE_WRITE_CONFLICT,
+    TxErrorCode::OCC_BREAK_REPEATABLE_READ,
+    TxErrorCode::DEAD_LOCK_ABORT,
+    TxErrorCode::GET_RANGE_ID_ERROR,
+    TxErrorCode::SI_R4W_ERR_KEY_WAS_UPDATED,
+    TxErrorCode::UPSERT_TABLE_ACQUIRE_WRITE_INTENT_FAIL,
+};
+
+// Errors the catalog record store's bounded retry loop must keep. Chief among
+// them READ_CATALOG_FAIL, which means "the catalog entry is not resident, a
+// fetch has been issued, try again" -- ejecting it would replace a working
+// retry with a client-visible error.
+const TxErrorCode kRetriedInLoopCodes[] = {
+    TxErrorCode::READ_CATALOG_FAIL,
+    TxErrorCode::NG_TERM_CHANGED,
+    TxErrorCode::REQUEST_LOST,
+    TxErrorCode::INTERNAL_ERR_TIMEOUT,
+};
+
+/**
+ * Mirrors the catch clause inside writeConflictRetry (write_conflict_exception.h).
+ * Returns true if the thrown object is caught by `catch (WriteConflictException const&)`,
+ * which is the only handler that loop has.
+ */
+bool caughtByWriteConflictRetry(TxErrorCode code) {
+    try {
+        ThrowIfWriteConflict(code);
+        return false;
+    } catch (const WriteConflictException&) {
+        return true;
+    } catch (const DBException&) {
+        return false;
+    }
+}
+
+TEST(EloqUtil, ConflictsAreCaughtByWriteConflictRetry) {
+    for (auto code : kConflictCodes) {
+        ASSERT_TRUE(caughtByWriteConflictRetry(code))
+            << "TxErrorCode " << static_cast<int>(code)
+            << " does not reach a writeConflictRetry boundary";
+    }
+}
+
+TEST(EloqUtil, ConflictIsNotExceptionForWriteConflict) {
+    // WriteConflictException is `final : public DBException`, while
+    // ExceptionFor<ErrorCodes::WriteConflict> resolves to
+    // ExceptionForImpl<code, AssertionException>. They are siblings with no
+    // inheritance relationship, so returning a Status carrying
+    // ErrorCodes::WriteConflict and uassertStatusOK-ing it would sail straight
+    // through writeConflictRetry. A test that only checked the error code
+    // would pass under that defective form; this one does not.
+    for (auto code : kConflictCodes) {
+        bool wrongType = false;
+        bool rightType = false;
+        try {
+            ThrowIfWriteConflict(code);
+        } catch (const ExceptionFor<ErrorCodes::WriteConflict>&) {
+            wrongType = true;
+        } catch (const WriteConflictException&) {
+            rightType = true;
+        }
+        ASSERT_FALSE(wrongType) << "TxErrorCode " << static_cast<int>(code)
+                                << " threw ExceptionFor<WriteConflict>";
+        ASSERT_TRUE(rightType);
+    }
+}
+
+TEST(EloqUtil, DeadlockAbortIsInTheConflictGroup) {
+    // The live reproduction logged exactly this code being swallowed for 90
+    // seconds by the in-transaction retry loop. Retrying it inside the same
+    // transaction is unsound: the transaction has already been chosen as the
+    // deadlock victim.
+    ASSERT_TRUE(caughtByWriteConflictRetry(TxErrorCode::DEAD_LOCK_ABORT));
+}
+
+TEST(EloqUtil, TransientErrorsDoNotThrow) {
+    // These stay inside the catalog record store's bounded retry loop.
+    for (auto code : kRetriedInLoopCodes) {
+        ThrowIfWriteConflict(code);  // must return normally
+    }
+    ThrowIfWriteConflict(TxErrorCode::NO_ERROR);
+}
+
+TEST(EloqUtil, ConverterAndHelperCannotDrift) {
+    // TxErrorCodeToMongoStatus calls ThrowIfWriteConflict first, so the group
+    // has exactly one definition. Anything the helper throws for, the
+    // converter must throw for too.
+    for (auto code : kConflictCodes) {
+        bool threw = false;
+        try {
+            TxErrorCodeToMongoStatus(code);
+        } catch (const WriteConflictException&) {
+            threw = true;
+        }
+        ASSERT_TRUE(threw) << "TxErrorCodeToMongoStatus returned a Status for "
+                              "conflict code "
+                           << static_cast<int>(code);
+    }
+}
+
+TEST(EloqUtil, ConverterStillReturnsStatusForOtherErrors) {
+    ASSERT_OK(TxErrorCodeToMongoStatus(TxErrorCode::NO_ERROR));
+
+    // READ_CATALOG_FAIL has no dedicated mapping, so it lands in the default
+    // arm. What matters is that it comes back as a Status rather than an
+    // exception: the record store loop tests the code and retries.
+    Status readCatalogFail = TxErrorCodeToMongoStatus(TxErrorCode::READ_CATALOG_FAIL);
+    ASSERT_NOT_OK(readCatalogFail);
+
+    ASSERT_EQUALS(TxErrorCodeToMongoStatus(TxErrorCode::DUPLICATE_KEY).code(),
+                  ErrorCodes::DuplicateKey);
+    ASSERT_EQUALS(TxErrorCodeToMongoStatus(TxErrorCode::OUT_OF_MEMORY).code(),
+                  ErrorCodes::ExceededMemoryLimit);
+    ASSERT_EQUALS(TxErrorCodeToMongoStatus(TxErrorCode::READ_WRITE_CONFLICT).code(),
+                  ErrorCodes::SnapshotUnavailable);
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/modules/eloq/src/eloq_index.cpp b/src/mongo/db/modules/eloq/src/eloq_index.cpp
index bfb7c3549b..caa483ec1d 100644
--- a/src/mongo/db/modules/eloq/src/eloq_index.cpp
+++ b/src/mongo/db/modules/eloq/src/eloq_index.cpp
@@ -218,6 +218,14 @@ class EloqIndexCursor final : public SortedDataInterface::Cursor {
     void detachFromOperationContext() override {
         MONGO_LOG(1) << "EloqIndexCursor::detachFromOperationContext";
         assert(_opCtx);
+        // Close the scan here rather than in save(): detach is the point where the cursor
+        // leaves its operation, so this is the last moment the transaction that opened the
+        // scan is still live. A stashed cursor would otherwise keep an EloqCursor bound to
+        // that txm, which is recycled once the transaction commits, and its next batch
+        // request would land on a foreign transaction's queue. save() must not do this --
+        // DeleteStage saves state once per deleted document, so closing there costs one
+        // scan close and re-seek per document. restore() re-seeks exclusively after '_key'.
+        _cursor.reset();
         _opCtx = nullptr;
         _ru = nullptr;
     }
diff --git a/src/mongo/db/modules/eloq/src/eloq_record_store.cpp b/src/mongo/db/modules/eloq/src/eloq_record_store.cpp
index d069b1003e..445eb9cb91 100644
--- a/src/mongo/db/modules/eloq/src/eloq_record_store.cpp
+++ b/src/mongo/db/modules/eloq/src/eloq_record_store.cpp
@@ -129,6 +129,10 @@ class EloqCatalogRecordStoreCursor : public SeekableRecordCursor {
         MONGO_UNREACHABLE;
     }
 
+    // Catalog scans are fully prefetched at construction and are never stashed across wire
+    // operations today; if one ever is (a future catalog cursor surviving a getMore), these
+    // MONGO_UNREACHABLEs abort the process -- implement the save/restore contract like
+    // EloqRecordStoreCursor's (close at save, lazy re-seek at next use) before allowing that.
     void save() override {
         MONGO_UNREACHABLE;
     }
@@ -224,8 +228,17 @@ void EloqCatalogRecordStore::deleteRecord(OperationContext* opCtx, const RecordI
     for (uint16_t i = 1; i < kMaxRetryLimit; ++i) {
         auto [exist, errorCode] = ru->readCatalog(catalogKey, catalogRecord, true);
         if (errorCode != txservice::TxErrorCode::NO_ERROR) {
-            MONGO_LOG(1) << "Eloq readCatalog error with write intent. Another transaction "
-                            "may do DDL on the same table.";
+            // A conflict must leave this loop and abort the transaction, so the
+            // catalog read lock this transaction holds is released and the
+            // competing DDL can upgrade its write intent. Retrying here holds
+            // that lock for the whole retry budget and can wedge both sides.
+            // Retry belongs at the command layer, which owns the transaction.
+            ThrowIfWriteConflict(errorCode);
+            // Transient errors -- most importantly READ_CATALOG_FAIL, which
+            // means "the catalog entry is not resident, a fetch has been
+            // issued, try again" -- stay in the loop below.
+            MONGO_LOG(1) << "Eloq readCatalog error with write intent: "
+                         << txservice::TxErrorMessage(errorCode);
         } else {
             if (!exist) {
                 return;
@@ -282,16 +295,16 @@ StatusWith<RecordId> EloqCatalogRecordStore::insertRecord(
 
     auto [exist, errorCode] = ru->readCatalog(catalogKey, catalogRecord, true);
     if (errorCode != txservice::TxErrorCode::NO_ERROR) {
-        if (errorCode == txservice::TxErrorCode::WRITE_WRITE_CONFLICT) {
-            MONGO_LOG(1) << "Eloq readCatalog error with write intent. Another transaction "
-                            "may do DDL on the same table.";
-            return {ErrorCodes::WriteConflict,
-                    "[Create Table] Another transaction may do DDL on the same table"};
-        } else {
-            MONGO_LOG(0) << "Eloq readCatalog error with write intent."
-                         << txservice::TxErrorMessage(errorCode);
-            return {ErrorCodes::InternalError, txservice::TxErrorMessage(errorCode)};
-        }
+        // The whole conflict group has to leave as a WriteConflictException. Returning a
+        // Status carrying ErrorCodes::WriteConflict looks equivalent but is not: at the
+        // first uassertStatusOK it becomes ExceptionFor<ErrorCodes::WriteConflict>, a
+        // sibling type that no writeConflictRetry catches. The previous form also mapped
+        // DEAD_LOCK_ABORT and UPSERT_TABLE_ACQUIRE_WRITE_INTENT_FAIL to InternalError,
+        // which no caller retries at all.
+        ThrowIfWriteConflict(errorCode);
+        MONGO_LOG(0) << "Eloq readCatalog error with write intent."
+                     << txservice::TxErrorMessage(errorCode);
+        return {ErrorCodes::InternalError, txservice::TxErrorMessage(errorCode)};
     } else {
         if (exist) {
             const char* msg = "Collection already exists in Eloq storage engine";
@@ -345,8 +358,11 @@ Status EloqCatalogRecordStore::updateRecord(OperationContext* opCtx,
     for (uint16_t i = 1; i < kMaxRetryLimit; ++i) {
         auto [exist, errorCode] = ru->readCatalog(catalogKey, catalogRecord, true);
         if (errorCode != txservice::TxErrorCode::NO_ERROR) {
-            MONGO_LOG(1) << "Eloq readCatalog error with write intent. Another transaction "
-                            "may do DDL on the same table.";
+            // See deleteRecord for why a conflict must not be retried here.
+            ThrowIfWriteConflict(errorCode);
+            // Transient errors stay in the loop below.
+            MONGO_LOG(1) << "Eloq readCatalog error with write intent: "
+                         << txservice::TxErrorMessage(errorCode);
         } else {
             if (!exist) {
                 return {ErrorCodes::InternalError, "Try to Update a non-exist table"};
@@ -531,6 +547,14 @@ class EloqRecordStoreCursor : public SeekableRecordCursor {
     void detachFromOperationContext() override {
         MONGO_LOG(1) << "EloqRecordStoreCursor::detachFromOperationContext";
         assert(_opCtx);
+        // Close the scan here rather than in save(): detach is the point where the cursor
+        // leaves its operation, so this is the last moment the transaction that opened the
+        // scan is still live. A stashed cursor would otherwise keep an EloqCursor bound to
+        // that txm, which is recycled once the transaction commits, and its next batch
+        // request would land on a foreign transaction's queue. save() must not do this --
+        // DeleteStage saves state once per deleted document, so closing there costs one
+        // scan close and re-seek per document. next() re-seeks from _lastMongoKey.
+        _cursor.reset();
         _opCtx = nullptr;
         _ru = nullptr;
     }
diff --git a/src/mongo/db/modules/eloq/src/eloq_recovery_unit.cpp b/src/mongo/db/modules/eloq/src/eloq_recovery_unit.cpp
index cb095dbf48..4417c47175 100644
--- a/src/mongo/db/modules/eloq/src/eloq_recovery_unit.cpp
+++ b/src/mongo/db/modules/eloq/src/eloq_recovery_unit.cpp
@@ -67,6 +67,36 @@ namespace {
 
 std::atomic<uint64_t> nextSnapshotId{1};
 
+/**
+ * Restores an entry erased from _unreadyTableMap when the enclosing
+ * WriteUnitOfWork rolls back.
+ *
+ * The staged unready-index metadata lives only in that map: updateRecord takes
+ * an early return for a catalog object carrying a not-yet-ready index, so the
+ * catalog record itself never gets the index. An erase that is not undone on
+ * abort therefore destroys the only copy, and the in-process retry falls back
+ * to the pre-build metadata, which has neither the unready nor the ready index.
+ */
+class RestoreUnreadyTableChange : public RecoveryUnit::Change {
+public:
+    RestoreUnreadyTableChange(std::unordered_map<txservice::TableName, BSONObj>* map,
+                              txservice::TableName tableName,
+                              BSONObj obj)
+        : _map(map), _tableName(std::move(tableName)), _obj(std::move(obj)) {}
+
+    void commit(boost::optional<Timestamp>) override {}
+
+    void rollback() override {
+        _map->insert_or_assign(_tableName, _obj);
+    }
+
+private:
+    std::unordered_map<txservice::TableName, BSONObj>* const _map;
+    const txservice::TableName _tableName;
+    // Owned; getOwned() was called before the entry went into the map.
+    const BSONObj _obj;
+};
+
 }  // namespace
 
 txservice::AlterTableInfo getAlterTableInfo(std::string_view oldMetadata,
@@ -156,6 +186,9 @@ void EloqRecoveryUnit::reset() {
     _changes.clear();
     _discoveredTableMap.clear();
     _unreadyTableMap.clear();
+    // Last: evicted Collections pinned by the previous operation may be destroyed here, and
+    // nothing torn down above may touch them afterwards.
+    clearPinnedResources();
 }
 
 EloqRecoveryUnit::~EloqRecoveryUnit() {
@@ -628,6 +661,11 @@ Status EloqRecoveryUnit::createTable(const txservice::TableName& tableName,
         case txservice::UpsertResult::Failed:
             MONGO_LOG(1) << "UpsertTableTxRequest error. UpsertTableOp on multiple nodes at the "
                             "same time may conflict and then backoff.";
+            // A lock conflict has to leave as a WriteConflictException, not as a Status: a
+            // Status carrying ErrorCodes::WriteConflict becomes ExceptionFor<WriteConflict>
+            // at the first uassertStatusOK, which no writeConflictRetry catches. See
+            // ThrowIfWriteConflict in eloq_util.h.
+            ThrowIfWriteConflict(upsertTableTxReq.ErrorCode());
             return {ErrorCodes::Error::WriteConflict, upsertTableTxReq.ErrorMsg()};
             break;
         case txservice::UpsertResult::Unverified:
@@ -672,6 +710,9 @@ Status EloqRecoveryUnit::dropTable(const txservice::TableName& tableName,
         case txservice::UpsertResult::Failed:
             MONGO_LOG(1) << "UpsertTableTxRequest error. Drop temporary table "
                          << tableName.StringView() << " failed at launch.";
+            // See createTable: a conflict must be raised as WriteConflictException so the
+            // command layer retries it, rather than surfacing as InternalError.
+            ThrowIfWriteConflict(dropTableTxReq.ErrorCode());
             return {ErrorCodes::Error::InternalError, dropTableTxReq.ErrorMsg()};
             break;
         case txservice::UpsertResult::Unverified:
@@ -812,6 +853,11 @@ Status EloqRecoveryUnit::updateTable(const txservice::TableName& tableName,
                     MONGO_LOG(1)
                         << "UpsertTableTxRequest error. UpsertTableOp on multiple nodes at the "
                            "same time may conflict and then backoff.";
+                    // A conflict here reaches ~MultiIndexBlockImpl's cleanup loop, which
+                    // retries only WriteConflictException and hits a fatal assertion on
+                    // anything else (index_create_impl.cpp:194). Reporting a lock conflict as
+                    // InternalError therefore kills the node. See ThrowIfWriteConflict.
+                    ThrowIfWriteConflict(txErr);
                     return {ErrorCodes::Error::InternalError, upsertTableTxReq.ErrorMsg()};
                 }
                 break;
@@ -830,7 +876,25 @@ Status EloqRecoveryUnit::updateTable(const txservice::TableName& tableName,
     }
 }
 void EloqRecoveryUnit::eraseUnreadyTable(const txservice::TableName& tableName) {
-    _unreadyTableMap.erase(tableName);
+    auto iter = _unreadyTableMap.find(tableName);
+    if (iter == _unreadyTableMap.end()) {
+        return;
+    }
+
+    // The erase must be undone if the enclosing WriteUnitOfWork aborts; see
+    // RestoreUnreadyTableChange. The reachable case -- indexer.commit() inside
+    // the writeConflictRetry of createIndexes -- always runs in one. If there
+    // is no unit of work there is also no rollback path that could run the
+    // change, so the plain erase is all that can be done.
+    if (_inUnitOfWork) {
+        registerChange(new RestoreUnreadyTableChange{&_unreadyTableMap, tableName, iter->second});
+    } else {
+        MONGO_LOG(1) << "eraseUnreadyTable outside a unit of work; the erase cannot be rolled "
+                        "back. tableName: "
+                     << tableName.StringView();
+    }
+
+    _unreadyTableMap.erase(iter);
 }
 BSONObj EloqRecoveryUnit::getUnreadyTable(const txservice::TableName& tableName) {
     auto iter = _unreadyTableMap.find(tableName);
diff --git a/src/mongo/db/pipeline/document_source_cursor.cpp b/src/mongo/db/pipeline/document_source_cursor.cpp
index 10fe39a148..82542d0a8d 100644
--- a/src/mongo/db/pipeline/document_source_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_cursor.cpp
@@ -252,19 +252,24 @@ void DocumentSourceCursor::doDispose() {
 void DocumentSourceCursor::cleanupExecutor() {
     invariant(_exec);
     auto* opCtx = pExpCtx->opCtx;
-    // We need to be careful to not use AutoGetCollection here, since we only need the lock to
-    // protect potential access to the Collection's CursorManager, and AutoGetCollection may throw
-    // if this namespace has since turned into a view. Using Database::getCollection() will simply
-    // return nullptr if the collection has since turned into a view. In this case, '_exec' will
-    // already have been marked as killed when the collection was dropped, and we won't need to
-    // access the CursorManager to properly dispose of it.
+    // Keep upstream's lock discipline: PlanExecutor::dispose()'s deregistration path asserts a
+    // MODE_IS collection lock and non-Eloq engines rely on it for exclusion against concurrent
+    // invalidation. What upstream additionally did -- re-resolving the namespace via
+    // Database::getCollection() to find the CursorManager, chosen because upstream's version
+    // cannot throw -- is gone: EloqDoc's getCollection performs a transactional catalog read
+    // that can throw under concurrent DDL (Pipeline::dispose() turns that into std::terminate),
+    // and re-resolution can return a rebuilt Collection whose CursorManager never saw '_exec'.
+    //
+    // '_cursorManager' was captured from the instance '_exec' registered with;
+    // '_collectionPin' keeps it alive in the shared-ownership case. If the collection was
+    // destroyed or evicted instead, its invalidateAll() killed '_exec' first and killed
+    // executors never dereference the manager argument. Lock::DBLock (not AutoGetDb) so the
+    // whole path stays free of catalog and database-holder access.
     UninterruptibleLockGuard noInterrupt(opCtx->lockState());
     auto lockMode = getLockModeForQuery(opCtx);
-    AutoGetDb dbLock(opCtx, _exec->nss().db(), lockMode);
+    Lock::DBLock dbLock(opCtx, _exec->nss().db(), lockMode);
     Lock::CollectionLock collLock(opCtx->lockState(), _exec->nss().ns(), lockMode);
-    auto collection = dbLock.getDb() ? dbLock.getDb()->getCollection(opCtx, _exec->nss()) : nullptr;
-    auto cursorManager = collection ? collection->getCursorManager() : nullptr;
-    _exec->dispose(opCtx, cursorManager);
+    _exec->dispose(opCtx, _cursorManager);
 
     // Not freeing _exec if we're in explain mode since it will be used in serialize() to gather
     // execution stats.
@@ -300,6 +305,8 @@ DocumentSourceCursor::DocumentSourceCursor(
     const intrusive_ptr<ExpressionContext>& pCtx)
     : DocumentSource(pCtx),
       _docsAddedToBatches(0),
+      _cursorManager(collection ? collection->getCursorManager() : nullptr),
+      _collectionPin(collection ? collection->sharedFromThisIfShared() : nullptr),
       _exec(std::move(exec)),
       _outputSorts(_exec->getOutputSorts()) {
 
diff --git a/src/mongo/db/pipeline/document_source_cursor.h b/src/mongo/db/pipeline/document_source_cursor.h
index fc9b37f940..2175285e90 100644
--- a/src/mongo/db/pipeline/document_source_cursor.h
+++ b/src/mongo/db/pipeline/document_source_cursor.h
@@ -40,6 +40,8 @@
 
 namespace mongo {
 
+class CursorManager;
+
 /**
  * Constructs and returns Documents from the BSONObj objects produced by a supplied PlanExecutor.
  */
@@ -197,6 +199,25 @@ class DocumentSourceCursor final : public DocumentSource {
     boost::intrusive_ptr<DocumentSourceLimit> _limit;
     long long _docsAddedToBatches;  // for _limit enforcement
 
+    // EloqDoc: the CursorManager '_exec' registered with, captured at construction from the same
+    // Collection instance the executor embeds. cleanupExecutor() deregisters from it instead of
+    // re-resolving the namespace (EloqDoc's Database::getCollection() performs a transactional
+    // catalog read that can throw on the no-throw disposal path, and re-resolution can return a
+    // rebuilt Collection whose CursorManager never saw '_exec'). Non-null whenever the executor
+    // was built over a collection, shared-owned or not; if the collection has since been
+    // destroyed, '_exec' was killed by its invalidateAll() first and killed executors never
+    // dereference the manager argument.
+    CursorManager* _cursorManager = nullptr;
+
+    // EloqDoc: shared ownership of the Collection '_exec''s stages reference, keeping
+    // '_cursorManager' alive. Pipelines embed cursor sources over the top-level *and* foreign
+    // ($lookup/$graphLookup makePipeline) collections, and both can outlive the operation that
+    // resolved them inside a globally-managed aggregation cursor; per-operation RecoveryUnit
+    // pins do not cover them. Null for uniquely-owned Collection instances, whose lifetime is
+    // their owner's problem exactly as upstream assumed. Declared before '_exec' so the executor
+    // is destroyed first.
+    std::shared_ptr<Collection> _collectionPin;
+
     // The underlying query plan which feeds this pipeline. Must be destroyed while holding the
     // collection lock.
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> _exec;
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
index 58c8606b74..19c2798381 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
@@ -38,6 +38,7 @@
 #include "mongo/db/storage/kv/kv_catalog.h"
 #include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
 #include "mongo/db/storage/kv/kv_engine.h"
+#include "mongo/util/scopeguard.h"
 
 namespace mongo {
 
@@ -203,6 +204,21 @@ Status KVCollectionCatalogEntry::prepareForIndexBuild(OperationContext* opCtx,
             RecoveryUnit* newRU = opCtx->getServiceContext()->getStorageEngine()->newRecoveryUnit();
             WriteUnitOfWork::RecoveryUnitState oldState =
                 opCtx->setRecoveryUnit(newRU, WriteUnitOfWork::kNotInUnitOfWork);
+            // Everything below has to unwind cleanly. Marking the feature in use writes the
+            // catalog, and a catalog write conflict leaves as a WriteConflictException, so the
+            // loop can be exited by an exception at any point.
+            //
+            // Two things must happen on that path. The borrowed RecoveryUnit's unit of work has
+            // to be closed, or destroying it trips ~EloqRecoveryUnit's invariant(!_inUnitOfWork).
+            // And oldRU has to be reinstated, or it leaks and the caller's still-live toplevel
+            // WriteUnitOfWork finds _ruState == kNotInUnitOfWork and aborts the process.
+            bool newRuInUnitOfWork = false;
+            const auto restoreRecoveryUnit = MakeGuard([&] {
+                if (newRuInUnitOfWork) {
+                    newRU->abortUnitOfWork();
+                }
+                opCtx->setRecoveryUnit(oldRU, oldState);
+            });
 
             int retryCount = 0;
             const int maxRetry = 1000;
@@ -210,24 +226,33 @@ Status KVCollectionCatalogEntry::prepareForIndexBuild(OperationContext* opCtx,
 
             while (retryCount++ < maxRetry) {
                 newRU->beginUnitOfWork(opCtx);
+                newRuInUnitOfWork = true;
                 if (!_catalog->getFeatureTracker()->isRepairableFeatureInUse(opCtx, feature)) {
                     tmp_st =
                         _catalog->getFeatureTracker()->markRepairableFeatureAsInUse(opCtx, feature);
                     if (tmp_st.isOK()) {
+                        // Cleared before the call, not after: commitUnitOfWork() clears
+                        // _inUnitOfWork before doing the work that can throw, so on a
+                        // throwing commit the unit of work is already closed and the
+                        // guard must not try to abort it again.
+                        newRuInUnitOfWork = false;
                         newRU->commitUnitOfWork();
                         break;
                     } else {
+                        // Cleared before the call; see the commit path above.
+                        newRuInUnitOfWork = false;
                         newRU->abortUnitOfWork();
                         opCtx->sleepForRandomMilliseconds();
                     }
                 } else {
                     tmp_st = Status::OK();
+                    // Cleared before the call; see the commit path above.
+                    newRuInUnitOfWork = false;
                     newRU->commitUnitOfWork();
                     break;
                 }
             }
-            // Must restore the old recovery unit state before leaving the scope.
-            opCtx->setRecoveryUnit(oldRU, oldState);
+            // Restored by restoreRecoveryUnit above.
             if (!tmp_st.isOK()) {
                 return tmp_st;
             }
@@ -245,30 +270,54 @@ Status KVCollectionCatalogEntry::prepareForIndexBuild(OperationContext* opCtx,
             RecoveryUnit* newRU = opCtx->getServiceContext()->getStorageEngine()->newRecoveryUnit();
             WriteUnitOfWork::RecoveryUnitState oldState =
                 opCtx->setRecoveryUnit(newRU, WriteUnitOfWork::kNotInUnitOfWork);
+            // Everything below has to unwind cleanly. Marking the feature in use writes the
+            // catalog, and a catalog write conflict leaves as a WriteConflictException, so the
+            // loop can be exited by an exception at any point.
+            //
+            // Two things must happen on that path. The borrowed RecoveryUnit's unit of work has
+            // to be closed, or destroying it trips ~EloqRecoveryUnit's invariant(!_inUnitOfWork).
+            // And oldRU has to be reinstated, or it leaks and the caller's still-live toplevel
+            // WriteUnitOfWork finds _ruState == kNotInUnitOfWork and aborts the process.
+            bool newRuInUnitOfWork = false;
+            const auto restoreRecoveryUnit = MakeGuard([&] {
+                if (newRuInUnitOfWork) {
+                    newRU->abortUnitOfWork();
+                }
+                opCtx->setRecoveryUnit(oldRU, oldState);
+            });
 
             int retryCount = 0;
             const int maxRetry = 1000;
             Status tmp_st = Status::OK();
             while (retryCount++ < maxRetry) {
                 newRU->beginUnitOfWork(opCtx);
+                newRuInUnitOfWork = true;
                 if (!_catalog->getFeatureTracker()->isNonRepairableFeatureInUse(opCtx, feature)) {
                     tmp_st = _catalog->getFeatureTracker()->markNonRepairableFeatureAsInUse(
                         opCtx, feature);
                     if (tmp_st.isOK()) {
+                        // Cleared before the call, not after: commitUnitOfWork() clears
+                        // _inUnitOfWork before doing the work that can throw, so on a
+                        // throwing commit the unit of work is already closed and the
+                        // guard must not try to abort it again.
+                        newRuInUnitOfWork = false;
                         newRU->commitUnitOfWork();
                         break;
                     } else {
+                        // Cleared before the call; see the commit path above.
+                        newRuInUnitOfWork = false;
                         newRU->abortUnitOfWork();
                         opCtx->sleepForRandomMilliseconds();
                     }
                 } else {
                     tmp_st = Status::OK();
+                    // Cleared before the call; see the commit path above.
+                    newRuInUnitOfWork = false;
                     newRU->commitUnitOfWork();
                     break;
                 }
             }
-            // Must restore the old recovery unit state before leaving the scope.
-            opCtx->setRecoveryUnit(oldRU, oldState);
+            // Restored by restoreRecoveryUnit above.
             if (!tmp_st.isOK()) {
                 return tmp_st;
             }
diff --git a/src/mongo/db/storage/recovery_unit.h b/src/mongo/db/storage/recovery_unit.h
index 0e49fcba61..07b1872558 100644
--- a/src/mongo/db/storage/recovery_unit.h
+++ b/src/mongo/db/storage/recovery_unit.h
@@ -33,6 +33,7 @@
 #include <memory>
 #include <stdlib.h>
 #include <string>
+#include <vector>
 
 #include "mongo/base/disallow_copying.h"
 #include "mongo/base/status.h"
@@ -66,6 +67,33 @@ class RecoveryUnit {
 
     virtual void setOperationContext(OperationContext* opCtx) {}
 
+    /**
+     * Keeps 'resource' alive until this recovery unit is reset for reuse (or destroyed).
+     *
+     * EloqDoc: DatabaseImpl::getCollection() rebuilds the cached Collection in place whenever the
+     * catalog version moves, which happens constantly under concurrent DDL. Operations hold raw
+     * Collection* across coroutine yields and into RecoveryUnit onCommit callbacks, and
+     * EloqLockerNoop provides none of the lock-manager exclusion upstream relies on to make the
+     * eviction safe. Pinning the shared_ptr here defers destruction of the evicted object until
+     * the pooled OperationContext is recycled for its next operation
+     * (StorageClientObserver::onCreateOperationContext -> resetRecoveryUnit -> reset()), which is
+     * strictly after every use the pinning operation can make of the raw pointer. Multi-statement
+     * transactions stash and restore the RecoveryUnit between statements, so pins follow the
+     * transaction, not the wire operation.
+     */
+    void pinResource(std::shared_ptr<void> resource) {
+        // Deduplicate against the full live pin set so growth is bounded by distinct resources,
+        // not acquisition count (an operation alternating between two collections would otherwise
+        // grow the vector per call). Linear scan: the set is a handful of collections plus their
+        // superseded versions per operation/transaction.
+        for (const auto& pinned : _pinnedResources) {
+            if (pinned == resource) {
+                return;
+            }
+        }
+        _pinnedResources.push_back(std::move(resource));
+    }
+
     /**
      * These should be called through WriteUnitOfWork rather than directly.
      *
@@ -389,6 +417,17 @@ class RecoveryUnit {
 
 protected:
     RecoveryUnit() {}
+
+    /**
+     * Drops all resources pinned with pinResource(). Engines whose reset() supports pooled reuse
+     * (EloqRecoveryUnit) must call this from reset(); destruction handles the non-pooled case.
+     */
+    void clearPinnedResources() {
+        _pinnedResources.clear();
+    }
+
+private:
+    std::vector<std::shared_ptr<void>> _pinnedResources;
 };
 
 }  // namespace mongo