Update upstream source from tag 'upstream/6.0.0'master

Update to upstream version '6.0.0'
with Debian dir 5604a80ec1c96ca76f25f40d78e6ef855abec322

1 files changed, 58 insertions, 181 deletions

diff --git a/src/mongo/db/s/collmod_coordinator.cpp b/src/mongo/db/s/collmod_coordinator.cpp
index 2dbf5e176e3..9e75175964f 100644
--- a/src/mongo/db/s/collmod_coordinator.cpp
+++ b/src/mongo/db/s/collmod_coordinator.cpp
@@ -31,7 +31,6 @@
 
 #include "mongo/db/s/collmod_coordinator.h"
 
-#include "mongo/db/catalog/coll_mod.h"
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_uuid_mismatch.h"
 #include "mongo/db/catalog/database_holder.h"
@@ -40,7 +39,6 @@
 #include "mongo/db/ops/insert.h"
 #include "mongo/db/s/participant_block_gen.h"
 #include "mongo/db/s/sharded_collmod_gen.h"
-#include "mongo/db/s/sharding_ddl_coordinator_gen.h"
 #include "mongo/db/s/sharding_ddl_util.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/db/timeseries/catalog_helper.h"
@@ -48,6 +46,7 @@
 #include "mongo/db/timeseries/timeseries_options.h"
 #include "mongo/idl/idl_parser.h"
 #include "mongo/logv2/log.h"
+#include "mongo/s/async_requests_sender.h"
 #include "mongo/s/cluster_commands_helpers.h"
 #include "mongo/s/grid.h"
 #include "mongo/util/fail_point.h"
@@ -135,13 +134,7 @@ void CollModCoordinator::_enterPhase(Phase newPhase) {
     if (_doc.getPhase() == Phase::kUnset) {
         newDoc = _insertStateDocument(std::move(newDoc));
     } else {
-        ServiceContext::UniqueOperationContext uniqueOpCtx;
-        auto opCtx = cc().getOperationContext();
-        if (!opCtx) {
-            uniqueOpCtx = cc().makeOperationContext();
-            opCtx = uniqueOpCtx.get();
-        }
-        newDoc = _updateStateDocument(opCtx, std::move(newDoc));
+        newDoc = _updateStateDocument(cc().makeOperationContext().get(), std::move(newDoc));
     }
 
     {
@@ -180,85 +173,19 @@ void CollModCoordinator::_saveShardingInfoOnCoordinatorIfNecessary(OperationCont
         6522700, "Sharding information must be gathered after collection information", _collInfo);
     if (!_shardingInfo && _collInfo->isSharded) {
         ShardingInfo info;
-        info.isPrimaryOwningChunks = false;
         const auto chunkManager =
             uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfoWithRefresh(
                 opCtx, _collInfo->nsForTargeting));
 
-        // Coordinator is guaranteed to be running on primary shard
-        info.primaryShard = ShardingState::get(opCtx)->shardId();
-
+        info.primaryShard = chunkManager.dbPrimary();
         std::set<ShardId> shardIdsSet;
         chunkManager.getAllShardIds(&shardIdsSet);
-        std::vector<ShardId> participantsNotOwningChunks;
-
-        std::vector<ShardId> shardIdsVec;
-        shardIdsVec.reserve(shardIdsSet.size());
-        for (const auto& shard : shardIdsSet) {
-            if (shard != info.primaryShard) {
-                shardIdsVec.push_back(shard);
-            } else {
-                info.isPrimaryOwningChunks = true;
-            }
-        }
-
-        auto allShards = Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx);
-        for (const auto& shard : allShards) {
-            if (std::find(shardIdsVec.begin(), shardIdsVec.end(), shard) == shardIdsVec.end() &&
-                shard != info.primaryShard) {
-                participantsNotOwningChunks.push_back(shard);
-            }
-        }
-
-        info.participantsOwningChunks = std::move(shardIdsVec);
-        info.participantsNotOwningChunks = std::move(participantsNotOwningChunks);
+        std::vector<ShardId> shardIdsVec{shardIdsSet.begin(), shardIdsSet.end()};
+        info.shardsOwningChunks = std::move(shardIdsVec);
         _shardingInfo = std::move(info);
     }
 }
 
