feat(storage): add default-off PUT admission gate (#6197)

Add an experimental fixed-count foreground PutObject admission gate for #1882 Phase 0 validation. The gate is default-off, returns SlowDown before body ingest when saturated, and keeps the admission permit with the spawned store commit owner until store PUT returns.

Co-authored-by: heihutu <heihutu@gmail.com>

crates/config/src/constants/object.rs

@@ -234,6 +234,31 @@ pub const ENV_OBJECT_DISK_WRITE_ABSOLUTE_CAP: &str = "RUSTFS_OBJECT_DISK_WRITE_A
 /// Default absolute per-object erasure write cap in seconds (`0` = disabled).
 pub const DEFAULT_OBJECT_DISK_WRITE_ABSOLUTE_CAP: u64 = 0;
 
+/// Enable foreground PutObject request admission.
+///
+/// This is an experimental, default-off foreground write backpressure gate for
+/// strict commit tail investigations. When disabled, PUTs follow the legacy
+/// path and only the existing request counters are updated.
+pub const ENV_PUT_FOREGROUND_ADMISSION_ENABLE: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_ENABLE";
+pub const DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE: bool = false;
+
+/// Maximum foreground PutObject requests admitted concurrently per process.
+///
+/// The limit is used only when [`ENV_PUT_FOREGROUND_ADMISSION_ENABLE`] is true.
+/// A value of `0` disables the gate even when the enable flag is present, so a
+/// partially configured rollout cannot reject every PUT.
+pub const ENV_PUT_FOREGROUND_ADMISSION_LIMIT: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_LIMIT";
+pub const DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT: usize = 0;
+
+/// Time in milliseconds a foreground PutObject waits for an admission permit.
+///
+/// Once this timeout expires the request fails before body ingest/storage
+/// mutation with S3 `SlowDown`/503. `0` means fail fast when the limit is full.
+pub const ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS";
+pub const DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 0;
+
+const _: () = assert!(!DEFAULT_PUT_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_usecase.rs

@@ -56,8 +56,8 @@ use super::storage_api::object_usecase::bucket::{
 };
 use super::storage_api::object_usecase::compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible};
 use super::storage_api::object_usecase::concurrency::{
-    self, ConcurrencyManager, DiskReadAdmission, GetObjectGuard, PutObjectGuard, get_concurrency_aware_buffer_size,
-    get_concurrency_manager, get_put_concurrency_aware_buffer_size,
+    self, ConcurrencyManager, DiskReadAdmission, GetObjectGuard, PutObjectAdmission, PutObjectGuard,
+    get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size,
 };
 #[cfg(test)]
 use super::storage_api::object_usecase::contract::http::HTTPPreconditions;
@@ -5681,6 +5681,35 @@ impl DefaultObjectUsecase {
         let server_side_encryption_requested =
             server_side_encryption.is_some() || sse_customer_algorithm.is_some() || ssekms_key_id.is_some();
 
+        // Resolve the store through the request-bound server context
+        // (backlog#1052 S6), not the process-global handle, so an embedded
+        // second server never writes into the first server's store.
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+        let bucket_validate_stage_start = put_stage_metrics_enabled.then(Instant::now);
+        validate_bucket_exists(&store, &bucket).await?;
+        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()
+            .await
+            .map_err(|_| s3_error!(InternalError, "foreground write admission closed"))?
+        {
+            PutObjectAdmission::Disabled => None,
+            PutObjectAdmission::Admitted(permit) => {
+                counter!("rustfs.put_object.foreground_admission.total", "result" => "admitted").increment(1);
+                Some(permit)
+            }
+            PutObjectAdmission::Rejected => {
+                counter!("rustfs.put_object.foreground_admission.total", "result" => "rejected").increment(1);
+                return Err(s3_error!(
+                    SlowDown,
+                    "foreground write concurrency limit reached, please reduce your request rate"
+                ));
+            }
+        };
+
         let mut put_request_guard = PutObjectGuard::new();
         let concurrent_put_requests = PutObjectGuard::concurrent_requests();
 
@@ -5733,16 +5762,6 @@ impl DefaultObjectUsecase {
             use_large_put_concurrency_tuning,
         );
 
-        // Resolve the store through the request-bound server context
-        // (backlog#1052 S6), not the process-global handle, so an embedded
-        // second server never writes into the first server's store.
-        let Some(store) = self.object_store() else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
-        };
-        let bucket_validate_stage_start = put_stage_metrics_enabled.then(Instant::now);
-        validate_bucket_exists(&store, &bucket).await?;
-        rustfs_io_metrics::record_put_object_stage_duration_from("app_bucket_validate", bucket_validate_stage_start);
-
         let sse_config_stage_start = put_stage_metrics_enabled.then(Instant::now);
         let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok();
         rustfs_io_metrics::record_put_object_stage_duration_from("app_sse_config_lookup", sse_config_stage_start);
@@ -6132,7 +6151,9 @@ impl DefaultObjectUsecase {
             let cache_adapter = cache_adapter.clone();
             let request_id = request_id.clone();
             let put_path = put_path.to_string();
+            let put_admission = put_admission;
             async move {
+                let _put_admission = put_admission;
                 let object_traffic_progress = object_traffic_health
                     .as_deref()
                     .and_then(ObjectTrafficHealth::track_write_storage);
@@ -6183,6 +6204,7 @@ impl DefaultObjectUsecase {
                     }
                 };
                 rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start);
+                drop(_put_admission);
                 drop(object_traffic_progress);
                 #[cfg(test)]
                 wait_for_put_post_store_test_hook(&bucket).await;

rustfs/src/app/storage_api.rs

@@ -936,7 +936,7 @@ pub(crate) mod bucket {
 
 pub(crate) mod concurrency {
     pub(crate) use crate::storage::storage_api::concurrency_consumer::{
-        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard,
+        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectAdmission, PutObjectGuard,
         get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size,
     };
 }

rustfs/src/storage/concurrency/manager.rs

@@ -65,6 +65,11 @@ 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.
+    put_admission_semaphore: Arc<Semaphore>,
+    put_admission_enabled: bool,
+    put_admission_limit: usize,
+    put_admission_wait_timeout: Duration,
 }
 
 impl std::fmt::Debug for ConcurrencyManager {
@@ -114,6 +119,18 @@ pub enum DiskReadAdmission {
     Rejected,
 }
 
+/// Outcome of foreground PutObject request admission.
+#[derive(Debug)]
+pub enum PutObjectAdmission {
+    /// Foreground PUT admission is disabled; proceed on the legacy path.
+    Disabled,
+    /// 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.
+    Rejected,
+}
+
 #[allow(dead_code)]
 impl ConcurrencyManager {
     /// Create a new concurrency manager with default settings
@@ -161,6 +178,18 @@ 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,
+        ));
 
         // Build queue config directly from scheduler config.
         let queue_config = IoPriorityQueueConfig::from_scheduler_config(&scheduler_config);
@@ -176,6 +205,10 @@ 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,
         }
     }
 
