ref(server): Introduce a queue to the concurrency limiter (#562)

The concurrency limiter can now absorb brief capacity bursts by queuing
requests instead of rejecting them immediately. This is implemented
through semaphores: an outer semaphore (capacity `max + queue`) gates
total participants, and an inner semaphore (capacity `max`) gates
execution. Requests that pass the outer gate wait up to a configurable
timeout for an execution permit; requests that exceed both limits are
rejected with `AtCapacity` as before.

With the default `concurrency_queue = 0`, both semaphores have the same
capacity and behavior is identical to the current immediate-reject
model. In this case, there is no queuing and no added latency.

Two new config fields control the queue. Sizing guidance: set the queue
depth to roughly `permit_release_rate × acceptable_added_latency`, and
keep the timeout comfortably below client request timeouts.

```yaml
service:
  max_concurrency: 500
  concurrency_queue: 100
  concurrency_queue_timeout: 1s
```

Batch `stream()` reservations continue to use non-blocking
`try_acquire_many` and do not enter the queue.

Ref FS-447

Author: jan-auer

objectstore-server/docs/architecture.md

@@ -232,9 +232,9 @@ Direct 503 rejection is preferred over readiness-based backpressure:
 Beyond rate limiting and the web concurrency limit, the
 [`StorageService`](objectstore_service::StorageService) enforces a second
 layer of backpressure through a concurrency limit on in-flight backend
-operations, configured via `service.max_concurrency`. When exceeded, requests
-receive HTTP 429. See the [service architecture docs](objectstore_service) for
-details.
+operations, configured via `service.max_concurrency` and related options.
+When exceeded, requests receive HTTP 429. See the [service architecture
+docs](objectstore_service) for details.
 
 ## KEDA Metrics
 

objectstore-server/src/config.rs

@@ -508,6 +508,8 @@ pub struct Config {
 /// # Environment Variables
 ///
 /// - `OS__SERVICE__MAX_CONCURRENCY`
+/// - `OS__SERVICE__CONCURRENCY_QUEUE`
+/// - `OS__SERVICE__CONCURRENCY_QUEUE_TIMEOUT`
 #[derive(Debug, Deserialize, Serialize)]
 #[serde(default)]
 pub struct Service {
@@ -521,12 +523,39 @@ pub struct Service {
     ///
     /// [`DEFAULT_CONCURRENCY_LIMIT`](objectstore_service::service::DEFAULT_CONCURRENCY_LIMIT)
     pub max_concurrency: u32,
+
+    /// Maximum number of requests that may wait for a concurrency permit.
+    ///
+    /// When all `max_concurrency` execution slots are held, up to this many
+    /// additional requests will park and wait (for at most
+    /// `concurrency_queue_timeout`) instead of being rejected immediately.
+    /// Requests beyond that are rejected with HTTP 429.
+    ///
+    /// Sizing guidance: `concurrency_queue ≈ permit_release_rate ×
+    /// acceptable_added_latency`.
+    ///
+    /// # Default
+    ///
+    /// `0`
+    pub concurrency_queue: u32,
+
+    /// Maximum time a request may wait in the concurrency queue.
+    ///
+    /// Ignored when `concurrency_queue` is `0`.
+    ///
+    /// # Default
+    ///
+    /// `1s`
+    #[serde(with = "humantime_serde")]
+    pub concurrency_queue_timeout: Duration,
 }
 
 impl Default for Service {
     fn default() -> Self {
         Self {
             max_concurrency: objectstore_service::service::DEFAULT_CONCURRENCY_LIMIT,
+            concurrency_queue: 0,
+            concurrency_queue_timeout: Duration::from_secs(1),
         }
     }
 }

objectstore-server/src/state.rs

@@ -9,6 +9,7 @@ use std::time::Duration;
 use anyhow::Result;
 use bytes::Bytes;
 use futures_util::Stream;
+use objectstore_service::concurrency::ConcurrencyLimiter;
 use objectstore_service::id::ObjectContext;
 use objectstore_service::{StorageService, backend};
 use tokio::runtime::Handle;
@@ -61,8 +62,11 @@ impl Services {
         tokio::spawn(track_allocator_metrics(config.runtime.metrics_interval));
 
         let backend = backend::from_config(config.storage.clone()).await?;
-        let service =
-            StorageService::new(backend).with_concurrency_limit(config.service.max_concurrency);
+        let concurrency = ConcurrencyLimiter::new(config.service.max_concurrency).with_queue(
+            config.service.concurrency_queue,
+            config.service.concurrency_queue_timeout,
+        );
+        let service = StorageService::new(backend).with_concurrency(concurrency);
         service.start();
 
         let key_directory = Arc::new(PublicKeyDirectory::from_config(&config.auth).await?);

objectstore-server/tests/limits.rs

@@ -629,7 +629,10 @@ async fn test_batch_at_capacity_returns_429() -> Result<()> {
     // With max_concurrency=0 the service has no permits available, so
     // BatchExecutor::new() returns AtCapacity and the endpoint responds 429.
     let server = TestServer::with_config(Config {
-        service: Service { max_concurrency: 0 },
+        service: Service {
+            max_concurrency: 0,
+            ..Default::default()
+        },
         auth: AuthZ {
             enforce: false,
             ..Default::default()

objectstore-service/docs/architecture.md

@@ -175,20 +175,20 @@ during metadata creation by the server.
 
 # Backpressure
 
-The service applies backpressure to protect backends from overload. Rather than
-queueing work when capacity is exhausted, the service rejects operations
-immediately so the caller can shed load or retry.
+The service applies backpressure to protect backends from overload and to
+prevent exhaustion of internal resources such as memory.
 
 ## Concurrency Limit
 
-A semaphore caps the total number of in-flight backend operations across all
-callers. A permit is acquired before each operation is spawned and held until
-the task completes — including on panic — so the limit counts *running*
-operations, not queued ones. When no permits are available, the operation fails
-with [`Error::AtCapacity`](error::Error::AtCapacity).
+A concurrency limiter caps in-flight
+backend operations. When all execution permits are held, new operations are
+queued — adding latency instead of rejecting immediately. The queue itself is
+bounded in both depth and time: operations that cannot be served within those
+limits fail with [`Error::AtCapacity`](error::Error::AtCapacity).
 
-The default limit is [`DEFAULT_CONCURRENCY_LIMIT`](service::DEFAULT_CONCURRENCY_LIMIT). Callers can override it via
-[`StorageService::with_concurrency_limit`].
+The default execution limit is
+[`DEFAULT_CONCURRENCY_LIMIT`](service::DEFAULT_CONCURRENCY_LIMIT). See
+[`StorageService::with_concurrency`] for configuration.
 
 ## Multipart Uploads