fix: use adjustable retry delays when sending data

tziemek · tziemek · commit 2d78552cb7e7 · 2025-11-28T10:58:26.000+01:00
diff --git a/common/src/cli/client.rs b/common/src/cli/client.rs
@@ -18,11 +18,10 @@ pub struct ClientArguments {
 
 impl From<ClientArguments> for FetcherOptions {
     fn from(value: ClientArguments) -> Self {
-        FetcherOptions {
-            timeout: value.timeout.into(),
-            retries: value.retries,
-            default_retry_after: value.default_retry_after.into(),
-        }
+        FetcherOptions::new()
+            .timeout(value.timeout)
+            .retries(value.retries)
+            .retry_after(value.default_retry_after.into())
     }
 }
 
diff --git a/common/src/fetcher/mod.rs b/common/src/fetcher/mod.rs
@@ -4,7 +4,7 @@ mod data;
 use backon::{ExponentialBuilder, Retryable};
 pub use data::*;
 
-use crate::http::get_retry_after_from_response_header;
+use crate::http::calculate_retry_after_from_response_header;
 use reqwest::{Client, ClientBuilder, IntoUrl, Method, Response};
 use std::fmt::Debug;
 use std::future::Future;
@@ -37,9 +37,10 @@ pub enum Error {
 #[non_exhaustive]
 #[derive(Clone, Debug)]
 pub struct FetcherOptions {
-    pub timeout: Duration,
-    pub retries: usize,
-    pub default_retry_after: Duration,
+    timeout: Duration,
+    retries: usize,
+    default_retry_after: Duration,
+    max_retry_after: Duration,
 }
 
 impl FetcherOptions {
@@ -61,8 +62,22 @@ impl FetcherOptions {
     }
 
     /// Set the default retry-after duration when a 429 response doesn't include a Retry-After header.
-    pub fn default_retry_after(mut self, duration: impl Into<Duration>) -> Self {
-        self.default_retry_after = duration.into();
+    pub fn retry_after(mut self, duration: Duration) -> Self {
+        if duration > self.max_retry_after {
+            panic!("Default retry-after cannot be greater than max retry-after (300s)");
+        }
+        self.default_retry_after = duration;
+        self
+    }
+
+    /// Set the default retry-after duration when a 429 response doesn't include a Retry-After header
+    /// and checks the duration against the maximum retry-after.
+    pub fn retry_after_with_max(mut self, default: Duration, max: Duration) -> Self {
+        if default > max {
+            panic!("Default retry-after cannot be greater than max retry-after");
+        }
+        self.default_retry_after = default;
+        self.max_retry_after = max;
         self
     }
 }
@@ -73,6 +88,7 @@ impl Default for FetcherOptions {
             timeout: Duration::from_secs(30),
             retries: 5,
             default_retry_after: Duration::from_secs(10),
+            max_retry_after: Duration::from_mins(5),
         }
     }
 }
@@ -124,33 +140,23 @@ impl Fetcher {
         let url = url.into_url()?;
 
         let retries = self.retries;
-        let backoff = ExponentialBuilder::default();
-
-        (|| async {
-            match self.fetch_once(url.clone(), &processor).await {
-                Ok(result) => Ok(result),
-                Err(err) => {
-                    log::info!("Failed to retrieve: {err}");
-                    Err(err)
-                }
-            }
-        })
-        .retry(&backoff.with_max_times(retries))
-        .notify(|err, dur| {
-            // If rate limited, ensure we wait at least the Retry-After duration
-            if let Error::RateLimited(retry_after) = err {
-                if dur < *retry_after {
-                    log::info!(
-                        "Rate limited, extending wait from {:?} to {:?}",
-                        dur,
-                        retry_after
-                    );
-                    let additional = *retry_after - dur;
-                    std::thread::sleep(additional);
+        let retry = ExponentialBuilder::default().with_max_times(retries);
+
+        (|| async { self.fetch_once(url.clone(), &processor).await })
+            .retry(retry)
+            .adjust(|e, dur| {
+                if let Error::RateLimited(retry_after) = e {
+                    if let Some(dur_value) = dur
+                        && dur_value > *retry_after
+                    {
+                        return dur;
+                    }
+                    Some(*retry_after) // only use server-provided delay if it's longer
+                } else {
+                    dur // minimum delay as per backoff strategy
                 }
-            }
-        })
-        .await
+            })
+            .await
     }
 
     async fn fetch_once<D: DataProcessor>(
@@ -164,7 +170,7 @@ impl Fetcher {
 
         // Check for rate limiting
         if let Some(retry_after) =
-            get_retry_after_from_response_header(&response, self.default_retry_after)
+            calculate_retry_after_from_response_header(&response, self.default_retry_after)
         {
             log::info!("Rate limited (429), retry after: {:?}", retry_after);
             return Err(Error::RateLimited(retry_after));
diff --git a/common/src/http.rs b/common/src/http.rs
@@ -24,7 +24,7 @@ fn parse_retry_after(value: &str) -> Option<RetryAfter> {
     None
 }
 
-pub fn get_retry_after_from_response_header(
+pub fn calculate_retry_after_from_response_header(
     response: &Response,
     default_duration: Duration,
 ) -> Option<Duration> {
diff --git a/common/tests/fetcher.rs b/common/tests/fetcher.rs
@@ -229,7 +229,7 @@ async fn test_configurable_default_retry_after(#[case] custom_default_secs: u64)
     let fetcher = Fetcher::new(
         FetcherOptions::new()
             .retries(3)
-            .default_retry_after(Duration::from_secs(custom_default_secs)),
+            .retry_after(Duration::from_secs(custom_default_secs)),
     )
     .await
     .unwrap();
diff --git a/extras/src/visitors/send/clap.rs b/extras/src/visitors/send/clap.rs
@@ -107,13 +107,9 @@ impl SendArguments {
         )
         .await?;
 
-        Ok(SendVisitor {
-            url: target,
-            sender,
-            retries,
-            min_delay: Some(min_delay.into()),
-            max_delay: Some(max_delay.into()),
-            default_retry_after: None,
-        })
+        Ok(SendVisitor::new(target, sender)
+            .retries(retries)
+            .min_delay(min_delay)
+            .max_delay(max_delay))
     }
 }
diff --git a/extras/src/visitors/send/mod.rs b/extras/src/visitors/send/mod.rs
@@ -3,7 +3,7 @@ use bytes::Bytes;
 use reqwest::{Body, Method, StatusCode, Url, header};
 use std::time::Duration;
 use walker_common::{
-    http::get_retry_after_from_response_header,
+    http::calculate_retry_after_from_response_header,
     sender::{self, HttpSender},
 };
 
@@ -49,16 +49,16 @@ pub struct SendVisitor {
     pub sender: HttpSender,
 
     /// The number of retries in case of a server or transmission failure
-    pub retries: usize,
+    retries: usize,
 
-    /// The minimum delay between retries
-    pub min_delay: Option<Duration>,
+    /// The minimum delay between retries, will be overruled by the retry-after header if present.
+    min_delay: Option<Duration>,
 
-    /// The maximum delay between retries
-    pub max_delay: Option<Duration>,
+    /// The maximum delay between retries, will be overruled by the retry-after header if present.
+    max_delay: Option<Duration>,
 
     /// The default retry-after duration when a 429 response doesn't include a Retry-After header
-    pub default_retry_after: Option<Duration>,
+    default_retry_after: Duration,
 }
 
 impl SendVisitor {
@@ -69,7 +69,7 @@ impl SendVisitor {
             retries: 0,
             min_delay: None,
             max_delay: None,
-            default_retry_after: Some(Duration::from_secs(10)),
+            default_retry_after: Duration::from_secs(10),
         }
     }
 
@@ -87,12 +87,6 @@ impl SendVisitor {
         self.max_delay = Some(retry_delay.into());
         self
     }
-
-    /// Set the default retry-after duration when a 429 response doesn't include a Retry-After header.
-    pub fn default_retry_after(mut self, duration: impl Into<Duration>) -> Self {
-        self.default_retry_after = Some(duration.into());
-        self
-    }
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -135,10 +129,9 @@ impl SendVisitor {
             .await
             .map_err(|err| SendOnceError::Temporary(err.into()))?;
 
-        if let Some(retry_after) = get_retry_after_from_response_header(
-            &response,
-            self.default_retry_after.unwrap_or(Duration::from_secs(10)),
-        ) {
+        if let Some(retry_after) =
+            calculate_retry_after_from_response_header(&response, self.default_retry_after)
+        {
             log::warn!(
                 "Rate limited (429) when uploading {name}, retry after: {:?}",
                 retry_after
@@ -185,20 +178,17 @@ impl SendVisitor {
         Ok(
             (|| async { self.send_once(name, data.clone(), &customizer).await })
                 .retry(retry)
-                .sleep(tokio::time::sleep)
                 .when(|e| matches!(e, SendOnceError::Temporary(_)))
-                .notify(|err, dur| {
-                    // If rate limited, ensure we wait at least the Retry-After duration
-                    if let SendOnceError::Temporary(SendError::RateLimited(retry_after)) = err {
-                        if dur < *retry_after {
-                            log::info!(
-                                "Rate limited, extending wait from {:?} to {:?}",
-                                dur,
-                                retry_after
-                            );
-                            let additional = *retry_after - dur;
-                            std::thread::sleep(additional);
+                .adjust(|e, dur| {
+                    if let SendOnceError::Temporary(SendError::RateLimited(retry_after)) = e {
+                        if let Some(dur_value) = dur
+                            && dur_value > *retry_after
+                        {
+                            return dur;
                         }
+                        Some(*retry_after) // only use server-provided delay if it's longer
+                    } else {
+                        dur // minimum delay as per backoff strategy
                     }
                 })
                 .await?,

Original file line number	Diff line number	Diff line change
`@@ -18,11 +18,10 @@ pub struct ClientArguments {`
`18`	`18`
`19`	`19`	`impl From<ClientArguments> for FetcherOptions {`
`20`	`20`	`fn from(value: ClientArguments) -> Self {`
`21`		`- FetcherOptions {`
`22`		`- timeout: value.timeout.into(),`
`23`		`- retries: value.retries,`
`24`		`- default_retry_after: value.default_retry_after.into(),`
`25`		`- }`
	`21`	`+ FetcherOptions::new()`
	`22`	`+ .timeout(value.timeout)`
	`23`	`+ .retries(value.retries)`
	`24`	`+ .retry_after(value.default_retry_after.into())`
`26`	`25`	`}`
`27`	`26`	`}`
`28`	`27`
Original file line number	Diff line number	Diff line change
`@@ -24,7 +24,7 @@ fn parse_retry_after(value: &str) -> Option<RetryAfter> {`
`24`	`24`	`None`
`25`	`25`	`}`
`26`	`26`
`27`		`-pub fn get_retry_after_from_response_header(`
	`27`	`+pub fn calculate_retry_after_from_response_header(`
`28`	`28`	`response: &Response,`
`29`	`29`	`default_duration: Duration,`
`30`	`30`	`) -> Option<Duration> {`
Original file line number	Diff line number	Diff line change
`@@ -229,7 +229,7 @@ async fn test_configurable_default_retry_after(#[case] custom_default_secs: u64)`
`229`	`229`	`let fetcher = Fetcher::new(`
`230`	`230`	`FetcherOptions::new()`
`231`	`231`	`.retries(3)`
`232`		`- .default_retry_after(Duration::from_secs(custom_default_secs)),`
	`232`	`+ .retry_after(Duration::from_secs(custom_default_secs)),`
`233`	`233`	`)`
`234`	`234`	`.await`
`235`	`235`	`.unwrap();`
Original file line number	Diff line number	Diff line change
`@@ -107,13 +107,9 @@ impl SendArguments {`
`107`	`107`	`)`
`108`	`108`	`.await?;`
`109`	`109`
`110`		`- Ok(SendVisitor {`
`111`		`- url: target,`
`112`		`- sender,`
`113`		`- retries,`
`114`		`- min_delay: Some(min_delay.into()),`
`115`		`- max_delay: Some(max_delay.into()),`
`116`		`- default_retry_after: None,`
`117`		`- })`
	`110`	`+ Ok(SendVisitor::new(target, sender)`
	`111`	`+ .retries(retries)`
	`112`	`+ .min_delay(min_delay)`
	`113`	`+ .max_delay(max_delay))`
`118`	`114`	`}`
`119`	`115`	`}`