feat: Add job cancellation for early stop of the job execution.

src/spider/CMakeLists.txt

@@ -61,6 +61,8 @@ set(SPIDER_WORKER_SOURCES
     worker/ChildPid.cpp
     worker/DllLoader.hpp
     worker/DllLoader.cpp
+    worker/ExecutorHandle.hpp
+    worker/ExecutorHandle.cpp
     worker/Process.hpp
     worker/Process.cpp
     worker/TaskExecutor.hpp

src/spider/client/Job.hpp

@@ -82,7 +82,19 @@ class Job {
      *
      * @throw spider::ConnectionException
      */
-    auto cancel();
+    auto cancel() -> void {
+        std::variant<std::unique_ptr<core::StorageConnection>, core::StorageErr> conn_result
+                = m_storage_factory->provide_storage_connection();
+        if (std::holds_alternative<core::StorageErr>(conn_result)) {
+            throw ConnectionException(std::get<core::StorageErr>(conn_result).description);
+        }
+        auto conn = std::move(std::get<std::unique_ptr<core::StorageConnection>>(conn_result));
+
+        core::StorageErr const err = m_metadata_storage->cancel_job(*conn, m_id);
+        if (!err.success()) {
+            throw ConnectionException(err.description);
+        }
+    }
 
     /**
      * @return Status of the job.
@@ -153,7 +165,30 @@ class Job {
      * @throw spider::ConnectionException
      */
     auto get_error() -> std::pair<std::string, std::string> {
-        throw ConnectionException{"Not implemented"};
+        if (nullptr == m_conn) {
+            std::variant<std::unique_ptr<core::StorageConnection>, core::StorageErr> conn_result
+                    = m_storage_factory->provide_storage_connection();
+            if (std::holds_alternative<core::StorageErr>(conn_result)) {
+                throw ConnectionException(std::get<core::StorageErr>(conn_result).description);
+            }
+            auto conn = std::move(std::get<std::unique_ptr<core::StorageConnection>>(conn_result));
+
+            std::pair<std::string, std::string> res;
+            core::StorageErr const err
+                    = m_metadata_storage->get_job_message(*conn, m_id, &res.first, &res.second);
+            if (false == err.success()) {
+                throw ConnectionException{err.description};
+            }
+            return res;
+        }
+
+        std::pair<std::string, std::string> res;
+        core::StorageErr const err
+                = m_metadata_storage->get_job_message(*m_conn, m_id, &res.first, &res.second);
+        if (false == err.success()) {
+            throw ConnectionException{err.description};
+        }
+        return res;
     }
 
 private:

src/spider/client/TaskContext.cpp

@@ -1,5 +1,6 @@
 #include "TaskContext.hpp"
 
+#include <cstdlib>
 #include <memory>
 #include <optional>
 #include <string>
@@ -69,4 +70,19 @@ auto TaskContext::get_jobs() -> std::vector<boost::uuids::uuid> {
     }
     return job_ids;
 }
+
+auto TaskContext::abort(std::string const& message) -> void {
+    std::variant<std::unique_ptr<core::StorageConnection>, core::StorageErr> conn_result
+            = m_storage_factory->provide_storage_connection();
+    if (std::holds_alternative<core::StorageErr>(conn_result)) {
+        throw ConnectionException(std::get<core::StorageErr>(conn_result).description);
+    }
+    auto conn = std::move(std::get<std::unique_ptr<core::StorageConnection>>(conn_result));
+
+    core::StorageErr const err = m_metadata_store->cancel_job_by_task(*conn, m_task_id, message);
+    if (!err.success()) {
+        throw ConnectionException(err.description);
+    }
+    std::quick_exit(1);
+}
 }  // namespace spider

src/spider/storage/MetadataStorage.hpp

@@ -72,6 +72,41 @@ class MetadataStorage {
             std::vector<boost::uuids::uuid>* job_ids
     ) -> StorageErr
             = 0;
