feat: handle rate limiting

common/Cargo.toml

@@ -86,6 +86,11 @@ denylist = [
     "_semver",
 ]
 
+[dev-dependencies]
+hyper = { version = "1", features = ["server", "http1"] }
+hyper-util = { version = "0.1", features = ["tokio"] }
+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,13 +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,
+            default_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::get_retry_after_from_response_header;
 use reqwest::{Client, ClientBuilder, IntoUrl, Method, Response};
 use std::fmt::Debug;
 use std::future::Future;
@@ -15,17 +16,21 @@ use url::Url;
 ///
 /// This is some functionality sitting on top an HTTP client, allowing for additional options like
 /// retries.
+/// *default_retry_after* is used when a 429 response does not include a Retry-After header.
 #[derive(Clone, Debug)]
 pub struct Fetcher {
     client: Client,
     retries: usize,
+    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`]
@@ -34,6 +39,7 @@ pub enum Error {
 pub struct FetcherOptions {
     pub timeout: Duration,
     pub retries: usize,
+    pub default_retry_after: Duration,
 }
 
 impl FetcherOptions {
@@ -53,13 +59,20 @@ 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 default_retry_after(mut self, duration: impl Into<Duration>) -> Self {
+        self.default_retry_after = duration.into();
+        self
+    }
 }
 
 impl Default for FetcherOptions {
     fn default() -> Self {
         Self {
             timeout: Duration::from_secs(30),
             retries: 5,
+            default_retry_after: Duration::from_secs(10),
         }
     }
 }
@@ -83,6 +96,7 @@ impl Fetcher {
         Self {
             client,
             retries: options.retries,
+            default_retry_after: options.default_retry_after,
         }
     }
 
@@ -122,6 +136,20 @@ impl Fetcher {
             }
         })
         .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);
+                }
+            }
+        })
         .await
     }
 
@@ -134,6 +162,14 @@ impl Fetcher {
 
         log::debug!("Response Status: {}", response.status());
 
+        // Check for rate limiting
+        if let Some(retry_after) =
+            get_retry_after_from_response_header(&response, self.default_retry_after)
+        {
+            log::warn!("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,29 @@
+use std::time::Duration;
+
+use reqwest::{Response, StatusCode, header};
+
+/// Parse Retry-After header value
+fn parse_retry_after(value: &str) -> Option<Duration> {
+    // Try parsing as seconds (numeric)
+    if let Ok(seconds) = value.parse::<u64>() {
+        return Some(Duration::from_secs(seconds));
+    }
+    // Could also parse HTTP-date format here if needed
+    None
+}
+
+pub fn get_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)
+            .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;