refactor: compose workload admission providers (#3685)

ARCHITECTURE.md

@@ -250,7 +250,8 @@ Depth 8 — TOP:
 
 - **Three-layer BackpressureConfig/DeadlockConfig duplication** across io-core,
   concurrency, and rustfs/storage. Storage policies now expose and consume
-  explicit projections into the concurrency/io-core policy shapes; later work
+  explicit projections into the concurrency/io-core policy shapes, and workload
+  admission snapshots are composed through provider registries; later work
   should use those bridges before deleting compatibility wrappers.
 
 ### Medium

crates/concurrency/src/workload.rs

@@ -150,6 +150,20 @@ impl WorkloadAdmissionRegistrySnapshot {
     pub fn get(&self, class: WorkloadClass) -> Option<&WorkloadAdmissionSnapshot> {
         self.entries.iter().find(|entry| entry.class == class)
     }
+
+    /// Return a registry where matching entries from `overlay` replace this
+    /// registry's entries, and new classes are appended in overlay order.
+    pub fn overlay(mut self, overlay: Self) -> Self {
+        for entry in overlay.entries {
+            if let Some(existing) = self.entries.iter_mut().find(|existing| existing.class == entry.class) {
+                *existing = entry;
+            } else {
+                self.entries.push(entry);
+            }
+        }
+
+        self
+    }
 }
 
 /// Provider boundary for future read-only workload admission snapshots.
@@ -206,4 +220,33 @@ mod tests {
             Some(8)
         );
     }
+
+    #[test]
+    fn admission_registry_overlay_replaces_existing_classes_and_appends_new_ones() {
+        let base = WorkloadAdmissionRegistrySnapshot::new(vec![
+            WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, AdmissionState::Open),
+            WorkloadAdmissionSnapshot::new(WorkloadClass::Scanner, AdmissionState::Unknown),
+        ]);
+        let overlay = WorkloadAdmissionRegistrySnapshot::new(vec![
+            WorkloadAdmissionSnapshot::new(WorkloadClass::Scanner, AdmissionState::Open).with_counts(Some(2), None, None),
+            WorkloadAdmissionSnapshot::new(WorkloadClass::Repair, AdmissionState::Open),
+        ]);
+
+        let registry = base.overlay(overlay);
+
+        assert_eq!(registry.entries().len(), 3);
+        assert_eq!(
+            registry.get(WorkloadClass::ForegroundRead).map(|snapshot| snapshot.state),
+            Some(AdmissionState::Open)
+        );
+        assert_eq!(
+            registry.get(WorkloadClass::Scanner).map(|snapshot| snapshot.state),
+            Some(AdmissionState::Open)
+        );
+        assert_eq!(registry.get(WorkloadClass::Scanner).and_then(|snapshot| snapshot.active), Some(2));
+        assert_eq!(
+            registry.get(WorkloadClass::Repair).map(|snapshot| snapshot.state),
+            Some(AdmissionState::Open)
+        );
+    }
 }

docs/architecture/migration-progress.md

