SERVER-62194 Make AutoGetCollection accept multiple namespaces to lock.

7 files changed, 524 insertions, 84 deletions

diff --git a/src/mongo/db/catalog_raii.cpp b/src/mongo/db/catalog_raii.cpp
index 155cdabc8b0..576dc81a1ec 100644
--- a/src/mongo/db/catalog_raii.cpp
+++ b/src/mongo/db/catalog_raii.cpp
@@ -46,19 +46,161 @@ namespace {
 
 MONGO_FAIL_POINT_DEFINE(setAutoGetCollectionWait);
 
+/**
+ * Returns true if 'nss' is a view. False if the namespace or view doesn't exist.
+ */
+bool isSecondaryNssAView(OperationContext* opCtx, const NamespaceString& nss) {
+    auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, nss.db());
+    return viewCatalog && viewCatalog->lookup(opCtx, nss);
+}
+
+/**
+ * Performs some sanity checks on the collection and database.
+ */
+void verifyDbAndCollection(OperationContext* opCtx,
+                           LockMode modeColl,
+                           const NamespaceStringOrUUID& nsOrUUID,
+                           const NamespaceString& resolvedNss,
+                           CollectionPtr& coll,
+                           Database* db) {
+    invariant(!nsOrUUID.uuid() || coll,
+              str::stream() << "Collection for " << resolvedNss.ns()
+                            << " disappeared after successfully resolving " << nsOrUUID.toString());
+
+    invariant(!nsOrUUID.uuid() || db,
+              str::stream() << "Database for " << resolvedNss.ns()
+                            << " disappeared after successfully resolving " << nsOrUUID.toString());
+
+    // In most cases we expect modifications for system.views to upgrade MODE_IX to MODE_X before
+    // taking the lock. One exception is a query by UUID of system.views in a transaction. Usual
+    // queries of system.views (by name, not UUID) within a transaction are rejected. However, if
+    // the query is by UUID we can't determine whether the namespace is actually system.views until
+    // we take the lock here. So we have this one last assertion.
+    uassert(51070,
+            "Modifications to system.views must take an exclusive lock",
+            !resolvedNss.isSystemDotViews() || modeColl != MODE_IX);
+
+    if (!db || !coll) {
+        return;
+    }
+
+    // If we are in a transaction, we cannot yield and wait when there are pending catalog changes.
+    // Instead, we must return an error in such situations. We ignore this restriction for the
+    // oplog, since it never has pending catalog changes.
+    if (opCtx->inMultiDocumentTransaction() && resolvedNss != NamespaceString::kRsOplogNamespace) {
+        if (auto minSnapshot = coll->getMinimumVisibleSnapshot()) {
+            auto mySnapshot =
+                opCtx->recoveryUnit()->getPointInTimeReadTimestamp(opCtx).get_value_or(
+                    opCtx->recoveryUnit()->getCatalogConflictingTimestamp());
+
+            uassert(
+                ErrorCodes::SnapshotUnavailable,
+                str::stream() << "Unable to read from a snapshot due to pending collection catalog "
+                                 "changes; please retry the operation. Snapshot timestamp is "
+                              << mySnapshot.toString() << ". Collection minimum is "
+                              << minSnapshot->toString(),
+                mySnapshot.isNull() || mySnapshot >= minSnapshot.get());
+        }
+    }
+}
+
+/**
+ * Defines sorting order for NamespaceStrings based on what their ResourceId would be for locking.
+ */
+struct ResourceIdNssComparator {
+    bool operator()(const NamespaceString& lhs, const NamespaceString& rhs) const {
+        return ResourceId(RESOURCE_COLLECTION, lhs.ns()) <
+            ResourceId(RESOURCE_COLLECTION, rhs.ns());
+    }
+};
+
+/**
+ * Fills the input 'collLocks' with CollectionLocks, acquiring locks on namespaces 'nsOrUUID' and
+ * 'secondaryNssOrUUIDs' in ResourceId(RESOURCE_COLLECTION, nss.ns()) order.
+ *
+ * The namespaces will be resolved, the locks acquired, and then the namespaces will be checked for
+ * changes in case there is a race with rename and a UUID no longer matches the locked namespace.
+ *
+ * Handles duplicate namespaces across 'nsOrUUID' and 'secondaryNssOrUUIDs'. Only one lock will be
+ * taken on each namespace.
+ */
+void acquireCollectionLocksInResourceIdOrder(
+    OperationContext* opCtx,
+    const NamespaceStringOrUUID& nsOrUUID,
+    LockMode modeColl,
+    Date_t deadline,
+    const std::vector<NamespaceStringOrUUID>& secondaryNssOrUUIDs,
+    std::vector<Lock::CollectionLock>* collLocks) {
+    invariant(collLocks->empty());
+    auto catalog = CollectionCatalog::get(opCtx);
+
+    // Use a set so that we can easily dedupe namespaces to avoid locking the same collection twice.
+    std::set<NamespaceString, ResourceIdNssComparator> temp;
+    std::set<NamespaceString, ResourceIdNssComparator> verifyTemp;
+    do {
+        // Clear the data structures when/if we loop more than once.
+        collLocks->clear();
+        temp.clear();
+        verifyTemp.clear();
+
+        // Create a single set with all the resolved namespaces sorted by ascending
+        // ResourceId(RESOURCE_COLLECTION, nss.ns()).
+        temp.insert(catalog->resolveNamespaceStringOrUUID(opCtx, nsOrUUID));
+        for (const auto& secondaryNssOrUUID : secondaryNssOrUUIDs) {
+            temp.insert(catalog->resolveNamespaceStringOrUUID(opCtx, secondaryNssOrUUID));
+        }
+
+        // Acquire all of the locks in order.
+        for (auto& nss : temp) {
+            collLocks->emplace_back(opCtx, nss, modeColl, deadline);
+        }
+
+        // Check that the namespaces have NOT changed after acquiring locks. It's possible to race
+        // with a rename collection when the given NamespaceStringOrUUID is a UUID, and consequently
+        // fail to lock the correct namespace.
+        verifyTemp.insert(catalog->resolveNamespaceStringOrUUID(opCtx, nsOrUUID));
+        for (const auto& secondaryNssOrUUID : secondaryNssOrUUIDs) {
+            verifyTemp.insert(catalog->resolveNamespaceStringOrUUID(opCtx, secondaryNssOrUUID));
+        }
+    } while (temp != verifyTemp);
+}
+
 }  // namespace
 
