From ba4e743b44ba0ac558fd2794222505a68828517b Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Thu, 16 Jul 2026 17:50:38 +0200 Subject: [PATCH 1/7] ref(server): Introduce a queue to the concurrency limiter --- objectstore-server/docs/architecture.md | 6 +- objectstore-server/src/config.rs | 30 ++ objectstore-server/src/state.rs | 8 +- objectstore-server/tests/limits.rs | 5 +- objectstore-service/docs/architecture.md | 21 +- objectstore-service/src/concurrency.rs | 337 ++++++++++++++++++++--- objectstore-service/src/service.rs | 71 ++++- 7 files changed, 403 insertions(+), 75 deletions(-) diff --git a/objectstore-server/docs/architecture.md b/objectstore-server/docs/architecture.md index 3f8b49ae..e665fc96 100644 --- a/objectstore-server/docs/architecture.md +++ b/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 diff --git a/objectstore-server/src/config.rs b/objectstore-server/src/config.rs index 3fd4d469..3e8e9925 100644 --- a/objectstore-server/src/config.rs +++ b/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,40 @@ 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 `max_concurrency + concurrency_queue` are rejected + /// with HTTP 429. + /// + /// Both this and `concurrency_queue_timeout` are inert while this is `0` + /// (the default), preserving the immediate-reject behavior. + /// + /// Sizing guidance: `concurrency_queue ≈ permit_release_rate × + /// acceptable_added_latency`. `concurrency_queue_timeout` must stay + /// comfortably below client request timeouts. + 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), } } } diff --git a/objectstore-server/src/state.rs b/objectstore-server/src/state.rs index f7cae5af..f09f872d 100644 --- a/objectstore-server/src/state.rs +++ b/objectstore-server/src/state.rs @@ -61,8 +61,12 @@ 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 service = StorageService::new(backend) + .with_concurrency_limit(config.service.max_concurrency) + .with_concurrency_queue( + config.service.concurrency_queue, + config.service.concurrency_queue_timeout, + ); service.start(); let key_directory = Arc::new(PublicKeyDirectory::from_config(&config.auth).await?); diff --git a/objectstore-server/tests/limits.rs b/objectstore-server/tests/limits.rs index 524d975b..d2ad5a3a 100644 --- a/objectstore-server/tests/limits.rs +++ b/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() diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 73ad50b3..6e969acb 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -175,20 +175,21 @@ 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_limit`] and +[`StorageService::with_concurrency_queue`] for configuration. ## Multipart Uploads diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index ac9bb416..eb26c258 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -23,65 +23,126 @@ const EMITTER_INTERVAL: Duration = Duration::from_secs(1); /// Limits concurrent backend operations and tracks in-flight count. /// -/// Permits are acquired with [`try_acquire`](Self::try_acquire) and -/// automatically returned when the [`ConcurrencyPermit`] is dropped. +/// Permits are acquired with [`acquire`](Self::acquire) or +/// [`try_acquire_many`](Self::try_acquire_many) and automatically returned +/// when the [`ConcurrencyPermit`] is dropped. #[derive(Clone, Debug)] pub struct ConcurrencyLimiter { - semaphore: Arc, - max: u32, + tasks: Arc, + queue: Arc, + tasks_total: u32, + queue_total: u32, + timeout: Duration, released: Arc, } impl ConcurrencyLimiter { /// Creates a new limiter with the given maximum number of permits. + /// + /// By default the queue depth is zero, preserving the original + /// try-or-reject behavior. Use [`with_queue`](Self::with_queue) to + /// enable bounded waiting. pub fn new(max: u32) -> Self { Self { - semaphore: Arc::new(Semaphore::new(max as usize)), - max, + tasks: Arc::new(Semaphore::new(max as usize)), + queue: Arc::new(Semaphore::new(max as usize)), + tasks_total: max, + queue_total: max, + timeout: Duration::ZERO, released: Arc::new(Notify::new()), } } + /// Enables bounded waiting when all execution permits are held. + /// + /// Up to `size` additional callers may park in [`acquire`](Self::acquire) + /// waiting for a permit, each for at most `timeout`. Callers beyond + /// that are rejected immediately. + pub fn with_queue(mut self, size: u32, timeout: Duration) -> Self { + self.queue_total = self.tasks_total + size; + self.queue = Arc::new(Semaphore::new((self.queue_total) as usize)); + self.timeout = timeout; + self + } + + /// Acquires a single concurrency permit, waiting if necessary. + /// + /// If all `max + queue` slots are occupied, returns + /// [`Error::AtCapacity`] immediately. Otherwise, waits up to the + /// configured queue timeout for an execution permit to become + /// available. Returns [`Error::AtCapacity`] on timeout. + pub async fn acquire(&self) -> Result { + let queue_permit = self + .queue + .clone() + .try_acquire_owned() + .map_err(|_| Error::AtCapacity)?; + + let acquire = self.tasks.clone().acquire_owned(); + let task_permit = tokio::time::timeout(self.timeout, acquire) + .await + .map_err(|_| Error::AtCapacity)? + .map_err(|_| Error::AtCapacity)?; + + Ok(ConcurrencyPermit { + task_permit: Some(task_permit), + queue_permit: Some(queue_permit), + released: Arc::clone(&self.released), + }) + } + /// Tries to acquire `count` permits at once as a single bulk reservation. /// /// Returns a [`ConcurrencyPermit`] that releases all `count` permits and - /// notifies waiters on drop, just like single-permit acquisition. + /// notifies waiters on drop, just like single-permit acquisition. Both + /// the execution semaphore and the queue semaphore are checked, so the + /// total-capacity invariant (`max + queue`) is maintained. /// /// Returns [`Error::AtCapacity`] when fewer than `count` permits are available. pub fn try_acquire_many(&self, count: u32) -> Result { - let permit = self - .semaphore + let queue_permit = self + .queue .clone() .try_acquire_many_owned(count) .map_err(|_| Error::AtCapacity)?; + + let task_permit = self + .tasks + .clone() + .try_acquire_many_owned(count) + .map_err(|_| Error::AtCapacity)?; + Ok(ConcurrencyPermit { - permit: Some(permit), + task_permit: Some(task_permit), + queue_permit: Some(queue_permit), released: Arc::clone(&self.released), }) } - /// Tries to acquire a single concurrency permit. - /// - /// Convenience shorthand for `try_acquire_many(1)`. - /// - /// Returns [`Error::AtCapacity`] when all permits are held. - pub fn try_acquire(&self) -> Result { - self.try_acquire_many(1) - } - /// Returns the number of permits currently available. pub fn available_permits(&self) -> u32 { - u32::try_from(self.semaphore.available_permits()).unwrap_or(self.max) + u32::try_from(self.tasks.available_permits()).unwrap_or(self.tasks_total) } /// Returns the number of permits currently held. pub fn used_permits(&self) -> u32 { - self.max - self.available_permits() + self.tasks_total - self.available_permits() } /// Returns the total number of permits. pub fn total_permits(&self) -> u32 { - self.max + self.tasks_total + } + + /// Returns the number of callers currently waiting in the queue. + pub fn queued_permits(&self) -> u32 { + let available = u32::try_from(self.queue.available_permits()).unwrap_or(self.queue_total); + (self.queue_total - available).saturating_sub(self.used_permits()) + } + + /// Returns the configured queue capacity. + pub fn total_queue(&self) -> u32 { + self.queue_total - self.tasks_total } /// Waits until all permits have been returned. @@ -96,19 +157,19 @@ impl ConcurrencyLimiter { } } - /// Periodically calls `emit` with the current in-use count. + /// Periodically calls `emit` with the current in-use and queued counts. /// /// This future runs forever and is intended to be spawned as a background /// task alongside the service. pub async fn run_emitter(&self, mut emit: F) where - F: FnMut(u32) -> Fut, + F: FnMut(u32, u32) -> Fut, Fut: Future, { let mut ticker = tokio::time::interval(EMITTER_INTERVAL); loop { ticker.tick().await; - emit(self.used_permits()).await; + emit(self.used_permits(), self.queued_permits()).await; } } } @@ -117,8 +178,13 @@ impl ConcurrencyLimiter { /// /// Dropping this permit releases it back to the [`ConcurrencyLimiter`] and /// notifies any task waiting in [`ConcurrencyLimiter::wait_all`]. +/// +/// Fields are ordered so that the inner (execution) permit drops before the +/// outer (queue) permit — a waiter that claims the freed queue slot can +/// immediately see the freed execution slot. pub struct ConcurrencyPermit { - permit: Option, + task_permit: Option, + queue_permit: Option, released: Arc, } @@ -130,7 +196,8 @@ impl std::fmt::Debug for ConcurrencyPermit { impl Drop for ConcurrencyPermit { fn drop(&mut self) { - drop(self.permit.take()); + drop(self.task_permit.take()); + drop(self.queue_permit.take()); self.released.notify_waiters(); } } @@ -212,10 +279,10 @@ mod tests { let limiter = ConcurrencyLimiter::new(5); assert_eq!(limiter.available_permits(), 5); - let p1 = limiter.try_acquire().unwrap(); + let p1 = limiter.try_acquire_many(1).unwrap(); assert_eq!(limiter.available_permits(), 4); - let p2 = limiter.try_acquire().unwrap(); + let p2 = limiter.try_acquire_many(1).unwrap(); assert_eq!(limiter.available_permits(), 3); drop(p1); @@ -236,10 +303,10 @@ mod tests { let limiter = ConcurrencyLimiter::new(2); assert_eq!(limiter.used_permits(), 0); - let p1 = limiter.try_acquire().unwrap(); + let p1 = limiter.try_acquire_many(1).unwrap(); assert_eq!(limiter.used_permits(), 1); - let p2 = limiter.try_acquire().unwrap(); + let p2 = limiter.try_acquire_many(1).unwrap(); assert_eq!(limiter.used_permits(), 2); drop(p1); @@ -252,9 +319,9 @@ mod tests { #[test] fn at_capacity_rejects() { let limiter = ConcurrencyLimiter::new(1); - let _permit = limiter.try_acquire().unwrap(); + let _permit = limiter.try_acquire_many(1).unwrap(); - let result = limiter.try_acquire(); + let result = limiter.try_acquire_many(1); assert!(matches!(result, Err(Error::AtCapacity))); } @@ -262,25 +329,29 @@ mod tests { fn permit_recovery_after_drop() { let limiter = ConcurrencyLimiter::new(1); - let permit = limiter.try_acquire().unwrap(); - assert!(limiter.try_acquire().is_err()); + let permit = limiter.try_acquire_many(1).unwrap(); + assert!(limiter.try_acquire_many(1).is_err()); drop(permit); - assert!(limiter.try_acquire().is_ok()); + assert!(limiter.try_acquire_many(1).is_ok()); } #[tokio::test(start_paused = true)] async fn emitter_calls_callback() { let limiter = ConcurrencyLimiter::new(5); - let _permit = limiter.try_acquire().unwrap(); + let _permit = limiter.try_acquire_many(1).unwrap(); - let emitted = Arc::new(AtomicU32::new(0)); - let emitted_clone = Arc::clone(&emitted); + let emitted_in_use = Arc::new(AtomicU32::new(0)); + let emitted_queued = Arc::new(AtomicU32::new(0)); + let in_use_clone = Arc::clone(&emitted_in_use); + let queued_clone = Arc::clone(&emitted_queued); - let emitter = limiter.run_emitter(move |count| { - let emitted = Arc::clone(&emitted_clone); + let emitter = limiter.run_emitter(move |in_use, queued| { + let in_use_ref = Arc::clone(&in_use_clone); + let queued_ref = Arc::clone(&queued_clone); async move { - emitted.store(count, Ordering::Relaxed); + in_use_ref.store(in_use, Ordering::Relaxed); + queued_ref.store(queued, Ordering::Relaxed); } }); @@ -289,14 +360,15 @@ mod tests { _ = tokio::time::sleep(EMITTER_INTERVAL) => {} } - assert_eq!(emitted.load(Ordering::Relaxed), 1); + assert_eq!(emitted_in_use.load(Ordering::Relaxed), 1); + assert_eq!(emitted_queued.load(Ordering::Relaxed), 0); } #[tokio::test] async fn wait_all_resolves_when_permits_returned() { let limiter = ConcurrencyLimiter::new(2); - let p1 = limiter.try_acquire().unwrap(); - let p2 = limiter.try_acquire().unwrap(); + let p1 = limiter.try_acquire_many(1).unwrap(); + let p2 = limiter.try_acquire_many(1).unwrap(); let mut wait = Box::pin(limiter.wait_all()); @@ -315,4 +387,179 @@ mod tests { let wait = Box::pin(limiter.wait_all()); assert!(futures::poll!(wait).is_ready()); } + + // --- Queue tests --- + + #[test] + fn queue_zero_rejects_immediately() { + let limiter = ConcurrencyLimiter::new(2); + assert_eq!(limiter.total_queue(), 0); + + let p1 = limiter.try_acquire_many(1).unwrap(); + let p2 = limiter.try_acquire_many(1).unwrap(); + assert!(matches!( + limiter.try_acquire_many(1), + Err(Error::AtCapacity) + )); + + drop(p1); + assert!(limiter.try_acquire_many(1).is_ok()); + drop(p2); + } + + #[tokio::test(start_paused = true)] + async fn acquire_succeeds_immediately_when_available() { + let limiter = ConcurrencyLimiter::new(2).with_queue(3, Duration::from_secs(5)); + + let permit = limiter.acquire().await.unwrap(); + assert_eq!(limiter.used_permits(), 1); + assert_eq!(limiter.queued_permits(), 0); + drop(permit); + } + + #[tokio::test(start_paused = true)] + async fn acquire_waits_and_succeeds_after_release() { + let limiter = ConcurrencyLimiter::new(1).with_queue(2, Duration::from_secs(5)); + + let held = limiter.acquire().await.unwrap(); + assert_eq!(limiter.used_permits(), 1); + + let limiter2 = limiter.clone(); + let waiter = tokio::spawn(async move { limiter2.acquire().await }); + + tokio::task::yield_now().await; + assert_eq!(limiter.queued_permits(), 1); + + drop(held); + + let permit = waiter.await.unwrap().unwrap(); + assert_eq!(limiter.used_permits(), 1); + assert_eq!(limiter.queued_permits(), 0); + drop(permit); + } + + #[tokio::test(start_paused = true)] + async fn acquire_times_out() { + let limiter = ConcurrencyLimiter::new(1).with_queue(2, Duration::from_secs(1)); + + let _held = limiter.acquire().await.unwrap(); + + let limiter2 = limiter.clone(); + let waiter = tokio::spawn(async move { limiter2.acquire().await }); + + tokio::time::sleep(Duration::from_secs(2)).await; + + let result = waiter.await.unwrap(); + assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(limiter.queued_permits(), 0); + } + + #[tokio::test(start_paused = true)] + async fn acquire_rejects_over_max_plus_queue() { + let limiter = ConcurrencyLimiter::new(1).with_queue(1, Duration::from_secs(5)); + + let _held = limiter.acquire().await.unwrap(); + + let limiter2 = limiter.clone(); + let _waiter = tokio::spawn(async move { limiter2.acquire().await }); + tokio::task::yield_now().await; + + assert_eq!(limiter.queued_permits(), 1); + let result = limiter.acquire().await; + assert!(matches!(result, Err(Error::AtCapacity))); + } + + #[test] + fn try_acquire_many_respects_outer_capacity() { + let limiter = ConcurrencyLimiter::new(1).with_queue(2, Duration::from_secs(5)); + + let _p = limiter.try_acquire_many(1).unwrap(); + assert!(matches!( + limiter.try_acquire_many(1), + Err(Error::AtCapacity) + )); + } + + #[test] + fn try_acquire_many_consumes_n_outer_permits() { + let limiter = ConcurrencyLimiter::new(4).with_queue(4, Duration::from_secs(5)); + + let _bulk = limiter.try_acquire_many(3).unwrap(); + assert_eq!(limiter.used_permits(), 3); + + assert!(matches!( + limiter.try_acquire_many(2), + Err(Error::AtCapacity) + )); + + let _single = limiter.try_acquire_many(1).unwrap(); + assert_eq!(limiter.used_permits(), 4); + } + + #[tokio::test(start_paused = true)] + async fn dropping_parked_acquire_releases_queue_slot() { + let limiter = ConcurrencyLimiter::new(1).with_queue(1, Duration::from_secs(60)); + + let _held = limiter.acquire().await.unwrap(); + + let limiter2 = limiter.clone(); + let waiter = tokio::spawn(async move { limiter2.acquire().await }); + tokio::task::yield_now().await; + assert_eq!(limiter.queued_permits(), 1); + + waiter.abort(); + let _ = waiter.await; + tokio::task::yield_now().await; + + assert_eq!(limiter.queued_permits(), 0); + + let limiter3 = limiter.clone(); + let replacement = tokio::spawn(async move { limiter3.acquire().await }); + tokio::task::yield_now().await; + assert_eq!(limiter.queued_permits(), 1); + drop(replacement); + } + + #[test] + fn queued_permits_reflects_state() { + let limiter = ConcurrencyLimiter::new(2).with_queue(3, Duration::from_secs(5)); + + assert_eq!(limiter.queued_permits(), 0); + let _p1 = limiter.try_acquire_many(1).unwrap(); + assert_eq!(limiter.queued_permits(), 0); + let _p2 = limiter.try_acquire_many(1).unwrap(); + assert_eq!(limiter.queued_permits(), 0); + } + + #[tokio::test(start_paused = true)] + async fn emitter_reports_queued_count() { + let limiter = ConcurrencyLimiter::new(1).with_queue(2, Duration::from_secs(60)); + let _held = limiter.acquire().await.unwrap(); + + let limiter2 = limiter.clone(); + let _waiter = tokio::spawn(async move { limiter2.acquire().await }); + tokio::task::yield_now().await; + + let emitted_in_use = Arc::new(AtomicU32::new(0)); + let emitted_queued = Arc::new(AtomicU32::new(0)); + let in_use_clone = Arc::clone(&emitted_in_use); + let queued_clone = Arc::clone(&emitted_queued); + + let emitter = limiter.run_emitter(move |in_use, queued| { + let in_use_ref = Arc::clone(&in_use_clone); + let queued_ref = Arc::clone(&queued_clone); + async move { + in_use_ref.store(in_use, Ordering::Relaxed); + queued_ref.store(queued, Ordering::Relaxed); + } + }); + + tokio::select! { + _ = emitter => unreachable!("emitter runs forever"), + _ = tokio::time::sleep(EMITTER_INTERVAL) => {} + } + + assert_eq!(emitted_in_use.load(Ordering::Relaxed), 1); + assert_eq!(emitted_queued.load(Ordering::Relaxed), 1); + } } diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index 74479aff..0681f866 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -69,8 +69,12 @@ pub const DEFAULT_CONCURRENCY_LIMIT: u32 = 500; /// A semaphore caps the number of in-flight backend operations. The limit is /// configured via [`with_concurrency_limit`](StorageService::with_concurrency_limit); /// without an explicit value the default is [`DEFAULT_CONCURRENCY_LIMIT`]. -/// Operations that exceed the limit are rejected immediately with -/// [`Error::AtCapacity`]. +/// +/// Optionally, a bounded queue can be enabled with +/// [`with_concurrency_queue`](StorageService::with_concurrency_queue) so that +/// requests exceeding `max` but within `max + queue` wait briefly instead of +/// being rejected immediately. Operations beyond `max + queue` are rejected +/// with [`Error::AtCapacity`]. #[derive(Clone, Debug)] pub struct StorageService { inner: Arc, @@ -100,6 +104,17 @@ impl StorageService { self } + /// Enables a bounded queue for backend concurrency. + /// + /// When all execution permits are held, up to `queue` additional callers + /// wait for a permit, each for at most `timeout`. Must be called after + /// [`with_concurrency_limit`](Self::with_concurrency_limit) and before + /// [`start`](Self::start). + pub fn with_concurrency_queue(mut self, queue: u32, timeout: std::time::Duration) -> Self { + self.concurrency = self.concurrency.with_queue(queue, timeout); + self + } + /// Returns the number of backend task slots currently available. pub fn tasks_available(&self) -> u32 { self.concurrency.available_permits() @@ -143,16 +158,20 @@ impl StorageService { /// Starts background processes for the storage service. /// - /// Currently spawns a task that emits the `service.concurrency.in_use` - /// and `service.concurrency.limit` gauges once per second. + /// Spawns a task that emits concurrency gauges once per second: + /// `service.concurrency.in_use`, `service.concurrency.queued`, + /// `service.concurrency.limit`, and `service.concurrency.queue_limit`. pub fn start(&self) { let concurrency = self.concurrency.clone(); let limit = concurrency.total_permits(); + let queue_limit = concurrency.total_queue(); tokio::spawn(async move { concurrency - .run_emitter(|permits| async move { - objectstore_metrics::gauge!("service.concurrency.in_use" = permits); + .run_emitter(|in_use, queued| async move { + objectstore_metrics::gauge!("service.concurrency.in_use" = in_use); + objectstore_metrics::gauge!("service.concurrency.queued" = queued); objectstore_metrics::gauge!("service.concurrency.limit" = limit); + objectstore_metrics::gauge!("service.concurrency.queue_limit" = queue_limit); }) .await; }); @@ -160,10 +179,10 @@ impl StorageService { /// Spawns a future in a separate task and awaits its result. /// - /// Returns [`Error::AtCapacity`] if the concurrency limit is reached, - /// [`Error::Panic`] if the spawned task panics (the panic message - /// is captured for diagnostics), or [`Error::Dropped`] if the task is - /// dropped before sending its result. + /// Returns [`Error::AtCapacity`] if the concurrency limit or queue is + /// exhausted, [`Error::Panic`] if the spawned task panics (the panic + /// message is captured for diagnostics), or [`Error::Dropped`] if the + /// task is dropped before sending its result. /// /// Emits `service.task.start` (counter) after acquiring a permit and /// `service.task.duration` (distribution) when the task completes, tagged @@ -174,10 +193,34 @@ impl StorageService { T: Send + 'static, F: Future> + Send + 'static, { - let permit = self.concurrency.try_acquire().inspect_err(|_| { - objectstore_metrics::count!("service.concurrency.rejected"); - objectstore_log::warn!("Request rejected: service at capacity"); - })?; + let start = tokio::time::Instant::now(); + let permit = self.concurrency.acquire().await; + let waited = start.elapsed(); + + let permit = match permit { + Ok(p) => { + if waited > std::time::Duration::from_millis(1) { + objectstore_metrics::record!( + "service.concurrency.wait" = waited, + outcome = "acquired", + ); + } + p + } + Err(e) => { + if waited > std::time::Duration::from_millis(1) { + objectstore_metrics::count!("service.concurrency.queue_timeout"); + objectstore_metrics::record!( + "service.concurrency.wait" = waited, + outcome = "timeout", + ); + } else { + objectstore_metrics::count!("service.concurrency.rejected"); + } + objectstore_log::warn!("Request rejected: service at capacity"); + return Err(e); + } + }; crate::concurrency::spawn_metered(operation, permit, f).await } From 151acd01516f6157eaf3ab1336431d7e0e246f82 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Mon, 20 Jul 2026 12:32:02 +0200 Subject: [PATCH 2/7] fix(service): Reject immediately in acquire when max concurrency is zero With max=0 and a non-zero queue, callers were admitted into the queue and waited the full timeout for a task permit that could never exist. Guard at the top of `acquire` so the at-capacity mode stays instant. --- objectstore-service/src/concurrency.rs | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index eb26c258..dfbcdd7f 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -72,6 +72,10 @@ impl ConcurrencyLimiter { /// configured queue timeout for an execution permit to become /// available. Returns [`Error::AtCapacity`] on timeout. pub async fn acquire(&self) -> Result { + if self.tasks_total == 0 { + return Err(Error::AtCapacity); + } + let queue_permit = self .queue .clone() @@ -531,6 +535,16 @@ mod tests { assert_eq!(limiter.queued_permits(), 0); } + #[tokio::test(start_paused = true)] + async fn acquire_rejects_immediately_when_max_is_zero() { + let limiter = ConcurrencyLimiter::new(0).with_queue(5, Duration::from_secs(10)); + + let start = tokio::time::Instant::now(); + let result = limiter.acquire().await; + assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(start.elapsed(), Duration::ZERO); + } + #[tokio::test(start_paused = true)] async fn emitter_reports_queued_count() { let limiter = ConcurrencyLimiter::new(1).with_queue(2, Duration::from_secs(60)); From 3bc02fcea95747b52b970a614707aab924e92eb1 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Mon, 20 Jul 2026 12:34:21 +0200 Subject: [PATCH 3/7] fix(service): Use saturating arithmetic for queue capacity Prevents u32 wrapping when max + queue overflows in release builds. --- objectstore-service/src/concurrency.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index dfbcdd7f..7f43d095 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -59,7 +59,7 @@ impl ConcurrencyLimiter { /// waiting for a permit, each for at most `timeout`. Callers beyond /// that are rejected immediately. pub fn with_queue(mut self, size: u32, timeout: Duration) -> Self { - self.queue_total = self.tasks_total + size; + self.queue_total = self.tasks_total.saturating_add(size); self.queue = Arc::new(Semaphore::new((self.queue_total) as usize)); self.timeout = timeout; self From aa2fec040a3093cb69e641a9afb1dc96c9e705ac Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Mon, 20 Jul 2026 12:39:49 +0200 Subject: [PATCH 4/7] ref(service): Pass Stats struct to concurrency emitter callback Replace positional (u32, u32) arguments with a #[non_exhaustive] Stats struct so new fields can be added without breaking callers. --- objectstore-service/src/concurrency.rs | 37 ++++++++++++++++++++------ objectstore-service/src/service.rs | 21 +++++++-------- 2 files changed, 39 insertions(+), 19 deletions(-) diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index 7f43d095..d3e75eaf 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -21,6 +21,19 @@ use crate::error::{Error, Result}; /// Interval for the periodic metrics emitter. const EMITTER_INTERVAL: Duration = Duration::from_secs(1); +/// Snapshot of concurrency limiter state. +/// +/// Passed to the callback registered via +/// [`ConcurrencyLimiter::run_emitter`]. +#[non_exhaustive] +#[derive(Clone, Copy, Debug)] +pub struct Stats { + /// Number of execution permits currently held. + pub in_use: u32, + /// Number of callers waiting in the queue for a permit. + pub queued: u32, +} + /// Limits concurrent backend operations and tracks in-flight count. /// /// Permits are acquired with [`acquire`](Self::acquire) or @@ -161,19 +174,27 @@ impl ConcurrencyLimiter { } } + /// Returns a snapshot of the current counts. + pub fn stats(&self) -> Stats { + Stats { + in_use: self.used_permits(), + queued: self.queued_permits(), + } + } + /// Periodically calls `emit` with the current in-use and queued counts. /// /// This future runs forever and is intended to be spawned as a background /// task alongside the service. pub async fn run_emitter(&self, mut emit: F) where - F: FnMut(u32, u32) -> Fut, + F: FnMut(Stats) -> Fut, Fut: Future, { let mut ticker = tokio::time::interval(EMITTER_INTERVAL); loop { ticker.tick().await; - emit(self.used_permits(), self.queued_permits()).await; + emit(self.stats()).await; } } } @@ -350,12 +371,12 @@ mod tests { let in_use_clone = Arc::clone(&emitted_in_use); let queued_clone = Arc::clone(&emitted_queued); - let emitter = limiter.run_emitter(move |in_use, queued| { + let emitter = limiter.run_emitter(move |stats| { let in_use_ref = Arc::clone(&in_use_clone); let queued_ref = Arc::clone(&queued_clone); async move { - in_use_ref.store(in_use, Ordering::Relaxed); - queued_ref.store(queued, Ordering::Relaxed); + in_use_ref.store(stats.in_use, Ordering::Relaxed); + queued_ref.store(stats.queued, Ordering::Relaxed); } }); @@ -559,12 +580,12 @@ mod tests { let in_use_clone = Arc::clone(&emitted_in_use); let queued_clone = Arc::clone(&emitted_queued); - let emitter = limiter.run_emitter(move |in_use, queued| { + let emitter = limiter.run_emitter(move |stats| { let in_use_ref = Arc::clone(&in_use_clone); let queued_ref = Arc::clone(&queued_clone); async move { - in_use_ref.store(in_use, Ordering::Relaxed); - queued_ref.store(queued, Ordering::Relaxed); + in_use_ref.store(stats.in_use, Ordering::Relaxed); + queued_ref.store(stats.queued, Ordering::Relaxed); } }); diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index 0681f866..864090b8 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -163,15 +163,14 @@ impl StorageService { /// `service.concurrency.limit`, and `service.concurrency.queue_limit`. pub fn start(&self) { let concurrency = self.concurrency.clone(); - let limit = concurrency.total_permits(); - let queue_limit = concurrency.total_queue(); + objectstore_metrics::gauge!("service.concurrency.limit" = concurrency.total_permits()); + objectstore_metrics::gauge!("service.concurrency.queue_limit" = concurrency.total_queue()); + tokio::spawn(async move { concurrency - .run_emitter(|in_use, queued| async move { - objectstore_metrics::gauge!("service.concurrency.in_use" = in_use); - objectstore_metrics::gauge!("service.concurrency.queued" = queued); - objectstore_metrics::gauge!("service.concurrency.limit" = limit); - objectstore_metrics::gauge!("service.concurrency.queue_limit" = queue_limit); + .run_emitter(|stats| async move { + objectstore_metrics::gauge!("service.concurrency.in_use" = stats.in_use); + objectstore_metrics::gauge!("service.concurrency.queued" = stats.queued); }) .await; }); @@ -179,10 +178,10 @@ impl StorageService { /// Spawns a future in a separate task and awaits its result. /// - /// Returns [`Error::AtCapacity`] if the concurrency limit or queue is - /// exhausted, [`Error::Panic`] if the spawned task panics (the panic - /// message is captured for diagnostics), or [`Error::Dropped`] if the - /// task is dropped before sending its result. + /// Returns [`Error::AtCapacity`] if the concurrency limit is reached, + /// [`Error::Panic`] if the spawned task panics (the panic message + /// is captured for diagnostics), or [`Error::Dropped`] if the task is + /// dropped before sending its result. /// /// Emits `service.task.start` (counter) after acquiring a permit and /// `service.task.duration` (distribution) when the task completes, tagged From 4d9d4365e4db3b3338ac33394bc295bbcfe11b0f Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Mon, 20 Jul 2026 12:44:06 +0200 Subject: [PATCH 5/7] ref(service): Simplify spawn acquire and add timer metric Replace the manual elapsed/match block with a timer guard around acquire and a single inspect_err for rejection logging. --- objectstore-service/src/service.rs | 34 ++++++------------------------ 1 file changed, 6 insertions(+), 28 deletions(-) diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index 864090b8..d7d65865 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -192,35 +192,13 @@ impl StorageService { T: Send + 'static, F: Future> + Send + 'static, { - let start = tokio::time::Instant::now(); - let permit = self.concurrency.acquire().await; - let waited = start.elapsed(); - - let permit = match permit { - Ok(p) => { - if waited > std::time::Duration::from_millis(1) { - objectstore_metrics::record!( - "service.concurrency.wait" = waited, - outcome = "acquired", - ); - } - p - } - Err(e) => { - if waited > std::time::Duration::from_millis(1) { - objectstore_metrics::count!("service.concurrency.queue_timeout"); - objectstore_metrics::record!( - "service.concurrency.wait" = waited, - outcome = "timeout", - ); - } else { - objectstore_metrics::count!("service.concurrency.rejected"); - } - objectstore_log::warn!("Request rejected: service at capacity"); - return Err(e); - } - }; + let timer = objectstore_metrics::timer!("service.concurrency.wait"); + let permit = self.concurrency.acquire().await.inspect_err(|_| { + objectstore_metrics::count!("service.concurrency.rejected"); + objectstore_log::warn!("Request rejected: service at capacity"); + })?; + timer.record(); crate::concurrency::spawn_metered(operation, permit, f).await } From 1682ece5c4a8dce2486955c26117e56fa7d3321e Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Mon, 20 Jul 2026 12:56:56 +0200 Subject: [PATCH 6/7] ref(service): Accept ConcurrencyLimiter instance in builder Replace with_concurrency_limit + with_concurrency_queue with a single with_concurrency that takes a pre-configured ConcurrencyLimiter. The limiter module is now pub so callers can construct it directly. --- objectstore-server/src/state.rs | 12 +++--- objectstore-service/docs/architecture.md | 3 +- objectstore-service/src/lib.rs | 2 +- objectstore-service/src/service.rs | 48 +++++++++--------------- objectstore-service/src/streaming.rs | 9 +++-- 5 files changed, 31 insertions(+), 43 deletions(-) diff --git a/objectstore-server/src/state.rs b/objectstore-server/src/state.rs index f09f872d..61e7c3d2 100644 --- a/objectstore-server/src/state.rs +++ b/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,12 +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) - .with_concurrency_queue( - config.service.concurrency_queue, - config.service.concurrency_queue_timeout, - ); + 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?); diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 6e969acb..96e51980 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -188,8 +188,7 @@ limits fail with [`Error::AtCapacity`](error::Error::AtCapacity). The default execution limit is [`DEFAULT_CONCURRENCY_LIMIT`](service::DEFAULT_CONCURRENCY_LIMIT). See -[`StorageService::with_concurrency_limit`] and -[`StorageService::with_concurrency_queue`] for configuration. +[`StorageService::with_concurrency`] for configuration. ## Multipart Uploads diff --git a/objectstore-service/src/lib.rs b/objectstore-service/src/lib.rs index f2e2eb5c..e33a2f7c 100644 --- a/objectstore-service/src/lib.rs +++ b/objectstore-service/src/lib.rs @@ -3,7 +3,7 @@ #![warn(missing_debug_implementations)] pub mod backend; -mod concurrency; +pub mod concurrency; pub mod error; mod gcp_auth; pub mod id; diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index d7d65865..1e70f2bd 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -34,8 +34,8 @@ pub type DeleteResponse = (); /// Default concurrency limit for [`StorageService`]. /// -/// This value is used when no explicit limit is set via -/// [`StorageService::with_concurrency_limit`]. +/// This value is used when no explicit limiter is set via +/// [`StorageService::with_concurrency`]. pub const DEFAULT_CONCURRENCY_LIMIT: u32 = 500; /// Asynchronous storage service wrapping a single [`Backend`]. @@ -66,15 +66,10 @@ pub const DEFAULT_CONCURRENCY_LIMIT: u32 = 500; /// /// # Concurrency Limit /// -/// A semaphore caps the number of in-flight backend operations. The limit is -/// configured via [`with_concurrency_limit`](StorageService::with_concurrency_limit); -/// without an explicit value the default is [`DEFAULT_CONCURRENCY_LIMIT`]. -/// -/// Optionally, a bounded queue can be enabled with -/// [`with_concurrency_queue`](StorageService::with_concurrency_queue) so that -/// requests exceeding `max` but within `max + queue` wait briefly instead of -/// being rejected immediately. Operations beyond `max + queue` are rejected -/// with [`Error::AtCapacity`]. +/// A [`ConcurrencyLimiter`] caps the number of in-flight backend operations. +/// Pass a custom limiter via +/// [`with_concurrency`](StorageService::with_concurrency); without one the +/// default is [`DEFAULT_CONCURRENCY_LIMIT`] permits with no queue. #[derive(Clone, Debug)] pub struct StorageService { inner: Arc, @@ -95,23 +90,13 @@ impl StorageService { } } - /// Sets the maximum number of concurrent backend operations. - /// - /// Must be called before [`start`](Self::start). Operations beyond this - /// limit are rejected with [`Error::AtCapacity`]. - pub fn with_concurrency_limit(mut self, max: u32) -> Self { - self.concurrency = ConcurrencyLimiter::new(max); - self - } - - /// Enables a bounded queue for backend concurrency. + /// Replaces the default concurrency limiter. /// - /// When all execution permits are held, up to `queue` additional callers - /// wait for a permit, each for at most `timeout`. Must be called after - /// [`with_concurrency_limit`](Self::with_concurrency_limit) and before - /// [`start`](Self::start). - pub fn with_concurrency_queue(mut self, queue: u32, timeout: std::time::Duration) -> Self { - self.concurrency = self.concurrency.with_queue(queue, timeout); + /// Must be called before [`start`](Self::start). Without this, the + /// service uses a limiter with [`DEFAULT_CONCURRENCY_LIMIT`] permits + /// and no queue. + pub fn with_concurrency(mut self, limiter: ConcurrencyLimiter) -> Self { + self.concurrency = limiter; self } @@ -698,7 +683,8 @@ mod tests { fn make_limited_service(limit: u32) -> (StorageService, TestBackend) { let backend = TestBackend::new(GateOnPut::with_pause()); - let service = StorageService::new(Box::new(backend.clone())).with_concurrency_limit(limit); + let service = StorageService::new(Box::new(backend.clone())) + .with_concurrency(ConcurrencyLimiter::new(limit)); (service, backend) } @@ -750,7 +736,7 @@ mod tests { #[tokio::test] async fn tasks_limit_returns_configured_limit() { let backend = Box::new(InMemoryBackend::new("cap")); - let service = StorageService::new(backend).with_concurrency_limit(7); + let service = StorageService::new(backend).with_concurrency(ConcurrencyLimiter::new(7)); assert_eq!(service.tasks_limit(), 7); } @@ -780,8 +766,8 @@ mod tests { #[tokio::test] async fn permits_released_after_panic() { - let service = - StorageService::new(Box::new(TestBackend::new(PanicOnGet))).with_concurrency_limit(1); + let service = StorageService::new(Box::new(TestBackend::new(PanicOnGet))) + .with_concurrency(ConcurrencyLimiter::new(1)); // First operation panics — the permit must still be released. let id = ObjectId::new(make_context(), "panic-permit".into()); diff --git a/objectstore-service/src/streaming.rs b/objectstore-service/src/streaming.rs index e666f055..e7811885 100644 --- a/objectstore-service/src/streaming.rs +++ b/objectstore-service/src/streaming.rs @@ -331,6 +331,7 @@ mod tests { use crate::backend::common::PutResponse; use crate::backend::in_memory::InMemoryBackend; use crate::backend::testing::{Hooks, TestBackend}; + use crate::concurrency::ConcurrencyLimiter; use crate::error::Error; use crate::service::StorageService; use crate::stream::{self, ClientStream}; @@ -344,7 +345,7 @@ mod tests { fn make_service_with_limit(limit: u32) -> StorageService { StorageService::new(Box::new(InMemoryBackend::new("in-memory"))) - .with_concurrency_limit(limit) + .with_concurrency(ConcurrencyLimiter::new(limit)) } fn make_service() -> StorageService { @@ -539,7 +540,8 @@ mod tests { resume: Arc::clone(&resume), in_flight: Arc::clone(&in_flight), }); - let service = StorageService::new(Box::new(gated)).with_concurrency_limit(100); + let service = + StorageService::new(Box::new(gated)).with_concurrency(ConcurrencyLimiter::new(100)); let executor = service.stream().unwrap(); assert_eq!(executor.window(), 10); @@ -593,7 +595,8 @@ mod tests { resume: Arc::clone(&resume), in_flight: Arc::new(AtomicUsize::new(0)), }); - let service = StorageService::new(Box::new(gated)).with_concurrency_limit(1); + let service = + StorageService::new(Box::new(gated)).with_concurrency(ConcurrencyLimiter::new(1)); // Hold the only permit via a blocking insert. let svc = service.clone(); From f7acaac7766ae8fd464695d983e10af3514659a6 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Mon, 20 Jul 2026 12:57:45 +0200 Subject: [PATCH 7/7] docs(server): Simplify concurrency queue config doc comments --- objectstore-server/src/config.rs | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/objectstore-server/src/config.rs b/objectstore-server/src/config.rs index 3e8e9925..73df89d2 100644 --- a/objectstore-server/src/config.rs +++ b/objectstore-server/src/config.rs @@ -529,15 +529,14 @@ pub struct Service { /// 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 `max_concurrency + concurrency_queue` are rejected - /// with HTTP 429. - /// - /// Both this and `concurrency_queue_timeout` are inert while this is `0` - /// (the default), preserving the immediate-reject behavior. + /// Requests beyond that are rejected with HTTP 429. /// /// Sizing guidance: `concurrency_queue ≈ permit_release_rate × - /// acceptable_added_latency`. `concurrency_queue_timeout` must stay - /// comfortably below client request timeouts. + /// acceptable_added_latency`. + /// + /// # Default + /// + /// `0` pub concurrency_queue: u32, /// Maximum time a request may wait in the concurrency queue.