fix(server): Prevent Envoy retries of capacity rejections (#732)

Since #703, capacity rejections return 503. Our Envoy routes retry on
`gateway-error`, which includes 503. As a result, a saturated instance
gets retries of its own rejections.

This PR sets `x-envoy-ratelimited: true` on rate limit and capacity
rejections. Envoy does not retry a response that has this header. Real
gateway errors are still retried, and outlier detection still counts the
503s.

Ref FS-577, INC-2579

<!-- junior-request-attribution:start -->
via **async jauer**.
<!-- junior-request-attribution:end -->

<!-- junior-session-footer:start -->
<!-- junior-conversation-id:slack%3AD0AQCS7VDC2%3A1792685297.094279 -->

--

[View Junior
Session](https://junior-prod.sentry.dev/conversations/slack%3AD0AQCS7VDC2%3A1792685297.094279)
[[Sentry]](https://sentry.sentry.io/explore/conversations/slack%3AD0AQCS7VDC2%3A1792685297.094279/?project=4510944073809921)

<!-- junior-session-footer:end -->

---------

Co-authored-by: sentry-junior[bot] <264270552+sentry-junior[bot]@users.noreply.github.com>
Co-authored-by: Jan Michael Auer <jan.auer@sentry.io>

Author: sentry-junior[bot]

objectstore-server/docs/architecture.md

@@ -151,6 +151,11 @@ scope, then explicit rules. The first exceeded limit determines the response.
 Responses identify only capacity protection or an exceeded traffic allowance;
 the specific limit is recorded in logs and metrics.
 
+Rejected requests, including those rejected by the web concurrency limit and
+service backpressure below, carry `x-envoy-ratelimited: true`. Envoy does not retry responses with
+this header, even if the route retries on `gateway-error`. Outlier detection
+still counts the 503 responses.
+
 ### Throughput
 
 Throughput limits use **token bucket** rate limiting with configurable burst.

objectstore-server/src/endpoints/common.rs

@@ -13,6 +13,7 @@ use thiserror::Error;
 
 use crate::auth::AuthError;
 use crate::extractors::batch::BatchError;
+use crate::rate_limits::LimitExceeded;
 
 /// A JSON error response returned by the API.
 #[derive(Serialize, Deserialize, Debug)]
@@ -151,7 +152,7 @@ impl ApiError {
                 ServiceErrorKind::RangeNotSatisfiable { .. } => StatusCode::RANGE_NOT_SATISFIABLE,
                 ServiceErrorKind::UploadOffsetMismatch { .. } => StatusCode::CONFLICT,
                 ServiceErrorKind::UploadSessionGone => StatusCode::GONE,
-                ServiceErrorKind::AtCapacity => StatusCode::SERVICE_UNAVAILABLE,
+                ServiceErrorKind::AtCapacity => LimitExceeded::Capacity.status(),
                 ServiceErrorKind::Unsupported => StatusCode::NOT_IMPLEMENTED,
                 ServiceErrorKind::BackendRateLimited => StatusCode::TOO_MANY_REQUESTS,
                 ServiceErrorKind::BackendTimeout | ServiceErrorKind::BackendUnavailable => {
@@ -189,6 +190,12 @@ impl ApiError {
 
 impl IntoResponse for ApiError {
     fn into_response(self) -> Response {
+        if let ApiError::Service(e) = &self
+            && e.kind() == ServiceErrorKind::AtCapacity
+        {
+            return LimitExceeded::Capacity.into_response();
+        }
+
         self.capture();
         let body = ApiErrorResponse::from_error(&self);
         (self.status(), Json(body)).into_response()

objectstore-server/src/extractors/id.rs

@@ -30,9 +30,7 @@ impl IntoResponse for ObjectRejection {
                 "Object access is disabled for this scope through killswitches",
             )
                 .into_response(),
-            ObjectRejection::RateLimited(error) => {
-                (error.status(), error.to_string()).into_response()
-            }
+            ObjectRejection::RateLimited(error) => error.into_response(),
         }
     }
 }

objectstore-server/src/rate_limits.rs

@@ -19,16 +19,30 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::task::{Context, Poll};
 use std::time::{Duration, Instant};
 
+use axum::response::{IntoResponse, Response};
 use bytes::Bytes;
 use futures_util::Stream;
-use http::StatusCode;
+use http::{HeaderValue, StatusCode};
 use objectstore_service::id::ObjectContext;
 use objectstore_types::scope::Scopes;
 use objectstore_types::usecase::Usecase;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
 
+/// Response header that prevents Envoy from retrying a response.
+///
+/// Envoy does not retry a response with this header, even if the route retries on
+/// `gateway-error` or `5xx`. Only the `envoy-ratelimited` retry condition retries it. Outlier
+/// detection still counts the response. See the [Envoy router
+/// documentation](https://www.envoyproxy.io/docs/envoy/latest/configuration/http/http_filters/router_filter#x-envoy-ratelimited).
+pub const HEADER_ENVOY_RATELIMITED: &str = "x-envoy-ratelimited";
+
 /// Classifies a rejected request as capacity protection or an exceeded traffic allowance.
+///
+/// This is the single conversion for rate limit and capacity rejections into HTTP responses.
+/// Its [`IntoResponse`] implementation sets the status from [`status`](Self::status) and marks
+/// the response with [`HEADER_ENVOY_RATELIMITED`], so that Envoy does not retry a rejection and
+/// add load to an instance that is already at capacity.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
 pub enum LimitExceeded {
     /// A general capacity protection limit was exceeded.
@@ -49,6 +63,16 @@ impl LimitExceeded {
     }
 }
 
+impl IntoResponse for LimitExceeded {
+    fn into_response(self) -> Response {
+        let mut response = (self.status(), self.to_string()).into_response();
+        response
+            .headers_mut()
+            .insert(HEADER_ENVOY_RATELIMITED, HeaderValue::from_static("true"));
+        response
+    }
+}
+
 /// Identifies which rate limit triggered a rejection.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum RateLimitRejection {

objectstore-server/src/web/middleware.rs

@@ -11,6 +11,7 @@ use tower_http::set_header::SetResponseHeaderLayer;
 use crate::endpoints;
 use crate::extractors::downstream_service::DownstreamService;
 use crate::extractors::request_time::RequestTime;
+use crate::rate_limits::LimitExceeded;
 use crate::web::RequestCounter;
 use crate::web::metrics_body::{EmitMetricsGuard, MetricsBody};
 use crate::web::sentry_body::SentryBody;
@@ -41,7 +42,7 @@ pub async fn limit_web_concurrency(
         let service = request.extract_parts::<DownstreamService>().await.unwrap();
         objectstore_metrics::count!("web.concurrency.rejected", service = service.to_string());
         objectstore_log::warn!("Request rejected: web concurrency limit reached");
-        return StatusCode::SERVICE_UNAVAILABLE.into_response();
+        return LimitExceeded::Capacity.into_response();
     }
 
     next.run(request).await

objectstore-server/tests/limits.rs

@@ -15,6 +15,17 @@ use objectstore_server::rate_limits::{
 use objectstore_server::usecases::{ExpirationConfig, PolicyConfig, UseCaseConfig, UseCases};
 use objectstore_test::server::TestServer;
 
+/// Asserts that a rejection tells Envoy not to retry it.
+fn assert_no_retry(response: &reqwest::Response) {
+    assert_eq!(
+        response
+            .headers()
+            .get(objectstore_server::rate_limits::HEADER_ENVOY_RATELIMITED)
+            .expect("rejection must prevent Envoy retries"),
+        "true"
+    );
+}
+
 #[tokio::test]
 async fn test_web_concurrency_limit() -> Result<()> {
     // Setting max_requests = 0 means every non-exempt request is rejected immediately
@@ -37,6 +48,8 @@ async fn test_web_concurrency_limit() -> Result<()> {
         reqwest::StatusCode::SERVICE_UNAVAILABLE,
         "expected 503 for regular request at limit"
     );
+    assert_no_retry(&response);
+    assert_eq!(response.text().await?, "service at capacity");
 
     // Health endpoints need no auth and bypass the concurrency limit.
     for path in ["/health", "/ready"] {
@@ -52,6 +65,33 @@ async fn test_web_concurrency_limit() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_service_concurrency_limit() -> Result<()> {
+    // With max_concurrency = 0, every operation fails to acquire a service permit.
+    let server = TestServer::with_config(Config {
+        service: Service {
+            max_concurrency: 0,
+            ..Default::default()
+        },
+        auth: AuthZ {
+            enforce: false,
+            ..Default::default()
+        },
+        ..Default::default()
+    })
+    .await;
+
+    let response = reqwest::Client::new()
+        .get(server.url("/v1/objects/test/org=1/key"))
+        .send()
+        .await?;
+    assert_eq!(response.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE);
+    assert_no_retry(&response);
+    assert_eq!(response.text().await?, "service at capacity");
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_killswitches() -> Result<()> {
     let server = TestServer::with_config(Config {
@@ -134,6 +174,7 @@ async fn test_throughput_global_rps_limit() -> Result<()> {
         .send()
         .await?;
     assert_eq!(response.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE);
+    assert_no_retry(&response);
     assert_eq!(response.text().await?, "service at capacity");
 
     // Refill bucket
@@ -305,6 +346,7 @@ async fn test_throughput_rule() -> Result<()> {
         .send()
         .await?;
     assert_eq!(response.status(), reqwest::StatusCode::TOO_MANY_REQUESTS);
+    assert_no_retry(&response);
     assert_eq!(response.text().await?, "rate limit exceeded");
 
     // Different usecase should not be affected by rule