diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 081308754..874933aa6 100644 --- a/crates/config/src/constants/object.rs +++ b/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 diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 643b4fb6e..1bb337c69 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/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; diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 839b8a38f..c6f03f240 100644 --- a/rustfs/src/app/storage_api.rs +++ b/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, }; } diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index 13b509f93..5bc456148 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -65,6 +65,11 @@ 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, } 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 { + 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() { diff --git a/rustfs/src/storage/concurrency/mod.rs b/rustfs/src/storage/concurrency/mod.rs index 110e14cf6..9b14eebda 100644 --- a/rustfs/src/storage/concurrency/mod.rs +++ b/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) diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index d1e7c0f3a..be9548619 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/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, }; }