sync: drop upstream's subscribeRepos rate limit
Reverts upstream 8fd032f ("Rate limit subscribeRepos", in v0.7.0), which limited com.atproto.sync.subscribeRepos to 5 connections per 60s per IP. Ana asked for the fork to run without it. Every other rate limit is kept. No test covered the limit (the test env sets DISABLE_RATE_LIMITING), so no tests change. Upstream's CHANGELOG entry for it is left as history.
97 unmodified lines989910010110210210310410517 unmodified lines12312412512612812712812913077 unmodified lines208209210211214215216217212213214215272 unmodified linesExpand all488489490491498492493494495304 unmodified linesExpand all97 unmodified lines SsoCallback, SsoUnlink, OAuthRegisterComplete, HandleVerification, SubscribeRepos,}
impl RateLimitKind { const fn key_prefix(&self) -> &'static str {17 unmodified lines Self::SsoCallback => "sso_callback", Self::SsoUnlink => "sso_unlink", Self::OAuthRegisterComplete => "oauth_register_complete", Self::HandleVerification => "handle_verification", Self::SubscribeRepos => "subscribe_repos", } }
pub const fn params(&self) -> RateLimitParams {77 unmodified lines Self::HandleVerification => RateLimitParams { limit: 10, window_ms: 60_000, }, Self::SubscribeRepos => RateLimitParams { limit: 5, window_ms: 60_000, }, } }}
272 unmodified linesExpand all RateLimitKind::SsoCallback => &self.rate_limiters.sso_callback, RateLimitKind::SsoUnlink => &self.rate_limiters.sso_unlink, RateLimitKind::OAuthRegisterComplete => &self.rate_limiters.oauth_register_complete, RateLimitKind::HandleVerification => &self.rate_limiters.handle_verification, RateLimitKind::SubscribeRepos => &self.rate_limiters.subscribe_repos, };
if limiter.check_key(&client_ip.to_string()).is_err() { crate::metrics::record_rate_limit_rejection(limiter_name);304 unmodified linesExpand all6 unmodified lines78910111112131414 unmodified lines293031323433343536278 unmodified linesExpand all6 unmodified linesuse std::sync::atomic::{AtomicUsize, Ordering};use tokio::sync::broadcast::error::RecvError;use tracing::{error, info, warn};use tranquil_db_traits::SequenceNumber;use tranquil_pds::rate_limit::{RateLimited, SubscribeReposLimit};use tranquil_pds::state::AppState;use tranquil_pds::sync::firehose::SequencedEvent;use tranquil_pds::sync::frame::{ErrorFrameName, InfoFrameName};use tranquil_pds::sync::util::{14 unmodified linespub async fn subscribe_repos( ws: WebSocketUpgrade, State(state): State<AppState>, Query(params): Query<SubscribeReposParams>, _rate_limit: RateLimited<SubscribeReposLimit>,) -> Response { ws.on_upgrade(move |socket| handle_socket(socket, state, params))}
278 unmodified linesExpand all113 unmodified linesExpand all114115116117118119120121122118119120121156 unmodified linesExpand all113 unmodified linesExpand allimpl RateLimitPolicy for HandleVerificationLimit { const KIND: RateLimitKind = RateLimitKind::HandleVerification;}
pub struct SubscribeReposLimit;impl RateLimitPolicy for SubscribeReposLimit { const KIND: RateLimitKind = RateLimitKind::SubscribeRepos;}
pub trait RateLimitRejection: IntoResponse + Send + 'static { fn new() -> Self;}
156 unmodified linesExpand all37 unmodified lines38394041424243444523 unmodified lines69707172747374757685 unmodified lines37 unmodified lines pub sso_callback: Arc<KeyedRateLimiter>, pub sso_unlink: Arc<KeyedRateLimiter>, pub oauth_register_complete: Arc<KeyedRateLimiter>, pub handle_verification: Arc<KeyedRateLimiter>, pub subscribe_repos: Arc<KeyedRateLimiter>,}
impl Default for RateLimiters { fn default() -> Self {23 unmodified lines sso_callback: keyed_limiter(RateLimitKind::SsoCallback), sso_unlink: keyed_limiter(RateLimitKind::SsoUnlink), oauth_register_complete: keyed_limiter(RateLimitKind::OAuthRegisterComplete), handle_verification: keyed_limiter(RateLimitKind::HandleVerification), subscribe_repos: keyed_limiter(RateLimitKind::SubscribeRepos), } }
pub fn override_limit(kind: RateLimitKind, limit: u32) -> Arc<KeyedRateLimiter> {85 unmodified lines