diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index da869bf5b..eb3338bf2 100644 --- a/ARCHITECTURE.md +++ b/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 diff --git a/crates/concurrency/src/workload.rs b/crates/concurrency/src/workload.rs index f9aee7109..064e2ede0 100644 --- a/crates/concurrency/src/workload.rs +++ b/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) + ); + } } diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index b31834fb5..2c91c049c 100644 --- a/docs/architecture/migration-progress.md +++ b/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. diff --git a/docs/architecture/workload-admission-contracts.md b/docs/architecture/workload-admission-contracts.md index d8e948618..34f260022 100644 --- a/docs/architecture/workload-admission-contracts.md +++ b/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. diff --git a/rustfs/src/workload_admission.rs b/rustfs/src/workload_admission.rs index e3fc4ba9f..0a7eaf907 100644 --- a/rustfs/src/workload_admission.rs +++ b/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) + ); + } } diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index db219301f..e6c1b825e 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/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