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

src/spider/CMakeLists.txt

@@ -68,6 +68,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

@@ -84,7 +84,20 @@ 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_by_user(*conn, m_id, "Job cancelled by user.");
+        if (!err.success()) {
+            throw ConnectionException(err.description);
+        }
+    }
 
     /**
      * @return Status of the job.
@@ -146,16 +159,41 @@ class Job {
     }
 
     /**
-     * NOTE: It is undefined behavior to call this method for a job that is not in the `Failed`
+     * NOTE: It is undefined behavior to call this method for a job that is not in the `Cancelled`
      * state.
      *
      * @return A pair:
-     * - the name of the task function that failed.
-     * - the error message sent from the task through `TaskContext::abort` or from Spider.
+     * - the name of the task function that called `TaskContext::abort`, or "user" if job is
+     *   cancelled by calling `Job::cancel`.
+     * - the error message sent from the task through `TaskContext::abort` or "Job cancelled by
+     *   user." if job is cancelled through `Job::cancel`.
      * @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_error_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_error_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

@@ -75,6 +75,47 @@ class MetadataStorage {
             std::vector<boost::uuids::uuid>* job_ids
     ) -> StorageErr
             = 0;
+    /**
+     * Cancel a job by user. Set the job state to CANCEL and set all tasks that have not finished
+     * or started to CANCEL. Set the error message of the job and offender to "user".
+     * @param conn
+     * @param id The job id.
+     * @param message The error message of the cancellation.
+     * @return The error code.
+     */
+    virtual auto
+    cancel_job_by_user(StorageConnection& conn, boost::uuids::uuid id, std::string const& message)
+            -> StorageErr
+            = 0;
+    /**
+     * Cancel the job from the task. Set the job state to CANCEL and set all tasks that have not
+     * finished or started to CANCEL. Se the error message of the job and offender to the function
+     * name of the task.
+     * @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 offender The function name of the cancelling task if job cancelled by task, "user" if
+     *                 the job is cancelled by user.
+     * @param message The error message of the cancellation.
+     * @return The error code.
+     */
+    virtual auto get_error_message(
+            StorageConnection& conn,
+            boost::uuids::uuid id,
+            std::string* offender,
+            std::string* message
+    ) -> StorageErr
+            = 0;
     virtual auto remove_job(StorageConnection& conn, boost::uuids::uuid id) noexcept -> StorageErr
             = 0;
     virtual auto reset_job(StorageConnection& conn, boost::uuids::uuid id) -> StorageErr = 0;
@@ -96,6 +137,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

@@ -1076,6 +1076,138 @@ auto MySqlMetadataStorage::get_jobs_by_client_id(
     return StorageErr{};
 }
 
+auto MySqlMetadataStorage::cancel_job_by_user(
+        StorageConnection& conn,
+        boost::uuids::uuid const id,
+        std::string const& message
+) -> 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();
+        // Set the cancel message
+        std::unique_ptr<sql::PreparedStatement> message_statement(
+                static_cast<MySqlConnection&>(conn)->prepareStatement(
+                        "INSERT INTO `job_errors` (`job_id`, `offender`, `message`) VALUES (?, ?, "
+                        "?) "
+                )
+        );
+        message_statement->setBytes(1, &id_bytes);
+        message_statement->setString(2, "user");
+        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::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`, `offender`, `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_error_message(
+        StorageConnection& conn,
+        boost::uuids::uuid const id,
+        std::string* offender,
+        std::string* message
+) -> StorageErr {
+    try {
+        std::unique_ptr<sql::PreparedStatement> statement{
+                static_cast<MySqlConnection&>(conn)->prepareStatement(
+                        "SELECT `offender`, `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();
+        *offender = get_sql_string(res->getString("offender"));
+        *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) noexcept
         -> StorageErr {
     try {
@@ -1422,6 +1554,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 {
@@ -1705,11 +1870,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

@@ -73,6 +73,18 @@ class MySqlMetadataStorage : public MetadataStorage {
             boost::uuids::uuid client_id,
             std::vector<boost::uuids::uuid>* job_ids
     ) -> StorageErr override;
+    auto
+    cancel_job_by_user(StorageConnection& conn, boost::uuids::uuid id, std::string const& message)
+            -> StorageErr override;
+    auto
+    cancel_job_by_task(StorageConnection& conn, boost::uuids::uuid id, std::string const& message)
+            -> StorageErr override;
+    auto get_error_message(
+            StorageConnection& conn,
+            boost::uuids::uuid id,
+            std::string* offender,
+            std::string* message
+    ) -> StorageErr override;
     auto remove_job(StorageConnection& conn, boost::uuids::uuid id) noexcept -> 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)
@@ -88,6 +100,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;