diff --git a/Cargo.lock b/Cargo.lock index 91b0e5a25..bfd252fd5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9331,6 +9331,7 @@ dependencies = [ "rustfs-io-metrics", "thiserror 2.0.18", "tokio", + "tracing", ] [[package]] diff --git a/crates/concurrency/src/config.rs b/crates/concurrency/src/config.rs index 6fe7e1733..8f010c104 100644 --- a/crates/concurrency/src/config.rs +++ b/crates/concurrency/src/config.rs @@ -117,27 +117,19 @@ impl ConcurrencyConfig { let mut config = Self::default(); // Read from environment if available - if let Ok(val) = std::env::var("RUSTFS_TIMEOUT_DEFAULT") - && let Ok(secs) = val.parse::() - { + if let Some(secs) = parse_env::("RUSTFS_TIMEOUT_DEFAULT") { config.timeout_policy.default_timeout = Duration::from_secs(secs); } - if let Ok(val) = std::env::var("RUSTFS_TIMEOUT_MAX") - && let Ok(secs) = val.parse::() - { + if let Some(secs) = parse_env::("RUSTFS_TIMEOUT_MAX") { config.timeout_policy.max_timeout = Duration::from_secs(secs); } - if let Ok(val) = std::env::var("RUSTFS_BACKPRESSURE_BUFFER_SIZE") - && let Ok(size) = val.parse::() - { + if let Some(size) = parse_env::("RUSTFS_BACKPRESSURE_BUFFER_SIZE") { config.backpressure_policy.buffer_size = size; } - if let Ok(val) = std::env::var("RUSTFS_IO_BUFFER_SIZE") - && let Ok(size) = val.parse::() - { + if let Some(size) = parse_env::("RUSTFS_IO_BUFFER_SIZE") { config.scheduler_policy.base_buffer_size = size; } @@ -146,6 +138,9 @@ impl ConcurrencyConfig { /// Validate configuration pub fn validate(&self) -> Result<(), ConfigError> { + if self.timeout_policy.default_timeout.is_zero() || self.timeout_policy.max_timeout.is_zero() { + return Err(ConfigError::InvalidTimeout("timeouts must be > 0".to_string())); + } if self.timeout_policy.default_timeout > self.timeout_policy.max_timeout { return Err(ConfigError::InvalidTimeout("default_timeout cannot exceed max_timeout".to_string())); } @@ -153,6 +148,10 @@ impl ConcurrencyConfig { return Err(ConfigError::InvalidTimeout("min_timeout cannot exceed max_timeout".to_string())); } + // A zero-capacity pipe (duplex(0)) never accepts writes, hanging producers forever. + if self.backpressure_policy.buffer_size == 0 { + return Err(ConfigError::InvalidBackpressure("buffer_size must be > 0".to_string())); + } if self.backpressure_policy.high_watermark <= self.backpressure_policy.low_watermark || self.backpressure_policy.high_watermark > 100 { @@ -161,16 +160,37 @@ impl ConcurrencyConfig { )); } + if self.scheduler_policy.base_buffer_size == 0 { + return Err(ConfigError::InvalidScheduler("base_buffer_size must be > 0".to_string())); + } if self.scheduler_policy.base_buffer_size > self.scheduler_policy.max_buffer_size { return Err(ConfigError::InvalidScheduler( "base_buffer_size cannot exceed max_buffer_size".to_string(), )); } + // Ensure the derived io-core config also holds its invariants. + self.scheduler_policy + .to_core_config() + .validate() + .map_err(|e| ConfigError::InvalidScheduler(e.to_string()))?; Ok(()) } } +/// Parse an environment variable, logging a warning instead of silently +/// falling back to the default when the value is set but unparsable. +fn parse_env(name: &str) -> Option { + let val = std::env::var(name).ok()?; + match val.parse() { + Ok(parsed) => Some(parsed), + Err(_) => { + tracing::warn!(env_var = name, value = %val, "ignoring unparsable environment variable"); + None + } + } +} + /// Configuration error #[allow(clippy::enum_variant_names)] #[derive(Debug, Clone, thiserror::Error)] @@ -231,6 +251,69 @@ mod tests { ); } + #[test] + fn test_zero_backpressure_buffer_size_rejected() { + let config = ConcurrencyConfig { + backpressure_policy: PipeBackpressurePolicy { + buffer_size: 0, + ..Default::default() + }, + ..Default::default() + }; + assert!( + matches!(config.validate(), Err(ConfigError::InvalidBackpressure(_))), + "validate() must reject buffer_size=0 (duplex(0) hangs all writes)" + ); + } + + #[test] + fn test_zero_timeouts_rejected() { + let config = ConcurrencyConfig { + timeout_policy: TimeoutManagerPolicy { + default_timeout: Duration::ZERO, + ..Default::default() + }, + ..Default::default() + }; + assert!(matches!(config.validate(), Err(ConfigError::InvalidTimeout(_)))); + + let config = ConcurrencyConfig { + timeout_policy: TimeoutManagerPolicy { + max_timeout: Duration::ZERO, + ..Default::default() + }, + ..Default::default() + }; + assert!(matches!(config.validate(), Err(ConfigError::InvalidTimeout(_)))); + } + + #[test] + fn test_zero_scheduler_base_buffer_size_rejected() { + let config = ConcurrencyConfig { + scheduler_policy: SchedulerPolicy { + base_buffer_size: 0, + ..Default::default() + }, + ..Default::default() + }; + assert!(matches!(config.validate(), Err(ConfigError::InvalidScheduler(_)))); + } + + #[test] + fn test_small_max_buffer_size_passes_core_validation() { + // max_buffer_size below the io-core default min_buffer_size (4KB) must + // still yield a valid core config (min lowered by to_core_config). + let config = ConcurrencyConfig { + scheduler_policy: SchedulerPolicy { + base_buffer_size: 1024, + max_buffer_size: 2048, + ..Default::default() + }, + ..Default::default() + }; + assert!(config.validate().is_ok()); + } + #[test] fn test_features() { let features = ConcurrencyFeatures::all(); diff --git a/crates/concurrency/src/scheduler.rs b/crates/concurrency/src/scheduler.rs index 34e989d8a..fc892f10e 100644 --- a/crates/concurrency/src/scheduler.rs +++ b/crates/concurrency/src/scheduler.rs @@ -49,12 +49,16 @@ impl Default for SchedulerPolicy { impl SchedulerPolicy { /// Convert facade policy to io-core scheduler config. pub fn to_core_config(&self) -> rustfs_io_core::IoSchedulerConfig { + let default = rustfs_io_core::IoSchedulerConfig::default(); rustfs_io_core::IoSchedulerConfig { base_buffer_size: self.base_buffer_size, max_buffer_size: self.max_buffer_size, + // Keep min <= base <= max: the io-core default min (4KB) would make + // `clamp(min, max)` panic for policies with a smaller max_buffer_size. + min_buffer_size: default.min_buffer_size.min(self.base_buffer_size).min(self.max_buffer_size), high_priority_size_threshold: self.high_priority_threshold, low_priority_size_threshold: self.low_priority_threshold, - ..rustfs_io_core::IoSchedulerConfig::default() + ..default } } } @@ -266,6 +270,16 @@ mod tests { assert!(priority.is_high()); } + #[test] + fn test_small_max_buffer_size_does_not_panic() { + // Regression: max_buffer_size below the io-core default min (4KB) used to + // panic in the core clamp (min > max) on the first buffer-size calculation. + let manager = SchedulerManager::new(1024, 2048, 512, 2048); + let size = manager.calculate_buffer_size(1024, StorageMedia::Ssd, AccessPattern::Sequential, IoLoadLevel::Medium, 1); + assert!(size > 0 && size <= 2048); + assert!(manager.core_config().validate().is_ok()); + } + #[test] fn test_io_strategy() { let manager = SchedulerManager::new(1024, 4096, 512, 2048); diff --git a/crates/io-core/Cargo.toml b/crates/io-core/Cargo.toml index 72c7801ab..604ef61fb 100644 --- a/crates/io-core/Cargo.toml +++ b/crates/io-core/Cargo.toml @@ -33,6 +33,7 @@ thiserror = { workspace = true } tokio = { workspace = true, features = ["io-util", "fs", "rt", "sync"] } memmap2 = { workspace = true } rustfs-io-metrics = { workspace = true } +tracing = { workspace = true } [dev-dependencies] tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } diff --git a/crates/io-core/src/backpressure.rs b/crates/io-core/src/backpressure.rs index e23637cbe..fd127a949 100644 --- a/crates/io-core/src/backpressure.rs +++ b/crates/io-core/src/backpressure.rs @@ -223,7 +223,18 @@ impl BackpressureMonitor { /// Release a slot after operation completes. pub fn release(&self) { - let prev = self.current.fetch_sub(1, Ordering::Relaxed); + // Guard against underflow: an unpaired release at 0 must not wrap to + // usize::MAX, which would permanently reject all future acquisitions. + let prev = match self + .current + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)) + { + Ok(prev) => prev, + Err(_) => { + tracing::warn!("BackpressureMonitor::release called with no outstanding acquisition; ignoring"); + 0 + } + }; let low_threshold = self.config.low_threshold(); // Update state if needed @@ -362,6 +373,22 @@ mod tests { assert_eq!(monitor.state(), BackpressureState::Normal); } + #[test] + fn test_release_underflow_stays_at_zero() { + // An unpaired release at current==0 must not wrap to usize::MAX, + // which would make try_acquire reject everything forever. + let monitor = BackpressureMonitor::with_defaults(); + + monitor.release(); + assert_eq!(monitor.current(), 0); + + // The monitor must still be usable. + assert!(monitor.try_acquire()); + assert_eq!(monitor.current(), 1); + monitor.release(); + assert_eq!(monitor.current(), 0); + } + #[test] fn test_rejection_rate() { let config = BackpressureConfig { diff --git a/crates/io-core/src/scheduler.rs b/crates/io-core/src/scheduler.rs index 53ce3748a..76afa58cf 100644 --- a/crates/io-core/src/scheduler.rs +++ b/crates/io-core/src/scheduler.rs @@ -36,7 +36,14 @@ pub enum IoPriority { impl IoPriority { /// Determine priority based on request size. + /// + /// A negative `size` means the size is unknown (-1 by convention) and maps + /// to `Normal`; casting it to `usize` would wrap to a huge value and + /// misclassify the request as `Low`. pub fn from_size(size: i64, high_threshold: usize, low_threshold: usize) -> Self { + if size < 0 { + return IoPriority::Normal; + } let size = size as usize; if size < high_threshold { IoPriority::High @@ -731,6 +738,13 @@ mod tests { assert_eq!(IoPriority::from_size(10 * 1024 * 1024, 64 * 1024, 4 * 1024 * 1024), IoPriority::Low); } + #[test] + fn test_io_priority_unknown_size_is_normal() { + // -1 means "size unknown" and must not wrap to usize::MAX (=> Low). + assert_eq!(IoPriority::from_size(-1, 64 * 1024, 4 * 1024 * 1024), IoPriority::Normal); + assert_eq!(IoPriority::from_size(i64::MIN, 64 * 1024, 4 * 1024 * 1024), IoPriority::Normal); + } + #[test] fn test_io_load_level() { let low = Duration::from_millis(5); diff --git a/crates/io-core/src/timeout_wrapper.rs b/crates/io-core/src/timeout_wrapper.rs index 426911925..9d9ce4b22 100644 --- a/crates/io-core/src/timeout_wrapper.rs +++ b/crates/io-core/src/timeout_wrapper.rs @@ -375,23 +375,19 @@ pub fn calculate_adaptive_timeout( 1.0 // No adjustment }; + // Adaptive timeout bounds: 5 seconds minimum, 10 minutes maximum. + const MIN_SECS: f64 = 5.0; + const MAX_SECS: f64 = 600.0; + // If we have historical rate data, use it for estimation - let estimated_duration = if let Some(rate) = historical_rate_bps { - if rate > 0 { - let estimated_secs = (object_size as f64 / rate as f64) * 1.2; // 20% buffer - Duration::from_secs_f64(estimated_secs) - } else { - base_timeout - } - } else { - base_timeout + let estimated_secs = match historical_rate_bps { + Some(rate) if rate > 0 => (object_size as f64 / rate as f64) * 1.2, // 20% buffer + _ => base_timeout.as_secs_f64(), }; - // Apply timeout multiplier but clamp to reasonable bounds - let adaptive_duration = Duration::from_secs_f64(estimated_duration.as_secs_f64() * timeout_multiplier); - - // Clamp to 5 seconds minimum and 10 minutes maximum - adaptive_duration.clamp(Duration::from_secs(5), Duration::from_secs(600)) + // Clamp BEFORE constructing the Duration: `from_secs_f64` panics when the + // estimate overflows Duration (huge object_size with a tiny historical rate). + Duration::from_secs_f64((estimated_secs * timeout_multiplier).clamp(MIN_SECS, MAX_SECS)) } /// Estimate bytes per second transfer rate. @@ -435,6 +431,18 @@ mod tests { assert!(config.validate().is_err()); } + #[test] + fn test_adaptive_timeout_extreme_estimate_does_not_panic() { + // A huge object with a tiny historical rate used to overflow + // Duration::from_secs_f64 and panic; it must clamp to the upper bound. + let timeout = calculate_adaptive_timeout(Duration::from_secs(30), Some(1), 0, u64::MAX); + assert_eq!(timeout, Duration::from_secs(600)); + + // Tiny estimates clamp to the lower bound. + let timeout = calculate_adaptive_timeout(Duration::from_secs(30), Some(u64::MAX), 0, 1); + assert_eq!(timeout, Duration::from_secs(5)); + } + #[test] fn test_operation_progress() { let progress = OperationProgress::new(Some(1000), Duration::from_secs(5));