SERVER-73289 Support yield/restore on CollectionSnapshots API

9 files changed, 797 insertions, 299 deletions

diff --git a/src/mongo/db/s/collection_sharding_runtime.cpp b/src/mongo/db/s/collection_sharding_runtime.cpp
index be3c4ff4b4a..f007c17c777 100644
--- a/src/mongo/db/s/collection_sharding_runtime.cpp
+++ b/src/mongo/db/s/collection_sharding_runtime.cpp
@@ -147,6 +147,14 @@ ScopedCollectionFilter CollectionShardingRuntime::getOwnershipFilter(
     return {std::move(metadata)};
 }
 
+ScopedCollectionFilter CollectionShardingRuntime::getOwnershipFilter(
+    OperationContext* opCtx,
+    OrphanCleanupPolicy orphanCleanupPolicy,
+    const ShardVersion& receivedShardVersion) const {
+    return _getMetadataWithVersionCheckAt(
+        opCtx, repl::ReadConcernArgs::get(opCtx).getArgsAtClusterTime(), receivedShardVersion);
+}
+
 ScopedCollectionDescription CollectionShardingRuntime::getCollectionDescription(
     OperationContext* opCtx) const {
     const bool operationIsVersioned = OperationShardingState::isComingFromRouter(opCtx);
diff --git a/src/mongo/db/s/collection_sharding_runtime.h b/src/mongo/db/s/collection_sharding_runtime.h
index ec707f9e7a1..6575399c7f8 100644
--- a/src/mongo/db/s/collection_sharding_runtime.h
+++ b/src/mongo/db/s/collection_sharding_runtime.h
@@ -128,6 +128,10 @@ public:
     ScopedCollectionFilter getOwnershipFilter(OperationContext* opCtx,
                                               OrphanCleanupPolicy orphanCleanupPolicy,
                                               bool supportNonVersionedOperations) const override;
+    ScopedCollectionFilter getOwnershipFilter(
+        OperationContext* opCtx,
+        OrphanCleanupPolicy orphanCleanupPolicy,
+        const ShardVersion& receivedShardVersion) const override;
 
     void checkShardVersionOrThrow(OperationContext* opCtx) const override;
 
diff --git a/src/mongo/db/s/collection_sharding_state.h b/src/mongo/db/s/collection_sharding_state.h
index d01c3802e90..cf69a4dd369 100644
--- a/src/mongo/db/s/collection_sharding_state.h
+++ b/src/mongo/db/s/collection_sharding_state.h
@@ -166,6 +166,11 @@ public:
         OrphanCleanupPolicy orphanCleanupPolicy,
         bool supportNonVersionedOperations = false) const = 0;
 
