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..73df89d2 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,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), } } } diff --git a/objectstore-server/src/state.rs b/objectstore-server/src/state.rs index f7cae5af..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,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?); 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..96e51980 100644 --- a/objectstore-service/docs/architecture.md +++ b/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 diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index ac9bb416..d3e75eaf 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -21,67 +21,145 @@ 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 [`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.saturating_add(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 { + if self.tasks_total == 0 { + return Err(Error::AtCapacity); + } + + 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 +174,27 @@ impl ConcurrencyLimiter { } } - /// Periodically calls `emit` with the current in-use count. + /// 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) -> Fut, + F: FnMut(Stats) -> Fut, Fut: Future, { let mut ticker = tokio::time::interval(EMITTER_INTERVAL); loop { ticker.tick().await; - emit(self.used_permits()).await; + emit(self.stats()).await; } } } @@ -117,8 +203,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 +221,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 +304,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 +328,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 +344,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 +354,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 |stats| { + 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(stats.in_use, Ordering::Relaxed); + queued_ref.store(stats.queued, Ordering::Relaxed); } }); @@ -289,14 +385,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 +412,189 @@ 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 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)); + 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 |stats| { + let in_use_ref = Arc::clone(&in_use_clone); + let queued_ref = Arc::clone(&queued_clone); + async move { + in_use_ref.store(stats.in_use, Ordering::Relaxed); + queued_ref.store(stats.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/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 74479aff..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,11 +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`]. -/// Operations that exceed the limit are rejected immediately 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, @@ -91,12 +90,13 @@ impl StorageService { } } - /// Sets the maximum number of concurrent backend operations. + /// Replaces the default concurrency limiter. /// - /// 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); + /// 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 } @@ -143,16 +143,19 @@ 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(); + 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(|permits| async move { - objectstore_metrics::gauge!("service.concurrency.in_use" = permits); - objectstore_metrics::gauge!("service.concurrency.limit" = 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; }); @@ -174,11 +177,13 @@ impl StorageService { T: Send + 'static, F: Future> + Send + 'static, { - let permit = self.concurrency.try_acquire().inspect_err(|_| { + 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 } @@ -678,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) } @@ -730,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); } @@ -760,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();