fix: harden io-core and concurrency boundaries against panics (#4503)

fix(io-core,concurrency): harden boundary validation and arithmetic against panics (backlog#1024)

Cargo.lock

@@ -9331,6 +9331,7 @@ dependencies = [
  "rustfs-io-metrics",
  "thiserror 2.0.18",
  "tokio",
+ "tracing",
 ]
 
 [[package]]

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::<u64>()
-        {
+        if let Some(secs) = parse_env::<u64>("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::<u64>()
-        {
+        if let Some(secs) = parse_env::<u64>("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::<usize>()
-        {
+        if let Some(size) = parse_env::<usize>("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::<usize>()
-        {
+        if let Some(size) = parse_env::<usize>("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<T: std::str::FromStr>(name: &str) -> Option<T> {
+    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();

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);

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"] }

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 {

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);

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));