SERVER-67633 Use the RecoverableCriticalSection to synch createCollection() with Sharding DDL ops

4 files changed, 52 insertions, 20 deletions

diff --git a/jstests/sharding/create_new_collections_prepared_transactions.js b/jstests/sharding/create_new_collections_prepared_transactions.js
index d38d2dbfabd..52aa344971f 100644
--- a/jstests/sharding/create_new_collections_prepared_transactions.js
+++ b/jstests/sharding/create_new_collections_prepared_transactions.js
@@ -9,6 +9,8 @@
 (function() {
 "use strict";
 
+load("jstests/libs/auto_retry_transaction_in_sharding.js");
+
 const dbNameShard0 = "test";
 const dbNameShard2 = "testOther";
 const collName = "foo";
@@ -44,16 +46,20 @@ let doc2 = st.s.getDB(dbNameShard2).getCollection(collName).findOne({_id: 4});
 assert.eq(doc2._id, 4);
 
 jsTest.log("Testing collection creation in a cross-shard write transaction.");
-session.startTransaction({writeConcern: {w: "majority"}});
-assert.commandWorked(sessionDBShard0.createCollection(newCollName));
-assert.commandWorked(sessionDBShard2.createCollection(newCollName));
-
+const txnOptions = {
+    writeConcern: {w: "majority"}
+};
+session.startTransaction(txnOptions);
+retryOnceOnTransientAndRestartTxnOnMongos(session, () => {
+    assert.commandWorked(sessionDBShard0.createCollection(newCollName));
+    assert.commandWorked(sessionDBShard2.createCollection(newCollName));
+}, txnOptions);
 assert.commandFailedWithCode(session.commitTransaction_forTesting(),
                              ErrorCodes.OperationNotSupportedInTransaction);
 
 jsTest.log("Testing collection creation in a single-shard write transaction.");
 // TODO (SERVER-48340): Re-enable the single-write-shard transaction commit optimization.
-session.startTransaction({writeConcern: {w: "majority"}});
+session.startTransaction(txnOptions);
 assert.commandWorked(sessionDBShard0.createCollection(newCollName));
 doc2 = sessionDBShard2.getCollection(collName).findOne({_id: 4});
 assert.eq(doc2._id, 4);
diff --git a/jstests/sharding/timeseries_sharding_admin_commands.js b/jstests/sharding/timeseries_sharding_admin_commands.js
index 5e6177bb4b0..0219d8c103e 100644
--- a/jstests/sharding/timeseries_sharding_admin_commands.js
+++ b/jstests/sharding/timeseries_sharding_admin_commands.js
@@ -150,13 +150,17 @@ function assertRangeMatch(savedRange, paramRange) {
     check({[metaField]: 1}, {[metaField]: 10}, true);
 })();
 
-// Check shardingState commands will return collection info in bucket namespace.
+// Check shardingState commands returns the expected collection info about buckets & view nss.
 (function checkShardingStateCommand() {
     createTimeSeriesColl(
         {index: {[metaField]: 1, [timeField]: 1}, shardKey: {[metaField]: 1, [timeField]: 1}});
     const shardingStateRes = mongo.getPrimaryShard(dbName).adminCommand({shardingState: 1});
-    const shardingStateColls = Object.keys(shardingStateRes.versions);
-    assert(shardingStateColls.includes(bucketNss) && !shardingStateColls.includes(viewNss));
+    const shardingStateColls = shardingStateRes.versions;
+    const bucketNssIsSharded = (bucketNss in shardingStateColls &&
+                                timestampCmp(shardingStateColls[bucketNss], Timestamp(0, 0)) !== 0);
+    const viewNssIsSharded = (viewNss in shardingStateColls &&
+                              timestampCmp(shardingStateColls[viewNss], Timestamp(0, 0)) !== 0);
+    assert(bucketNssIsSharded && !viewNssIsSharded);
     dropTimeSeriesColl();
 })();
 
diff --git a/src/mongo/db/catalog/create_collection.cpp b/src/mongo/db/catalog/create_collection.cpp
index 9b1e61cbc9c..1202eb66696 100644
--- a/src/mongo/db/catalog/create_collection.cpp
+++ b/src/mongo/db/catalog/create_collection.cpp
@@ -53,6 +53,7 @@
 #include "mongo/db/ops/insert.h"
 #include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/repl/replication_coordinator.h"
+#include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/storage/storage_parameters_gen.h"
 #include "mongo/db/timeseries/timeseries_index_schema_conversion_functions.h"
 #include "mongo/db/timeseries/timeseries_options.h"
@@ -173,6 +174,8 @@ Status _createView(OperationContext* opCtx,
                           str::stream() << "Not primary while creating collection " << nss);
         }
 
