diff --git a/docker-compose.yml b/docker-compose.yml index d63d415..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,9 +21,11 @@ services: LOG: debug TZ: "Europe/Paris" # TMPDIR: /scratch - EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiZjlkZjhiNWYtM2I0MC00NWM3LWI3N2UtYzY4NzQ1YmU2NjMwIiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ==" + EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiY2UxNjRiZDItZGZkMy00YzY4LThlZGItNmQ3OTczODAzZWEyIiwibWFzdGVyS2V5QjY0IjoiMUh0djdtWCtYVkJxL0IzUEV2WDlZZjlQeUdVZW5oRHlXemo5THRqNW90WT0ifQ==" #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/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/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/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/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/services/backup/uploader.rs b/src/services/backup/uploader.rs index be2ba9e..64b6325 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" }; @@ -117,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( diff --git a/src/services/restore/downloader.rs b/src/services/restore/downloader.rs index a154160..4efaaf9 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,34 @@ impl RestoreService { tmp_path: &Path, logger: Arc, expected_size: Option, + ) -> Result { + let policy = RetryPolicy::default(); + + let logger_ref = &logger; + + let outcome = 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; + + if let Err(e) = &outcome { + logger.log("error", format!("Download failed: {e}")); + } + + outcome + } + + 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/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/services/backup_runner_tests.rs b/src/tests/services/backup_runner_tests.rs new file mode 100644 index 0000000..6473e10 --- /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("Backup failed:")), + "expected a single terminal error from the runner" + ); + + 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/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") + ); +} diff --git a/src/tests/services/mod.rs b/src/tests/services/mod.rs index b3a352c..aa4be3a 100644 --- a/src/tests/services/mod.rs +++ b/src/tests/services/mod.rs @@ -1,4 +1,6 @@ mod api_models_tests; +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") + ); +} 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..e991ca3 --- /dev/null +++ b/src/tests/utils/retry_tests.rs @@ -0,0 +1,135 @@ +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_no_terminal_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(), 0); +} + +#[tokio::test] +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), move |attempt| async move { + seen_ref.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..9b6ce6c --- /dev/null +++ b/src/utils/retry.rs @@ -0,0 +1,75 @@ +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 { + 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: FnMut(u32) -> Fut, + Fut: Future> + Send, + T: Send, + E: Display + Send, +{ + 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) => { + return Err(e); + } + } + } +}