Merge pull request #122 from Portabase/dev

Dev

src/core/agent.rs

@@ -9,6 +9,7 @@ use crate::services::restore::RestoreService;
 use crate::services::status::StatusService;
 use crate::settings::CONFIG;
 use crate::utils::common::BackupMethod;
+use std::collections::HashSet;
 use std::path::PathBuf;
 use std::sync::Arc;
 use tracing::{error, info, warn};
@@ -59,6 +60,13 @@ impl Agent {
             error!("Failed to persist dashboard cache: {e}");
         }
 
+        let live_ids: HashSet<&str> = ping_result
+            .databases
+            .iter()
+            .map(|d| d.generated_id.as_str())
+            .collect();
+        self.cron_service.prune(&live_ids).await;
+
         let merged = merge(&local.databases, &self.dashboard_cache);
 
         for db in ping_result.databases.iter() {

src/services/cron.rs

@@ -5,9 +5,31 @@ use crate::services::api::models::agent::status::DatabaseStatus;
 use crate::utils::common::vec_to_option_json;
 use crate::utils::redis_client;
 use crate::utils::task_manager::cron::check_and_update_cron;
+use crate::utils::task_manager::tasks::{SCHEDULE_KEY, remove_task};
+use redis::AsyncCommands;
 use redis::aio::MultiplexedConnection;
 use serde_json::{Value, json};
+use std::collections::HashSet;
 use std::sync::Arc;
+use tracing::{error, info};
+
+const BACKUP_TASK_PREFIX: &str = "periodic.backup_";
+
+pub fn backup_task_name(generated_id: &str) -> String {
+    format!("{BACKUP_TASK_PREFIX}{generated_id}")
+}
+
+pub fn stale_backup_tasks(scheduled_keys: &[String], live_ids: &HashSet<&str>) -> Vec<String> {
+    scheduled_keys
+        .iter()
+        .filter_map(|key| key.strip_prefix("redbeat:"))
+        .filter(|name| {
+            name.strip_prefix(BACKUP_TASK_PREFIX)
+                .is_some_and(|id| !live_ids.contains(id))
+        })
+        .map(str::to_string)
+        .collect()
+}
 
 pub struct CronService {
     ctx: Arc<Context>,
@@ -23,7 +45,7 @@ impl CronService {
     pub async fn sync(&mut self, database: &DatabaseStatus) -> Result<bool, String> {
         let generated_id = database.generated_id.as_str();
         let dbms = database.dbms.as_str();
-        let task_name = format!("periodic.backup_{}", generated_id);
+        let task_name = backup_task_name(generated_id);
         let args = vec![generated_id.to_string(), dbms.to_string()];
         let storages: Option<Value> = vec_to_option_json(database.storages.clone());
         let encrypt: bool = database.encrypt;
@@ -44,4 +66,24 @@ impl CronService {
 
         Ok(true)
     }
+
+    pub async fn prune(&mut self, live_ids: &HashSet<&str>) {
+        let keys: Vec<String> = match self.conn.zrange(SCHEDULE_KEY, 0, -1).await {
+            Ok(keys) => keys,
+            Err(e) => {
+                error!("Failed to list scheduled tasks: {:?}", e);
+                return;
+            }
+        };
+
+        for task_name in stale_backup_tasks(&keys, live_ids) {
+            match remove_task(&mut self.conn, &task_name).await {
+                Ok(()) => info!(
+                    "Task {} removed: database no longer returned by dashboard",
+                    task_name
+                ),
+                Err(e) => error!("Failed to remove task {}: {:?}", task_name, e),
+            }
+        }
+    }
 }

src/tests/services/cron_tests.rs

@@ -0,0 +1,40 @@
+use crate::services::cron::{backup_task_name, stale_backup_tasks};
+use std::collections::HashSet;
+
+const ID_A: &str = "16678159-ff7e-4c97-8c83-0adeff214681";
+const ID_B: &str = "16678124-ff7e-4c97-8c83-0adeff214681";
+
+fn key(id: &str) -> String {
+    format!("redbeat:{}", backup_task_name(id))
+}
+
+#[test]
+fn backup_task_name_format() {
+    assert_eq!(backup_task_name(ID_A), format!("periodic.backup_{ID_A}"));
+}
+
+#[test]
+fn stale_backup_tasks_returns_ids_missing_from_response() {
+    let live: HashSet<&str> = [ID_A].into();
+    assert_eq!(
+        stale_backup_tasks(&[key(ID_A), key(ID_B)], &live),
+        vec![backup_task_name(ID_B)]
+    );
+}
+
+#[test]
+fn stale_backup_tasks_empty_response_prunes_all() {
+    assert_eq!(
+        stale_backup_tasks(&[key(ID_A), key(ID_B)], &HashSet::new()),
+        vec![backup_task_name(ID_A), backup_task_name(ID_B)]
+    );
+}
+
+#[test]
+fn stale_backup_tasks_ignores_other_keys() {
+    let keys = vec![
+        "redbeat:periodic.other_x".to_string(),
+        "periodic.backup_raw".to_string(),
+    ];
+    assert!(stale_backup_tasks(&keys, &HashSet::new()).is_empty());
+}

src/tests/services/mod.rs

@@ -2,5 +2,6 @@ mod api_models_tests;
 mod backup_runner_tests;
 mod backup_uploader_tests;
 mod config_tests;
+mod cron_tests;
 mod dashboard_config_tests;
 mod restore_downloader_tests;

src/utils/file.rs

@@ -1,7 +1,6 @@
 #![allow(dead_code)]
 
 use anyhow::Result;
-use chrono::Utc;
 use serde::{Deserialize, Serialize};
 use std::path::{Path, PathBuf};
 use tokio::fs::File;
@@ -54,7 +53,7 @@ pub fn full_file_path(file_name: &String, folder_name: Option<&str>) -> String {
         .filter(|folder_name| !folder_name.is_empty())
         .unwrap_or("backups");
 
-    format!("{}/{}/{}", folder_name, Utc::now().format("%Y-%m-%d"), file_name)
+    format!("{}/{}/{}", folder_name, chrono::Local::now().format("%Y-%m-%d"), file_name)
 }
 
 const CHUNK_SIZE: usize = 16 * 1024 * 1024;