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, 123 deletions

diff --git a/src/mongo/db/s/move_primary_source_manager.cpp b/src/mongo/db/s/move_primary_source_manager.cpp
index 356ce6c8f88..dea310c8430 100644
--- a/src/mongo/db/s/move_primary_source_manager.cpp
+++ b/src/mongo/db/s/move_primary_source_manager.cpp
@@ -34,7 +34,6 @@
 #include "mongo/db/s/move_primary_source_manager.h"
 
 #include "mongo/client/connpool.h"
-#include "mongo/db/catalog/drop_collection.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/dbdirectclient.h"
@@ -47,7 +46,6 @@
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/s/catalog_cache.h"
 #include "mongo/s/grid.h"
-#include "mongo/s/request_types/move_primary_gen.h"
 #include "mongo/util/exit.h"
 #include "mongo/util/scopeguard.h"
 
@@ -217,8 +215,6 @@ Status MovePrimarySourceManager::commitOnConfig(OperationContext* opCtx) {
     invariant(_state == kCriticalSection);
     ScopeGuard scopedGuard([&] { cleanupOnError(opCtx); });
 
-    boost::optional<DatabaseVersion> expectedDbVersion;
-
     {
         AutoGetDb autoDb(opCtx, getNss().toString(), MODE_X);
 
@@ -234,13 +230,13 @@ Status MovePrimarySourceManager::commitOnConfig(OperationContext* opCtx) {
         // Read operations must begin to wait on the critical section just before we send the
         // commit operation to the config server
         dss->enterCriticalSectionCommitPhase(opCtx, dssLock, _critSecReason);
-
-        expectedDbVersion = dss->getDbVersion(opCtx, dssLock);
     }
 
+    auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
+
     auto commitStatus = [&]() {
         try {
-            return _commitOnConfig(opCtx, *expectedDbVersion);
+            return _commitOnConfig(opCtx);
         } catch (const DBException& ex) {
             return ex.toStatus();
         }
@@ -248,7 +244,8 @@ Status MovePrimarySourceManager::commitOnConfig(OperationContext* opCtx) {
 
     if (!commitStatus.isOK()) {
         // Need to get the latest optime in case the refresh request goes to a secondary --
-        // otherwise the read won't wait for the write that commit on config server may have done.
+        // otherwise the read won't wait for the write that _commitOnConfig may have
+        // done
         LOGV2(22044,
               "Error occurred while committing the movePrimary. Performing a majority write "
               "against the config server to obtain its latest optime: {error}",
@@ -332,121 +329,62 @@ Status MovePrimarySourceManager::commitOnConfig(OperationContext* opCtx) {
     return Status::OK();
 }
 
-Status MovePrimarySourceManager::_commitOnConfig(OperationContext* opCtx,
-                                                 const DatabaseVersion& expectedDbVersion) {
-    LOGV2_DEBUG(6854100,
-                3,
-                "Committing movePrimary",
-                "db"_attr = _dbname,
-                "fromShard"_attr = _fromShard,
-                "toShard"_attr = _toShard,
-                "expectedDbVersion"_attr = expectedDbVersion);
-
-    const auto commitStatus = [&] {
-        ConfigsvrCommitMovePrimary commitRequest(_dbname.toString(), expectedDbVersion, _toShard);
-        commitRequest.setDbName(NamespaceString::kAdminDb);
-
-        const auto commitResponse =
-            Grid::get(opCtx)->shardRegistry()->getConfigShard()->runCommandWithFixedRetryAttempts(
-                opCtx,
-                ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                NamespaceString::kAdminDb.toString(),
-                CommandHelpers::appendMajorityWriteConcern(commitRequest.toBSON({})),
-                Shard::RetryPolicy::kIdempotent);
-
-        const auto status = Shard::CommandResponse::getEffectiveStatus(commitResponse);
-        if (status != ErrorCodes::CommandNotFound) {
-            return status;
-        }
+Status MovePrimarySourceManager::_commitOnConfig(OperationContext* opCtx) {
+    auto const configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
+
+    auto findResponse = uassertStatusOK(
+        configShard->exhaustiveFindOnConfig(opCtx,
+                                            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+                                            repl::ReadConcernLevel::kMajorityReadConcern,
+                                            NamespaceString::kConfigDatabasesNamespace,
+                                            BSON(DatabaseType::kNameFieldName << _dbname),
+                                            BSON(DatabaseType::kNameFieldName << -1),
+                                            1));
+
+    const auto databasesVector = std::move(findResponse.docs);
+    uassert(ErrorCodes::IncompatibleShardingMetadata,
+            str::stream() << "Tried to find max database version for database '" << _dbname
+                          << "', but found no databases",
+            !databasesVector.empty());
+
+    const auto dbType =
+        DatabaseType::parse(IDLParserErrorContext("DatabaseType"), databasesVector.front());
+
+    if (dbType.getPrimary() == _toShard) {
+        return Status::OK();
+    }
 
-        LOGV2(6854101,
-              "_configsvrCommitMovePrimary command not found on config server, so try to update "
-              "the metadata document directly",
-              "db"_attr = _dbname);
+    auto newDbType = dbType;
+    newDbType.setPrimary(_toShard);
 
-        // The fallback logic is not synchronized with the removeShard command and simultaneous
-        // invocations of movePrimary and removeShard can lead to data loss.
-        return _fallbackCommitOnConfig(opCtx, expectedDbVersion);
-    }();
+    auto const currentDatabaseVersion = dbType.getVersion();
 
-    if (!commitStatus.isOK()) {
-        LOGV2(6854102,
+    newDbType.setVersion(currentDatabaseVersion.makeUpdated());
+
+    auto const updateQuery =
+        BSON(DatabaseType::kNameFieldName << _dbname << DatabaseType::kVersionFieldName
+                                          << currentDatabaseVersion.toBSON());
+
+    auto updateStatus = Grid::get(opCtx)->catalogClient()->updateConfigDocument(
+        opCtx,
+        NamespaceString::kConfigDatabasesNamespace,
+        updateQuery,
+        newDbType.toBSON(),
+        false,
+        ShardingCatalogClient::kMajorityWriteConcern);
+
+    if (!updateStatus.isOK()) {
+        LOGV2(5448803,
+              "Error committing movePrimary for {db}: {error}",
               "Error committing movePrimary",
               "db"_attr = _dbname,
-              "error"_attr = redact(commitStatus));
-        return commitStatus;
+              "error"_attr = redact(updateStatus.getStatus()));
+        return updateStatus.getStatus();
     }
 
-    const auto updatedDbType = [&]() {
-        const auto findResponse = uassertStatusOK(
-            Grid::get(opCtx)->shardRegistry()->getConfigShard()->exhaustiveFindOnConfig(
-                opCtx,
-                ReadPreferenceSetting{ReadPreference::PrimaryOnly},
-                repl::ReadConcernLevel::kMajorityReadConcern,
-                NamespaceString::kConfigDatabasesNamespace,
-                BSON(DatabaseType::kNameFieldName << _dbname),
-                BSON(DatabaseType::kNameFieldName << -1),
-                1));
-
-        const auto databases = std::move(findResponse.docs);
-        uassert(ErrorCodes::IncompatibleShardingMetadata,
-                "Tried to find version for database {}, but found no databases"_format(_dbname),
-                !databases.empty());
-
-        return DatabaseType::parse(IDLParserErrorContext("DatabaseType"), databases.front());
-    }();
-    tassert(6851100,
-            "Error committing movePrimary: database version went backwards",
-            updatedDbType.getVersion() > expectedDbVersion);
-    uassert(6851101,
-            "Error committing movePrimary: update of config.databases failed",
-            updatedDbType.getPrimary() != _fromShard);
-
-    LOGV2_DEBUG(6854103,
-                3,
-                "Commited movePrimary",
-                "db"_attr = _dbname,
-                "fromShard"_attr = _fromShard,
-                "toShard"_attr = _toShard,
-                "updatedDbVersion"_attr = updatedDbType.getVersion());
-
     return Status::OK();
 }
 
-Status MovePrimarySourceManager::_fallbackCommitOnConfig(OperationContext* opCtx,
-                                                         const DatabaseVersion& expectedDbVersion) {
-    const auto query = [&] {
-        BSONObjBuilder bsonBuilder;
-        bsonBuilder.append(DatabaseType::kNameFieldName, _dbname);
-        // Include the version in the update filter to be resilient to potential network retries and
-        // delayed messages.
-        for (const auto [fieldName, fieldValue] : expectedDbVersion.toBSON()) {
-            const auto dottedFieldName = DatabaseType::kVersionFieldName + "." + fieldName;
-            bsonBuilder.appendAs(fieldValue, dottedFieldName);
-        }
-        return bsonBuilder.obj();
-    }();
-
-    const auto update = [&] {
-        const auto newDbVersion = expectedDbVersion.makeUpdated();
-
-        BSONObjBuilder bsonBuilder;
-        bsonBuilder.append(DatabaseType::kPrimaryFieldName, _toShard);
-        bsonBuilder.append(DatabaseType::kVersionFieldName, newDbVersion.toBSON());
-        return BSON("$set" << bsonBuilder.obj());
-    }();
-
-    return Grid::get(opCtx)
-        ->catalogClient()
-        ->updateConfigDocument(opCtx,
-                               NamespaceString::kConfigDatabasesNamespace,
-                               query,
-                               update,
-                               false,
-                               ShardingCatalogClient::kMajorityWriteConcern)
-        .getStatus();
-}
-
 Status MovePrimarySourceManager::cleanStaleData(OperationContext* opCtx) {
     invariant(!opCtx->lockState()->isLocked());
     invariant(_state == kNeedCleanStaleData);
@@ -457,21 +395,17 @@ Status MovePrimarySourceManager::cleanStaleData(OperationContext* opCtx) {
     }
 
     // Only drop the cloned (unsharded) collections.
+    DBDirectClient client(opCtx);
     for (auto& coll : _clonedColls) {
-        DropReply unusedDropReply;
-        try {
-            uassertStatusOK(
-                dropCollection(opCtx,
-                               coll,
-                               &unusedDropReply,
-                               DropCollectionSystemCollectionMode::kAllowSystemCollectionDrops,
-                               false /* fromMigrate */));
-        } catch (const DBException& e) {
+        BSONObj dropCollResult;
+        client.runCommand(_dbname.toString(), BSON("drop" << coll.coll()), dropCollResult);
+        Status dropStatus = getStatusFromCommandResult(dropCollResult);
+        if (!dropStatus.isOK()) {
             LOGV2(22045,
                   "Failed to drop cloned collection {namespace} in movePrimary: {error}",
                   "Failed to drop cloned collection in movePrimary",
                   "namespace"_attr = coll,
-                  "error"_attr = redact(e));
+                  "error"_attr = redact(dropStatus));
         }
     }
 
@@ -479,6 +413,7 @@ Status MovePrimarySourceManager::cleanStaleData(OperationContext* opCtx) {
     return Status::OK();
 }
 
+
 void MovePrimarySourceManager::cleanupOnError(OperationContext* opCtx) {
     if (_state == kDone) {
         return;