@@ -5,16 +5,17 @@ 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-consumers`
-- Baseline: completed `C-004/C-005/C-006/C-011`.
-- Stacked on: storage concurrency policy bridges.
+- Branch: `overtrue/arch-workload-admission-provider-compose`
+- Baseline: completed `C-011/C-012/API-055/API-059`.
+- Stacked on: storage concurrency policy consumers.
 - PR type for this branch: `api-extraction`
 - Runtime behavior changes: none.
 - Rust code changes: route storage object backpressure and request hang/deadlock
-  runtime config consumption through the shared concurrency facade policies.
+  runtime config consumption through shared concurrency facade policies, then
+  compose RustFS workload admission through provider registries.
 - CI/script changes: none.
-- Docs changes: record the C-012 policy consumer precondition for later
-  controller status work.
+- Docs changes: record the C-013 admission provider composition precondition
+  for later controller status work.
 
 ## Phase 0 Tasks
 
@@ -95,6 +96,21 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
   - Verification: storage backpressure/deadlock consumer tests, compile
     coverage, formatting, diff hygiene, risk scan, architecture guard,
     pre-commit quality gate, and three-expert review.
+- [x] `C-013-ADMISSION` Compose workload admission providers.
+  - Completed slice: add workload admission registry overlay support, compose
+    the RustFS workload admission provider from the storage concurrency
+    provider plus RustFS runtime owner snapshots, and guard the composition
+    boundary.
+  - Acceptance: foreground-read admission remains owned by the storage
+    concurrency provider, RustFS runtime owner snapshots overlay metadata,
+    scanner, repair, replication, and foreground-write status, and later
+    controller/status work can consume one provider-composed registry.
+  - Must preserve: disk-read semaphore acquisition, scanner activity counter,
+    heal task/queue counters, replication worker/queue stats, metadata runtime
+    initialization checks, object write paths, and queue behavior.
+  - Verification: workload contract tests, RustFS workload admission 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
@@ -3128,14 +3144,27 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
 
 | Expert | Status | Notes |
 |---|---|---|
-| 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 | 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 consumer tests, RustFS lib compile coverage, migration guard, formatting, diff hygiene, added-line risk scan, full pre-commit, and three-expert review passed. |
+| Quality/architecture | passed | C-013 composes workload admission providers through the shared registry contract without adding ownership cycles or ECStore dependencies. |
+| Migration preservation | passed | Provider composition preserves storage foreground-read ownership, scanner activity, heal counters, replication stats, metadata runtime checks, and object/queue behavior. |
+| Testing/verification | passed | Focused workload admission tests, RustFS/concurrency 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-013 current slice:
+  - `cargo test -p rustfs-concurrency workload::tests:: -- --nocapture`:
+    passed.
+  - `cargo test -p rustfs --lib workload_admission::tests:: -- --nocapture`:
+    passed.
+  - `cargo check -p rustfs-concurrency`: 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 Rust files: passed.
+  - `make pre-commit`: passed.
+
 - 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.

docs/architecture/workload-admission-contracts.md

@@ -112,3 +112,19 @@ additional read-only owner mappings:
 This is an observation surface only. Disk-read permit acquisition, scanner
 cycle scheduling, bucket metadata loading, metadata locks, object write paths,
 and queue behavior are unchanged.
+
+## Provider Composition Boundary
+
+`WorkloadAdmissionRegistrySnapshot::overlay` composes provider-owned registry
+snapshots without mutating runtime owners:
+
+- The storage concurrency provider remains the source of truth for
+  `ForegroundRead`.
+- The RustFS runtime owner provider overlays metadata, scanner, repair,
+  replication, and foreground-write status on top of the storage registry.
+- Matching workload classes are replaced by the later provider snapshot; new
+  classes are appended without reordering existing unrelated entries.
+
+This keeps the later controller/status layer consuming a single read-only
+registry while preserving the existing storage, scanner, heal, replication, and
+metadata ownership boundaries.

rustfs/src/workload_admission.rs

@@ -26,6 +26,8 @@ const NOT_EXPOSED_BY_PROVIDER: &str = "not exposed by RustFS workload admission
 const REPLICATION_RUNTIME_NOT_INITIALIZED: &str = "replication runtime not initialized";
 const REPLICATION_QUEUE_STATS_UNAVAILABLE: &str = "replication queue stats unavailable";
 const SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED: &str = "scanner activity idle or not initialized";
+const STORAGE_CONCURRENCY_PROVIDER_MISSING_FOREGROUND_READ: &str =
+    "storage concurrency provider did not expose foreground read admission";
 
 #[derive(Debug, Default, Clone, Copy)]
 pub struct RustFsWorkloadAdmissionSnapshotProvider;
@@ -37,16 +39,25 @@ impl WorkloadAdmissionSnapshotProvider for RustFsWorkloadAdmissionSnapshotProvid
 }
 
 pub fn workload_admission_registry_snapshot() -> WorkloadAdmissionRegistrySnapshot {
+    storage_concurrency_workload_admission_snapshot().overlay(runtime_owner_workload_admission_registry_snapshot())
+}
+
+fn storage_concurrency_workload_admission_snapshot() -> WorkloadAdmissionRegistrySnapshot {
+    let storage_provider: &dyn WorkloadAdmissionSnapshotProvider = get_concurrency_manager();
+    storage_provider.workload_admission_snapshot()
+}
+
+fn runtime_owner_workload_admission_registry_snapshot() -> WorkloadAdmissionRegistrySnapshot {
     let entries = WorkloadClass::REQUIRED
         .iter()
         .copied()
-        .map(|class| match class {
-            WorkloadClass::ForegroundRead => foreground_read_workload_admission_snapshot(),
-            WorkloadClass::Metadata => metadata_workload_admission_snapshot(),
-            WorkloadClass::Scanner => scanner_workload_admission_snapshot(),
-            WorkloadClass::Repair => repair_workload_admission_snapshot(),
-            WorkloadClass::Replication => replication_workload_admission_snapshot(),
-            class => WorkloadAdmissionSnapshot::new(class, AdmissionState::Unknown).with_reason(NOT_EXPOSED_BY_PROVIDER),
+        .filter_map(|class| match class {
+            WorkloadClass::ForegroundRead => None,
+            WorkloadClass::Metadata => Some(metadata_workload_admission_snapshot()),
+            WorkloadClass::Scanner => Some(scanner_workload_admission_snapshot()),
+            WorkloadClass::Repair => Some(repair_workload_admission_snapshot()),
+            WorkloadClass::Replication => Some(replication_workload_admission_snapshot()),
+            class => Some(WorkloadAdmissionSnapshot::new(class, AdmissionState::Unknown).with_reason(NOT_EXPOSED_BY_PROVIDER)),
         })
         .collect();
 
@@ -54,7 +65,13 @@ pub fn workload_admission_registry_snapshot() -> WorkloadAdmissionRegistrySnapsh
 }
 
 pub fn foreground_read_workload_admission_snapshot() -> WorkloadAdmissionSnapshot {
-    get_concurrency_manager().get_object_admission_snapshot()
+    storage_concurrency_workload_admission_snapshot()
+        .get(WorkloadClass::ForegroundRead)
+        .cloned()
+        .unwrap_or_else(|| {
+            WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, AdmissionState::Unknown)
+                .with_reason(STORAGE_CONCURRENCY_PROVIDER_MISSING_FOREGROUND_READ)
+        })
 }
 
 pub fn metadata_workload_admission_snapshot() -> WorkloadAdmissionSnapshot {
@@ -192,9 +209,13 @@ mod tests {
     use super::*;
 
     #[test]
-    fn foreground_read_snapshot_uses_storage_concurrency_admission() {
+    fn foreground_read_snapshot_uses_storage_concurrency_provider_contract() {
         let snapshot = foreground_read_workload_admission_snapshot();
+        let storage_snapshot = storage_concurrency_workload_admission_snapshot()
+            .get(WorkloadClass::ForegroundRead)
+            .cloned();
 
+        assert_eq!(Some(snapshot.clone()), storage_snapshot);
         assert_eq!(snapshot.class, WorkloadClass::ForegroundRead);
         assert_ne!(snapshot.state, AdmissionState::Unknown);
         assert!(snapshot.active.is_some());
@@ -319,4 +340,23 @@ mod tests {
         );
         assert_eq!(registry.get(WorkloadClass::Scanner).and_then(|snapshot| snapshot.active), Some(0));
     }
+
+    #[test]
+    fn provider_overlays_runtime_owner_snapshots_on_storage_registry() {
+        let storage_registry = storage_concurrency_workload_admission_snapshot();
+        let registry = workload_admission_registry_snapshot();
+
+        assert_eq!(
+            registry.get(WorkloadClass::ForegroundRead),
+            storage_registry.get(WorkloadClass::ForegroundRead)
+        );
+        assert_eq!(registry.get(WorkloadClass::Metadata), Some(&metadata_workload_admission_snapshot()));
+        assert_eq!(registry.get(WorkloadClass::Scanner), Some(&scanner_workload_admission_snapshot()));
+        assert_eq!(
+            registry
+                .get(WorkloadClass::ForegroundWrite)
+                .and_then(|snapshot| snapshot.reason.as_deref()),
+            Some(NOT_EXPOSED_BY_PROVIDER)
+        );
+    }
 }

scripts/check_architecture_migration_rules.sh

@@ -89,6 +89,7 @@ STORE_API_MODULE_PATH_HITS_FILE="${TMP_DIR}/store_api_module_path_hits.txt"
 ECSTORE_COMPAT_PASSTHROUGH_EXPECTED_FILE="${TMP_DIR}/ecstore_compat_passthrough_expected.txt"
 ECSTORE_COMPAT_PASSTHROUGH_ACTUAL_FILE="${TMP_DIR}/ecstore_compat_passthrough_actual.txt"
 ECSTORE_COMPAT_PASSTHROUGH_DIFF_FILE="${TMP_DIR}/ecstore_compat_passthrough_diff.txt"
+RUSTFS_WORKLOAD_DIRECT_FOREGROUND_MAPPING_HITS_FILE="${TMP_DIR}/rustfs_workload_direct_foreground_mapping_hits.txt"
 
 awk '
   /^## PR Types$/ {
@@ -370,19 +371,23 @@ require_source_contains \
   "RustFS workload admission snapshot provider implementation"
 require_source_contains \
   "rustfs/src/workload_admission.rs" \
-  "pub fn foreground_read_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \
-  "RustFS foreground-read workload admission snapshot helper"
+  "storage_concurrency_workload_admission_snapshot().overlay(runtime_owner_workload_admission_registry_snapshot())" \
+  "RustFS workload admission provider composition"
 require_source_contains \
   "rustfs/src/workload_admission.rs" \
-  "WorkloadClass::ForegroundRead => foreground_read_workload_admission_snapshot()" \
-  "RustFS foreground-read workload admission class mapping"
+  "let storage_provider: &dyn WorkloadAdmissionSnapshotProvider = get_concurrency_manager();" \
+  "RustFS workload admission storage provider boundary"
+require_source_contains \
+  "rustfs/src/workload_admission.rs" \
+  "pub fn foreground_read_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \
+  "RustFS foreground-read workload admission snapshot helper"
 require_source_contains \
   "rustfs/src/workload_admission.rs" \
   "pub fn metadata_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \
   "RustFS metadata workload admission snapshot helper"
 require_source_contains \
   "rustfs/src/workload_admission.rs" \
-  "WorkloadClass::Metadata => metadata_workload_admission_snapshot()" \
+  "WorkloadClass::Metadata => Some(metadata_workload_admission_snapshot())" \
   "RustFS metadata workload admission class mapping"
 require_source_contains \
   "rustfs/src/workload_admission.rs" \
@@ -390,7 +395,7 @@ require_source_contains \
   "RustFS scanner workload admission snapshot helper"
 require_source_contains \
   "rustfs/src/workload_admission.rs" \
-  "WorkloadClass::Scanner => scanner_workload_admission_snapshot()" \
+  "WorkloadClass::Scanner => Some(scanner_workload_admission_snapshot())" \
   "RustFS scanner workload admission class mapping"
 require_source_contains \
   "rustfs/src/workload_admission.rs" \
@@ -398,7 +403,7 @@ require_source_contains \
   "RustFS repair workload admission snapshot helper"
 require_source_contains \
   "rustfs/src/workload_admission.rs" \
-  "WorkloadClass::Repair => repair_workload_admission_snapshot()" \
+  "WorkloadClass::Repair => Some(repair_workload_admission_snapshot())" \
   "RustFS repair workload admission class mapping"
 require_source_contains \
   "rustfs/src/workload_admission.rs" \
@@ -406,9 +411,19 @@ require_source_contains \
   "RustFS replication workload admission snapshot helper"
 require_source_contains \
   "rustfs/src/workload_admission.rs" \
-  "WorkloadClass::Replication => replication_workload_admission_snapshot()" \
+  "WorkloadClass::Replication => Some(replication_workload_admission_snapshot())" \
   "RustFS replication workload admission class mapping"
 
+(
+  cd "$ROOT_DIR"
+  rg -n --no-heading 'WorkloadClass::ForegroundRead => foreground_read_workload_admission_snapshot\(\)' \
+    rustfs/src/workload_admission.rs || true
+) >"$RUSTFS_WORKLOAD_DIRECT_FOREGROUND_MAPPING_HITS_FILE"
+
+if [[ -s "$RUSTFS_WORKLOAD_DIRECT_FOREGROUND_MAPPING_HITS_FILE" ]]; then
+  report_failure "RustFS workload admission foreground reads must be composed through the storage provider registry: $(paste -sd '; ' "$RUSTFS_WORKLOAD_DIRECT_FOREGROUND_MAPPING_HITS_FILE")"
+fi
+
 (
   cd "$ROOT_DIR"
   rg -n --no-heading '\bStorageAPI\b' crates/ecstore/src rustfs/src || true