fix: Remove dangling reference to msgpack::object in task executor; Increase test value sizes to cover this case.

src/spider/storage/mysql/mysql_stmt.hpp

@@ -67,10 +67,10 @@ std::string const cCreateOutputTaskTable = R"(CREATE TABLE IF NOT EXISTS output_
 std::string const cCreateTaskInputTable = R"(CREATE TABLE IF NOT EXISTS `task_inputs` (
     `task_id` BINARY(16) NOT NULL,
     `position` INT UNSIGNED NOT NULL,
-    `type` VARCHAR(64) NOT NULL,
+    `type` VARCHAR(999) NOT NULL,
     `output_task_id` BINARY(16),
     `output_task_position` INT UNSIGNED,
-    `value` VARBINARY(64), -- Use VARBINARY for all types of values
+    `value` VARBINARY(999), -- Use VARBINARY for all types of values
     `data_id` BINARY(16),
     CONSTRAINT `input_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE,
     CONSTRAINT `input_task_output_match` FOREIGN KEY (`output_task_id`, `output_task_position`) REFERENCES task_outputs (`task_id`, `position`) ON UPDATE NO ACTION ON DELETE SET NULL,
@@ -81,8 +81,8 @@ std::string const cCreateTaskInputTable = R"(CREATE TABLE IF NOT EXISTS `task_in
 std::string const cCreateTaskOutputTable = R"(CREATE TABLE IF NOT EXISTS `task_outputs` (
     `task_id` BINARY(16) NOT NULL,
     `position` INT UNSIGNED NOT NULL,
-    `type` VARCHAR(64) NOT NULL,
-    `value` VARBINARY(64),
+    `type` VARCHAR(999) NOT NULL,
+    `value` VARBINARY(999),
     `data_id` BINARY(16),
     CONSTRAINT `output_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE,
     CONSTRAINT `output_data_id` FOREIGN KEY (`data_id`) REFERENCES `data` (`id`) ON UPDATE NO ACTION ON DELETE NO ACTION,
@@ -108,7 +108,7 @@ std::string const cCreateTaskInstanceTable = R"(CREATE TABLE IF NOT EXISTS `task
 
 std::string const cCreateDataTable = R"(CREATE TABLE IF NOT EXISTS `data` (
     `id` BINARY(16) NOT NULL,
-    `value` VARBINARY(256) NOT NULL,
+    `value` VARBINARY(999) NOT NULL,
     `hard_locality` BOOL DEFAULT FALSE,
     `persisted` BOOL DEFAULT FALSE,
     PRIMARY KEY (`id`)
@@ -141,14 +141,14 @@ std::string const cCreateDataRefTaskTable = R"(CREATE TABLE IF NOT EXISTS `data_
 
 std::string const cCreateClientKVDataTable = R"(CREATE TABLE IF NOT EXISTS `client_kv_data` (
     `kv_key` VARCHAR(64) NOT NULL,
-    `value` VARBINARY(128) NOT NULL,
+    `value` VARBINARY(999) NOT NULL,
     `client_id` BINARY(16) NOT NULL,
     PRIMARY KEY (`client_id`, `kv_key`)
 ))";
 
 std::string const cCreateTaskKVDataTable = R"(CREATE TABLE IF NOT EXISTS `task_kv_data` (
     `kv_key` VARCHAR(64) NOT NULL,
