diff --git a/objectstore-server/src/config.rs b/objectstore-server/src/config.rs index 73df89d2..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 { @@ -528,7 +529,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 +540,31 @@ 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, + pub concurrency_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 1..=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 { @@ -555,7 +572,8 @@ 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/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..1a1dc8c1 100644 --- a/objectstore-server/src/state.rs +++ b/objectstore-server/src/state.rs @@ -62,10 +62,10 @@ 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) + .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-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..7715568e 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,16 +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`](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)), - 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, - timeout: Duration::ZERO, + queue_total: 0, + bulk_total: max, + timeout: Duration::from_secs(1), released: Arc::new(Notify::new()), } } @@ -69,27 +83,61 @@ 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)); + /// 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 } + /// Sets the bulk concurrency budget as a percentage of `max`. + /// + /// `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 + /// 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).div_ceil(100).max(1); + 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 - /// [`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() @@ -103,35 +151,59 @@ impl ConcurrencyLimiter { Ok(ConcurrencyPermit { task_permit: Some(task_permit), - queue_permit: Some(queue_permit), + bulk_permit: None, 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 { - let queue_permit = self - .queue - .clone() - .try_acquire_many_owned(count) - .map_err(|_| Error::AtCapacity)?; - + /// Returns [`Error::AtCapacity`] when no permits are available. + pub fn try_acquire(&self) -> Result { 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), - queue_permit: Some(queue_permit), + bulk_permit: None, + 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 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 { + 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), released: Arc::clone(&self.released), }) } @@ -154,12 +226,23 @@ 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. + 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. @@ -179,6 +262,7 @@ impl ConcurrencyLimiter { Stats { in_use: self.used_permits(), queued: self.queued_permits(), + bulk_in_use: self.used_bulk_permits(), } } @@ -204,12 +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 before the -/// outer (queue) permit — a waiter that claims the freed queue 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, - queue_permit: Option, + bulk_permit: Option, released: Arc, } @@ -222,7 +305,7 @@ impl std::fmt::Debug for ConcurrencyPermit { impl Drop for ConcurrencyPermit { fn drop(&mut self) { drop(self.task_permit.take()); - drop(self.queue_permit.take()); + drop(self.bulk_permit.take()); self.released.notify_waiters(); } } @@ -304,10 +387,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 +411,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 +427,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 +437,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 +475,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()); @@ -415,26 +498,36 @@ 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); - 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); + + // 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); @@ -444,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); @@ -465,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(); @@ -481,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(); @@ -494,36 +587,9 @@ 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)); + let limiter = ConcurrencyLimiter::new(1).with_queue(1); let _held = limiter.acquire().await.unwrap(); @@ -545,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_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); + 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; @@ -568,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(); @@ -597,4 +675,176 @@ 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); + + // 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)] + async fn bulk_caps_at_budget() { + let limiter = ConcurrencyLimiter::new(10).with_queue(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); + + 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); + + 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); + + 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); + + 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); + + let start = tokio::time::Instant::now(); + let result = limiter.acquire_bulk().await; + assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(start.elapsed(), Duration::ZERO); + } + + #[tokio::test(start_paused = true)] + async fn bulk_with_zero_percent_allows_one() { + let limiter = ConcurrencyLimiter::new(10).with_bulk(0); + + 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(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/service.rs b/objectstore-service/src/service.rs index 1e70f2bd..0953fcd8 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, @@ -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 /// @@ -100,9 +99,9 @@ impl StorageService { self } - /// Returns the number of backend task slots currently available. - pub fn tasks_available(&self) -> u32 { - self.concurrency.available_permits() + /// Returns a reference to the concurrency limiter. + pub fn concurrency_limiter(&self) -> &ConcurrencyLimiter { + &self.concurrency } /// Returns the number of backend tasks currently running. @@ -117,45 +116,40 @@ 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 { - backend: Arc::clone(&self.inner), - window, - reservation, - }) + /// 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::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.limit`, and `service.concurrency.queue_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()); 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; }); @@ -163,15 +157,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 + /// + /// 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 /// - /// 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"`. + /// - `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, @@ -179,7 +179,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 e7811885..27eed920 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 bring 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,41 @@ 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, + backend: Arc, + concurrency: ConcurrencyLimiter, } impl StreamExecutor { - /// Returns the concurrency window computed at construction. - pub fn window(&self) -> u32 { - self.window + /// 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 /// 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`]. /// - /// 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. + /// 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. + /// + /// 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 +241,52 @@ 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) => { + objectstore_metrics::count!( + "service.concurrency.rejected", + class = "bulk" + ); + objectstore_log::warn!("Bulk operation rejected: service at capacity"); + (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 +330,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 +367,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 +412,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 +452,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 +509,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 +521,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 +531,7 @@ mod tests { }) .collect(); + let executor = service.stream(); let exec_handle = tokio::spawn(async move { executor .execute(make_context(), indexed_ok(ops)) @@ -584,40 +559,36 @@ 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) + .with_timeout(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_limiter().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(); } }