SERVER-43859: Take MODE_IX locks for collection creation.

Two concurrent storage transactions can now create collections with the same
collection name. These transactions will conflict at commit time; the first
committer will win and register their collection into the global catalog. The
losing transactions will bubble a WriteConflictException.

Top-level callers that should fail if the collection already existed must now
check and fail with a NamespaceExists error code. Previously, those callers
could rely on lower level code returning the NamespaceExists error.

Callers that were implicitly creating a collection may retry the operation,
using the now-registered collection.

These transaction-local collections (UncommittedCollections) are returned when
doing any CollectionCatalog::lookup* call.

1 files changed, 3 insertions, 2 deletions

diff --git a/src/mongo/db/ops/update.cpp b/src/mongo/db/ops/update.cpp
index 6f12b861cc4..38cf8eb8c12 100644
--- a/src/mongo/db/ops/update.cpp
+++ b/src/mongo/db/ops/update.cpp
@@ -62,7 +62,8 @@ UpdateResult update(OperationContext* opCtx, Database* db, const UpdateRequest&
     invariant(!request.isExplain());
 
     const NamespaceString& nsString = request.getNamespaceString();
-    Collection* collection = CollectionCatalog::get(opCtx).lookupCollectionByNamespace(nsString);
+    Collection* collection =
+        CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, nsString);
 
     // The update stage does not create its own collection.  As such, if the update is
     // an upsert, create the collection that the update stage inserts into beforehand.
@@ -71,7 +72,7 @@ UpdateResult update(OperationContext* opCtx, Database* db, const UpdateRequest&
         // Callers should either get an X or create the collection.
         const Locker* locker = opCtx->lockState();
         invariant(locker->isW() ||
-                  locker->isLockHeldForMode(ResourceId(RESOURCE_DATABASE, nsString.db()), MODE_X));
+                  locker->isLockHeldForMode(ResourceId(RESOURCE_DATABASE, nsString.db()), MODE_IX));
 
         writeConflictRetry(opCtx, "createCollection", nsString.ns(), [&] {
             Lock::DBLock lk(opCtx, nsString.db(), MODE_X);