feat(server): Implement throughput rate limits

Author: jan-auer

Cargo.lock

@@ -52,6 +52,7 @@ tower-http = { version = "0.6.6", default-features = false, features = [
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 uuid = { workspace = true, features = ["v7"] }
+papaya = "0.2.3"
 
 [dev-dependencies]
 nix = { version = "0.30.1", features = ["signal"] }

objectstore-server/Cargo.toml

@@ -46,6 +46,7 @@ use serde::{Deserialize, Serialize};
 use tracing::level_filters::LevelFilter;
 
 use crate::killswitches::Killswitches;
+use crate::rate_limits::RateLimits;
 
 /// Environment variable prefix for all configuration options.
 const ENV_PREFIX: &str = "OS__";
@@ -870,6 +871,9 @@ pub struct Config {
 
     /// A list of matchers for requests to discard without processing.
     pub killswitches: Killswitches,
+
+    /// Definitions for rate limits to enforce on incoming requests.
+    pub rate_limits: RateLimits,
 }
 
 impl Default for Config {
@@ -890,6 +894,7 @@ impl Default for Config {
             metrics: Metrics::default(),
             auth: AuthZ::default(),
             killswitches: Killswitches::default(),
+            rate_limits: RateLimits::default(),
         }
     }
 }

objectstore-server/src/config.rs

@@ -15,6 +15,7 @@ use crate::state::ServiceState;
 pub enum ObjectRejection {
     Path(PathRejection),
     Killswitched,
+    RateLimited,
 }
 
 impl IntoResponse for ObjectRejection {
@@ -26,6 +27,11 @@ impl IntoResponse for ObjectRejection {
                 "Object access is disabled for this scope through killswitches",
             )
                 .into_response(),
+            ObjectRejection::RateLimited => (
+                axum::http::StatusCode::TOO_MANY_REQUESTS,
+                "Object access is rate limited",
+            )
+                .into_response(),
         }
     }
 }
@@ -53,6 +59,10 @@ impl FromRequestParts<ServiceState> for Xt<ObjectId> {
             return Err(ObjectRejection::Killswitched);
         }
 
+        if !state.rate_limiter.check(id.context()) {
+            return Err(ObjectRejection::RateLimited);
+        }
+
         Ok(Xt(id))
     }
 }
@@ -114,6 +124,10 @@ impl FromRequestParts<ServiceState> for Xt<ObjectContext> {
             return Err(ObjectRejection::Killswitched);
         }
 
+        if !state.rate_limiter.check(&context) {
+            return Err(ObjectRejection::RateLimited);
+        }
+
         Ok(Xt(context))
     }
 }

objectstore-server/src/extractors/id.rs

@@ -11,5 +11,6 @@ pub mod extractors;
 pub mod healthcheck;
 pub mod killswitches;
 pub mod observability;
+pub mod rate_limits;
 pub mod state;
 pub mod web;