refactor: bridge storage concurrency policies (#3683)

ARCHITECTURE.md

@@ -244,11 +244,14 @@ Depth 8 — TOP:
 
 ### High
 
-- **Dependency inversions.** `utils → config` and `common → filemeta/madmin` break
-  the layering model. These need to be untangled.
+- **Dependency inversions.** Historical `utils → config` and
+  `common → filemeta/madmin` edges must stay removed so leaf/helper crates do
+  not regain upward dependencies.
 
 - **Three-layer BackpressureConfig/DeadlockConfig duplication** across io-core,
-  concurrency, and rustfs/storage. Should be defined once with builder/composition.
+  concurrency, and rustfs/storage. Storage policies now expose explicit
+  projections into the concurrency/io-core policy shapes; later work should use
+  those bridges before deleting compatibility wrappers.
 
 ### Medium
 

docs/architecture/migration-progress.md

@@ -5,15 +5,16 @@ 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-cluster-control-plane-status-snapshots`
-- Baseline: completed `C-001/C-002/C-003`.
-- Stacked on: ECStore cluster control-plane read models.
+- Branch: `overtrue/arch-storage-concurrency-policy-bridges`
+- Baseline: completed `C-004/C-005/C-006`.
+- Stacked on: ECStore cluster status snapshots.
 - PR type for this branch: `api-extraction`
 - Runtime behavior changes: none.
-- Rust code changes: add pool-state, local-node storage, and peer-health status
-  read models to the ECStore ClusterControlPlane snapshot.
+- Rust code changes: add storage-to-concurrency policy bridge projections for
+  object backpressure and request hang/deadlock detection.
 - CI/script changes: none.
-- Docs changes: record the C-004/C-005/C-006 status snapshot slice.
+- Docs changes: record the C-011 policy bridge precondition for later
+  controller status work.
 
 ## Phase 0 Tasks
 
@@ -68,6 +69,19 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
     budget, heal admission, and the Tokio runtime builder.
   - Must preserve: no Rust source changes, no scheduler/controller contract
     changes, and no runtime behavior changes.
+- [x] `C-011-POLICY` Bridge storage concurrency policies.
+  - Completed slice: add explicit projections from storage object backpressure
+    and request hang/deadlock policies into the shared `rustfs-concurrency`
+    facade and reusable `rustfs-io-core` configs.
+  - Acceptance: storage keeps existing env/default ownership and runtime
+    behavior, while later controller/read-only status work can consume the
+    shared facade policy shape instead of duplicating field mapping.
+  - 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 policy 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
@@ -3101,14 +3115,24 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
 
 | Expert | Status | Notes |
 |---|---|---|
-| Quality/architecture | passed | C-004/C-005/C-006 extends ClusterControlPlane with pool-state, local-node storage, and peer-health status read models behind `api::cluster`. |
-| Migration preservation | passed | New status surfaces remain static read-only projections; no placement, readiness, health loop, lock, or object I/O behavior changes. |
-| Testing/verification | passed | ECStore cluster tests, ECStore/RustFS compile coverage, migration guard, formatting, diff hygiene, risk scan, full pre-commit, and three-expert review passed. |
+| 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. |
+| Migration preservation | passed | Bridge methods 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. |
 
 ## Verification Notes
 
 Passed before push:
 
+- 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.
+  - `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-004/C-005/C-006 current slice:
   - `cargo test -p rustfs-ecstore cluster -- --nocapture`: passed, 7 tests.
   - `cargo check -p rustfs-ecstore --all-targets`: passed.

rustfs/src/storage/backpressure.rs

@@ -42,6 +42,8 @@ use tokio::io::{DuplexStream, duplex};
 use tracing::{debug, warn};
 
 use metrics::counter;
+use rustfs_concurrency::PipeBackpressurePolicy;
+use rustfs_io_core::BackpressureConfig as CoreBackpressureConfig;
 
 /// Object-transfer duplex pipe backpressure policy.
 #[derive(Debug, Clone, Copy)]
@@ -98,6 +100,20 @@ impl ObjectPipeBackpressurePolicy {
     pub fn low_watermark_bytes(&self) -> usize {
         (self.buffer_size as u64 * self.low_watermark as u64 / 100) as usize
     }
+
+    /// Project this object-transfer policy into the shared concurrency facade policy.
+    pub fn to_concurrency_policy(&self) -> PipeBackpressurePolicy {
+        PipeBackpressurePolicy {
+            buffer_size: self.buffer_size,
+            high_watermark: self.high_watermark,
+            low_watermark: self.low_watermark,
+        }
+    }
+
+    /// Project this object-transfer policy into the reusable io-core admission config.
+    pub fn to_core_config(&self) -> CoreBackpressureConfig {
+        self.to_concurrency_policy().to_core_config()
+    }
 }
 
 /// Backpressure state.
@@ -487,6 +503,24 @@ mod tests {
         assert_eq!(config.low_watermark_bytes(), 500);
     }
 
+    #[test]
+    fn test_backpressure_policy_projects_to_concurrency_and_core_config() {
+        let config = ObjectPipeBackpressurePolicy {
+            buffer_size: 2000,
+            high_watermark: 75,
+            low_watermark: 40,
+        };
+        let concurrency = config.to_concurrency_policy();
+        let core = config.to_core_config();
+
+        assert_eq!(concurrency.buffer_size, config.buffer_size);
+        assert_eq!(concurrency.high_watermark, config.high_watermark);
+        assert_eq!(concurrency.low_watermark, config.low_watermark);
+        assert_eq!(core.high_water_mark, 0.75);
+        assert_eq!(core.low_water_mark, 0.40);
+        assert!(core.enabled);
+    }
+
     #[test]
     fn test_backpressure_state_display() {
         assert_eq!(format!("{}", BackpressureState::Normal), "normal");

rustfs/src/storage/deadlock_detector.rs

@@ -66,6 +66,7 @@ use tokio::sync::broadcast;
 use tracing::{debug, error, warn};
 
 use metrics::counter;
+use rustfs_concurrency::DeadlockMonitorPolicy;
 use rustfs_io_core::DeadlockDetectorConfig as CoreDeadlockConfig;
 
 /// Request identifier type.
@@ -124,10 +125,15 @@ impl RequestHangDetectionPolicy {
 
     /// Convert the request-level policy into the shared io-core deadlock config.
     pub fn to_core_config(&self) -> CoreDeadlockConfig {
-        CoreDeadlockConfig {
+        self.to_concurrency_policy().to_core_config()
+    }
+
+    /// Convert the request-level policy into the shared concurrency facade policy.
+    pub fn to_concurrency_policy(&self) -> DeadlockMonitorPolicy {
+        DeadlockMonitorPolicy {
             enabled: self.enabled,
-            detection_interval: self.check_interval,
-            max_hold_time: self.hang_threshold,
+            check_interval: self.check_interval,
+            hang_threshold: self.hang_threshold,
         }
     }
 }
@@ -612,6 +618,26 @@ mod tests {
         assert_eq!(config.hang_threshold, Duration::from_secs(10));
     }
 
+    #[test]
+    fn test_request_hang_policy_projects_to_concurrency_and_core_config() {
+        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 core = config.to_core_config();
+
+        assert!(concurrency.enabled);
+        assert_eq!(concurrency.check_interval, config.check_interval);
+        assert_eq!(concurrency.hang_threshold, config.hang_threshold);
+        assert!(core.enabled);
+        assert_eq!(core.detection_interval, config.check_interval);
+        assert_eq!(core.max_hold_time, config.hang_threshold);
+        assert!(config.capture_backtrace);
+    }
+
     #[test]
     fn test_request_resource_tracker() {
         let tracker = RequestResourceTracker::new("req-1".to_string(), "GetObject bucket/key");