From 55e20d48e7984fb7019265927cc8d308f841c01f Mon Sep 17 00:00:00 2001 From: charles-gauthereau Date: Thu, 27 Aug 2026 18:13:17 +0200 Subject: [PATCH] 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); + } + } + } +}