+    /**
+     * Cancel a job. This will set the job state to CANCEL and set all tasks that have not
+     * finished or started to CANCEL.
+     * @param conn
+     * @param id The job id.
+     * @return The error code.
+     */
+    virtual auto cancel_job(StorageConnection& conn, boost::uuids::uuid id) -> StorageErr = 0;
+    /**
+     * Cancel a job that owns the task. This will set the job state to CANCEL and set all tasks
+     * that have not finished or started to CANCEL.
+     * @param conn
+     * @param id The task id.
+     * @param message The error message of the cancellation.
+     * @return The error code.
+     */
+    virtual auto
+    cancel_job_by_task(StorageConnection& conn, boost::uuids::uuid id, std::string const& message)
+            -> StorageErr
+            = 0;
+    /**
+     * Get the error message of a cancelled job.
+     * @param conn
+     * @param id The job id.
+     * @param function_name The function name of the cancelled task.
+     * @param message The error message of the cancellation.
+     * @return The error code.
+     */
+    virtual auto get_job_message(
+            StorageConnection& conn,
+            boost::uuids::uuid id,
+            std::string* function_name,
+            std::string* message
+    ) -> StorageErr
+            = 0;
     virtual auto remove_job(StorageConnection& conn, boost::uuids::uuid id) -> StorageErr = 0;
     virtual auto reset_job(StorageConnection& conn, boost::uuids::uuid id) -> StorageErr = 0;
     virtual auto add_child(StorageConnection& conn, boost::uuids::uuid parent_id, Task const& child)
