fix: Add classes to clean-up Spider's internal Data, Driver, and Job objects from the database when clients destroy the corresponding client objects (fixes #135).

src/spider/CMakeLists.txt

@@ -1,5 +1,8 @@
 # set variable as CACHE INTERNAL to access it from other scope
 set(SPIDER_CORE_SOURCES
+    core/DataCleaner.cpp
+    core/DriverCleaner.cpp
+    core/JobCleaner.cpp
     core/Task.cpp
     storage/mysql/MySqlConnection.cpp
     storage/mysql/MySqlStorageFactory.cpp
@@ -13,9 +16,13 @@ set(SPIDER_CORE_SOURCES
 )
 
 set(SPIDER_CORE_HEADERS
+    core/Context.hpp
     core/Error.hpp
     core/Data.hpp
+    core/DataCleaner.hpp
     core/Driver.hpp
+    core/DriverCleaner.hpp
+    core/JobCleaner.hpp
     core/KeyValueData.hpp
     core/Task.hpp
     core/TaskGraph.hpp

src/spider/client/Data.hpp

@@ -1,17 +1,16 @@
 #ifndef SPIDER_CLIENT_DATA_HPP
 #define SPIDER_CLIENT_DATA_HPP
 
-#include <cstdint>
 #include <functional>
 #include <memory>
 #include <string>
 #include <utility>
 #include <variant>
 #include <vector>
 
-#include <boost/uuid/uuid.hpp>
-
 #include <spider/client/Exception.hpp>
+#include <spider/core/Context.hpp>
+#include <spider/core/DataCleaner.hpp>
 #include <spider/core/Error.hpp>
 #include <spider/io/MsgPack.hpp>  // IWYU pragma: keep
 #include <spider/io/Serializer.hpp>
@@ -130,46 +129,37 @@ class Data {
                 conn = std::move(std::get<std::unique_ptr<core::StorageConnection>>(conn_result));
             }
             core::StorageErr err;
-            switch (m_data_source) {
-                case DataSource::Driver:
-                    err = m_data_store->add_driver_data(*conn, m_source_id, *data);
+            switch (m_context.get_source()) {
+                case core::Context::Source::Driver:
+                    err = m_data_store->add_driver_data(*conn, m_context.get_id(), *data);
                     if (!err.success()) {
                         throw ConnectionException(err.description);
                     }
                     break;
-                case DataSource::TaskContext:
-                    err = m_data_store->add_task_data(*conn, m_source_id, *data);
+                case core::Context::Source::Task:
+                    err = m_data_store->add_task_data(*conn, m_context.get_id(), *data);
                     if (!err.success()) {
                         throw ConnectionException(err.description);
                     }
                     break;
             }
-            return Data{std::move(data), m_data_store, m_storage_factory, m_connection};
+            return Data{std::move(data), m_context, m_data_store, m_storage_factory, m_connection};
         }
 
     private:
-        enum class DataSource : std::uint8_t {
-            Driver,
-            TaskContext
-        };
-
-        Builder(std::shared_ptr<core::DataStorage> data_store,
-                boost::uuids::uuid const source_id,
-                DataSource const data_source,
+        Builder(core::Context context,
+                std::shared_ptr<core::DataStorage> data_store,
                 std::shared_ptr<core::StorageFactory> storage_factory)
-                : m_data_store{std::move(data_store)},
-                  m_source_id{source_id},
-                  m_data_source{data_source},
+                : m_context{context},
+                  m_data_store{std::move(data_store)},
                   m_storage_factory{std::move(storage_factory)} {}
 
-        Builder(std::shared_ptr<core::DataStorage> data_store,
-                boost::uuids::uuid const source_id,
-                DataSource const data_source,
+        Builder(core::Context context,
+                std::shared_ptr<core::DataStorage> data_store,
                 std::shared_ptr<core::StorageFactory> storage_factory,
                 std::shared_ptr<core::StorageConnection> connection)
-                : m_data_store{std::move(data_store)},
-                  m_source_id{source_id},
-                  m_data_source{data_source},
+                : m_context{context},
+                  m_data_store{std::move(data_store)},
                   m_storage_factory{std::move(storage_factory)},
                   m_connection{std::move(connection)} {}
 
@@ -181,8 +171,7 @@ class Data {
         std::shared_ptr<core::StorageFactory> m_storage_factory;
         std::shared_ptr<core::StorageConnection> m_connection = nullptr;
 
-        boost::uuids::uuid m_source_id;
-        DataSource m_data_source;
+        core::Context m_context;
 
         friend class Driver;
         friend class TaskContext;
@@ -192,23 +181,40 @@ class Data {
 
 private:
     Data(std::unique_ptr<core::Data> impl,
+         core::Context context,
          std::shared_ptr<core::DataStorage> data_store,
          std::shared_ptr<core::StorageFactory> storage_factory)
-            : m_impl{std::move(impl)},
+            : m_data_cleaner{std::make_unique<core::DataCleaner>(
+                      impl->get_id(),
+                      context,
+                      data_store,
+                      storage_factory,
+                      nullptr
+              )},
+              m_impl{std::move(impl)},
               m_data_store{std::move(data_store)},
               m_storage_factory{std::move(storage_factory)} {}
 
     Data(std::unique_ptr<core::Data> impl,
+         core::Context context,
          std::shared_ptr<core::DataStorage> data_store,
          std::shared_ptr<core::StorageFactory> storage_factory,
          std::shared_ptr<core::StorageConnection> connection)
-            : m_impl{std::move(impl)},
+            : m_data_cleaner{std::make_unique<core::DataCleaner>(
+                      impl->get_id(),
+                      context,
+                      data_store,
+                      storage_factory,
+                      connection
+              )},
+              m_impl{std::move(impl)},
               m_data_store{std::move(data_store)},
               m_storage_factory{std::move(storage_factory)},
               m_connection{std::move(connection)} {}
 
     [[nodiscard]] auto get_impl() const -> std::unique_ptr<core::Data> const& { return m_impl; }
 
+    std::unique_ptr<core::DataCleaner> m_data_cleaner;
     std::unique_ptr<core::Data> m_impl;
     std::shared_ptr<core::DataStorage> m_data_store;
     std::shared_ptr<core::StorageFactory> m_storage_factory;

src/spider/client/Driver.cpp

@@ -14,6 +14,7 @@
 
 #include <spider/client/Exception.hpp>
 #include <spider/core/Driver.hpp>
+#include <spider/core/DriverCleaner.hpp>
 #include <spider/core/Error.hpp>
 #include <spider/core/KeyValueData.hpp>
 #include <spider/io/BoostAsio.hpp>  // IWYU pragma: keep
@@ -44,6 +45,13 @@ Driver::Driver(std::string const& storage_url)
         throw ConnectionException(err.description);
     }
 
+    m_driver_cleaner = std::make_unique<core::DriverCleaner>(
+            m_id,
+            m_metadata_storage,
+            m_storage_factory,
+            m_conn
+    );
+
     // Start a thread to send heartbeats
     // NOLINTNEXTLINE(performance-unnecessary-value-param)
     m_heartbeat_thread = std::jthread([this](std::stop_token stoken) {
@@ -85,6 +93,13 @@ Driver::Driver(std::string const& storage_url, boost::uuids::uuid const id)
         throw ConnectionException(err.description);
     }
 
+    m_driver_cleaner = std::make_unique<core::DriverCleaner>(
+            m_id,
+            m_metadata_storage,
+            m_storage_factory,
+            m_conn
+    );
+
     // Start a thread to send heartbeats
     // NOLINTNEXTLINE(performance-unnecessary-value-param)
     m_heartbeat_thread = std::jthread([this](std::stop_token stoken) {

src/spider/client/Driver.hpp

@@ -19,7 +19,7 @@
 #include <spider/client/Exception.hpp>
 #include <spider/client/Job.hpp>
 #include <spider/client/task.hpp>
-#include <spider/client/TaskGraph.hpp>
+#include <spider/core/DriverCleaner.hpp>
 #include <spider/core/Error.hpp>
 #include <spider/core/TaskGraphImpl.hpp>
 #include <spider/io/Serializer.hpp>
@@ -83,9 +83,8 @@ class Driver {
     auto get_data_builder() -> Data<T>::Builder {
         using DataBuilder = typename Data<T>::Builder;
         return DataBuilder{
+                core::Context{core::Context::Source::Driver, m_id},
                 m_data_storage,
-                m_id,
-                DataBuilder::DataSource::Driver,
                 m_storage_factory,
                 m_conn
         };
@@ -228,8 +227,7 @@ class Driver {
 
         return Job<ReturnType>{
                 job_id,
-                Job<ReturnType>::JobSource::Driver,
-                m_id,
+                core::Context{core::Context::Source::Driver, m_id},
                 m_metadata_storage,
                 m_data_storage,
                 m_storage_factory,
@@ -293,8 +291,7 @@ class Driver {
 
         return Job<ReturnType>{
                 job_id,
-                Job<ReturnType>::JobSource::Driver,
-                m_id,
+                core::Context{core::Context::Source::Driver, m_id},
                 m_metadata_storage,
                 m_data_storage,
                 m_storage_factory,
@@ -322,6 +319,7 @@ class Driver {
 
 private:
     boost::uuids::uuid m_id;
+    std::unique_ptr<core::DriverCleaner> m_driver_cleaner;
     std::shared_ptr<core::MetadataStorage> m_metadata_storage;
     std::shared_ptr<core::DataStorage> m_data_storage;
     std::shared_ptr<core::StorageFactory> m_storage_factory;

src/spider/client/Job.hpp

@@ -20,8 +20,10 @@
 #include <spider/client/Exception.hpp>
 #include <spider/client/task.hpp>
 #include <spider/client/type_utils.hpp>
+#include <spider/core/Context.hpp>
 #include <spider/core/DataImpl.hpp>
 #include <spider/core/Error.hpp>
+#include <spider/core/JobCleaner.hpp>
 #include <spider/core/JobMetadata.hpp>
 #include <spider/io/MsgPack.hpp>  // IWYU pragma: keep
 #include <spider/storage/MetadataStorage.hpp>
@@ -157,34 +159,37 @@ class Job {
     }
 
 private:
-    enum class JobSource : uint8_t {
-        Driver,
-        Task,
-    };
-
     Job(boost::uuids::uuid const id,
-        JobSource const source,
-        boost::uuids::uuid const source_id,
+        core::Context context,
         std::shared_ptr<core::MetadataStorage> metadata_storage,
         std::shared_ptr<core::DataStorage> data_storage,
         std::shared_ptr<core::StorageFactory> storage_factory)
             : m_id{id},
-              m_source{source},
-              m_source_id{source_id},
+              m_context{context},
+              m_job_cleaner{std::make_unique<core::JobCleaner>(
+                      id,
+                      metadata_storage,
+                      storage_factory,
+                      nullptr
+              )},
               m_metadata_storage{std::move(metadata_storage)},
               m_data_storage{std::move(data_storage)},
               m_storage_factory{std::move(storage_factory)} {}
 
     Job(boost::uuids::uuid const id,
-        JobSource const source,
-        boost::uuids::uuid const source_id,
+        core::Context context,
         std::shared_ptr<core::MetadataStorage> metadata_storage,
         std::shared_ptr<core::DataStorage> data_storage,
         std::shared_ptr<core::StorageFactory> storage_factory,
         std::shared_ptr<core::StorageConnection> conn)
             : m_id{id},
-              m_source{source},
-              m_source_id{source_id},
+              m_context{context},
+              m_job_cleaner{std::make_unique<core::JobCleaner>(
+                      id,
+                      metadata_storage,
+                      storage_factory,
+                      conn
+              )},
               m_metadata_storage{std::move(metadata_storage)},
               m_data_storage{std::move(data_storage)},
               m_storage_factory{std::move(storage_factory)},
@@ -255,17 +260,17 @@ class Job {
                     if (!optional_data_id.has_value()) {
                         throw ConnectionException{fmt::format("Output data ID is missing")};
                     }
-                    if (m_source == JobSource::Driver) {
+                    if (m_context.get_source() == core::Context::Source::Driver) {
                         err = m_data_storage->get_driver_data(
                                 conn,
-                                m_source_id,
+                                m_context.get_id(),
                                 optional_data_id.value(),
                                 &data
                         );
                     } else {
                         err = m_data_storage->get_task_data(
                                 conn,
-                                m_source_id,
+                                m_context.get_id(),
                                 optional_data_id.value(),
                                 &data
                         );
@@ -277,7 +282,9 @@ class Job {
                     }
                     std::get<i.cValue>(result) = core::DataImpl::create_data<DataType>(
                             std::make_unique<core::Data>(std::move(data)),
-                            m_data_storage
+                            m_context,
+                            m_data_storage,
+                            m_storage_factory
                     );
                 } else {
                     if (output.get_type() != typeid(T).name()) {
@@ -329,17 +336,17 @@ class Job {
                 if (!optional_data_id.has_value()) {
                     throw ConnectionException{fmt::format("Output data ID is missing")};
                 }
-                if (m_source == JobSource::Driver) {
+                if (m_context.get_source() == core::Context::Source::Driver) {
                     err = m_data_storage->get_driver_data(
                             conn,
-                            m_source_id,
+                            m_context.get_id(),
                             optional_data_id.value(),
                             &data
                     );
                 } else {
                     err = m_data_storage->get_task_data(
                             conn,
-                            m_source_id,
+                            m_context.get_id(),
                             optional_data_id.value(),
                             &data
                     );
@@ -351,6 +358,7 @@ class Job {
                 }
                 return core::DataImpl::create_data<DataType>(
                         std::make_unique<core::Data>(std::move(data)),
+                        m_context,
                         m_data_storage,
                         m_storage_factory
                 );
@@ -378,8 +386,8 @@ class Job {
     // NOLINTEND(readability-function-cognitive-complexity)
 
     boost::uuids::uuid m_id;
-    JobSource m_source;
-    boost::uuids::uuid m_source_id;
+    core::Context m_context;
+    std::unique_ptr<core::JobCleaner> m_job_cleaner;
     std::shared_ptr<core::MetadataStorage> m_metadata_storage;
     std::shared_ptr<core::DataStorage> m_data_storage;
     std::shared_ptr<core::StorageFactory> m_storage_factory;