refactor: consume storage concurrency policies (#3684)

ARCHITECTURE.md

@@ -249,9 +249,9 @@ Depth 8 — TOP:
   not regain upward dependencies.
 
 - **Three-layer BackpressureConfig/DeadlockConfig duplication** across io-core,
-  concurrency, and rustfs/storage. Storage policies now expose explicit
+  concurrency, and rustfs/storage. Storage policies now expose and consume
-  projections into the concurrency/io-core policy shapes; later work should use
+  explicit projections into the concurrency/io-core policy shapes; later work
-  those bridges before deleting compatibility wrappers.
+  should use those bridges before deleting compatibility wrappers.
 
 ### Medium
 

docs/architecture/migration-progress.md

@@ -5,15 +5,15 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
 ## Current Context
 
 - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
-- Branch: `overtrue/arch-storage-concurrency-policy-bridges`
+- Branch: `overtrue/arch-storage-concurrency-policy-consumers`
-- Baseline: completed `C-004/C-005/C-006`.
+- Baseline: completed `C-004/C-005/C-006/C-011`.
-- Stacked on: ECStore cluster status snapshots.
+- Stacked on: storage concurrency policy bridges.
 - PR type for this branch: `api-extraction`
 - Runtime behavior changes: none.
-- Rust code changes: add storage-to-concurrency policy bridge projections for
+- Rust code changes: route storage object backpressure and request hang/deadlock
-  object backpressure and request hang/deadlock detection.
+  runtime config consumption through the shared concurrency facade policies.
 - CI/script changes: none.
-- Docs changes: record the C-011 policy bridge precondition for later
+- Docs changes: record the C-012 policy consumer precondition for later
   controller status work.
 
 ## Phase 0 Tasks
@@ -82,6 +82,19 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
   - Verification: storage backpressure/deadlock policy tests, compile coverage,
     formatting, diff hygiene, risk scan, architecture guard, pre-commit quality
     gate, and three-expert review.
+- [x] `C-012-POLICY` Consume storage concurrency policy bridges.
+  - Completed slice: route object backpressure threshold derivation and request
+    hang/deadlock runtime policy reads through the shared `rustfs-concurrency`
+    facade policies.
+  - Acceptance: storage keeps env/default ownership and local state machines,
+    while threshold and hang-policy consumption is anchored on the shared
+    concurrency policy shapes.
+  - Must preserve: no worker start/stop, no object pipe state-machine change, no
+    deadlock detector lifecycle change, no metrics label change, and no S3 I/O
+    behavior change.
+  - Verification: storage backpressure/deadlock consumer tests, compile
+    coverage, formatting, diff hygiene, risk scan, architecture guard,
+    pre-commit quality gate, and three-expert review.
 - [x] `G-012` Inventory placement and repair invariants.
   - Acceptance:
     [`placement-repair-invariants.md`](placement-repair-invariants.md) records
@@ -3115,14 +3128,24 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
 
 | Expert | Status | Notes |
 |---|---|---|
-| Quality/architecture | passed | C-011 exposes storage backpressure and request-hang policies through existing `rustfs-concurrency` and `rustfs-io-core` policy shapes without adding ownership cycles. |
+| Quality/architecture | passed | C-012 routes storage backpressure and request-hang runtime consumers through existing `rustfs-concurrency` policy shapes without adding ownership cycles. |
-| Migration preservation | passed | Bridge methods preserve storage env/default ownership, object pipe state, deadlock detector lifecycle, metrics labels, and S3 I/O behavior. |
+| Migration preservation | passed | Consumer changes preserve storage env/default ownership, object pipe state, deadlock detector lifecycle, metrics labels, and S3 I/O behavior. |
-| Testing/verification | passed | Focused storage policy tests, RustFS lib compile coverage, migration guard, formatting, diff hygiene, added-line risk scan, full pre-commit, and three-expert review passed. |
+| Testing/verification | passed | Focused storage consumer tests, RustFS lib compile coverage, migration guard, formatting, diff hygiene, added-line risk scan, full pre-commit, and three-expert review passed. |
 
 ## Verification Notes
 
 Passed before push:
 
+- Issue #660 C-012 current slice:
+  - `cargo test -p rustfs --lib storage::backpressure::tests:: -- --nocapture`: passed.
+  - `cargo test -p rustfs --lib storage::deadlock_detector::tests:: -- --nocapture`: passed.
+  - `cargo check -p rustfs --lib`: passed.
+  - `cargo fmt --all --check`: passed.
+  - `git diff --check`: passed.
+  - `./scripts/check_architecture_migration_rules.sh`: passed.
+  - Rust added-line risk scan on changed storage Rust files: passed.
+  - `make pre-commit`: passed.
+
 - Issue #660 C-011 current slice:
   - `cargo test -p rustfs --lib storage::deadlock_detector::tests::test_request_hang_policy_projects_to_concurrency_and_core_config -- --nocapture`: passed.
   - `cargo test -p rustfs --lib storage::backpressure::tests::test_backpressure_policy_projects_to_concurrency_and_core_config -- --nocapture`: passed.

rustfs/src/storage/backpressure.rs

@@ -229,9 +229,10 @@ impl BackpressurePipe {
 
     /// Create a new backpressure-aware pipe with custom configuration.
     pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self {
-        let (reader, writer) = duplex(config.buffer_size);
+        let policy = config.to_concurrency_policy();
-        let high_watermark_bytes = config.high_watermark_bytes();
+        let (reader, writer) = duplex(policy.buffer_size);
-        let low_watermark_bytes = config.low_watermark_bytes();
+        let high_watermark_bytes = policy.high_watermark_bytes();
+        let low_watermark_bytes = policy.low_watermark_bytes();
 
         debug!(
             buffer_size = config.buffer_size,
@@ -396,8 +397,9 @@ impl BackpressureMonitor {
 
     /// Create a new monitor with custom configuration.
     pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self {
-        let high_watermark_bytes = config.high_watermark_bytes();
+        let policy = config.to_concurrency_policy();
-        let low_watermark_bytes = config.low_watermark_bytes();
+        let high_watermark_bytes = policy.high_watermark_bytes();
+        let low_watermark_bytes = policy.low_watermark_bytes();
         Self {
             config,
             buffer_usage: AtomicUsize::new(0),
@@ -521,6 +523,36 @@ mod tests {
         assert!(core.enabled);
     }
 
+    #[test]
+    fn test_backpressure_pipe_consumes_concurrency_policy_thresholds() {
+        let config = ObjectPipeBackpressurePolicy {
+            buffer_size: 2000,
+            high_watermark: 75,
+            low_watermark: 40,
+        };
+        let concurrency = config.to_concurrency_policy();
+        let pipe = BackpressurePipe::with_config(config);
+
+        assert_eq!(pipe.capacity(), concurrency.buffer_size);
+        assert_eq!(pipe.high_watermark_bytes, concurrency.high_watermark_bytes());
+        assert_eq!(pipe.low_watermark_bytes, concurrency.low_watermark_bytes());
+    }
+
+    #[test]
+    fn test_backpressure_monitor_consumes_concurrency_policy_thresholds() {
+        let config = ObjectPipeBackpressurePolicy {
+            buffer_size: 2000,
+            high_watermark: 75,
+            low_watermark: 40,
+        };
+        let concurrency = config.to_concurrency_policy();
+        let monitor = BackpressureMonitor::with_config(config);
+
+        assert_eq!(monitor.meta().buffer_capacity, concurrency.buffer_size);
+        assert_eq!(monitor.high_watermark_bytes, concurrency.high_watermark_bytes());
+        assert_eq!(monitor.low_watermark_bytes, concurrency.low_watermark_bytes());
+    }
+
     #[test]
     fn test_backpressure_state_display() {
         assert_eq!(format!("{}", BackpressureState::Normal), "normal");

rustfs/src/storage/deadlock_detector.rs

@@ -264,6 +264,8 @@ pub struct ResourceUsage {
 pub struct DeadlockDetector {
     /// Configuration.
     config: RequestHangDetectionPolicy,
+    /// Shared concurrency facade policy.
+    policy: DeadlockMonitorPolicy,
     /// Active request trackers.
     requests: Arc<RwLock<HashMap<RequestId, RequestResourceTracker>>>,
     /// Detection task handle.
@@ -280,9 +282,11 @@ impl DeadlockDetector {
     /// Create a new deadlock detector.
     pub fn new(config: RequestHangDetectionPolicy) -> Self {
         let (shutdown_tx, _) = broadcast::channel(1);
+        let policy = config.to_concurrency_policy();
 
         Self {
             config,
+            policy,
             requests: Arc::new(RwLock::new(HashMap::new())),
             detector_task: Arc::new(Mutex::new(None)),
             shutdown_tx,
@@ -293,12 +297,12 @@ impl DeadlockDetector {
 
     /// Check if detection is enabled.
     pub fn is_enabled(&self) -> bool {
-        self.config.enabled
+        self.policy.enabled
     }
 
     /// Start the detection task.
     pub fn start(&self) {
-        if !self.config.enabled {
+        if !self.policy.enabled {
             debug!("Deadlock detection is disabled");
             return;
         }
@@ -309,22 +313,22 @@ impl DeadlockDetector {
         }
 
         let requests = self.requests.clone();
-        let config = self.config.clone();
+        let policy = self.policy;
         let deadlocks_detected = self.deadlocks_detected.clone();
         let mut shutdown_rx = self.shutdown_tx.subscribe();
         let running = self.running.clone();
 
         let handle = tokio::spawn(async move {
             debug!(
-                check_interval_secs = config.check_interval.as_secs(),
+                check_interval_secs = policy.check_interval.as_secs(),
-                hang_threshold_secs = config.hang_threshold.as_secs(),
+                hang_threshold_secs = policy.hang_threshold.as_secs(),
                 "Deadlock detector started"
             );
 
             loop {
                 tokio::select! {
-                    _ = tokio::time::sleep(config.check_interval) => {
+                    _ = tokio::time::sleep(policy.check_interval) => {
-                        Self::detect_cycle(&requests, &config, &deadlocks_detected);
+                        Self::detect_cycle(&requests, &policy, &deadlocks_detected);
                     }
                     _ = shutdown_rx.recv() => {
                         debug!("Deadlock detector shutting down");
@@ -351,7 +355,7 @@ impl DeadlockDetector {
 
     /// Register a new request for tracking.
     pub fn register_request(&self, request_id: impl Into<String>, description: impl Into<String>) {
-        if !self.config.enabled {
+        if !self.policy.enabled {
             return;
         }
 
@@ -365,7 +369,7 @@ impl DeadlockDetector {
 
     /// Unregister a request (it completed or was cancelled).
     pub fn unregister_request(&self, request_id: &str) {
-        if !self.config.enabled {
+        if !self.policy.enabled {
             return;
         }
 
@@ -376,7 +380,7 @@ impl DeadlockDetector {
 
     /// Record a lock acquisition.
     pub fn record_lock_acquire(&self, request_id: &str, lock: LockInfo) {
-        if !self.config.enabled {
+        if !self.policy.enabled {
             return;
         }
 
@@ -387,7 +391,7 @@ impl DeadlockDetector {
 
     /// Record a lock release.
     pub fn record_lock_release(&self, request_id: &str, lock_id: &LockId) {
-        if !self.config.enabled {
+        if !self.policy.enabled {
             return;
         }
 
@@ -398,7 +402,7 @@ impl DeadlockDetector {
 
     /// Record that a request is waiting for a lock.
     pub fn record_lock_wait(&self, request_id: &str, lock: LockInfo) {
-        if !self.config.enabled {
+        if !self.policy.enabled {
             return;
         }
 
@@ -409,7 +413,7 @@ impl DeadlockDetector {
 
     /// Clear the waiting lock (acquired or gave up).
     pub fn clear_lock_wait(&self, request_id: &str) {
-        if !self.config.enabled {
+        if !self.policy.enabled {
             return;
         }
 
@@ -420,7 +424,7 @@ impl DeadlockDetector {
 
     /// Record resource usage.
     pub fn record_resource(&self, request_id: &str, resource_type: ResourceType, amount: usize) {
-        if !self.config.enabled {
+        if !self.policy.enabled {
             return;
         }
 
@@ -442,13 +446,13 @@ impl DeadlockDetector {
     /// Detect deadlock cycles in the lock wait graph.
     fn detect_cycle(
         requests: &Arc<RwLock<HashMap<RequestId, RequestResourceTracker>>>,
-        config: &RequestHangDetectionPolicy,
+        policy: &DeadlockMonitorPolicy,
         deadlocks_detected: &Arc<AtomicU64>,
     ) {
         let requests_guard = requests.read().unwrap();
 
         // Find hung requests
-        let hung_requests: Vec<_> = requests_guard.values().filter(|r| r.is_hung(config.hang_threshold)).collect();
+        let hung_requests: Vec<_> = requests_guard.values().filter(|r| r.is_hung(policy.hang_threshold)).collect();
 
         if hung_requests.is_empty() {
             return;
@@ -638,6 +642,23 @@ mod tests {
         assert!(config.capture_backtrace);
     }
 
+    #[test]
+    fn test_deadlock_detector_consumes_concurrency_policy() {
+        let config = RequestHangDetectionPolicy {
+            enabled: true,
+            check_interval: Duration::from_secs(7),
+            hang_threshold: Duration::from_secs(11),
+            capture_backtrace: true,
+        };
+        let concurrency = config.to_concurrency_policy();
+        let detector = DeadlockDetector::new(config);
+
+        assert!(detector.is_enabled());
+        assert_eq!(detector.policy.check_interval, concurrency.check_interval);
+        assert_eq!(detector.policy.hang_threshold, concurrency.hang_threshold);
+        assert!(detector.config.capture_backtrace);
+    }
+
     #[test]
     fn test_request_resource_tracker() {
         let tracker = RequestResourceTracker::new("req-1".to_string(), "GetObject bucket/key");