diff --git a/src/core/agent.rs b/src/core/agent.rs index 9c78d4e..e6b9a48 100644 --- a/src/core/agent.rs +++ b/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() { diff --git a/src/services/cron.rs b/src/services/cron.rs index 87aa1a3..85ba05f 100644 --- a/src/services/cron.rs +++ b/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 { + 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, @@ -23,7 +45,7 @@ impl CronService { pub async fn sync(&mut self, database: &DatabaseStatus) -> Result { 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 = 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 = 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), + } + } + } } diff --git a/src/tests/services/cron_tests.rs b/src/tests/services/cron_tests.rs new file mode 100644 index 0000000..7960d80 --- /dev/null +++ b/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()); +} diff --git a/src/tests/services/mod.rs b/src/tests/services/mod.rs index aa4be3a..d825608 100644 --- a/src/tests/services/mod.rs +++ b/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; diff --git a/src/utils/file.rs b/src/utils/file.rs index b39dd92..10bac58 100644 --- a/src/utils/file.rs +++ b/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;