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

@@ -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"

helm/values.yaml

@@ -11,6 +11,8 @@ env:
   POLLING: "5"
   APP_ENV: "production"
   LOG: "info"
+  RETRY_ATTEMPTS: "3"
+  RETRY_BACKOFF_MS: "1000"
 
 resources:
   limits:

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::<u32>()
+            .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::<u64>()
+            .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,
         }
     }
 }

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;

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<u32, anyhow::Error> =
+        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<u32, anyhow::Error> =
+        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));
+}

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;

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<T, E, F>(
+    op: &str,
+    logger: &JobLogger,
+    policy: &RetryPolicy,
+    mut f: F,
+) -> Result<T, E>
+where
+    F: AsyncFnMut(u32) -> Result<T, E>,
+    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);
+            }
+        }
+    }
+}