-std::vector<AsyncRequestsSender::Response> CollModCoordinator::_sendCollModToPrimaryShard(
-    OperationContext* opCtx,
-    ShardsvrCollModParticipant& request,
-    const std::shared_ptr<executor::ScopedTaskExecutor>& executor) {
-    // A view definition will only be present on the primary shard. So we pass an addition
-    // 'performViewChange' flag only to the primary shard.
-    request.setPerformViewChange(true);
-
-    return sharding_ddl_util::sendAuthenticatedCommandToShards(
-        opCtx,
-        nss().db(),
-        CommandHelpers::appendMajorityWriteConcern(request.toBSON({})),
-        {_shardingInfo->primaryShard},
-        **executor,
-        _shardingInfo->isPrimaryOwningChunks);
-}
-
-std::vector<AsyncRequestsSender::Response> CollModCoordinator::_sendCollModToParticipantShards(
-    OperationContext* opCtx,
-    ShardsvrCollModParticipant& request,
-    const std::shared_ptr<executor::ScopedTaskExecutor>& executor) {
-    request.setPerformViewChange(false);
-
-    // The collMod command targets all shards, regardless of whether they have chunks. The shards
-    // that have no chunks for the collection will not throw nor will be included in the responses.
-
-    sharding_ddl_util::sendAuthenticatedCommandToShards(
-        opCtx,
-        nss().db(),
-        CommandHelpers::appendMajorityWriteConcern(request.toBSON({})),
-        _shardingInfo->participantsNotOwningChunks,
-        **executor,
-        false /* throwOnError */);
-
-    return sharding_ddl_util::sendAuthenticatedCommandToShards(
-        opCtx,
-        nss().db(),
-        CommandHelpers::appendMajorityWriteConcern(request.toBSON({})),
-        _shardingInfo->participantsOwningChunks,
-        **executor,
-        true /* throwOnError */);
-}
-
 ExecutorFuture<void> CollModCoordinator::_runImpl(
     std::shared_ptr<executor::ScopedTaskExecutor> executor,
     const CancellationToken& token) noexcept {
@@ -305,37 +232,19 @@ ExecutorFuture<void> CollModCoordinator::_runImpl(
                 _saveCollectionInfoOnCoordinatorIfNecessary(opCtx);
 
                 if (_collInfo->isSharded) {
-                    const auto migrationsAlreadyBlockedForBucketNss =
-                        hasTimeSeriesGranularityUpdate(_request) &&
-                        _doc.getMigrationsAlreadyBlockedForBucketNss();
-
-                    if (!migrationsAlreadyBlockedForBucketNss) {
-                        _doc.setCollUUID(sharding_ddl_util::getCollectionUUID(
-                            opCtx, _collInfo->nsForTargeting, true /* allowViews */));
-                        sharding_ddl_util::stopMigrations(
-                            opCtx, _collInfo->nsForTargeting, _doc.getCollUUID());
-                    }
+                    _doc.setCollUUID(
+                        sharding_ddl_util::getCollectionUUID(opCtx, nss(), true /* allowViews */));
+                    sharding_ddl_util::stopMigrations(opCtx, nss(), _doc.getCollUUID());
                 }
 
                 _saveShardingInfoOnCoordinatorIfNecessary(opCtx);
 
                 if (_collInfo->isSharded && hasTimeSeriesGranularityUpdate(_request)) {
-                    {
-                        // Persist the migrationAlreadyBlocked flag on the coordinator document
-                        auto newDoc = _doc;
-                        newDoc.setMigrationsAlreadyBlockedForBucketNss(true);
-                        _updateStateDocument(opCtx, std::move(newDoc));
-                    }
-
                     ShardsvrParticipantBlock blockCRUDOperationsRequest(_collInfo->nsForTargeting);
                     const auto cmdObj = CommandHelpers::appendMajorityWriteConcern(
                         blockCRUDOperationsRequest.toBSON({}));
-                    std::vector<ShardId> shards = _shardingInfo->participantsOwningChunks;
-                    if (_shardingInfo->isPrimaryOwningChunks) {
-                        shards.push_back(_shardingInfo->primaryShard);
-                    }
                     sharding_ddl_util::sendAuthenticatedCommandToShards(
-                        opCtx, nss().db(), cmdObj, shards, **executor, true /* throwOnError */);
+                        opCtx, nss().db(), cmdObj, _shardingInfo->shardsOwningChunks, **executor);
                 }
             }))
         .then(_executePhase(
@@ -380,92 +289,51 @@ ExecutorFuture<void> CollModCoordinator::_runImpl(
                 _saveShardingInfoOnCoordinatorIfNecessary(opCtx);
 
                 if (_collInfo->isSharded) {
-                    try {
-                        if (!_firstExecution) {
-                            bool allowMigrations = sharding_ddl_util::checkAllowMigrations(
-                                opCtx, _collInfo->nsForTargeting);
-                            if (_result.is_initialized() && allowMigrations) {
-                                // The command finished and we have the response. Return it.
-                                return;
-                            } else if (allowMigrations) {
-                                // Previous run on a different node completed, but we lost the
-                                // result in the stepdown. Restart from stage in which we disallow
-                                // migrations.
-                                _enterPhase(Phase::kBlockShards);
-                                uasserted(ErrorCodes::Interrupted,
-                                          "Retriable error to move to previous stage");
-                            }
-                        }
-
-                        ShardsvrCollModParticipant request(nss(), _request);
-                        bool needsUnblock = _collInfo->timeSeriesOptions &&
-                            hasTimeSeriesGranularityUpdate(_request);
-                        request.setNeedsUnblock(needsUnblock);
-
-                        // If trying to convert an index to unique, executes a dryRun first to find
-                        // any duplicates without actually changing the indexes to avoid
-                        // inconsistent index specs on different shards. Example:
-                        //   Shard0: {_id: 0, a: 1}
-                        //   Shard1: {_id: 1, a: 2}, {_id: 2, a: 2}
-                        //   When trying to convert index {a: 1} to unique, the dry run will return
-                        //   the duplicate errors to the user without converting the indexes.
-                        if (isCollModIndexUniqueConversion(_request)) {
-                            // The 'dryRun' option only works with 'unique' index option. We need to
-                            // strip out other incompatible options.
-                            auto dryRunRequest = ShardsvrCollModParticipant{
-                                nss(), makeCollModDryRunRequest(_request)};
-                            std::vector<ShardId> shards = _shardingInfo->participantsOwningChunks;
-                            if (_shardingInfo->isPrimaryOwningChunks) {
-                                shards.push_back(_shardingInfo->primaryShard);
-                            }
+                    ShardsvrCollModParticipant request(nss(), _request);
+                    bool needsUnblock =
+                        _collInfo->timeSeriesOptions && hasTimeSeriesGranularityUpdate(_request);
+                    request.setNeedsUnblock(needsUnblock);
+
+                    std::vector<AsyncRequestsSender::Response> responses;
+                    auto shardsOwningChunks = _shardingInfo->shardsOwningChunks;
+                    auto primaryShardOwningChunk = std::find(shardsOwningChunks.begin(),
+                                                             shardsOwningChunks.end(),
+                                                             _shardingInfo->primaryShard);
+                    // A view definition will only be present on the primary shard. So we pass an
+                    // addition 'performViewChange' flag only to the primary shard.
+                    if (primaryShardOwningChunk != shardsOwningChunks.end()) {
+                        request.setPerformViewChange(true);
+                        const auto& primaryResponse =
                             sharding_ddl_util::sendAuthenticatedCommandToShards(
                                 opCtx,
                                 nss().db(),
-                                CommandHelpers::appendMajorityWriteConcern(
-                                    dryRunRequest.toBSON({})),
-                                shards,
+                                CommandHelpers::appendMajorityWriteConcern(request.toBSON({})),
+                                {_shardingInfo->primaryShard},
                                 **executor);
-                        }
-
-                        std::vector<AsyncRequestsSender::Response> responses;
-
-                        // In the case of the participants, we are broadcasting the collMod to all
-                        // the shards. On one hand, if the shard contains chunks for the
-                        // collections, we parse all the responses. On the other hand, if the shard
-                        // does not contain chunks, we make a best effort to not process the
-                        // returned responses or throw any errors.
-
-                        auto primaryResponse = _sendCollModToPrimaryShard(opCtx, request, executor);
-                        if (_shardingInfo->isPrimaryOwningChunks) {
-                            responses.insert(responses.end(),
-                                             std::make_move_iterator(primaryResponse.begin()),
-                                             std::make_move_iterator(primaryResponse.end()));
-                        }
-
-                        auto participantsResponses =
-                            _sendCollModToParticipantShards(opCtx, request, executor);
-                        responses.insert(responses.end(),
-                                         std::make_move_iterator(participantsResponses.begin()),
-                                         std::make_move_iterator(participantsResponses.end()));
-
-
-                        BSONObjBuilder builder;
-                        std::string errmsg;
-                        auto ok =
-                            appendRawResponses(opCtx, &errmsg, &builder, responses).responseOK;
-                        if (!errmsg.empty()) {
-                            CommandHelpers::appendSimpleCommandStatus(builder, ok, errmsg);
-                        }
-                        _result = builder.obj();
-                        sharding_ddl_util::resumeMigrations(
-                            opCtx, _collInfo->nsForTargeting, _doc.getCollUUID());
-                    } catch (DBException& ex) {
-                        if (!_isRetriableErrorForDDLCoordinator(ex.toStatus())) {
-                            sharding_ddl_util::resumeMigrations(
-                                opCtx, _collInfo->nsForTargeting, _doc.getCollUUID());
-                        }
-                        throw;
+                        responses.insert(
+                            responses.end(), primaryResponse.begin(), primaryResponse.end());
+                        shardsOwningChunks.erase(primaryShardOwningChunk);
                     }
+
+                    request.setPerformViewChange(false);
+                    const auto& secondaryResponses =
+                        sharding_ddl_util::sendAuthenticatedCommandToShards(
+                            opCtx,
+                            nss().db(),
+                            CommandHelpers::appendMajorityWriteConcern(request.toBSON({})),
+                            shardsOwningChunks,
+                            **executor);
+                    responses.insert(
+                        responses.end(), secondaryResponses.begin(), secondaryResponses.end());
+
+                    BSONObjBuilder builder;
+                    std::string errmsg;
+                    auto ok = appendRawResponses(opCtx, &errmsg, &builder, responses).responseOK;
+                    if (!errmsg.empty()) {
+                        CommandHelpers::appendSimpleCommandStatus(builder, ok, errmsg);
+                    }
+                    _result = builder.obj();
+                    sharding_ddl_util::resumeMigrations(opCtx, nss(), _doc.getCollUUID());
                 } else {
                     CollMod cmd(nss());
                     cmd.setCollModRequest(_request);
@@ -493,6 +361,15 @@ ExecutorFuture<void> CollModCoordinator::_runImpl(
                             "Error running collMod",
                             "namespace"_attr = nss(),
                             "error"_attr = redact(status));
+                // If we have the collection UUID set, this error happened in a sharded collection,
+                // we should restore the migrations.
+                if (_doc.getCollUUID()) {
+                    auto opCtxHolder = cc().makeOperationContext();
+                    auto* opCtx = opCtxHolder.get();
+                    getForwardableOpMetadata().setOn(opCtx);
+
+                    sharding_ddl_util::resumeMigrations(opCtx, nss(), _doc.getCollUUID());
+                }
             }
             return status;
         });