feat: handle rate limiting

Cargo.toml

@@ -13,7 +13,7 @@ members = [
 edition = "2024"
 license = "Apache-2.0"
 rust-version = "1.87.0"
-version = "0.15.0"
+version = "0.16.0"
 
 [workspace.dependencies]
 anyhow = "1"
@@ -43,6 +43,7 @@ futures = "0.3"
 futures-util = "0.3"
 hickory-resolver = "0.25.1"
 html-escape = "0.2.13"
+httpdate = "1"
 humantime = "2"
 indicatif = "0.18.0"
 indicatif-log-bridge = "0.2.1"
@@ -55,6 +56,7 @@ parking_lot = "0.12"
 pem = "3"
 percent-encoding = "2.3"
 reqwest = "0.12"
+rstest = "0.23"
 sectxtlib = "0.3.1"
 sequoia-openpgp = { version = "2", default-features = false }
 serde = "1"
@@ -73,10 +75,10 @@ walkdir = "2.4"
 
 # internal dependencies
 
-csaf-walker = { version = "0.15.0", path = "csaf", default-features = false }
-sbom-walker = { version = "0.15.0", path = "sbom", default-features = false }
-walker-common = { version = "0.15.0", path = "common" }
-walker-extras = { version = "0.15.0", path = "extras" }
+csaf-walker = { version = "0.16.0", path = "csaf", default-features = false }
+sbom-walker = { version = "0.16.0", path = "sbom", default-features = false }
+walker-common = { version = "0.16.0", path = "common" }
+walker-extras = { version = "0.16.0", path = "extras" }
 
 [workspace.metadata.release]
 tag = false

common/Cargo.toml

@@ -28,6 +28,7 @@ fluent-uri = { workspace = true }
 fsquirrel = { workspace = true }
 futures-util = { workspace = true }
 html-escape = { workspace = true }
+httpdate = { workspace = true }
 humantime = { workspace = true }
 indicatif = { workspace = true }
 indicatif-log-bridge = { workspace = true }
@@ -86,6 +87,12 @@ denylist = [
     "_semver",
 ]
 
+[dev-dependencies]
+hyper = { version = "1", features = ["server", "http1"] }
+hyper-util = { version = "0.1", features = ["tokio"] }
+rstest = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "net"] }
+
 [package.metadata.release]
 enable-features = ["sequoia-openpgp/crypto-nettle"]
 tag = true

common/src/cli/client.rs

@@ -10,14 +10,18 @@ pub struct ClientArguments {
     /// Per-request retries count
     #[arg(short, long, default_value = "5")]
     pub retries: usize,
+
+    /// Per-request minimum delay after rate limit (429).
+    #[arg(long, default_value = "10s")]
+    pub default_retry_after: humantime::Duration,
 }
 
 impl From<ClientArguments> for FetcherOptions {
     fn from(value: ClientArguments) -> Self {
-        FetcherOptions {
-            timeout: value.timeout.into(),
-            retries: value.retries,
-        }
+        FetcherOptions::new()
+            .timeout(value.timeout)
+            .retries(value.retries)
+            .retry_after(value.default_retry_after.into())
     }
 }
 

common/src/fetcher/mod.rs

@@ -4,6 +4,7 @@ mod data;
 use backon::{ExponentialBuilder, Retryable};
 pub use data::*;
 
+use crate::http::calculate_retry_after_from_response_header;
 use reqwest::{Client, ClientBuilder, IntoUrl, Method, Response};
 use std::fmt::Debug;
 use std::future::Future;