+        CollectionShardingState::get(opCtx, nss)->checkShardVersionOrThrow(opCtx);
+
         if (collectionOptions.changeStreamPreAndPostImagesOptions.getEnabled()) {
             return Status(ErrorCodes::InvalidOptions,
                           "option not supported on a view: changeStreamPreAndPostImages");
@@ -332,6 +335,8 @@ Status _createTimeseries(OperationContext* opCtx,
                               str::stream() << "Not primary while creating collection " << ns);
             }
 
+            CollectionShardingState::get(opCtx, bucketsNs)->checkShardVersionOrThrow(opCtx);
+
             WriteUnitOfWork wuow(opCtx);
             AutoStatsTracker bucketsStatsTracker(
                 opCtx,
@@ -413,6 +418,7 @@ Status _createTimeseries(OperationContext* opCtx,
                     str::stream() << "Not primary while creating collection " << ns};
         }
 
+        CollectionShardingState::get(opCtx, ns)->checkShardVersionOrThrow(opCtx);
 
         _createSystemDotViewsIfNecessary(opCtx, db);
 
@@ -527,6 +533,8 @@ Status _createCollection(OperationContext* opCtx,
                           str::stream() << "Not primary while creating collection " << nss);
         }
 
+        CollectionShardingState::get(opCtx, nss)->checkShardVersionOrThrow(opCtx);
+
         WriteUnitOfWork wunit(opCtx);
 
         AutoStatsTracker statsTracker(
diff --git a/src/mongo/s/commands/cluster_create_cmd.cpp b/src/mongo/s/commands/cluster_create_cmd.cpp
index eba994dfb74..05cec7b1606 100644
--- a/src/mongo/s/commands/cluster_create_cmd.cpp
+++ b/src/mongo/s/commands/cluster_create_cmd.cpp
@@ -130,18 +130,32 @@ public:
             // Manually forward the create collection command to the primary shard.
             const auto dbInfo =
                 uassertStatusOK(Grid::get(opCtx)->catalogCache()->getDatabase(opCtx, dbName));
-            auto response = uassertStatusOK(
-                executeCommandAgainstDatabasePrimary(
-                    opCtx,
-                    dbName,
-                    dbInfo,
-                    applyReadWriteConcern(
-                        opCtx,
-                        this,
-                        CommandHelpers::filterCommandRequestForPassthrough(cmd.toBSON({}))),
-                    ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                    Shard::RetryPolicy::kIdempotent)
-                    .swResponse);
+            auto response = [&] {
+                auto cmdToSend = cmd.toBSON({});
+                cmdToSend = CommandHelpers::filterCommandRequestForPassthrough(cmd.toBSON({}));
+                cmdToSend = applyReadWriteConcern(opCtx, this, cmdToSend);
+                // If the DB primary resides on a shard, forward the create command applying the
+                // metadata that characterise an unsharded CRUD request according to the versioning
+                // protocol (the received DB version + an "UNSHARDED" shard version): this will
+                // guarantee that sharded DDL ops targeting the same namespace won't be executed in
+                // parallel.
+                // If the DB primary is hosted by the config server, apply the original metadata.
+                if (dbInfo->getPrimary() != ShardId::kConfigServerId) {
+                    cmdToSend = appendShardVersion(cmdToSend, ChunkVersion::UNSHARDED());
+                }
+                cmdToSend = appendDbVersionIfPresent(cmdToSend, dbInfo);
+
+                std::vector<AsyncRequestsSender::Request> arsRequests{
+                    AsyncRequestsSender::Request(dbInfo->getPrimary(), cmdToSend)};
+
+                auto arsResponses =
+                    gatherResponses(opCtx,
+                                    dbName,
+                                    ReadPreferenceSetting(ReadPreference::PrimaryOnly),
+                                    Shard::RetryPolicy::kIdempotent,
+                                    arsRequests);
+                return uassertStatusOK(arsResponses.front().swResponse);
+            }();
 
             const auto createStatus = mongo::getStatusFromCommandResult(response.data);
             if (createStatus == ErrorCodes::NamespaceExists &&