adapt Intervals request limiting to official rate limits

src/config.rs

@@ -5,42 +5,116 @@ use std::{
     env,
     sync::{Arc, Mutex},
 };
-use tokio::sync::Semaphore;
+use tokio::time::{Duration, Instant, sleep_until};
 
-const MAX_ACTIVE_REQUESTS_PER_TOKEN: usize = 5;
+/// Intervals.icu currently enforces an additional limit of ten calls per
+/// second per source IP address. We pace request starts at 100 ms intervals
+/// so this application never intentionally exceeds that limit.
+const IP_REQUEST_INTERVAL: Duration = Duration::from_millis(100);
 
 #[derive(Clone, Debug)]
-pub struct TokenRequestLimiters {
+pub struct IntervalsRateLimiters {
-    inner: Arc<Mutex<HashMap<[u8; 32], Arc<Semaphore>>>>,
+    inner: Arc<RateLimiterState>,
 }
 
-impl Default for TokenRequestLimiters {
+#[derive(Debug)]
+struct RateLimiterState {
+    /// One global pacing gate because all credentials used by this process
+    /// originate from the same source IP.
+    ip: Mutex<Option<Instant>>,
+
+    /// Per-caller backoff. A 429 response supplies Retry-After; all
+    /// subsequent requests using that credential wait until the server says
+    /// the call may succeed again.
+    caller_blocked_until: Mutex<HashMap<[u8; 32], Instant>>,
+}
+
+impl Default for IntervalsRateLimiters {
     fn default() -> Self {
         Self {
-            inner: Arc::new(Mutex::new(HashMap::new())),
+            inner: Arc::new(RateLimiterState {
+                ip: Mutex::new(None),
+                caller_blocked_until: Mutex::new(HashMap::new()),
+            }),
         }
     }
 }
 
-impl TokenRequestLimiters {
+impl IntervalsRateLimiters {
-    pub fn semaphore(&self, token: &str) -> Arc<Semaphore> {
+    /// Pace the next request for this process so request starts do not exceed
-        let digest = Sha256::digest(token.as_bytes());
+    /// ten per second from the application's source IP.
-        let key: [u8; 32] = digest.into();
+    pub async fn wait_for_ip_slot(&self) {
+        let now = Instant::now();
 
-        let mut limiters = self
+        let wait_until = {
-            .inner
+            let mut last = self
-            .lock()
+                .inner
-            .expect("token request limiter mutex poisoned");
+                .ip
+                .lock()
+                .expect("Intervals IP limiter mutex poisoned");
 
-        limiters
+            let next = match *last {
-            .entry(key)
+                Some(previous) => {
-            .or_insert_with(|| {
+                    std::cmp::max(now, previous + IP_REQUEST_INTERVAL)
-                Arc::new(Semaphore::new(
+                }
-                    MAX_ACTIVE_REQUESTS_PER_TOKEN,
+                None => now,
-                ))
+            };
-            })
+
-            .clone()
+            *last = Some(next);
+            next
+        };
+
+        sleep_until(wait_until).await;
     }
+
+    /// Wait for a Retry-After backoff associated with this credential.
+    pub async fn wait_for_caller(&self, caller: &str) {
+        let key = caller_key(caller);
+        let blocked_until = {
+            let blocked = self
+                .inner
+                .caller_blocked_until
+                .lock()
+                .expect("Intervals credential limiter mutex poisoned");
+
+            blocked.get(&key).copied()
+        };
+
+        if let Some(blocked_until) = blocked_until {
+            let now = Instant::now();
+
+            if blocked_until > now {
+                sleep_until(blocked_until).await;
+            }
+        }
+    }
+
+    /// Block this credential for the supplied duration after a 429 response.
+    /// Multiple concurrent 429s extend the block to the latest deadline.
+    pub fn block_caller(&self, caller: &str, duration: Duration) {
+        let key = caller_key(caller);
+        let blocked_until = Instant::now() + duration;
+
+        let mut blocked = self
+            .inner
+            .caller_blocked_until
+            .lock()
+            .expect("Intervals credential limiter mutex poisoned");
+
+        blocked
+            .entry(key)
+            .and_modify(|current| {
+                if blocked_until > *current {
+                    *current = blocked_until;
+                }
+            })
+            .or_insert(blocked_until);
+    }
+}
+
+fn caller_key(caller: &str) -> [u8; 32] {
+    let digest = Sha256::digest(caller.as_bytes());
+    digest.into()
 }
 
 #[derive(Clone, Debug)]
@@ -59,9 +133,9 @@ pub struct Config {
 
     pub cookie_secure: bool,
 
-    /// Shared per-credential concurrency limit for Intervals.icu requests.
+    /// Shared Intervals.icu rate-limit state. It applies to all clients
-    /// Each API/OAuth credential is limited to five active requests.
+    /// created from this Config, including API-key and OAuth callers.
-    pub intervals_request_limiters: TokenRequestLimiters,
+    pub intervals_rate_limiters: IntervalsRateLimiters,
 }
 
 impl Config {
@@ -109,13 +183,15 @@ impl Config {
 
             cookie_secure,
 
-            intervals_request_limiters: TokenRequestLimiters::default(),
+            intervals_rate_limiters:
+                IntervalsRateLimiters::default(),
         })
     }
 }
 
 fn required(name: &str) -> Result<String> {
-    let value = env::var(name).with_context(|| format!("{name} is not configured"))?;
+    let value = env::var(name)
+        .with_context(|| format!("{name} is not configured"))?;
 
     if value.trim().is_empty() {
         anyhow::bail!("{name} is empty");
@@ -125,60 +201,47 @@ fn required(name: &str) -> Result<String> {
 }
 
 fn optional(name: &str) -> Option<String> {
-    env::var(name).ok().filter(|value| !value.trim().is_empty())
+    env::var(name)
+        .ok()
+        .filter(|value| !value.trim().is_empty())
 }
 
 #[cfg(test)]
 mod tests {
-    use super::TokenRequestLimiters;
+    use super::IntervalsRateLimiters;
-    use std::sync::Arc;
+    use tokio::time::{Duration, Instant};
-    use tokio::time::{Duration, timeout};
 
     #[tokio::test]
-    async fn allows_at_most_five_active_requests_per_token() {
+    async fn different_credentials_share_the_ip_limit() {
-        let limiters = TokenRequestLimiters::default();
+        let limiters = IntervalsRateLimiters::default();
-        let semaphore = limiters.semaphore("same-token");
 
-        let mut permits = Vec::new();
+        let start = Instant::now();
-        for _ in 0..5 {
+
-            permits.push(
+        for _ in 0..3 {
-                semaphore
+            limiters.wait_for_ip_slot().await;
-                    .clone()
-                    .acquire_owned()
-                    .await
-                    .expect("semaphore should be open"),
-            );
         }
 
-        let sixth = timeout(
+        assert!(
-            Duration::from_millis(20),
+            Instant::now().duration_since(start) >= Duration::from_millis(190)
-            semaphore.clone().acquire_owned(),
+        );
-        )
-        .await;
-
-        assert!(sixth.is_err(), "sixth request must wait");
-
-        drop(permits.pop());
-
-        let sixth = timeout(
-            Duration::from_millis(100),
-            semaphore.acquire_owned(),
-        )
-        .await
-        .expect("a freed permit should become available")
-        .expect("semaphore should be open");
-
-        drop(sixth);
     }
 
-    #[test]
+    #[tokio::test]
-    fn different_tokens_get_independent_limiters() {
+    async fn caller_backoff_is_independent() {
-        let limiters = TokenRequestLimiters::default();
+        let limiters = IntervalsRateLimiters::default();
-        let first = limiters.semaphore("token-a");
-        let second = limiters.semaphore("token-b");
 
-        assert!(!Arc::ptr_eq(&first, &second));
+        limiters.block_caller("caller-a", Duration::from_millis(80));
-        assert_eq!(first.available_permits(), 5);
+
-        assert_eq!(second.available_permits(), 5);
+        let start = Instant::now();
+        limiters.wait_for_caller("caller-a").await;
+        assert!(
+            Instant::now().duration_since(start) >= Duration::from_millis(70)
+        );
+
+        let start = Instant::now();
+        limiters.wait_for_caller("caller-b").await;
+        assert!(
+            Instant::now().duration_since(start) < Duration::from_millis(40)
+        );
     }
 }

src/intervals.rs

@@ -58,10 +58,18 @@ impl IntervalsClient {
         self.config.intervals_base_url.trim_end_matches('/')
     }
 
-    fn authentication_token(&self) -> Result<&str> {
+    /// Return the identifier used for Intervals.icu's application-level
+    /// rate limiting. Personal API keys are limited per API-key caller.
+    /// OAuth requests are rate-limited per OAuth client app, not per access
+    /// token, so all tokens belonging to this application share one bucket.
+    fn rate_limit_caller(&self) -> Result<String> {
         match self.authentication.as_ref() {
-            Some(Authentication::ApiKey(token))
+            Some(Authentication::ApiKey(token)) => Ok(token.clone()),
-            | Some(Authentication::Bearer(token)) => Ok(token),
+            Some(Authentication::Bearer(token)) => Ok(self
+                .config
+                .intervals_client_id
+                .clone()
+                .unwrap_or_else(|| token.clone())),
             None => Err(anyhow!(
                 "Intervals.icu authentication is not configured"
             )),
@@ -94,39 +102,103 @@ impl IntervalsClient {
     }
 
     async fn get_json(&self, url: String) -> Result<Value> {
-        let token = self.authentication_token()?;
+        let caller = self.rate_limit_caller()?;
-        let semaphore = self
+        let limiters = self
             .config
-            .intervals_request_limiters
+            .intervals_rate_limiters
-            .semaphore(token);
+            .clone();
 
-        // The permit covers the complete active request, including reading
+        // Intervals.icu enforces an additional limit of ten calls/sec per
-        // the response body. This enforces a real concurrency limit rather
+        // source IP. Pace every request start globally for this process.
-        // than merely limiting request creation.
+        const MAX_RATE_LIMIT_RETRIES: usize = 3;
-        let _permit = semaphore
-            .acquire()
-            .await
-            .context("Intervals.icu request limiter closed")?;
 
-        let response = self
+        for attempt in 0..=MAX_RATE_LIMIT_RETRIES {
-            .authenticated(self.client.get(&url))?
+            // Retry-After backoff is scoped to the API-key caller or OAuth
-            .send()
+            // client app, while the ten-calls/sec limit is scoped to the
-            .await
+            // source IP. Apply both gates before each actual request.
-            .context("Intervals.icu request failed")?;
+            limiters.wait_for_caller(&caller).await;
+            limiters.wait_for_ip_slot().await;
 
-        let status = response.status();
+            let response = self
+                .authenticated(self.client.get(&url))?
+                .send()
+                .await
+                .context("Intervals.icu request failed")?;
 
-        let body = response
+            let status = response.status();
-            .text()
+            let retry_after =
-            .await
+                parse_retry_after(response.headers().get("Retry-After"));
-            .context("cannot read Intervals.icu response")?;
+            let rate_limit =
+                parse_rate_limit_headers(response.headers());
 
-        if !status.is_success() {
+            if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
-            return Err(anyhow!("Intervals.icu returned HTTP {}: {}", status, body));
+                let body = response
+                    .text()
+                    .await
+                    .context("cannot read Intervals.icu 429 response")?;
+
+                let Some(delay) = retry_after else {
+                    return Err(anyhow!(
+                        "Intervals.icu returned HTTP 429 without Retry-After: {}",
+                        body
+                    ));
+                };
+
+                limiters.block_caller(&caller, delay);
+
+                tracing::warn!(
+                    attempt,
+                    retry_after_seconds = delay.as_secs(),
+                    rate_limit_15m = rate_limit.0,
+                    rate_limit_daily = rate_limit.1,
+                    remaining_15m = rate_limit.2,
+                    remaining_daily = rate_limit.3,
+                    "Intervals.icu rate limit reached; waiting before retry"
+                );
+
+                if attempt == MAX_RATE_LIMIT_RETRIES {
+                    return Err(anyhow!(
+                        "Intervals.icu returned HTTP 429 after {} retries: {}",
+                        MAX_RATE_LIMIT_RETRIES,
+                        body
+                    ));
+                }
+
+                continue;
+            }
+
+            let body = response
+                .text()
+                .await
+                .context("cannot read Intervals.icu response")?;
+
+            if !status.is_success() {
+                return Err(anyhow!(
+                    "Intervals.icu returned HTTP {}: {}",
+                    status,
+                    body
+                ));
+            }
+
+            tracing::debug!(
+                status = %status,
+                rate_limit_15m = rate_limit.0,
+                rate_limit_daily = rate_limit.1,
+                remaining_15m = rate_limit.2,
+                remaining_daily = rate_limit.3,
+                "Intervals.icu request completed"
+            );
+
+            return serde_json::from_str(&body)
+                .with_context(|| {
+                    format!(
+                        "invalid JSON returned by Intervals.icu: {}",
+                        body
+                    )
+                });
         }
 
-        serde_json::from_str(&body)
+        unreachable!("rate-limit retry loop must return")
-            .with_context(|| format!("invalid JSON returned by Intervals.icu: {}", body))
     }
 
     /// Get the athlete belonging to the configured API key.
@@ -429,6 +501,56 @@ impl IntervalsClient {
     }
 }
 
+
+fn parse_retry_after(
+    value: Option<&reqwest::header::HeaderValue>,
+) -> Option<std::time::Duration> {
+    let value = value?.to_str().ok()?.trim();
+    let seconds = value.parse::<u64>().ok()?;
+    Some(std::time::Duration::from_secs(seconds))
+}
+
+/// Parse the two rate-limit headers returned by Intervals.icu:
+///
+///     X-RateLimit-Limit: <15m>,<daily>
+///     X-RateLimit-Remaining: <15m>,<daily>
+///
+/// Missing or malformed headers are represented as `None` values.
+fn parse_rate_limit_headers(
+    headers: &reqwest::header::HeaderMap,
+) -> (Option<u64>, Option<u64>, Option<u64>, Option<u64>) {
+    fn pair(
+        value: Option<&reqwest::header::HeaderValue>,
+    ) -> (Option<u64>, Option<u64>) {
+        let Some(value) = value else {
+            return (None, None);
+        };
+
+        let Ok(value) = value.to_str() else {
+            return (None, None);
+        };
+
+        let mut parts = value.split(',').map(str::trim);
+
+        let first = parts.next().and_then(|value| value.parse().ok());
+        let second = parts.next().and_then(|value| value.parse().ok());
+
+        (first, second)
+    }
+
+    let (limit_15m, limit_daily) =
+        pair(headers.get("X-RateLimit-Limit"));
+    let (remaining_15m, remaining_daily) =
+        pair(headers.get("X-RateLimit-Remaining"));
+
+    (
+        limit_15m,
+        limit_daily,
+        remaining_15m,
+        remaining_daily,
+    )
+}
+
 fn parse_datetime(value: &str) -> Result<DateTime<Utc>> {
     if let Ok(value) = DateTime::parse_from_rfc3339(value) {
         return Ok(value.with_timezone(&Utc));