From c2a15f52149ce89ca3833f23ffeebbcc572ab516 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 13 Aug 2026 08:06:21 +0800 Subject: [PATCH] refactor(utils): add shared retry_with_backoff and migrate target_descriptor (#6026) --- crates/utils/src/retry.rs | 131 ++++++++++++++++++ .../src/admin/handlers/target_descriptor.rs | 32 +---- 2 files changed, 135 insertions(+), 28 deletions(-) diff --git a/crates/utils/src/retry.rs b/crates/utils/src/retry.rs index 82103049f..c5fd5ef85 100644 --- a/crates/utils/src/retry.rs +++ b/crates/utils/src/retry.rs @@ -101,6 +101,56 @@ impl Stream for RetryTimer { } } +/// Drives `operation` with capped, jittered exponential backoff, returning the +/// first success or the last error once `max_attempts` attempts are exhausted. +/// +/// The sleep before retry `n` (1-based) is `min(base_delay * 2^(n-1), max_delay)`, +/// reduced by up to half through a cheap clock-derived jitter so concurrent +/// retriers decorrelate — the same backoff shape as [`RetryTimer`] without +/// needing a caller-supplied random seed or the Stream API. `max_attempts` is +/// clamped to at least 1. +pub async fn retry_with_backoff( + mut operation: F, + max_attempts: usize, + base_delay: Duration, + max_delay: Duration, +) -> Result +where + F: FnMut() -> Fut, + Fut: std::future::Future>, +{ + let max_attempts = max_attempts.max(1); + let mut last_err = None; + + for attempt in 0..max_attempts { + match operation().await { + Ok(value) => return Ok(value), + Err(err) => { + last_err = Some(err); + if attempt + 1 < max_attempts { + // Cap the shift so the multiplier cannot overflow; the cap + // below bounds the result anyway. + let exp = base_delay.saturating_mul(1u32 << attempt.min(16)); + let mut sleep_duration = exp.min(max_delay); + // Up to 50% reduction, derived from the clock's sub-second + // nanoseconds — cheap decorrelation without a rand dependency. + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.subsec_nanos()) + .unwrap_or(0); + let reduction_percent = u64::from(nanos % 50); + let sleep_ms = sleep_duration.as_millis() as u64; + let jittered_ms = sleep_ms.saturating_sub(sleep_ms * reduction_percent / 100).max(1); + sleep_duration = Duration::from_millis(jittered_ms); + tokio::time::sleep(sleep_duration).await; + } + } + } + } + + Err(last_err.expect("max_attempts is clamped to at least 1, so at least one attempt ran")) +} + static RETRYABLE_S3CODES: LazyLock> = LazyLock::new(|| { vec![ "RequestError".to_string(), @@ -241,6 +291,87 @@ mod tests { assert!(!is_s3code_in_message_retryable("")); } + #[tokio::test] + async fn retry_with_backoff_returns_first_success_without_retrying() { + let mut calls = 0; + let result: Result = retry_with_backoff( + || { + calls += 1; + async { Ok(42) } + }, + 3, + Duration::from_millis(1), + Duration::from_millis(2), + ) + .await; + + assert_eq!(result.expect("first attempt succeeds"), 42); + assert_eq!(calls, 1, "a success must not trigger further attempts"); + } + + #[tokio::test] + async fn retry_with_backoff_retries_until_success() { + let mut calls = 0; + let result: Result = retry_with_backoff( + || { + calls += 1; + let attempt = calls; + async move { + if attempt < 3 { + Err(std::io::Error::other("transient")) + } else { + Ok(7) + } + } + }, + 5, + Duration::from_millis(1), + Duration::from_millis(2), + ) + .await; + + assert_eq!(result.expect("third attempt succeeds"), 7); + assert_eq!(calls, 3); + } + + #[tokio::test] + async fn retry_with_backoff_exhausts_attempts_and_returns_last_error() { + let mut calls = 0; + let result: Result<(), std::io::Error> = retry_with_backoff( + || { + calls += 1; + let attempt = calls; + async move { Err(std::io::Error::other(format!("attempt {attempt}"))) } + }, + 3, + Duration::from_millis(1), + Duration::from_millis(2), + ) + .await; + + let err = result.expect_err("all attempts fail"); + assert_eq!(err.to_string(), "attempt 3", "the LAST error must be returned"); + assert_eq!(calls, 3); + } + + #[tokio::test] + async fn retry_with_backoff_clamps_zero_attempts_to_one() { + let mut calls = 0; + let result: Result<(), std::io::Error> = retry_with_backoff( + || { + calls += 1; + async { Err(std::io::Error::other("always")) } + }, + 0, + Duration::from_millis(1), + Duration::from_millis(2), + ) + .await; + + assert!(result.is_err()); + assert_eq!(calls, 1, "zero attempts clamps to a single attempt instead of panicking"); + } + #[test] fn is_s3code_in_message_retryable_is_case_sensitive() { // Pin the contract: a backend that down-cases its error diff --git a/rustfs/src/admin/handlers/target_descriptor.rs b/rustfs/src/admin/handlers/target_descriptor.rs index dab02d4bc..ac0e5681e 100644 --- a/rustfs/src/admin/handlers/target_descriptor.rs +++ b/rustfs/src/admin/handlers/target_descriptor.rs @@ -38,10 +38,10 @@ use rustfs_utils::egress::OutboundPolicy; use s3s::{Body, S3Response, S3Result, header::CONTENT_TYPE, s3_error}; use serde::Serialize; use std::collections::{HashMap, HashSet}; -use std::io::{Error, ErrorKind}; +use std::io::ErrorKind; use std::path::Path; use std::sync::Arc; -use tokio::time::{Duration, sleep, timeout}; +use tokio::time::{Duration, timeout}; use url::Url; pub(crate) type EndpointKey = (String, String); @@ -535,10 +535,11 @@ pub(crate) async fn validate_queue_dir(queue_dir: &str) -> S3Result<()> { if !Path::new(queue_dir).is_absolute() { return Err(s3_error!(InvalidArgument, "queue_dir must be an absolute path")); } - retry_with_backoff( + rustfs_utils::retry::retry_with_backoff( || async { tokio::fs::metadata(queue_dir).await.map(|_| ()) }, 3, Duration::from_millis(100), + rustfs_utils::retry::DEFAULT_RETRY_CAP, ) .await .map_err(|e| match e.kind() { @@ -665,31 +666,6 @@ fn collect_endpoint_snapshot(specs: &[AdminTargetSpec], route_prefix: &str, conf }) } -async fn retry_with_backoff(mut operation: F, max_attempts: usize, base_delay: Duration) -> Result -where - F: FnMut() -> Fut, - Fut: std::future::Future>, -{ - let mut attempts = 0; - let mut delay = base_delay; - let mut last_err = None; - - while attempts < max_attempts { - match operation().await { - Ok(result) => return Ok(result), - Err(e) => { - last_err = Some(e); - attempts += 1; - if attempts < max_attempts { - sleep(delay).await; - delay = delay.saturating_mul(2); - } - } - } - } - Err(last_err.unwrap_or_else(|| Error::other("retry_with_backoff: unknown error"))) -} - async fn validate_webhook_request(kv_map: &HashMap) -> S3Result<()> { let endpoint = kv_map .get("endpoint")