refactor(concurrency): remove zero-caller facade modules, fix feature build (#4530)

refactor(concurrency): remove zero-caller facade modules and fix no-default-features build (backlog#1025) The audit in rustfs/backlog#1010 (consistent with #805) established that most of crates/concurrency was a decorative facade with zero production callers; the real runtime concurrency control lives in rustfs/src/storage/*. This deletes the dead facades and keeps only what the workspace actually consumes. Deleted (zero callers verified by workspace-wide grep): - manager.rs: ConcurrencyManager, lifecycle start/stop, misleading 'started' lifecycle logs - config.rs: ConcurrencyConfig, ConcurrencyFeatures, from_env - timeout.rs: TimeoutManager, TimeoutGuard, TimeoutManagerPolicy - lock.rs: LockManager, LockScopeGuard, OptimizedLockGuard - scheduler.rs: SchedulerManager, SchedulerPolicy, IoStrategy - deadlock.rs facade: DeadlockManager, RequestTracker - backpressure.rs facade: BackpressureManager, BackpressurePipe - the prelude module, unused io-core re-exports, and all feature flags Kept (real callers in ecstore/heal/rustfs): - workload.rs admission contract types (unchanged) - workers.rs Workers pool (unchanged, retained per #4498) - GetObjectQueueSnapshot (moved from manager.rs to new queue.rs) - PipeBackpressurePolicy (used by rustfs/src/storage/backpressure.rs) - DeadlockMonitorPolicy (used by rustfs/src/storage/deadlock_detector.rs) - OperationProgress re-export (used by rustfs/src/storage/timeout_wrapper.rs) Removing the feature flags fixes the previously broken cargo check -p rustfs-concurrency --no-default-features (E0432). Docs and the logging guardrail file list are updated to match. Ref: rustfs/backlog#1025
2026-08-28 07:57:01 +00:00 · 2026-07-09 01:12:43 +08:00
parent a7b9659e75
commit 05833063c7
15 changed files with 152 additions and 2022 deletions
@@ -141,7 +141,7 @@ Depth 8 — TOP:
 | `io-metrics` | 4.5K | I/O operation metrics and counters |
 | `rio` | 6.9K | Composable reader chain (encrypt → compress → hash → limit) |
 | `object-io` | 2.4K | High-level object read/write using rio + ecstore |
-| `concurrency` | 1.8K | Concurrency control wrappers over io-core |
+| `concurrency` | 0.8K | Shared concurrency contract types: workload admission snapshots, worker-slot pool, policy types (runtime control lives in `rustfs/src/storage`) |
 **Storage Engine:**
@@ -9027,12 +9027,9 @@ version = "1.0.0-beta.8"
 dependencies = [
 "insta",
 "rustfs-io-core",
 "rustfs-io-metrics",
 "serde",
 "serde_json",
 "thiserror 2.0.18",
 "tokio",
 "tokio-util",
 "tracing",
 ]
@@ -6,22 +6,17 @@ license.workspace = true
 repository.workspace = true
 rust-version.workspace = true
 homepage.workspace = true
-description = "Concurrency management for RustFS - timeout, locking, backpressure, and I/O scheduling"
+description = "Shared concurrency contract types for RustFS - workload admission, queue snapshots, worker pools, and policy types"
-keywords = ["rustfs", "concurrency", "timeout", "backpressure", "scheduling"]
+keywords = ["rustfs", "concurrency", "admission", "backpressure", "workers"]
 categories = ["concurrency", "filesystem"]
 [dependencies]
 # Internal crates
 rustfs-io-core = { workspace = true }
 rustfs-io-metrics = { workspace = true }
 serde = { workspace = true }
 # Async runtime
-tokio = { workspace = true, features = ["sync", "time", "rt"] }
+tokio = { workspace = true, features = ["sync"] }
 tokio-util = { workspace = true }
 # Error handling
 thiserror = { workspace = true }
 # Logging
 tracing = { workspace = true }
@@ -29,18 +24,4 @@ tracing = { workspace = true }
 [dev-dependencies]
 insta = { workspace = true }
 serde_json = { workspace = true }
-tokio = { workspace = true, features = ["test-util","macros","rt-multi-thread"] }
+tokio = { workspace = true, features = ["test-util", "macros", "rt-multi-thread"] }
 [features]
 default = ["timeout", "lock", "deadlock", "backpressure", "scheduler"]
 # Feature modules
 timeout = []
 lock = []
 deadlock = []
 backpressure = []
 scheduler = []
 [package.metadata.docs.rs]
 all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
@@ -12,17 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
-//! Backpressure management
+//! Shared backpressure policy type.
 //!
 //! The runtime backpressure implementation (byte-watermark pipes and
 //! monitors) lives in `rustfs/src/storage/backpressure.rs`; this module only
 //! carries the watermark policy type that implementation shares.
-use rustfs_io_core::{
+use rustfs_io_core::BackpressureConfig as CoreBackpressureConfig;
    BackpressureConfig as CoreBackpressureConfig, BackpressureMonitor as CoreBackpressureMonitor, BackpressureState,
 };
 use rustfs_io_metrics::backpressure_metrics;
 use std::sync::Arc;
 use std::time::Instant;
 use tokio::io::{DuplexStream, duplex};
-/// Facade policy for duplex-pipe watermark backpressure.
+/// Watermark policy for duplex-pipe backpressure.
 #[derive(Debug, Clone, Copy)]
 pub struct PipeBackpressurePolicy {
    /// Buffer size in bytes
@@ -54,9 +52,9 @@ impl PipeBackpressurePolicy {
        (self.buffer_size as u64 * self.low_watermark as u64 / 100) as usize
    }
-    /// Convert the facade policy into the reusable io-core admission-pressure config.
+    /// Convert the policy into the reusable io-core admission-pressure config.
    ///
-    /// The concurrency layer still owns duplex buffer sizing, but the shared
+    /// The caller still owns duplex buffer sizing, but the shared
    /// overload/admission primitive lives in `io-core`.
    pub fn to_core_config(&self) -> CoreBackpressureConfig {
        CoreBackpressureConfig {
@@ -69,157 +67,28 @@ impl PipeBackpressurePolicy {
    }
 }
 /// Backpressure manager
 pub struct BackpressureManager {
    config: PipeBackpressurePolicy,
    core_config: CoreBackpressureConfig,
    monitor: Arc<CoreBackpressureMonitor>,
 }
 impl BackpressureManager {
    /// Create a new backpressure manager
    pub fn new(buffer_size: usize, high_watermark: u32, low_watermark: u32) -> Self {
        Self::from_policy(PipeBackpressurePolicy {
            buffer_size,
            high_watermark,
            low_watermark,
        })
    }
    /// Create a new backpressure manager from the facade policy type.
    pub fn from_policy(config: PipeBackpressurePolicy) -> Self {
        let core_config = config.to_core_config();
        Self {
            config,
            core_config: core_config.clone(),
            monitor: Arc::new(CoreBackpressureMonitor::new(core_config)),
        }
    }
    /// Get the configuration
    pub fn config(&self) -> &PipeBackpressurePolicy {
        &self.config
    }
    /// Get the derived io-core admission-pressure configuration.
    pub fn core_config(&self) -> &CoreBackpressureConfig {
        &self.core_config
    }
    /// Get the monitor
    pub fn monitor(&self) -> Arc<CoreBackpressureMonitor> {
        self.monitor.clone()
    }
    /// Create a backpressure pipe
    pub fn create_pipe(&self) -> BackpressurePipe {
        BackpressurePipe::new(self.config, self.monitor.clone())
    }
    /// Get current state
    pub fn state(&self) -> BackpressureState {
        self.monitor.state()
    }
    /// Check if backpressure is active
    pub fn is_active(&self) -> bool {
        self.monitor.is_active()
    }
 }
 /// Backpressure pipe wrapping tokio's duplex
 pub struct BackpressurePipe {
    reader: DuplexStream,
    writer: DuplexStream,
    config: PipeBackpressurePolicy,
    monitor: Arc<CoreBackpressureMonitor>,
    created_at: Instant,
 }
 /// Shared pipe metadata snapshot for facade-level backpressure pipes.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct BackpressurePipeMeta {
    /// Configured duplex buffer capacity in bytes.
    pub buffer_capacity: usize,
    /// Current backpressure state reported by the shared core monitor.
    pub state: BackpressureState,
    /// Age of the pipe since creation.
    pub age: std::time::Duration,
 }
 impl BackpressurePipe {
    fn new(config: PipeBackpressurePolicy, monitor: Arc<CoreBackpressureMonitor>) -> Self {
        let (reader, writer) = duplex(config.buffer_size);
        Self {
            reader,
            writer,
            config,
            monitor,
            created_at: Instant::now(),
        }
    }
    /// Get the reader end
    pub fn reader(&mut self) -> &mut DuplexStream {
        &mut self.reader
    }
    /// Get the writer end
    pub fn writer(&mut self) -> &mut DuplexStream {
        &mut self.writer
    }
    /// Split into reader and writer
    pub fn into_split(self) -> (DuplexStream, DuplexStream) {
        (self.reader, self.writer)
    }
    /// Get the configuration
    pub fn config(&self) -> &PipeBackpressurePolicy {
        &self.config
    }
    /// Get current state
    pub fn state(&self) -> BackpressureState {
        self.monitor.state()
    }
    /// Get the age of this pipe
    pub fn age(&self) -> std::time::Duration {
        self.created_at.elapsed()
    }
    /// Get a compact metadata snapshot for the pipe.
    pub fn meta(&self) -> BackpressurePipeMeta {
        BackpressurePipeMeta {
            buffer_capacity: self.config.buffer_size,
            state: self.state(),
            age: self.age(),
        }
    }
    /// Check if should apply backpressure
    pub fn should_apply_backpressure(&self) -> bool {
        let should = self.monitor.should_apply_backpressure();
        if should {
            backpressure_metrics::record_backpressure_activation();
        }
        should
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
-    fn test_backpressure_config() {
+    fn test_backpressure_policy_defaults() {
        let config = PipeBackpressurePolicy::default();
        assert_eq!(config.buffer_size, 4 * 1024 * 1024);
        assert!(config.high_watermark > config.low_watermark);
    }
    #[test]
    fn test_backpressure_policy_watermark_bytes() {
        let config = PipeBackpressurePolicy {
            buffer_size: 1000,
            high_watermark: 80,
            low_watermark: 50,
        };
        assert_eq!(config.high_watermark_bytes(), 800);
        assert_eq!(config.low_watermark_bytes(), 500);
    }
    #[test]
    fn test_backpressure_policy_to_core_config() {
        let policy = PipeBackpressurePolicy::default();
@@ -228,33 +97,4 @@ mod tests {
        assert_eq!(core.low_water_mark, policy.low_watermark as f64 / 100.0);
        assert!(core.enabled);
    }
    #[test]
    fn test_backpressure_manager() {
        let manager = BackpressureManager::new(1024, 80, 50);
        assert_eq!(manager.state(), BackpressureState::Normal);
    }
    #[test]
    fn test_backpressure_pipe() {
        let manager = BackpressureManager::new(1024, 80, 50);
        let pipe = manager.create_pipe();
        assert_eq!(pipe.state(), BackpressureState::Normal);
        assert_eq!(pipe.meta().buffer_capacity, 1024);
    }
    #[test]
    fn test_backpressure_pipe_meta_is_read_only() {
        let manager = BackpressureManager::new(2048, 80, 50);
        let pipe = manager.create_pipe();
        let first = pipe.meta();
        let second = pipe.meta();
        assert_eq!(first.buffer_capacity, 2048);
        assert_eq!(first.state, BackpressureState::Normal);
        assert_eq!(second.buffer_capacity, first.buffer_capacity);
        assert_eq!(second.state, first.state);
        assert!(second.age >= first.age);
        assert_eq!(manager.state(), BackpressureState::Normal);
    }
 }
@@ -1,325 +0,0 @@
 // Copyright 2024 RustFS Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //! Configuration for concurrency management
 use crate::{
    backpressure::PipeBackpressurePolicy, deadlock::DeadlockMonitorPolicy, scheduler::SchedulerPolicy,
    timeout::TimeoutManagerPolicy,
 };
 use std::time::Duration;
 /// Feature flags for concurrency modules
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct ConcurrencyFeatures {
    /// Enable timeout control
    pub timeout: bool,
    /// Enable lock optimization
    pub lock: bool,
    /// Enable deadlock detection
    pub deadlock: bool,
    /// Enable backpressure management
    pub backpressure: bool,
    /// Enable I/O scheduling
    pub scheduler: bool,
 }
 impl Default for ConcurrencyFeatures {
    fn default() -> Self {
        Self {
            timeout: cfg!(feature = "timeout"),
            lock: cfg!(feature = "lock"),
            deadlock: cfg!(feature = "deadlock"),
            backpressure: cfg!(feature = "backpressure"),
            scheduler: cfg!(feature = "scheduler"),
        }
    }
 }
 impl ConcurrencyFeatures {
    /// Create with all features enabled
    pub fn all() -> Self {
        Self {
            timeout: true,
            lock: true,
            deadlock: true,
            backpressure: true,
            scheduler: true,
        }
    }
    /// Create with no features enabled
    pub fn none() -> Self {
        Self {
            timeout: false,
            lock: false,
            deadlock: false,
            backpressure: false,
            scheduler: false,
        }
    }
    /// Check if any feature is enabled
    pub fn any_enabled(&self) -> bool {
        self.timeout || self.lock || self.deadlock || self.backpressure || self.scheduler
    }
 }
 /// Facade policy for lock manager behavior.
 #[derive(Debug, Clone, Copy)]
 pub struct LockManagerPolicy {
    /// Enable lock optimization.
    pub enabled: bool,
    /// Lock acquisition timeout.
    pub acquire_timeout: Duration,
 }
 impl Default for LockManagerPolicy {
    fn default() -> Self {
        Self {
            enabled: true,
            acquire_timeout: Duration::from_secs(5),
        }
    }
 }
 /// Main configuration for concurrency management
 #[derive(Debug, Clone, Default)]
 pub struct ConcurrencyConfig {
    /// Feature flags
    pub features: ConcurrencyFeatures,
    /// Timeout facade policy.
    pub timeout_policy: TimeoutManagerPolicy,
    /// Lock facade policy.
    pub lock_policy: LockManagerPolicy,
    /// Deadlock facade policy.
    pub deadlock_policy: DeadlockMonitorPolicy,
    /// Backpressure facade policy.
    pub backpressure_policy: PipeBackpressurePolicy,
    /// Scheduler facade policy.
    pub scheduler_policy: SchedulerPolicy,
 }
 impl ConcurrencyConfig {
    /// Create configuration from environment variables
    pub fn from_env() -> Self {
        let mut config = Self::default();
        // Read from environment if available
        if let Some(secs) = parse_env::<u64>("RUSTFS_TIMEOUT_DEFAULT") {
            config.timeout_policy.default_timeout = Duration::from_secs(secs);
        }
        if let Some(secs) = parse_env::<u64>("RUSTFS_TIMEOUT_MAX") {
            config.timeout_policy.max_timeout = Duration::from_secs(secs);
        }
        if let Some(size) = parse_env::<usize>("RUSTFS_BACKPRESSURE_BUFFER_SIZE") {
            config.backpressure_policy.buffer_size = size;
        }
        if let Some(size) = parse_env::<usize>("RUSTFS_IO_BUFFER_SIZE") {
            config.scheduler_policy.base_buffer_size = size;
        }
        config
    }
    /// 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()));
        }
        if self.timeout_policy.min_timeout > self.timeout_policy.max_timeout {
            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
        {
            return Err(ConfigError::InvalidBackpressure(
                "high_watermark must be > low_watermark and <= 100".to_string(),
            ));
        }
        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)]
 pub enum ConfigError {
    /// Invalid timeout configuration
    #[error("Invalid timeout config: {0}")]
    InvalidTimeout(String),
    /// Invalid backpressure configuration
    #[error("Invalid backpressure config: {0}")]
    InvalidBackpressure(String),
    /// Invalid scheduler configuration
    #[error("Invalid scheduler config: {0}")]
    InvalidScheduler(String),
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn test_default_config() {
        let config = ConcurrencyConfig::default();
        assert!(config.validate().is_ok());
    }
    #[test]
    fn test_invalid_timeout() {
        let config = ConcurrencyConfig {
            timeout_policy: TimeoutManagerPolicy {
                default_timeout: Duration::from_secs(100),
                max_timeout: Duration::from_secs(50),
                enable_dynamic: true,
                ..Default::default()
            },
            ..Default::default()
        };
        assert!(
            config.validate().is_err(),
            "validate() should return an error when default_timeout > max_timeout"
        );
    }
    #[test]
    fn test_invalid_min_timeout() {
        let config = ConcurrencyConfig {
            timeout_policy: TimeoutManagerPolicy {
                min_timeout: Duration::from_secs(100),
                max_timeout: Duration::from_secs(50),
                ..Default::default()
            },
            ..Default::default()
        };
        assert!(
            config.validate().is_err(),
            "validate() should return an error when min_timeout > max_timeout"
        );
    }
    #[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();
        assert!(features.any_enabled());
        let features = ConcurrencyFeatures::none();
        assert!(!features.any_enabled());
    }
 }
@@ -12,15 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
-//! Deadlock detection management
+//! Shared deadlock-monitor policy type.
 //!
 //! The runtime request-hang / deadlock detection loop lives in
 //! `rustfs/src/storage/deadlock_detector.rs`; this module only carries the
 //! monitor policy type that implementation shares.
-use rustfs_io_core::{DeadlockDetector as CoreDeadlockDetector, DeadlockDetectorConfig as CoreDeadlockConfig, LockType};
+use rustfs_io_core::DeadlockDetectorConfig as CoreDeadlockConfig;
-use rustfs_io_metrics::deadlock_metrics;
+use std::time::Duration;
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
-/// Facade policy for the concurrency-layer deadlock monitor.
+/// Policy for the request-hang deadlock monitor.
 #[derive(Debug, Clone, Copy)]
 pub struct DeadlockMonitorPolicy {
    /// Enable deadlock detection
@@ -42,7 +43,7 @@ impl Default for DeadlockMonitorPolicy {
 }
 impl DeadlockMonitorPolicy {
-    /// Convert the facade policy into the reusable io-core deadlock config.
+    /// Convert the policy into the reusable io-core deadlock config.
    pub fn to_core_config(&self) -> CoreDeadlockConfig {
        CoreDeadlockConfig {
            enabled: self.enabled,
@@ -52,182 +53,14 @@ impl DeadlockMonitorPolicy {
    }
 }
 /// Deadlock manager
 pub struct DeadlockManager {
    config: DeadlockMonitorPolicy,
    detector: Arc<CoreDeadlockDetector>,
    running: Arc<tokio::sync::Mutex<bool>>,
 }
 impl DeadlockManager {
    /// Create a new deadlock manager
    pub fn new(enabled: bool, check_interval: Duration, hang_threshold: Duration) -> Self {
        Self::from_policy(DeadlockMonitorPolicy {
            enabled,
            check_interval,
            hang_threshold,
        })
    }
    /// Create a new deadlock manager from the facade policy type.
    pub fn from_policy(config: DeadlockMonitorPolicy) -> Self {
        let core_config = config.to_core_config();
        Self {
            config,
            detector: Arc::new(CoreDeadlockDetector::new(core_config)),
            running: Arc::new(tokio::sync::Mutex::new(false)),
        }
    }
    /// Get the configuration
    pub fn config(&self) -> &DeadlockMonitorPolicy {
        &self.config
    }
    /// Get the core detector
    pub fn detector(&self) -> Arc<CoreDeadlockDetector> {
        self.detector.clone()
    }
    /// Start the deadlock detection background task
    pub async fn start(&self) {
        if !self.config.enabled {
            return;
        }
        let mut running = self.running.lock().await;
        if *running {
            return;
        }
        *running = true;
        drop(running);
        tracing::info!(
            event = "deadlock_monitor.lifecycle",
            component = "concurrency",
            subsystem = "deadlock",
            state = "started",
            check_interval_ms = self.config.check_interval.as_millis(),
            hang_threshold_ms = self.config.hang_threshold.as_millis(),
            "deadlock monitor state changed"
        );
    }
    /// Stop the deadlock detection
    pub async fn stop(&self) {
        let mut running = self.running.lock().await;
        *running = false;
        tracing::info!(
            event = "deadlock_monitor.lifecycle",
            component = "concurrency",
            subsystem = "deadlock",
            state = "stopped",
            check_interval_ms = self.config.check_interval.as_millis(),
            hang_threshold_ms = self.config.hang_threshold.as_millis(),
            "deadlock monitor state changed"
        );
    }
    /// Create a request tracker
    pub fn track_request(&self, request_id: String, description: String) -> RequestTracker {
        RequestTracker::new(request_id, description, self.detector.clone())
    }
    /// Register a lock
    pub fn register_lock(&self, lock_type: LockType) -> u64 {
        self.detector.register_lock(lock_type)
    }
    /// Unregister a lock
    pub fn unregister_lock(&self, lock_id: u64) {
        self.detector.unregister_lock(lock_id);
    }
    /// Detect deadlock
    pub fn detect_deadlock(&self) -> Option<Vec<u64>> {
        let result = self.detector.detect_deadlock();
        if let Some(ref cycle) = result {
            deadlock_metrics::record_deadlock_detected(cycle.len());
        }
        result
    }
 }
 /// Lightweight compatibility wrapper for request-scoped deadlock bookkeeping.
 ///
 /// This type intentionally stays minimal in the concurrency layer. Rich
 /// request-level lock/resource diagnostics belong to
 /// `rustfs::storage::deadlock_detector::RequestResourceTracker`.
 pub struct RequestTracker {
    request_id: String,
    description: String,
    start_time: Instant,
    resources: HashMap<String, Vec<String>>,
    detector: Arc<CoreDeadlockDetector>,
 }
 impl RequestTracker {
    fn new(request_id: String, description: String, detector: Arc<CoreDeadlockDetector>) -> Self {
        let start_time = Instant::now();
        detector.register_request(&request_id, 1); // Use placeholder thread ID
        Self {
            request_id,
            description,
            start_time,
            resources: HashMap::new(),
            detector,
        }
    }
    /// Get the request ID
    pub fn request_id(&self) -> &str {
        &self.request_id
    }
    /// Get the description
    pub fn description(&self) -> &str {
        &self.description
    }
    /// Get the elapsed time
    pub fn elapsed(&self) -> Duration {
        self.start_time.elapsed()
    }
    /// Record a lock acquisition
    pub fn record_lock_acquire(&mut self, lock_id: u64, resource: String) {
        self.resources.entry("locks".to_string()).or_default().push(resource);
        self.detector.record_acquire(lock_id, 1); // Use placeholder thread ID
        deadlock_metrics::record_lock_acquisition("read");
    }
    /// Return a read-only view of tracked resource names.
    pub fn resources(&self) -> &HashMap<String, Vec<String>> {
        &self.resources
    }
    /// Record a lock release
    pub fn record_lock_release(&mut self, lock_id: u64) {
        self.detector.record_release(lock_id);
    }
 }
 impl Drop for RequestTracker {
    fn drop(&mut self) {
        self.detector.unregister_request(&self.request_id);
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
-    fn test_deadlock_manager_creation() {
+    fn test_deadlock_policy_defaults_disabled() {
-        let manager = DeadlockManager::new(false, Duration::from_secs(10), Duration::from_secs(60));
+        let policy = DeadlockMonitorPolicy::default();
-        assert!(!manager.config().enabled);
+        assert!(!policy.enabled);
    }
    #[test]
@@ -238,16 +71,4 @@ mod tests {
        assert_eq!(core.detection_interval, policy.check_interval);
        assert_eq!(core.max_hold_time, policy.hang_threshold);
    }
    #[tokio::test]
    async fn test_request_tracker() {
        let manager = DeadlockManager::new(true, Duration::from_secs(10), Duration::from_secs(60));
        let mut tracker = manager.track_request("req-1".to_string(), "test request".to_string());
        let lock_id = manager.register_lock(LockType::Mutex);
        tracker.record_lock_acquire(lock_id, "bucket/key".to_string());
        assert_eq!(tracker.request_id(), "req-1");
        assert_eq!(tracker.description(), "test request");
        assert_eq!(tracker.resources().get("locks").map(Vec::len), Some(1));
    }
 }
@@ -12,167 +12,40 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
-//! # RustFS Concurrency Management
+//! # RustFS Concurrency Contracts
 //!
-//! This crate provides comprehensive concurrency management for RustFS,
+//! Shared concurrency contract types for RustFS:
 //! including timeout control, lock optimization, deadlock detection,
 //! backpressure management, and I/O scheduling.
 //!
-//! ## Features
+//! - [`workload`]: workload admission snapshot/contract types consumed by
 //!   ecstore, heal, and the server for admission-aware throttling.
 //! - [`GetObjectQueueSnapshot`]: disk permit queue usage snapshot for
 //!   GetObject orchestration.
 //! - [`workers`]: a bounded semaphore-backed worker-slot pool.
 //! - [`PipeBackpressurePolicy`] / [`DeadlockMonitorPolicy`]: policy types
 //!   shared with the runtime implementations.
 //!
-//! All features are controlled by feature flags and can be enabled/disabled at compile time:
+//! The actual runtime concurrency control — size-aware timeouts, byte-watermark
-//!
+//! backpressure, request-hang/deadlock detection, and I/O scheduling — is
-//! - **timeout**: Dynamic timeout calculation based on data size and transfer rate
+//! implemented in `rustfs/src/storage/*` on top of the `rustfs-io-core`
-//! - **lock**: Early lock release to reduce contention
+//! primitives. This crate only carries the data and contract types those
-//! - **deadlock**: Request tracking and cycle detection
+//! implementations share; it does not run any background tasks itself.
 //! - **backpressure**: Buffer-based flow control
 //! - **scheduler**: Adaptive buffer sizing and priority queuing
 //!
 //! ## Architecture
 //!
 //! ```text
 //! rustfs-concurrency (Business Layer)
 //!     ├── timeout (Timeout Control)
 //!     ├── lock (Lock Optimization)
 //!     ├── deadlock (Deadlock Detection)
 //!     ├── backpressure (Backpressure Management)
 //!     └── scheduler (I/O Scheduling)
 //!          │
 //!          ├── rustfs-io-core (Core Algorithms)
 //!          └── rustfs-io-metrics (Metrics Collection)
 //! ```
 //!
 //! ## Usage
 //!
 //! ```rust,no_run
 //! use rustfs_concurrency::{ConcurrencyConfig, ConcurrencyManager};
 //!
 //! # #[tokio::main]
 //! # async fn main() {
 //! // Create manager with all features enabled
 //! let config = ConcurrencyConfig::default();
 //! let manager = ConcurrencyManager::new(config);
 //!
 //! // Start services
 //! manager.start().await;
 //!
 //! // Use timeout control (if enabled)
 //! if manager.is_timeout_enabled() {
 //!     let timeout_manager = manager.timeout();
 //!     let _ = timeout_manager;
 //! }
 //!
 //! // Use lock optimization (if enabled)
 //! if manager.is_lock_enabled() {
 //!     let lock_manager = manager.lock();
 //!     let _ = lock_manager;
 //! }
 //!
 //! // Stop services
 //! manager.stop().await;
 //! # }
 //! ```
 #![deny(missing_docs)]
 #![deny(unsafe_code)]
 #![cfg_attr(docsrs, feature(doc_cfg))]
-// Re-export core types from io-core
+// Re-exported io-core type used by downstream timeout implementations.
-pub use rustfs_io_core::{
+pub use rustfs_io_core::OperationProgress;
    // Backpressure types
    BackpressureConfig as CoreBackpressureConfig,
    BackpressureMonitor as CoreBackpressureMonitor,
    BackpressureState,
    // Deadlock types
    DeadlockDetector as CoreDeadlockDetector,
    IoLoadLevel,
    IoLoadMetrics,
    IoPriority,
    // Scheduler types
    IoScheduler,
    IoSchedulingContext,
    LockInfo,
    LockOptimizer as CoreLockOptimizer,
    // Lock types
    LockStats as CoreLockStats,
    LockType,
    // Timeout types
    OperationProgress,
    TimeoutError,
    TimeoutStats,
    WaitGraphEdge,
    calculate_adaptive_timeout,
    estimate_bytes_per_second,
 };
 // Module declarations with feature gates
 #[cfg(feature = "timeout")]
 mod timeout;
 #[cfg(feature = "lock")]
 mod lock;
 #[cfg(feature = "deadlock")]
 mod deadlock;
 #[cfg(feature = "backpressure")]
 mod backpressure;
-
+mod deadlock;
-#[cfg(feature = "scheduler")]
+mod queue;
 mod scheduler;
 pub mod workers;
 pub mod workload;
-// Public module exports with feature gates
+pub use backpressure::PipeBackpressurePolicy;
-#[cfg(feature = "timeout")]
+pub use deadlock::DeadlockMonitorPolicy;
-pub use timeout::{TimeoutGuard, TimeoutManager, TimeoutManagerPolicy};
+pub use queue::GetObjectQueueSnapshot;
 #[cfg(feature = "lock")]
 pub use lock::{LockConfig, LockManager, LockScopeGuard, OptimizedLockGuard};
 #[cfg(feature = "deadlock")]
 pub use deadlock::{DeadlockManager, DeadlockMonitorPolicy, RequestTracker};
 #[cfg(feature = "backpressure")]
 pub use backpressure::{BackpressureManager, BackpressurePipe, PipeBackpressurePolicy};
 #[cfg(feature = "scheduler")]
 pub use scheduler::{IoStrategy, SchedulerManager, SchedulerPolicy};
 // Configuration
 mod config;
 pub use config::{ConcurrencyConfig, ConcurrencyFeatures};
 // Manager
 mod manager;
 pub use manager::{ConcurrencyManager, GetObjectQueueSnapshot};
 pub use workload::{
    AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadAdmissionSnapshotProvider,
    WorkloadClass,
 };
 // Prelude for convenient imports
 pub mod prelude {
    //! Prelude module for convenient imports
    #[cfg(feature = "timeout")]
    pub use crate::timeout::{TimeoutGuard, TimeoutManager, TimeoutManagerPolicy};
    #[cfg(feature = "lock")]
    pub use crate::lock::{LockConfig, LockManager, LockScopeGuard, OptimizedLockGuard};
    #[cfg(feature = "deadlock")]
    pub use crate::deadlock::{DeadlockManager, DeadlockMonitorPolicy, RequestTracker};
    #[cfg(feature = "backpressure")]
    pub use crate::backpressure::{BackpressureManager, BackpressurePipe, PipeBackpressurePolicy};
    #[cfg(feature = "scheduler")]
    pub use crate::scheduler::{IoStrategy, SchedulerManager, SchedulerPolicy};
    pub use crate::{AdmissionState, ConcurrencyConfig, ConcurrencyFeatures, ConcurrencyManager, WorkloadAdmissionSnapshot};
 }
@@ -1,227 +0,0 @@
 // Copyright 2024 RustFS Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //! Lock optimization management
 use rustfs_io_core::{LockOptimizer as CoreLockOptimizer, LockStats};
 use rustfs_io_metrics::lock_metrics;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 /// Lock configuration
 #[derive(Debug, Clone)]
 pub struct LockConfig {
    /// Enable lock optimization
    pub enabled: bool,
    /// Lock acquisition timeout
    pub acquire_timeout: Duration,
 }
 impl Default for LockConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            acquire_timeout: Duration::from_secs(5),
        }
    }
 }
 /// Lock manager
 pub struct LockManager {
    config: LockConfig,
    optimizer: Arc<CoreLockOptimizer>,
 }
 impl LockManager {
    /// Create a new lock manager
    pub fn new(enabled: bool, acquire_timeout: Duration) -> Self {
        let config = LockConfig {
            enabled,
            acquire_timeout,
        };
        let core_config = rustfs_io_core::LockOptimizeConfig {
            enabled,
            acquire_timeout,
            max_hold_time_warning: Duration::from_millis(100),
            adaptive_spin: true,
            max_spin_iterations: 1000,
        };
        Self {
            config,
            optimizer: Arc::new(CoreLockOptimizer::new(core_config)),
        }
    }
    /// Get the configuration
    pub fn config(&self) -> &LockConfig {
        &self.config
    }
    /// Get the core optimizer
    pub fn optimizer(&self) -> Arc<CoreLockOptimizer> {
        self.optimizer.clone()
    }
    /// Get lock statistics
    pub fn stats(&self) -> &LockStats {
        self.optimizer.stats()
    }
    /// Optimize a lock guard
    pub fn optimize<G>(&self, guard: G, resource: impl Into<String>) -> OptimizedLockGuard<G> {
        OptimizedLockGuard::new(guard, resource, self.optimizer.clone())
    }
    /// Check if optimization is enabled
    pub fn is_enabled(&self) -> bool {
        self.config.enabled
    }
 }
 /// Optimized lock guard with early release support
 pub struct OptimizedLockGuard<G> {
    guard: Option<G>,
    acquire_time: Instant,
    released: bool,
    resource: String,
    optimizer: Arc<CoreLockOptimizer>,
 }
 impl<G> OptimizedLockGuard<G> {
    fn new(guard: G, resource: impl Into<String>, optimizer: Arc<CoreLockOptimizer>) -> Self {
        optimizer.on_acquire();
        lock_metrics::record_lock_optimization_enabled(optimizer.config().enabled);
        Self {
            guard: Some(guard),
            acquire_time: Instant::now(),
            released: false,
            resource: resource.into(),
            optimizer,
        }
    }
    /// Get the lock hold time
    pub fn hold_time(&self) -> Duration {
        self.acquire_time.elapsed()
    }
    /// Check if the lock has been released
    pub fn is_released(&self) -> bool {
        self.released
    }
    /// Release the lock early
    pub fn early_release(&mut self) {
        if self.released {
            return;
        }
        let hold_time = self.hold_time();
        self.guard.take();
        self.released = true;
        self.optimizer.on_release(hold_time);
        lock_metrics::record_lock_hold_time(hold_time);
        tracing::debug!(
            event = "lock_guard.release",
            component = "concurrency",
            subsystem = "lock",
            release_mode = "early",
            resource = %self.resource,
            hold_time_ms = hold_time.as_millis(),
            "lock guard released"
        );
    }
    /// Get a reference to the underlying guard
    pub fn as_ref(&self) -> Option<&G> {
        if self.released { None } else { self.guard.as_ref() }
    }
 }
 impl<G> Drop for OptimizedLockGuard<G> {
    fn drop(&mut self) {
        if !self.released {
            let hold_time = self.hold_time();
            self.guard.take();
            self.released = true;
            self.optimizer.on_release(hold_time);
            lock_metrics::record_lock_hold_time(hold_time);
            tracing::debug!(
                event = "lock_guard.release",
                component = "concurrency",
                subsystem = "lock",
                release_mode = "drop",
                resource = %self.resource,
                hold_time_ms = hold_time.as_millis(),
                "lock guard released"
            );
        }
    }
 }
 /// Lock scope guard for RAII semantics
 pub struct LockScopeGuard<G> {
    guard: Option<G>,
 }
 impl<G> LockScopeGuard<G> {
    /// Create a new scope guard
    pub fn new(guard: G) -> Self {
        Self { guard: Some(guard) }
    }
    /// Get a reference to the guard
    pub fn as_ref(&self) -> Option<&G> {
        self.guard.as_ref()
    }
 }
 impl<G> Drop for LockScopeGuard<G> {
    fn drop(&mut self) {
        self.guard.take();
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    use std::sync::Mutex;
    #[test]
    fn test_lock_manager_creation() {
        let manager = LockManager::new(true, Duration::from_secs(5));
        assert!(manager.is_enabled());
    }
    #[test]
    fn test_optimized_lock_guard() {
        let manager = LockManager::new(true, Duration::from_secs(5));
        let mutex = Mutex::new(42);
        let guard = mutex.lock().unwrap();
        let optimized = manager.optimize(guard, "test_resource");
        assert!(!optimized.is_released());
        let mut optimized = optimized;
        optimized.early_release();
        assert!(optimized.is_released());
    }
 }
@@ -1,397 +0,0 @@
 // Copyright 2024 RustFS Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //! Main concurrency manager
 use crate::config::{ConcurrencyConfig, ConfigError};
 use std::sync::Arc;
 /// Snapshot of disk permit queue usage for GetObject orchestration.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct GetObjectQueueSnapshot {
    /// Total permits configured for disk reads.
    pub total_permits: usize,
    /// Permits currently in use.
    pub permits_in_use: usize,
 }
 impl GetObjectQueueSnapshot {
    /// Create a queue snapshot from total and available permits.
    pub fn from_available_permits(total_permits: usize, available_permits: usize) -> Self {
        Self {
            total_permits,
            permits_in_use: total_permits.saturating_sub(available_permits),
        }
    }
    /// Return currently available permits.
    pub fn permits_available(&self) -> usize {
        self.total_permits.saturating_sub(self.permits_in_use)
    }
    /// Return queue utilization percentage in the 0-100 range.
    pub fn utilization_percent(&self) -> f64 {
        if self.total_permits == 0 {
            0.0
        } else {
            (self.permits_in_use as f64 / self.total_permits as f64) * 100.0
        }
    }
    /// Return whether the queue is considered congested.
    pub fn is_congested(&self, threshold_percent: f64) -> bool {
        self.utilization_percent() > threshold_percent
    }
 }
 /// Main concurrency manager that provides access to all concurrency features
 pub struct ConcurrencyManager {
    config: ConcurrencyConfig,
    #[cfg(feature = "timeout")]
    timeout: Arc<crate::timeout::TimeoutManager>,
    #[cfg(feature = "lock")]
    lock: Arc<crate::lock::LockManager>,
    #[cfg(feature = "deadlock")]
    deadlock: Arc<crate::deadlock::DeadlockManager>,
    #[cfg(feature = "backpressure")]
    backpressure: Arc<crate::backpressure::BackpressureManager>,
    #[cfg(feature = "scheduler")]
    scheduler: Arc<crate::scheduler::SchedulerManager>,
 }
 impl ConcurrencyManager {
    fn build(config: ConcurrencyConfig) -> Self {
        Self {
            #[cfg(feature = "timeout")]
            timeout: Arc::new(crate::timeout::TimeoutManager::from_policy(config.timeout_policy)),
            #[cfg(feature = "lock")]
            lock: Arc::new(crate::lock::LockManager::new(
                config.lock_policy.enabled,
                config.lock_policy.acquire_timeout,
            )),
            #[cfg(feature = "deadlock")]
            deadlock: Arc::new(crate::deadlock::DeadlockManager::from_policy(config.deadlock_policy)),
            #[cfg(feature = "backpressure")]
            backpressure: Arc::new(crate::backpressure::BackpressureManager::from_policy(config.backpressure_policy)),
            #[cfg(feature = "scheduler")]
            scheduler: Arc::new(crate::scheduler::SchedulerManager::from_policy(config.scheduler_policy)),
            config,
        }
    }
    /// Try to create a new concurrency manager with the given configuration.
    pub fn try_new(config: ConcurrencyConfig) -> Result<Self, ConfigError> {
        config.validate()?;
        Ok(Self::build(config))
    }
    /// Create a new concurrency manager with the given configuration.
    ///
    /// Invalid configurations are downgraded to the default configuration instead of
    /// panicking so startup/runtime callers can remain fail-safe in production paths.
    pub fn new(config: ConcurrencyConfig) -> Self {
        match Self::try_new(config) {
            Ok(manager) => manager,
            Err(err) => {
                tracing::warn!(
                    event = "concurrency_manager.invalid_config_fallback",
                    component = "concurrency",
                    subsystem = "manager",
                    error = %err,
                    "Invalid concurrency configuration detected; falling back to defaults"
                );
                Self::build(ConcurrencyConfig::default())
            }
        }
    }
    /// Create with default configuration
    pub fn with_defaults() -> Self {
        Self::build(ConcurrencyConfig::default())
    }
    /// Try to create a manager from environment-derived configuration.
    pub fn try_from_env() -> Result<Self, ConfigError> {
        Self::try_new(ConcurrencyConfig::from_env())
    }
    /// Create from environment variables
    pub fn from_env() -> Self {
        match Self::try_from_env() {
            Ok(manager) => manager,
            Err(err) => {
                tracing::warn!(
                    event = "concurrency_manager.invalid_env_fallback",
                    component = "concurrency",
                    subsystem = "manager",
                    error = %err,
                    "Invalid environment-derived concurrency configuration detected; falling back to defaults"
                );
                Self::build(ConcurrencyConfig::default())
            }
        }
    }
    /// Get the configuration
    pub fn config(&self) -> &ConcurrencyConfig {
        &self.config
    }
    // ============================================
    // Feature enable checks
    // ============================================
    /// Check if timeout feature is enabled
    pub fn is_timeout_enabled(&self) -> bool {
        #[cfg(feature = "timeout")]
        {
            self.config.features.timeout
        }
        #[cfg(not(feature = "timeout"))]
        {
            false
        }
    }
    /// Check if lock feature is enabled
    pub fn is_lock_enabled(&self) -> bool {
        #[cfg(feature = "lock")]
        {
            self.config.features.lock
        }
        #[cfg(not(feature = "lock"))]
        {
            false
        }
    }
    /// Check if deadlock feature is enabled
    pub fn is_deadlock_enabled(&self) -> bool {
        #[cfg(feature = "deadlock")]
        {
            self.config.features.deadlock
        }
        #[cfg(not(feature = "deadlock"))]
        {
            false
        }
    }
    /// Check if backpressure feature is enabled
    pub fn is_backpressure_enabled(&self) -> bool {
        #[cfg(feature = "backpressure")]
        {
            self.config.features.backpressure
        }
        #[cfg(not(feature = "backpressure"))]
        {
            false
        }
    }
    /// Check if scheduler feature is enabled
    pub fn is_scheduler_enabled(&self) -> bool {
        #[cfg(feature = "scheduler")]
        {
            self.config.features.scheduler
        }
        #[cfg(not(feature = "scheduler"))]
        {
            false
        }
    }
    // ============================================
    // Feature accessors
    // ============================================
    /// Get timeout manager
    #[cfg(feature = "timeout")]
    pub fn timeout(&self) -> Arc<crate::timeout::TimeoutManager> {
        self.timeout.clone()
    }
    /// Get lock manager
    #[cfg(feature = "lock")]
    pub fn lock(&self) -> Arc<crate::lock::LockManager> {
        self.lock.clone()
    }
    /// Get deadlock manager
    #[cfg(feature = "deadlock")]
    pub fn deadlock(&self) -> Arc<crate::deadlock::DeadlockManager> {
        self.deadlock.clone()
    }
    /// Get backpressure manager
    #[cfg(feature = "backpressure")]
    pub fn backpressure(&self) -> Arc<crate::backpressure::BackpressureManager> {
        self.backpressure.clone()
    }
    /// Get scheduler manager
    #[cfg(feature = "scheduler")]
    pub fn scheduler(&self) -> Arc<crate::scheduler::SchedulerManager> {
        self.scheduler.clone()
    }
    // ============================================
    // Lifecycle management
    // ============================================
    /// Start all enabled services (e.g., deadlock detection background task)
    pub async fn start(&self) {
        #[cfg(feature = "deadlock")]
        {
            if self.config.deadlock_policy.enabled {
                self.deadlock.start().await;
            }
        }
        tracing::info!(
            event = "concurrency_manager.lifecycle",
            component = "concurrency",
            subsystem = "manager",
            state = "started",
            timeout_enabled = self.is_timeout_enabled(),
            lock_enabled = self.is_lock_enabled(),
            deadlock_enabled = self.is_deadlock_enabled(),
            backpressure_enabled = self.is_backpressure_enabled(),
            scheduler_enabled = self.is_scheduler_enabled(),
            "concurrency manager state changed"
        );
    }
    /// Stop all services
    pub async fn stop(&self) {
        #[cfg(feature = "deadlock")]
        {
            self.deadlock.stop().await;
        }
        tracing::info!(
            event = "concurrency_manager.lifecycle",
            component = "concurrency",
            subsystem = "manager",
            state = "stopped",
            "concurrency manager state changed"
        );
    }
 }
 impl Default for ConcurrencyManager {
    fn default() -> Self {
        Self::with_defaults()
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn test_queue_snapshot() {
        let snapshot = GetObjectQueueSnapshot::from_available_permits(64, 16);
        assert_eq!(snapshot.permits_in_use, 48);
        assert_eq!(snapshot.permits_available(), 16);
        assert!(snapshot.is_congested(70.0));
    }
    #[test]
    fn test_queue_snapshot_clamps_over_available_permits() {
        let snapshot = GetObjectQueueSnapshot::from_available_permits(64, 96);
        assert_eq!(snapshot.permits_in_use, 0);
        assert_eq!(snapshot.permits_available(), 64);
        assert_eq!(snapshot.utilization_percent(), 0.0);
        assert!(!snapshot.is_congested(0.0));
    }
    #[test]
    fn test_queue_snapshot_handles_zero_total_permits() {
        let snapshot = GetObjectQueueSnapshot::from_available_permits(0, 0);
        assert_eq!(snapshot.permits_in_use, 0);
        assert_eq!(snapshot.permits_available(), 0);
        assert_eq!(snapshot.utilization_percent(), 0.0);
        assert!(!snapshot.is_congested(0.0));
    }
    #[test]
    fn test_manager_creation() {
        let manager = ConcurrencyManager::with_defaults();
        assert!(manager.config().validate().is_ok());
    }
    #[test]
    fn test_try_new_returns_error_for_invalid_config() {
        let config = ConcurrencyConfig {
            timeout_policy: crate::TimeoutManagerPolicy {
                default_timeout: std::time::Duration::from_secs(10),
                max_timeout: std::time::Duration::from_secs(1),
                ..Default::default()
            },
            ..Default::default()
        };
        let result = ConcurrencyManager::try_new(config);
        assert!(matches!(result, Err(ConfigError::InvalidTimeout(_))));
    }
    #[test]
    fn test_new_falls_back_to_default_for_invalid_config() {
        let config = ConcurrencyConfig {
            timeout_policy: crate::TimeoutManagerPolicy {
                default_timeout: std::time::Duration::from_secs(10),
                max_timeout: std::time::Duration::from_secs(1),
                ..Default::default()
            },
            ..Default::default()
        };
        let manager = ConcurrencyManager::new(config);
        assert!(manager.config().validate().is_ok());
        assert_eq!(
            manager.config().timeout_policy.default_timeout,
            ConcurrencyConfig::default().timeout_policy.default_timeout
        );
    }
    #[tokio::test]
    async fn test_manager_lifecycle() {
        let manager = ConcurrencyManager::with_defaults();
        manager.start().await;
        manager.stop().await;
    }
    #[test]
    fn test_feature_checks() {
        let manager = ConcurrencyManager::with_defaults();
        // These should return the feature flag status
        let _ = manager.is_timeout_enabled();
        let _ = manager.is_lock_enabled();
        let _ = manager.is_deadlock_enabled();
        let _ = manager.is_backpressure_enabled();
        let _ = manager.is_scheduler_enabled();
    }
 }
@@ -0,0 +1,84 @@
 // Copyright 2024 RustFS Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //! Disk permit queue snapshot for GetObject orchestration.
 /// Snapshot of disk permit queue usage for GetObject orchestration.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct GetObjectQueueSnapshot {
    /// Total permits configured for disk reads.
    pub total_permits: usize,
    /// Permits currently in use.
    pub permits_in_use: usize,
 }
 impl GetObjectQueueSnapshot {
    /// Create a queue snapshot from total and available permits.
    pub fn from_available_permits(total_permits: usize, available_permits: usize) -> Self {
        Self {
            total_permits,
            permits_in_use: total_permits.saturating_sub(available_permits),
        }
    }
    /// Return currently available permits.
    pub fn permits_available(&self) -> usize {
        self.total_permits.saturating_sub(self.permits_in_use)
    }
    /// Return queue utilization percentage in the 0-100 range.
    pub fn utilization_percent(&self) -> f64 {
        if self.total_permits == 0 {
            0.0
        } else {
            (self.permits_in_use as f64 / self.total_permits as f64) * 100.0
        }
    }
    /// Return whether the queue is considered congested.
    pub fn is_congested(&self, threshold_percent: f64) -> bool {
        self.utilization_percent() > threshold_percent
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn test_queue_snapshot() {
        let snapshot = GetObjectQueueSnapshot::from_available_permits(64, 16);
        assert_eq!(snapshot.permits_in_use, 48);
        assert_eq!(snapshot.permits_available(), 16);
        assert!(snapshot.is_congested(70.0));
    }
    #[test]
    fn test_queue_snapshot_clamps_over_available_permits() {
        let snapshot = GetObjectQueueSnapshot::from_available_permits(64, 96);
        assert_eq!(snapshot.permits_in_use, 0);
        assert_eq!(snapshot.permits_available(), 64);
        assert_eq!(snapshot.utilization_percent(), 0.0);
        assert!(!snapshot.is_congested(0.0));
    }
    #[test]
    fn test_queue_snapshot_handles_zero_total_permits() {
        let snapshot = GetObjectQueueSnapshot::from_available_permits(0, 0);
        assert_eq!(snapshot.permits_in_use, 0);
        assert_eq!(snapshot.permits_available(), 0);
        assert_eq!(snapshot.utilization_percent(), 0.0);
        assert!(!snapshot.is_congested(0.0));
    }
 }
@@ -1,292 +0,0 @@
 // Copyright 2024 RustFS Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //! I/O scheduler management
 use rustfs_io_core::{
    IoLoadLevel, IoPriority, IoScheduler as CoreIoScheduler, IoSchedulingContext,
    io_profile::{AccessPattern, StorageMedia},
 };
 use rustfs_io_metrics::io_metrics;
 use std::sync::Arc;
 use std::time::Duration;
 /// Facade policy for the concurrency-layer scheduler manager.
 #[derive(Debug, Clone, Copy)]
 pub struct SchedulerPolicy {
    /// Base buffer size
    pub base_buffer_size: usize,
    /// Maximum buffer size
    pub max_buffer_size: usize,
    /// High priority threshold
    pub high_priority_threshold: usize,
    /// Low priority threshold
    pub low_priority_threshold: usize,
 }
 impl Default for SchedulerPolicy {
    fn default() -> Self {
        Self {
            base_buffer_size: 64 * 1024,              // 64KB
            max_buffer_size: 4 * 1024 * 1024,         // 4MB
            high_priority_threshold: 1024 * 1024,     // 1MB
            low_priority_threshold: 10 * 1024 * 1024, // 10MB
        }
    }
 }
 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,
            ..default
        }
    }
 }
 /// Scheduler manager
 pub struct SchedulerManager {
    config: SchedulerPolicy,
    core_config: rustfs_io_core::IoSchedulerConfig,
    scheduler: Arc<CoreIoScheduler>,
 }
 impl SchedulerManager {
    /// Create a new scheduler manager
    pub fn new(
        base_buffer_size: usize,
        max_buffer_size: usize,
        high_priority_threshold: usize,
        low_priority_threshold: usize,
    ) -> Self {
        Self::from_policy(SchedulerPolicy {
            base_buffer_size,
            max_buffer_size,
            high_priority_threshold,
            low_priority_threshold,
        })
    }
    /// Create a scheduler manager from facade policy.
    pub fn from_policy(config: SchedulerPolicy) -> Self {
        let core_config = config.to_core_config();
        Self {
            config,
            core_config: core_config.clone(),
            scheduler: Arc::new(CoreIoScheduler::new(core_config)),
        }
    }
    /// Get the configuration
    pub fn config(&self) -> &SchedulerPolicy {
        &self.config
    }
    /// Get the derived io-core scheduler config.
    pub fn core_config(&self) -> &rustfs_io_core::IoSchedulerConfig {
        &self.core_config
    }
    /// Get the scheduler
    pub fn scheduler(&self) -> Arc<CoreIoScheduler> {
        self.scheduler.clone()
    }
    /// Create an I/O strategy
    pub fn create_strategy(&self) -> IoStrategy {
        IoStrategy::new(self.config, self.scheduler.clone())
    }
    /// Calculate buffer size
    pub fn calculate_buffer_size(
        &self,
        file_size: i64,
        media: StorageMedia,
        pattern: AccessPattern,
        load: IoLoadLevel,
        concurrent: usize,
    ) -> usize {
        let strategy = self.create_strategy();
        strategy.calculate_buffer_size(file_size, media, pattern, load, concurrent)
    }
    /// Get I/O priority
    pub fn get_priority(&self, size: i64) -> IoPriority {
        IoPriority::from_size(size, self.config.high_priority_threshold, self.config.low_priority_threshold)
    }
 }
 /// I/O strategy
 pub struct IoStrategy {
    config: SchedulerPolicy,
    scheduler: Arc<CoreIoScheduler>,
 }
 impl IoStrategy {
    fn new(config: SchedulerPolicy, scheduler: Arc<CoreIoScheduler>) -> Self {
        Self { config, scheduler }
    }
    /// Calculate buffer size with multi-factor strategy
    pub fn calculate_buffer_size(
        &self,
        file_size: i64,
        media: StorageMedia,
        pattern: AccessPattern,
        load: IoLoadLevel,
        concurrent: usize,
    ) -> usize {
        // Create scheduling context
        let _ctx = IoSchedulingContext::new(file_size, self.config.base_buffer_size)
            .with_sequential(matches!(pattern, AccessPattern::Sequential))
            .with_media(media);
        // Get base buffer size from core scheduler
        let permit_wait = Duration::from_millis(10); // Default wait time
        let is_sequential = matches!(pattern, AccessPattern::Sequential);
        let core_strategy = self.scheduler.calculate_strategy(file_size, permit_wait, is_sequential);
        let base_size = core_strategy.buffer_size;
        // Apply multi-factor adjustments
        let adjusted_size = self.apply_adjustments(base_size, media, pattern, load, concurrent);
        // Record metrics
        io_metrics::record_io_scheduler_decision(adjusted_size, load.as_str(), pattern.as_str());
        adjusted_size.min(self.config.max_buffer_size)
    }
    fn apply_adjustments(
        &self,
        base_size: usize,
        media: StorageMedia,
        pattern: AccessPattern,
        load: IoLoadLevel,
        concurrent: usize,
    ) -> usize {
        let mut size = base_size;
        // Media adjustment
        size = match media {
            StorageMedia::Nvme => (size as f64 * 1.5) as usize,
            StorageMedia::Ssd => (size as f64 * 1.2) as usize,
            StorageMedia::Hdd => size,
            _ => size,
        };
        // Pattern adjustment
        size = match pattern {
            AccessPattern::Sequential => (size as f64 * 1.5) as usize,
            AccessPattern::Random => (size as f64 * 0.5) as usize,
            _ => size,
        };
        // Load adjustment
        size = match load {
            IoLoadLevel::Low => (size as f64 * 1.2) as usize,
            IoLoadLevel::Medium => size,
            IoLoadLevel::High => (size as f64 * 0.7) as usize,
            IoLoadLevel::Critical => (size as f64 * 0.5) as usize,
        };
        // Concurrency adjustment
        if concurrent > 10 {
            size = (size as f64 * 0.8) as usize;
        }
        size
    }
    /// Get the configuration
    pub fn config(&self) -> &SchedulerPolicy {
        &self.config
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn test_scheduler_config() {
        let config = SchedulerPolicy::default();
        assert!(config.base_buffer_size < config.max_buffer_size);
    }
    #[test]
    fn test_scheduler_policy_to_core_config() {
        let policy = SchedulerPolicy::default();
        let core = policy.to_core_config();
        assert_eq!(core.base_buffer_size, policy.base_buffer_size);
        assert_eq!(core.max_buffer_size, policy.max_buffer_size);
        assert_eq!(core.high_priority_size_threshold, policy.high_priority_threshold);
        assert_eq!(core.low_priority_size_threshold, policy.low_priority_threshold);
    }
    #[test]
    fn test_scheduler_default_thresholds_remain_stable() {
        let policy = SchedulerPolicy::default();
        assert_eq!(policy.base_buffer_size, 64 * 1024);
        assert_eq!(policy.max_buffer_size, 4 * 1024 * 1024);
        assert_eq!(policy.high_priority_threshold, 1024 * 1024);
        assert_eq!(policy.low_priority_threshold, 10 * 1024 * 1024);
    }
    #[test]
    fn test_scheduler_priority_boundaries_remain_stable() {
        let policy = SchedulerPolicy::default();
        let manager = SchedulerManager::from_policy(policy);
        assert_eq!(manager.get_priority(1_048_575), IoPriority::High);
        assert_eq!(manager.get_priority(1_048_576), IoPriority::Normal);
        assert_eq!(manager.get_priority(10_485_760), IoPriority::Normal);
        assert_eq!(manager.get_priority(10_485_761), IoPriority::Low);
    }
    #[test]
    fn test_scheduler_manager() {
        let manager = SchedulerManager::new(1024, 4096, 512, 2048);
        let priority = manager.get_priority(100);
        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);
        let strategy = manager.create_strategy();
        let size = strategy.calculate_buffer_size(1024 * 1024, StorageMedia::Ssd, AccessPattern::Sequential, IoLoadLevel::Low, 1);
        assert!(size > 0);
    }
 }
@@ -1,222 +0,0 @@
 // Copyright 2024 RustFS Team
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //! Timeout management for operations
 use rustfs_io_core::{TimeoutConfig as CoreTimeoutConfig, TimeoutError, calculate_adaptive_timeout};
 use std::time::{Duration, Instant};
 use tokio_util::sync::CancellationToken;
 /// Facade policy for the concurrency-layer timeout manager.
 #[derive(Debug, Clone, Copy)]
 pub struct TimeoutManagerPolicy {
    /// Default timeout duration
    pub default_timeout: Duration,
    /// Maximum timeout duration
    pub max_timeout: Duration,
    /// Minimum timeout floor (prevents dynamic calculation from going too low).
    pub min_timeout: Duration,
    /// Enable dynamic timeout calculation
    pub enable_dynamic: bool,
 }
 impl Default for TimeoutManagerPolicy {
    fn default() -> Self {
        Self {
            default_timeout: Duration::from_secs(30),
            max_timeout: Duration::from_secs(300),
            min_timeout: Duration::from_secs(5),
            enable_dynamic: true,
        }
    }
 }
 impl TimeoutManagerPolicy {
    /// Convert the facade policy into the reusable io-core timeout configuration.
    ///
    /// This keeps the concurrency layer explicitly wired to the shared core
    /// timeout primitives without changing the facade's public behavior.
    pub fn to_core_config(&self) -> CoreTimeoutConfig {
        CoreTimeoutConfig {
            base_timeout: self.default_timeout,
            timeout_per_mb: Duration::ZERO,
            max_timeout: self.max_timeout,
            min_timeout: self.min_timeout,
            get_object_timeout: self.default_timeout,
            put_object_timeout: self.max_timeout,
            list_objects_timeout: self.default_timeout,
            enable_dynamic_timeout: self.enable_dynamic,
        }
    }
 }
 /// Timeout manager
 pub struct TimeoutManager {
    config: TimeoutManagerPolicy,
    core_config: CoreTimeoutConfig,
 }
 impl TimeoutManager {
    /// Create a new timeout manager
    pub fn new(default_timeout: Duration, max_timeout: Duration, enable_dynamic: bool) -> Self {
        let min_timeout = default_timeout.min(max_timeout);
        Self::from_policy(TimeoutManagerPolicy {
            default_timeout,
            max_timeout,
            min_timeout,
            enable_dynamic,
        })
    }
    /// Create a new timeout manager from the facade policy type.
    pub fn from_policy(config: TimeoutManagerPolicy) -> Self {
        let config = TimeoutManagerPolicy {
            // Guard clamp(min, max) from panic when callers provide an
            // out-of-order policy (or very small max_timeout).
            min_timeout: config.min_timeout.min(config.max_timeout),
            ..config
        };
        let core_config = config.to_core_config();
        Self { config, core_config }
    }
    /// Get the configuration
    pub fn config(&self) -> &TimeoutManagerPolicy {
        &self.config
    }
    /// Get the derived io-core timeout configuration.
    pub fn core_config(&self) -> &CoreTimeoutConfig {
        &self.core_config
    }
    /// Calculate timeout for a given size
    pub fn calculate_timeout(&self, size: u64, _history: &[Duration]) -> Duration {
        if !self.config.enable_dynamic {
            return self.config.default_timeout;
        }
        calculate_adaptive_timeout(self.core_config.base_timeout, None, 0, size)
            .clamp(self.core_config.min_timeout, self.core_config.max_timeout)
    }
    /// Wrap an operation with timeout control
    pub async fn wrap_operation<F, T, E>(&self, operation: F, timeout: Option<Duration>) -> Result<T, TimeoutError>
    where
        F: std::future::Future<Output = Result<T, E>>,
        E: Into<TimeoutError>,
    {
        let timeout = timeout.unwrap_or(self.config.default_timeout);
        match tokio::time::timeout(timeout, operation).await {
            Ok(Ok(result)) => Ok(result),
            Ok(Err(e)) => Err(e.into()),
            Err(_) => Err(TimeoutError::TimedOut(timeout)),
        }
    }
    /// Create a timeout guard for manual timeout control
    pub fn create_guard(&self, timeout: Option<Duration>) -> TimeoutGuard {
        TimeoutGuard::new(timeout.unwrap_or(self.core_config.base_timeout))
    }
 }
 /// Timeout guard for manual timeout control
 pub struct TimeoutGuard {
    timeout: Duration,
    start: Instant,
    cancel_token: CancellationToken,
 }
 impl TimeoutGuard {
    fn new(timeout: Duration) -> Self {
        Self {
            timeout,
            start: Instant::now(),
            cancel_token: CancellationToken::new(),
        }
    }
    /// Check if timeout has elapsed
    pub fn is_timed_out(&self) -> bool {
        self.start.elapsed() > self.timeout
    }
    /// Get remaining time
    pub fn remaining(&self) -> Duration {
        self.timeout.saturating_sub(self.start.elapsed())
    }
    /// Get the cancellation token
    pub fn cancel_token(&self) -> CancellationToken {
        self.cancel_token.clone()
    }
    /// Cancel the operation
    pub fn cancel(&self) {
        self.cancel_token.cancel();
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn test_timeout_config() {
        let config = TimeoutManagerPolicy::default();
        assert!(config.default_timeout < config.max_timeout);
    }
    #[test]
    fn test_timeout_policy_to_core_config() {
        let policy = TimeoutManagerPolicy::default();
        let core = policy.to_core_config();
        assert_eq!(core.base_timeout, policy.default_timeout);
        assert_eq!(core.max_timeout, policy.max_timeout);
        assert_eq!(core.min_timeout, policy.min_timeout);
        assert_eq!(core.get_object_timeout, policy.default_timeout);
        assert!(core.enable_dynamic_timeout);
    }
    #[test]
    fn test_timeout_manager_new_sanitizes_min_timeout_with_small_max_timeout() {
        let manager = TimeoutManager::new(Duration::from_secs(1), Duration::from_secs(1), true);
        let timeout = manager.calculate_timeout(1024, &[]);
        assert_eq!(timeout, Duration::from_secs(1));
    }
    #[test]
    fn test_timeout_manager_from_policy_sanitizes_min_timeout() {
        let manager = TimeoutManager::from_policy(TimeoutManagerPolicy {
            default_timeout: Duration::from_secs(30),
            max_timeout: Duration::from_secs(1),
            min_timeout: Duration::from_secs(5),
            enable_dynamic: true,
        });
        assert_eq!(manager.config().min_timeout, Duration::from_secs(1));
        assert_eq!(manager.core_config().min_timeout, Duration::from_secs(1));
    }
    #[tokio::test]
    async fn test_wrap_operation_success() {
        let manager = TimeoutManager::new(Duration::from_secs(5), Duration::from_secs(10), true);
        let result = manager.wrap_operation(async { Ok::<_, TimeoutError>(42) }, None).await;
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), 42);
    }
 }
@@ -9,9 +9,8 @@ runtime-builder ownership. It does not define new behavior.
 | Surface | Current owner | Current responsibility | Migration boundary |
 |---|---|---|---|
 | `ConcurrencyManager` | `rustfs/src/storage/concurrency/manager.rs` | Owns the RustFS S3 read-path disk-read semaphore, I/O metrics, priority queue, storage media detection, access-pattern detection, and buffer strategy. | Keep request admission and I/O metrics behavior stable until a controller can consume the same state explicitly. |
-| `SchedulerManager` | `crates/concurrency/src/scheduler.rs` | Provides a reusable facade over `rustfs-io-core::IoScheduler` and derives buffer/priority decisions from `SchedulerPolicy`. | Treat it as reusable library surface; do not assume the RustFS S3 read path has already moved to this facade. |
+| I/O scheduler core | `crates/io-core/src/scheduler.rs` | Owns the reusable buffer-size and priority algorithms consumed by the RustFS S3 read path (`rustfs/src/storage/concurrency/io_schedule.rs`). The former `SchedulerManager` facade in `rustfs-concurrency` was removed as zero-caller dead code (backlog#1025). | Treat `rustfs-io-core` as the reusable algorithm surface; the RustFS S3 read path owns its own scheduling wiring. |
-| `BackpressureManager` | `crates/concurrency/src/backpressure.rs` | Provides a reusable duplex-pipe backpressure facade over `rustfs-io-core::BackpressureMonitor`. | Keep pipe sizing and watermark policy separate from object-read disk semaphore admission. |
+| RustFS backpressure monitor | `rustfs/src/storage/backpressure.rs` | Tracks object-pipe watermark state used by RustFS storage backpressure tests and helpers, using the shared `PipeBackpressurePolicy` from `crates/concurrency/src/backpressure.rs`. The former `BackpressureManager`/`BackpressurePipe` facade in `rustfs-concurrency` was removed as zero-caller dead code (backlog#1025). | Preserve current state labels and watermark semantics; keep pipe sizing and watermark policy separate from object-read disk semaphore admission. |
 | RustFS backpressure monitor | `rustfs/src/storage/backpressure.rs` | Tracks object-pipe watermark state used by RustFS storage backpressure tests and helpers. | Preserve current state labels and watermark semantics when consolidating with reusable facades. |
 | `Workers` | `crates/concurrency/src/workers.rs` | Provides cooperative worker-slot admission with `take`, `give`, and `wait`; current background workflows use it for bounded set workers. | Preserve blocking/wakeup semantics and over-release clamping. |
 | Scanner cycle budget | `crates/scanner/src/scanner_budget.rs` | Cancels a child token when runtime, object-count, or directory-count budget is reached. | Preserve partial-cycle reason mapping and checkpoint accounting. |
 | Heal admission | `crates/heal/src/heal/manager.rs`, `crates/heal/src/heal/channel.rs`, `rustfs_common::heal_channel` | Owns priority queue admission, duplicate merge/drop/full results, active-task tracking, retry admission, and channel responses. | Preserve low-priority scanner behavior and high-priority escalation gates. |
@@ -5,18 +5,18 @@ preservation and runtime workload-class contract slice.
 ## Preservation Coverage
-The `rustfs-concurrency` tests pin the current reusable scheduler and
+The `rustfs-concurrency` tests pin the current reusable admission-facing
-admission-facing behavior before later snapshot extraction:
+behavior before later snapshot extraction:
 - Worker slot over-release remains clamped by the configured worker limit.
 - Scheduler default buffer and priority thresholds remain unchanged.
 - Scheduler priority boundaries remain high below the high threshold, normal at
  both thresholds, and low above the low threshold.
 - Backpressure pipe metadata reads preserve buffer capacity and state without
  mutating the manager state.
 - `GetObjectQueueSnapshot` preserves saturated, over-available, and zero-total
  permit semantics.
 The former reusable scheduler and backpressure-pipe facades (and their
 preservation tests) were removed as zero-caller dead code in backlog#1025;
 scheduler buffer/priority behavior is now pinned by `rustfs-io-core` and
 `rustfs/src/storage/concurrency` tests.
 ## Workload Class Contract
 `WorkloadClass` defines the required future admission categories:
@@ -54,8 +54,6 @@ checked_files=(
  "crates/scanner/src/scanner_io.rs"
  "crates/scanner/src/scanner_folder.rs"
  "crates/concurrency/src/workers.rs"
  "crates/concurrency/src/manager.rs"
  "crates/concurrency/src/lock.rs"
  "crates/concurrency/src/deadlock.rs"
  "crates/trusted-proxies/src/global.rs"
  "crates/trusted-proxies/src/config/loader.rs"