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 <heihutu@gmail.com>

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

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"))?
         {

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<ConcurrencyManager> = LazyLock::new(ConcurrencyManager::new);
 
@@ -65,11 +67,8 @@ pub struct ConcurrencyManager {
     bandwidth_monitor: Arc<Mutex<BandwidthMonitor>>,
     /// Metrics collector for I/O latency tracking (P50, P95, P99)
     metrics_collector: Arc<MetricsCollector>,
-    /// Experimental fixed-count foreground PutObject admission gate.
+    /// Foreground PutObject admission policy, resolved once at startup.
-    put_admission_semaphore: Arc<Semaphore>,
+    put_admission_policy: PutAdmissionPolicy,
-    put_admission_enabled: bool,
-    put_admission_limit: usize,
-    put_admission_wait_timeout: Duration,
 }
 
 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<Semaphore>,
+    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<PutObjectAdmission, tokio::sync::AcquireError> {
+        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<PutObjectAdmission, tokio::sync::AcquireError> {
+        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(
+        let put_admission_policy = PutAdmissionPolicy::from_env(max_disk_reads);
-            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,
-        ));
 
         // 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_policy,
-            put_admission_enabled,
-            put_admission_limit,
-            put_admission_wait_timeout,
         }
     }
 
@@ -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_policy = PutAdmissionPolicy::strict_for_test(enabled, limit, wait_timeout);
-        manager.put_admission_enabled = enabled;
+        manager
-        manager.put_admission_limit = limit;
+    }
-        manager.put_admission_wait_timeout = wait_timeout;
+
+    #[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
+    /// The strict experimental gate applies to every PUT only when explicitly
-    /// touching the semaphore, preserving legacy behavior. When enabled, the
+    /// enabled. Otherwise the default-on large-object gate protects sustained
-    /// permit must be acquired before body ingest and held until the store write
+    /// erasure/RPC pressure while keeping small PUTs on the legacy path.
-    /// returns, so saturated foreground writes can fail with `SlowDown` before
+    pub async fn admit_put_object(&self, size: i64) -> Result<PutObjectAdmission, tokio::sync::AcquireError> {
-    /// creating visible side effects.
+        self.put_admission_policy.admit(size).await
-    pub async fn admit_put_object(&self) -> Result<PutObjectAdmission, tokio::sync::AcquireError> {
-        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),
-        }
     }
 
     // ============================================
@@ -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_policy.snapshot(self.scheduler_config.max_concurrent_reads)
-            (
-                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,
-        }
     }
 
     /// 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() {