SERVER-58663 Create a lock-free multi collection AutoGet* helper

7 files changed, 485 insertions, 1 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 40d29dff810..028c0d3d3f5 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -2377,6 +2377,7 @@ if wiredtiger:
             'dbdirectclient_test.cpp',
             'dbmessage_test.cpp',
             'db_raii_test.cpp',
+            'db_raii_multi_collection_test.cpp',
             "explain_test.cpp",
             'field_parser_test.cpp',
             'field_ref_set_test.cpp',
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index 66dc7be085d..4cba32ed567 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -827,6 +827,14 @@ public:
     void yield() const override;
     void restore() const override;
 
+    RestoreFn detachRestoreFn() {
+        return std::move(_restoreFn);
+    }
+
+    void attachRestoreFn(RestoreFn newRestoreFn) {
+        _restoreFn = std::move(newRestoreFn);
+    }
+
     friend std::ostream& operator<<(std::ostream& os, const CollectionPtr& coll);
 
     void setShardKeyPattern(const BSONObj& shardKeyPattern) {
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index eb9ce38cc41..5eed6683a7d 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -218,9 +218,18 @@ public:
 
     /**
      * This function gets the Collection pointer that corresponds to the CollectionUUID.
+     *
      * The required locks must be obtained prior to calling this function, or else the found
      * Collection pointer might no longer be valid when the call returns.
      *
+     * 'lookupCollectionByUUIDForMetadataWrite' requires a MODE_X collection lock, returns a copy to
+     * the caller because catalog updates are copy-on-write.
+     *
+     * 'lookupCollectionByUUID' requires a MODE_IS collection lock.
+     *
+     * 'lookupCollectionByUUIDForRead' does not require locks and should only be used in the context
+     * of a lock-free read wherein we also have a consistent storage snapshot.
+     *
      * Returns nullptr if the 'uuid' is not known.
      */
     Collection* lookupCollectionByUUIDForMetadataWrite(OperationContext* opCtx,
@@ -237,10 +246,19 @@ public:
     bool isCollectionAwaitingVisibility(CollectionUUID uuid) const;
 
     /**
-     * This function gets the Collection pointer that corresponds to the NamespaceString.
+     * These functions fetch a Collection pointer that corresponds to the NamespaceString.
+     *
      * The required locks must be obtained prior to calling this function, or else the found
      * Collection pointer may no longer be valid when the call returns.
      *
+     * 'lookupCollectionByNamespaceForMetadataWrite' requires a MODE_X collection lock, returns a
+     * copy to the caller because catalog updates are copy-on-write.
+     *
+     * 'lookupCollectionByNamespace' requires a MODE_IS collection lock.
+     *
+     * 'lookupCollectionByNamespaceForRead' does not require locks and should only be used in the
+     * context of a lock-free read wherein we also have a consistent storage snapshot.
+     *
      * Returns nullptr if the namespace is unknown.
      */
     Collection* lookupCollectionByNamespaceForMetadataWrite(OperationContext* opCtx,
diff --git a/src/mongo/db/catalog_raii.h b/src/mongo/db/catalog_raii.h
index 3c9c541fb9a..84445f3ae6f 100644
--- a/src/mongo/db/catalog_raii.h
+++ b/src/mongo/db/catalog_raii.h
@@ -183,6 +183,16 @@ public:
     }
 
 protected:
+    template <typename AutoGetCollectionType, typename EmplaceAutoGetCollectionFunc>
+    friend class AutoGetCollectionForReadBase;
+
+    /**
+     * Allow access to the CollectionPtr as non-const, for friend classes.
+     */
+    CollectionPtr& _getCollectionPtrForModify() {
+        return _coll;
+    }
+
     OperationContext* _opCtx = nullptr;
     AutoGetDb _autoDb;
     boost::optional<Lock::CollectionLock> _collLock;
@@ -273,6 +283,16 @@ public:
     }
 
 private:
+    template <typename AutoGetCollectionType, typename EmplaceAutoGetCollectionFunc>
+    friend class AutoGetCollectionForReadBase;
+
+    /**
+     * Allow access to the CollectionPtr as non-const, for friend classes.
+     */
+    CollectionPtr& _getCollectionPtrForModify() {
+        return _collectionPtr;
+    }
+
     // Indicate that we are lock-free on code paths that can run either lock-free or locked for
     // different kinds of operations. Note: this class member is currently declared first so that it
     // destructs last, as a safety measure, but not because it is currently depended upon behavior.
diff --git a/src/mongo/db/db_raii.cpp b/src/mongo/db/db_raii.cpp
index 7a0eda9b506..1cda869dc71 100644
--- a/src/mongo/db/db_raii.cpp
+++ b/src/mongo/db/db_raii.cpp
@@ -186,6 +186,51 @@ auto acquireCollectionAndConsistentSnapshot(
     return collection;
 }
 
+/**
+ * Checks that the 'collection' is not null, that the 'collection' is not sharded and that the
+ * minimum visible timestamp of 'collection' is compatible with 'readTimestamp', if 'readTimestamp;
+ * is set.
+ *
+ * Returns OK, or either SnapshotUnavailable or NamespaceNotFound.
+ * Invariants that the collection is not sharded.
+ */
+Status checkSecondaryCollection(OperationContext* opCtx,
+                                boost::optional<NamespaceString> nss,
+                                boost::optional<CollectionUUID> uuid,
+                                const std::shared_ptr<const Collection>& collection,
+                                boost::optional<Timestamp> readTimestamp) {
+    invariant(nss || uuid);
+
+    // Check that the collection exists.
+    if (!collection) {
+        return Status(ErrorCodes::NamespaceNotFound,
+                      str::stream() << "Could not find collection '"
+                                    << (nss ? nss->toString() : uuid->toString()) << "'");
+    }
+
+    // Secondary collections of a query are not allowed to be sharded.
+    auto collDesc = CollectionShardingState::getSharedForLockFreeReads(opCtx, collection->ns())
+                        ->getCollectionDescription(opCtx);
+    invariant(!collDesc.isSharded());
+
+    // Ensure the readTimestamp is not older than the collection's minimum visible timestamp.
+    auto minSnapshot = collection->getMinimumVisibleSnapshot();
+    if (SnapshotHelper::collectionChangesConflictWithRead(minSnapshot, readTimestamp)) {
+        // Note: SnapshotHelper::collectionChangesConflictWithRead returns false if either
+        // minSnapshot or readTimestamp is not set, so it's safe to print them below.
+        return Status(ErrorCodes::SnapshotUnavailable,
+                      str::stream()
+                          << "Unable to read from a snapshot due to pending collection catalog "
+                             "changes to collection '"
+                          << collection->ns()
+                          << "'; please retry the operation. Snapshot timestamp is "
+                          << readTimestamp->toString() << ". Collection minimum timestamp is "
+                          << minSnapshot->toString());
+    }
+
+    return Status::OK();
+}
+
 }  // namespace
 
 AutoStatsTracker::AutoStatsTracker(OperationContext* opCtx,
@@ -592,6 +637,71 @@ AutoGetCollectionForReadCommandBase<AutoGetCollectionForReadType>::
     }
 }
 
