From 55e20d48e7984fb7019265927cc8d308f841c01f Mon Sep 17 00:00:00 2001 From: charles-gauthereau Date: Thu, 27 Aug 2026 18:13:17 +0200 Subject: [PATCH 1/7] feat: configurable retry policy and combinator Adds RETRY_ATTEMPTS (3..=5, default 3) and RETRY_BACKOFF_MS (100..=30000, default 1000) to Settings, validated with the same panic-on-invalid contract as POOLING and CHUNK_SIZE_MB. The combinator logs every failed attempt and any late success through the JobLogger it borrows, so retries reach the server on the existing job-log path with no API change. It borrows rather than clones the Arc so Arc::try_unwrap in the backup executor keeps working. Backoff is exponential with equal jitter, because the uploader retries storages concurrently and would otherwise retry them in lockstep. --- docker-compose.yml | 2 + helm/values.yaml | 2 + src/settings.rs | 24 +++++- src/tests/utils/mod.rs | 1 + src/tests/utils/retry_tests.rs | 138 +++++++++++++++++++++++++++++++++ src/utils/mod.rs | 1 + src/utils/retry.rs | 73 +++++++++++++++++ 7 files changed, 240 insertions(+), 1 deletion(-) create mode 100644 src/tests/utils/retry_tests.rs create mode 100644 src/utils/retry.rs diff --git a/docker-compose.yml b/docker-compose.yml index d63d415..8859988 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -24,6 +24,8 @@ services: EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiZjlkZjhiNWYtM2I0MC00NWM3LWI3N2UtYzY4NzQ1YmU2NjMwIiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ==" #CHUNK_SIZE_MB: "1" #POOLING: 1 + #RETRY_ATTEMPTS: 3 + #RETRY_BACKOFF_MS: 1000 #DATABASES_CONFIG_FILE: "config.toml" extra_hosts: - "localhost:host-gateway" diff --git a/helm/values.yaml b/helm/values.yaml index b7a9876..dad96c4 100644 --- a/helm/values.yaml +++ b/helm/values.yaml @@ -11,6 +11,8 @@ env: POLLING: "5" APP_ENV: "production" LOG: "info" + RETRY_ATTEMPTS: "3" + RETRY_BACKOFF_MS: "1000" resources: limits: diff --git a/src/settings.rs b/src/settings.rs index b4076d1..5131fb3 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -16,6 +16,8 @@ pub struct Settings { pub timezone: String, pub log: String, pub chunk_size: usize, // bytes + pub retry_attempts: u32, + pub retry_backoff_ms: u64, } impl Settings { @@ -49,6 +51,24 @@ impl Settings { let chunk_size = chunk_size_mb * 1024 * 1024; + let retry_attempts = env::var("RETRY_ATTEMPTS") + .unwrap_or_else(|_| "3".to_string()) + .parse::() + .expect("RETRY_ATTEMPTS must be a valid positive integer"); + + if retry_attempts < 3 || retry_attempts > 5 { + panic!("RETRY_ATTEMPTS must be between 3 and 5"); + } + + let retry_backoff_ms = env::var("RETRY_BACKOFF_MS") + .unwrap_or_else(|_| "1000".to_string()) + .parse::() + .expect("RETRY_BACKOFF_MS must be a valid positive integer"); + + if retry_backoff_ms < 100 || retry_backoff_ms > 30_000 { + panic!("RETRY_BACKOFF_MS must be between 100 and 30000 milliseconds"); + } + let tz = env::var("TZ").unwrap_or_else(|_| "UTC".to_string()); Self { @@ -64,7 +84,9 @@ impl Settings { pooling: pooling_seconds, timezone: tz, log: env::var("LOG").unwrap_or_else(|_| "info".into()), - chunk_size + chunk_size, + retry_attempts, + retry_backoff_ms, } } } diff --git a/src/tests/utils/mod.rs b/src/tests/utils/mod.rs index 858d6c1..084a7e5 100644 --- a/src/tests/utils/mod.rs +++ b/src/tests/utils/mod.rs @@ -4,4 +4,5 @@ mod deserializer; mod edge_key_tests; mod file_tests; mod normalize_cron_tests; +mod retry_tests; mod stream_tests; diff --git a/src/tests/utils/retry_tests.rs b/src/tests/utils/retry_tests.rs new file mode 100644 index 0000000..115790c --- /dev/null +++ b/src/tests/utils/retry_tests.rs @@ -0,0 +1,138 @@ +use crate::services::backup::logger::JobLogger; +use crate::tests::init_tracing_for_test; +use crate::utils::retry::{RetryPolicy, retry}; + +use std::sync::Mutex; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::time::Duration; + +fn fast_policy(attempts: u32) -> RetryPolicy { + RetryPolicy { + attempts, + base_backoff: Duration::from_millis(1), + max_backoff: Duration::from_millis(4), + } +} + +#[test] +fn delay_grows_with_the_attempt_number() { + let policy = RetryPolicy { + attempts: 5, + base_backoff: Duration::from_millis(100), + max_backoff: Duration::from_secs(30), + }; + + assert!(policy.delay(1) >= Duration::from_millis(50)); + assert!(policy.delay(1) <= Duration::from_millis(100)); + assert!(policy.delay(2) >= Duration::from_millis(100)); + assert!(policy.delay(2) <= Duration::from_millis(200)); + assert!(policy.delay(3) >= Duration::from_millis(200)); + assert!(policy.delay(3) <= Duration::from_millis(400)); +} + +#[test] +fn delay_never_exceeds_max_backoff() { + let policy = RetryPolicy { + attempts: 5, + base_backoff: Duration::from_millis(1000), + max_backoff: Duration::from_millis(2000), + }; + + for attempt in 1..=5 { + assert!(policy.delay(attempt) <= Duration::from_millis(2000)); + } +} + +#[tokio::test] +async fn first_attempt_success_logs_nothing() { + init_tracing_for_test(); + let logger = JobLogger::new(); + + let result: Result = + retry("Test op", &logger, &fast_policy(3), async |_| Ok(7)).await; + + assert_eq!(result.unwrap(), 7); + assert!(logger.into_entries().is_empty()); +} + +#[tokio::test] +async fn retries_until_success_and_logs_each_attempt() { + init_tracing_for_test(); + let logger = JobLogger::new(); + let calls = AtomicU32::new(0); + + let result: Result = + retry("Test op", &logger, &fast_policy(3), async |_| { + let n = calls.fetch_add(1, Ordering::SeqCst) + 1; + if n < 3 { + Err(anyhow::anyhow!("boom {n}")) + } else { + Ok(n) + } + }) + .await; + + assert_eq!(result.unwrap(), 3); + assert_eq!(calls.load(Ordering::SeqCst), 3); + + let entries = logger.into_entries(); + + let warns: Vec<_> = entries.iter().filter(|e| e.level == "warn").collect(); + assert_eq!(warns.len(), 2); + assert!(warns[0].message.starts_with("Test op attempt 1/3 failed: boom 1")); + assert!(warns[1].message.starts_with("Test op attempt 2/3 failed: boom 2")); + + let infos: Vec<_> = entries.iter().filter(|e| e.level == "info").collect(); + assert_eq!(infos.len(), 1); + assert_eq!(infos[0].message, "Test op succeeded on attempt 3/3"); +} + +#[tokio::test] +async fn exhausts_attempts_and_logs_a_single_error() { + init_tracing_for_test(); + let logger = JobLogger::new(); + let calls = AtomicU32::new(0); + + let result: Result<(), anyhow::Error> = + retry("Test op", &logger, &fast_policy(3), async |_| { + calls.fetch_add(1, Ordering::SeqCst); + Err(anyhow::anyhow!("always")) + }) + .await; + + assert!(result.is_err()); + assert_eq!(calls.load(Ordering::SeqCst), 3); + + let entries = logger.into_entries(); + assert_eq!(entries.iter().filter(|e| e.level == "warn").count(), 2); + assert_eq!(entries.iter().filter(|e| e.level == "error").count(), 1); + assert_eq!( + entries.iter().find(|e| e.level == "error").unwrap().message, + "Test op failed after 3 attempts: always" + ); +} + +#[tokio::test] +async fn closure_receives_the_attempt_number() { + init_tracing_for_test(); + let logger = JobLogger::new(); + let seen = Mutex::new(Vec::new()); + + let result: Result<(), anyhow::Error> = + retry("Test op", &logger, &fast_policy(3), async |attempt| { + seen.lock().unwrap().push(attempt); + Err(anyhow::anyhow!("nope")) + }) + .await; + + assert!(result.is_err()); + assert_eq!(*seen.lock().unwrap(), vec![1, 2, 3]); +} + +#[test] +fn config_defaults_are_within_the_documented_range() { + let policy = RetryPolicy::default(); + assert!(policy.attempts >= 3 && policy.attempts <= 5); + assert!(policy.base_backoff >= Duration::from_millis(100)); + assert!(policy.base_backoff <= Duration::from_millis(30_000)); +} diff --git a/src/utils/mod.rs b/src/utils/mod.rs index ebe8b00..a84e528 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -6,6 +6,7 @@ pub mod file; pub mod locks; pub mod logging; pub mod redis_client; +pub mod retry; pub mod stream; pub mod task_manager; pub mod text; diff --git a/src/utils/retry.rs b/src/utils/retry.rs new file mode 100644 index 0000000..8231e77 --- /dev/null +++ b/src/utils/retry.rs @@ -0,0 +1,73 @@ +use crate::services::backup::logger::JobLogger; +use crate::settings::CONFIG; +use rand::Rng; +use std::fmt::Display; +use std::time::Duration; + +pub struct RetryPolicy { + pub attempts: u32, + pub base_backoff: Duration, + pub max_backoff: Duration, +} + +impl Default for RetryPolicy { + fn default() -> Self { + Self { + attempts: CONFIG.retry_attempts, + base_backoff: Duration::from_millis(CONFIG.retry_backoff_ms), + max_backoff: Duration::from_secs(30), + } + } +} + +impl RetryPolicy { + pub(crate) fn delay(&self, attempt: u32) -> Duration { + let exp = self.base_backoff.saturating_mul(1u32 << (attempt - 1).min(16)); + let capped = exp.min(self.max_backoff); + let half = capped / 2; + let jitter = rand::rng().random_range(0..=half.as_millis() as u64); + + half + Duration::from_millis(jitter) + } +} + +pub async fn retry( + op: &str, + logger: &JobLogger, + policy: &RetryPolicy, + mut f: F, +) -> Result +where + F: AsyncFnMut(u32) -> Result, + E: Display, +{ + let total = policy.attempts; + let mut attempt = 1; + + loop { + match f(attempt).await { + Ok(v) => { + if attempt > 1 { + logger.log("info", format!("{op} succeeded on attempt {attempt}/{total}")); + } + return Ok(v); + } + Err(e) if attempt < total => { + let delay = policy.delay(attempt); + logger.log( + "warn", + format!( + "{op} attempt {attempt}/{total} failed: {e} — retrying in {}ms", + delay.as_millis() + ), + ); + tokio::time::sleep(delay).await; + attempt += 1; + } + Err(e) => { + logger.log("error", format!("{op} failed after {total} attempts: {e}")); + return Err(e); + } + } + } +} From b0da2e40a3f4a099e19d2f4f79559e6a7f651f1c Mon Sep 17 00:00:00 2001 From: charles-gauthereau Date: Thu, 27 Aug 2026 18:39:41 +0200 Subject: [PATCH 2/7] feat: retry the database backup Each attempt now dumps into its own tmp_path/attempt-{n} directory, which is removed when the attempt fails. Without the per-attempt directory pg_dump -Fd would refuse every retry, because it will not write into a directory a previous attempt left behind; removing it on failure keeps peak disk at one attempt's artifacts rather than five. A backup blocked by a concurrent job is retried before surfacing the same backup_already_in_progress code, since FileLock reports it as an ordinary error and the combinator cannot tell it apart. Reshapes retry()'s bound from the native AsyncFnMut sugar to the classic F: FnMut(u32) -> Fut, Fut: Future> + Send shape (the pattern tokio-retry and backoff both use). AsyncFnMut's produced future is a lifetime-quantified associated type (F::CallRefFuture<'_>) that cannot be named or bounded as Send on stable Rust, so wiring a retried call through it into a future that eventually gets polled inside tokio::spawn (dispatcher.rs, via execute_backup) made rustc's opaque-type Send inference fail with "implementation of Send is not general enough" at the spawn site, several call layers away from the actual retry call. Naming Fut as its own type parameter lets Send be asserted on it directly instead, which resolves cleanly. The combinator's control flow, log messages, and formats are unchanged; only the bound and the call sites' closure shape (async |x| { } becomes |x| async { }, with shared references bound outside a move closure so the inner async move block only moves Copy references, not the originals) are affected. --- src/services/backup/runner.rs | 27 ++++++++- src/tests/services/backup_runner_tests.rs | 68 +++++++++++++++++++++++ src/tests/services/mod.rs | 1 + src/tests/utils/retry_tests.rs | 11 ++-- src/utils/retry.rs | 9 ++- 5 files changed, 107 insertions(+), 9 deletions(-) create mode 100644 src/tests/services/backup_runner_tests.rs diff --git a/src/services/backup/runner.rs b/src/services/backup/runner.rs index 52aa5d5..e9543a5 100644 --- a/src/services/backup/runner.rs +++ b/src/services/backup/runner.rs @@ -4,6 +4,7 @@ use super::service::BackupService; use crate::domain::factory::DatabaseFactory; use crate::services::config::DatabaseConfig; +use crate::utils::retry::{RetryPolicy, retry}; use anyhow::Result; use std::path::Path; @@ -39,7 +40,31 @@ impl BackupService { }); } - match db.backup(tmp_path, Arc::clone(&logger)).await { + let policy = RetryPolicy::default(); + + let db_ref = &db; + let logger_ref = &logger; + + let outcome = retry("Database backup", &logger, &policy, move |attempt| { + let dir = tmp_path.join(format!("attempt-{attempt}")); + + async move { + if let Err(e) = tokio::fs::create_dir_all(&dir).await { + return Err(anyhow::Error::from(e)); + } + + match db_ref.backup(&dir, Arc::clone(logger_ref)).await { + Ok(f) => Ok(f), + Err(e) => { + let _ = tokio::fs::remove_dir_all(&dir).await; + Err(e) + } + } + } + }) + .await; + + match outcome { Ok(file) => Ok(BackupResult { generated_id, db_type, diff --git a/src/tests/services/backup_runner_tests.rs b/src/tests/services/backup_runner_tests.rs new file mode 100644 index 0000000..883a4c6 --- /dev/null +++ b/src/tests/services/backup_runner_tests.rs @@ -0,0 +1,68 @@ +use crate::services::backup::BackupService; +use crate::services::backup::logger::JobLogger; +use crate::services::config::{DatabaseConfig, DbType}; +use crate::tests::init_tracing_for_test; + +use std::collections::HashMap; +use std::sync::Arc; +use tempfile::TempDir; + +fn sqlite_config(path: &str) -> DatabaseConfig { + DatabaseConfig { + name: "retry-test".to_string(), + database: String::new(), + db_type: DbType::Sqlite, + username: String::new(), + password: String::new(), + port: 0, + host: String::new(), + generated_id: "retry-test-gen".to_string(), + path: path.to_string(), + max_packet_size: String::new(), + volume_name: String::new(), + container_name: None, + options: HashMap::new(), + } +} + +#[tokio::test] +async fn a_failing_backup_is_retried_and_leaves_no_attempt_directory() { + init_tracing_for_test(); + + let temp_dir = TempDir::new().unwrap(); + let tmp_path = temp_dir.path(); + let logger = Arc::new(JobLogger::new()); + + let cfg = sqlite_config("/nonexistent/definitely-not-here.sqlite"); + + let result = BackupService::run(cfg, tmp_path, Arc::clone(&logger)) + .await + .unwrap(); + + assert_eq!(result.status, "failed"); + assert!(result.backup_file.is_none()); + + let entries = Arc::try_unwrap(logger).unwrap().into_entries(); + assert_eq!( + entries.iter().filter(|e| e.level == "warn").count(), + 2, + "expected one warn per non-final failed attempt" + ); + assert!( + entries + .iter() + .any(|e| e.level == "error" && e.message.starts_with("Database backup failed after 3 attempts")), + "expected a single terminal error naming the attempt count" + ); + + let leftovers: Vec<_> = std::fs::read_dir(tmp_path) + .unwrap() + .filter_map(|e| e.ok()) + .filter(|e| e.file_name().to_string_lossy().starts_with("attempt-")) + .collect(); + assert!( + leftovers.is_empty(), + "failed attempt directories must be cleaned up, found {:?}", + leftovers.iter().map(|e| e.file_name()).collect::>() + ); +} diff --git a/src/tests/services/mod.rs b/src/tests/services/mod.rs index b3a352c..9f6f9b6 100644 --- a/src/tests/services/mod.rs +++ b/src/tests/services/mod.rs @@ -1,4 +1,5 @@ mod api_models_tests; +mod backup_runner_tests; mod backup_uploader_tests; mod config_tests; mod dashboard_config_tests; diff --git a/src/tests/utils/retry_tests.rs b/src/tests/utils/retry_tests.rs index 115790c..fe4279a 100644 --- a/src/tests/utils/retry_tests.rs +++ b/src/tests/utils/retry_tests.rs @@ -49,7 +49,7 @@ async fn first_attempt_success_logs_nothing() { let logger = JobLogger::new(); let result: Result = - retry("Test op", &logger, &fast_policy(3), async |_| Ok(7)).await; + retry("Test op", &logger, &fast_policy(3), |_| async { Ok(7) }).await; assert_eq!(result.unwrap(), 7); assert!(logger.into_entries().is_empty()); @@ -62,7 +62,7 @@ async fn retries_until_success_and_logs_each_attempt() { let calls = AtomicU32::new(0); let result: Result = - retry("Test op", &logger, &fast_policy(3), async |_| { + retry("Test op", &logger, &fast_policy(3), |_| async { let n = calls.fetch_add(1, Ordering::SeqCst) + 1; if n < 3 { Err(anyhow::anyhow!("boom {n}")) @@ -94,7 +94,7 @@ async fn exhausts_attempts_and_logs_a_single_error() { let calls = AtomicU32::new(0); let result: Result<(), anyhow::Error> = - retry("Test op", &logger, &fast_policy(3), async |_| { + retry("Test op", &logger, &fast_policy(3), |_| async { calls.fetch_add(1, Ordering::SeqCst); Err(anyhow::anyhow!("always")) }) @@ -117,10 +117,11 @@ async fn closure_receives_the_attempt_number() { init_tracing_for_test(); let logger = JobLogger::new(); let seen = Mutex::new(Vec::new()); + let seen_ref = &seen; let result: Result<(), anyhow::Error> = - retry("Test op", &logger, &fast_policy(3), async |attempt| { - seen.lock().unwrap().push(attempt); + retry("Test op", &logger, &fast_policy(3), move |attempt| async move { + seen_ref.lock().unwrap().push(attempt); Err(anyhow::anyhow!("nope")) }) .await; diff --git a/src/utils/retry.rs b/src/utils/retry.rs index 8231e77..c4df2c6 100644 --- a/src/utils/retry.rs +++ b/src/utils/retry.rs @@ -2,6 +2,7 @@ use crate::services::backup::logger::JobLogger; use crate::settings::CONFIG; use rand::Rng; use std::fmt::Display; +use std::future::Future; use std::time::Duration; pub struct RetryPolicy { @@ -31,15 +32,17 @@ impl RetryPolicy { } } -pub async fn retry( +pub async fn retry( op: &str, logger: &JobLogger, policy: &RetryPolicy, mut f: F, ) -> Result where - F: AsyncFnMut(u32) -> Result, - E: Display, + F: FnMut(u32) -> Fut, + Fut: Future> + Send, + T: Send, + E: Display + Send, { let total = policy.attempts; let mut attempt = 1; From 5298d82576ea13454346154acf5505384858c13e Mon Sep 17 00:00:00 2001 From: charles-gauthereau Date: Thu, 27 Aug 2026 18:53:06 +0200 Subject: [PATCH 3/7] feat: retry storage uploads Wraps provider.upload in the retry combinator. No provider changes are needed: each one builds its upload stream from the file on disk inside upload(), so every attempt gets a fresh handle and a fresh nonce. Result is collapsed with an or-pattern so the last attempt's error and metadata survive into the existing failure branch. A missing backup file short-circuits into Err rather than returning early, which skips the retry without skipping the backup_upload_status(failed) call that closes the server-side record. backup_upload_init and backup_upload_status are left unwrapped; they are control-plane calls, not storage uploads. --- src/services/backup/models.rs | 7 ++ src/services/backup/uploader.rs | 47 +++++++-- src/tests/services/backup_uploader_tests.rs | 110 ++++++++++++++++++++ 3 files changed, 155 insertions(+), 9 deletions(-) diff --git a/src/services/backup/models.rs b/src/services/backup/models.rs index def8787..ba5d88d 100644 --- a/src/services/backup/models.rs +++ b/src/services/backup/models.rs @@ -1,6 +1,7 @@ #![allow(dead_code)] use crate::services::config::DbType; +use std::fmt::{self, Display, Formatter}; use std::path::PathBuf; #[derive(Debug, Clone)] @@ -20,3 +21,9 @@ pub struct UploadResult { pub remote_file_path: Option, pub total_size: Option, } + +impl Display for UploadResult { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.error.as_deref().unwrap_or("unknown error")) + } +} diff --git a/src/services/backup/uploader.rs b/src/services/backup/uploader.rs index be2ba9e..0d4609f 100644 --- a/src/services/backup/uploader.rs +++ b/src/services/backup/uploader.rs @@ -4,6 +4,7 @@ use super::service::BackupService; use crate::services::api::models::agent::status::DatabaseStorage; use crate::services::storage; use crate::utils::common::BackupMethod; +use crate::utils::retry::{RetryPolicy, retry}; use anyhow::{Result, bail}; use futures::future::join_all; use std::sync::Arc; @@ -97,16 +98,44 @@ impl BackupService { /* STORAGE UPLOAD */ - let upload_result = provider - .upload( - ctx_clone.clone(), - result_clone, - method, - &storage, - Some(encrypt), - &backup_storage_id, + let policy = RetryPolicy::default(); + + let attempt_result = if result_clone.backup_file.is_none() { + logger_clone.log("error", format!("Missing backup file for storage {}", storage_id)); + + Err(UploadResult { + storage_id: storage_id.clone(), + success: false, + error: Some("Missing backup file path".into()), + remote_file_path: None, + total_size: None, + }) + } else { + retry( + &format!("Upload to storage {storage_id}"), + &logger_clone, + &policy, + |_| async { + let r = provider + .upload( + ctx_clone.clone(), + result_clone.clone(), + method, + &storage, + Some(encrypt), + &backup_storage_id, + ) + .await; + + if r.success { Ok(r) } else { Err(r) } + }, ) - .await; + .await + }; + + let upload_result = match attempt_result { + Ok(r) | Err(r) => r, + }; let status = if upload_result.success { "success" } else { "failed" }; diff --git a/src/tests/services/backup_uploader_tests.rs b/src/tests/services/backup_uploader_tests.rs index 886f578..f23b99f 100644 --- a/src/tests/services/backup_uploader_tests.rs +++ b/src/tests/services/backup_uploader_tests.rs @@ -14,7 +14,9 @@ use crate::utils::common::BackupMethod; use crate::utils::edge_key::EdgeKey; use serde_json::json; +use std::io::Write; use std::sync::Arc; +use tempfile::NamedTempFile; use wiremock::matchers::{body_partial_json, method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -93,3 +95,111 @@ async fn failed_upload_reports_failed_status_to_server() { // MockServer drop verifies both `.expect(1)` mounts were hit — including the "failed" PATCH. } + +#[tokio::test] +async fn a_failing_upload_is_retried_until_it_succeeds() { + init_tracing_for_test(); + let server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/agent/agent-1/backup/upload/init")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "message": "ok", + "backupStorage": { "id": "bs-1" } + }))) + .expect(1) + .mount(&server) + .await; + + Mock::given(method("POST")) + .and(path("/tus/files")) + .respond_with(ResponseTemplate::new(500)) + .up_to_n_times(2) + .with_priority(1) + .expect(2) + .mount(&server) + .await; + + Mock::given(method("POST")) + .and(path("/tus/files")) + .respond_with( + ResponseTemplate::new(201) + .insert_header("Location", format!("{}/tus/files/upload-1", server.uri()).as_str()), + ) + .with_priority(2) + .expect(1) + .mount(&server) + .await; + + Mock::given(method("PATCH")) + .and(path("/tus/files/upload-1")) + .respond_with(ResponseTemplate::new(204)) + .mount(&server) + .await; + + Mock::given(method("PATCH")) + .and(path("/agent/agent-1/backup/upload/status")) + .and(body_partial_json(json!({ "status": "success" }))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "message": "ok", + "backupStorage": { "id": "bs-1" } + }))) + .expect(1) + .mount(&server) + .await; + + let mut backup_file = NamedTempFile::new().unwrap(); + backup_file.write_all(b"portabase-retry-test-payload").unwrap(); + backup_file.flush().unwrap(); + + let ctx = Context { + edge_key: EdgeKey { + server_url: server.uri(), + agent_id: "agent-1".to_string(), + master_key_b64: String::new(), + }, + api: ApiClient::new(server.uri()), + }; + + let service = BackupService::new(Arc::new(ctx)); + + let result = BackupResult { + generated_id: "gen-1".to_string(), + db_type: DbType::Postgresql, + status: "success".to_string(), + backup_file: Some(backup_file.path().to_path_buf()), + code: None, + }; + + let storage: DatabaseStorage = serde_json::from_value(json!({ + "id": "storage-1", + "provider": "local", + "config": {} + })) + .unwrap(); + + let backup_id = "backup-1".to_string(); + let logger = Arc::new(JobLogger::new()); + + let results = service + .upload( + result, + BackupMethod::Manual, + vec![storage], + false, + &backup_id, + Arc::clone(&logger), + ) + .await + .unwrap(); + + assert_eq!(results.len(), 1); + assert!(results[0].success); + + let entries = Arc::try_unwrap(logger).unwrap().into_entries(); + assert_eq!(entries.iter().filter(|e| e.level == "warn").count(), 2); + assert!( + entries.iter().any(|e| e.message + == "Upload to storage storage-1 succeeded on attempt 3/3") + ); +} From b6a120fcbfab89cb0ae41212a83f780cfe13095e Mon Sep 17 00:00:00 2001 From: charles-gauthereau Date: Thu, 27 Aug 2026 19:01:10 +0200 Subject: [PATCH 4/7] feat: retry the restore backup download Extracts the download body to download_once and makes download_backup a retry wrapper around it. This path had no retry at all before, so a single dropped connection failed the whole restore job. Retrying is safe because File::create truncates and the target filename is derived from Content-Disposition or the URL, so it is stable across attempts. There is no Range resume: a download that fails at 90% starts over. --- src/services/restore/downloader.rs | 23 +++++++ src/tests/services/mod.rs | 1 + .../services/restore_downloader_tests.rs | 64 +++++++++++++++++++ 3 files changed, 88 insertions(+) create mode 100644 src/tests/services/restore_downloader_tests.rs diff --git a/src/services/restore/downloader.rs b/src/services/restore/downloader.rs index a154160..da65f96 100644 --- a/src/services/restore/downloader.rs +++ b/src/services/restore/downloader.rs @@ -8,6 +8,7 @@ use std::sync::Arc; use std::time::Instant; use tokio::io::AsyncWriteExt; use crate::services::backup::logger::JobLogger; +use crate::utils::retry::{RetryPolicy, retry}; fn human_size(bytes: u64) -> String { if bytes >= 1024 * 1024 { @@ -26,6 +27,28 @@ impl RestoreService { tmp_path: &Path, logger: Arc, expected_size: Option, + ) -> Result { + let policy = RetryPolicy::default(); + + let logger_ref = &logger; + + retry("Backup download", &logger, &policy, move |_| { + let expected = expected_size.clone(); + + async move { + self.download_once(file_url, tmp_path, Arc::clone(logger_ref), expected) + .await + } + }) + .await + } + + pub async fn download_once( + &self, + file_url: &str, + tmp_path: &Path, + logger: Arc, + expected_size: Option, ) -> Result { logger.log("info", "Start downloading backup archive".to_string()); diff --git a/src/tests/services/mod.rs b/src/tests/services/mod.rs index 9f6f9b6..aa4be3a 100644 --- a/src/tests/services/mod.rs +++ b/src/tests/services/mod.rs @@ -3,3 +3,4 @@ mod backup_runner_tests; mod backup_uploader_tests; mod config_tests; mod dashboard_config_tests; +mod restore_downloader_tests; diff --git a/src/tests/services/restore_downloader_tests.rs b/src/tests/services/restore_downloader_tests.rs new file mode 100644 index 0000000..905545e --- /dev/null +++ b/src/tests/services/restore_downloader_tests.rs @@ -0,0 +1,64 @@ +use crate::core::context::Context; +use crate::services::api::ApiClient; +use crate::services::backup::logger::JobLogger; +use crate::services::restore::RestoreService; +use crate::tests::init_tracing_for_test; +use crate::utils::edge_key::EdgeKey; + +use std::sync::Arc; +use tempfile::TempDir; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +#[tokio::test] +async fn a_failing_download_is_retried_until_it_succeeds() { + init_tracing_for_test(); + let server = MockServer::start().await; + + Mock::given(method("GET")) + .and(path("/backups/archive.tar.gz")) + .respond_with(ResponseTemplate::new(503)) + .up_to_n_times(2) + .with_priority(1) + .expect(2) + .mount(&server) + .await; + + Mock::given(method("GET")) + .and(path("/backups/archive.tar.gz")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"portabase-archive".to_vec())) + .with_priority(2) + .expect(1) + .mount(&server) + .await; + + let ctx = Context { + edge_key: EdgeKey { + server_url: server.uri(), + agent_id: "agent-1".to_string(), + master_key_b64: String::new(), + }, + api: ApiClient::new(server.uri()), + }; + + let service = RestoreService::new(Arc::new(ctx)); + + let temp_dir = TempDir::new().unwrap(); + let logger = Arc::new(JobLogger::new()); + let url = format!("{}/backups/archive.tar.gz", server.uri()); + + let downloaded = service + .download_backup(&url, temp_dir.path(), Arc::clone(&logger), None) + .await + .unwrap(); + + assert_eq!(std::fs::read(&downloaded).unwrap(), b"portabase-archive"); + + let entries = Arc::try_unwrap(logger).unwrap().into_entries(); + assert_eq!(entries.iter().filter(|e| e.level == "warn").count(), 2); + assert!( + entries + .iter() + .any(|e| e.message == "Backup download succeeded on attempt 3/3") + ); +} From 87f33af772ada1f368cdbc6755aaf1b9b2a1977f Mon Sep 17 00:00:00 2001 From: charles-gauthereau Date: Thu, 27 Aug 2026 22:46:49 +0200 Subject: [PATCH 5/7] fix: wire retry env vars into helm configmap, dedupe terminal retry log helm/templates/env-configmap.yaml never listed RETRY_ATTEMPTS and RETRY_BACKOFF_MS even though values.yaml gained them, so --set env.RETRY_ATTEMPTS=N was silently ignored by Kubernetes deployments. Add both keys in the same explicit style as the existing entries. src/utils/retry.rs logged its own "failed after N attempts" error on exhaustion, on top of the terminal log each call site already writes, producing two error entries per failure. Worse, it changed a log level: FileLock::acquire's "backup_already_in_progress" bails through the combinator, which now logged it as error before runner.rs got a chance to reclassify it as the routine warn it always was. A manual backup colliding with a scheduled one would show up as a hard error on the dashboard instead of the harmless warn it used to be, breaking the "fails exactly as it does today" guarantee for job records. Drop the combinator's terminal error log and give download_backup its own terminal error log so all three call sites (runner, uploader, downloader) own their failure logging uniformly. Update the two tests that asserted the removed message to assert the new behavior instead. --- helm/templates/env-configmap.yaml | 4 +++- src/services/restore/downloader.rs | 10 ++++++++-- src/tests/services/backup_runner_tests.rs | 4 ++-- src/tests/utils/retry_tests.rs | 8 ++------ src/utils/retry.rs | 1 - 5 files changed, 15 insertions(+), 12 deletions(-) diff --git a/helm/templates/env-configmap.yaml b/helm/templates/env-configmap.yaml index 9afc8e9..a7894ed 100644 --- a/helm/templates/env-configmap.yaml +++ b/helm/templates/env-configmap.yaml @@ -7,4 +7,6 @@ data: TZ: {{ .Values.env.TZ | quote }} POLLING: {{ .Values.env.POLLING | quote }} APP_ENV: {{ .Values.env.APP_ENV | quote }} - LOG: {{ .Values.env.LOG | quote }} \ No newline at end of file + LOG: {{ .Values.env.LOG | quote }} + RETRY_ATTEMPTS: {{ .Values.env.RETRY_ATTEMPTS | quote }} + RETRY_BACKOFF_MS: {{ .Values.env.RETRY_BACKOFF_MS | quote }} diff --git a/src/services/restore/downloader.rs b/src/services/restore/downloader.rs index da65f96..4efaaf9 100644 --- a/src/services/restore/downloader.rs +++ b/src/services/restore/downloader.rs @@ -32,7 +32,7 @@ impl RestoreService { let logger_ref = &logger; - retry("Backup download", &logger, &policy, move |_| { + let outcome = retry("Backup download", &logger, &policy, move |_| { let expected = expected_size.clone(); async move { @@ -40,7 +40,13 @@ impl RestoreService { .await } }) - .await + .await; + + if let Err(e) = &outcome { + logger.log("error", format!("Download failed: {e}")); + } + + outcome } pub async fn download_once( diff --git a/src/tests/services/backup_runner_tests.rs b/src/tests/services/backup_runner_tests.rs index 883a4c6..6473e10 100644 --- a/src/tests/services/backup_runner_tests.rs +++ b/src/tests/services/backup_runner_tests.rs @@ -51,8 +51,8 @@ async fn a_failing_backup_is_retried_and_leaves_no_attempt_directory() { assert!( entries .iter() - .any(|e| e.level == "error" && e.message.starts_with("Database backup failed after 3 attempts")), - "expected a single terminal error naming the attempt count" + .any(|e| e.level == "error" && e.message.starts_with("Backup failed:")), + "expected a single terminal error from the runner" ); let leftovers: Vec<_> = std::fs::read_dir(tmp_path) diff --git a/src/tests/utils/retry_tests.rs b/src/tests/utils/retry_tests.rs index fe4279a..e991ca3 100644 --- a/src/tests/utils/retry_tests.rs +++ b/src/tests/utils/retry_tests.rs @@ -88,7 +88,7 @@ async fn retries_until_success_and_logs_each_attempt() { } #[tokio::test] -async fn exhausts_attempts_and_logs_a_single_error() { +async fn exhausts_attempts_and_logs_no_terminal_error() { init_tracing_for_test(); let logger = JobLogger::new(); let calls = AtomicU32::new(0); @@ -105,11 +105,7 @@ async fn exhausts_attempts_and_logs_a_single_error() { let entries = logger.into_entries(); assert_eq!(entries.iter().filter(|e| e.level == "warn").count(), 2); - assert_eq!(entries.iter().filter(|e| e.level == "error").count(), 1); - assert_eq!( - entries.iter().find(|e| e.level == "error").unwrap().message, - "Test op failed after 3 attempts: always" - ); + assert_eq!(entries.iter().filter(|e| e.level == "error").count(), 0); } #[tokio::test] diff --git a/src/utils/retry.rs b/src/utils/retry.rs index c4df2c6..b2c19f4 100644 --- a/src/utils/retry.rs +++ b/src/utils/retry.rs @@ -68,7 +68,6 @@ where attempt += 1; } Err(e) => { - logger.log("error", format!("{op} failed after {total} attempts: {e}")); return Err(e); } } From 42c3c5945a0e81fd58fcc02b1cfdddbf483340e4 Mon Sep 17 00:00:00 2001 From: charles-gauthereau Date: Fri, 28 Aug 2026 17:06:23 +0200 Subject: [PATCH 6/7] fix: refactoring --- src/services/backup/uploader.rs | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/services/backup/uploader.rs b/src/services/backup/uploader.rs index 0d4609f..64b6325 100644 --- a/src/services/backup/uploader.rs +++ b/src/services/backup/uploader.rs @@ -146,8 +146,6 @@ impl BackupService { upload_result.error.as_deref().unwrap_or("unknown error") )); - // `backup_upload_init` opened a per-storage record; close it as "failed" - // so the server is notified of the failure (no path/size on this path). if let Err(err) = ctx_clone .api .backup_upload_status( From 99f1ef20812ba5e129f6d17ab423757ce1296558 Mon Sep 17 00:00:00 2001 From: charles-gauthereau Date: Fri, 28 Aug 2026 17:25:30 +0200 Subject: [PATCH 7/7] fix: refactoring --- docker-compose.yml | 4 ++-- src/utils/retry.rs | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 8859988..e84f924 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -9,7 +9,7 @@ services: - .:/app - cargo-registry:/usr/local/cargo/registry - cargo-git:/usr/local/cargo/git -# - ./databases.json:/config/config.json + - ./databases.json:/config/config.json #- ./databases.toml:/config/config.toml - /var/run/docker.sock:/var/run/docker.sock # - cargo-target:/app/target @@ -21,7 +21,7 @@ services: LOG: debug TZ: "Europe/Paris" # TMPDIR: /scratch - EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiZjlkZjhiNWYtM2I0MC00NWM3LWI3N2UtYzY4NzQ1YmU2NjMwIiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ==" + EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiY2UxNjRiZDItZGZkMy00YzY4LThlZGItNmQ3OTczODAzZWEyIiwibWFzdGVyS2V5QjY0IjoiMUh0djdtWCtYVkJxL0IzUEV2WDlZZjlQeUdVZW5oRHlXemo5THRqNW90WT0ifQ==" #CHUNK_SIZE_MB: "1" #POOLING: 1 #RETRY_ATTEMPTS: 3 diff --git a/src/utils/retry.rs b/src/utils/retry.rs index b2c19f4..9b6ce6c 100644 --- a/src/utils/retry.rs +++ b/src/utils/retry.rs @@ -60,7 +60,7 @@ where logger.log( "warn", format!( - "{op} attempt {attempt}/{total} failed: {e} — retrying in {}ms", + "{op} attempt {attempt}/{total} failed: {e} - retrying in {}ms", delay.as_millis() ), );