From 5481cc12673ddc3f6be578b6a0eb4a3409e698da Mon Sep 17 00:00:00 2001
From: dawn <90008@klbr.net>
Date: Mon, 22 Jun 2026 02:56:33 +0300
Subject: [PATCH] [api] add concurrency rate limiting to public getRepo

---
 examples/statusphere.rs       | 5 ++++-
 src/api/xrpc/get_repo.rs      | 6 ++++++
 src/api/xrpc/mod.rs           | 4 ++--
 src/api/xrpc/request_crawl.rs | 1 +
 src/config.rs                 | 5 +++++
 src/config/env.rs             | 5 +++++
 src/state.rs                  | 4 +++-
 7 files changed, 26 insertions(+), 4 deletions(-)

diff --git a/examples/statusphere.rs b/examples/statusphere.rs
index dae7036..31a9993 100644
--- a/examples/statusphere.rs
+++ b/examples/statusphere.rs
@@ -125,7 +125,10 @@ async fn handle_stream(index: Arc<StatusIndex>, repos: ReposControl, mut stream:
             let did = rec.did.as_str().to_owned();
             match rec.action.as_str() {
                 "create" | "update" => {
-                    let Some(record) = rec.record else { continue };
+                    let Some(record_raw) = rec.record else { continue };
+                    let Ok(record) = serde_json::from_str::<serde_json::Value>(record_raw.get()) else {
+                        continue;
+                    };
                     let Some(emoji) = record
                         .get("status")
                         .and_then(|v| v.as_str())
diff --git a/src/api/xrpc/get_repo.rs b/src/api/xrpc/get_repo.rs
index 4146080..69901a4 100644
--- a/src/api/xrpc/get_repo.rs
+++ b/src/api/xrpc/get_repo.rs
@@ -44,6 +44,12 @@ pub async fn handle(
         });
     }
 
+    let _permit = hydrant
+        .state
+        .get_repo_semaphore
+        .try_acquire()
+        .map_err(|_| rate_limited(nsid, "GET", "too many concurrent getRepo requests"))?;
+
     let Some(car_bytes) = repo
         .generate_car()
         .await
diff --git a/src/api/xrpc/mod.rs b/src/api/xrpc/mod.rs
index 2cf09c3..382f874 100644
--- a/src/api/xrpc/mod.rs
+++ b/src/api/xrpc/mod.rs
@@ -219,9 +219,9 @@ fn payload_too_large<E: std::error::Error + IntoStatic>(
     }
 }
 
-#[cfg(feature = "relay")]
 fn rate_limited<E: std::error::Error + IntoStatic>(
     nsid: &'static str,
+    method: &'static str,
     message: impl Display,
 ) -> XrpcErrorResponse<E> {
     XrpcErrorResponse {
@@ -230,7 +230,7 @@ fn rate_limited<E: std::error::Error + IntoStatic>(
             error: "RateLimitExceeded".into(),
             message: Some(message.to_smolstr()),
             nsid,
-            method: "POST",
+            method,
             http_status: StatusCode::TOO_MANY_REQUESTS,
         }),
     }
diff --git a/src/api/xrpc/request_crawl.rs b/src/api/xrpc/request_crawl.rs
index 0957c29..644f576 100644
--- a/src/api/xrpc/request_crawl.rs
+++ b/src/api/xrpc/request_crawl.rs
@@ -35,6 +35,7 @@ pub async fn handle(
         if !allowed {
             return Err(rate_limited(
                 nsid,
+                "POST",
                 "daily limit for new PDS sources reached",
             ));
         }
diff --git a/src/config.rs b/src/config.rs
index 793acc3..1b6bbf2 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -111,6 +111,10 @@ pub struct Config {
     /// set via `HYDRANT_ENABLE_BACKLINKS=true`.
     pub enable_backlinks: bool,
 
+    /// maximum number of concurrent public getRepo CAR generations.
+    /// set via `HYDRANT_GET_REPO_CONCURRENCY_LIMIT`; defaults to 2.
+    pub get_repo_concurrency_limit: usize,
+
     /// if `true`, record blocks are not stored; only the index (records, counts, events) is kept.
     /// `getRecord`, `listRecords`, and `getRepo` will return errors when this is enabled.
     /// event stream still functions but create/update events will not include record values.
@@ -275,6 +279,7 @@ impl Default for Config {
             filter_collections: None,
             filter_excludes: None,
             enable_backlinks: false,
+            get_repo_concurrency_limit: 2,
             only_index_links: false,
             new_host_limit: Some(50),
             offline_host_retry_interval: Some(Duration::from_secs(30 * 60)),
diff --git a/src/config/env.rs b/src/config/env.rs
index ede50c7..90292ea 100644
--- a/src/config/env.rs
+++ b/src/config/env.rs
@@ -212,6 +212,10 @@ impl Config {
         });
 
         let enable_backlinks: bool = cfg!("ENABLE_BACKLINKS", defaults.enable_backlinks);
+        let get_repo_concurrency_limit = cfg!(
+            "GET_REPO_CONCURRENCY_LIMIT",
+            defaults.get_repo_concurrency_limit
+        );
         let only_index_links: bool = cfg!("ONLY_INDEX_LINKS", defaults.only_index_links);
         let max_pds_added_per_day = parse_new_host_limit(defaults.new_host_limit);
 
@@ -342,6 +346,7 @@ impl Config {
             filter_collections,
             filter_excludes,
             enable_backlinks,
+            get_repo_concurrency_limit,
             only_index_links,
             new_host_limit: max_pds_added_per_day,
             offline_host_retry_interval: offline_retry_interval,
diff --git a/src/state.rs b/src/state.rs
index 1f14076..e87033b 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -7,7 +7,7 @@ use miette::Result;
 use smol_str::SmolStr;
 #[cfg(feature = "indexer")]
 use tokio::sync::Notify;
-use tokio::sync::watch;
+use tokio::sync::{watch, Semaphore};
 use url::Url;
 
 #[cfg(feature = "firehose-diagnostics")]
@@ -45,6 +45,7 @@ pub struct AppState {
     pub ephemeral: bool,
     pub ephemeral_ttl: Duration,
     pub only_index_links: bool,
+    pub get_repo_semaphore: Semaphore,
 }
 
 impl AppState {
@@ -124,6 +125,7 @@ impl AppState {
             ephemeral: config.ephemeral,
             ephemeral_ttl: config.ephemeral_ttl,
             only_index_links: config.only_index_links,
+            get_repo_semaphore: Semaphore::new(config.get_repo_concurrency_limit),
             throttler: Throttler::new(),
             #[cfg(feature = "relay")]
             pds_daily_limit,
-- 
2.51.2