SERVER-63495 Support running cluster commands through the transaction API

5 files changed, 108 insertions, 5 deletions

diff --git a/src/mongo/s/commands/SConscript b/src/mongo/s/commands/SConscript
index adbc0a3c03c..df97cd30619 100644
--- a/src/mongo/s/commands/SConscript
+++ b/src/mongo/s/commands/SConscript
@@ -12,10 +12,16 @@ env.Library(
     source=[
         'flush_router_config_cmd.cpp',
         'get_shard_map_cmd.cpp',
+        'internal_transactions_test_commands.cpp',
+        'internal_transactions_test_commands.idl',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/db/cluster_transaction_api',
         '$BUILD_DIR/mongo/db/commands',
+        '$BUILD_DIR/mongo/db/transaction_api',
         '$BUILD_DIR/mongo/s/grid',
+        '$BUILD_DIR/mongo/s/sharding_router_api',
+        '$BUILD_DIR/mongo/s/startup_initialization',
     ]
 )
 
@@ -90,11 +96,9 @@ env.Library(
         'cluster_validate_db_metadata_cmd.cpp',
         'cluster_whats_my_uri_cmd.cpp',
         'cluster_write_cmd_s.cpp',
-        'internal_transactions_test_commands.cpp',
         'kill_sessions_remote.cpp',
         's_read_write_concern_defaults_server_status.cpp',
         'cluster_commands.idl',
-        'internal_transactions_test_commands.idl',
         'shard_collection.idl',
     ],
     LIBDEPS_PRIVATE=[
diff --git a/src/mongo/s/commands/internal_transactions_test_commands.cpp b/src/mongo/s/commands/internal_transactions_test_commands.cpp
index 81ac508f06e..9e827f8d35c 100644
--- a/src/mongo/s/commands/internal_transactions_test_commands.cpp
+++ b/src/mongo/s/commands/internal_transactions_test_commands.cpp
@@ -26,9 +26,17 @@
  *    exception statement from all source files in the program, then also delete
  *    it in the license file.
  */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand
+
 #include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/cluster_transaction_api.h"
 #include "mongo/db/commands.h"
+#include "mongo/db/transaction_api.h"
+#include "mongo/logv2/log.h"
 #include "mongo/s/commands/internal_transactions_test_commands_gen.h"
+#include "mongo/s/grid.h"
+#include "mongo/s/transaction_router_resource_yielder.h"
 
 namespace mongo {
 namespace {
@@ -41,7 +49,56 @@ public:
     public:
         using InvocationBase::InvocationBase;
 
-        void typedRun(OperationContext* opCtx){};
+        TestInternalTransactionsReply typedRun(OperationContext* opCtx) {
+            Grid::get(opCtx)->assertShardingIsInitialized();
+
+            auto fixedExec = Grid::get(opCtx)->getExecutorPool()->getFixedExecutor();
+            auto txn = txn_api::TransactionWithRetries(
+                opCtx,
+                fixedExec,
+                std::make_unique<txn_api::details::SEPTransactionClient>(
+                    opCtx,
+                    fixedExec,
+                    std::make_unique<txn_api::details::ClusterSEPTransactionClientBehaviors>(
+                        opCtx->getServiceContext())),
+                TransactionRouterResourceYielder::makeForLocalHandoff());
+
+            struct SharedBlock {
+                SharedBlock(std::vector<TestInternalTransactionsCommandInfo> commandInfos_)
+                    : commandInfos(commandInfos_) {}
+
+                std::vector<TestInternalTransactionsCommandInfo> commandInfos;
+                std::vector<BSONObj> responses;
+            };
+            auto sharedBlock = std::make_shared<SharedBlock>(request().getCommandInfos());
+
+            // Swallow errors and let clients inspect the responses array to determine success /
+            // failure.
+            (void)txn.runSyncNoThrow(
+                opCtx,
+                [sharedBlock](const txn_api::TransactionClient& txnClient, ExecutorPtr txnExec) {
+                    for (const auto& commandInfo : sharedBlock->commandInfos) {
+                        const auto& dbName = commandInfo.getDbName();
+                        const auto& command = commandInfo.getCommand();
+                        auto assertSucceeds = commandInfo.getAssertSucceeds();
+
+                        auto res = txnClient.runCommand(dbName, command).get();
+                        sharedBlock->responses.emplace_back(
+                            CommandHelpers::filterCommandReplyForPassthrough(
+                                res.removeField("recoveryToken")));
+
+                        if (assertSucceeds) {
+                            // Note this only inspects the top level ok field for non-write
+                            // commands.
+                            uassertStatusOK(getStatusFromWriteCommandReply(res));
+                        }
+                    }
+
+                    return SemiFuture<void>::makeReady();
+                });
+
+            return TestInternalTransactionsReply(std::move(sharedBlock->responses));
+        };
 
         NamespaceString ns() const override {
             return NamespaceString(request().getDbName(), "");
@@ -64,6 +121,8 @@ public:
         return "Internal command for testing internal transactions";
     }
 
+    // This command can use the transaction API to run commands on different databases, so a single
+    // user database doesn't apply and we restrict this to only the admin database.
     bool adminOnly() const override {
         return true;
     }
diff --git a/src/mongo/s/commands/internal_transactions_test_commands.idl b/src/mongo/s/commands/internal_transactions_test_commands.idl
index 846351967d3..e34d4adbbb9 100644
--- a/src/mongo/s/commands/internal_transactions_test_commands.idl
+++ b/src/mongo/s/commands/internal_transactions_test_commands.idl
@@ -32,9 +32,40 @@ global:
 imports:
   - "mongo/idl/basic_types.idl"
 
+structs:
+    TestInternalTransactionsReply:
+        description: "Response for testInternalTransactions command"
+        strict: false
+        fields:
+            responses:
+                type: array<object>
+
+    TestInternalTransactionsCommandInfo:
+        description: "A command, its database name, and other test options"
+        strict: false
+        fields:
+            dbName:
+                type: string
+            command:
+                type: object
+            assertSucceeds:
+                type: bool
+                default: true
+
 commands:
     testInternalTransactions:
         command_name: testInternalTransactions
         description: "The 'testInternalTransactions' command."
         namespace: ignored
         api_version: ""
+        fields:
+            useClusterClient:
+                description: "Whether the transaction API client used should opt into running the
+                             'cluster' versions of commands that enables a non-router node to run
+                             the router versions of commands. Only meaningful on mongod because a
+                             mongos will always run 'cluster' commands."
+                type: bool
+                default: false
+            commandInfos:
+                type: array<TestInternalTransactionsCommandInfo>
+        reply_type: TestInternalTransactionsReply
diff --git a/src/mongo/s/service_entry_point_mongos.cpp b/src/mongo/s/service_entry_point_mongos.cpp
index 89e71ea5e05..43e2d5213f0 100644
--- a/src/mongo/s/service_entry_point_mongos.cpp
+++ b/src/mongo/s/service_entry_point_mongos.cpp
@@ -202,12 +202,17 @@ Future<DbResponse> HandleRequest::run() {
     return future;
 }
 
-Future<DbResponse> ServiceEntryPointMongos::handleRequest(OperationContext* opCtx,
-                                                          const Message& message) noexcept {
+Future<DbResponse> ServiceEntryPointMongos::handleRequestImpl(OperationContext* opCtx,
+                                                              const Message& message) noexcept {
     auto hr = std::make_shared<HandleRequest>(opCtx, message);
     return hr->run();
 }
 
+Future<DbResponse> ServiceEntryPointMongos::handleRequest(OperationContext* opCtx,
+                                                          const Message& message) noexcept {
+    return handleRequestImpl(opCtx, message);
+}
+
 void ServiceEntryPointMongos::onClientConnect(Client* client) {
     if (load_balancer_support::isFromLoadBalancer(client)) {
         _loadBalancedConnections.increment();
diff --git a/src/mongo/s/service_entry_point_mongos.h b/src/mongo/s/service_entry_point_mongos.h
index c821ab184ef..c5c6530d2a9 100644
--- a/src/mongo/s/service_entry_point_mongos.h
+++ b/src/mongo/s/service_entry_point_mongos.h
@@ -44,6 +44,10 @@ class ServiceEntryPointMongos final : public ServiceEntryPointImpl {
 
 public:
     using ServiceEntryPointImpl::ServiceEntryPointImpl;
+
+    static Future<DbResponse> handleRequestImpl(OperationContext* opCtx,
+                                                const Message& request) noexcept;
+
     Future<DbResponse> handleRequest(OperationContext* opCtx,
                                      const Message& request) noexcept override;