+    virtual ScopedCollectionFilter getOwnershipFilter(
+        OperationContext* opCtx,
+        OrphanCleanupPolicy orphanCleanupPolicy,
+        const ShardVersion& receivedShardVersion) const = 0;
+
     /**
      * Checks whether the shard version in the operation context is compatible with the shard
      * version of the collection and if not, throws StaleConfigException populated with the received
diff --git a/src/mongo/db/s/collection_sharding_state_factory_standalone.cpp b/src/mongo/db/s/collection_sharding_state_factory_standalone.cpp
index 2695a9d0898..794dae98eb0 100644
--- a/src/mongo/db/s/collection_sharding_state_factory_standalone.cpp
+++ b/src/mongo/db/s/collection_sharding_state_factory_standalone.cpp
@@ -69,6 +69,13 @@ public:
         return {kUnshardedCollection};
     }
 
+    ScopedCollectionFilter getOwnershipFilter(
+        OperationContext*,
+        OrphanCleanupPolicy orphanCleanupPolicy,
+        const ShardVersion& receivedShardVersion) const override {
+        return {kUnshardedCollection};
+    }
+
     void checkShardVersionOrThrow(OperationContext*) const override {}
 
     void checkShardVersionOrThrow(OperationContext*, const ShardVersion&) const override {}
diff --git a/src/mongo/db/shard_role.cpp b/src/mongo/db/shard_role.cpp
index 39e289c9096..131e7cab441 100644
--- a/src/mongo/db/shard_role.cpp
+++ b/src/mongo/db/shard_role.cpp
@@ -42,7 +42,6 @@
 #include "mongo/db/s/database_sharding_state.h"
 #include "mongo/db/s/operation_sharding_state.h"
 #include "mongo/db/s/scoped_collection_metadata.h"
-#include "mongo/db/transaction_resources.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/decorable.h"
 
@@ -52,20 +51,21 @@ namespace mongo {
 namespace {
 
 auto getTransactionResources = OperationContext::declareDecoration<
-    boost::optional<shard_role_details::TransactionResources>>();
+    std::unique_ptr<shard_role_details::TransactionResources>>();
 
 shard_role_details::TransactionResources& getOrMakeTransactionResources(OperationContext* opCtx) {
     auto& readConcern = repl::ReadConcernArgs::get(opCtx);
     auto& optTransactionResources = getTransactionResources(opCtx);
     if (!optTransactionResources) {
-        optTransactionResources.emplace(readConcern);
+        optTransactionResources =
+            std::make_unique<shard_role_details::TransactionResources>(readConcern);
     }
 
     return *optTransactionResources;
 }
 
-NamespaceOrViewAcquisitionRequest::PlacementConcern getPlacementConcernFromOSS(
-    OperationContext* opCtx, const NamespaceString& nss) {
+AcquisitionPrerequisites::PlacementConcern getPlacementConcernFromOSS(OperationContext* opCtx,
+                                                                      const NamespaceString& nss) {
     const auto optDbVersion = OperationShardingState::get(opCtx).getDbVersion(nss.db());
     const auto optShardVersion = OperationShardingState::get(opCtx).getShardVersion(nss);
     return {optDbVersion, optShardVersion};
@@ -73,18 +73,11 @@ NamespaceOrViewAcquisitionRequest::PlacementConcern getPlacementConcernFromOSS(
 
 struct ResolvedNamespaceOrViewAcquisitionRequest {
     // Populated in the first phase of collection(s) acquisition
-    NamespaceString nss;
-
-    boost::optional<UUID> uuid;
-
-    NamespaceOrViewAcquisitionRequest::PlacementConcern tsPlacement;
-    NamespaceOrViewAcquisitionRequest::ViewMode viewMode;
+    AcquisitionPrerequisites prerequisites;
 
     // Populated optionally in the second phase of collection(s) acquisition
     boost::optional<Lock::DBLock> dbLock;
     boost::optional<Lock::CollectionLock> collLock;
-
-    boost::optional<ScopedCollectionFilter> filter;
 };
 
 using ResolvedNamespaceOrViewAcquisitionRequestsMap =
@@ -104,14 +97,11 @@ ResolvedNamespaceOrViewAcquisitionRequestsMap resolveNamespaceOrViewAcquisitionR
                 checkCollectionUUIDMismatch(opCtx, *ar.nss, coll, *ar.uuid);
             }
 
+            AcquisitionPrerequisites prerequisites(
+                *ar.nss, ar.uuid, ar.placementConcern, ar.operationType, ar.viewMode);
+
             ResolvedNamespaceOrViewAcquisitionRequest resolvedAcquisitionRequest{
-                *ar.nss,
-                ar.uuid,
-                ar.placementConcern,
-                ar.viewMode,
-                boost::none,
-                boost::none,
-                boost::none};
+                prerequisites, boost::none, boost::none};
             sortedAcquisitionRequests.emplace(ResourceId(RESOURCE_COLLECTION, *ar.nss),
                                               std::move(resolvedAcquisitionRequest));
         } else if (ar.dbname) {
@@ -129,14 +119,11 @@ ResolvedNamespaceOrViewAcquisitionRequestsMap resolveNamespaceOrViewAcquisitionR
                 checkCollectionUUIDMismatch(opCtx, *ar.nss, coll, *ar.uuid);
             }
 
+            AcquisitionPrerequisites prerequisites(
+                coll->ns(), coll->uuid(), ar.placementConcern, ar.operationType, ar.viewMode);
+
             ResolvedNamespaceOrViewAcquisitionRequest resolvedAcquisitionRequest{
-                coll->ns(),
-                coll->uuid(),
-                ar.placementConcern,
-                ar.viewMode,
-                boost::none,
-                boost::none,
-                boost::none};
+                prerequisites, boost::none, boost::none};
 
             sortedAcquisitionRequests.emplace(ResourceId(RESOURCE_COLLECTION, coll->ns()),
                                               std::move(resolvedAcquisitionRequest));
@@ -180,68 +167,84 @@ void verifyDbAndCollection(OperationContext* opCtx,
     }
 }
 
-void checkPlacementVersion(
-    OperationContext* opCtx,
-    const ResolvedNamespaceOrViewAcquisitionRequest& resolvedAcquisitionRequest) {
-    const auto& nss = resolvedAcquisitionRequest.nss;
+void checkPlacementVersion(OperationContext* opCtx, const AcquisitionPrerequisites& prerequisites) {
+    const auto& nss = prerequisites.nss;
 
-    const auto& receivedDbVersion = resolvedAcquisitionRequest.tsPlacement.dbVersion;
+    const auto& receivedDbVersion = prerequisites.placementConcern.dbVersion;
     if (receivedDbVersion) {
         DatabaseShardingState::assertMatchingDbVersion(opCtx, nss.db(), *receivedDbVersion);
     }
 
-    const auto& receivedShardVersion = resolvedAcquisitionRequest.tsPlacement.shardVersion;
+    const auto& receivedShardVersion = prerequisites.placementConcern.shardVersion;
     if (receivedShardVersion) {
         const auto scopedCSS = CollectionShardingState::acquire(opCtx, nss);
         scopedCSS->checkShardVersionOrThrow(opCtx, *receivedShardVersion);
     }
 }
 
-ScopedCollectionOrViewAcquisition acquireResolvedCollectionOrView(
-    OperationContext* opCtx,
-    ResolvedNamespaceOrViewAcquisitionRequest& resolvedAcquisitionRequest) {
-    const auto& nss = resolvedAcquisitionRequest.nss;
+struct SnapshotedServices {
+    CollectionPtr collectionPtr;
+    ScopedCollectionDescription collectionDescription;
+    boost::optional<ScopedCollectionFilter> ownershipFilter;
+};
 
-    // Check placement version before acquiring the catalog snapshot
-    checkPlacementVersion(opCtx, resolvedAcquisitionRequest);
+CollectionPtr acquireCollectionPtr(OperationContext* opCtx,
+                                   const AcquisitionPrerequisites& prerequisites) {
+    const auto& nss = prerequisites.nss;
 
     const auto catalog = CollectionCatalog::get(opCtx);
     auto coll = catalog->lookupCollectionByNamespace(opCtx, nss);
 
-    // Checks after having established the storage catalog snapshot
-    if (resolvedAcquisitionRequest.uuid) {
-        checkCollectionUUIDMismatch(opCtx, nss, coll, resolvedAcquisitionRequest.uuid);
-    }
-
     if (coll) {
         verifyDbAndCollection(opCtx, nss, coll);
     } else if (catalog->lookupView(opCtx, nss)) {
         uassert(ErrorCodes::CommandNotSupportedOnView,
                 str::stream() << "Namespace " << nss << " is a view, not a collection",
-                resolvedAcquisitionRequest.viewMode ==
-                    NamespaceOrViewAcquisitionRequest::ViewMode::kCanBeView);
+                prerequisites.viewMode == AcquisitionPrerequisites::kCanBeView);
     } else {
         uasserted(ErrorCodes::NamespaceNotFound,
                   str::stream() << "Namespace " << nss << "does not exist.");
     }
 
-    const bool isPlacementConcernVersioned = resolvedAcquisitionRequest.tsPlacement.dbVersion ||
-        resolvedAcquisitionRequest.tsPlacement.shardVersion;
+    // Checks after having established the storage catalog snapshot
+    checkCollectionUUIDMismatch(opCtx, nss, coll, prerequisites.uuid);
+
+    return coll;
+}
+
+SnapshotedServices acquireServicesSnapshot(OperationContext* opCtx,
+                                           const AcquisitionPrerequisites& prerequisites) {
+    const auto& nss = prerequisites.nss;
+
+    // Check placement version before acquiring the catalog snapshot
+    checkPlacementVersion(opCtx, prerequisites);
+
+    auto coll = acquireCollectionPtr(opCtx, prerequisites);
+
+    const bool isPlacementConcernVersioned =
+        prerequisites.placementConcern.dbVersion || prerequisites.placementConcern.shardVersion;
 
     const auto scopedCSS = CollectionShardingState::acquire(opCtx, nss);
     auto collectionDescription =
         scopedCSS->getCollectionDescription(opCtx, isPlacementConcernVersioned);
 
+    invariant(!collectionDescription.isSharded() || prerequisites.placementConcern.shardVersion);
+    auto optOwnershipFilter = collectionDescription.isSharded()
+        ? boost::optional<ScopedCollectionFilter>(scopedCSS->getOwnershipFilter(
+              opCtx,
+              prerequisites.operationType == AcquisitionPrerequisites::kRead
+                  ? CollectionShardingState::OrphanCleanupPolicy::kDisallowOrphanCleanup
+                  : CollectionShardingState::OrphanCleanupPolicy::kAllowOrphanCleanup,
+              *prerequisites.placementConcern.shardVersion))
+        : boost::none;
+
     // Recheck the placement version after having acquired the catalog snapshot. If the placement
     // version still matches, then the catalog we snapshoted is consistent with the placement
     // concern too.
-    checkPlacementVersion(opCtx, resolvedAcquisitionRequest);
+    checkPlacementVersion(opCtx, prerequisites);
 
-    return ScopedCollectionOrViewAcquisition::make(opCtx,
-                                                   std::move(coll),
-                                                   std::move(collectionDescription),
-                                                   std::move(resolvedAcquisitionRequest.dbLock),
-                                                   std::move(resolvedAcquisitionRequest.collLock));
+    return SnapshotedServices{
+        std::move(coll), std::move(collectionDescription), std::move(optOwnershipFilter)};
 }
 
 std::vector<ScopedCollectionOrViewAcquisition> acquireResolvedCollectionsOrViewsWithoutTakingLocks(
@@ -249,8 +252,36 @@ std::vector<ScopedCollectionOrViewAcquisition> acquireResolvedCollectionsOrViews
     ResolvedNamespaceOrViewAcquisitionRequestsMap sortedAcquisitionRequests) {
     std::vector<ScopedCollectionOrViewAcquisition> acquisitions;
     for (auto& acquisitionRequest : sortedAcquisitionRequests) {
-        auto acquisition = acquireResolvedCollectionOrView(opCtx, acquisitionRequest.second);
-        acquisitions.emplace_back(std::move(acquisition));
+        tassert(7328900,
+                "Cannot acquire for write without locks",
+                acquisitionRequest.second.prerequisites.operationType ==
+                        AcquisitionPrerequisites::kRead ||
+                    acquisitionRequest.second.collLock);
+
+        auto& prerequisites = acquisitionRequest.second.prerequisites;
+        auto snapshotedServices = acquireServicesSnapshot(opCtx, prerequisites);
+
+        invariant(snapshotedServices.collectionPtr);
+        invariant(!prerequisites.uuid ||
+                  prerequisites.uuid == snapshotedServices.collectionPtr->uuid());
+        if (!prerequisites.uuid) {
+            // If the uuid wasn't originally set on the AcquisitionRequest, set it now on the
+            // prerequisites so that on restore from yield we can check we are restoring the same
+            // instance of the ns.
+            prerequisites.uuid = snapshotedServices.collectionPtr->uuid();
+        }
+
+        const shard_role_details::AcquiredCollection& acquiredCollection =
+            getOrMakeTransactionResources(opCtx).addAcquiredCollection(
+                {prerequisites,
+                 std::move(acquisitionRequest.second.dbLock),
+                 std::move(acquisitionRequest.second.collLock),
+                 std::move(snapshotedServices.collectionDescription),
+                 std::move(snapshotedServices.ownershipFilter),
+                 std::move(snapshotedServices.collectionPtr)});
+
+        ScopedCollectionOrViewAcquisition scopedAcquisition(opCtx, acquiredCollection);
+        acquisitions.emplace_back(std::move(scopedAcquisition));
     }
 
     return acquisitions;
@@ -258,7 +289,7 @@ std::vector<ScopedCollectionOrViewAcquisition> acquireResolvedCollectionsOrViews
 
 }  // namespace
 
-const NamespaceOrViewAcquisitionRequest::PlacementConcern
+const AcquisitionPrerequisites::PlacementConcern
     NamespaceOrViewAcquisitionRequest::kPretendUnshardedDueToDirectConnection{boost::none,
                                                                               boost::none};
 
@@ -266,43 +297,42 @@ NamespaceOrViewAcquisitionRequest::NamespaceOrViewAcquisitionRequest(
     OperationContext* opCtx,
     NamespaceString nss,
     repl::ReadConcernArgs readConcern,
-    ViewMode viewMode)
+    AcquisitionPrerequisites::OperationType operationType,
+    AcquisitionPrerequisites::ViewMode viewMode)
     : nss(nss),
       placementConcern(getPlacementConcernFromOSS(opCtx, nss)),
       readConcern(readConcern),
+      operationType(operationType),
       viewMode(viewMode) {}
 
 ScopedCollectionOrViewAcquisition::~ScopedCollectionOrViewAcquisition() {
     if (_opCtx) {
-        auto& transactionResources = getOrMakeTransactionResources(_opCtx);
-        transactionResources.acquiredCollections.remove_if(
-            [& acquiredCollection = _acquiredCollection](
-                const shard_role_details::TransactionResources::AcquiredCollection&
-                    txnResourceAcquiredColl) {
-                return &txnResourceAcquiredColl == &acquiredCollection;
-            });
+        const auto& transactionResources = getTransactionResources(_opCtx);
+        if (transactionResources) {
+            transactionResources->acquiredCollections.remove_if(
+                [this](const shard_role_details::AcquiredCollection& txnResourceAcquiredColl) {
+                    return &txnResourceAcquiredColl == &(this->_acquiredCollection);
+                });
+        }
     }
 }
 
-ScopedCollectionOrViewAcquisition ScopedCollectionOrViewAcquisition::make(
-    OperationContext* opCtx,
-    CollectionPtr&& collectionPtr,
-    ScopedCollectionDescription&& collectionDescription,
-    boost::optional<Lock::DBLock>&& dbLock,
-    boost::optional<Lock::CollectionLock>&& collectionLock) {
-    invariant(collectionPtr);
-
-    const auto nss = collectionPtr->ns();
-    const auto uuid = collectionPtr->uuid();
-    const shard_role_details::TransactionResources::AcquiredCollection& acquiredCollection =
-        getOrMakeTransactionResources(opCtx).addAcquiredCollection({nss,
-                                                                    uuid,
-                                                                    std::move(collectionPtr),
-                                                                    collectionDescription,
-                                                                    std::move(dbLock),
-                                                                    std::move(collectionLock)});
-
-    return ScopedCollectionOrViewAcquisition(opCtx, std::move(acquiredCollection));
+const NamespaceString& ScopedCollectionOrViewAcquisition::nss() const {
+    return _acquiredCollection.prerequisites.nss;
+}
+
+const ScopedCollectionDescription& ScopedCollectionOrViewAcquisition::getShardingDescription()
+    const {
+    return _acquiredCollection.collectionDescription;
+}
+
+const boost::optional<ScopedCollectionFilter>&
+ScopedCollectionOrViewAcquisition::getCollectionFilter() const {
+    return _acquiredCollection.ownershipFilter;
+}
+
+const CollectionPtr& ScopedCollectionOrViewAcquisition::getCollectionPtr() const {
+    return _acquiredCollection.collectionPtr;
 }
 
 std::vector<ScopedCollectionOrViewAcquisition> acquireCollectionsOrViews(
@@ -326,9 +356,9 @@ std::vector<ScopedCollectionOrViewAcquisition> acquireCollectionsOrViews(
             // TODO: SERVER-73004 When acquiring multiple collections, avoid recursively locking the
             // dbLock because that causes recursive locking of the globalLock which prevents
             // yielding.
-            ar.second.dbLock.emplace(
-                opCtx, ar.second.nss.db(), isSharedLockMode(mode) ? MODE_IS : MODE_IX);
-            ar.second.collLock.emplace(opCtx, ar.second.nss, mode);
+            const auto& nss = ar.second.prerequisites.nss;
+            ar.second.dbLock.emplace(opCtx, nss.db(), isSharedLockMode(mode) ? MODE_IS : MODE_IX);
+            ar.second.collLock.emplace(opCtx, nss, mode);
         }
 
         try {
@@ -356,9 +386,79 @@ std::vector<ScopedCollectionOrViewAcquisition> acquireCollectionsOrViewsWithoutT
     }
 }
 
-YieldedTransactionResources yieldTransactionResources(OperationContext* opCtx);
+YieldedTransactionResources::~YieldedTransactionResources() {
+    invariant(!_yieldedResources);
+}
+
+YieldedTransactionResources::YieldedTransactionResources(
+    std::unique_ptr<shard_role_details::TransactionResources>&& yieldedResources)
+    : _yieldedResources(std::move(yieldedResources)) {}
+
+YieldedTransactionResources yieldTransactionResourcesFromOperationContext(OperationContext* opCtx) {
+    auto transactionResources = std::move(getTransactionResources(opCtx));
+    if (!transactionResources) {
+        return YieldedTransactionResources();
+    }
+
+    invariant(!transactionResources->yielded);
+
+    invariant(!transactionResources->lockSnapshot.is_initialized());
+    transactionResources->lockSnapshot.emplace();
+    opCtx->lockState()->saveLockStateAndUnlock(&(*transactionResources->lockSnapshot));
+
+    transactionResources->yielded = true;
+
+    return YieldedTransactionResources(std::move(transactionResources));
+}
+
+void restoreTransactionResourcesToOperationContext(OperationContext* opCtx,
+                                                   YieldedTransactionResources&& yieldedResources) {
+    if (!yieldedResources._yieldedResources) {
+        // Nothing to restore.
+        return;
+    }
+
+    // On failure to restore, release the yielded resources.
+    ScopeGuard scopeGuard([&] {
+        yieldedResources._yieldedResources->releaseAllResourcesOnCommitOrAbort();
+        yieldedResources._yieldedResources.reset();
+    });
+
+    // Reacquire locks.
+    if (yieldedResources._yieldedResources->lockSnapshot) {
+        opCtx->lockState()->restoreLockState(opCtx,
+                                             *yieldedResources._yieldedResources->lockSnapshot);
+        yieldedResources._yieldedResources->lockSnapshot.reset();
+    }
+
+    // Reacquire service snapshots. Will throw if placement concern can no longer be met.
+    for (auto& acquiredCollection : yieldedResources._yieldedResources->acquiredCollections) {
+        const auto& prerequisites = acquiredCollection.prerequisites;
+
+        if (prerequisites.operationType == AcquisitionPrerequisites::OperationType::kRead) {
+            // Just reacquire the CollectionPtr. Reads don't care about placement changes because
+            // they have already established a ScopedCollectionFilter that acts as RangePreserver.
+            auto collectionPtr = acquireCollectionPtr(opCtx, prerequisites);
 
-void restoreTransactionResources(OperationContext* opCtx,
-                                 YieldedTransactionResources&& yieldedResources);
+            // Update the services snapshot on TransactionResources
+            acquiredCollection.collectionPtr = std::move(collectionPtr);
+
+        } else {
+            auto reacquiredServicesSnapshot = acquireServicesSnapshot(opCtx, prerequisites);
+
+            // Update the services snapshot on TransactionResources
+            acquiredCollection.collectionPtr = std::move(reacquiredServicesSnapshot.collectionPtr);
+            acquiredCollection.collectionDescription =
+                std::move(reacquiredServicesSnapshot.collectionDescription);
+            acquiredCollection.ownershipFilter =
+                std::move(reacquiredServicesSnapshot.ownershipFilter);
+        }
+    }
+
+    // Restore TransactionsResource on opCtx.
+    yieldedResources._yieldedResources->yielded = false;
+    getTransactionResources(opCtx) = std::move(yieldedResources)._yieldedResources;
+    scopeGuard.dismiss();
+}
 
 }  // namespace mongo
diff --git a/src/mongo/db/shard_role.h b/src/mongo/db/shard_role.h
index 92adecb674a..228748e4534 100644
--- a/src/mongo/db/shard_role.h
+++ b/src/mongo/db/shard_role.h
@@ -40,133 +40,128 @@
 namespace mongo {
 
 /**
- * See the comments on the TransactionResources class for the semantics of this class.
- */
-class ScopedCollectionOrViewAcquisition {
-public:
-    ScopedCollectionOrViewAcquisition() = delete;
-    ScopedCollectionOrViewAcquisition(const mongo::ScopedCollectionOrViewAcquisition&) = delete;
-
-    ScopedCollectionOrViewAcquisition(mongo::ScopedCollectionOrViewAcquisition&& other)
-        : _opCtx(other._opCtx), _acquiredCollection(other._acquiredCollection) {
-        other._opCtx = nullptr;
-    }
-
-    ~ScopedCollectionOrViewAcquisition();
-
-    static ScopedCollectionOrViewAcquisition make(
-        OperationContext* opCtx,
-        CollectionPtr&& collectionPtr,
-        ScopedCollectionDescription&& collectionDescription,
-        boost::optional<Lock::DBLock>&& dbLock,
-        boost::optional<Lock::CollectionLock>&& collectionLock);
-
-    const NamespaceString& nss() const {
-        return _acquiredCollection.nss;
-    }
-
-    bool isView() const {
-        // TODO: SERVER-73005 Support views
-        return false;
-    }
-
-    // Access to services associated with the specified collection top to bottom on the hierarchical
-    // stack
-
-    // Sharding services
-    ScopedCollectionDescription getShardingDescription() const {
-        return _acquiredCollection.collectionDescription;
-    }
-
-    // StorEx services
-    const CollectionPtr& getCollectionPtr() const {
-        return _acquiredCollection.collectionPtr;
-    }
-
-private:
-    ScopedCollectionOrViewAcquisition(
-        OperationContext* opCtx,
-        const shard_role_details::TransactionResources::AcquiredCollection& acquiredCollection)
-        : _opCtx(opCtx), _acquiredCollection(acquiredCollection) {}
-
-    OperationContext* _opCtx;
-
-    const shard_role_details::TransactionResources::AcquiredCollection& _acquiredCollection;
-};
-
-/**
  * Contains all the required properties for the acquisition of a collection. These properties are
  * taken into account in addition to the ReadConcern of the transaction, which is stored in the
  * OperationContext.
  */
 struct NamespaceOrViewAcquisitionRequest {
-    struct PlacementConcern {
-        boost::optional<DatabaseVersion> dbVersion;
-        boost::optional<ShardVersion> shardVersion;
-    };
-
-    static const PlacementConcern kPretendUnshardedDueToDirectConnection;
-
-    enum ViewMode { kMustBeCollection, kCanBeView };
+    static const AcquisitionPrerequisites::PlacementConcern kPretendUnshardedDueToDirectConnection;
 
     /**
      * Overload, which acquires a collection by NSS, ignoring the current UUID mapping.
      */
-    NamespaceOrViewAcquisitionRequest(NamespaceString nss,
-                                      PlacementConcern placementConcern,
-                                      repl::ReadConcernArgs readConcern,
-                                      ViewMode viewMode = kMustBeCollection)
+    NamespaceOrViewAcquisitionRequest(
+        NamespaceString nss,
+        AcquisitionPrerequisites::PlacementConcern placementConcern,
+        repl::ReadConcernArgs readConcern,
+        AcquisitionPrerequisites::OperationType operationType,
+        AcquisitionPrerequisites::ViewMode viewMode = AcquisitionPrerequisites::kMustBeCollection)
         : nss(nss),
           placementConcern(placementConcern),
           readConcern(readConcern),
+          operationType(operationType),
           viewMode(viewMode) {}
 
     /**
      * Overload, which acquires a collection by NSS/UUID combination, requiring that the UUID of the
      * namespace matches exactly.
      */
-    NamespaceOrViewAcquisitionRequest(NamespaceString nss,
-                                      UUID uuid,
-                                      PlacementConcern placementConcern,
-                                      repl::ReadConcernArgs readConcern,
-                                      ViewMode viewMode = kMustBeCollection)
+    NamespaceOrViewAcquisitionRequest(
+        NamespaceString nss,
+        UUID uuid,
+        AcquisitionPrerequisites::PlacementConcern placementConcern,
+        repl::ReadConcernArgs readConcern,
+        AcquisitionPrerequisites::OperationType operationType,
+        AcquisitionPrerequisites::ViewMode viewMode = AcquisitionPrerequisites::kMustBeCollection)
         : nss(nss),
           uuid(uuid),
           placementConcern(placementConcern),
           readConcern(readConcern),
+          operationType(operationType),
           viewMode(viewMode) {}
 
     /**
      * Overload, which acquires a collection by NSS or DB/UUID, without imposing an expected
      * relationship between NSS and UUID.
      */
-    NamespaceOrViewAcquisitionRequest(NamespaceStringOrUUID nssOrUUID,
-                                      PlacementConcern placementConcern,
-                                      repl::ReadConcernArgs readConcern,
-                                      ViewMode viewMode = kMustBeCollection)
+    NamespaceOrViewAcquisitionRequest(
+        NamespaceStringOrUUID nssOrUUID,
+        AcquisitionPrerequisites::PlacementConcern placementConcern,
+        repl::ReadConcernArgs readConcern,
+        AcquisitionPrerequisites::OperationType operationType,
+        AcquisitionPrerequisites::ViewMode viewMode = AcquisitionPrerequisites::kMustBeCollection)
         : dbname(nssOrUUID.dbName()),
           nss(nssOrUUID.nss()),
           uuid(nssOrUUID.uuid()),
           placementConcern(placementConcern),
           readConcern(readConcern),
+          operationType(operationType),
           viewMode(viewMode) {}
 
     /**
      * Overload, which acquires a collection by NSS, ignoring the current UUID mapping. Takes the
      * placement concern from the opCtx's OperationShardingState.
      */
-    NamespaceOrViewAcquisitionRequest(OperationContext* opCtx,
-                                      NamespaceString nss,
-                                      repl::ReadConcernArgs readConcern,
-                                      ViewMode viewMode = kMustBeCollection);
+    NamespaceOrViewAcquisitionRequest(
+        OperationContext* opCtx,
+        NamespaceString nss,
+        repl::ReadConcernArgs readConcern,
+        AcquisitionPrerequisites::OperationType operationType,
+        AcquisitionPrerequisites::ViewMode viewMode = AcquisitionPrerequisites::kMustBeCollection);
 
     boost::optional<DatabaseName> dbname;
     boost::optional<NamespaceString> nss;
     boost::optional<UUID> uuid;
 
-    PlacementConcern placementConcern;
+    AcquisitionPrerequisites::PlacementConcern placementConcern;
     repl::ReadConcernArgs readConcern;
-    ViewMode viewMode;
+    AcquisitionPrerequisites::OperationType operationType;
+    AcquisitionPrerequisites::ViewMode viewMode;
+};
+
+/**
+ * See the comments on the TransactionResources class for the semantics of this class.
+ */
+class ScopedCollectionOrViewAcquisition {
+public:
+    ScopedCollectionOrViewAcquisition() = delete;
+    ScopedCollectionOrViewAcquisition(const mongo::ScopedCollectionOrViewAcquisition&) = delete;
+
+    ScopedCollectionOrViewAcquisition(mongo::ScopedCollectionOrViewAcquisition&& other)
+        : _opCtx(other._opCtx), _acquiredCollection(other._acquiredCollection) {
+        other._opCtx = nullptr;
+    }
+
+    ~ScopedCollectionOrViewAcquisition();
+
+    ScopedCollectionOrViewAcquisition(
+        OperationContext* opCtx, const shard_role_details::AcquiredCollection& acquiredCollection)
+        : _opCtx(opCtx), _acquiredCollection(acquiredCollection) {}
+
+    const NamespaceString& nss() const;
+
+    bool isView() const {
+        // TODO: SERVER-73005 Support views
+        return false;
+    }
+
+    // Access to services associated with the specified collection top to bottom on the hierarchical
+    // stack
+
+    // Sharding services
+    const ScopedCollectionDescription& getShardingDescription() const;
+    const boost::optional<ScopedCollectionFilter>& getCollectionFilter() const;
+
+    // StorEx services
+    const CollectionPtr& getCollectionPtr() const;
+
+private:
+    OperationContext* _opCtx;
+
+    // Points to the acquired resources that live on the TransactionResources opCtx decoration. The
+    // lifetime of these resources is tied to the lifetime of this
+    // ScopedCollectionOrViewAcquisition.
+    const shard_role_details::AcquiredCollection& _acquiredCollection;
 };
 
 /**
@@ -198,10 +193,12 @@ class YieldedTransactionResources {
 public:
     ~YieldedTransactionResources();
 
-private:
-    YieldedTransactionResources(shard_role_details::TransactionResources&& yieldedResources);
+    YieldedTransactionResources() = default;
+
+    YieldedTransactionResources(
+        std::unique_ptr<shard_role_details::TransactionResources>&& yieldedResources);
 
-    shard_role_details::TransactionResources _yieldedResources;
+    std::unique_ptr<shard_role_details::TransactionResources> _yieldedResources;
 };
 
 YieldedTransactionResources yieldTransactionResourcesFromOperationContext(OperationContext* opCtx);
diff --git a/src/mongo/db/shard_role_test.cpp b/src/mongo/db/shard_role_test.cpp
index eade9a4bff1..6d625a095f4 100644
--- a/src/mongo/db/shard_role_test.cpp
+++ b/src/mongo/db/shard_role_test.cpp
@@ -30,6 +30,7 @@
 #include "mongo/db/catalog/collection_uuid_mismatch_info.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/database_holder.h"
+#include "mongo/db/dbdirectclient.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/s/collection_sharding_runtime.h"
 #include "mongo/db/s/database_sharding_state.h"
@@ -64,7 +65,10 @@ void installUnshardedCollectionMetadata(OperationContext* opCtx, const Namespace
 void installShardedCollectionMetadata(OperationContext* opCtx,
                                       const NamespaceString& nss,
                                       const DatabaseVersion& dbVersion,
-                                      const ShardVersion& shardVersion) {
+                                      std::vector<ChunkType> chunks,
+                                      ShardId thisShardId) {
+    ASSERT(!chunks.empty());
+
     const auto uuid = [&] {
         AutoGetCollection autoColl(opCtx, nss, MODE_IX);
         return autoColl.getCollection()->uuid();
@@ -72,24 +76,21 @@ void installShardedCollectionMetadata(OperationContext* opCtx,
 
     const std::string shardKey("skey");
     const ShardKeyPattern shardKeyPattern{BSON(shardKey << 1)};
-    const ShardId thisShardId("this");
-
-    auto rt = RoutingTableHistory::makeNew(
-        nss,
-        uuid,
-        shardKeyPattern.getKeyPattern(),
-        nullptr,
-        false,
-        shardVersion.placementVersion().epoch(),
-        shardVersion.placementVersion().getTimestamp(),
-        boost::none /* timeseriesFields */,
-        boost::none /* resharding Fields */,
-        boost::none /* chunkSizeBytes */,
-        true /* allowMigrations */,
-        {ChunkType{uuid,
-                   ChunkRange{BSON(shardKey << MINKEY), BSON(shardKey << MAXKEY)},
-                   shardVersion.placementVersion(),
-                   thisShardId}});
+    const auto epoch = chunks.front().getVersion().epoch();
+    const auto timestamp = chunks.front().getVersion().getTimestamp();
+
+    auto rt = RoutingTableHistory::makeNew(nss,
+                                           uuid,
+                                           shardKeyPattern.getKeyPattern(),
+                                           nullptr,
+                                           false,
+                                           epoch,
+                                           timestamp,
+                                           boost::none /* timeseriesFields */,
+                                           boost::none /* resharding Fields */,
+                                           boost::none /* chunkSizeBytes */,
+                                           true /* allowMigrations */,
+                                           chunks);
 
     const auto version = rt.getVersion();
     const auto rtHandle =
