SERVER-63081 Create a test command for internal transactions

8 files changed, 295 insertions, 70 deletions

diff --git a/jstests/sharding/internal_txns/transaction_api_test_command_basic.js b/jstests/sharding/internal_txns/transaction_api_test_command_basic.js
new file mode 100644
index 00000000000..97a2e308c77
--- /dev/null
+++ b/jstests/sharding/internal_txns/transaction_api_test_command_basic.js
@@ -0,0 +1,85 @@
+/*
+ * Basic tests confirming functionality of the interalTransactionsTestCommand.
+ *
+ * @tags: [requires_fcv_60]
+ */
+(function() {
+'use strict';
+
+// This test intentionally runs commands without a logical session id, which is not compatible
+// with implicit sessions.
+TestData.disableImplicitSessions = true;
+
+const kDbName = "testDb";
+const kCollName = "testColl";
+const kNs = kDbName + "." + kCollName;
+
+const st = new ShardingTest({shards: 1});
+const rst = new ReplSetTest({nodes: 1});
+rst.startSet();
+rst.initiate();
+
+let primary = rst.getPrimary();
+let db = primary.getDB(kDbName);
+let rstColl = db.getCollection(kCollName);
+
+function runTxn(connection, commandInfos, collection) {
+    const res = assert.commandWorked(
+        connection.adminCommand({testInternalTransactions: 1, commandInfos: commandInfos}));
+    jsTest.log(res);
+
+    let i = 0;
+    commandInfos.forEach(commandInfo => {
+        assert.eq(1, res.responses[i].ok);
+        assert.eq(commandInfo.command.documents.length, res.responses[i].n);
+        commandInfo.command.documents.forEach(document => {
+            assert.eq(document, collection.findOne(document));
+        });
+        ++i;
+    });
+}
+
+// Insert initial data.
+assert.commandWorked(st.s.getCollection(kNs).insert([{_id: 0}]));
+assert.commandWorked(rstColl.insert([{_id: 0}]));
+
+const commandInfos0 = [{
+    dbName: kDbName,
+    command: {
+        insert: kCollName,
+        documents: [{_id: 1}],
+    }
+}];
+
+const commandInfos1 = [
+    {
+        dbName: kDbName,
+        command: {
+            insert: kCollName,
+            documents: [{_id: 2}],
+        }
+    },
+    {
+        dbName: kDbName,
+        command: {
+            insert: kCollName,
+            documents: [{_id: 3}, {_id: 4}],
+        }
+    }
+];
+
+jsTest.log(
+    "Insert documents without a session into a sharded cluster, using internal transactions test command.");
+runTxn(st.s, commandInfos0, st.s.getCollection(kNs));
+runTxn(st.s, commandInfos1, st.s.getCollection(kNs));
+
+jsTest.log(
+    "Insert documents without a session into a replica set, using internal transactions test command.");
+runTxn(primary, commandInfos0, rstColl);
+runTxn(primary, commandInfos1, rstColl);
+
+// TODO SERVER-65048: Add testing for retryable writes and txns run in sessions.
+
+rst.stopSet();
+st.stop();
+})();
diff --git a/jstests/sharding/transaction_api_distributed_from_shard.js b/jstests/sharding/transaction_api_distributed_from_shard.js
index 2fa38fc5481..3c5a98a1784 100644
--- a/jstests/sharding/transaction_api_distributed_from_shard.js
+++ b/jstests/sharding/transaction_api_distributed_from_shard.js
@@ -32,8 +32,8 @@ function runTestSuccess() {
     // Insert initial data.
     assert.commandWorked(st.s.getCollection(kNs).insert([{_id: 1}]));
 
-    const res = assert.commandWorked(
-        shard0Primary.adminCommand({testInternalTransactions: 1, commandInfos: commands}));
+    const res = assert.commandWorked(shard0Primary.adminCommand(
+        {testInternalTransactions: 1, commandInfos: commands, useClusterClient: true}));
     res.responses.forEach((innerRes) => {
         assert.commandWorked(innerRes, tojson(res));
     });
@@ -70,8 +70,8 @@ function runTestFailure() {
     // Insert initial data.
     assert.commandWorked(st.s.getCollection(kNs).insert([{_id: 1}]));
 
-    const res = assert.commandWorked(
-        shard0Primary.adminCommand({testInternalTransactions: 1, commandInfos: commands}));
+    const res = assert.commandWorked(shard0Primary.adminCommand(
+        {testInternalTransactions: 1, commandInfos: commands, useClusterClient: true}));
     // The clusterCount is rejected without being run, so expect one fewer response.
     assert.eq(res.responses.length, commands.length - 1, tojson(res));
 
@@ -108,8 +108,8 @@ function runTestGetMore() {
 
     const commandMetricsBefore = shard0Primary.getDB(kDbName).serverStatus().metrics.commands;
 
-    const res = assert.commandWorked(
-        shard0Primary.adminCommand({testInternalTransactions: 1, commandInfos: commands}));
+    const res = assert.commandWorked(shard0Primary.adminCommand(
+        {testInternalTransactions: 1, commandInfos: commands, useClusterClient: true}));
     assert.eq(res.responses.length, 1, tojson(res));
 
     // The response from an exhausted cursor is an array of BSON objects, so we don't assert the
diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index 6a08a6b7e9e..c6d2d8232e0 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -142,6 +142,7 @@ env.Library(
         'rotate_certificates_command.cpp',
         'generic_servers.cpp',
         'generic_servers.idl',
+        'internal_transactions_test_command.idl',
         'isself.cpp',
         'logical_session_server_status_section.cpp',
         'mr_common.cpp',
@@ -512,6 +513,7 @@ env.Library(
         "driverHelpers.cpp",
         'get_cluster_parameter_command.cpp',
         "internal_rename_if_options_and_indexes_match_cmd.cpp",
+        "internal_transactions_test_command_d.cpp",
         "fle2_compact_cmd.cpp",
         "map_reduce_command.cpp",
         "oplog_application_checks.cpp",
@@ -554,6 +556,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/database_holder',
         '$BUILD_DIR/mongo/db/catalog/index_key_validate',
         '$BUILD_DIR/mongo/db/change_stream_options_manager',
+        '$BUILD_DIR/mongo/db/cluster_transaction_api',
         '$BUILD_DIR/mongo/db/commands',
         '$BUILD_DIR/mongo/db/curop_failpoint_helpers',
         '$BUILD_DIR/mongo/db/dbhelpers',
@@ -581,6 +584,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/server_feature_flags',
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_conversion_util',
+        '$BUILD_DIR/mongo/db/transaction_api',
         '$BUILD_DIR/mongo/idl/idl_parser',
         '$BUILD_DIR/mongo/util/net/ssl_manager',
         'core',
diff --git a/src/mongo/s/commands/internal_transactions_test_commands.idl b/src/mongo/db/commands/internal_transactions_test_command.idl
index 1e68776ba94..292355cf255 100644
--- a/src/mongo/s/commands/internal_transactions_test_commands.idl
+++ b/src/mongo/db/commands/internal_transactions_test_command.idl
@@ -33,27 +33,29 @@ imports:
   - "mongo/idl/basic_types.idl"
 
 structs:
-    TestInternalTransactionsReply:
-        description: "Response for testInternalTransactions command"
-        strict: false
-        fields:
-            responses:
-                type: array<object>
+  TestInternalTransactionsCommandReply:
+    description: "Response for internal transaction command."
+    strict: false
+    fields:
+      responses:
+        description: "List of responses to commands in the transaction."
+        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
-            exhaustCursor:
-                type: bool
-                optional: true
+  TestInternalTransactionsCommandInfo:
+    description: "A command, its database name, and other test options."
+    strict: false
+    fields:
+      dbName:
+        type: string
+      command:
+        type: object
+      # TODO SERVER-64986: Remove assertSucceeds parameter.
+      assertSucceeds:
+        type: bool
+        default: true
+      exhaustCursor:
+        type: bool
+        optional: true
 
 commands:
     testInternalTransactions:
@@ -62,13 +64,14 @@ commands:
         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
+          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:
+              description: "List of TestInternalTransactionsCommandInfos to run in a single transaction."
+              type: array<TestInternalTransactionsCommandInfo>
+        reply_type: TestInternalTransactionsCommandReply
diff --git a/src/mongo/db/commands/internal_transactions_test_command_d.cpp b/src/mongo/db/commands/internal_transactions_test_command_d.cpp
new file mode 100644
index 00000000000..6892bbb8e62
--- /dev/null
+++ b/src/mongo/db/commands/internal_transactions_test_command_d.cpp
@@ -0,0 +1,66 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    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/cluster_transaction_api.h"
+#include "mongo/db/transaction_participant_resource_yielder.h"
+#include "mongo/s/commands/internal_transactions_test_command.h"
+
+namespace mongo {
+namespace {
+
+class InternalTransactionsTestCommandD
+    : public InternalTransactionsTestCommandBase<InternalTransactionsTestCommandD> {
+public:
+    static txn_api::TransactionWithRetries getTxn(OperationContext* opCtx,
+                                                  ExecutorPtr executor,
+                                                  StringData commandName,
+                                                  bool useClusterClient) {
+        // If a sharded mongod is acting as a mongos, it will need special routing behaviors.
+        if (useClusterClient) {
+            return txn_api::TransactionWithRetries(
+                opCtx,
+                executor,
+                std::make_unique<txn_api::details::SEPTransactionClient>(
+                    opCtx,
+                    executor,
+                    std::make_unique<txn_api::details::ClusterSEPTransactionClientBehaviors>(
+                        opCtx->getServiceContext())),
+                TransactionParticipantResourceYielder::make(commandName));
+        }
+
+        return txn_api::TransactionWithRetries(
+            opCtx, executor, TransactionParticipantResourceYielder::make(commandName));
+    }
+};
+
+MONGO_REGISTER_TEST_COMMAND(InternalTransactionsTestCommandD);
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/s/commands/SConscript b/src/mongo/s/commands/SConscript
index 91a07c18245..be0eb7d5203 100644
--- a/src/mongo/s/commands/SConscript
+++ b/src/mongo/s/commands/SConscript
@@ -12,15 +12,10 @@ 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',
     ]
 )
@@ -98,6 +93,7 @@ env.Library(
         'cluster_validate_db_metadata_cmd.cpp',
         'cluster_whats_my_uri_cmd.cpp',
         'cluster_write_cmd_s.cpp',
+        'internal_transactions_test_command_s.cpp',
         'kill_sessions_remote.cpp',
         'refine_collection_shard_key.idl',
         's_read_write_concern_defaults_server_status.cpp',
diff --git a/src/mongo/s/commands/internal_transactions_test_commands.cpp b/src/mongo/s/commands/internal_transactions_test_command.h
index 1843dd9c28e..dc4b7fc00c8 100644
--- a/src/mongo/s/commands/internal_transactions_test_commands.cpp
+++ b/src/mongo/s/commands/internal_transactions_test_command.h
@@ -30,40 +30,28 @@
 #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/commands/internal_transactions_test_command_gen.h"
 #include "mongo/db/query/find_command_gen.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"
+#include "mongo/stdx/future.h"
 
 namespace mongo {
 namespace {
 
-class InternalTransactionsTestCommand final : public TypedCommand<InternalTransactionsTestCommand> {
+template <typename Impl>
+class InternalTransactionsTestCommandBase : public TypedCommand<Impl> {
 public:
     using Request = TestInternalTransactions;
 
-    class Invocation final : public InvocationBase {
+    class Invocation final : public TypedCommand<Impl>::InvocationBase {
     public:
-        using InvocationBase::InvocationBase;
-
-        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());
+        using Base = typename TypedCommand<Impl>::InvocationBase;
+        using Base::Base;
 
+        TestInternalTransactionsCommandReply typedRun(OperationContext* opCtx) {
             struct SharedBlock {
                 SharedBlock(std::vector<TestInternalTransactionsCommandInfo> commandInfos_)
                     : commandInfos(commandInfos_) {}
@@ -71,13 +59,29 @@ public:
                 std::vector<TestInternalTransactionsCommandInfo> commandInfos;
                 std::vector<BSONObj> responses;
             };
-            auto sharedBlock = std::make_shared<SharedBlock>(request().getCommandInfos());
+
+            auto sharedBlock = std::make_shared<SharedBlock>(Base::request().getCommandInfos());
+
+            const auto executor = Grid::get(opCtx)->isShardingInitialized()
+                ? static_cast<ExecutorPtr>(Grid::get(opCtx)->getExecutorPool()->getFixedExecutor())
+                : static_cast<ExecutorPtr>(getTransactionExecutor());
+
+            // If internalTransactionsTestCommand is received by a mongod, it should be instantiated
+            // with the TransactionParticipant's resource yielder. If on a mongos, txn should be
+            // instantiated with the TransactionRouter's resource yielder.
+            auto txn = Impl::getTxn(opCtx,
+                                    std::move(executor),
+                                    Base::request().kCommandName,
+                                    Base::request().getUseClusterClient());
 
             // 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) {
+                    // Iterate through commands and record responses for each. Return immediately if
+                    // we encounter a response with a retriedStmtId. This field indicates that the
+                    // command and everything following it have already been executed.
                     for (const auto& commandInfo : sharedBlock->commandInfos) {
                         const auto& dbName = commandInfo.getDbName();
                         const auto& command = commandInfo.getCommand();
@@ -101,26 +105,27 @@ public:
                             continue;
                         }
 
-                        auto res = txnClient.runCommand(dbName, command).get();
+                        const auto res = txnClient.runCommand(dbName, command).get();
                         sharedBlock->responses.emplace_back(
                             CommandHelpers::filterCommandReplyForPassthrough(
                                 res.removeField("recoveryToken")));
 
+                        // TODO SERVER-64986: Remove assert check.
                         if (assertSucceeds) {
                             // Note this only inspects the top level ok field for non-write
                             // commands.
                             uassertStatusOK(getStatusFromWriteCommandReply(res));
                         }
+                        // TODO SERVER-65048: Check if result has retriedStmtId & retriedStmtIds
+                        // field, exit.
                     }
-
                     return SemiFuture<void>::makeReady();
                 });
-
-            return TestInternalTransactionsReply(std::move(sharedBlock->responses));
+            return TestInternalTransactionsCommandReply(std::move(sharedBlock->responses));
         };
 
         NamespaceString ns() const override {
-            return NamespaceString(request().getDbName(), "");
+            return NamespaceString(Base::request().getDbName(), "");
         }
 
         bool supportsWriteConcern() const override {
@@ -134,6 +139,23 @@ public:
                         ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(),
                                                            ActionType::internal));
         }
+
+        const std::shared_ptr<ThreadPool>& getTransactionExecutor() {
+            static Mutex mutex =
+                MONGO_MAKE_LATCH("InternalTransactionsTestCommandExecutor::_mutex");
+            static std::shared_ptr<ThreadPool> executor;
+
+            stdx::lock_guard<Latch> lg(mutex);
+            if (!executor) {
+                ThreadPool::Options options;
+                options.poolName = "InternalTransaction";
+                options.minThreads = 0;
+                options.maxThreads = 4;
+                executor = std::make_shared<ThreadPool>(std::move(options));
+                executor->startup();
+            }
+            return executor;
+        }
     };
 
     std::string help() const override {
@@ -146,11 +168,10 @@ public:
         return true;
     }
 
-    AllowedOnSecondary secondaryAllowed(ServiceContext*) const override {
-        return AllowedOnSecondary::kNever;
+    BasicCommand::AllowedOnSecondary secondaryAllowed(ServiceContext*) const override {
+        return BasicCommand::AllowedOnSecondary::kNever;
     }
 };
-MONGO_REGISTER_TEST_COMMAND(InternalTransactionsTestCommand);
 
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/s/commands/internal_transactions_test_command_s.cpp b/src/mongo/s/commands/internal_transactions_test_command_s.cpp
new file mode 100644
index 00000000000..16f141fb5e3
--- /dev/null
+++ b/src/mongo/s/commands/internal_transactions_test_command_s.cpp
@@ -0,0 +1,50 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/s/commands/internal_transactions_test_command.h"
+#include "mongo/s/transaction_router_resource_yielder.h"
+
+namespace mongo {
+namespace {
+
+class InternalTransactionsTestCommandS
+    : public InternalTransactionsTestCommandBase<InternalTransactionsTestCommandS> {
+public:
+    static txn_api::TransactionWithRetries getTxn(OperationContext* opCtx,
+                                                  ExecutorPtr executor,
+                                                  StringData commandName,
+                                                  bool useClusterClient) {
+        return txn_api::TransactionWithRetries(
+            opCtx, executor, TransactionRouterResourceYielder::makeForLocalHandoff());
+    }
+};
+
+MONGO_REGISTER_TEST_COMMAND(InternalTransactionsTestCommandS);
+}  // namespace
+}  // namespace mongo