From c393ce80b33d278796427b59c3746c6e87966d2a Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Mon, 20 Jul 2026 15:36:11 +0200 Subject: [PATCH 01/11] ref(service): Replace upfront batch reservation with per-op bulk permits --- objectstore-server/src/config.rs | 16 ++ objectstore-server/src/endpoints/batch.rs | 9 +- objectstore-server/src/state.rs | 10 +- objectstore-server/tests/limits.rs | 17 +- objectstore-service/docs/architecture.md | 19 +- objectstore-service/src/concurrency.rs | 304 +++++++++++++++++----- objectstore-service/src/service.rs | 46 ++-- objectstore-service/src/streaming.rs | 221 +++++++--------- 8 files changed, 392 insertions(+), 250 deletions(-) diff --git a/objectstore-server/src/config.rs b/objectstore-server/src/config.rs index 73df89d2..afa20247 100644 --- a/objectstore-server/src/config.rs +++ b/objectstore-server/src/config.rs @@ -548,6 +548,21 @@ pub struct Service { /// `1s` #[serde(with = "humantime_serde")] pub concurrency_queue_timeout: Duration, + + /// Percentage of `max_concurrency` available to bulk operations + /// (e.g. parallelized batch requests). + /// + /// This sets a safe operating point: below this level there is + /// little-to-no performance degradation, leaving room for more tasks + /// to be admitted via the queue before rejection is necessary. + /// + /// Clamped to 0..=100. At 100, bulk operations can use all execution + /// slots. Lower values leave headroom for single-object requests. + /// + /// # Default + /// + /// `60` + pub bulk_concurrency_pct: u32, } impl Default for Service { @@ -556,6 +571,7 @@ impl Default for Service { max_concurrency: objectstore_service::service::DEFAULT_CONCURRENCY_LIMIT, concurrency_queue: 0, concurrency_queue_timeout: Duration::from_secs(1), + bulk_concurrency_pct: 60, } } } diff --git a/objectstore-server/src/endpoints/batch.rs b/objectstore-server/src/endpoints/batch.rs index 11ff1091..8cbee6d2 100644 --- a/objectstore-server/src/endpoints/batch.rs +++ b/objectstore-server/src/endpoints/batch.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use axum::Router; use axum::extract::{DefaultBodyLimit, State}; -use axum::response::{IntoResponse, Response}; +use axum::response::Response; use axum::routing; use bytes::{Bytes, BytesMut}; use futures::StreamExt; @@ -63,12 +63,7 @@ async fn batch( Xt(context): Xt, requests: BatchOperationStream, ) -> Response { - let batch = match state.service.stream() { - Ok(b) => b, - Err(e) => return ApiError::Service(e).into_response(), - }; - - objectstore_metrics::gauge!("service.batch.window" = batch.window()); + let batch = state.service.stream(); // Step 1: parse multipart fields → (idx, Result) let parsed = requests.0.map(|r| r.map_err(ApiError::from)).enumerate(); diff --git a/objectstore-server/src/state.rs b/objectstore-server/src/state.rs index 61e7c3d2..2a3fb0b2 100644 --- a/objectstore-server/src/state.rs +++ b/objectstore-server/src/state.rs @@ -62,10 +62,12 @@ impl Services { tokio::spawn(track_allocator_metrics(config.runtime.metrics_interval)); let backend = backend::from_config(config.storage.clone()).await?; - let concurrency = ConcurrencyLimiter::new(config.service.max_concurrency).with_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, + ) + .with_bulk(config.service.bulk_concurrency_pct); let service = StorageService::new(backend).with_concurrency(concurrency); service.start(); diff --git a/objectstore-server/tests/limits.rs b/objectstore-server/tests/limits.rs index d2ad5a3a..f6df29ea 100644 --- a/objectstore-server/tests/limits.rs +++ b/objectstore-server/tests/limits.rs @@ -625,9 +625,10 @@ async fn test_bandwidth_scope_pct_limit() -> Result<()> { } #[tokio::test] -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. +async fn test_batch_at_capacity_returns_per_op_error() -> Result<()> { + // With max_concurrency=0 no batch operations can acquire a permit. + // Each operation fails individually with AtCapacity inside the + // multipart response (the endpoint itself returns 200). let server = TestServer::with_config(Config { service: Service { max_concurrency: 0, @@ -660,8 +661,14 @@ async fn test_batch_at_capacity_returns_429() -> Result<()> { assert_eq!( response.status(), - reqwest::StatusCode::TOO_MANY_REQUESTS, - "expected 429 when service has no available permits" + reqwest::StatusCode::OK, + "batch endpoint returns 200 with per-operation errors" + ); + + let body = response.text().await?; + assert!( + body.contains("429"), + "expected per-operation 429 in multipart body, got: {body}" ); Ok(()) diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 96e51980..0b745d46 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -201,11 +201,20 @@ Multipart operations share the same concurrency limiter as regular operations. ## Streaming Concurrency The [`streaming`](streaming) module provides [`StreamExecutor`](streaming::StreamExecutor) -for running a stream of operations concurrently within a bounded window. It is -intended for efficient handling of batch requests, where multiple operations -arrive together and should be dispatched in parallel rather than sequentially. -See the [module documentation](streaming) for the window formula, permit -reservation, lazy pulling, memory bounds, and concurrency model. +for running a stream of operations concurrently. It is intended for efficient +handling of batch requests, where multiple operations arrive together and should +be dispatched in parallel rather than sequentially. + +Each streaming operation acquires a "bulk" permit. These permits set a safe +operating point: below this level there is little-to-no performance degradation, +leaving room for more tasks to be admitted via the queue before rejection is +necessary. The percentage is configurable. + +Normal requests never touch the bulk semaphore, so they can always use 100% of +permits when no bulk operations are running. Tokio's FIFO semaphore fairness +ensures parked bulk operations cannot be starved by sustained normal traffic. + +See the [module documentation](streaming) for details. ## Further Plans diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index d3e75eaf..b84c6695 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -14,7 +14,7 @@ use std::time::Duration; use futures_util::FutureExt; use sentry::{Hub, SentryFutureExt, TransactionContext}; -use tokio::sync::{Notify, OwnedSemaphorePermit, Semaphore}; +use tokio::sync::{AcquireError, Notify, OwnedSemaphorePermit, Semaphore}; use crate::error::{Error, Result}; @@ -32,19 +32,29 @@ pub struct Stats { pub in_use: u32, /// Number of callers waiting in the queue for a permit. pub queued: u32, + /// Number of bulk operations currently in flight. + pub bulk_in_use: u32, } /// Limits concurrent backend operations and tracks in-flight count. /// /// Permits are acquired with [`acquire`](Self::acquire) or -/// [`try_acquire_many`](Self::try_acquire_many) and automatically returned -/// when the [`ConcurrencyPermit`] is dropped. +/// [`acquire_bulk`](Self::acquire_bulk) and automatically returned when +/// the [`ConcurrencyPermit`] is dropped. +/// +/// Bulk operations use a separate budget semaphore that limits how many +/// execution slots they may occupy. This is intended as a safe operating +/// point — below this level there should be little-to-no performance +/// degradation, leaving room for more tasks to be admitted via the queue +/// before rejection is necessary. #[derive(Clone, Debug)] pub struct ConcurrencyLimiter { tasks: Arc, queue: Arc, + bulk: Arc, tasks_total: u32, queue_total: u32, + bulk_total: u32, timeout: Duration, released: Arc, } @@ -52,15 +62,20 @@ pub struct ConcurrencyLimiter { 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. + /// By default the queue depth is zero, preserving the original try-or-reject behavior. Use + /// [`with_queue`](Self::with_queue) to enable bounded waiting. + /// + /// The bulk budget defaults to 100% of `max` (no restriction); use + /// [`with_bulk_percent`](Self::with_bulk_percent) to set a safe operating point for bulk + /// traffic. pub fn new(max: u32) -> Self { Self { tasks: Arc::new(Semaphore::new(max as usize)), queue: Arc::new(Semaphore::new(max as usize)), + bulk: Arc::new(Semaphore::new(max as usize)), tasks_total: max, queue_total: max, + bulk_total: max, timeout: Duration::ZERO, released: Arc::new(Notify::new()), } @@ -78,6 +93,21 @@ impl ConcurrencyLimiter { self } + /// Sets the bulk concurrency budget as a percentage of `max`. + /// + /// `percent` is clamped to `0..=100`. At `100` (the default), bulk + /// operations can use all execution slots. Lower values set a safe + /// operating point below which there is little-to-no performance + /// degradation, allowing more tasks to queue before rejection is + /// necessary — e.g. `60` means bulk operations can hold at most 60% + /// of permits. + pub fn with_bulk(mut self, percent: u32) -> Self { + let clamped = percent.min(100); + self.bulk_total = self.tasks_total * clamped / 100; + self.bulk = Arc::new(Semaphore::new(self.bulk_total as usize)); + self + } + /// Acquires a single concurrency permit, waiting if necessary. /// /// If all `max + queue` slots are occupied, returns @@ -103,39 +133,72 @@ impl ConcurrencyLimiter { Ok(ConcurrencyPermit { task_permit: Some(task_permit), + bulk_permit: None, 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. Both - /// the execution semaphore and the queue semaphore are checked, so the - /// total-capacity invariant (`max + queue`) is maintained. + /// Tries to acquire a single permit without waiting. /// - /// Returns [`Error::AtCapacity`] when fewer than `count` permits are available. - pub fn try_acquire_many(&self, count: u32) -> Result { + /// Returns [`Error::AtCapacity`] when no permits are available. + pub fn try_acquire(&self) -> Result { let queue_permit = self .queue .clone() - .try_acquire_many_owned(count) + .try_acquire_owned() .map_err(|_| Error::AtCapacity)?; let task_permit = self .tasks .clone() - .try_acquire_many_owned(count) + .try_acquire_owned() .map_err(|_| Error::AtCapacity)?; Ok(ConcurrencyPermit { task_permit: Some(task_permit), + bulk_permit: None, queue_permit: Some(queue_permit), released: Arc::clone(&self.released), }) } + /// Acquires a single permit for a bulk operation, waiting if necessary. + /// + /// Bulk operations are bounded by the bulk budget — a safe operating + /// point below which there is little-to-no performance degradation. + /// Both the bulk semaphore and the inner execution semaphore are + /// acquired under a single [`queue_timeout`](Self::with_queue) + /// deadline. + /// + /// Returns [`Error::AtCapacity`] on timeout or when `max` is zero. + pub async fn acquire_bulk(&self) -> Result { + if self.tasks_total == 0 { + return Err(Error::AtCapacity); + } + + let bulk_sem = self.bulk.clone(); + let tasks_sem = self.tasks.clone(); + + let acquire = async move { + let bulk_permit = bulk_sem.acquire_owned().await?; + let task_permit = tasks_sem.acquire_owned().await?; + Ok((task_permit, bulk_permit)) + }; + + let (task_permit, bulk_permit) = tokio::time::timeout(self.timeout, acquire) + .await + .map_err(|_| Error::AtCapacity)? + .map_err(|_: AcquireError| Error::AtCapacity)?; + + Ok(ConcurrencyPermit { + task_permit: Some(task_permit), + bulk_permit: Some(bulk_permit), + queue_permit: None, + released: Arc::clone(&self.released), + }) + } + /// Returns the number of permits currently available. pub fn available_permits(&self) -> u32 { u32::try_from(self.tasks.available_permits()).unwrap_or(self.tasks_total) @@ -162,6 +225,17 @@ impl ConcurrencyLimiter { self.queue_total - self.tasks_total } + /// Returns the number of bulk permits currently held. + pub fn used_bulk_permits(&self) -> u32 { + let available = u32::try_from(self.bulk.available_permits()).unwrap_or(self.bulk_total); + self.bulk_total - available + } + + /// Returns the bulk concurrency budget. + pub fn total_bulk(&self) -> u32 { + self.bulk_total + } + /// Waits until all permits have been returned. #[allow(dead_code)] pub async fn wait_all(&self) { @@ -179,6 +253,7 @@ impl ConcurrencyLimiter { Stats { in_use: self.used_permits(), queued: self.queued_permits(), + bulk_in_use: self.used_bulk_permits(), } } @@ -204,11 +279,12 @@ 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. +/// Fields are ordered so that the inner (execution) permit drops first, +/// then the bulk and queue permits — a waiter that claims a freed slot +/// can immediately see the freed execution slot. pub struct ConcurrencyPermit { task_permit: Option, + bulk_permit: Option, queue_permit: Option, released: Arc, } @@ -222,6 +298,7 @@ impl std::fmt::Debug for ConcurrencyPermit { impl Drop for ConcurrencyPermit { fn drop(&mut self) { drop(self.task_permit.take()); + drop(self.bulk_permit.take()); drop(self.queue_permit.take()); self.released.notify_waiters(); } @@ -304,10 +381,10 @@ mod tests { let limiter = ConcurrencyLimiter::new(5); assert_eq!(limiter.available_permits(), 5); - let p1 = limiter.try_acquire_many(1).unwrap(); + let p1 = limiter.try_acquire().unwrap(); assert_eq!(limiter.available_permits(), 4); - let p2 = limiter.try_acquire_many(1).unwrap(); + let p2 = limiter.try_acquire().unwrap(); assert_eq!(limiter.available_permits(), 3); drop(p1); @@ -328,10 +405,10 @@ mod tests { let limiter = ConcurrencyLimiter::new(2); assert_eq!(limiter.used_permits(), 0); - let p1 = limiter.try_acquire_many(1).unwrap(); + let p1 = limiter.try_acquire().unwrap(); assert_eq!(limiter.used_permits(), 1); - let p2 = limiter.try_acquire_many(1).unwrap(); + let p2 = limiter.try_acquire().unwrap(); assert_eq!(limiter.used_permits(), 2); drop(p1); @@ -344,9 +421,9 @@ mod tests { #[test] fn at_capacity_rejects() { let limiter = ConcurrencyLimiter::new(1); - let _permit = limiter.try_acquire_many(1).unwrap(); + let _permit = limiter.try_acquire().unwrap(); - let result = limiter.try_acquire_many(1); + let result = limiter.try_acquire(); assert!(matches!(result, Err(Error::AtCapacity))); } @@ -354,17 +431,17 @@ mod tests { fn permit_recovery_after_drop() { let limiter = ConcurrencyLimiter::new(1); - let permit = limiter.try_acquire_many(1).unwrap(); - assert!(limiter.try_acquire_many(1).is_err()); + let permit = limiter.try_acquire().unwrap(); + assert!(limiter.try_acquire().is_err()); drop(permit); - assert!(limiter.try_acquire_many(1).is_ok()); + assert!(limiter.try_acquire().is_ok()); } #[tokio::test(start_paused = true)] async fn emitter_calls_callback() { let limiter = ConcurrencyLimiter::new(5); - let _permit = limiter.try_acquire_many(1).unwrap(); + let _permit = limiter.try_acquire().unwrap(); let emitted_in_use = Arc::new(AtomicU32::new(0)); let emitted_queued = Arc::new(AtomicU32::new(0)); @@ -392,8 +469,8 @@ mod tests { #[tokio::test] async fn wait_all_resolves_when_permits_returned() { let limiter = ConcurrencyLimiter::new(2); - let p1 = limiter.try_acquire_many(1).unwrap(); - let p2 = limiter.try_acquire_many(1).unwrap(); + let p1 = limiter.try_acquire().unwrap(); + let p2 = limiter.try_acquire().unwrap(); let mut wait = Box::pin(limiter.wait_all()); @@ -420,15 +497,12 @@ mod tests { 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) - )); + let p1 = limiter.try_acquire().unwrap(); + let p2 = limiter.try_acquire().unwrap(); + assert!(matches!(limiter.try_acquire(), Err(Error::AtCapacity))); drop(p1); - assert!(limiter.try_acquire_many(1).is_ok()); + assert!(limiter.try_acquire().is_ok()); drop(p2); } @@ -494,33 +568,6 @@ mod tests { 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)); @@ -550,9 +597,9 @@ mod tests { 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(); + let _p1 = limiter.try_acquire().unwrap(); assert_eq!(limiter.queued_permits(), 0); - let _p2 = limiter.try_acquire_many(1).unwrap(); + let _p2 = limiter.try_acquire().unwrap(); assert_eq!(limiter.queued_permits(), 0); } @@ -597,4 +644,127 @@ mod tests { assert_eq!(emitted_in_use.load(Ordering::Relaxed), 1); assert_eq!(emitted_queued.load(Ordering::Relaxed), 1); } + + // --- Bulk tests --- + + #[test] + fn bulk_defaults_to_full_capacity() { + let limiter = ConcurrencyLimiter::new(100); + assert_eq!(limiter.total_bulk(), 100); + } + + #[test] + fn bulk_percent_computes_correctly() { + let limiter = ConcurrencyLimiter::new(100).with_bulk(60); + assert_eq!(limiter.total_bulk(), 60); + + let limiter = ConcurrencyLimiter::new(10).with_bulk(90); + assert_eq!(limiter.total_bulk(), 9); + + let limiter = ConcurrencyLimiter::new(100).with_bulk(150); + assert_eq!(limiter.total_bulk(), 100); + } + + #[tokio::test(start_paused = true)] + async fn bulk_caps_at_budget() { + let limiter = ConcurrencyLimiter::new(10) + .with_queue(5, Duration::from_secs(5)) + .with_bulk(90); + let bulk_budget = limiter.total_bulk(); + assert_eq!(bulk_budget, 9); + + let mut permits = Vec::new(); + for _ in 0..bulk_budget { + let permit = limiter.acquire_bulk().await.unwrap(); + permits.push(permit); + } + + assert_eq!(limiter.used_bulk_permits(), bulk_budget); + assert_eq!(limiter.used_permits(), bulk_budget); + + // Normal request can still acquire the remaining permit. + let normal = limiter.acquire().await.unwrap(); + assert_eq!(limiter.used_permits(), 10); + drop(normal); + drop(permits); + } + + #[tokio::test(start_paused = true)] + async fn normal_uses_all_permits_when_bulk_idle() { + let limiter = ConcurrencyLimiter::new(10).with_queue(5, Duration::from_secs(5)); + + let mut permits = Vec::new(); + for _ in 0..10 { + let permit = limiter.acquire().await.unwrap(); + permits.push(permit); + } + + assert_eq!(limiter.used_permits(), 10); + assert_eq!(limiter.used_bulk_permits(), 0); + drop(permits); + } + + #[tokio::test(start_paused = true)] + async fn bulk_waits_for_inner_permit() { + 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_bulk().await }); + tokio::task::yield_now().await; + + drop(held); + + let permit = waiter.await.unwrap().unwrap(); + assert_eq!(limiter.used_permits(), 1); + assert_eq!(limiter.used_bulk_permits(), 1); + drop(permit); + } + + #[tokio::test(start_paused = true)] + async fn bulk_timeout_spans_both_waits() { + let limiter = ConcurrencyLimiter::new(1).with_queue(0, Duration::from_secs(1)); + + let _held = limiter.acquire().await.unwrap(); + + let limiter2 = limiter.clone(); + let waiter = tokio::spawn(async move { limiter2.acquire_bulk().await }); + + tokio::time::sleep(Duration::from_secs(2)).await; + + let result = waiter.await.unwrap(); + assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(limiter.used_bulk_permits(), 0); + } + + #[tokio::test(start_paused = true)] + async fn bulk_cancellation_leaks_nothing() { + let limiter = ConcurrencyLimiter::new(2).with_queue(2, Duration::from_secs(60)); + + let _held1 = limiter.acquire().await.unwrap(); + let _held2 = limiter.acquire().await.unwrap(); + + let limiter2 = limiter.clone(); + let waiter = tokio::spawn(async move { limiter2.acquire_bulk().await }); + tokio::task::yield_now().await; + + waiter.abort(); + let _ = waiter.await; + tokio::task::yield_now().await; + + assert_eq!(limiter.used_bulk_permits(), 0); + assert_eq!(limiter.used_permits(), 2); + } + + #[tokio::test(start_paused = true)] + async fn bulk_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_bulk().await; + assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(start.elapsed(), Duration::ZERO); + } } diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index 1e70f2bd..ed4e7ee0 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -14,7 +14,7 @@ use objectstore_types::range::{ByteRange, ContentRange}; use crate::backend::common::Backend; use crate::backend::counting::CountingBackend; use crate::concurrency::ConcurrencyLimiter; -use crate::error::{Error, Result}; +use crate::error::Result; use crate::id::{ObjectContext, ObjectId}; use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, @@ -73,7 +73,7 @@ pub const DEFAULT_CONCURRENCY_LIMIT: u32 = 500; #[derive(Clone, Debug)] pub struct StorageService { inner: Arc, - concurrency: ConcurrencyLimiter, + pub(crate) concurrency: ConcurrencyLimiter, } impl StorageService { @@ -100,11 +100,6 @@ impl StorageService { self } - /// Returns the number of backend task slots currently available. - pub fn tasks_available(&self) -> u32 { - self.concurrency.available_permits() - } - /// Returns the number of backend tasks currently running. pub fn tasks_running(&self) -> u32 { self.concurrency.used_permits() @@ -117,45 +112,38 @@ impl StorageService { /// Prepares to stream multiple operations concurrently against this service. /// - /// Operations are executed concurrently up to a window derived from the - /// service's current capacity. The permits for that window are reserved - /// upfront — if the service is at capacity, this returns - /// [`Error::AtCapacity`] immediately before any operations are read. - pub fn stream(&self) -> Result { - let available = self.tasks_available(); - let window = available.div_ceil(10); - - let acquire_result = match window { - 0 => Err(Error::AtCapacity), - _ => self.concurrency.try_acquire_many(window), - }; - let reservation = acquire_result.inspect_err(|_| { - objectstore_metrics::count!("service.concurrency.rejected"); - objectstore_log::warn!("Request rejected: service at capacity"); - })?; - - Ok(StreamExecutor { + /// Each operation acquires a bulk permit individually via + /// [`ConcurrencyLimiter::acquire_bulk`], which caps bulk traffic at a + /// configurable percentage of execution slots while allowing operations + /// to queue for permits instead of requiring upfront reservation. + pub fn stream(&self) -> StreamExecutor { + StreamExecutor { backend: Arc::clone(&self.inner), - window, - reservation, - }) + concurrency: self.concurrency.clone(), + } } /// Starts background processes for the storage service. /// /// 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`. + /// `service.concurrency.bulk_in_use`, `service.concurrency.limit`, + /// `service.concurrency.queue_limit`, and + /// `service.concurrency.bulk_limit`. pub fn start(&self) { let concurrency = self.concurrency.clone(); objectstore_metrics::gauge!("service.concurrency.limit" = concurrency.total_permits()); objectstore_metrics::gauge!("service.concurrency.queue_limit" = concurrency.total_queue()); + objectstore_metrics::gauge!("service.concurrency.bulk_limit" = concurrency.total_bulk()); tokio::spawn(async move { concurrency .run_emitter(|stats| async move { objectstore_metrics::gauge!("service.concurrency.in_use" = stats.in_use); objectstore_metrics::gauge!("service.concurrency.queued" = stats.queued); + objectstore_metrics::gauge!( + "service.concurrency.bulk_in_use" = stats.bulk_in_use + ); }) .await; }); diff --git a/objectstore-service/src/streaming.rs b/objectstore-service/src/streaming.rs index e7811885..7fd09d8a 100644 --- a/objectstore-service/src/streaming.rs +++ b/objectstore-service/src/streaming.rs @@ -1,41 +1,26 @@ //! Streaming operation types and concurrent executor. //! -//! [`StreamExecutor`] processes a stream of `(idx, Result<`[`Operation`]`, E>)` tuples -//! concurrently within a bounded window. Errors in the input stream pass through -//! unchanged; successful operations are executed against the backend directly, -//! with [`tokio::spawn`] for panic isolation and run-to-completion guarantees. +//! [`StreamExecutor`] processes a stream of `(idx, Result<`[`Operation`]`, E>)` tuples concurrently +//! within a bounded window. Errors in the input stream pass through unchanged; successful +//! operations are executed against the backend directly, with [`tokio::spawn`] for panic isolation +//! and run-to-completion guarantees. //! -//! ## Window and Permit Reservation +//! ## Permit Acquisition //! -//! The concurrency window is derived from the service's available permits at the time -//! [`StorageService::stream`](crate::service::StorageService::stream) is called: `ceil(tasks_available × 0.10)`. -//! The executor pre-acquires exactly `window` permits from the service's -//! `ConcurrencyLimiter` as a single bulk reservation. The reservation is shared -//! (via `Arc`) with every spawned task, so permits are released only after every -//! in-flight task completes — even if the output stream is dropped early. +//! Streaming / batch operations are subject to the service's concurrency limiter. They count as +//! "bulk" operations, which are capped at a lower limit than regular operations. This ensures a +//! large bulk request doesn't automatically brings the service to its limits and leaves room for +//! regular requests. //! -//! This means: -//! - If the service is at capacity, [`StorageService::stream`](crate::service::StorageService::stream) fails immediately with -//! [`Error::AtCapacity`] before any operations are read. -//! - During execution, operations call the storage backend directly without acquiring -//! additional per-operation permits. +//! The regular acquire timeout applies: Operations that cannot acquire a permit within the +//! configured queue timeout fail with [`Error::AtCapacity`]. //! //! ## Concurrency Model //! -//! [`StreamExecutor::execute`] uses `buffer_unordered` to drive up to `window` -//! operations concurrently. The input stream is pulled lazily — at most `window` -//! operations are in-flight at once, bounding memory to roughly -//! `window × max_operation_size`. Results are yielded in completion order. -//! -//! Each operation is wrapped in a [`tokio::spawn`] for panic isolation: a panic in -//! one operation surfaces as [`Error::Panic`] for that item and does not affect the -//! others. -//! -//! ## Future Scope -//! -//! The window fraction (10%) is hard-coded. Configurable fractions, adaptive window -//! sizing, and backend-level optimizations (e.g. BigTable multi-read, GCS batch API) -//! are out of scope for the current implementation. +//! [`StreamExecutor::execute`] uses `buffer_unordered` with the bulk budget as the concurrency +//! bound. The input stream is pulled lazily and results are yielded in completion order. Each +//! operation is wrapped in a [`tokio::spawn`] for panic isolation: a panic in one operation +//! surfaces as [`Error::Panic`] for that item and does not affect the others. use std::sync::Arc; @@ -43,7 +28,7 @@ use futures_util::{Stream, StreamExt}; use objectstore_types::metadata::Metadata; use crate::backend::common::Backend; -use crate::concurrency::ConcurrencyPermit; +use crate::concurrency::ConcurrencyLimiter; use crate::error::{Error, Result}; use crate::id::{ObjectContext, ObjectId, ObjectKey}; use crate::service::GetResponse; @@ -211,35 +196,33 @@ impl std::fmt::Debug for OpResponse { /// Executes streaming operations with bounded concurrency. /// -/// Construct via [`StorageService::stream`](crate::service::StorageService::stream), -/// which pre-acquires the concurrency window from the service's available permits. +/// Construct via [`StorageService::stream`](crate::service::StorageService::stream). +/// Each operation acquires a bulk permit individually; aggregate bulk +/// concurrency is bounded by the bulk semaphore on the limiter. /// -/// See the [module documentation](self) for a full description of the window -/// calculation, permit reservation, and concurrency model. +/// See the [module documentation](self) for the concurrency model. #[derive(Debug)] pub struct StreamExecutor { pub(crate) backend: Arc, - pub(crate) window: u32, - pub(crate) reservation: ConcurrencyPermit, + pub(crate) concurrency: ConcurrencyLimiter, } impl StreamExecutor { - /// Returns the concurrency window computed at construction. - pub fn window(&self) -> u32 { - self.window - } - /// Executes the operations stream with bounded concurrency. /// /// Each item is a `(index, Result)` tuple where `index` is the /// 0-based position of the operation in the original request. Error items pass - /// through immediately; successful items are executed concurrently up to `window` - /// at a time, each in an isolated [`tokio::spawn`]. + /// through immediately; successful items acquire a bulk permit and execute + /// in an isolated [`tokio::spawn`]. + /// + /// Permit acquisition is sequential — only one acquire is in flight at a + /// time, ensuring fairness with other streams and preventing a large + /// batch from racing for all permits at once. Execution of operations + /// that already hold a permit proceeds concurrently. /// - /// Results are yielded in completion order (not submission order). The permit - /// reservation is held until every spawned task has completed — if the stream - /// is dropped early, in-flight tasks run to completion before the permits are - /// released. + /// Operations that cannot acquire a permit within the configured queue + /// timeout fail with [`Error::AtCapacity`]. Results are yielded in + /// completion order (not submission order). pub fn execute( self, context: ObjectContext, @@ -250,34 +233,45 @@ impl StreamExecutor { { let StreamExecutor { backend, - window, - reservation, + concurrency, } = self; - // Arc-wrap so each spawned task can hold a clone. Permits are released - // only when the last clone is dropped — i.e. after every spawned task - // completes, even if the output stream is dropped early. - let reservation = Arc::new(reservation); + let buffer = concurrency.total_bulk().max(1) as usize; operations - .map(move |(idx, item)| { - let permit = Arc::clone(&reservation); - let backend = Arc::clone(&backend); - let context = context.clone(); + // `then` awaits each closure before pulling the next item, making + // permit acquisition sequential. this ensures fairness with other + // streams and prevents a large batch from racing for all permits + // at once. + .then(move |(idx, item)| { + let concurrency = concurrency.clone(); async move { let op = match item { Ok(op) => op, Err(e) => return (idx, Err(e)), }; + match concurrency.acquire_bulk().await { + Ok(permit) => (idx, Ok((op, permit))), + Err(e) => (idx, Err(E::from(e))), + } + } + }) + .map(move |(idx, result)| { + let backend = Arc::clone(&backend); + let context = context.clone(); + async move { + let (op, permit) = match result { + Ok(pair) => pair, + Err(e) => return (idx, Err(e)), + }; - let spawn = crate::concurrency::spawn_metered(op.kind(), permit, async move { - execute_operation(backend, context, op).await + let spawn = crate::concurrency::spawn_metered(op.kind(), permit, { + execute_operation(backend, context, op) }); - (idx, spawn.await.map_err(E::from)) } }) - .buffer_unordered(window as usize) + .buffer_unordered(buffer) } } @@ -321,6 +315,7 @@ async fn execute_operation( mod tests { use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; use bytes::Bytes; use futures_util::StreamExt; @@ -357,43 +352,12 @@ mod tests { futures_util::stream::iter(ops.into_iter().enumerate().map(|(i, op)| (i, Ok(op)))) } - // --- StreamExecutor window and capacity tests --- - - #[test] - fn at_capacity_when_no_permits() { - let service = make_service_with_limit(0); - assert!(matches!(service.stream(), Err(Error::AtCapacity))); - } - - #[test] - fn window_computation() { - // ceil(1 × 0.10) = 1 - let s = make_service_with_limit(1); - assert_eq!(s.stream().unwrap().window(), 1); - - // ceil(10 × 0.10) = 1 - let s = make_service_with_limit(10); - assert_eq!(s.stream().unwrap().window(), 1); - - // ceil(100 × 0.10) = 10 - let s = make_service_with_limit(100); - assert_eq!(s.stream().unwrap().window(), 10); - - // ceil(500 × 0.10) = 50 - let s = make_service_with_limit(500); - assert_eq!(s.stream().unwrap().window(), 50); - - // ceil(1000 × 0.10) = 100 - let s = make_service_with_limit(1000); - assert_eq!(s.stream().unwrap().window(), 100); - } - - // --- StreamExecutor::execute() correctness tests --- + // --- StreamExecutor correctness tests --- #[tokio::test] async fn execute_empty_stream() { let service = make_service(); - let executor = service.stream().unwrap(); + let executor = service.stream(); let outcomes: Vec<_> = executor .execute( make_context(), @@ -433,7 +397,7 @@ mod tests { Operation::Delete(Delete { key: "key1".into() }), ]; - let executor = service.stream().unwrap(); + let executor = service.stream(); let outcomes: Vec<_> = executor.execute(context, indexed_ok(ops)).collect().await; assert_eq!(outcomes.len(), 4); @@ -473,7 +437,7 @@ mod tests { }), ]; - let executor = service.stream().unwrap(); + let executor = service.stream(); let mut outcomes: Vec<_> = executor.execute(context, indexed_ok(ops)).collect().await; outcomes.sort_by_key(|(idx, _)| *idx); @@ -530,7 +494,6 @@ mod tests { #[tokio::test] async fn concurrent_execution() { - // Window = ceil(100 × 0.10) = 10. let (paused_tx, mut paused_rx) = tokio::sync::mpsc::channel::<()>(20); let resume = Arc::new(tokio::sync::Notify::new()); let in_flight = Arc::new(AtomicUsize::new(0)); @@ -543,10 +506,6 @@ mod tests { let service = StorageService::new(Box::new(gated)).with_concurrency(ConcurrencyLimiter::new(100)); - let executor = service.stream().unwrap(); - assert_eq!(executor.window(), 10); - - // Submit 10 inserts. With window=10, all should be in-flight simultaneously. let ops: Vec = (0..10) .map(|i| { Operation::Insert(Box::new(Insert { @@ -557,6 +516,7 @@ mod tests { }) .collect(); + let executor = service.stream(); let exec_handle = tokio::spawn(async move { executor .execute(make_context(), indexed_ok(ops)) @@ -584,40 +544,35 @@ mod tests { } #[tokio::test] - async fn batch_rejected_when_permits_exhausted() { - // Service with limit=1. One background insert holds the only permit. - // service.stream() must fail with AtCapacity. - let (paused_tx, mut paused_rx) = tokio::sync::mpsc::channel::<()>(2); - let resume = Arc::new(tokio::sync::Notify::new()); + async fn bulk_respects_budget() { + // Bulk budget = 1 (100% of max=1). Hold the permit via a normal + // acquire; the bulk op should wait and eventually time out. + let service = StorageService::new(Box::new(InMemoryBackend::new("in-memory"))) + .with_concurrency( + ConcurrencyLimiter::new(1) + .with_queue(0, Duration::from_millis(1)) + .with_bulk(100), + ); - let gated = TestBackend::new(GateOnPut { - paused_tx, - resume: Arc::clone(&resume), - in_flight: Arc::new(AtomicUsize::new(0)), - }); - let service = - StorageService::new(Box::new(gated)).with_concurrency(ConcurrencyLimiter::new(1)); - - // Hold the only permit via a blocking insert. - let svc = service.clone(); - tokio::spawn(async move { - let _ = svc - .insert_object( - make_context(), - Some("blocker".into()), - Metadata::default(), - stream::single("x"), - ) - .await; - }); - paused_rx.recv().await.unwrap(); + let _held = service.concurrency.acquire().await.unwrap(); + + let ops = vec![Operation::Insert(Box::new(Insert { + key: Some("blocked".into()), + metadata: Metadata::default(), + payload: Bytes::from("data"), + }))]; + + let executor = service.stream(); + let outcomes: Vec<_> = executor + .execute(make_context(), indexed_ok(ops)) + .collect() + .await; - // Permit is held — stream() must fail immediately with AtCapacity. + assert_eq!(outcomes.len(), 1); assert!( - matches!(service.stream(), Err(Error::AtCapacity)), - "expected AtCapacity when all permits are held" + matches!(&outcomes[0].1, Err(Error::AtCapacity)), + "expected AtCapacity, got {:?}", + outcomes[0].1, ); - - resume.notify_waiters(); } } From acab0fd85177b9f63620bd970f9720f899638550 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 21 Jul 2026 11:29:06 +0200 Subject: [PATCH 02/11] fix: Intra-doc links --- objectstore-service/src/concurrency.rs | 3 +-- objectstore-service/src/service.rs | 23 ++++++++++++++--------- 2 files changed, 15 insertions(+), 11 deletions(-) diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index b84c6695..e72b8e41 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -66,8 +66,7 @@ impl ConcurrencyLimiter { /// [`with_queue`](Self::with_queue) to enable bounded waiting. /// /// The bulk budget defaults to 100% of `max` (no restriction); use - /// [`with_bulk_percent`](Self::with_bulk_percent) to set a safe operating point for bulk - /// traffic. + /// [`with_bulk`](Self::with_bulk) to set a safe operating point for bulk traffic. pub fn new(max: u32) -> Self { Self { tasks: Arc::new(Semaphore::new(max as usize)), diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index ed4e7ee0..38414369 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -62,7 +62,6 @@ pub const DEFAULT_CONCURRENCY_LIMIT: u32 = 500; /// blocked. Call [`join`](StorageService::join) during shutdown to wait for /// outstanding cleanup. Operations are also isolated from panics in backend /// code — a failure in one operation does not bring down other in-flight work. -/// See [`Error::Panic`]. /// /// # Concurrency Limit /// @@ -151,15 +150,21 @@ 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. + /// # Observability /// - /// Emits `service.task.start` (counter) after acquiring a permit and - /// `service.task.duration` (distribution) when the task completes, tagged - /// with the given `operation` name and an `outcome` of `"success"` or - /// `"error"`. + /// This tracks two metrics: + /// + /// - `service.task.start` (counter) after acquiring a permit + /// - `service.task.duration` (distribution) when the task completes + /// + /// Both are tagged with the given `operation` name and an `outcome` + /// of `"success"` or `"error"`. + /// + /// # Errors + /// + /// - `AtCapacity` if the concurrency limit is reached + /// - `Panic` if the spawned task panics (the panic message is captured for diagnostics) + /// - `Dropped` if the task is dropped before sending its result. async fn spawn(&self, operation: &'static str, f: F) -> Result where T: Send + 'static, From 5031eaf339a00378f6815d750bc75211f1cf400d Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 21 Jul 2026 11:33:45 +0200 Subject: [PATCH 03/11] test(service): Add test for bulk acquire with zero budget --- objectstore-service/src/concurrency.rs | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index e72b8e41..4dba15f1 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -172,7 +172,7 @@ impl ConcurrencyLimiter { /// /// Returns [`Error::AtCapacity`] on timeout or when `max` is zero. pub async fn acquire_bulk(&self) -> Result { - if self.tasks_total == 0 { + if self.tasks_total == 0 || self.bulk_total == 0 { return Err(Error::AtCapacity); } @@ -766,4 +766,16 @@ mod tests { assert!(matches!(result, Err(Error::AtCapacity))); assert_eq!(start.elapsed(), Duration::ZERO); } + + #[tokio::test(start_paused = true)] + async fn bulk_rejects_immediately_when_bulk_budget_is_zero() { + let limiter = ConcurrencyLimiter::new(10) + .with_queue(5, Duration::from_secs(10)) + .with_bulk(0); + + let start = tokio::time::Instant::now(); + let result = limiter.acquire_bulk().await; + assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(start.elapsed(), Duration::ZERO); + } } From 21d1d0d65bea736fd91b19b8e62cba13d9607597 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 21 Jul 2026 14:54:25 +0200 Subject: [PATCH 04/11] fix: Small bulk limits --- objectstore-service/src/concurrency.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index 4dba15f1..6eb3c645 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -102,7 +102,7 @@ impl ConcurrencyLimiter { /// of permits. pub fn with_bulk(mut self, percent: u32) -> Self { let clamped = percent.min(100); - self.bulk_total = self.tasks_total * clamped / 100; + self.bulk_total = (self.tasks_total * clamped / 100).max(1); self.bulk = Arc::new(Semaphore::new(self.bulk_total as usize)); self } @@ -662,6 +662,10 @@ mod tests { let limiter = ConcurrencyLimiter::new(100).with_bulk(150); assert_eq!(limiter.total_bulk(), 100); + + // Allow at least one bulk permit even when the percentage is zero. + let limiter = ConcurrencyLimiter::new(1).with_bulk(60); + assert_eq!(limiter.total_bulk(), 1); } #[tokio::test(start_paused = true)] From 0afde606e4e77d8ef5d7736a9a61ed73e9e2b173 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 21 Jul 2026 17:40:56 +0200 Subject: [PATCH 05/11] fix(service): Remove dead bulk_total == 0 check in acquire_bulk --- objectstore-service/src/concurrency.rs | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index 6eb3c645..27b91d53 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -89,6 +89,7 @@ impl ConcurrencyLimiter { self.queue_total = self.tasks_total.saturating_add(size); self.queue = Arc::new(Semaphore::new((self.queue_total) as usize)); self.timeout = timeout; + self } @@ -172,7 +173,7 @@ impl ConcurrencyLimiter { /// /// Returns [`Error::AtCapacity`] on timeout or when `max` is zero. pub async fn acquire_bulk(&self) -> Result { - if self.tasks_total == 0 || self.bulk_total == 0 { + if self.tasks_total == 0 { return Err(Error::AtCapacity); } @@ -772,14 +773,22 @@ mod tests { } #[tokio::test(start_paused = true)] - async fn bulk_rejects_immediately_when_bulk_budget_is_zero() { + async fn bulk_with_zero_percent_allows_one() { let limiter = ConcurrencyLimiter::new(10) .with_queue(5, Duration::from_secs(10)) .with_bulk(0); - let start = tokio::time::Instant::now(); - let result = limiter.acquire_bulk().await; + assert_eq!(limiter.total_bulk(), 1); + + let permit = limiter.acquire_bulk().await.unwrap(); + assert_eq!(limiter.used_bulk_permits(), 1); + + let limiter2 = limiter.clone(); + let waiter = tokio::spawn(async move { limiter2.acquire_bulk().await }); + tokio::time::sleep(Duration::from_secs(11)).await; + + let result = waiter.await.unwrap(); assert!(matches!(result, Err(Error::AtCapacity))); - assert_eq!(start.elapsed(), Duration::ZERO); + drop(permit); } } From 58e8a44afe1bd0a0772bc59cc39396d32501709d Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 21 Jul 2026 17:47:47 +0200 Subject: [PATCH 06/11] fix(service): Emit concurrency rejected metric for bulk operations --- objectstore-service/src/service.rs | 2 +- objectstore-service/src/streaming.rs | 9 ++++++++- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index 38414369..47cbd364 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -172,7 +172,7 @@ impl StorageService { { let timer = objectstore_metrics::timer!("service.concurrency.wait"); let permit = self.concurrency.acquire().await.inspect_err(|_| { - objectstore_metrics::count!("service.concurrency.rejected"); + objectstore_metrics::count!("service.concurrency.rejected", class = "normal"); objectstore_log::warn!("Request rejected: service at capacity"); })?; diff --git a/objectstore-service/src/streaming.rs b/objectstore-service/src/streaming.rs index 7fd09d8a..2d1750c1 100644 --- a/objectstore-service/src/streaming.rs +++ b/objectstore-service/src/streaming.rs @@ -252,7 +252,14 @@ impl StreamExecutor { }; match concurrency.acquire_bulk().await { Ok(permit) => (idx, Ok((op, permit))), - Err(e) => (idx, Err(E::from(e))), + Err(e) => { + objectstore_metrics::count!( + "service.concurrency.rejected", + class = "bulk" + ); + objectstore_log::warn!("Bulk operation rejected: service at capacity"); + (idx, Err(E::from(e))) + } } } }) From 4a14810c48f88e2b71b44a44def3516b5d36c1fa Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 21 Jul 2026 19:09:30 +0200 Subject: [PATCH 07/11] fix(service): Use ticket semantics for queue semaphore in concurrency limiter --- objectstore-server/src/config.rs | 13 +- objectstore-server/src/state.rs | 6 +- objectstore-service/src/concurrency.rs | 168 ++++++++++++++++--------- objectstore-service/src/streaming.rs | 3 +- 4 files changed, 123 insertions(+), 67 deletions(-) diff --git a/objectstore-server/src/config.rs b/objectstore-server/src/config.rs index afa20247..6c3c76c5 100644 --- a/objectstore-server/src/config.rs +++ b/objectstore-server/src/config.rs @@ -528,7 +528,7 @@ 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. + /// `concurrency_timeout`) instead of being rejected immediately. /// Requests beyond that are rejected with HTTP 429. /// /// Sizing guidance: `concurrency_queue ≈ permit_release_rate × @@ -539,15 +539,16 @@ pub struct Service { /// `0` pub concurrency_queue: u32, - /// Maximum time a request may wait in the concurrency queue. + /// Maximum time a caller may wait for a concurrency permit. /// - /// Ignored when `concurrency_queue` is `0`. + /// Applies to both queued normal requests and bulk operations + /// waiting for the bulk and execution semaphores. /// /// # Default /// /// `1s` - #[serde(with = "humantime_serde")] - pub concurrency_queue_timeout: Duration, + #[serde(alias = "concurrency_queue_timeout", with = "humantime_serde")] + pub concurrency_timeout: Duration, /// Percentage of `max_concurrency` available to bulk operations /// (e.g. parallelized batch requests). @@ -570,7 +571,7 @@ impl Default for Service { Self { max_concurrency: objectstore_service::service::DEFAULT_CONCURRENCY_LIMIT, concurrency_queue: 0, - concurrency_queue_timeout: Duration::from_secs(1), + concurrency_timeout: Duration::from_secs(1), bulk_concurrency_pct: 60, } } diff --git a/objectstore-server/src/state.rs b/objectstore-server/src/state.rs index 2a3fb0b2..1a1dc8c1 100644 --- a/objectstore-server/src/state.rs +++ b/objectstore-server/src/state.rs @@ -63,10 +63,8 @@ impl Services { let backend = backend::from_config(config.storage.clone()).await?; let concurrency = ConcurrencyLimiter::new(config.service.max_concurrency) - .with_queue( - config.service.concurrency_queue, - config.service.concurrency_queue_timeout, - ) + .with_queue(config.service.concurrency_queue) + .with_timeout(config.service.concurrency_timeout) .with_bulk(config.service.bulk_concurrency_pct); let service = StorageService::new(backend).with_concurrency(concurrency); service.start(); diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index 27b91d53..831a3616 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -70,12 +70,12 @@ impl ConcurrencyLimiter { pub fn new(max: u32) -> Self { Self { tasks: Arc::new(Semaphore::new(max as usize)), - queue: Arc::new(Semaphore::new(max as usize)), + queue: Arc::new(Semaphore::new(0)), bulk: Arc::new(Semaphore::new(max as usize)), tasks_total: max, - queue_total: max, + queue_total: 0, bulk_total: max, - timeout: Duration::ZERO, + timeout: Duration::from_secs(1), released: Arc::new(Notify::new()), } } @@ -83,13 +83,20 @@ impl ConcurrencyLimiter { /// 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; + /// waiting for a permit. Callers beyond that are rejected immediately. + pub fn with_queue(mut self, size: u32) -> Self { + self.queue_total = size; + self.queue = Arc::new(Semaphore::new(size as usize)); + self + } + /// Sets the maximum time a caller may wait for a permit. + /// + /// Applies to both [`acquire`](Self::acquire) (when parked in the + /// queue) and [`acquire_bulk`](Self::acquire_bulk) (waiting for the + /// bulk and execution semaphores). Defaults to 1 second. + pub fn with_timeout(mut self, timeout: Duration) -> Self { + self.timeout = timeout; self } @@ -110,16 +117,27 @@ impl ConcurrencyLimiter { /// 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. + /// If a permit is free, returns immediately without touching the + /// queue. Otherwise, acquires a queue ticket (bounded by the queue + /// depth) and waits up to the configured timeout. Returns + /// [`Error::AtCapacity`] if the queue is full or on timeout. pub async fn acquire(&self) -> Result { if self.tasks_total == 0 { return Err(Error::AtCapacity); } - let queue_permit = self + // Fast path: Instantly grab a free permit without parking. + if let Ok(task_permit) = self.tasks.clone().try_acquire_owned() { + return Ok(ConcurrencyPermit { + task_permit: Some(task_permit), + bulk_permit: None, + released: Arc::clone(&self.released), + }); + } + + // Slow path: acquire a temporary queue ticket to bound concurrent + // waiters. Released in this scope when the permit is constructed. + let _ticket = self .queue .clone() .try_acquire_owned() @@ -134,7 +152,6 @@ impl ConcurrencyLimiter { Ok(ConcurrencyPermit { task_permit: Some(task_permit), bulk_permit: None, - queue_permit: Some(queue_permit), released: Arc::clone(&self.released), }) } @@ -143,12 +160,6 @@ impl ConcurrencyLimiter { /// /// Returns [`Error::AtCapacity`] when no permits are available. pub fn try_acquire(&self) -> Result { - let queue_permit = self - .queue - .clone() - .try_acquire_owned() - .map_err(|_| Error::AtCapacity)?; - let task_permit = self .tasks .clone() @@ -158,7 +169,6 @@ impl ConcurrencyLimiter { Ok(ConcurrencyPermit { task_permit: Some(task_permit), bulk_permit: None, - queue_permit: Some(queue_permit), released: Arc::clone(&self.released), }) } @@ -168,8 +178,8 @@ impl ConcurrencyLimiter { /// Bulk operations are bounded by the bulk budget — a safe operating /// point below which there is little-to-no performance degradation. /// Both the bulk semaphore and the inner execution semaphore are - /// acquired under a single [`queue_timeout`](Self::with_queue) - /// deadline. + /// acquired under a single timeout deadline configured via + /// [`with_timeout`](Self::with_timeout). /// /// Returns [`Error::AtCapacity`] on timeout or when `max` is zero. pub async fn acquire_bulk(&self) -> Result { @@ -194,7 +204,6 @@ impl ConcurrencyLimiter { Ok(ConcurrencyPermit { task_permit: Some(task_permit), bulk_permit: Some(bulk_permit), - queue_permit: None, released: Arc::clone(&self.released), }) } @@ -217,12 +226,12 @@ impl ConcurrencyLimiter { /// 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()) + self.queue_total - available } /// Returns the configured queue capacity. pub fn total_queue(&self) -> u32 { - self.queue_total - self.tasks_total + self.queue_total } /// Returns the number of bulk permits currently held. @@ -279,13 +288,11 @@ 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 first, -/// then the bulk and queue permits — a waiter that claims a freed slot -/// can immediately see the freed execution slot. +/// The execution permit drops before the bulk permit so that a waiter +/// blocked on the task semaphore sees the freed slot immediately. pub struct ConcurrencyPermit { task_permit: Option, bulk_permit: Option, - queue_permit: Option, released: Arc, } @@ -299,7 +306,6 @@ impl Drop for ConcurrencyPermit { fn drop(&mut self) { drop(self.task_permit.take()); drop(self.bulk_permit.take()); - drop(self.queue_permit.take()); self.released.notify_waiters(); } } @@ -492,8 +498,8 @@ mod tests { // --- Queue tests --- - #[test] - fn queue_zero_rejects_immediately() { + #[tokio::test(start_paused = true)] + async fn queue_zero_rejects_immediately() { let limiter = ConcurrencyLimiter::new(2); assert_eq!(limiter.total_queue(), 0); @@ -504,11 +510,24 @@ mod tests { drop(p1); assert!(limiter.try_acquire().is_ok()); drop(p2); + + // Bulk holding all permits also rejects instantly (no timeout wait). + let mut bulk_permits = Vec::new(); + for _ in 0..2 { + let permit = limiter.acquire_bulk().await.unwrap(); + bulk_permits.push(permit); + } + + let start = tokio::time::Instant::now(); + let result = limiter.acquire().await; + assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(start.elapsed(), Duration::ZERO); + drop(bulk_permits); } #[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 limiter = ConcurrencyLimiter::new(2).with_queue(3); let permit = limiter.acquire().await.unwrap(); assert_eq!(limiter.used_permits(), 1); @@ -518,7 +537,7 @@ mod tests { #[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 limiter = ConcurrencyLimiter::new(1).with_queue(2); let held = limiter.acquire().await.unwrap(); assert_eq!(limiter.used_permits(), 1); @@ -539,7 +558,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn acquire_times_out() { - let limiter = ConcurrencyLimiter::new(1).with_queue(2, Duration::from_secs(1)); + let limiter = ConcurrencyLimiter::new(1).with_queue(2); let _held = limiter.acquire().await.unwrap(); @@ -555,7 +574,7 @@ mod tests { #[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 limiter = ConcurrencyLimiter::new(1).with_queue(1); let _held = limiter.acquire().await.unwrap(); @@ -570,7 +589,7 @@ mod tests { #[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 limiter = ConcurrencyLimiter::new(1).with_queue(1); let _held = limiter.acquire().await.unwrap(); @@ -592,20 +611,32 @@ mod tests { drop(replacement); } - #[test] - fn queued_permits_reflects_state() { - let limiter = ConcurrencyLimiter::new(2).with_queue(3, Duration::from_secs(5)); + #[tokio::test(start_paused = true)] + async fn queued_permits_reflects_state() { + let limiter = ConcurrencyLimiter::new(2).with_queue(3); assert_eq!(limiter.queued_permits(), 0); let _p1 = limiter.try_acquire().unwrap(); assert_eq!(limiter.queued_permits(), 0); let _p2 = limiter.try_acquire().unwrap(); assert_eq!(limiter.queued_permits(), 0); + drop(_p1); + drop(_p2); + + // Exact count when bulk holds permits. + let _bulk = limiter.acquire_bulk().await.unwrap(); + let _bulk2 = limiter.acquire_bulk().await.unwrap(); + assert_eq!(limiter.queued_permits(), 0); + + let limiter2 = limiter.clone(); + let _waiter = tokio::spawn(async move { limiter2.acquire().await }); + tokio::task::yield_now().await; + assert_eq!(limiter.queued_permits(), 1); } #[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 limiter = ConcurrencyLimiter::new(0).with_queue(5); let start = tokio::time::Instant::now(); let result = limiter.acquire().await; @@ -615,7 +646,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn emitter_reports_queued_count() { - let limiter = ConcurrencyLimiter::new(1).with_queue(2, Duration::from_secs(60)); + let limiter = ConcurrencyLimiter::new(1).with_queue(2); let _held = limiter.acquire().await.unwrap(); let limiter2 = limiter.clone(); @@ -671,9 +702,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn bulk_caps_at_budget() { - let limiter = ConcurrencyLimiter::new(10) - .with_queue(5, Duration::from_secs(5)) - .with_bulk(90); + let limiter = ConcurrencyLimiter::new(10).with_queue(5).with_bulk(90); let bulk_budget = limiter.total_bulk(); assert_eq!(bulk_budget, 9); @@ -695,7 +724,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn normal_uses_all_permits_when_bulk_idle() { - let limiter = ConcurrencyLimiter::new(10).with_queue(5, Duration::from_secs(5)); + let limiter = ConcurrencyLimiter::new(10).with_queue(5); let mut permits = Vec::new(); for _ in 0..10 { @@ -710,7 +739,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn bulk_waits_for_inner_permit() { - let limiter = ConcurrencyLimiter::new(1).with_queue(2, Duration::from_secs(5)); + let limiter = ConcurrencyLimiter::new(1).with_queue(2); let held = limiter.acquire().await.unwrap(); assert_eq!(limiter.used_permits(), 1); @@ -729,7 +758,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn bulk_timeout_spans_both_waits() { - let limiter = ConcurrencyLimiter::new(1).with_queue(0, Duration::from_secs(1)); + let limiter = ConcurrencyLimiter::new(1); let _held = limiter.acquire().await.unwrap(); @@ -745,7 +774,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn bulk_cancellation_leaks_nothing() { - let limiter = ConcurrencyLimiter::new(2).with_queue(2, Duration::from_secs(60)); + let limiter = ConcurrencyLimiter::new(2).with_queue(2); let _held1 = limiter.acquire().await.unwrap(); let _held2 = limiter.acquire().await.unwrap(); @@ -764,7 +793,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn bulk_rejects_immediately_when_max_is_zero() { - let limiter = ConcurrencyLimiter::new(0).with_queue(5, Duration::from_secs(10)); + let limiter = ConcurrencyLimiter::new(0).with_queue(5); let start = tokio::time::Instant::now(); let result = limiter.acquire_bulk().await; @@ -774,9 +803,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn bulk_with_zero_percent_allows_one() { - let limiter = ConcurrencyLimiter::new(10) - .with_queue(5, Duration::from_secs(10)) - .with_bulk(0); + let limiter = ConcurrencyLimiter::new(10).with_bulk(0); assert_eq!(limiter.total_bulk(), 1); @@ -785,10 +812,39 @@ mod tests { let limiter2 = limiter.clone(); let waiter = tokio::spawn(async move { limiter2.acquire_bulk().await }); - tokio::time::sleep(Duration::from_secs(11)).await; + tokio::time::sleep(Duration::from_secs(2)).await; let result = waiter.await.unwrap(); assert!(matches!(result, Err(Error::AtCapacity))); drop(permit); } + + #[tokio::test(start_paused = true)] + async fn queue_bounded_under_bulk_load() { + let limiter = ConcurrencyLimiter::new(3).with_queue(2); + + let mut bulk_permits = Vec::new(); + for _ in 0..3 { + let permit = limiter.acquire_bulk().await.unwrap(); + bulk_permits.push(permit); + } + + let limiter2 = limiter.clone(); + let _w1 = tokio::spawn(async move { limiter2.acquire().await }); + tokio::task::yield_now().await; + assert_eq!(limiter.queued_permits(), 1); + + let limiter3 = limiter.clone(); + let _w2 = tokio::spawn(async move { limiter3.acquire().await }); + tokio::task::yield_now().await; + assert_eq!(limiter.queued_permits(), 2); + + // Third waiter exceeds queue depth — rejected instantly. + let start = tokio::time::Instant::now(); + let result = limiter.acquire().await; + assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(start.elapsed(), Duration::ZERO); + + drop(bulk_permits); + } } diff --git a/objectstore-service/src/streaming.rs b/objectstore-service/src/streaming.rs index 2d1750c1..1e682724 100644 --- a/objectstore-service/src/streaming.rs +++ b/objectstore-service/src/streaming.rs @@ -557,7 +557,8 @@ mod tests { let service = StorageService::new(Box::new(InMemoryBackend::new("in-memory"))) .with_concurrency( ConcurrencyLimiter::new(1) - .with_queue(0, Duration::from_millis(1)) + .with_queue(0) + .with_timeout(Duration::from_millis(1)) .with_bulk(100), ); From f503e5620d8060ab8e93f3d23f6012bba3c88467 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 21 Jul 2026 19:17:06 +0200 Subject: [PATCH 08/11] fix(service): Fix docs and use div_ceil for bulk budget computation --- objectstore-server/src/config.rs | 2 +- objectstore-service/src/concurrency.rs | 4 ++-- objectstore-service/src/streaming.rs | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/objectstore-server/src/config.rs b/objectstore-server/src/config.rs index 6c3c76c5..af8c5cca 100644 --- a/objectstore-server/src/config.rs +++ b/objectstore-server/src/config.rs @@ -557,7 +557,7 @@ pub struct Service { /// little-to-no performance degradation, leaving room for more tasks /// to be admitted via the queue before rejection is necessary. /// - /// Clamped to 0..=100. At 100, bulk operations can use all execution + /// Clamped to 1..=100. At 100, bulk operations can use all execution /// slots. Lower values leave headroom for single-object requests. /// /// # Default diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index 831a3616..7715568e 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -102,7 +102,7 @@ impl ConcurrencyLimiter { /// Sets the bulk concurrency budget as a percentage of `max`. /// - /// `percent` is clamped to `0..=100`. At `100` (the default), bulk + /// `percent` is clamped to `1..=100`. At `100` (the default), bulk /// operations can use all execution slots. Lower values set a safe /// operating point below which there is little-to-no performance /// degradation, allowing more tasks to queue before rejection is @@ -110,7 +110,7 @@ impl ConcurrencyLimiter { /// of permits. pub fn with_bulk(mut self, percent: u32) -> Self { let clamped = percent.min(100); - self.bulk_total = (self.tasks_total * clamped / 100).max(1); + self.bulk_total = (self.tasks_total * clamped).div_ceil(100).max(1); self.bulk = Arc::new(Semaphore::new(self.bulk_total as usize)); self } diff --git a/objectstore-service/src/streaming.rs b/objectstore-service/src/streaming.rs index 1e682724..86d061b8 100644 --- a/objectstore-service/src/streaming.rs +++ b/objectstore-service/src/streaming.rs @@ -9,7 +9,7 @@ //! //! Streaming / batch operations are subject to the service's concurrency limiter. They count as //! "bulk" operations, which are capped at a lower limit than regular operations. This ensures a -//! large bulk request doesn't automatically brings the service to its limits and leaves room for +//! large bulk request doesn't automatically bring the service to its limits and leaves room for //! regular requests. //! //! The regular acquire timeout applies: Operations that cannot acquire a permit within the From 7e59ba6d27f243549558ab78b6b2f1247f626b16 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 21 Jul 2026 21:12:36 +0200 Subject: [PATCH 09/11] fix: Remove alias to avoid "duplicate field" error --- objectstore-server/src/config.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/objectstore-server/src/config.rs b/objectstore-server/src/config.rs index af8c5cca..5b66a23d 100644 --- a/objectstore-server/src/config.rs +++ b/objectstore-server/src/config.rs @@ -547,7 +547,7 @@ pub struct Service { /// # Default /// /// `1s` - #[serde(alias = "concurrency_queue_timeout", with = "humantime_serde")] + #[serde(with = "humantime_serde")] pub concurrency_timeout: Duration, /// Percentage of `max_concurrency` available to bulk operations From 38d90d61dd3a0b1cd3f59e2432f08c05eff9c5d9 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 21 Jul 2026 22:02:33 +0200 Subject: [PATCH 10/11] fix(server): Update Service struct doc to list current env var names --- objectstore-server/src/config.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/objectstore-server/src/config.rs b/objectstore-server/src/config.rs index 5b66a23d..b7f633f7 100644 --- a/objectstore-server/src/config.rs +++ b/objectstore-server/src/config.rs @@ -509,7 +509,8 @@ pub struct Config { /// /// - `OS__SERVICE__MAX_CONCURRENCY` /// - `OS__SERVICE__CONCURRENCY_QUEUE` -/// - `OS__SERVICE__CONCURRENCY_QUEUE_TIMEOUT` +/// - `OS__SERVICE__CONCURRENCY_TIMEOUT` +/// - `OS__SERVICE__BULK_CONCURRENCY_PCT` #[derive(Debug, Deserialize, Serialize)] #[serde(default)] pub struct Service { From 84c63b01058e88771418ff25ef12dddcca117c9f Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 21 Jul 2026 22:13:27 +0200 Subject: [PATCH 11/11] ref(service): Encapsulate concurrency limiter behind accessor methods --- objectstore-service/src/service.rs | 27 +++++++++++++++++---------- objectstore-service/src/streaming.rs | 14 +++++++++++--- 2 files changed, 28 insertions(+), 13 deletions(-) diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index 47cbd364..0953fcd8 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -72,7 +72,7 @@ pub const DEFAULT_CONCURRENCY_LIMIT: u32 = 500; #[derive(Clone, Debug)] pub struct StorageService { inner: Arc, - pub(crate) concurrency: ConcurrencyLimiter, + concurrency: ConcurrencyLimiter, } impl StorageService { @@ -99,6 +99,11 @@ impl StorageService { self } + /// Returns a reference to the concurrency limiter. + pub fn concurrency_limiter(&self) -> &ConcurrencyLimiter { + &self.concurrency + } + /// Returns the number of backend tasks currently running. pub fn tasks_running(&self) -> u32 { self.concurrency.used_permits() @@ -116,19 +121,21 @@ impl StorageService { /// configurable percentage of execution slots while allowing operations /// to queue for permits instead of requiring upfront reservation. pub fn stream(&self) -> StreamExecutor { - StreamExecutor { - backend: Arc::clone(&self.inner), - concurrency: self.concurrency.clone(), - } + StreamExecutor::new(Arc::clone(&self.inner), self.concurrency.clone()) } /// Starts background processes for the storage service. /// - /// Spawns a task that emits concurrency gauges once per second: - /// `service.concurrency.in_use`, `service.concurrency.queued`, - /// `service.concurrency.bulk_in_use`, `service.concurrency.limit`, - /// `service.concurrency.queue_limit`, and - /// `service.concurrency.bulk_limit`. + /// At startup, this tracks the following gauges: + /// + /// - `service.concurrency.limit`: concurrent task execution slots + /// - `service.concurrency.queue_limit`: queue size for waiting tasks + /// - `service.concurrency.bulk_limit`: concurrent task execution slots for bulk operations + /// + /// Also spawns a task that emits concurrency gauges once per second: + /// - `service.concurrency.in_use`: currently running tasks + /// - `service.concurrency.queued`: currently queued tasks + /// - `service.concurrency.bulk_in_use`: currently running bulk tasks pub fn start(&self) { let concurrency = self.concurrency.clone(); objectstore_metrics::gauge!("service.concurrency.limit" = concurrency.total_permits()); diff --git a/objectstore-service/src/streaming.rs b/objectstore-service/src/streaming.rs index 86d061b8..27eed920 100644 --- a/objectstore-service/src/streaming.rs +++ b/objectstore-service/src/streaming.rs @@ -203,11 +203,19 @@ impl std::fmt::Debug for OpResponse { /// See the [module documentation](self) for the concurrency model. #[derive(Debug)] pub struct StreamExecutor { - pub(crate) backend: Arc, - pub(crate) concurrency: ConcurrencyLimiter, + backend: Arc, + concurrency: ConcurrencyLimiter, } impl StreamExecutor { + /// Creates a new `StreamExecutor` with the given backend and limiter. + pub fn new(backend: Arc, concurrency: ConcurrencyLimiter) -> Self { + Self { + backend, + concurrency, + } + } + /// Executes the operations stream with bounded concurrency. /// /// Each item is a `(index, Result)` tuple where `index` is the @@ -562,7 +570,7 @@ mod tests { .with_bulk(100), ); - let _held = service.concurrency.acquire().await.unwrap(); + let _held = service.concurrency_limiter().acquire().await.unwrap(); let ops = vec![Operation::Insert(Box::new(Insert { key: Some("blocked".into()),