ref(service): Replace upfront batch reservation with per-op bulk permits

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,23 +540,40 @@ 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 {
     fn default() -> Self {
         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,
         }
     }
 }

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<ObjectContext>,
     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<Operation, ApiError>)
     let parsed = requests.0.map(|r| r.map_err(ApiError::from)).enumerate();

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();
 

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(())

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