@@ -92,6 +127,9 @@ class MetadataStorage {
     virtual auto set_task_state(StorageConnection& conn, boost::uuids::uuid id, TaskState state)
             -> StorageErr
             = 0;
+    virtual auto get_task_state(StorageConnection& conn, boost::uuids::uuid id, TaskState* state)
+            -> StorageErr
+            = 0;
     virtual auto set_task_running(StorageConnection& conn, boost::uuids::uuid id) -> StorageErr = 0;
     virtual auto add_task_instance(StorageConnection& conn, TaskInstance const& instance)
             -> StorageErr

src/spider/storage/mysql/MySqlStorage.cpp

@@ -1053,6 +1053,124 @@ auto MySqlMetadataStorage::get_jobs_by_client_id(
     return StorageErr{};
 }
 
+auto MySqlMetadataStorage::cancel_job(StorageConnection& conn, boost::uuids::uuid const id)
+        -> StorageErr {
+    try {
+        // Set all pending/ready/running tasks from the job to cancelled
+        std::unique_ptr<sql::PreparedStatement> task_statement(
+                static_cast<MySqlConnection&>(conn)->prepareStatement(
+                        "UPDATE `tasks` SET `state` = 'cancel' WHERE `job_id` = ? AND "
+                        "`state` IN ('pending', 'ready', 'running')"
+                )
+        );
+        sql::bytes id_bytes = uuid_get_bytes(id);
+        task_statement->setBytes(1, &id_bytes);
+        task_statement->executeUpdate();
+        // Set job state to cancelled
+        std::unique_ptr<sql::PreparedStatement> job_statement(
+                static_cast<MySqlConnection&>(conn)->prepareStatement(
+                        "UPDATE `jobs` SET `state` = 'cancel' WHERE `id` = ?"
+                )
+        );
+        job_statement->setBytes(1, &id_bytes);
+        job_statement->executeUpdate();
+    } catch (sql::SQLException& e) {
+        static_cast<MySqlConnection&>(conn)->rollback();
+        return StorageErr{StorageErrType::OtherErr, e.what()};
+    }
+    static_cast<MySqlConnection&>(conn)->commit();
+    return StorageErr{};
+}
+
+auto MySqlMetadataStorage::cancel_job_by_task(
+        StorageConnection& conn,
+        boost::uuids::uuid id,
+        std::string const& message
+) -> StorageErr {
+    try {
+        // Get job id
+        sql::bytes task_id_bytes = uuid_get_bytes(id);
+        std::unique_ptr<sql::PreparedStatement> statement(
+                static_cast<MySqlConnection&>(conn)->prepareStatement(
+                        "SELECT `job_id`, `func_name` FROM `tasks` WHERE `id` = ?"
+                )
+        );
+        statement->setBytes(1, &task_id_bytes);
+        std::unique_ptr<sql::ResultSet> const res{statement->executeQuery()};
+        if (res->rowsCount() == 0) {
+            static_cast<MySqlConnection&>(conn)->rollback();
+            return StorageErr{StorageErrType::KeyNotFoundErr, "No task with id"};
+        }
+        res->next();
+        boost::uuids::uuid const job_id = read_id(res->getBinaryStream("job_id"));
+        sql::bytes job_id_bytes = uuid_get_bytes(job_id);
+        std::string const function_name = get_sql_string(res->getString("func_name"));
+        // Set all pending/ready/running tasks from the job to cancelled
+        std::unique_ptr<sql::PreparedStatement> task_statement(
+                static_cast<MySqlConnection&>(conn)->prepareStatement(
+                        "UPDATE `tasks` SET `state` = 'cancel' WHERE `job_id` = ? AND "
+                        "`state` IN ('pending', 'ready', 'running')"
+                )
+        );
+        task_statement->setBytes(1, &job_id_bytes);
+        task_statement->executeUpdate();
+        // Set job state to cancelled
+        std::unique_ptr<sql::PreparedStatement> job_statement(
+                static_cast<MySqlConnection&>(conn)->prepareStatement(
+                        "UPDATE `jobs` SET `state` = 'cancel' WHERE `id` = ?"
+                )
+        );
+        job_statement->setBytes(1, &job_id_bytes);
+        job_statement->executeUpdate();
+        // Set the cancel message
+        std::unique_ptr<sql::PreparedStatement> message_statement(
+                static_cast<MySqlConnection&>(conn)->prepareStatement(
+                        "INSERT INTO `job_errors` (`job_id`, `func_name`, `message`) VALUES (?, ?, "
+                        "?) "
+                )
+        );
+        message_statement->setBytes(1, &job_id_bytes);
+        message_statement->setString(2, function_name);
+        message_statement->setString(3, message);
+        message_statement->executeUpdate();
+    } catch (sql::SQLException& e) {
+        static_cast<MySqlConnection&>(conn)->rollback();
+        return StorageErr{StorageErrType::OtherErr, e.what()};
+    }
+    static_cast<MySqlConnection&>(conn)->commit();
+    return StorageErr{};
+}
+
+auto MySqlMetadataStorage::get_job_message(
+        StorageConnection& conn,
+        boost::uuids::uuid const id,
+        std::string* function_name,
+        std::string* message
+) -> StorageErr {
+    try {
+        std::unique_ptr<sql::PreparedStatement> statement{
+                static_cast<MySqlConnection&>(conn)->prepareStatement(
+                        "SELECT `func_name`, `message` FROM `job_errors` WHERE `job_id` = ?"
+                )
+        };
+        sql::bytes id_bytes = uuid_get_bytes(id);
+        statement->setBytes(1, &id_bytes);
+        std::unique_ptr<sql::ResultSet> const res{statement->executeQuery()};
+        if (res->rowsCount() == 0) {
+            static_cast<MySqlConnection&>(conn)->commit();
+            return StorageErr{StorageErrType::KeyNotFoundErr, "No messages found"};
+        }
+        res->next();
+        *function_name = get_sql_string(res->getString("func_name"));
+        *message = get_sql_string(res->getString("message"));
+    } catch (sql::SQLException& e) {
+        static_cast<MySqlConnection&>(conn)->rollback();
+        return StorageErr{StorageErrType::OtherErr, e.what()};
+    }
+    static_cast<MySqlConnection&>(conn)->commit();
+    return StorageErr{};
+}
+
 auto MySqlMetadataStorage::remove_job(StorageConnection& conn, boost::uuids::uuid id)
         -> StorageErr {
     try {
@@ -1399,6 +1517,39 @@ auto MySqlMetadataStorage::set_task_state(
     return StorageErr{};
 }
 
+auto MySqlMetadataStorage::get_task_state(
+        StorageConnection& conn,
+        boost::uuids::uuid const id,
+        TaskState* state
+) -> StorageErr {
+    try {
+        // Get the state of the task
+        std::unique_ptr<sql::PreparedStatement> statement(
+                static_cast<MySqlConnection&>(conn)->prepareStatement(
+                        "SELECT `state` FROM `tasks` WHERE `id` = ?"
+                )
+        );
+        sql::bytes id_bytes = uuid_get_bytes(id);
+        statement->setBytes(1, &id_bytes);
+        std::unique_ptr<sql::ResultSet> const res(statement->executeQuery());
+        if (res->rowsCount() == 0) {
+            static_cast<MySqlConnection&>(conn)->commit();
+            return StorageErr{
+                    StorageErrType::KeyNotFoundErr,
+                    fmt::format("No task with id {} ", boost::uuids::to_string(id))
+            };
+        }
+        res->next();
+        std::string const state_str = get_sql_string(res->getString("state"));
+        *state = string_to_task_state(state_str);
+    } catch (sql::SQLException& e) {
+        static_cast<MySqlConnection&>(conn)->rollback();
+        return StorageErr{StorageErrType::OtherErr, e.what()};
+    }
+    static_cast<MySqlConnection&>(conn)->commit();
+    return StorageErr{};
+}
+
 auto MySqlMetadataStorage::set_task_running(StorageConnection& conn, boost::uuids::uuid id)
         -> StorageErr {
     try {
@@ -1682,11 +1833,16 @@ auto MySqlMetadataStorage::task_fail(
             // Set the task fail if the last task instance fails
             std::unique_ptr<sql::PreparedStatement> const task_statement(
                     static_cast<MySqlConnection&>(conn)->prepareStatement(
-                            "UPDATE `tasks` SET `state` = 'fail' WHERE `id` = ?"
+                            "UPDATE `tasks` SET `state` = 'fail' WHERE `id` = ? AND `state` = "
+                            "'running'"
                     )
             );
             task_statement->setBytes(1, &task_id_bytes);
-            task_statement->executeUpdate();
+            int32_t const task_count = task_statement->executeUpdate();
+            if (task_count == 0) {
+                static_cast<MySqlConnection&>(conn)->commit();
+                return StorageErr{};
+            }
             // Set the job fails
             std::unique_ptr<sql::PreparedStatement> const job_statement(
                     static_cast<MySqlConnection&>(conn)->prepareStatement(

src/spider/storage/mysql/MySqlStorage.hpp

@@ -71,6 +71,16 @@ class MySqlMetadataStorage : public MetadataStorage {
             boost::uuids::uuid client_id,
             std::vector<boost::uuids::uuid>* job_ids
     ) -> StorageErr override;
+    auto cancel_job(StorageConnection& conn, boost::uuids::uuid id) -> StorageErr override;
+    auto
+    cancel_job_by_task(StorageConnection& conn, boost::uuids::uuid id, std::string const& message)
+            -> StorageErr override;
+    auto get_job_message(
+            StorageConnection& conn,
+            boost::uuids::uuid id,
+            std::string* function_name,
+            std::string* message
+    ) -> StorageErr override;
     auto remove_job(StorageConnection& conn, boost::uuids::uuid id) -> StorageErr override;
     auto reset_job(StorageConnection& conn, boost::uuids::uuid id) -> StorageErr override;
     auto add_child(StorageConnection& conn, boost::uuids::uuid parent_id, Task const& child)
@@ -86,6 +96,8 @@ class MySqlMetadataStorage : public MetadataStorage {
     ) -> StorageErr override;
     auto set_task_state(StorageConnection& conn, boost::uuids::uuid id, TaskState state)
             -> StorageErr override;
+    auto get_task_state(StorageConnection& conn, boost::uuids::uuid id, TaskState* state)
+            -> StorageErr override;
     auto set_task_running(StorageConnection& conn, boost::uuids::uuid id) -> StorageErr override;
     auto add_task_instance(StorageConnection& conn, TaskInstance const& instance)
             -> StorageErr override;

src/spider/storage/mysql/mysql_stmt.hpp

@@ -32,6 +32,14 @@ std::string const cCreateJobTable = R"(CREATE TABLE IF NOT EXISTS jobs (
     PRIMARY KEY (`id`)
 ))";
 
+std::string const cCreateJobErrorTable = R"(CREATE TABLE IF NOT EXISTS `job_errors` (
+    `job_id` BINARY(16) NOT NULL,
+    `func_name` VARCHAR(64) NOT NULL,
+    `message` VARCHAR(999) NOT NULL,
+    CONSTRAINT `job_error_job_id` FOREIGN KEY (`job_id`) REFERENCES `jobs` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE,
+    PRIMARY KEY (`job_id`)
+))";
+
 std::string const cCreateTaskTable = R"(CREATE TABLE IF NOT EXISTS tasks (
     `id` BINARY(16) NOT NULL,
     `job_id` BINARY(16) NOT NULL,
@@ -167,10 +175,11 @@ std::string const cCreateTaskKVDataTable = R"(CREATE TABLE IF NOT EXISTS `task_k
     CONSTRAINT `kv_data_task_id` FOREIGN KEY (`task_id`) REFERENCES `tasks` (`id`) ON UPDATE NO ACTION ON DELETE CASCADE
 ))";
 
-std::array<std::string const, 17> const cCreateStorage = {
+std::array<std::string const, 18> const cCreateStorage = {
         cCreateDriverTable,  // drivers table must be created before data_ref_driver
         cCreateSchedulerTable,
-        cCreateJobTable,  // jobs table must be created before task
+        cCreateJobTable,  // jobs table must be created before task and job_error
+        cCreateJobErrorTable,
         cCreateTaskTable,  // tasks table must be created before data_ref_task
         cCreateDataTable,  // data table must be created before task_outputs
         cCreateDataLocalityTable,