feat(app): handle rate limits (#6790)

* feat(app): handle rate limits + governor rate limits

* feat(app): add hysteresis to rate limiter

* fix(ci): make sure sccache directory exists if mount fail

* fix(ci): also apply sccache mount fix to daedalus-docker

* fix(ci): sudo

packages/app-lib/Cargo.toml

@@ -43,8 +43,10 @@ flate2 = { workspace = true }
 fs4 = { workspace = true, features = ["tokio"] }
 futures = { workspace = true, features = ["alloc", "async-await"] }
 futures-lite = { workspace = true }
+governor = { workspace = true }
 heck = { workspace = true }
 hickory-resolver = { workspace = true }
+httpdate = { workspace = true }
 indicatif = { workspace = true, optional = true }
 itertools = { workspace = true }
 modrinth-content-management = { workspace = true }

packages/app-lib/src/error.rs

@@ -79,6 +79,9 @@ pub enum ErrorKind {
     #[error("Too many API errors, try again in {0} minutes")]
     ApiIsDownError(u32),
 
+    #[error("Too many requests, retry in {}", format_seconds(*.retry_in_seconds))]
+    Ratelimited { retry_in_seconds: u64 },
+
     #[error("{0}")]
     LabrinthError(LabrinthError),
 
@@ -192,6 +195,27 @@ pub enum ErrorKind {
     DiscordRichPresenceError(#[from] discord_rich_presence::error::Error),
 }
 
+fn format_seconds(seconds: u64) -> String {
+    let plural = |unit: u64| if unit == 1 { "" } else { "s" };
+
+    if seconds <= 59 {
+        return format!("{seconds} second{}", plural(seconds));
+    }
+
+    let minutes = seconds / 60;
+    let rem_seconds = seconds % 60;
+
+    if rem_seconds == 0 {
+        return format!("{minutes} minutes");
+    }
+
+    format!(
+        "{minutes} minute{} and {rem_seconds} second{}",
+        plural(minutes),
+        plural(rem_seconds)
+    )
+}
+
 #[derive(Debug)]
 pub struct Error {
     pub raw: Arc<ErrorKind>,

packages/app-lib/src/install/runner.rs

@@ -1074,9 +1074,9 @@ fn install_error_code(
             PreparingJava => "java_error",
             _ => "metadata_error",
         },
-        ErrorKind::FetchError(_) | ErrorKind::ApiIsDownError(_) => {
-            "network_error"
-        }
+        ErrorKind::FetchError(_)
+        | ErrorKind::ApiIsDownError(_)
+        | ErrorKind::Ratelimited { .. } => "network_error",
         ErrorKind::Any(_)
             if matches!(
                 phase,

packages/app-lib/src/util/fetch.rs

@@ -6,6 +6,8 @@ use crate::{ErrorKind, LabrinthError};
 use bytes::Bytes;
 use chrono::{DateTime, TimeDelta, Utc};
 use eyre::{Context, eyre};
+use governor::clock::{Clock, DefaultClock};
+use governor::{DefaultDirectRateLimiter, Quota, RateLimiter};
 use parking_lot::Mutex;
 use rand::Rng;
 use reqwest::Method;
@@ -14,12 +16,14 @@ use serde::{Deserialize, Serialize};
 use std::collections::{HashMap, VecDeque};
 use std::ffi::OsStr;
 use std::future::Future;
+use std::num::NonZeroU32;
 use std::path::{Path, PathBuf};
 use std::pin::Pin;
-use std::sync::LazyLock;
-use std::time::{self};
+use std::sync::{Arc, LazyLock};
+use std::time::{self, Duration, Instant, SystemTime};
 use tokio::sync::Semaphore;
 use tokio::{fs::File, io::AsyncReadExt, io::AsyncWriteExt};
+use tracing::{debug, info};
 
 pub const DOWNLOAD_META_HEADER: &str = "modrinth-download-meta";
 
@@ -188,6 +192,145 @@ static GLOBAL_FETCH_FENCE: LazyLock<FetchFence> =
         inner: Mutex::new(HashMap::new()),
     });
 
+const API_RETRY_AFTER_FALLBACK: Duration = Duration::from_secs(60);
+
+// This means the unit recovery time will be:
+// replenish one unit time in seconds = (60 / (units recovered per minute))
+// smooth recovery time = replenish one unit time in seconds * API_RATE_LIMIT_RECOVERY_SIZE.
+//
+// At 20 it means (60 / 120) * 20 = 10 seconds.
+const API_RATE_LIMIT_RECOVERY_SIZE: u32 = 20;
+
+/// Implements request-rate-limit handling as well as a local rate limiter of 120 RPM + 50 burst.
+struct ApiRateLimit {
+    block_until: Mutex<Option<Instant>>,
+    check_lock: Mutex<()>,
+    local: Arc<DefaultDirectRateLimiter>,
+    recovery_padding: Duration,
+}
+
+impl ApiRateLimit {
+    fn new() -> Self {
+        let quota = Quota::per_minute(NonZeroU32::new(120).unwrap())
+            .allow_burst(NonZeroU32::new(50).unwrap());
+        let recovery_size =
+            API_RATE_LIMIT_RECOVERY_SIZE.min(quota.burst_size().get());
+
+        Self {
+            block_until: Mutex::new(None),
+            check_lock: Mutex::new(()),
+            local: Arc::new(RateLimiter::direct(quota)),
+            recovery_padding: quota
+                .replenish_interval()
+                .saturating_mul(recovery_size.saturating_sub(1)),
+        }
+    }
+
+    fn check(&self) -> crate::Result<()> {
+        let _check_guard = self.check_lock.lock();
+        self.ensure_not_blocked()?;
+
+        if let Err(not_until) = self.local.check() {
+            // Adds hysteresis to the rate limiting system, ensuring recovery happens
+            // for longer but for more units, avoiding the "flapping" effect when running
+            // out of units.
+
+            let retry_after = not_until
+                .wait_time_from(DefaultClock::default().now())
+                .saturating_add(self.recovery_padding);
+            info!(
+                ?retry_after,
+                "Hit builtin rate limiter; waiting for recovery"
+            );
+            self.block_for(retry_after);
+
+            let retry_in_seconds = self
+                .retry_in_seconds()
+                .unwrap_or_else(|| duration_seconds_ceil(retry_after));
+
+            return Err(ErrorKind::Ratelimited { retry_in_seconds }.into());
+        }
+
+        self.ensure_not_blocked()
+    }
+
+    fn handle_response(
+        &self,
+        response: &reqwest::Response,
+    ) -> Option<ErrorKind> {
+        if response.status() != reqwest::StatusCode::TOO_MANY_REQUESTS {
+            return None;
+        }
+
+        debug!("Received 429 response; blocking");
+
+        let retry_after = response
+            .headers()
+            .get(reqwest::header::RETRY_AFTER)
+            .and_then(|value| value.to_str().ok())
+            .and_then(|value| parse_retry_after(value, SystemTime::now()))
+            .unwrap_or(API_RETRY_AFTER_FALLBACK);
+
+        self.block_for(retry_after);
+
+        Some(ErrorKind::Ratelimited {
+            retry_in_seconds: self.retry_in_seconds().unwrap_or(0),
+        })
+    }
+
+    fn ensure_not_blocked(&self) -> crate::Result<()> {
+        if let Some(retry_in_seconds) = self.retry_in_seconds() {
+            debug!("Hit builtin rate limiter; blocking");
+            return Err(ErrorKind::Ratelimited { retry_in_seconds }.into());
+        }
+
+        Ok(())
+    }
+
+    fn block_for(&self, duration: Duration) {
+        let Some(block_until) = Instant::now().checked_add(duration) else {
+            return;
+        };
+        let mut current_block = self.block_until.lock();
+
+        if current_block.is_none_or(|current| current < block_until) {
+            *current_block = Some(block_until);
+        }
+    }
+
+    fn retry_in_seconds(&self) -> Option<u64> {
+        let mut block_until = self.block_until.lock();
+        let remaining = (*block_until)?.checked_duration_since(Instant::now());
+
+        if let Some(remaining) = remaining
+            && !remaining.is_zero()
+        {
+            return Some(duration_seconds_ceil(remaining));
+        }
+
+        *block_until = None;
+        None
+    }
+}
+
+static GLOBAL_API_RATE_LIMIT: LazyLock<ApiRateLimit> =
+    LazyLock::new(ApiRateLimit::new);
+
+fn parse_retry_after(value: &str, now: SystemTime) -> Option<Duration> {
+    if let Ok(seconds) = value.parse::<u64>() {
+        return Some(Duration::from_secs(seconds));
+    }
+
+    let retry_at = httpdate::parse_http_date(value).ok()?;
+    Some(retry_at.duration_since(now).unwrap_or(Duration::ZERO))
+}
+
+fn duration_seconds_ceil(duration: Duration) -> u64 {
+    duration
+        .as_secs()
+        .saturating_add(u64::from(duration.subsec_nanos() > 0))
+}
+
 fn reqwest_client_builder() -> reqwest::ClientBuilder {
     reqwest::Client::builder()
         .connect_timeout(time::Duration::from_secs(15))
@@ -452,6 +595,10 @@ async fn fetch_advanced_with_client_and_progress(
         .map(|m| (DOWNLOAD_META_HEADER.to_string(), m.to_header_value()));
 
     for attempt in 1..=(FETCH_ATTEMPTS + 1) {
+        if is_api_url {
+            GLOBAL_API_RATE_LIMIT.check()?;
+        }
+
         if let Some(fence_key) = fence_key
             && GLOBAL_FETCH_FENCE.is_blocked(fence_key)
         {
@@ -483,6 +630,13 @@ async fn fetch_advanced_with_client_and_progress(
         let result = req.send().await;
         match result {
             Ok(resp) => {
+                if is_api_url
+                    && let Some(error) =
+                        GLOBAL_API_RATE_LIMIT.handle_response(&resp)
+                {
+                    return Err(error.into());
+                }
+
                 if resp.status().is_server_error() {
                     if let Some(fence_key) = fence_key {
                         GLOBAL_FETCH_FENCE.record_fail(fence_key);