fix: Move create_task_instance from scheduler to worker

src/spider/scheduler/SchedulerMessage.hpp

@@ -3,7 +3,6 @@
 
 #include <optional>
 #include <string>
-#include <tuple>
 #include <utility>
 
 #include <boost/uuid/uuid.hpp>
@@ -55,28 +54,19 @@ class ScheduleTaskResponse {
 public:
     ScheduleTaskResponse() = default;
 
-    ScheduleTaskResponse(
-            boost::uuids::uuid const task_id,
-            boost::uuids::uuid const task_instance_id
-    )
-            : m_task_ids{std::make_tuple(task_id, task_instance_id)} {}
-
-    explicit ScheduleTaskResponse(std::tuple<boost::uuids::uuid, boost::uuids::uuid> const& task_ids
-    )
-            : m_task_ids{task_ids} {}
+    explicit ScheduleTaskResponse(boost::uuids::uuid const task_id) : m_task_id{task_id} {}
 
-    [[nodiscard]] auto has_task_id() const -> bool { return m_task_ids.has_value(); }
+    [[nodiscard]] auto has_task_id() const -> bool { return m_task_id.has_value(); }
 
-    [[nodiscard]] auto get_task_ids(
-    ) const -> std::tuple<boost::uuids::uuid, boost::uuids::uuid> const& {
+    [[nodiscard]] auto get_task_id() const -> boost::uuids::uuid const& {
         // NOLINTNEXTLINE(bugprone-unchecked-optional-access)
-        return m_task_ids.value();
+        return m_task_id.value();
     }
 
-    MSGPACK_DEFINE_ARRAY(m_task_ids);
+    MSGPACK_DEFINE_ARRAY(m_task_id);
 
 private:
-    std::optional<std::tuple<boost::uuids::uuid, boost::uuids::uuid>> m_task_ids = std::nullopt;
+    std::optional<boost::uuids::uuid> m_task_id = std::nullopt;
 };
 
 }  // namespace spider::scheduler

src/spider/scheduler/SchedulerServer.cpp

@@ -12,7 +12,6 @@
 #include <spdlog/spdlog.h>
 
 #include "../core/Error.hpp"
-#include "../core/Task.hpp"
 #include "../io/BoostAsio.hpp"  // IWYU pragma: keep
 #include "../io/MsgPack.hpp"  // IWYU pragma: keep
 #include "../io/msgpack_message.hpp"
@@ -160,17 +159,7 @@ auto SchedulerServer::process_message(boost::asio::ip::tcp::socket socket
             = m_policy->schedule_next(request.get_worker_id(), request.get_worker_addr());
     ScheduleTaskResponse response{};
     if (task_id.has_value()) {
-        core::TaskInstance const instance{task_id.value()};
-        core::StorageErr const err = m_metadata_store->create_task_instance(m_conn, instance);
-        if (err.success()) {
-            response = ScheduleTaskResponse{task_id.value(), instance.id};
-        } else {
-            spdlog::error(
-                    "Cannot create task instance {}: {}",
-                    boost::uuids::to_string(task_id.value()),
-                    err.description
-            );
-        }
+        response = ScheduleTaskResponse{task_id.value()};
     }
     msgpack::sbuffer response_buffer;
     msgpack::pack(response_buffer, response);

src/spider/worker/WorkerClient.cpp

@@ -17,6 +17,7 @@
 
 #include "../core/Driver.hpp"
 #include "../core/Error.hpp"
+#include "../core/Task.hpp"
 #include "../io/BoostAsio.hpp"  // IWYU pragma: keep
 #include "../io/MsgPack.hpp"  // IWYU pragma: keep
 #include "../io/msgpack_message.hpp"
@@ -112,7 +113,23 @@ auto WorkerClient::get_next_task(std::optional<boost::uuids::uuid> const& fail_t
         if (!response.has_task_id()) {
             return std::nullopt;
         }
-        return response.get_task_ids();
+        boost::uuids::uuid const task_id = response.get_task_id();
+        std::variant<core::MySqlConnection, core::StorageErr> conn_result
+                = core::MySqlConnection::create(m_metadata_store->get_url());
+        if (std::holds_alternative<core::StorageErr>(conn_result)) {
+            spdlog::error(
+                    "Failed to connect to storage: {}",
+                    std::get<core::StorageErr>(conn_result).description
+            );
+            return std::nullopt;
+        }
+        auto& conn = std::get<core::MySqlConnection>(conn_result);
+        core::TaskInstance const instance{task_id};
+        core::StorageErr const err = m_metadata_store->create_task_instance(conn, instance);
+        if (!err.success()) {
+            return std::nullopt;
+        }
+        return std::make_tuple(task_id, instance.id);
     } catch (boost::system::system_error const& e) {
         return std::nullopt;
     } catch (std::runtime_error const& e) {

tests/scheduler/test-SchedulerPolicy.cpp

@@ -66,16 +66,28 @@ TEMPLATE_LIST_TEST_CASE(
 
     spider::scheduler::FifoPolicy policy{metadata_store, data_store, conn};
 
-    // Scheduler the earlier task
-    std::optional<boost::uuids::uuid> const optional_task_id = policy.schedule_next(gen(), "");
+    // Schedule the earlier task
+    std::optional<boost::uuids::uuid> optional_task_id = policy.schedule_next(gen(), "");
     REQUIRE(optional_task_id.has_value());
     if (optional_task_id.has_value()) {
         boost::uuids::uuid const& task_id = optional_task_id.value();
         REQUIRE(task_id == task_1.get_id());
     }
 
+    // Schedule the later task
+    optional_task_id = policy.schedule_next(gen(), "");
+    REQUIRE(optional_task_id.has_value());
+    if (optional_task_id.has_value()) {
+        boost::uuids::uuid const& task_id = optional_task_id.value();
+        REQUIRE(task_id == task_2.get_id());
+    }
+
     REQUIRE(metadata_store->remove_job(conn, job_id_1).success());
     REQUIRE(metadata_store->remove_job(conn, job_id_2).success());
+
+    // Schedule when no task available
+    optional_task_id = policy.schedule_next(gen(), "");
+    REQUIRE(!optional_task_id.has_value());
 }
 
 TEMPLATE_LIST_TEST_CASE(

tests/scheduler/test-SchedulerServer.cpp

@@ -108,7 +108,7 @@ TEMPLATE_LIST_TEST_CASE(
         spider::scheduler::ScheduleTaskResponse const res
                 = object.as<spider::scheduler::ScheduleTaskResponse>();
         REQUIRE(res.has_task_id());
-        REQUIRE(std::get<0>(res.get_task_ids()) == parent_task.get_id());
+        REQUIRE(res.get_task_id() == parent_task.get_id());
     }
     socket.close();
     server.stop();