diff --git a/src/config.rs b/src/config.rs index 64ac134..40ced39 100644 --- a/src/config.rs +++ b/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 { - inner: Arc>>>, +pub struct IntervalsRateLimiters { + inner: Arc, } -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>, + + /// 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>, +} + +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 { - pub fn semaphore(&self, token: &str) -> Arc { - let digest = Sha256::digest(token.as_bytes()); - let key: [u8; 32] = digest.into(); +impl IntervalsRateLimiters { + /// Pace the next request for this process so request starts do not exceed + /// ten per second from the application's source IP. + pub async fn wait_for_ip_slot(&self) { + let now = Instant::now(); - let mut limiters = self - .inner - .lock() - .expect("token request limiter mutex poisoned"); + let wait_until = { + let mut last = self + .inner + .ip + .lock() + .expect("Intervals IP limiter mutex poisoned"); - limiters - .entry(key) - .or_insert_with(|| { - Arc::new(Semaphore::new( - MAX_ACTIVE_REQUESTS_PER_TOKEN, - )) - }) - .clone() + let next = match *last { + Some(previous) => { + std::cmp::max(now, previous + IP_REQUEST_INTERVAL) + } + None => now, + }; + + *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. - /// Each API/OAuth credential is limited to five active requests. - pub intervals_request_limiters: TokenRequestLimiters, + /// Shared Intervals.icu rate-limit state. It applies to all clients + /// created from this Config, including API-key and OAuth callers. + 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 { - 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 { } fn optional(name: &str) -> Option { - 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 std::sync::Arc; - use tokio::time::{Duration, timeout}; + use super::IntervalsRateLimiters; + use tokio::time::{Duration, Instant}; #[tokio::test] - async fn allows_at_most_five_active_requests_per_token() { - let limiters = TokenRequestLimiters::default(); - let semaphore = limiters.semaphore("same-token"); + async fn different_credentials_share_the_ip_limit() { + let limiters = IntervalsRateLimiters::default(); - let mut permits = Vec::new(); - for _ in 0..5 { - permits.push( - semaphore - .clone() - .acquire_owned() - .await - .expect("semaphore should be open"), - ); + let start = Instant::now(); + + for _ in 0..3 { + limiters.wait_for_ip_slot().await; } - let sixth = timeout( - Duration::from_millis(20), - 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); + assert!( + Instant::now().duration_since(start) >= Duration::from_millis(190) + ); } - #[test] - fn different_tokens_get_independent_limiters() { - let limiters = TokenRequestLimiters::default(); - let first = limiters.semaphore("token-a"); - let second = limiters.semaphore("token-b"); + #[tokio::test] + async fn caller_backoff_is_independent() { + let limiters = IntervalsRateLimiters::default(); - assert!(!Arc::ptr_eq(&first, &second)); - assert_eq!(first.available_permits(), 5); - assert_eq!(second.available_permits(), 5); + limiters.block_caller("caller-a", Duration::from_millis(80)); + + 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) + ); } } diff --git a/src/intervals.rs b/src/intervals.rs index 28c36c0..6cf6a86 100644 --- a/src/intervals.rs +++ b/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 { match self.authentication.as_ref() { - Some(Authentication::ApiKey(token)) - | Some(Authentication::Bearer(token)) => Ok(token), + Some(Authentication::ApiKey(token)) => Ok(token.clone()), + 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 { - let token = self.authentication_token()?; - let semaphore = self + let caller = self.rate_limit_caller()?; + let limiters = self .config - .intervals_request_limiters - .semaphore(token); + .intervals_rate_limiters + .clone(); - // The permit covers the complete active request, including reading - // the response body. This enforces a real concurrency limit rather - // than merely limiting request creation. - let _permit = semaphore - .acquire() - .await - .context("Intervals.icu request limiter closed")?; + // Intervals.icu enforces an additional limit of ten calls/sec per + // source IP. Pace every request start globally for this process. + const MAX_RATE_LIMIT_RETRIES: usize = 3; - let response = self - .authenticated(self.client.get(&url))? - .send() - .await - .context("Intervals.icu request failed")?; + for attempt in 0..=MAX_RATE_LIMIT_RETRIES { + // Retry-After backoff is scoped to the API-key caller or OAuth + // client app, while the ten-calls/sec limit is scoped to the + // source IP. Apply both gates before each actual request. + 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 - .text() - .await - .context("cannot read Intervals.icu response")?; + let status = response.status(); + let retry_after = + parse_retry_after(response.headers().get("Retry-After")); + let rate_limit = + parse_rate_limit_headers(response.headers()); - if !status.is_success() { - return Err(anyhow!("Intervals.icu returned HTTP {}: {}", status, body)); + if status == reqwest::StatusCode::TOO_MANY_REQUESTS { + 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) - .with_context(|| format!("invalid JSON returned by Intervals.icu: {}", body)) + unreachable!("rate-limit retry loop must return") } /// 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 { + let value = value?.to_str().ok()?.trim(); + let seconds = value.parse::().ok()?; + Some(std::time::Duration::from_secs(seconds)) +} + +/// Parse the two rate-limit headers returned by Intervals.icu: +/// +/// X-RateLimit-Limit: <15m>, +/// X-RateLimit-Remaining: <15m>, +/// +/// Missing or malformed headers are represented as `None` values. +fn parse_rate_limit_headers( + headers: &reqwest::header::HeaderMap, +) -> (Option, Option, Option, Option) { + fn pair( + value: Option<&reqwest::header::HeaderValue>, + ) -> (Option, Option) { + 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> { if let Ok(value) = DateTime::parse_from_rfc3339(value) { return Ok(value.with_timezone(&Utc));