@@ -199,6 +232,16 @@ impl ConcurrencyManager {
         self.degraded_read_semaphore.close();
     }
 
+    #[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
+    }
+
     /// Track a GetObject request
     pub fn track_request() -> GetObjectGuard {
         GetObjectGuard::new()
@@ -284,6 +327,32 @@ impl ConcurrencyManager {
         }
     }
 
+    /// Admit a foreground PutObject request under the experimental fixed-count 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<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),
+        }
+    }
+
     // ============================================
     // Adaptive I/O Strategy Methods
     // ============================================
@@ -692,8 +761,16 @@ impl ConcurrencyManager {
 
     /// Get a read-only workload admission snapshot for foreground writes.
     pub fn put_object_admission_snapshot(&self) -> WorkloadAdmissionSnapshot {
-        let active = PutObjectGuard::concurrent_count();
-        let limit = self.scheduler_config.max_concurrent_reads;
+        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 {
@@ -706,7 +783,10 @@ impl ConcurrencyManager {
             WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit));
 
         match state {
-            AdmissionState::Disabled => admission.with_reason("foreground write pressure tracking disabled"),
+            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,
         }
@@ -783,7 +863,7 @@ impl Default for ConcurrencyManager {
 mod integration_tests {
     use super::super::io_schedule::{IoLoadLevel, IoPriority};
     use super::super::request_guard::GetObjectGuard;
-    use super::ConcurrencyManager;
+    use super::{ConcurrencyManager, PutObjectAdmission};
     use crate::storage::storage_api::concurrency_consumer::PutObjectGuard;
     use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass};
     use rustfs_io_core::io_profile::{AccessPattern, StorageMedia};
@@ -880,6 +960,71 @@ mod integration_tests {
         crate::storage::concurrency::reset_active_put_requests();
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn test_concurrency_manager_put_admission_disabled_does_not_touch_gate() {
+        let manager = ConcurrencyManager::with_put_admission_for_test(false, 1, Duration::ZERO);
+
+        let admission = manager
+            .admit_put_object()
+            .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_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");
+        assert!(matches!(first, PutObjectAdmission::Admitted(_)));
+        assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Saturated);
+
+        let second = manager
+            .admit_put_object()
+            .await
+            .expect("full put admission gate should reject, not close");
+        assert!(matches!(second, PutObjectAdmission::Rejected));
+    }
+
+    #[tokio::test]
+    #[serial]
+    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");
+        drop(first);
+
+        let second = manager
+            .admit_put_object()
+            .await
+            .expect("released put admission permit should be reusable");
+        assert!(matches!(second, PutObjectAdmission::Admitted(_)));
+    }
+
+    #[tokio::test(start_paused = true)]
+    #[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 waiter_manager = manager.clone();
+
+        let waiter = tokio::spawn(async move { waiter_manager.admit_put_object().await });
+        tokio::task::yield_now().await;
+        tokio::time::advance(Duration::from_secs(5)).await;
+
+        let admission = waiter
+            .await
+            .expect("put admission waiter task must not panic")
+            .expect("put admission gate must stay open");
+        assert!(matches!(admission, PutObjectAdmission::Rejected));
+        drop(held);
+    }
+
     #[tokio::test]
     #[serial]
     async fn test_concurrency_manager_workload_admission_registry_covers_required_classes() {

rustfs/src/storage/concurrency/mod.rs

@@ -54,7 +54,7 @@ pub use io_schedule::{
 pub use request_guard::{GetObjectGuard, PutObjectGuard};
 
 // Concurrency manager
-pub use manager::{ConcurrencyManager, DiskReadAdmission};
+pub use manager::{ConcurrencyManager, DiskReadAdmission, PutObjectAdmission};
 
 // ============================================
 // New Module Re-exports (for gradual migration)

rustfs/src/storage/storage_api.rs

@@ -117,7 +117,7 @@ pub(crate) mod access_consumer {
 
 pub(crate) mod concurrency_consumer {
     pub(crate) use super::super::concurrency::{
-        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard,
+        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectAdmission, PutObjectGuard,
         get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size,
     };
 }