@@ -19,21 +20,27 @@ use url::Url;
 pub struct Fetcher {
     client: Client,
     retries: usize,
+    /// *default_retry_after* is used when a 429 response does not include a Retry-After header
+    default_retry_after: Duration,
 }
 
 /// Error when retrieving
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
     #[error("Request error: {0}")]
     Request(#[from] reqwest::Error),
+    #[error("Rate limited (HTTP 429), retry after {0:?}")]
+    RateLimited(Duration),
 }
 
 /// Options for the [`Fetcher`]
 #[non_exhaustive]
 #[derive(Clone, Debug)]
 pub struct FetcherOptions {
-    pub timeout: Duration,
-    pub retries: usize,
+    timeout: Duration,
+    retries: usize,
+    default_retry_after: Duration,
+    max_retry_after: Duration,
 }
 
 impl FetcherOptions {
@@ -53,13 +60,35 @@ impl FetcherOptions {
         self.retries = retries;
         self
     }
+
+    /// Set the default retry-after duration when a 429 response doesn't include a Retry-After header.
+    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
+    }
 }
 
 impl Default for FetcherOptions {
     fn default() -> Self {
         Self {
             timeout: Duration::from_secs(30),
             retries: 5,
+            default_retry_after: Duration::from_secs(10),
+            max_retry_after: Duration::from_mins(5),
         }
     }
 }
@@ -83,6 +112,7 @@ impl Fetcher {
         Self {
             client,
             retries: options.retries,
+            default_retry_after: options.default_retry_after,
         }
     }
 
@@ -110,19 +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)
+        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
                 }
-            }
-        })
-        .retry(&backoff.with_max_times(retries))
-        .await
+            })
+            .await
     }
 
     async fn fetch_once<D: DataProcessor>(
@@ -134,6 +168,14 @@ impl Fetcher {
 
         log::debug!("Response Status: {}", response.status());
 
+        // Check for rate limiting
+        if let Some(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));
+        }
+
         Ok(processor.process(response).await?)
     }
 }

common/src/http.rs

@@ -0,0 +1,56 @@
+use std::time::Duration;
+
+use reqwest::{Response, StatusCode, header};
+
+pub enum RetryAfter {
+    Duration(Duration),
+    After(std::time::SystemTime),
+}
+
+/// Parse Retry-After header value.
+/// Supports both delay-seconds (numeric) and HTTP-date formats as per RFC7231
+fn parse_retry_after(value: &str) -> Option<RetryAfter> {
+    // Try parsing as seconds (numeric)
+    if let Ok(seconds) = value.parse::<u64>() {
+        return Some(RetryAfter::Duration(Duration::from_secs(seconds)));
+    }
+
+    // Try parsing as HTTP-date (RFC7231 format)
+    // Common formats: "Sun, 06 Nov 1994 08:49:37 GMT" (IMF-fixdate preferred)
+    if let Ok(datetime) = httpdate::parse_http_date(value) {
+        return Some(RetryAfter::After(datetime));
+    }
+
+    None
+}
+
+pub fn calculate_retry_after_from_response_header(
+    response: &Response,
+    default_duration: Duration,
+) -> Option<Duration> {
+    if response.status() == StatusCode::TOO_MANY_REQUESTS {
+        let retry_after = response
+            .headers()
+            .get(header::RETRY_AFTER)
+            .and_then(|v| v.to_str().ok())
+            .and_then(parse_retry_after)
+            .and_then(|retry| match retry {
+                RetryAfter::Duration(d) => Some(d),
+                RetryAfter::After(after) => {
+                    // Calculate duration from now until the specified time
+                    std::time::SystemTime::now()
+                        .duration_since(std::time::UNIX_EPOCH)
+                        .ok()
+                        .and_then(|now| {
+                            after
+                                .duration_since(std::time::UNIX_EPOCH)
+                                .ok()
+                                .and_then(|target| target.checked_sub(now))
+                        })
+                }
+            })
+            .unwrap_or(default_duration);
+        return Some(retry_after);
+    }
+    None
+}

common/src/lib.rs

@@ -4,6 +4,7 @@
 pub mod changes;
 pub mod compression;
 pub mod fetcher;
+pub mod http;
 pub mod locale;
 pub mod progress;
 pub mod report;