-AutoGetDb::AutoGetDb(OperationContext* opCtx, StringData dbName, LockMode mode, Date_t deadline)
+AutoGetDb::AutoGetDb(OperationContext* opCtx,
+                     StringData dbName,
+                     LockMode mode,
+                     Date_t deadline,
+                     const std::set<StringData>& secondaryDbNames)
     : _dbName(dbName), _dbLock(opCtx, dbName, mode, deadline), _db([&] {
           auto databaseHolder = DatabaseHolder::get(opCtx);
           return databaseHolder->getDb(opCtx, dbName);
       }()) {
+    // Locking multiple databases is only supported in intent read mode (MODE_IS).
+    invariant(secondaryDbNames.empty() || mode == MODE_IS);
+
+    // Take the secondary dbs' database locks only: no global or RSTL, as they are already acquired
+    // above. Note: no consistent ordering is when acquiring database locks because there are no
+    // occasions where multiple strong locks are acquired to make ordering matter (deadlock
+    // avoidance).
+    for (const auto& secondaryDbName : secondaryDbNames) {
+        // The primary database may be repeated in the secondary databases and the primary database
+        // should not be locked twice.
+        if (secondaryDbName != _dbName) {
+            _secondaryDbLocks.emplace_back(
+                opCtx, secondaryDbName, MODE_IS, deadline, true /*skipGlobalAndRSTLLocks*/);
+        }
+    }
+
+    // The 'primary' database must be version checked for sharding.
     auto dss = DatabaseShardingState::get(opCtx, dbName);
     auto dssLock = DatabaseShardingState::DSSLock::lockShared(opCtx, dss);
     dss->checkDbVersion(opCtx, dssLock);
 }
 
 Database* AutoGetDb::ensureDbExists(OperationContext* opCtx) {
+    invariant(_secondaryDbLocks.empty());
+
     if (_db) {
         return _db;
     }
@@ -73,24 +215,30 @@ Database* AutoGetDb::ensureDbExists(OperationContext* opCtx) {
     return _db;
 }
 
-AutoGetCollection::AutoGetCollection(OperationContext* opCtx,
-                                     const NamespaceStringOrUUID& nsOrUUID,
-                                     LockMode modeColl,
-                                     AutoGetCollectionViewMode viewMode,
-                                     Date_t deadline)
-    : _autoDb(opCtx,
-              !nsOrUUID.dbname().empty() ? nsOrUUID.dbname() : nsOrUUID.nss()->db(),
-              isSharedLockMode(modeColl) ? MODE_IS : MODE_IX,
-              deadline) {
+AutoGetCollection::AutoGetCollection(
+    OperationContext* opCtx,
+    const NamespaceStringOrUUID& nsOrUUID,
+    LockMode modeColl,
+    AutoGetCollectionViewMode viewMode,
+    Date_t deadline,
+    const std::vector<NamespaceStringOrUUID>& secondaryNssOrUUIDs) {
     invariant(!opCtx->isLockFreeReadsOp());
+    invariant(secondaryNssOrUUIDs.empty() || modeColl == MODE_IS);
 
-    auto& nss = nsOrUUID.nss();
-    if (nss) {
-        uassert(ErrorCodes::InvalidNamespace,
-                str::stream() << "Namespace " << *nss << " is not a valid collection name",
-                nss->isValid());
+    // Get a unique list of 'secondary' database names to pass into AutoGetDbMulti below.
+    std::set<StringData> secondaryDbNames;
+    for (auto& secondaryNssOrUUID : secondaryNssOrUUIDs) {
+        secondaryDbNames.emplace(secondaryNssOrUUID.db());
     }
 
+    // Acquire the global/RSTL and all the database locks (may or may not be multiple
+    // databases).
+    _autoDb.emplace(opCtx,
+                    !nsOrUUID.dbname().empty() ? nsOrUUID.dbname() : nsOrUUID.nss()->db(),
+                    isSharedLockMode(modeColl) ? MODE_IS : MODE_IX,
+                    deadline,
+                    secondaryDbNames);
+
     // Out of an abundance of caution, force operations to acquire new snapshots after
     // acquiring exclusive collection locks. Operations that hold MODE_X locks make an
     // assumption that all writes are visible in their snapshot and no new writes will commit.
@@ -101,85 +249,78 @@ AutoGetCollection::AutoGetCollection(OperationContext* opCtx,
                   str::stream() << "Snapshot opened before acquiring X lock for " << nsOrUUID);
     }
 
-    _collLock.emplace(opCtx, nsOrUUID, modeColl, deadline);
-    auto catalog = CollectionCatalog::get(opCtx);
-    _resolvedNss = catalog->resolveNamespaceStringOrUUID(opCtx, nsOrUUID);
+    // Acquire the collection locks. If there's only one lock, then it can simply be taken. If
+    // there are many, however, the locks must be taken in _ascending_ ResourceId order to avoid
+    // deadlocks across threads.
+    if (secondaryDbNames.empty()) {
+        uassertStatusOK(nsOrUUID.isNssValid());
+        _collLocks.emplace_back(opCtx, nsOrUUID, modeColl, deadline);
+    } else {
+        acquireCollectionLocksInResourceIdOrder(
+            opCtx, nsOrUUID, modeColl, deadline, secondaryNssOrUUIDs, &_collLocks);
+    }
 
     // Wait for a configured amount of time after acquiring locks if the failpoint is enabled
     setAutoGetCollectionWait.execute(
         [&](const BSONObj& data) { sleepFor(Milliseconds(data["waitForMillis"].numberInt())); });
 
-    Database* const db = _autoDb.getDb();
-    invariant(!nsOrUUID.uuid() || db,
-              str::stream() << "Database for " << _resolvedNss.ns()
-                            << " disappeared after successfully resolving " << nsOrUUID.toString());
-
-    // In most cases we expect modifications for system.views to upgrade MODE_IX to MODE_X before
-    // taking the lock. One exception is a query by UUID of system.views in a transaction. Usual
-    // queries of system.views (by name, not UUID) within a transaction are rejected. However, if
-    // the query is by UUID we can't determine whether the namespace is actually system.views until
-    // we take the lock here. So we have this one last assertion.
-    uassert(51070,
-            "Modifications to system.views must take an exclusive lock",
-            !_resolvedNss.isSystemDotViews() || modeColl != MODE_IX);
-
-    // If the database doesn't exists, we can't obtain a collection or check for views
-    if (!db)
-        return;
+    auto catalog = CollectionCatalog::get(opCtx);
+    auto databaseHolder = DatabaseHolder::get(opCtx);
 
+    // Check that the collections are all safe to use.
+    _resolvedNss = catalog->resolveNamespaceStringOrUUID(opCtx, nsOrUUID);
     _coll = catalog->lookupCollectionByNamespace(opCtx, _resolvedNss);
-    invariant(!nsOrUUID.uuid() || _coll,
-              str::stream() << "Collection for " << _resolvedNss.ns()
-                            << " disappeared after successfully resolving " << nsOrUUID.toString());
-
-    if (_coll) {
-        // If we are in a transaction, we cannot yield and wait when there are pending catalog
-        // changes. Instead, we must return an error in such situations. We
-        // ignore this restriction for the oplog, since it never has pending catalog changes.
-        if (opCtx->inMultiDocumentTransaction() &&
-            _resolvedNss != NamespaceString::kRsOplogNamespace) {
-
-            if (auto minSnapshot = _coll->getMinimumVisibleSnapshot()) {
-                auto mySnapshot =
-                    opCtx->recoveryUnit()->getPointInTimeReadTimestamp(opCtx).get_value_or(
-                        opCtx->recoveryUnit()->getCatalogConflictingTimestamp());
-
-                uassert(ErrorCodes::SnapshotUnavailable,
-                        str::stream()
-                            << "Unable to read from a snapshot due to pending collection catalog "
-                               "changes; please retry the operation. Snapshot timestamp is "
-                            << mySnapshot.toString() << ". Collection minimum is "
-                            << minSnapshot->toString(),
-                        mySnapshot.isNull() || mySnapshot >= minSnapshot.get());
-            }
+    verifyDbAndCollection(opCtx, modeColl, nsOrUUID, _resolvedNss, _coll, _autoDb->getDb());
+    for (auto& secondaryNssOrUUID : secondaryNssOrUUIDs) {
+        auto secondaryResolvedNss =
+            catalog->resolveNamespaceStringOrUUID(opCtx, secondaryNssOrUUID);
+        auto secondaryColl = catalog->lookupCollectionByNamespace(opCtx, secondaryResolvedNss);
+        verifyDbAndCollection(opCtx,
+                              MODE_IS,
+                              secondaryNssOrUUID,
+                              secondaryResolvedNss,
+                              secondaryColl,
+                              databaseHolder->getDb(opCtx, secondaryNssOrUUID.db()));
+
+        // Flag if a secondary namespace is a view.
+        if (!_secondaryNssIsView && isSecondaryNssAView(opCtx, secondaryResolvedNss)) {
+            _secondaryNssIsView = true;
         }
+    }
 
+    if (_coll) {
         // Fetch and store the sharding collection description data needed for use during the
         // operation. The shardVersion will be checked later if the shard filtering metadata is
         // fetched, ensuring both that the collection description info used here and the routing
         // table are consistent with the read request's shardVersion.
+        //
+        // Note: sharding versioning for an operation has no concept of multiple collections.
         auto collDesc =
-            CollectionShardingState::get(opCtx, getNss())->getCollectionDescription(opCtx);
+            CollectionShardingState::get(opCtx, _resolvedNss)->getCollectionDescription(opCtx);
         if (collDesc.isSharded()) {
             _coll.setShardKeyPattern(collDesc.getKeyPattern());
         }
 
-        // If the collection exists, there is no need to check for views.
         return;
     }
 
-    _view = ViewCatalog::get(db)->lookup(opCtx, _resolvedNss);
-    uassert(ErrorCodes::CommandNotSupportedOnView,
-            str::stream() << "Namespace " << _resolvedNss.ns() << " is a timeseries collection",
-            !_view || viewMode == AutoGetCollectionViewMode::kViewsPermitted ||
-                !_view->timeseries());
-    uassert(ErrorCodes::CommandNotSupportedOnView,
-            str::stream() << "Namespace " << _resolvedNss.ns() << " is a view, not a collection",
-            !_view || viewMode == AutoGetCollectionViewMode::kViewsPermitted);
+    if (_autoDb->getDb()) {
+        _view = ViewCatalog::get(_autoDb->getDb())->lookup(opCtx, _resolvedNss);
+        uassert(ErrorCodes::CommandNotSupportedOnView,
+                str::stream() << "Namespace " << _resolvedNss.ns() << " is a timeseries collection",
+                !_view || viewMode == AutoGetCollectionViewMode::kViewsPermitted ||
+                    !_view->timeseries());
+        uassert(ErrorCodes::CommandNotSupportedOnView,
+                str::stream() << "Namespace " << _resolvedNss.ns()
+                              << " is a view, not a collection",
+                !_view || viewMode == AutoGetCollectionViewMode::kViewsPermitted);
+    }
 }
 
 Collection* AutoGetCollection::getWritableCollection(OperationContext* opCtx,
                                                      CollectionCatalog::LifetimeMode mode) {
+    invariant(_collLocks.size() == 1);
+
     // Acquire writable instance if not already available
     if (!_writableColl) {
 
diff --git a/src/mongo/db/catalog_raii.h b/src/mongo/db/catalog_raii.h
index b62d791c1ce..b1740758d05 100644
--- a/src/mongo/db/catalog_raii.h
+++ b/src/mongo/db/catalog_raii.h
@@ -56,10 +56,16 @@ class AutoGetDb {
     AutoGetDb& operator=(const AutoGetDb&) = delete;
 
 public:
+    /**
+     * Database locks are also acquired for any 'secondaryDbNames' database names provided. Only
+     * MODE_IS is supported when 'secondaryDbNames' are provided. It is safe to repeat 'dbName' in
+     * 'secondaryDbNames'.
+     */
     AutoGetDb(OperationContext* opCtx,
               StringData dbName,
               LockMode mode,
-              Date_t deadline = Date_t::max());
+              Date_t deadline = Date_t::max(),
+              const std::set<StringData>& secondaryDbNames = {});
 
     AutoGetDb(AutoGetDb&&) = default;
 
@@ -78,8 +84,16 @@ public:
 private:
     std::string _dbName;
 
+    // Special note! The primary DBLock must destruct last (be declared first) so that the global
+    // and RSTL locks are not released until all the secondary DBLocks (without global and RSTL)
+    // have destructed.
     Lock::DBLock _dbLock;
+
     Database* _db;
+
+    // The secondary DBLocks will be acquired without the global or RSTL locks taken, re: the
+    // skipGlobalAndRSTLLocks flag in the DBLock constructor.
+    std::vector<Lock::DBLock> _secondaryDbLocks;
 };
 
 enum class AutoGetCollectionViewMode { kViewsPermitted, kViewsForbidden };
@@ -105,12 +119,23 @@ class AutoGetCollection {
     AutoGetCollection& operator=(const AutoGetCollection&) = delete;
 
 public:
+    /**
+     * Collection locks are also acquired for any 'secondaryNssOrUUIDs' namespaces provided.
+     * Collection locks are acquired in ascending ResourceId(RESOURCE_COLLECTION, nss) order to
+     * avoid deadlocks, consistent with other locations in the code wherein we take multiple
+     * collection locks.
+     *
+     * Invariants if any 'secondaryNssOrUUIDs' represent a view namespace. Only MODE_IS is supported
+     * when 'secondaryNssOrUUIDs' namespaces are provided. It is safe for 'nsOrUUID' to be
+     * duplicated in 'secondaryNssOrUUIDs', or 'secondaryNssOrUUIDs' to contain duplicates.
+     */
     AutoGetCollection(
         OperationContext* opCtx,
         const NamespaceStringOrUUID& nsOrUUID,
         LockMode modeColl,
         AutoGetCollectionViewMode viewMode = AutoGetCollectionViewMode::kViewsForbidden,
-        Date_t deadline = Date_t::max());
+        Date_t deadline = Date_t::max(),
+        const std::vector<NamespaceStringOrUUID>& secondaryNssOrUUIDs = {});
 
     explicit operator bool() const {
         return static_cast<bool>(getCollection());
@@ -131,14 +156,14 @@ public:
      * Returns the database, or nullptr if it didn't exist.
      */
     Database* getDb() const {
-        return _autoDb.getDb();
+        return _autoDb->getDb();
     }
 
     /**
      * Returns the database, creating it if it does not exist.
      */
     Database* ensureDbExists(OperationContext* opCtx) {
-        return _autoDb.ensureDbExists(opCtx);
+        return _autoDb->ensureDbExists(opCtx);
     }
 
     /**
@@ -165,6 +190,13 @@ public:
     }
 
     /**
+     * Indicates whether any of the 'secondaryNssOrUUIDs' namespaces are views.
+     */
+    bool isAnySecondaryNamespaceAView() const {
+        return _secondaryNssIsView;
+    }
+
+    /**
      * Returns a writable Collection copy that will be returned by current and future calls to this
      * function as well as getCollection(). Any previous Collection pointers that were returned may
      * be invalidated.
@@ -189,11 +221,17 @@ protected:
         return _coll;
     }
 
-    AutoGetDb _autoDb;
-    boost::optional<Lock::CollectionLock> _collLock;
+    // Ordering matters, the _collLocks should destruct before the _autoGetDb releases the
+    // rstl/global/database locks.
+    boost::optional<AutoGetDb> _autoDb;
+    std::vector<Lock::CollectionLock> _collLocks;
+
     CollectionPtr _coll = nullptr;
     std::shared_ptr<const ViewDefinition> _view;
 
+    // Tracks whether any secondary collection namespaces is a view.
+    bool _secondaryNssIsView = false;
+
     // If the object was instantiated with a UUID, contains the resolved namespace, otherwise it is
     // the same as the input namespace string
     NamespaceString _resolvedNss;
diff --git a/src/mongo/db/catalog_raii_test.cpp b/src/mongo/db/catalog_raii_test.cpp
index 604f590c045..e4a9b6f34bd 100644
--- a/src/mongo/db/catalog_raii_test.cpp
+++ b/src/mongo/db/catalog_raii_test.cpp
@@ -42,6 +42,7 @@
 #include "mongo/db/service_context_test_fixture.h"
 #include "mongo/db/storage/recovery_unit_noop.h"
 #include "mongo/logv2/log.h"
+#include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/time_support.h"
 
@@ -61,6 +62,10 @@ public:
     }
 
     const NamespaceString nss = NamespaceString("test", "coll");
+    const NamespaceString kSecondaryNss1 = NamespaceString("test", "secondaryColl1");
+    const NamespaceString kSecondaryNss2 = NamespaceString("test", "secondaryColl2");
+    const NamespaceString kSecondaryNssOtherDb1 = NamespaceString("test2", "secondaryColl1");
+    const NamespaceString kSecondaryNssOtherDb2 = NamespaceString("test2", "secondaryColl2");
     const Milliseconds timeoutMs = Seconds(1);
     const ClientAndCtx client1 = makeClientWithLocker("client1");
     const ClientAndCtx client2 = makeClientWithLocker("client2");
@@ -118,6 +123,44 @@ TEST_F(CatalogRAIITestFixture, AutoGetDBDeadlineMin) {
                          Milliseconds(0));
 }
 
+TEST_F(CatalogRAIITestFixture, AutoGetDBMultiDBDeadline) {
+    // Take the kSecondaryNssOtherDb1 database MODE_X lock to create a conflict later.
+    boost::optional<Lock::DBLock> dbLockXLock;
+    dbLockXLock.emplace(client1.second.get(), kSecondaryNssOtherDb1.db(), MODE_X);
+    ASSERT(client1.second->lockState()->isDbLockedForMode(kSecondaryNssOtherDb1.db(), MODE_X));
+
+    // Trying to acquire the kSecondaryNssOtherDb1 database MODE_IS lock should time out.
+    std::set<StringData> secondaryDbNamesConflicting{kSecondaryNss1.db(),
+                                                     kSecondaryNss2.db(),
+                                                     kSecondaryNssOtherDb1.db(),
+                                                     kSecondaryNssOtherDb2.db()};
+    failsWithLockTimeout(
+        [&] {
+            AutoGetDb autoGetDb(client2.second.get(),
+                                nss.db(),
+                                MODE_IS,
+                                Date_t::now() + timeoutMs,
+                                secondaryDbNamesConflicting);
+        },
+        timeoutMs);
+
+    {
+        // Acquiring multiple database locks without the kSecondaryNssOtherDb1 database should work.
+        std::set<StringData> secondaryDbNamesNoConflict{kSecondaryNss1.db()};
+        AutoGetDb autoGetDbNoConflict(client2.second.get(),
+                                      kSecondaryNss1.db(),
+                                      MODE_IS,
+                                      Date_t::max(),
+                                      secondaryDbNamesNoConflict);
+    }
+
+    // Lastly, with the MODE_X lock on kSecondaryNssOtherDb1.db() released, the original multi
+    // database lock request should work.
+    dbLockXLock.reset();
+    AutoGetDb autoGetDb(
+        client2.second.get(), nss.db(), MODE_IS, Date_t::max(), secondaryDbNamesConflicting);
+}
+
 TEST_F(CatalogRAIITestFixture, AutoGetCollectionCollLockDeadline) {
     Lock::DBLock dbLock1(client1.second.get(), nss.db(), MODE_IX);
     ASSERT(client1.second->lockState()->isDbLockedForMode(nss.db(), MODE_IX));
@@ -218,6 +261,177 @@ TEST_F(CatalogRAIITestFixture, AutoGetCollectionDBLockCompatibleX) {
     AutoGetCollection coll(client2.second.get(), nss, MODE_X);
 }
 
+// Test multiple collections being locked on the same database.
+TEST_F(CatalogRAIITestFixture, AutoGetCollectionSecondaryNamespacesSingleDb) {
+    auto opCtx1 = client1.second.get();
+
+    std::vector<NamespaceStringOrUUID> secondaryNamespaces{NamespaceStringOrUUID(kSecondaryNss1),
+                                                           NamespaceStringOrUUID(kSecondaryNss2)};
+
+    boost::optional<AutoGetCollection> autoGetColl;
+    autoGetColl.emplace(opCtx1,
+                        nss,
+                        MODE_IS,
+                        AutoGetCollectionViewMode::kViewsForbidden,
+                        Date_t::max(),
+                        secondaryNamespaces);
+
+    ASSERT(opCtx1->lockState()->isRSTLLocked());
+    ASSERT(opCtx1->lockState()->isReadLocked());  // Global lock check
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(nss.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNss1.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNss2.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isCollectionLockedForMode(nss, MODE_IS));
+    ASSERT(opCtx1->lockState()->isCollectionLockedForMode(kSecondaryNss1, MODE_IS));
+    ASSERT(opCtx1->lockState()->isCollectionLockedForMode(kSecondaryNss2, MODE_IS));
+
+    ASSERT(!opCtx1->lockState()->isRSTLExclusive());
+    ASSERT(!opCtx1->lockState()->isGlobalLockedRecursively());
+    ASSERT(!opCtx1->lockState()->isWriteLocked());
+    ASSERT(!opCtx1->lockState()->isDbLockedForMode(kSecondaryNssOtherDb1.db(), MODE_IS));
+    ASSERT(!opCtx1->lockState()->isDbLockedForMode(kSecondaryNssOtherDb2.db(), MODE_IS));
+    ASSERT(!opCtx1->lockState()->isCollectionLockedForMode(kSecondaryNssOtherDb1, MODE_IS));
+    ASSERT(!opCtx1->lockState()->isCollectionLockedForMode(kSecondaryNssOtherDb2, MODE_IS));
+
+    // All the locks should release.
+    autoGetColl.reset();
+    ASSERT(!opCtx1->lockState()->isLocked());  // Global lock check
+}
+
+TEST_F(CatalogRAIITestFixture, AutoGetCollectionSecondaryNamespacesMultiDb) {
+    auto opCtx1 = client1.second.get();
+
+    std::vector<NamespaceStringOrUUID> secondaryNamespaces{
+        NamespaceStringOrUUID(kSecondaryNss1),
+        NamespaceStringOrUUID(kSecondaryNss2),
+        NamespaceStringOrUUID(kSecondaryNssOtherDb1),
+        NamespaceStringOrUUID(kSecondaryNssOtherDb2)};
+    boost::optional<AutoGetCollection> autoGetColl;
+    autoGetColl.emplace(opCtx1,
+                        nss,
+                        MODE_IS,
+                        AutoGetCollectionViewMode::kViewsForbidden,
+                        Date_t::max(),
+                        secondaryNamespaces);
+
+    ASSERT(opCtx1->lockState()->isRSTLLocked());
+    ASSERT(opCtx1->lockState()->isReadLocked());  // Global lock check
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(nss.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNss1.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNss2.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNssOtherDb1.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNssOtherDb2.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isCollectionLockedForMode(nss, MODE_IS));
+    ASSERT(opCtx1->lockState()->isCollectionLockedForMode(kSecondaryNss1, MODE_IS));
+    ASSERT(opCtx1->lockState()->isCollectionLockedForMode(kSecondaryNss2, MODE_IS));
+    ASSERT(opCtx1->lockState()->isCollectionLockedForMode(kSecondaryNssOtherDb1, MODE_IS));
+    ASSERT(opCtx1->lockState()->isCollectionLockedForMode(kSecondaryNssOtherDb2, MODE_IS));
+
+    ASSERT(!opCtx1->lockState()->isRSTLExclusive());
+    ASSERT(!opCtx1->lockState()->isGlobalLockedRecursively());
+    ASSERT(!opCtx1->lockState()->isWriteLocked());
+
+    // All the locks should release.
+    autoGetColl.reset();
+    ASSERT(!opCtx1->lockState()->isLocked());  // Global lock check.
+}
+
+TEST_F(CatalogRAIITestFixture, AutoGetDbSecondaryNamespacesSingleDb) {
+    auto opCtx1 = client1.second.get();
+
+    std::set<StringData> secondaryDbNames{kSecondaryNss1.db(), kSecondaryNss2.db()};
+    boost::optional<AutoGetDb> autoGetDb;
+    autoGetDb.emplace(opCtx1, nss.db(), MODE_IS, Date_t::max(), secondaryDbNames);
+
+    ASSERT(opCtx1->lockState()->isRSTLLocked());
+    ASSERT(opCtx1->lockState()->isReadLocked());  // Global lock check
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(nss.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNss1.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNss2.db(), MODE_IS));
+
+    ASSERT(!opCtx1->lockState()->isDbLockedForMode(kSecondaryNssOtherDb1.db(), MODE_IS));
+    ASSERT(!opCtx1->lockState()->isDbLockedForMode(kSecondaryNssOtherDb2.db(), MODE_IS));
+    ASSERT(!opCtx1->lockState()->isRSTLExclusive());
+    ASSERT(!opCtx1->lockState()->isGlobalLockedRecursively());
+    ASSERT(!opCtx1->lockState()->isWriteLocked());
+
+    // All the locks should release.
+    autoGetDb.reset();
+    ASSERT(!opCtx1->lockState()->isLocked());  // Global lock check.
+}
+
+TEST_F(CatalogRAIITestFixture, AutoGetDbSecondaryNamespacesMultiDb) {
+    auto opCtx1 = client1.second.get();
+
+    std::set<StringData> secondaryDbNames{kSecondaryNss1.db(),
+                                          kSecondaryNss2.db(),
+                                          kSecondaryNssOtherDb1.db(),
+                                          kSecondaryNssOtherDb2.db()};
+    boost::optional<AutoGetDb> autoGetDb;
+    autoGetDb.emplace(opCtx1, nss.db(), MODE_IS, Date_t::max(), secondaryDbNames);
+
+    ASSERT(opCtx1->lockState()->isReadLocked());  // Global lock check
+    ASSERT(opCtx1->lockState()->isRSTLLocked());
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(nss.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNss1.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNss2.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNssOtherDb1.db(), MODE_IS));
+    ASSERT(opCtx1->lockState()->isDbLockedForMode(kSecondaryNssOtherDb2.db(), MODE_IS));
+
+    ASSERT(!opCtx1->lockState()->isRSTLExclusive());
+    ASSERT(!opCtx1->lockState()->isGlobalLockedRecursively());
+    ASSERT(!opCtx1->lockState()->isWriteLocked());
+
+    // All the locks should release.
+    autoGetDb.reset();
+    ASSERT(!opCtx1->lockState()->isLocked());  // Global lock check.
+}
+
+TEST_F(CatalogRAIITestFixture, AutoGetCollectionMultiNssCollLockDeadline) {
+    // Take a MODE_X collection lock on kSecondaryNss1.
+    boost::optional<AutoGetCollection> autoGetCollWithXLock;
+    autoGetCollWithXLock.emplace(client1.second.get(), kSecondaryNss1, MODE_X);
+    ASSERT(client1.second->lockState()->isDbLockedForMode(kSecondaryNss1.db(), MODE_IX));
+    ASSERT(client1.second->lockState()->isCollectionLockedForMode(kSecondaryNss1, MODE_X));
+
+    // Now trying to take a MODE_IS lock on kSecondaryNss1 as a secondary collection should fail.
+    const std::vector<NamespaceStringOrUUID> secondaryNamespacesConflict{
+        NamespaceStringOrUUID(kSecondaryNss1),
+        NamespaceStringOrUUID(kSecondaryNss2),
+        NamespaceStringOrUUID(kSecondaryNssOtherDb1)};
+    failsWithLockTimeout(
+        [&] {
+            AutoGetCollection coll(client2.second.get(),
+                                   nss,
+                                   MODE_IS,
+                                   AutoGetCollectionViewMode::kViewsForbidden,
+                                   Date_t::now() + timeoutMs,
+                                   secondaryNamespacesConflict);
+        },
+        timeoutMs);
+
+    {
+        // Sanity check that there's no conflict without kSecondaryNss1 that's MODE_X locked.
+        const std::vector<NamespaceStringOrUUID> secondaryNamespacesNoConflict{
+            NamespaceStringOrUUID(kSecondaryNss2), NamespaceStringOrUUID(kSecondaryNssOtherDb1)};
+        AutoGetCollection collNoConflict(client2.second.get(),
+                                         nss,
+                                         MODE_IS,
+                                         AutoGetCollectionViewMode::kViewsForbidden,
+                                         Date_t::now() + timeoutMs,
+                                         secondaryNamespacesNoConflict);
+    }
+
+    // Now without the MODE_X lock on kSecondaryNss1, should work fine.
+    autoGetCollWithXLock.reset();
+    AutoGetCollection coll(client2.second.get(),
+                           nss,
+                           MODE_IS,
+                           AutoGetCollectionViewMode::kViewsForbidden,
+                           Date_t::max(),
+                           secondaryNamespacesConflict);
+}
+
 TEST_F(CatalogRAIITestFixture, AutoGetCollectionLockFreeGlobalLockDeadline) {
     Lock::GlobalLock gLock1(client1.second.get(), MODE_X);
     ASSERT(client1.second->lockState()->isLocked());
@@ -321,5 +535,28 @@ TEST_F(ReadSourceScopeTest, RestoreReadSource) {
     ASSERT_EQ(opCtx()->recoveryUnit()->getPointInTimeReadTimestamp(opCtx()), Timestamp(1, 2));
 }
 
+// Placing DEATH_TESTs at the end of the file avoids causing problems for debuggers.
+
+DEATH_TEST_F(CatalogRAIITestFixture, AutoGetDbMultiIntentWriteLock, "invariant") {
+    AutoGetDb autoGetDb(
+        client1.second.get(), nss.db(), MODE_IX, Date_t::max(), {kSecondaryNss1.db()});
+
+    // This invariants because only MODE_IS is supported for multi namespace locking.
+}
+
+DEATH_TEST_F(CatalogRAIITestFixture, AutoGetDbMultiStrongWriteLock, "invariant") {
+    AutoGetDb autoGetDb(
+        client1.second.get(), nss.db(), MODE_X, Date_t::max(), {kSecondaryNss1.db()});
+
+    // This invariants because only MODE_IS is supported for multi namespace locking.
+}
+
+DEATH_TEST_F(CatalogRAIITestFixture, AutoGetDbMultiStrongReadLock, "invariant") {
+    AutoGetDb autoGetDb(
+        client1.second.get(), nss.db(), MODE_S, Date_t::max(), {kSecondaryNss1.db()});
+
+    // This invariants because only MODE_IS is supported for multi namespace locking.
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/concurrency/d_concurrency.cpp b/src/mongo/db/concurrency/d_concurrency.cpp
index 4a6b8cee9a1..db11862b65b 100644
--- a/src/mongo/db/concurrency/d_concurrency.cpp
+++ b/src/mongo/db/concurrency/d_concurrency.cpp
@@ -196,13 +196,19 @@ void Lock::GlobalLock::_unlock() {
     _result = LOCK_INVALID;
 }
 
-Lock::DBLock::DBLock(OperationContext* opCtx, StringData db, LockMode mode, Date_t deadline)
-    : _id(RESOURCE_DATABASE, db),
-      _opCtx(opCtx),
-      _result(LOCK_INVALID),
-      _mode(mode),
-      _globalLock(
-          opCtx, isSharedLockMode(_mode) ? MODE_IS : MODE_IX, deadline, InterruptBehavior::kThrow) {
+Lock::DBLock::DBLock(OperationContext* opCtx,
+                     StringData db,
+                     LockMode mode,
+                     Date_t deadline,
+                     bool skipGlobalAndRSTLLocks)
+    : _id(RESOURCE_DATABASE, db), _opCtx(opCtx), _result(LOCK_INVALID), _mode(mode) {
+
+    if (!skipGlobalAndRSTLLocks) {
+        _globalLock.emplace(opCtx,
+                            isSharedLockMode(_mode) ? MODE_IS : MODE_IX,
+                            deadline,
+                            InterruptBehavior::kThrow);
+    }
     massert(28539, "need a valid database name", !db.empty() && nsIsDbOnly(db));
 
     _opCtx->lockState()->lock(_opCtx, _id, _mode, deadline);
diff --git a/src/mongo/db/concurrency/d_concurrency.h b/src/mongo/db/concurrency/d_concurrency.h
index bf467472b12..1c21e9362af 100644
--- a/src/mongo/db/concurrency/d_concurrency.h
+++ b/src/mongo/db/concurrency/d_concurrency.h
@@ -306,7 +306,8 @@ public:
         DBLock(OperationContext* opCtx,
                StringData db,
                LockMode mode,
-               Date_t deadline = Date_t::max());
+               Date_t deadline = Date_t::max(),
+               bool skipGlobalAndRSTLLocks = false);
         DBLock(DBLock&&);
         ~DBLock();
 
@@ -337,7 +338,7 @@ public:
         LockMode _mode;
 
         // Acquires the global lock on our behalf.
-        GlobalLock _globalLock;
+        boost::optional<GlobalLock> _globalLock;
     };
 
     /**
diff --git a/src/mongo/db/namespace_string.cpp b/src/mongo/db/namespace_string.cpp
index 980392f4669..631177c4d63 100644
--- a/src/mongo/db/namespace_string.cpp
+++ b/src/mongo/db/namespace_string.cpp
@@ -34,6 +34,7 @@
 #include <ostream>
 
 #include "mongo/base/parse_number.h"
+#include "mongo/base/status.h"
 #include "mongo/db/server_options.h"
 #include "mongo/util/str.h"
 
@@ -374,6 +375,16 @@ bool NamespaceString::isReplicated() const {
     return true;
 }
 
+Status NamespaceStringOrUUID::isNssValid() const {
+    if (!_nss || _nss->isValid()) {
+        return Status::OK();
+    }
+
+    // _nss is set and not valid.
+    return {ErrorCodes::InvalidNamespace,
+            str::stream() << "Namespace " << _nss << " is not a valid collection name"};
+}
+
 std::string NamespaceStringOrUUID::toString() const {
     if (_nss)
         return _nss->toString();
diff --git a/src/mongo/db/namespace_string.h b/src/mongo/db/namespace_string.h
index 7f0c6c89ec5..17ea8855bc7 100644
--- a/src/mongo/db/namespace_string.h
+++ b/src/mongo/db/namespace_string.h
@@ -594,6 +594,12 @@ public:
         return _nss ? _nss->db() : StringData(_dbname);
     }
 
+    /**
+     * Returns OK if either the nss is not set or is a valid nss. Otherwise returns an
+     * InvalidNamespace error.
+     */
+    Status isNssValid() const;
+
     std::string toString() const;
 
     void serialize(BSONObjBuilder* builder, StringData fieldName) const;