@@ -119,6 +120,8 @@ protected:
     void setUp() override;
     void tearDown() override;
 
+    const ShardId thisShardId{"this"};
+
     const DatabaseName dbNameTestDb{"test"};
     const DatabaseVersion dbVersionTestDb{UUID::gen(), Timestamp(1, 0)};
 
@@ -131,6 +134,13 @@ protected:
         ChunkVersion(CollectionGeneration{OID::gen(), Timestamp(5, 0)}, CollectionPlacement(10, 1)),
         boost::optional<CollectionIndexes>(boost::none)};
 
+    // Workaround to be able to write parametrized TEST_F
+    void testRestoreFailsIfCollectionNoLongerExists(
+        AcquisitionPrerequisites::OperationType operationType);
+    void testRestoreFailsIfCollectionRenamed(AcquisitionPrerequisites::OperationType operationType);
+    void testRestoreFailsIfCollectionDroppedAndRecreated(
+        AcquisitionPrerequisites::OperationType operationType);
+
 private:
     ServiceContext::UniqueOperationContext _opCtx;
 };
@@ -159,9 +169,17 @@ void ShardRoleTest::setUp() {
     installUnshardedCollectionMetadata(opCtx(), nssUnshardedCollection1);
 
     createTestCollection(opCtx(), nssShardedCollection1);
+    const auto uuidShardedCollection1 = getCollectionUUID(_opCtx.get(), nssShardedCollection1);
     installDatabaseMetadata(opCtx(), dbNameTestDb, dbVersionTestDb);
     installShardedCollectionMetadata(
-        opCtx(), nssShardedCollection1, dbVersionTestDb, shardVersionShardedCollection1);
+        opCtx(),
+        nssShardedCollection1,
+        dbVersionTestDb,
+        {ChunkType(uuidShardedCollection1,
+                   ChunkRange{BSON("skey" << MINKEY), BSON("skey" << MAXKEY)},
+                   shardVersionShardedCollection1.placementVersion(),
+                   thisShardId)},
+        thisShardId);
 }
 
 void ShardRoleTest::tearDown() {
@@ -174,7 +192,8 @@ TEST_F(ShardRoleTest, NamespaceOrViewAcquisitionRequestWithOpCtxTakesPlacementFr
     const auto nss = nssUnshardedCollection1;
 
     {
-        NamespaceOrViewAcquisitionRequest acquisitionRequest(opCtx(), nss, {});
+        NamespaceOrViewAcquisitionRequest acquisitionRequest(
+            opCtx(), nss, {}, AcquisitionPrerequisites::kWrite);
         ASSERT_EQ(boost::none, acquisitionRequest.placementConcern.dbVersion);
         ASSERT_EQ(boost::none, acquisitionRequest.placementConcern.shardVersion);
     }
@@ -184,7 +203,8 @@ TEST_F(ShardRoleTest, NamespaceOrViewAcquisitionRequestWithOpCtxTakesPlacementFr
             NamespaceString::createNamespaceString_forTest("test2.foo");
         ScopedSetShardRole setShardRole(
             opCtx(), anotherCollection, ShardVersion::UNSHARDED(), dbVersionTestDb);
-        NamespaceOrViewAcquisitionRequest acquisitionRequest(opCtx(), nss, {});
+        NamespaceOrViewAcquisitionRequest acquisitionRequest(
+            opCtx(), nss, {}, AcquisitionPrerequisites::kWrite);
         ASSERT_EQ(boost::none, acquisitionRequest.placementConcern.dbVersion);
         ASSERT_EQ(boost::none, acquisitionRequest.placementConcern.shardVersion);
     }
@@ -193,7 +213,8 @@ TEST_F(ShardRoleTest, NamespaceOrViewAcquisitionRequestWithOpCtxTakesPlacementFr
         const auto dbVersion = boost::none;
         const auto shardVersion = boost::none;
         ScopedSetShardRole setShardRole(opCtx(), nss, shardVersion, dbVersion);
-        NamespaceOrViewAcquisitionRequest acquisitionRequest(opCtx(), nss, {});
+        NamespaceOrViewAcquisitionRequest acquisitionRequest(
+            opCtx(), nss, {}, AcquisitionPrerequisites::kWrite);
         ASSERT_EQ(dbVersion, acquisitionRequest.placementConcern.dbVersion);
         ASSERT_EQ(shardVersion, acquisitionRequest.placementConcern.shardVersion);
     }
@@ -202,7 +223,8 @@ TEST_F(ShardRoleTest, NamespaceOrViewAcquisitionRequestWithOpCtxTakesPlacementFr
         const auto dbVersion = dbVersionTestDb;
         const auto shardVersion = ShardVersion::UNSHARDED();
         ScopedSetShardRole setShardRole(opCtx(), nss, shardVersion, dbVersion);
-        NamespaceOrViewAcquisitionRequest acquisitionRequest(opCtx(), nss, {});
+        NamespaceOrViewAcquisitionRequest acquisitionRequest(
+            opCtx(), nss, {}, AcquisitionPrerequisites::kWrite);
         ASSERT_EQ(dbVersion, acquisitionRequest.placementConcern.dbVersion);
         ASSERT_EQ(shardVersion, acquisitionRequest.placementConcern.shardVersion);
     }
@@ -211,7 +233,8 @@ TEST_F(ShardRoleTest, NamespaceOrViewAcquisitionRequestWithOpCtxTakesPlacementFr
         const auto dbVersion = boost::none;
         const auto shardVersion = shardVersionShardedCollection1;
         ScopedSetShardRole setShardRole(opCtx(), nss, shardVersion, dbVersion);
-        NamespaceOrViewAcquisitionRequest acquisitionRequest(opCtx(), nss, {});
+        NamespaceOrViewAcquisitionRequest acquisitionRequest(
+            opCtx(), nss, {}, AcquisitionPrerequisites::kWrite);
         ASSERT_EQ(dbVersion, acquisitionRequest.placementConcern.dbVersion);
         ASSERT_EQ(shardVersion, acquisitionRequest.placementConcern.shardVersion);
     }
@@ -221,15 +244,15 @@ TEST_F(ShardRoleTest, NamespaceOrViewAcquisitionRequestWithOpCtxTakesPlacementFr
 // Placement checks when acquiring unsharded collections
 
 TEST_F(ShardRoleTest, AcquireUnshardedCollWithCorrectPlacementVersion) {
-    NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern =
-        NamespaceOrViewAcquisitionRequest::PlacementConcern{dbVersionTestDb,
-                                                            ShardVersion::UNSHARDED()};
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{dbVersionTestDb, ShardVersion::UNSHARDED()};
     const auto acquisitions =
         acquireCollectionsOrViews(opCtx(),
                                   {{nssUnshardedCollection1,
                                     placementConcern,
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX);
 
     ASSERT_EQ(1, acquisitions.size());
@@ -237,20 +260,21 @@ TEST_F(ShardRoleTest, AcquireUnshardedCollWithCorrectPlacementVersion) {
     ASSERT_EQ(nssUnshardedCollection1, acquisitions.front().getCollectionPtr()->ns());
     ASSERT_FALSE(acquisitions.front().isView());
     ASSERT_FALSE(acquisitions.front().getShardingDescription().isSharded());
+    ASSERT_FALSE(acquisitions.front().getCollectionFilter().has_value());
 }
 
 TEST_F(ShardRoleTest, AcquireUnshardedCollWithIncorrectPlacementVersionThrows) {
     const auto incorrectDbVersion = DatabaseVersion(UUID::gen(), Timestamp(50, 0));
 
-    NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern =
-        NamespaceOrViewAcquisitionRequest::PlacementConcern{incorrectDbVersion,
-                                                            ShardVersion::UNSHARDED()};
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{incorrectDbVersion, ShardVersion::UNSHARDED()};
     ASSERT_THROWS_WITH_CHECK(
         acquireCollectionsOrViews(opCtx(),
                                   {{nssUnshardedCollection1,
                                     placementConcern,
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX),
         ExceptionFor<ErrorCodes::StaleDbVersion>,
         [&](const DBException& ex) {
@@ -271,15 +295,15 @@ TEST_F(ShardRoleTest, AcquireUnshardedCollWhenShardDoesNotKnowThePlacementVersio
         scopedDss->clearDbInfo(opCtx());
     }
 
-    NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern =
-        NamespaceOrViewAcquisitionRequest::PlacementConcern{dbVersionTestDb,
-                                                            ShardVersion::UNSHARDED()};
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{dbVersionTestDb, ShardVersion::UNSHARDED()};
     ASSERT_THROWS_WITH_CHECK(
         acquireCollectionsOrViews(opCtx(),
                                   {{nssUnshardedCollection1,
                                     placementConcern,
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX),
         ExceptionFor<ErrorCodes::StaleDbVersion>,
         [&](const DBException& ex) {
@@ -303,15 +327,15 @@ TEST_F(ShardRoleTest, AcquireUnshardedCollWhenCriticalSectionIsActiveThrows) {
     }
 
     {
-        NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern =
-            NamespaceOrViewAcquisitionRequest::PlacementConcern{dbVersionTestDb,
-                                                                ShardVersion::UNSHARDED()};
+        AcquisitionPrerequisites::PlacementConcern placementConcern =
+            AcquisitionPrerequisites::PlacementConcern{dbVersionTestDb, ShardVersion::UNSHARDED()};
         ASSERT_THROWS_WITH_CHECK(
             acquireCollectionsOrViews(opCtx(),
                                       {{nssUnshardedCollection1,
                                         placementConcern,
                                         repl::ReadConcernArgs(),
-                                        NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                        AcquisitionPrerequisites::kWrite,
+                                        AcquisitionPrerequisites::kMustBeCollection}},
                                       MODE_IX),
             ExceptionFor<ErrorCodes::StaleDbVersion>,
             [&](const DBException& ex) {
@@ -334,14 +358,15 @@ TEST_F(ShardRoleTest, AcquireUnshardedCollWhenCriticalSectionIsActiveThrows) {
 }
 
 TEST_F(ShardRoleTest, AcquireUnshardedCollWithoutSpecifyingPlacementVersion) {
-    NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern =
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
         NamespaceOrViewAcquisitionRequest::kPretendUnshardedDueToDirectConnection;
     const auto acquisitions =
         acquireCollectionsOrViews(opCtx(),
                                   {{nssUnshardedCollection1,
                                     {placementConcern},
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX);
 
     ASSERT_EQ(1, acquisitions.size());
@@ -349,21 +374,23 @@ TEST_F(ShardRoleTest, AcquireUnshardedCollWithoutSpecifyingPlacementVersion) {
     ASSERT_EQ(nssUnshardedCollection1, acquisitions.front().getCollectionPtr()->ns());
     ASSERT_FALSE(acquisitions.front().isView());
     ASSERT_FALSE(acquisitions.front().getShardingDescription().isSharded());
+    ASSERT_FALSE(acquisitions.front().getCollectionFilter().has_value());
 }
 
 // ---------------------------------------------------------------------------
 // Placement checks when acquiring sharded collections
 
 TEST_F(ShardRoleTest, AcquireShardedCollWithCorrectPlacementVersion) {
-    NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern =
-        NamespaceOrViewAcquisitionRequest::PlacementConcern{{} /* dbVersion */,
-                                                            shardVersionShardedCollection1};
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{{} /* dbVersion */,
+                                                   shardVersionShardedCollection1};
     const auto acquisitions =
         acquireCollectionsOrViews(opCtx(),
                                   {{nssShardedCollection1,
                                     placementConcern,
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX);
 
     ASSERT_EQ(1, acquisitions.size());
@@ -371,18 +398,19 @@ TEST_F(ShardRoleTest, AcquireShardedCollWithCorrectPlacementVersion) {
     ASSERT_EQ(nssShardedCollection1, acquisitions.front().getCollectionPtr()->ns());
     ASSERT_FALSE(acquisitions.front().isView());
     ASSERT_TRUE(acquisitions.front().getShardingDescription().isSharded());
+    ASSERT_TRUE(acquisitions.front().getCollectionFilter().has_value());
 }
 
 TEST_F(ShardRoleTest, AcquireShardedCollWithIncorrectPlacementVersionThrows) {
-    NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern =
-        NamespaceOrViewAcquisitionRequest::PlacementConcern{dbVersionTestDb,
-                                                            ShardVersion::UNSHARDED()};
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{dbVersionTestDb, ShardVersion::UNSHARDED()};
     ASSERT_THROWS_WITH_CHECK(
         acquireCollectionsOrViews(opCtx(),
                                   {{nssShardedCollection1,
                                     placementConcern,
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX),
         ExceptionFor<ErrorCodes::StaleConfig>,
         [&](const DBException& ex) {
@@ -404,14 +432,15 @@ TEST_F(ShardRoleTest, AcquireShardedCollWhenShardDoesNotKnowThePlacementVersionT
             ->clearFilteringMetadata(opCtx());
     }
 
-    NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern =
-        NamespaceOrViewAcquisitionRequest::PlacementConcern{{}, shardVersionShardedCollection1};
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{{}, shardVersionShardedCollection1};
     ASSERT_THROWS_WITH_CHECK(
         acquireCollectionsOrViews(opCtx(),
                                   {{nssShardedCollection1,
                                     placementConcern,
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX),
         ExceptionFor<ErrorCodes::StaleConfig>,
         [&](const DBException& ex) {
@@ -436,14 +465,15 @@ TEST_F(ShardRoleTest, AcquireShardedCollWhenCriticalSectionIsActiveThrows) {
     }
 
     {
-        NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern =
-            NamespaceOrViewAcquisitionRequest::PlacementConcern{{}, shardVersionShardedCollection1};
+        AcquisitionPrerequisites::PlacementConcern placementConcern =
+            AcquisitionPrerequisites::PlacementConcern{{}, shardVersionShardedCollection1};
         ASSERT_THROWS_WITH_CHECK(
             acquireCollectionsOrViews(opCtx(),
                                       {{nssShardedCollection1,
                                         placementConcern,
                                         repl::ReadConcernArgs(),
-                                        NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                        AcquisitionPrerequisites::kWrite,
+                                        AcquisitionPrerequisites::kMustBeCollection}},
                                       MODE_IX),
             ExceptionFor<ErrorCodes::StaleConfig>,
             [&](const DBException& ex) {
@@ -466,14 +496,15 @@ TEST_F(ShardRoleTest, AcquireShardedCollWhenCriticalSectionIsActiveThrows) {
 }
 
 TEST_F(ShardRoleTest, AcquireShardedCollWithoutSpecifyingPlacementVersion) {
-    NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern =
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
         NamespaceOrViewAcquisitionRequest::kPretendUnshardedDueToDirectConnection;
     const auto acquisitions =
         acquireCollectionsOrViews(opCtx(),
                                   {{nssShardedCollection1,
                                     placementConcern,
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX);
 
     ASSERT_EQ(1, acquisitions.size());
@@ -483,6 +514,7 @@ TEST_F(ShardRoleTest, AcquireShardedCollWithoutSpecifyingPlacementVersion) {
 
     // Note that the collection is treated as unsharded because the operation is unversioned.
     ASSERT_FALSE(acquisitions.front().getShardingDescription().isSharded());
+    ASSERT_FALSE(acquisitions.front().getCollectionFilter().has_value());
 }
 
 // ---------------------------------------------------------------------------
@@ -491,16 +523,16 @@ TEST_F(ShardRoleTest, AcquireShardedCollWithoutSpecifyingPlacementVersion) {
 TEST_F(ShardRoleTest, AcquireCollectionFailsIfItDoesNotExist) {
     const NamespaceString inexistentNss =
         NamespaceString::createNamespaceString_forTest(dbNameTestDb, "inexistent");
-    NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern;
-    ASSERT_THROWS_CODE(
-        acquireCollectionsOrViews(opCtx(),
-                                  {{inexistentNss,
-                                    placementConcern,
-                                    repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
-                                  MODE_IX),
-        DBException,
-        ErrorCodes::NamespaceNotFound);
+    AcquisitionPrerequisites::PlacementConcern placementConcern;
+    ASSERT_THROWS_CODE(acquireCollectionsOrViews(opCtx(),
+                                                 {{inexistentNss,
+                                                   placementConcern,
+                                                   repl::ReadConcernArgs(),
+                                                   AcquisitionPrerequisites::kWrite,
+                                                   AcquisitionPrerequisites::kMustBeCollection}},
+                                                 MODE_IX),
+                       DBException,
+                       ErrorCodes::NamespaceNotFound);
 }
 
 TEST_F(ShardRoleTest, AcquireInexistentCollectionWithWrongPlacementThrowsBecauseWrongPlacement) {
@@ -508,13 +540,14 @@ TEST_F(ShardRoleTest, AcquireInexistentCollectionWithWrongPlacementThrowsBecause
     const NamespaceString inexistentNss =
         NamespaceString::createNamespaceString_forTest(dbNameTestDb, "inexistent");
 
-    NamespaceOrViewAcquisitionRequest::PlacementConcern placementConcern{incorrectDbVersion, {}};
+    AcquisitionPrerequisites::PlacementConcern placementConcern{incorrectDbVersion, {}};
     ASSERT_THROWS_WITH_CHECK(
         acquireCollectionsOrViews(opCtx(),
                                   {{inexistentNss,
                                     placementConcern,
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX),
         ExceptionFor<ErrorCodes::StaleDbVersion>,
         [&](const DBException& ex) {
@@ -533,14 +566,15 @@ TEST_F(ShardRoleTest, AcquireMultipleCollectionsAllWithCorrectPlacementConcern)
     const auto acquisitions = acquireCollectionsOrViews(
         opCtx(),
         {{nssUnshardedCollection1,
-          NamespaceOrViewAcquisitionRequest::PlacementConcern{dbVersionTestDb,
-                                                              ShardVersion::UNSHARDED()},
+          AcquisitionPrerequisites::PlacementConcern{dbVersionTestDb, ShardVersion::UNSHARDED()},
           repl::ReadConcernArgs(),
-          NamespaceOrViewAcquisitionRequest::kMustBeCollection},
+          AcquisitionPrerequisites::kWrite,
+          AcquisitionPrerequisites::kMustBeCollection},
          {nssShardedCollection1,
-          NamespaceOrViewAcquisitionRequest::PlacementConcern{{}, shardVersionShardedCollection1},
+          AcquisitionPrerequisites::PlacementConcern{{}, shardVersionShardedCollection1},
           repl::ReadConcernArgs(),
-          NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+          AcquisitionPrerequisites::kWrite,
+          AcquisitionPrerequisites::kMustBeCollection}},
         MODE_IX);
 
     ASSERT_EQ(2, acquisitions.size());
@@ -554,6 +588,7 @@ TEST_F(ShardRoleTest, AcquireMultipleCollectionsAllWithCorrectPlacementConcern)
     ASSERT(acquisitionUnshardedColl != acquisitions.end());
     ASSERT_FALSE(acquisitionUnshardedColl->isView());
     ASSERT_FALSE(acquisitionUnshardedColl->getShardingDescription().isSharded());
+    ASSERT_FALSE(acquisitionUnshardedColl->getCollectionFilter().has_value());
 
     const auto& acquisitionShardedColl =
         std::find_if(acquisitions.begin(),
@@ -564,21 +599,24 @@ TEST_F(ShardRoleTest, AcquireMultipleCollectionsAllWithCorrectPlacementConcern)
     ASSERT(acquisitionShardedColl != acquisitions.end());
     ASSERT_FALSE(acquisitionShardedColl->isView());
     ASSERT_TRUE(acquisitionShardedColl->getShardingDescription().isSharded());
+    ASSERT_TRUE(acquisitionShardedColl->getCollectionFilter().has_value());
 }
 
 TEST_F(ShardRoleTest, AcquireMultipleCollectionsWithIncorrectPlacementConcernThrows) {
     ASSERT_THROWS_WITH_CHECK(
         acquireCollectionsOrViews(opCtx(),
                                   {{nssUnshardedCollection1,
-                                    NamespaceOrViewAcquisitionRequest::PlacementConcern{
+                                    AcquisitionPrerequisites::PlacementConcern{
                                         dbVersionTestDb, ShardVersion::UNSHARDED()},
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection},
                                    {nssShardedCollection1,
-                                    NamespaceOrViewAcquisitionRequest::PlacementConcern{
+                                    AcquisitionPrerequisites::PlacementConcern{
                                         dbVersionTestDb, ShardVersion::UNSHARDED()},
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX),
         ExceptionFor<ErrorCodes::StaleConfig>,
         [&](const DBException& ex) {
@@ -596,14 +634,14 @@ TEST_F(ShardRoleTest, AcquireMultipleCollectionsWithIncorrectPlacementConcernThr
 
 TEST_F(ShardRoleTest, AcquireCollectionByUUID) {
     const auto uuid = getCollectionUUID(opCtx(), nssUnshardedCollection1);
-    const auto acquisitions =
-        acquireCollectionsOrViews(opCtx(),
-                                  {{NamespaceStringOrUUID(dbNameTestDb, uuid),
-                                    NamespaceOrViewAcquisitionRequest::PlacementConcern{
-                                        dbVersionTestDb, ShardVersion::UNSHARDED()},
-                                    repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
-                                  MODE_IX);
+    const auto acquisitions = acquireCollectionsOrViews(
+        opCtx(),
+        {{NamespaceStringOrUUID(dbNameTestDb, uuid),
+          AcquisitionPrerequisites::PlacementConcern{dbVersionTestDb, ShardVersion::UNSHARDED()},
+          repl::ReadConcernArgs(),
+          AcquisitionPrerequisites::kWrite,
+          AcquisitionPrerequisites::kMustBeCollection}},
+        MODE_IX);
 
     ASSERT_EQ(1, acquisitions.size());
     ASSERT_EQ(nssUnshardedCollection1, acquisitions.front().nss());
@@ -612,28 +650,28 @@ TEST_F(ShardRoleTest, AcquireCollectionByUUID) {
 
 TEST_F(ShardRoleTest, AcquireCollectionByUUIDButWrongDbNameThrows) {
     const auto uuid = getCollectionUUID(opCtx(), nssUnshardedCollection1);
-    ASSERT_THROWS_CODE(
-        acquireCollectionsOrViews(opCtx(),
-                                  {{NamespaceStringOrUUID("anotherDbName", uuid),
-                                    {},
-                                    repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
-                                  MODE_IX),
-        DBException,
-        ErrorCodes::NamespaceNotFound);
+    ASSERT_THROWS_CODE(acquireCollectionsOrViews(opCtx(),
+                                                 {{NamespaceStringOrUUID("anotherDbName", uuid),
+                                                   {},
+                                                   repl::ReadConcernArgs(),
+                                                   AcquisitionPrerequisites::kWrite,
+                                                   AcquisitionPrerequisites::kMustBeCollection}},
+                                                 MODE_IX),
+                       DBException,
+                       ErrorCodes::NamespaceNotFound);
 }
 
 TEST_F(ShardRoleTest, AcquireCollectionByWrongUUID) {
     const auto uuid = UUID::gen();
-    ASSERT_THROWS_CODE(
-        acquireCollectionsOrViews(opCtx(),
-                                  {{NamespaceStringOrUUID(dbNameTestDb, uuid),
-                                    {},
-                                    repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
-                                  MODE_IX),
-        DBException,
-        ErrorCodes::NamespaceNotFound);
+    ASSERT_THROWS_CODE(acquireCollectionsOrViews(opCtx(),
+                                                 {{NamespaceStringOrUUID(dbNameTestDb, uuid),
+                                                   {},
+                                                   repl::ReadConcernArgs(),
+                                                   AcquisitionPrerequisites::kWrite,
+                                                   AcquisitionPrerequisites::kMustBeCollection}},
+                                                 MODE_IX),
+                       DBException,
+                       ErrorCodes::NamespaceNotFound);
 }
 
 // ---------------------------------------------------------------------------
@@ -647,7 +685,8 @@ TEST_F(ShardRoleTest, AcquireCollectionByNssAndExpectedUUID) {
                                     uuid,
                                     {},
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX);
 
     ASSERT_EQ(1, acquisitions.size());
@@ -664,7 +703,8 @@ TEST_F(ShardRoleTest, AcquireCollectionByNssAndWrongExpectedUUIDThrows) {
                                     wrongUuid,
                                     {},
                                     repl::ReadConcernArgs(),
-                                    NamespaceOrViewAcquisitionRequest::kMustBeCollection}},
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
                                   MODE_IX),
         ExceptionFor<ErrorCodes::CollectionUUIDMismatch>,
         [&](const DBException& ex) {
@@ -676,5 +716,311 @@ TEST_F(ShardRoleTest, AcquireCollectionByNssAndWrongExpectedUUIDThrows) {
         });
 }
 
+// ---------------------------------------------------------------------------
+// Yield and restore
+
+TEST_F(ShardRoleTest, YieldAndRestoreAcquisitionWithLocks) {
+    const auto nss = nssUnshardedCollection1;
+
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{dbVersionTestDb, ShardVersion::UNSHARDED()};
+    const auto acquisition =
+        acquireCollectionsOrViews(opCtx(),
+                                  {{nss,
+                                    placementConcern,
+                                    repl::ReadConcernArgs(),
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
+                                  MODE_IX);
+
+    ASSERT_TRUE(opCtx()->lockState()->isCollectionLockedForMode(nss, MODE_IX));
+
+    // Yield the resources
+    auto yieldedTransactionResources = yieldTransactionResourcesFromOperationContext(opCtx());
+    ASSERT_FALSE(opCtx()->lockState()->isCollectionLockedForMode(nss, MODE_IX));
+
+    // Restore the resources
+    restoreTransactionResourcesToOperationContext(opCtx(), std::move(yieldedTransactionResources));
+    ASSERT_TRUE(opCtx()->lockState()->isCollectionLockedForMode(nss, MODE_IX));
+}
+
+TEST_F(ShardRoleTest, RestoreForWriteFailsIfPlacementConcernNoLongerMet) {
+    const auto nss = nssShardedCollection1;
+
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{{}, shardVersionShardedCollection1};
+    const auto acquisition =
+        acquireCollectionsOrViews(opCtx(),
+                                  {{nss,
+                                    placementConcern,
+                                    repl::ReadConcernArgs(),
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
+                                  MODE_IX);
+
+    // Yield the resources
+    auto yieldedTransactionResources = yieldTransactionResourcesFromOperationContext(opCtx());
+
+    // Placement changes
+    const auto newShardVersion = [&]() {
+        auto newPlacementVersion = shardVersionShardedCollection1.placementVersion();
+        newPlacementVersion.incMajor();
+        return ShardVersion(newPlacementVersion, boost::optional<CollectionIndexes>(boost::none));
+    }();
+    const auto uuid = getCollectionUUID(opCtx(), nss);
+    installShardedCollectionMetadata(
+        opCtx(),
+        nss,
+        dbVersionTestDb,
+        {ChunkType(uuid,
+                   ChunkRange{BSON("skey" << MINKEY), BSON("skey" << MAXKEY)},
+                   newShardVersion.placementVersion(),
+                   thisShardId)},
+        thisShardId);
+
+    // Try to restore the resources should fail because placement concern is no longer met.
+    ASSERT_THROWS_WITH_CHECK(restoreTransactionResourcesToOperationContext(
+                                 opCtx(), std::move(yieldedTransactionResources)),
+                             ExceptionFor<ErrorCodes::StaleConfig>,
+                             [&](const DBException& ex) {
+                                 const auto exInfo = ex.extraInfo<StaleConfigInfo>();
+                                 ASSERT_EQ(nssShardedCollection1, exInfo->getNss());
+                                 ASSERT_EQ(shardVersionShardedCollection1,
+                                           exInfo->getVersionReceived());
+                                 ASSERT_EQ(newShardVersion, exInfo->getVersionWanted());
+                                 ASSERT_EQ(ShardId("this"), exInfo->getShardId());
+                                 ASSERT_FALSE(exInfo->getCriticalSectionSignal().is_initialized());
+                             });
+
+    ASSERT_FALSE(opCtx()->lockState()->isCollectionLockedForMode(nss, MODE_IX));
+}
+
+TEST_F(ShardRoleTest, RestoreWithShardVersionIgnored) {
+    const auto nss = nssShardedCollection1;
+
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{{}, ShardVersion::IGNORED()};
+    const auto acquisition =
+        acquireCollectionsOrViews(opCtx(),
+                                  {{nss,
+                                    placementConcern,
+                                    repl::ReadConcernArgs(),
+                                    AcquisitionPrerequisites::kWrite,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
+                                  MODE_IX);
+
+    ASSERT_TRUE(acquisition.front().getShardingDescription().isSharded());
+    ASSERT_TRUE(acquisition.front().getCollectionFilter().has_value());
+
+    // Yield the resources
+    auto yieldedTransactionResources = yieldTransactionResourcesFromOperationContext(opCtx());
+
+    // Placement changes
+    const auto newShardVersion = [&]() {
+        auto newPlacementVersion = shardVersionShardedCollection1.placementVersion();
+        newPlacementVersion.incMajor();
+        return ShardVersion(newPlacementVersion, boost::optional<CollectionIndexes>(boost::none));
+    }();
+
+    const auto uuid = getCollectionUUID(opCtx(), nss);
+    installShardedCollectionMetadata(
+        opCtx(),
+        nss,
+        dbVersionTestDb,
+        {ChunkType(uuid,
+                   ChunkRange{BSON("skey" << MINKEY), BSON("skey" << MAXKEY)},
+                   newShardVersion.placementVersion(),
+                   thisShardId)},
+        thisShardId);
+
+    // Try to restore the resources should work because placement concern (IGNORED) can be met.
+    restoreTransactionResourcesToOperationContext(opCtx(), std::move(yieldedTransactionResources));
+    ASSERT_TRUE(opCtx()->lockState()->isCollectionLockedForMode(nss, MODE_IX));
+}
+
+void ShardRoleTest::testRestoreFailsIfCollectionNoLongerExists(
+    AcquisitionPrerequisites::OperationType operationType) {
+    const auto nss = nssShardedCollection1;
+
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{{}, shardVersionShardedCollection1};
+    const auto acquisition =
+        acquireCollectionsOrViews(opCtx(),
+                                  {{nss,
+                                    placementConcern,
+                                    repl::ReadConcernArgs(),
+                                    operationType,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
+                                  MODE_IX);
+
+    // Yield the resources
+    auto yieldedTransactionResources = yieldTransactionResourcesFromOperationContext(opCtx());
+
+    // Drop the collection
+    {
+        DBDirectClient client(opCtx());
+        client.dropCollection(nss);
+    }
+
+    // Try to restore the resources should fail because the collection no longer exists.
+    ASSERT_THROWS_CODE(restoreTransactionResourcesToOperationContext(
+                           opCtx(), std::move(yieldedTransactionResources)),
+                       DBException,
+                       ErrorCodes::NamespaceNotFound);
+}
+TEST_F(ShardRoleTest, RestoreForReadFailsIfCollectionNoLongerExists) {
+    testRestoreFailsIfCollectionNoLongerExists(AcquisitionPrerequisites::kRead);
+}
+TEST_F(ShardRoleTest, RestoreForWriteFailsIfCollectionNoLongerExists) {
+    testRestoreFailsIfCollectionNoLongerExists(AcquisitionPrerequisites::kWrite);
+}
+
+void ShardRoleTest::testRestoreFailsIfCollectionRenamed(
+    AcquisitionPrerequisites::OperationType operationType) {
+    const auto nss = nssUnshardedCollection1;
+
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{dbVersionTestDb, ShardVersion::UNSHARDED()};
+    const auto acquisition =
+        acquireCollectionsOrViews(opCtx(),
+                                  {{nss,
+                                    placementConcern,
+                                    repl::ReadConcernArgs(),
+                                    operationType,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
+                                  MODE_IX);
+
+    // Yield the resources
+    auto yieldedTransactionResources = yieldTransactionResourcesFromOperationContext(opCtx());
+
+    // Rename the collection.
+    {
+        DBDirectClient client(opCtx());
+        BSONObj info;
+        ASSERT_TRUE(client.runCommand(
+            DatabaseName(boost::none, dbNameTestDb.db()),
+            BSON("renameCollection"
+                 << nss.ns() << "to"
+                 << NamespaceString::createNamespaceString_forTest(dbNameTestDb, "foo2").ns()),
+            info));
+    }
+
+    // Try to restore the resources should fail because the collection has been renamed.
+    ASSERT_THROWS_CODE(restoreTransactionResourcesToOperationContext(
+                           opCtx(), std::move(yieldedTransactionResources)),
+                       DBException,
+                       ErrorCodes::NamespaceNotFound);
+}
+TEST_F(ShardRoleTest, RestoreForReadFailsIfCollectionRenamed) {
+    testRestoreFailsIfCollectionRenamed(AcquisitionPrerequisites::kRead);
+}
+TEST_F(ShardRoleTest, RestoreForWriteFailsIfCollectionRenamed) {
+    testRestoreFailsIfCollectionRenamed(AcquisitionPrerequisites::kWrite);
+}
+
+void ShardRoleTest::testRestoreFailsIfCollectionDroppedAndRecreated(
+    AcquisitionPrerequisites::OperationType operationType) {
+    const auto nss = nssUnshardedCollection1;
+
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{dbVersionTestDb, ShardVersion::UNSHARDED()};
+    const auto acquisition =
+        acquireCollectionsOrViews(opCtx(),
+                                  {{nss,
+                                    placementConcern,
+                                    repl::ReadConcernArgs(),
+                                    operationType,
+                                    AcquisitionPrerequisites::kMustBeCollection}},
+                                  MODE_IX);
+
+    // Yield the resources
+    auto yieldedTransactionResources = yieldTransactionResourcesFromOperationContext(opCtx());
+
+    // Drop the collection and create a new one with the same nss.
+    {
+        DBDirectClient client(opCtx());
+        client.dropCollection(nss);
+        createTestCollection(opCtx(), nss);
+    }
+
+    // Try to restore the resources should fail because the collection no longer exists.
+    ASSERT_THROWS_CODE(restoreTransactionResourcesToOperationContext(
+                           opCtx(), std::move(yieldedTransactionResources)),
+                       DBException,
+                       ErrorCodes::CollectionUUIDMismatch);
+}
+TEST_F(ShardRoleTest, RestoreForWriteFailsIfCollectionDroppedAndRecreated) {
+    testRestoreFailsIfCollectionDroppedAndRecreated(AcquisitionPrerequisites::kWrite);
+}
+TEST_F(ShardRoleTest, RestoreForReadFailsIfCollectionDroppedAndRecreated) {
+    testRestoreFailsIfCollectionDroppedAndRecreated(AcquisitionPrerequisites::kRead);
+}
+
+TEST_F(ShardRoleTest, RestoreForReadSucceedsEvenIfPlacementHasChanged) {
+    const auto nss = nssShardedCollection1;
+
+    AcquisitionPrerequisites::PlacementConcern placementConcern =
+        AcquisitionPrerequisites::PlacementConcern{{}, shardVersionShardedCollection1};
+
+    SharedSemiFuture<void> ongoingQueriesCompletionFuture;
+
+    {
+        const auto acquisition =
+            acquireCollectionsOrViews(opCtx(),
+                                      {{nss,
+                                        placementConcern,
+                                        repl::ReadConcernArgs(),
+                                        AcquisitionPrerequisites::kRead,
+                                        AcquisitionPrerequisites::kMustBeCollection}},
+                                      MODE_IX);
+
+        ongoingQueriesCompletionFuture =
+            CollectionShardingRuntime::assertCollectionLockedAndAcquireShared(opCtx(), nss)
+                ->getOngoingQueriesCompletionFuture(
+                    getCollectionUUID(opCtx(), nss),
+                    ChunkRange(BSON("skey" << MINKEY), BSON("skey" << MAXKEY)));
+
+        // Yield the resources
+        auto yieldedTransactionResources = yieldTransactionResourcesFromOperationContext(opCtx());
+
+        ASSERT_FALSE(ongoingQueriesCompletionFuture.isReady());
+        ASSERT_TRUE(acquisition.front().getCollectionFilter().has_value());
+        ASSERT_TRUE(acquisition.front().getCollectionFilter()->keyBelongsToMe(BSON("skey" << 0)));
+
+        // Placement changes
+        const auto newShardVersion = [&]() {
+            auto newPlacementVersion = shardVersionShardedCollection1.placementVersion();
+            newPlacementVersion.incMajor();
+            return ShardVersion(newPlacementVersion,
+                                boost::optional<CollectionIndexes>(boost::none));
+        }();
+
+        const auto uuid = getCollectionUUID(opCtx(), nss);
+        installShardedCollectionMetadata(
+            opCtx(),
+            nss,
+            dbVersionTestDb,
+            {ChunkType(uuid,
+                       ChunkRange{BSON("skey" << MINKEY), BSON("skey" << MAXKEY)},
+                       newShardVersion.placementVersion(),
+                       ShardId("anotherShard"))},
+            thisShardId);
+
+        // Restore should work for reads even though placement has changed.
+        restoreTransactionResourcesToOperationContext(opCtx(),
+                                                      std::move(yieldedTransactionResources));
+
+        ASSERT_FALSE(ongoingQueriesCompletionFuture.isReady());
+
+        // Even though placement has changed, the filter (and preserver) still point to the original
+        // placement.
+        ASSERT_TRUE(acquisition.front().getCollectionFilter().has_value());
+        ASSERT_TRUE(acquisition.front().getCollectionFilter()->keyBelongsToMe(BSON("skey" << 0)));
+    }
+
+    // Acquisition released. Now the range is no longer in use.
+    ASSERT_TRUE(ongoingQueriesCompletionFuture.isReady());
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/transaction_resources.cpp b/src/mongo/db/transaction_resources.cpp
index 5b1a33f1923..da48397c968 100644
--- a/src/mongo/db/transaction_resources.cpp
+++ b/src/mongo/db/transaction_resources.cpp
@@ -35,12 +35,12 @@ namespace shard_role_details {
 TransactionResources::TransactionResources(repl::ReadConcernArgs readConcern)
     : readConcern(std::move(readConcern)) {}
 
-TransactionResources::TransactionResources(TransactionResources&& other) {
-    *this = std::move(other);
+void TransactionResources::releaseAllResourcesOnCommitOrAbort() noexcept {
+    locker.reset();
+    lockSnapshot.reset();
+    acquiredCollections.clear();
 }
 
-TransactionResources& TransactionResources::operator=(TransactionResources&&) = default;
-
 TransactionResources::~TransactionResources() {
     invariant(!locker);
     invariant(!lockSnapshot);
diff --git a/src/mongo/db/transaction_resources.h b/src/mongo/db/transaction_resources.h
index 8978d83d5f3..594d59532e1 100644
--- a/src/mongo/db/transaction_resources.h
+++ b/src/mongo/db/transaction_resources.h
@@ -41,8 +41,48 @@
 #include "mongo/util/uuid.h"
 
 namespace mongo {
+
+struct AcquisitionPrerequisites {
+    struct PlacementConcern {
+        boost::optional<DatabaseVersion> dbVersion;
+        boost::optional<ShardVersion> shardVersion;
+    };
+
+    enum ViewMode { kMustBeCollection, kCanBeView };
+
+    enum OperationType { kRead, kWrite };
+
+    AcquisitionPrerequisites(NamespaceString nss,
+                             boost::optional<UUID> uuid,
+                             PlacementConcern placementConcern,
+                             OperationType operationType,
+                             ViewMode viewMode)
+        : nss(std::move(nss)),
+          uuid(std::move(uuid)),
+          placementConcern(std::move(placementConcern)),
+          operationType(operationType),
+          viewMode(viewMode) {}
+
+    NamespaceString nss;
+    boost::optional<UUID> uuid;
+
+    PlacementConcern placementConcern;
+    OperationType operationType;
+    ViewMode viewMode;
+};
+
 namespace shard_role_details {
 
+struct AcquiredCollection {
+    AcquisitionPrerequisites prerequisites;
+
+    boost::optional<Lock::DBLock> dbLock;
+    boost::optional<Lock::CollectionLock> collectionLock;
+    ScopedCollectionDescription collectionDescription;
+    boost::optional<ScopedCollectionFilter> ownershipFilter;
+    CollectionPtr collectionPtr;
+};
+
 /**
  * This class is a container for all the collection resources which are currently acquired by a
  * given operation. Operations consist of one or more transactions, which "acquire" and "release"
@@ -82,23 +122,14 @@ namespace shard_role_details {
 struct TransactionResources {
     TransactionResources(repl::ReadConcernArgs readConcern);
 
-    TransactionResources(TransactionResources&&);
-    TransactionResources& operator=(TransactionResources&&);
+    TransactionResources(TransactionResources&&) = delete;
+    TransactionResources& operator=(TransactionResources&&) = delete;
 
     TransactionResources(TransactionResources&) = delete;
     TransactionResources& operator=(TransactionResources&) = delete;
 
     ~TransactionResources();
 
-    struct AcquiredCollection {
-        NamespaceString nss;
-        UUID uuid;
-        CollectionPtr collectionPtr;
-        ScopedCollectionDescription collectionDescription;
-        boost::optional<Lock::DBLock> dbLock;
-        boost::optional<Lock::CollectionLock> collectionLock;
-    };
-
     // void addAcquiredCollection(const NamespaceString& nss, UUID uuid);
     const AcquiredCollection& addAcquiredCollection(AcquiredCollection&& acquiredCollection) {
         return acquiredCollections.emplace_back(std::move(acquiredCollection));
@@ -106,7 +137,7 @@ struct TransactionResources {
 
     void releaseCollection(UUID uuid);
 
-    void releaseAllResourcesOnCommitOrAbort();
+    void releaseAllResourcesOnCommitOrAbort() noexcept;
 
     // The read concern with which the whole operation started. Remains the same for the duration of
     // the entire operation.