-    `value` VARBINARY(128) NOT NULL,
+    `value` VARBINARY(999) NOT NULL,
     `task_id` BINARY(16) NOT NULL,
     PRIMARY KEY (`task_id`, `kv_key`),
     CONSTRAINT `kv_data_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE

src/spider/worker/TaskExecutorMessage.hpp

@@ -37,29 +37,46 @@ enum class TaskExecutorRequestType : std::uint8_t {
     Resume,
 };
 
-inline auto get_request_type(msgpack::sbuffer const& buffer) -> TaskExecutorRequestType {
-    // NOLINTBEGIN(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic)
-    msgpack::object_handle const handle = msgpack::unpack(buffer.data(), buffer.size());
-    msgpack::object const object = handle.get();
-    if (object.type != msgpack::type::ARRAY || object.via.array.size < 2) {
-        return TaskExecutorRequestType::Unknown;
+class TaskExecutorRequestParser {
+public:
+    /**
+     * @param buffer
+     * @throw std::bad_cast if the buffer does not store a valid msgpack object
+     */
+    explicit TaskExecutorRequestParser(msgpack::sbuffer const& buffer)
+            : m_obj(msgpack::unpack(buffer.data(), buffer.size())) {}
+
+    /**
+     * @return The type of the message.
+     */
+    [[nodiscard]] auto get_type() const -> TaskExecutorRequestType {
+        // NOLINTBEGIN(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic)
+        msgpack::object const object = m_obj.get();
+        if (object.type != msgpack::type::ARRAY || object.via.array.size < 2) {
+            return TaskExecutorRequestType::Unknown;
+        }
+        msgpack::object const header = object.via.array.ptr[0];
+        try {
+            return header.as<TaskExecutorRequestType>();
+        } catch (msgpack::type_error const&) {
+            return TaskExecutorRequestType::Unknown;
+        }
+        // NOLINTEND(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic)
     }
-    msgpack::object const header = object.via.array.ptr[0];
-    try {
-        return header.as<TaskExecutorRequestType>();
-    } catch (msgpack::type_error const&) {
-        return TaskExecutorRequestType::Unknown;
+
+    /**
+     * @return The body of the message. Cannot outlive the `TaskExecutorRequestParser` object.
+     */
+    [[nodiscard]] auto get_body() const -> msgpack::object {
+        // NOLINTBEGIN(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic)
+        msgpack::object const object = m_obj.get();
+        return object.via.array.ptr[1];
+        // NOLINTEND(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic)
     }
-    // NOLINTEND(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic)
-}
 
-inline auto get_message_body(msgpack::sbuffer const& buffer) -> msgpack::object {
-    // NOLINTBEGIN(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic)
-    msgpack::object_handle const handle = msgpack::unpack(buffer.data(), buffer.size());
-    msgpack::object const object = handle.get();
-    return object.via.array.ptr[1];
-    // NOLINTEND(cppcoreguidelines-pro-type-union-access,cppcoreguidelines-pro-bounds-pointer-arithmetic)
-}
+private:
+    msgpack::object_handle m_obj;
+};
 
 }  // namespace spider::worker
 

src/spider/worker/task_executor.cpp

@@ -142,13 +142,12 @@ auto main(int const argc, char** argv) -> int {
             return cFuncArgParseErr;
         }
         msgpack::sbuffer const& request_buffer = request_buffer_option.value();
-        if (spider::worker::TaskExecutorRequestType::Arguments
-            != spider::worker::get_request_type(request_buffer))
-        {
+        spider::worker::TaskExecutorRequestParser const request_parser{request_buffer};
+        if (spider::worker::TaskExecutorRequestType::Arguments != request_parser.get_type()) {
             spdlog::error("Expect args request.");
             return cFuncArgParseErr;
         }
-        msgpack::object const args_object = spider::worker::get_message_body(request_buffer);
+        msgpack::object const args_object = request_parser.get_body();
         msgpack::sbuffer args_buffer;
         msgpack::packer packer{args_buffer};
         packer.pack(args_object);

tests/client/client-test.cpp

@@ -1,3 +1,4 @@
+#include <cstddef>
 #include <string>
 #include <tuple>
 #include <vector>
@@ -179,6 +180,33 @@ auto test_graph_batch_submission(spider::Driver& driver) -> int {
     return 0;
 }
 
+auto test_large_input_output(
+        spider::Driver& driver,
+        size_t const input_size_1,
+        size_t const input_size_2
+) -> int {
+    std::string input_1(input_size_1, 'a');
+    std::string input_2(input_size_2, 'b');
+
+    spider::Job<std::string> job = driver.start(&join_string_test, input_1, input_2);
+    job.wait_complete();
+    if (job.get_status() != spider::JobStatus::Succeeded) {
+        spdlog::error("Large input job failed");
+        return cJobFailed;
+    }
+    if (job.get_result() != input_1 + input_2) {
+        spdlog::error(
+                "Large input job wrong result. Expect {}. Get {}.",
+                input_1 + input_2,
+                job.get_result()
+        );
+        return cJobFailed;
+    }
+    return 0;
+}
+
+constexpr size_t cLargeInputSize = 300;
+
 }  // namespace
 
 // NOLINTNEXTLINE(bugprone-exception-escape)
@@ -243,5 +271,10 @@ auto main(int argc, char** argv) -> int {
         return result;
     }
 
+    result = test_large_input_output(driver, cLargeInputSize, cLargeInputSize);
+    if (0 != result) {
+        return result;
+    }
+
     return 0;
 }

tests/worker/test-TaskExecutor.cpp

@@ -208,6 +208,43 @@ TEMPLATE_LIST_TEST_CASE(
     REQUIRE(data_storage->remove_data(*conn, data.get_id()).success());
 }
 
+constexpr int cLargeInputSize = 300;
+
+TEMPLATE_LIST_TEST_CASE(
+        "Task execute large input&output",
+        "[worker][storage]",
+        spider::test::StorageFactoryTypeList
+) {
+    absl::flat_hash_map<
+            boost::process::v2::environment::key,
+            boost::process::v2::environment::value> const environment_variable
+            = get_environment_variable();
+
+    boost::asio::io_context context;
+
+    boost::uuids::random_generator gen;
+
+    std::string const input_1(cLargeInputSize, 'a');
+    std::string const input_2(cLargeInputSize, 'b');
+
+    spider::worker::TaskExecutor executor{
+            context,
+            "join_string_test",
+            gen(),
+            spider::test::get_storage_url<TestType>(),
+            get_libraries(),
+            environment_variable,
+            input_1,
+            input_2
+    };
+    context.run();
+    executor.wait();
+    REQUIRE(executor.succeed());
+    std::optional<std::string> const result_option = executor.get_result<std::string>();
+    REQUIRE(result_option.has_value());
+    REQUIRE(input_1 + input_2 == result_option.value_or(""));
+}
+
 }  // namespace
 
 // NOLINTEND(cert-err58-cpp,cppcoreguidelines-avoid-do-while,readability-function-cognitive-complexity,cppcoreguidelines-avoid-non-const-global-variables,cppcoreguidelines-avoid-c-arrays,modernize-avoid-c-arrays,clang-analyzer-unix.BlockInCriticalSection)

tests/worker/worker-test.cpp

@@ -3,6 +3,7 @@
 #include <iostream>
 #include <random>
 #include <stdexcept>
+#include <string>
 #include <tuple>
 
 #include <spider/client/Data.hpp>
@@ -61,6 +62,14 @@ auto create_task_test(spider::TaskContext& context, int x, int y) -> int {
     return job.get_result();
 }
 
+auto join_string_test(
+        spider::TaskContext& /*context*/,
+        std::string const& input_1,
+        std::string const& input_2
+) -> std::string {
+    return input_1 + input_2;
+}
+
 // NOLINTBEGIN(cert-err58-cpp)
 SPIDER_REGISTER_TASK(sum_test);
 SPIDER_REGISTER_TASK(swap_test);
@@ -69,4 +78,5 @@ SPIDER_REGISTER_TASK(data_test);
 SPIDER_REGISTER_TASK(random_fail_test);
 SPIDER_REGISTER_TASK(create_data_test);
 SPIDER_REGISTER_TASK(create_task_test);
+SPIDER_REGISTER_TASK(join_string_test);
 // NOLINTEND(cert-err58-cpp)

tests/worker/worker-test.hpp

@@ -1,6 +1,7 @@
 #ifndef SPIDER_TEST_WORKER_TEST_HPP
 #define SPIDER_TEST_WORKER_TEST_HPP
 
+#include <string>
 #include <tuple>
 
 #include <spider/client/Data.hpp>
@@ -20,4 +21,10 @@ auto create_data_test(spider::TaskContext& context, int x) -> spider::Data<int>;
 
 auto create_task_test(spider::TaskContext& context, int x, int y) -> int;
 
+auto join_string_test(
+        spider::TaskContext& context,
+        std::string const& input_1,
+        std::string const& input_2
+) -> std::string;
+
 #endif

tools/scripts/storage/init_db.sql

@@ -58,7 +58,7 @@ CREATE TABLE IF NOT EXISTS output_tasks
 CREATE TABLE IF NOT EXISTS `data`
 (
     `id`            BINARY(16)     NOT NULL,
-    `value`         VARBINARY(256) NOT NULL,
+    `value`         VARBINARY(999) NOT NULL,
     `hard_locality` BOOL DEFAULT FALSE,
     `persisted`     BOOL DEFAULT FALSE,
     PRIMARY KEY (`id`)
@@ -67,8 +67,8 @@ CREATE TABLE IF NOT EXISTS `task_outputs`
 (
     `task_id`  BINARY(16)   NOT NULL,
     `position` INT UNSIGNED NOT NULL,
-    `type`     VARCHAR(64)  NOT NULL,
-    `value`    VARBINARY(64),
+    `type`     VARCHAR(999)  NOT NULL,
+    `value`    VARBINARY(999),
     `data_id`  BINARY(16),
     CONSTRAINT `output_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE,
     CONSTRAINT `output_data_id` FOREIGN KEY (`data_id`) REFERENCES `data` (`id`) ON UPDATE NO ACTION ON DELETE NO ACTION,