+void AutoGetCollectionMultiForReadCommandLockFree::_secondaryCollectionsRestoreFn(
+    OperationContext* opCtx) {
+    const auto readTimestamp = opCtx->recoveryUnit()->getPointInTimeReadTimestamp(opCtx);
+    auto catalog = CollectionCatalog::get(opCtx);
+
+    // Fetch secondary collections from the catalog and check they're valid to use.
+    for (auto& secondaryUUID : _secondaryCollectionUUIDs) {
+        auto sharedCollPtr = catalog->lookupCollectionByUUIDForRead(opCtx, secondaryUUID);
+        uassertStatusOK(checkSecondaryCollection(
+            opCtx, /*nss*/ boost::none, secondaryUUID, sharedCollPtr, readTimestamp));
+    }
+};
+
+AutoGetCollectionMultiForReadCommandLockFree::AutoGetCollectionMultiForReadCommandLockFree(
+    OperationContext* opCtx,
+    const NamespaceStringOrUUID& primaryNssOrUUID,
+    std::vector<NamespaceStringOrUUID>& secondaryNsOrUUIDs,
+    AutoGetCollectionViewMode viewMode,
+    Date_t deadline,
+    AutoStatsTracker::LogMode logMode)
+    // Set up state regularly for a single collection access. This will handle setting up a
+    // consistent storage snapshot and a PIT in-memory catalog. We can then use the catalog to fetch
+    // and verify secondary collection state.
+    : _autoCollForReadCommandLockFree(opCtx, primaryNssOrUUID, viewMode, deadline, logMode) {
+    if (!_autoCollForReadCommandLockFree) {
+        return;
+    }
+
+    // Fetch secondary collection and verify they're valid for use.
+    {
+        auto catalog = CollectionCatalog::get(opCtx);
+        const auto readTimestamp = opCtx->recoveryUnit()->getPointInTimeReadTimestamp(opCtx);
+        for (auto& secondaryNssOrUUID : secondaryNsOrUUIDs) {
+            auto nss = catalog->resolveNamespaceStringOrUUID(opCtx, secondaryNssOrUUID);
+            auto sharedCollPtr = catalog->lookupCollectionByNamespaceForRead(opCtx, nss);
+
+            // Check that 'sharedCollPtr' exists and is safe to use.
+            uassertStatusOK(checkSecondaryCollection(
+                opCtx, nss, /*uuid*/ boost::none, sharedCollPtr, readTimestamp));
+
+            // Duplicate collection names should not be provided via 'secondaryNsOrUUIDs'.
+            invariant(std::find(_secondaryCollectionUUIDs.begin(),
+                                _secondaryCollectionUUIDs.end(),
+                                sharedCollPtr->uuid()) == _secondaryCollectionUUIDs.end());
+            _secondaryCollectionUUIDs.push_back(sharedCollPtr->uuid());
+        }
+    }
+
+    // Create a new restore from yield function to pass into all of the 'primary' CollectionPtr
+    // instance. It should encapsulate the logic _autoCollForReadCommandLockFree's CollectionPtr
+    // currently has and then add logic to check the secondary collections' state.
+
+    // Save the 'primary' CollectionPtr's original restore function so that the new restore function
+    // can reference it.
+    _primaryCollectionRestoreFn =
+        _autoCollForReadCommandLockFree._getCollectionPtrForModify().detachRestoreFn();
+
+    _autoCollForReadCommandLockFree._getCollectionPtrForModify().attachRestoreFn(
+        [&](OperationContext* opCtx, CollectionUUID collUUID) {
+            const Collection* primaryCollection = _primaryCollectionRestoreFn(opCtx, collUUID);
+            _secondaryCollectionsRestoreFn(opCtx);
+            return primaryCollection;
+        });
+}
+
 OldClientContext::OldClientContext(OperationContext* opCtx, const std::string& ns, bool doVersion)
     : _opCtx(opCtx), _db(DatabaseHolder::get(opCtx)->getDb(opCtx, ns)) {
     if (!_db) {
diff --git a/src/mongo/db/db_raii.h b/src/mongo/db/db_raii.h
index 346b4b83f99..30f8d235fb1 100644
--- a/src/mongo/db/db_raii.h
+++ b/src/mongo/db/db_raii.h
@@ -123,6 +123,19 @@ public:
     }
 
 protected:
+    // So that AutoGetCollectionForReadLockFree & AutoGetCollectionForReadCommandBase can access a
+    // non-const CollectionPtr from this class. AutoGetCollectionForRead inherits the access.
+    friend class AutoGetCollectionForReadLockFree;
+    template <typename AutoGetCollectionForReadType>
+    friend class AutoGetCollectionForReadCommandBase;
+
+    /**
+     * Allow access to the CollectionPtr as non-const, for friend classes.
+     */
+    CollectionPtr& _getCollectionPtrForModify() {
+        return _autoColl->_getCollectionPtrForModify();
+    }
+
     // If this field is set, the reader will not take the ParallelBatchWriterMode lock and conflict
     // with secondary batch application. This stays in scope with the _autoColl so that locks are
     // taken and released in the right order.
@@ -219,6 +232,18 @@ public:
     }
 
 private:
+    // So that AutoGetCollectionForReadCommandBase can access a non-const CollectionPtr from this
+    // class.
+    template <typename AutoGetCollectionForReadType>
+    friend class AutoGetCollectionForReadCommandBase;
+
+    /**
+     * Allow access to the CollectionPtr as non-const, for friend classes.
+     */
+    CollectionPtr& _getCollectionPtrForModify() {
+        return _autoGetCollectionForReadBase->_getCollectionPtrForModify();
+    }
+
     /**
      * Helper for how AutoGetCollectionForReadBase instantiates its owned AutoGetCollectionLockFree.
      */
@@ -318,6 +343,13 @@ public:
     }
 
 protected:
+    /**
+     * Allow access to the CollectionPtr as non-const, for friend classes of inheriting classes.
+     */
+    CollectionPtr& _getCollectionPtrForModify() {
+        return _autoCollForRead._getCollectionPtrForModify();
+    }
+
     AutoGetCollectionForReadCommandBase(
         OperationContext* opCtx,
         const NamespaceStringOrUUID& nsOrUUID,
@@ -362,6 +394,83 @@ public:
         Date_t deadline = Date_t::max(),
         AutoStatsTracker::LogMode logMode = AutoStatsTracker::LogMode::kUpdateTopAndCurOp)
         : AutoGetCollectionForReadCommandBase(opCtx, nsOrUUID, viewMode, deadline, logMode) {}
+
+private:
+    // So that AutoGetCollectionMultiForReadCommandLockFree can access a non-const CollectionPtr
+    // from this class.
+    friend class AutoGetCollectionMultiForReadCommandLockFree;
+};
+
+/**
+ * Creates an AutoGetCollectionForReadCommandLockFree instance for the 'primary' collection and
+ * verifies 'secondary' collections are safe to use for the read. Secondary collections are checked
+ * for existence and that their minimum visible timestamp supports read concern, throwing a
+ * NamespaceNotFound or SnapshotUnavailable on error. Secondaries must be unsharded: this is an
+ * invariant.
+ *
+ * Acquires only the Global IS lock, and sets up consistent in-memory and on-disk state.
+ */
+class AutoGetCollectionMultiForReadCommandLockFree {
+public:
+    /**
+     * 'primaryNssOrUUID' specifies the collection that can be sharded and on which sharding checks
+     * will be made. 'secondaryNsOrUUIDs' specifies the additional collections that are not expected
+     * to be sharded and on which 'not sharded' checks will be made.
+     */
+    AutoGetCollectionMultiForReadCommandLockFree(
+        OperationContext* opCtx,
+        const NamespaceStringOrUUID& primaryNssOrUUID,
+        std::vector<NamespaceStringOrUUID>& secondaryNsOrUUIDs,
+        AutoGetCollectionViewMode viewMode = AutoGetCollectionViewMode::kViewsForbidden,
+        Date_t deadline = Date_t::max(),
+        AutoStatsTracker::LogMode logMode = AutoStatsTracker::LogMode::kUpdateTopAndCurOp);
+
+    explicit operator bool() const {
+        return static_cast<bool>(getCollection());
+    }
+
+    const Collection* operator->() const {
+        return getCollection().get();
+    }
+
+    const CollectionPtr& operator*() const {
+        return getCollection();
+    }
+
+    const CollectionPtr& getCollection() const {
+        return _autoCollForReadCommandLockFree.getCollection();
+    }
+
+    const ViewDefinition* getView() const {
+        return _autoCollForReadCommandLockFree.getView();
+    }
+
+    /**
+     * Returns the namespace for the 'primary' collection.
+     */
+    const NamespaceString& getNss() const {
+        return _autoCollForReadCommandLockFree.getNss();
+    }
+
+private:
+    /**
+     * Runs on restore from a query yield. The 'primary' CollectionPtr's restore function is
+     * extended to run this logic.
+     *
+     * Fetches the collections identified by '_secondaryCollectionUUIDs' and verifies they're still
+     * safe to use.
+     */
+    void _secondaryCollectionsRestoreFn(OperationContext* opCtx);
+
+    // Runs all the usual check on the 'primary' collection.
+    AutoGetCollectionForReadCommandLockFree _autoCollForReadCommandLockFree;
+
+    // Save a reference to the 'primary' CollectionPtr's original restore lambda so that the new one
+    // can reference it.
+    std::function<const Collection*(OperationContext*, CollectionUUID)> _primaryCollectionRestoreFn;
+
+    // Used on query yield restore to identify the 'secondary' collections to check.
+    std::vector<CollectionUUID> _secondaryCollectionUUIDs;
 };
 
 /**
diff --git a/src/mongo/db/db_raii_multi_collection_test.cpp b/src/mongo/db/db_raii_multi_collection_test.cpp
new file mode 100644
index 00000000000..8ed19af5a71
--- /dev/null
+++ b/src/mongo/db/db_raii_multi_collection_test.cpp
@@ -0,0 +1,218 @@
+/**
+ *    Copyright (C) 2021-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    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
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kTest
+
+#include "mongo/platform/basic.h"
+
+#include <string>
+
+#include "mongo/db/catalog/catalog_test_fixture.h"
+#include "mongo/db/client.h"
+#include "mongo/db/concurrency/lock_state.h"
+#include "mongo/db/db_raii.h"
+#include "mongo/logv2/log.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+namespace {
+
+class AutoGetCollectionMultiTest : public CatalogTestFixture {
+public:
+    AutoGetCollectionMultiTest() : CatalogTestFixture("wiredTiger") {}
+
+    typedef std::pair<ServiceContext::UniqueClient, ServiceContext::UniqueOperationContext>
+        ClientAndCtx;
+
+    ClientAndCtx makeClientWithLocker(const std::string& clientName) {
+        auto client = getServiceContext()->makeClient(clientName);
+        auto opCtx = client->makeOperationContext();
+        client->swapLockState(std::make_unique<LockerImpl>());
+        return std::make_pair(std::move(client), std::move(opCtx));
+    }
+
+    void createCollections(OperationContext* opCtx) {
+        CollectionOptions defaultCollectionOptions;
+        ASSERT_OK(
+            storageInterface()->createCollection(opCtx, _primaryNss, defaultCollectionOptions));
+        ASSERT_OK(
+            storageInterface()->createCollection(opCtx, _secondaryNss1, defaultCollectionOptions));
+        ASSERT_OK(
+            storageInterface()->createCollection(opCtx, _secondaryNss2, defaultCollectionOptions));
+        ASSERT_OK(
+            storageInterface()->createCollection(opCtx, _otherDBNss, defaultCollectionOptions));
+    }
+
+    const NamespaceString _primaryNss = NamespaceString("primary1.db1");
+    const NamespaceString _secondaryNss1 = NamespaceString("secondary1.db1");
+    const NamespaceString _secondaryNss2 = NamespaceString("secondary2.db1");
+    const NamespaceString _otherDBNss = NamespaceString("secondary2.db2");
+
+    std::vector<NamespaceStringOrUUID> _secondaryNssVec = {NamespaceStringOrUUID(_secondaryNss1),
+                                                           NamespaceStringOrUUID(_secondaryNss2)};
+    std::vector<NamespaceStringOrUUID> _otherDBNssVec = {NamespaceStringOrUUID(_otherDBNss)};
+
+    const ClientAndCtx _client1 = makeClientWithLocker("client1");
+    const ClientAndCtx _client2 = makeClientWithLocker("client2");
+};
+
+TEST_F(AutoGetCollectionMultiTest, SingleDB) {
+    auto opCtx1 = _client1.second.get();
+
+    createCollections(opCtx1);
+
+    AutoGetCollectionMultiForReadCommandLockFree autoGet(
+        opCtx1, NamespaceStringOrUUID(_primaryNss), _secondaryNssVec);
+
+    auto locker = opCtx1->lockState();
+    locker->dump();
+    invariant(locker->isLockHeldForMode(resourceIdGlobal, MODE_IS));
+    invariant(!locker->isDbLockedForMode(_primaryNss.db(), MODE_IS));
+    // Set 'shouldConflictWithSecondaryBatchApplication' to true so isCollectionLockedForMode()
+    // doesn't return true regardless of what locks are held.
+    opCtx1->lockState()->setShouldConflictWithSecondaryBatchApplication(true);
+    invariant(!locker->isCollectionLockedForMode(_primaryNss, MODE_IS));
+
+    const auto& coll = autoGet.getCollection();
+    ASSERT(coll);
+    ASSERT(CollectionCatalog::get(opCtx1)->lookupCollectionByNamespace(opCtx1, _secondaryNss1));
+    ASSERT(CollectionCatalog::get(opCtx1)->lookupCollectionByNamespace(opCtx1, _secondaryNss2));
+
+    coll.yield();
+    coll.restore();
+    ASSERT(coll);
+    ASSERT(CollectionCatalog::get(opCtx1)->lookupCollectionByNamespace(opCtx1, _secondaryNss1));
+    ASSERT(CollectionCatalog::get(opCtx1)->lookupCollectionByNamespace(opCtx1, _secondaryNss2));
+}
+
+TEST_F(AutoGetCollectionMultiTest, MultiDBs) {
+    auto opCtx1 = _client1.second.get();
+
+    createCollections(opCtx1);
+
+    AutoGetCollectionMultiForReadCommandLockFree autoGet(
+        opCtx1, NamespaceStringOrUUID(_primaryNss), _otherDBNssVec);
+
+    auto locker = opCtx1->lockState();
+    locker->dump();
+    invariant(locker->isLockHeldForMode(resourceIdGlobal, MODE_IS));
+    invariant(!locker->isDbLockedForMode(_primaryNss.db(), MODE_IS));
+    invariant(!locker->isDbLockedForMode(_otherDBNss.db(), MODE_IS));
+    // Set 'shouldConflictWithSecondaryBatchApplication' to true so isCollectionLockedForMode()
+    // doesn't return true regardless of what locks are held.
+    opCtx1->lockState()->setShouldConflictWithSecondaryBatchApplication(true);
+    invariant(!locker->isCollectionLockedForMode(_primaryNss, MODE_IS));
+
+    const auto& coll = autoGet.getCollection();
+    ASSERT(coll);
+    ASSERT(CollectionCatalog::get(opCtx1)->lookupCollectionByNamespace(opCtx1, _otherDBNss));
+
+    coll.yield();
+    coll.restore();
+    ASSERT(coll);
+    ASSERT(CollectionCatalog::get(opCtx1)->lookupCollectionByNamespace(opCtx1, _otherDBNss));
+}
+
+TEST_F(AutoGetCollectionMultiTest, DropCollection) {
+    auto opCtx1 = _client1.second.get();
+
+    createCollections(opCtx1);
+
+    AutoGetCollectionMultiForReadCommandLockFree autoGet(
+        opCtx1, NamespaceStringOrUUID(_primaryNss), _secondaryNssVec);
+
+    auto locker = opCtx1->lockState();
+    locker->dump();
+    invariant(locker->isLockHeldForMode(resourceIdGlobal, MODE_IS));
+    invariant(!locker->isDbLockedForMode(_primaryNss.db(), MODE_IS));
+    // Set 'shouldConflictWithSecondaryBatchApplication' to true so isCollectionLockedForMode()
+    // doesn't return true regardless of what locks are held.
+    opCtx1->lockState()->setShouldConflictWithSecondaryBatchApplication(true);
+    invariant(!locker->isCollectionLockedForMode(_primaryNss, MODE_IS));
+
+    const auto& coll = autoGet.getCollection();
+
+    // Drop a secondary collection via a different opCtx, so that we can test that yield restore
+    // throws on failing to verify the secondary collection.
+    {
+        auto opCtx2 = _client2.second.get();
+        AutoGetCollection secondClientAutoGet(opCtx2, _secondaryNss1, MODE_X);
+
+        // Disable replication so that it is not necessary to set up the infrastructure to timestamp
+        // catalog writes properly.
+        repl::UnreplicatedWritesBlock uwb(opCtx2);
+
+        ASSERT_OK(storageInterface()->dropCollection(opCtx2, _secondaryNss1));
+    }
+
+    coll.yield();
+    ASSERT_THROWS_CODE(coll.restore(), AssertionException, ErrorCodes::NamespaceNotFound);
+}
+
+TEST_F(AutoGetCollectionMultiTest, DropAndRecreateCollection) {
+    auto opCtx1 = _client1.second.get();
+
+    createCollections(opCtx1);
+
+    AutoGetCollectionMultiForReadCommandLockFree autoGet(
+        opCtx1, NamespaceStringOrUUID(_primaryNss), _secondaryNssVec);
+
+    auto locker = opCtx1->lockState();
+    locker->dump();
+    invariant(locker->isLockHeldForMode(resourceIdGlobal, MODE_IS));
+    invariant(!locker->isDbLockedForMode(_primaryNss.db(), MODE_IS));
+    // Set 'shouldConflictWithSecondaryBatchApplication' to true so isCollectionLockedForMode()
+    // doesn't return true regardless of what locks are held.
+    opCtx1->lockState()->setShouldConflictWithSecondaryBatchApplication(true);
+    invariant(!locker->isCollectionLockedForMode(_primaryNss, MODE_IS));
+
+    const auto& coll = autoGet.getCollection();
+
+    // Drop and recreate a secondary collection via a different opCtx, so that we can test that
+    // yield restore throws on failing to verify the secondary collection.
+    {
+        auto opCtx2 = _client2.second.get();
+        AutoGetCollection secondClientAutoGet(opCtx2, _secondaryNss1, MODE_X);
+
+        // Disable replication so that it is not necessary to set up the infrastructure to timestamp
+        // catalog writes properly.
+        repl::UnreplicatedWritesBlock uwb(opCtx2);
+
+        ASSERT_OK(storageInterface()->dropCollection(_client2.second.get(), _secondaryNss1));
+        CollectionOptions defaultCollectionOptions;
+        ASSERT_OK(
+            storageInterface()->createCollection(opCtx2, _secondaryNss1, defaultCollectionOptions));
+    }
+
+    coll.yield();
+    ASSERT_THROWS_CODE(coll.restore(), AssertionException, ErrorCodes::NamespaceNotFound);
+}
+
+}  // namespace
+}  // namespace mongo