From 44f3f0e73ef4ced4dc6674df8c467071d67f324b Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 27 Aug 2026 22:06:51 +0800 Subject: [PATCH] perf(storage): gate large foreground PUT pressure (#6751) Add a default-on, size-aware foreground PUT admission policy so large or unknown-size PutObject requests are backpressured before body ingest and erasure/RPC fan-out. Preserve the explicit strict gate semantics, including limit=0 as an opt-out, and keep small PUTs on the legacy fast path. Closes rustfs/backlog#2038 Co-authored-by: heihutu --- crates/config/src/constants/object.rs | 33 ++ rustfs/src/app/object/put.rs | 2 +- rustfs/src/storage/concurrency/manager.rs | 419 +++++++++++++++++----- 3 files changed, 364 insertions(+), 90 deletions(-) diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 9502253cf..095ea2b3b 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -288,6 +288,39 @@ pub const DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 0; const _: () = assert!(!DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE); +/// Enable automatic foreground admission for large or unknown-size PutObject requests. +/// +/// Unlike the strict experimental gate above, this default-on path only applies +/// to requests that are large enough to create sustained erasure/RPC pressure. +/// Small PUTs continue on the legacy path unless the strict gate is explicitly +/// enabled. +pub const ENV_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE: &str = "RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE"; +pub const DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE: bool = true; + +/// Maximum large foreground PutObject requests admitted concurrently per process. +/// +/// `0` derives a conservative default from the local disk-read scheduler cap, +/// currently clamped to protect the commit path without making ordinary high +/// throughput uploads single-file. +pub const ENV_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT: &str = "RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT"; +pub const DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT: usize = 0; + +/// Minimum object size that enters automatic large PutObject admission. +/// +/// Requests with an unknown size are treated as large because the write pressure +/// cannot be bounded from headers. +pub const ENV_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES: &str = "RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES"; +pub const DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES: usize = 32 * 1024 * 1024; + +/// Time in milliseconds a large foreground PutObject waits for a permit. +/// +/// A short wait smooths transient bursts while still returning S3 +/// `SlowDown`/503 before body ingest when the node is already saturated. +pub const ENV_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: &str = "RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS"; +pub const DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 250; + +const _: () = assert!(DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE); + /// Environment variable for minimum GetObject timeout in seconds. /// /// When dynamic timeout calculation is enabled, this is the minimum timeout diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index e44720ef7..a7211d8ef 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -962,7 +962,7 @@ impl DefaultObjectUsecase { rustfs_io_metrics::record_put_object_stage_duration_from("app_bucket_validate", bucket_validate_stage_start); let put_admission = match get_concurrency_manager() - .admit_put_object() + .admit_put_object(size) .await .map_err(|_| s3_error!(InternalError, "foreground write admission closed"))? { diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index bf51b55bf..646991ff1 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Concurrency manager for coordinating concurrent GetObject requests. +//! Concurrency manager for coordinating concurrent GetObject and PutObject requests. use super::io_schedule::{ IoLoadLevel, IoLoadMetrics, IoPriority, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus, IoSchedulerConfig, IoStrategy, @@ -34,6 +34,8 @@ use std::time::Duration; use tokio::sync::Semaphore; use tracing::debug; +const DERIVED_LARGE_PUT_ADMISSION_LIMIT_MAX: usize = 32; + /// Global concurrency manager instance pub(crate) static CONCURRENCY_MANAGER: LazyLock = LazyLock::new(ConcurrencyManager::new); @@ -65,11 +67,8 @@ pub struct ConcurrencyManager { bandwidth_monitor: Arc>, /// Metrics collector for I/O latency tracking (P50, P95, P99) metrics_collector: Arc, - /// Experimental fixed-count foreground PutObject admission gate. - put_admission_semaphore: Arc, - put_admission_enabled: bool, - put_admission_limit: usize, - put_admission_wait_timeout: Duration, + /// Foreground PutObject admission policy, resolved once at startup. + put_admission_policy: PutAdmissionPolicy, } impl std::fmt::Debug for ConcurrencyManager { @@ -127,10 +126,201 @@ pub enum PutObjectAdmission { /// Request is admitted and must hold the permit until the store write /// returns or the request fails before mutation. Admitted(tokio::sync::OwnedSemaphorePermit), - /// The fixed-count gate stayed full until the configured wait timeout. + /// The selected foreground PUT admission gate stayed full until the configured wait timeout. Rejected, } +#[derive(Clone)] +struct PutAdmissionGate { + semaphore: Arc, + limit: usize, + wait_timeout: Duration, +} + +impl PutAdmissionGate { + fn new(limit: usize, wait_timeout: Duration) -> Self { + Self { + semaphore: Arc::new(Semaphore::new(limit)), + limit, + wait_timeout, + } + } + + fn active(&self) -> usize { + self.limit.saturating_sub(self.semaphore.available_permits()) + } + + async fn admit(&self) -> Result { + if self.wait_timeout.is_zero() { + return Ok(match self.semaphore.clone().try_acquire_owned() { + Ok(permit) => PutObjectAdmission::Admitted(permit), + Err(tokio::sync::TryAcquireError::NoPermits) => PutObjectAdmission::Rejected, + Err(tokio::sync::TryAcquireError::Closed) => PutObjectAdmission::Rejected, + }); + } + + match tokio::time::timeout(self.wait_timeout, self.semaphore.clone().acquire_owned()).await { + Ok(permit) => Ok(PutObjectAdmission::Admitted(permit?)), + Err(_) => Ok(PutObjectAdmission::Rejected), + } + } +} + +#[derive(Clone)] +enum PutAdmissionPolicy { + /// Strict admission was explicitly enabled with limit `0`. + Disabled, + /// No hard PUT gate is configured; foreground write snapshots use the + /// existing active request counter as a soft pressure signal. + LegacyCounterOnly, + /// Explicit all-PUT admission gate. + Strict(PutAdmissionGate), + /// Default large/unknown-size PUT admission gate. + Large { gate: PutAdmissionGate, min_size_bytes: usize }, +} + +impl PutAdmissionPolicy { + fn from_env(max_disk_reads: usize) -> Self { + let strict_enabled = rustfs_utils::get_env_bool( + rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_ENABLE, + rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE, + ); + if strict_enabled { + let strict_limit = rustfs_utils::get_env_usize( + rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_LIMIT, + rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT, + ); + let strict_wait_timeout = Duration::from_millis(rustfs_utils::get_env_u64( + rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, + rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, + )); + return if strict_limit == 0 { + Self::Disabled + } else { + Self::Strict(PutAdmissionGate::new(strict_limit, strict_wait_timeout)) + }; + } + + let large_enabled = rustfs_utils::get_env_bool( + rustfs_config::ENV_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE, + rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE, + ); + if !large_enabled { + return Self::LegacyCounterOnly; + } + + let large_limit = derive_large_put_admission_limit( + rustfs_utils::get_env_usize( + rustfs_config::ENV_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT, + rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT, + ), + max_disk_reads, + ); + let min_size_bytes = rustfs_utils::get_env_usize( + rustfs_config::ENV_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES, + rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES, + ); + let wait_timeout = Duration::from_millis(rustfs_utils::get_env_u64( + rustfs_config::ENV_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, + rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, + )); + + Self::Large { + gate: PutAdmissionGate::new(large_limit, wait_timeout), + min_size_bytes, + } + } + + #[cfg(test)] + fn strict_for_test(enabled: bool, limit: usize, wait_timeout: Duration) -> Self { + if enabled { + if limit == 0 { + Self::Disabled + } else { + Self::Strict(PutAdmissionGate::new(limit, wait_timeout)) + } + } else { + Self::LegacyCounterOnly + } + } + + #[cfg(test)] + fn large_for_test(enabled: bool, limit: usize, min_size_bytes: usize, wait_timeout: Duration) -> Self { + if enabled && limit > 0 { + Self::Large { + gate: PutAdmissionGate::new(limit, wait_timeout), + min_size_bytes, + } + } else { + Self::LegacyCounterOnly + } + } + + async fn admit(&self, size: i64) -> Result { + match self { + Self::Disabled | Self::LegacyCounterOnly => Ok(PutObjectAdmission::Disabled), + Self::Strict(gate) => gate.admit().await, + Self::Large { gate, min_size_bytes } if should_gate_large_put(size, *min_size_bytes) => gate.admit().await, + Self::Large { .. } => Ok(PutObjectAdmission::Disabled), + } + } + + fn snapshot(&self, legacy_limit: usize) -> WorkloadAdmissionSnapshot { + match self { + Self::Disabled => put_admission_snapshot(0, 0, None), + Self::LegacyCounterOnly => put_admission_snapshot(PutObjectGuard::concurrent_count(), legacy_limit, None), + Self::Strict(gate) => { + put_admission_snapshot(gate.active(), gate.limit, Some("foreground write admission permits exhausted")) + } + Self::Large { gate, .. } => { + put_admission_snapshot(gate.active(), gate.limit, Some("large foreground write admission permits exhausted")) + } + } + } +} + +fn put_admission_snapshot(active: usize, limit: usize, hard_gate_reason: Option<&'static str>) -> WorkloadAdmissionSnapshot { + let state = if limit == 0 { + AdmissionState::Disabled + } else if active >= limit { + AdmissionState::Saturated + } else { + AdmissionState::Open + }; + + let admission = + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit)); + + match state { + AdmissionState::Disabled => admission.with_reason("foreground write admission disabled"), + AdmissionState::Saturated => { + admission.with_reason(hard_gate_reason.unwrap_or("foreground write concurrency reached local pressure limit")) + } + _ => admission, + } +} + +fn derive_large_put_admission_limit(configured_limit: usize, max_disk_reads: usize) -> usize { + if configured_limit > 0 { + return configured_limit; + } + + let scheduler_base = if max_disk_reads == 0 { + rustfs_config::DEFAULT_OBJECT_MAX_CONCURRENT_DISK_READS + } else { + max_disk_reads + }; + scheduler_base.div_ceil(2).clamp(1, DERIVED_LARGE_PUT_ADMISSION_LIMIT_MAX) +} + +fn should_gate_large_put(size: i64, min_size_bytes: usize) -> bool { + if min_size_bytes == 0 || size < 0 { + return true; + } + + usize::try_from(size).is_ok_and(|size| size >= min_size_bytes) +} + impl ConcurrencyManager { /// Create a new concurrency manager with default settings /// @@ -177,18 +367,7 @@ impl ConcurrencyManager { // Initialize metrics collector for I/O latency tracking // Keep 1000 samples for P95/P99 calculation let metrics_collector = Arc::new(MetricsCollector::new(performance_metrics, 1000)); - let put_admission_enabled = rustfs_utils::get_env_bool( - rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_ENABLE, - rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE, - ); - let put_admission_limit = rustfs_utils::get_env_usize( - rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_LIMIT, - rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT, - ); - let put_admission_wait_timeout = Duration::from_millis(rustfs_utils::get_env_u64( - rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, - rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS, - )); + let put_admission_policy = PutAdmissionPolicy::from_env(max_disk_reads); // Build queue config directly from scheduler config. let queue_config = IoPriorityQueueConfig::from_scheduler_config(&scheduler_config); @@ -204,10 +383,7 @@ impl ConcurrencyManager { pattern_detector, bandwidth_monitor, metrics_collector, - put_admission_semaphore: Arc::new(Semaphore::new(if put_admission_enabled { put_admission_limit } else { 0 })), - put_admission_enabled, - put_admission_limit, - put_admission_wait_timeout, + put_admission_policy, } } @@ -234,10 +410,19 @@ impl ConcurrencyManager { #[cfg(test)] pub(crate) fn with_put_admission_for_test(enabled: bool, limit: usize, wait_timeout: Duration) -> Self { let mut manager = Self::new(); - manager.put_admission_semaphore = Arc::new(Semaphore::new(if enabled { limit } else { 0 })); - manager.put_admission_enabled = enabled; - manager.put_admission_limit = limit; - manager.put_admission_wait_timeout = wait_timeout; + manager.put_admission_policy = PutAdmissionPolicy::strict_for_test(enabled, limit, wait_timeout); + manager + } + + #[cfg(test)] + pub(crate) fn with_large_put_admission_for_test( + enabled: bool, + limit: usize, + min_size_bytes: usize, + wait_timeout: Duration, + ) -> Self { + let mut manager = Self::new(); + manager.put_admission_policy = PutAdmissionPolicy::large_for_test(enabled, limit, min_size_bytes, wait_timeout); manager } @@ -326,30 +511,13 @@ impl ConcurrencyManager { } } - /// Admit a foreground PutObject request under the experimental fixed-count gate. + /// Admit a foreground PutObject request under the configured write gate. /// - /// The default-off path returns [`PutObjectAdmission::Disabled`] without - /// touching the semaphore, preserving legacy behavior. When enabled, the - /// permit must be acquired before body ingest and held until the store write - /// returns, so saturated foreground writes can fail with `SlowDown` before - /// creating visible side effects. - pub async fn admit_put_object(&self) -> Result { - if !self.put_admission_enabled || self.put_admission_limit == 0 { - return Ok(PutObjectAdmission::Disabled); - } - - if self.put_admission_wait_timeout.is_zero() { - return Ok(match self.put_admission_semaphore.clone().try_acquire_owned() { - Ok(permit) => PutObjectAdmission::Admitted(permit), - Err(tokio::sync::TryAcquireError::NoPermits) => PutObjectAdmission::Rejected, - Err(tokio::sync::TryAcquireError::Closed) => PutObjectAdmission::Rejected, - }); - } - - match tokio::time::timeout(self.put_admission_wait_timeout, self.put_admission_semaphore.clone().acquire_owned()).await { - Ok(permit) => Ok(PutObjectAdmission::Admitted(permit?)), - Err(_) => Ok(PutObjectAdmission::Rejected), - } + /// The strict experimental gate applies to every PUT only when explicitly + /// enabled. Otherwise the default-on large-object gate protects sustained + /// erasure/RPC pressure while keeping small PUTs on the legacy path. + pub async fn admit_put_object(&self, size: i64) -> Result { + self.put_admission_policy.admit(size).await } // ============================================ @@ -760,35 +928,7 @@ impl ConcurrencyManager { /// Get a read-only workload admission snapshot for foreground writes. pub fn put_object_admission_snapshot(&self) -> WorkloadAdmissionSnapshot { - let (active, limit, hard_gate_enabled) = if self.put_admission_enabled && self.put_admission_limit > 0 { - ( - self.put_admission_limit - .saturating_sub(self.put_admission_semaphore.available_permits()), - self.put_admission_limit, - true, - ) - } else { - (PutObjectGuard::concurrent_count(), self.scheduler_config.max_concurrent_reads, false) - }; - let state = if limit == 0 { - AdmissionState::Disabled - } else if active >= limit { - AdmissionState::Saturated - } else { - AdmissionState::Open - }; - - let admission = - WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit)); - - match state { - AdmissionState::Disabled => admission.with_reason("foreground write admission disabled"), - AdmissionState::Saturated if hard_gate_enabled => { - admission.with_reason("foreground write admission permits exhausted") - } - AdmissionState::Saturated => admission.with_reason("foreground write concurrency reached local pressure limit"), - _ => admission, - } + self.put_admission_policy.snapshot(self.scheduler_config.max_concurrent_reads) } /// Get a read-only workload admission registry snapshot for local storage concurrency. @@ -862,7 +1002,7 @@ impl Default for ConcurrencyManager { mod integration_tests { use super::super::io_schedule::{IoLoadLevel, IoPriority}; use super::super::request_guard::GetObjectGuard; - use super::{ConcurrencyManager, PutObjectAdmission}; + use super::{ConcurrencyManager, PutObjectAdmission, derive_large_put_admission_limit}; use crate::storage::storage_api::concurrency_consumer::PutObjectGuard; use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass}; use rustfs_io_core::io_profile::{AccessPattern, StorageMedia}; @@ -941,7 +1081,7 @@ mod integration_tests { #[serial] async fn test_concurrency_manager_workload_admission_snapshot_tracks_put_requests() { crate::storage::concurrency::reset_active_put_requests(); - let manager = ConcurrencyManager::new(); + let manager = ConcurrencyManager::with_put_admission_for_test(false, 0, Duration::ZERO); let initial = manager.put_object_admission_snapshot(); assert_eq!(initial.class, WorkloadClass::ForegroundWrite); @@ -965,26 +1105,42 @@ mod integration_tests { let manager = ConcurrencyManager::with_put_admission_for_test(false, 1, Duration::ZERO); let admission = manager - .admit_put_object() + .admit_put_object(1024) .await .expect("disabled put admission must not close"); assert!(matches!(admission, PutObjectAdmission::Disabled)); - assert_eq!(manager.put_admission_semaphore.available_permits(), 0); assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Open); } + #[tokio::test] + #[serial] + async fn test_concurrency_manager_strict_put_admission_zero_limit_disables_large_gate() { + let manager = ConcurrencyManager::with_put_admission_for_test(true, 0, Duration::ZERO); + + let admission = manager + .admit_put_object(32 * 1024 * 1024) + .await + .expect("strict zero-limit put admission must not close"); + + assert!(matches!(admission, PutObjectAdmission::Disabled)); + assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Disabled); + } + #[tokio::test] #[serial] async fn test_concurrency_manager_put_admission_rejects_when_limit_full() { let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::ZERO); - let first = manager.admit_put_object().await.expect("first put admission should acquire"); + let first = manager + .admit_put_object(1024) + .await + .expect("first put admission should acquire"); assert!(matches!(first, PutObjectAdmission::Admitted(_))); assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Saturated); let second = manager - .admit_put_object() + .admit_put_object(1024) .await .expect("full put admission gate should reject, not close"); assert!(matches!(second, PutObjectAdmission::Rejected)); @@ -995,11 +1151,14 @@ mod integration_tests { async fn test_concurrency_manager_put_admission_reuses_released_permit() { let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::ZERO); - let first = manager.admit_put_object().await.expect("first put admission should acquire"); + let first = manager + .admit_put_object(1024) + .await + .expect("first put admission should acquire"); drop(first); let second = manager - .admit_put_object() + .admit_put_object(1024) .await .expect("released put admission permit should be reusable"); assert!(matches!(second, PutObjectAdmission::Admitted(_))); @@ -1009,10 +1168,13 @@ mod integration_tests { #[serial] async fn test_concurrency_manager_put_admission_wait_timeout_rejects() { let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::from_secs(5)); - let held = manager.admit_put_object().await.expect("first put admission should acquire"); + let held = manager + .admit_put_object(1024) + .await + .expect("first put admission should acquire"); let waiter_manager = manager.clone(); - let waiter = tokio::spawn(async move { waiter_manager.admit_put_object().await }); + let waiter = tokio::spawn(async move { waiter_manager.admit_put_object(1024).await }); tokio::task::yield_now().await; tokio::time::advance(Duration::from_secs(5)).await; @@ -1024,6 +1186,85 @@ mod integration_tests { drop(held); } + #[tokio::test] + #[serial] + async fn test_concurrency_manager_large_put_admission_bypasses_small_puts() { + let min_size = rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES; + let manager = ConcurrencyManager::with_large_put_admission_for_test(true, 1, min_size, Duration::ZERO); + + let held = manager + .admit_put_object(min_size as i64) + .await + .expect("large put admission should acquire"); + assert!(matches!(held, PutObjectAdmission::Admitted(_))); + + let small = manager + .admit_put_object((min_size - 1) as i64) + .await + .expect("small put should bypass large admission"); + assert!(matches!(small, PutObjectAdmission::Disabled)); + + let large = manager + .admit_put_object(min_size as i64) + .await + .expect("second large put should reject when the gate is full"); + assert!(matches!(large, PutObjectAdmission::Rejected)); + } + + #[tokio::test] + #[serial] + async fn test_concurrency_manager_large_put_admission_gates_unknown_size() { + let manager = ConcurrencyManager::with_large_put_admission_for_test(true, 1, 32 * 1024 * 1024, Duration::ZERO); + + let held = manager + .admit_put_object(-1) + .await + .expect("unknown-size put admission should acquire"); + assert!(matches!(held, PutObjectAdmission::Admitted(_))); + + let second = manager + .admit_put_object(-1) + .await + .expect("unknown-size put admission should reject when the gate is full"); + assert!(matches!(second, PutObjectAdmission::Rejected)); + } + + #[tokio::test] + #[serial] + async fn test_concurrency_manager_large_put_snapshot_tracks_gate() { + let manager = ConcurrencyManager::with_large_put_admission_for_test(true, 2, 32 * 1024 * 1024, Duration::ZERO); + let first = manager + .admit_put_object(32 * 1024 * 1024) + .await + .expect("first large put admission should acquire"); + let initial = manager.put_object_admission_snapshot(); + + assert_eq!(initial.class, WorkloadClass::ForegroundWrite); + assert_eq!(initial.state, AdmissionState::Open); + assert_eq!(initial.active, Some(1)); + assert_eq!(initial.limit, Some(2)); + + let second = manager + .admit_put_object(32 * 1024 * 1024) + .await + .expect("second large put admission should acquire"); + let saturated = manager.put_object_admission_snapshot(); + + assert_eq!(saturated.state, AdmissionState::Saturated); + assert_eq!(saturated.active, Some(2)); + assert_eq!(saturated.limit, Some(2)); + drop((first, second)); + } + + #[test] + fn test_concurrency_manager_derives_large_put_admission_limit_from_scheduler_cap() { + assert_eq!(derive_large_put_admission_limit(7, 64), 7); + assert_eq!(derive_large_put_admission_limit(0, 64), 32); + assert_eq!(derive_large_put_admission_limit(0, 8), 4); + assert_eq!(derive_large_put_admission_limit(0, 1), 1); + assert_eq!(derive_large_put_admission_limit(0, 0), 32); + } + #[tokio::test] #[serial] async fn test_concurrency_manager_workload_admission_registry_covers_required_classes() {