feat(spider-storage): Add background garbage collection to remove expired terminated jobs from the cache.

components/spider-storage/src/cache/error.rs

@@ -75,12 +75,15 @@ pub enum InternalError {
     #[error("task instance pool corrupted: {0}")]
     TaskInstancePoolCorrupted(String),
 
-    #[error("invalid config: {0}")]
+    #[error("invalid ready-queue config: {0}")]
     ReadyQueueInvalidConfig(&'static str),
 
-    #[error("invalid config: {0}")]
+    #[error("invalid task instance pool config: {0}")]
     TaskInstancePoolInvalidConfig(&'static str),
 
+    #[error("invalid job cache GC config: {0}")]
+    JobCacheGcInvalidConfig(&'static str),
+
     #[error("ready queue channel is closed")]
     ReadyQueueChannelClosed,
 

components/spider-storage/src/state.rs

@@ -1,10 +1,12 @@
 pub mod error;
 pub mod job_cache;
+pub mod job_cache_gc;
 pub mod runtime;
 pub mod service;
 
 pub use error::StorageServerError;
 pub use job_cache::JobCache;
+pub use job_cache_gc::{JobCacheGcConfig, JobCacheGcHandle, create_job_cache_gc};
 pub use runtime::{Runtime, create_runtime};
 pub use service::ServiceState;
 

components/spider-storage/src/state/job_cache.rs

@@ -1,4 +1,7 @@
-use std::collections::{HashMap, hash_map::Entry};
+use std::{
+    collections::{HashMap, hash_map::Entry},
+    sync::Arc,
+};
 
 use spider_core::types::id::JobId;
 use tokio::sync::RwLock;
@@ -21,21 +24,13 @@ use crate::{
 /// * `ReadyQueueSenderType` - The type of the ready queue sender.
 /// * `DbConnectorType` - The type of the DB-layer connector.
 /// * `TaskInstancePoolConnectorType` - The type of the task instance pool connector.
+#[derive(Clone)]
 pub struct JobCache<
     ReadyQueueSenderType: ReadyQueueSender,
     DbConnectorType: InternalJobOrchestration,
     TaskInstancePoolConnectorType: TaskInstancePoolConnector,
 > {
-    jobs: RwLock<
-        HashMap<
-            JobId,
-            SharedJobControlBlock<
-                ReadyQueueSenderType,
-                DbConnectorType,
-                TaskInstancePoolConnectorType,
-            >,
-        >,
-    >,
+    jobs: SharedJobMap<ReadyQueueSenderType, DbConnectorType, TaskInstancePoolConnectorType>,
 }
 
 impl<
@@ -48,7 +43,7 @@ impl<
     #[must_use]
     pub fn new() -> Self {
         Self {
-            jobs: RwLock::new(HashMap::new()),
+            jobs: Arc::new(RwLock::new(HashMap::new())),
         }
     }
 
@@ -106,6 +101,19 @@ impl<
         self.jobs.write().await.remove(&job_id)
     }
 
+    /// Removes multiple job control blocks from the cache.
+    ///
+    /// # Returns
+    ///
+    /// The number of job control blocks that existed and were removed.
+    pub async fn remove_batch(&self, job_ids: &[JobId]) -> usize {
+        let mut jobs = self.jobs.write().await;
+        job_ids
+            .iter()
+            .filter(|job_id| jobs.remove(job_id).is_some())
+            .count()
+    }
+
     /// Resends all ready tasks for every job in the cache to the ready queue.
     ///
     /// # Errors
@@ -132,6 +140,14 @@ impl<
     }
 }
 
+type JobMap<ReadyQueueSenderType, DbConnectorType, TaskInstancePoolConnectorType> = HashMap<
+    JobId,
+    SharedJobControlBlock<ReadyQueueSenderType, DbConnectorType, TaskInstancePoolConnectorType>,
+>;
+
+type SharedJobMap<ReadyQueueSenderType, DbConnectorType, TaskInstancePoolConnectorType> =
+    Arc<RwLock<JobMap<ReadyQueueSenderType, DbConnectorType, TaskInstancePoolConnectorType>>>;
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
@@ -225,6 +241,36 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn job_cache_remove_batch_removes_existing_jobs_once() -> anyhow::Result<()> {
+        let cache: JobCache<MockReadyQueueSender, MockDbConnector, MockTaskInstancePoolConnector> =
+            JobCache::new();
+        let first_job_id = JobId::random();
+        let second_job_id = JobId::random();
+        let missing_job_id = JobId::random();
+
+        cache.insert(create_test_jcb(first_job_id).await).await?;
+        cache.insert(create_test_jcb(second_job_id).await).await?;
+
+        let num_removed_jobs = cache
+            .remove_batch(&[first_job_id, missing_job_id, second_job_id, second_job_id])
+            .await;
+
+        assert_eq!(
+            num_removed_jobs, 2,
+            "remove_batch should count only existing jobs"
+        );
+        assert!(
+            cache.get(first_job_id).await.is_none(),
+            "first job should be removed"
+        );
+        assert!(
+            cache.get(second_job_id).await.is_none(),
+            "second job should be removed"
+        );
+        Ok(())
+    }
+
     #[tokio::test]
     async fn job_cache_get_returns_none_for_nonexistent_job() -> anyhow::Result<()> {
         let cache: JobCache<MockReadyQueueSender, MockDbConnector, MockTaskInstancePoolConnector> =