@@ -78,10 +78,10 @@ CREATE TABLE IF NOT EXISTS `task_inputs`
 (
     `task_id`              BINARY(16)   NOT NULL,
     `position`             INT UNSIGNED NOT NULL,
-    `type`                 VARCHAR(64)  NOT NULL,
+    `type`                 VARCHAR(999)  NOT NULL,
     `output_task_id`       BINARY(16),
     `output_task_position` INT UNSIGNED,
-    `value`                VARBINARY(64), -- Use VARBINARY for all types of values
+    `value`                VARBINARY(999), -- Use VARBINARY for all types of values
     `data_id`              BINARY(16),
     CONSTRAINT `input_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE,
     CONSTRAINT `input_task_output_match` FOREIGN KEY (`output_task_id`, `output_task_position`) REFERENCES task_outputs (`task_id`, `position`) ON UPDATE NO ACTION ON DELETE SET NULL,
@@ -135,14 +135,14 @@ CREATE TABLE IF NOT EXISTS `data_ref_task`
 CREATE TABLE IF NOT EXISTS `client_kv_data`
 (
     `kv_key`    VARCHAR(64)    NOT NULL,
-    `value`     VARBINARY(128) NOT NULL,
+    `value`     VARBINARY(999) NOT NULL,
     `client_id` BINARY(16)     NOT NULL,
     PRIMARY KEY (`client_id`, `kv_key`)
 );
 CREATE TABLE IF NOT EXISTS `task_kv_data`
 (
     `kv_key`  VARCHAR(64)    NOT NULL,
-    `value`   VARBINARY(128) NOT NULL,
+    `value`   VARBINARY(999) NOT NULL,
     `task_id` BINARY(16)     NOT NULL,
     PRIMARY KEY (`task_id`, `kv_key`),
     CONSTRAINT `kv_data_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE