diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index efbf293ff..c4356781f 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,18 +5,18 @@ 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-runtime-capability-snapshots` -- Baseline: `origin/main` after `rustfs/rustfs#3603` - (`b106b628c164bf6cdca9e71ff5118711f99d4ab0`). +- Branch: `overtrue/arch-workload-admission-owner-snapshots` +- Baseline: `origin/main` after `rustfs/rustfs#3590` + (`8cf3c0bf0100d9d2234f08b48df31fd73b92ed43`). - PR type for this branch: `consumer-migration` - Runtime behavior changes: none. -- Rust code changes: wire read-only observability and endpoint-topology - snapshot providers in the RustFS runtime. -- CI/script changes: extend migration guard coverage for RustFS runtime - capability provider implementations. -- Docs changes: add RustFS provider notes to - [`runtime-capability-contracts.md`](runtime-capability-contracts.md) and - record the API-056/R-016 provider slice. +- Rust code changes: expose read-only repair workload admission snapshots from + existing heal runtime counters. +- CI/script changes: extend migration guard coverage for RustFS workload + admission provider implementations. +- Docs changes: add RustFS heal provider notes to + [`workload-admission-contracts.md`](workload-admission-contracts.md) and + record the API-057/R-017 provider slice. ## Phase 0 Tasks @@ -170,6 +170,19 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block check, migration and layer guards, formatting, diff hygiene, risk scan, and three-expert review. +- [x] `API-057/R-017` Expose heal repair admission snapshot. + - Completed slice: implement a RustFS workload admission snapshot provider + that maps existing heal active-task and queue-length counters to the + `Repair` workload class. + - Acceptance: repair admission state is observable through the + `rustfs-concurrency` workload snapshot contract without changing heal + queueing, scheduling, retry, priority merge/drop, or repair behavior. + - Must preserve: heal request admission, queue capacity, scheduler wakeups, + task retry handling, active-task accounting, and repair execution. + - Verification: focused workload admission tests, focused RustFS library + check, migration and layer guards, formatting, diff hygiene, risk scan, and + three-expert review. + - [x] `TEST-PRTYPE-001` Check PR type enum consistency. - Acceptance: `./scripts/check_architecture_migration_rules.sh` parses the allowed PR types from [`crate-boundaries.md`](crate-boundaries.md) and fails diff --git a/docs/architecture/workload-admission-contracts.md b/docs/architecture/workload-admission-contracts.md index 26a653b89..bacef374d 100644 --- a/docs/architecture/workload-admission-contracts.md +++ b/docs/architecture/workload-admission-contracts.md @@ -59,3 +59,19 @@ The RustFS storage `ConcurrencyManager` now implements This is an observation surface only. Permit acquisition, priority assignment, buffer sizing, storage media detection, request guards, and queue behavior are unchanged. + +## Heal Repair Snapshot Extraction + +The RustFS integration layer now exposes a read-only repair admission snapshot +from the heal runtime counters: + +- `Repair` reports the current heal active task count. +- `queued` reports the current heal queue length. +- `limit` remains `None` because the configured heal queue and concurrency + limits live behind the async heal manager state. +- Other workload classes remain `Unknown` in this provider until their owning + runtime components expose read-only status. + +This is an observation surface only. Heal request admission, queue capacity, +priority merge/drop policy, task scheduling, retry handling, and repair +behavior are unchanged. diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs index fddcd51ba..8486bcd89 100644 --- a/rustfs/src/lib.rs +++ b/rustfs/src/lib.rs @@ -83,6 +83,7 @@ pub(crate) mod table_catalog; pub mod tls; pub mod update; pub mod version; +pub mod workload_admission; // Re-export from rustfs_utils so that config sub-modules can use // `crate::apply_external_env_compat` without breaking. diff --git a/rustfs/src/workload_admission.rs b/rustfs/src/workload_admission.rs new file mode 100644 index 000000000..cca7248e7 --- /dev/null +++ b/rustfs/src/workload_admission.rs @@ -0,0 +1,124 @@ +// 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. + +use rustfs_concurrency::{ + AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadAdmissionSnapshotProvider, + WorkloadClass, +}; + +const HEAL_MANAGER_NOT_INITIALIZED: &str = "heal manager not initialized"; +const NOT_EXPOSED_BY_PROVIDER: &str = "not exposed by RustFS workload admission provider"; + +#[derive(Debug, Default, Clone, Copy)] +pub struct RustFsWorkloadAdmissionSnapshotProvider; + +impl WorkloadAdmissionSnapshotProvider for RustFsWorkloadAdmissionSnapshotProvider { + fn workload_admission_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot { + workload_admission_registry_snapshot() + } +} + +pub fn workload_admission_registry_snapshot() -> WorkloadAdmissionRegistrySnapshot { + let entries = WorkloadClass::REQUIRED + .iter() + .copied() + .map(|class| match class { + WorkloadClass::Repair => repair_workload_admission_snapshot(), + class => WorkloadAdmissionSnapshot::new(class, AdmissionState::Unknown).with_reason(NOT_EXPOSED_BY_PROVIDER), + }) + .collect(); + + WorkloadAdmissionRegistrySnapshot::new(entries) +} + +pub fn repair_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { + repair_workload_admission_snapshot_from_counts( + rustfs_heal::get_heal_manager().is_some(), + rustfs_heal::current_heal_active_tasks(), + rustfs_heal::current_heal_queue_length(), + ) +} + +fn repair_workload_admission_snapshot_from_counts( + manager_initialized: bool, + active: u64, + queued: u64, +) -> WorkloadAdmissionSnapshot { + let state = if manager_initialized || active > 0 || queued > 0 { + AdmissionState::Open + } else { + AdmissionState::Unknown + }; + + let snapshot = WorkloadAdmissionSnapshot::new(WorkloadClass::Repair, state).with_counts( + Some(u64_to_usize_saturated(active)), + Some(u64_to_usize_saturated(queued)), + None, + ); + + if state == AdmissionState::Unknown { + snapshot.with_reason(HEAL_MANAGER_NOT_INITIALIZED) + } else { + snapshot + } +} + +fn u64_to_usize_saturated(value: u64) -> usize { + usize::try_from(value).unwrap_or(usize::MAX) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn repair_snapshot_reports_heal_counters() { + let snapshot = repair_workload_admission_snapshot_from_counts(true, 2, 3); + + assert_eq!(snapshot.class, WorkloadClass::Repair); + assert_eq!(snapshot.state, AdmissionState::Open); + assert_eq!(snapshot.active, Some(2)); + assert_eq!(snapshot.queued, Some(3)); + assert_eq!(snapshot.limit, None); + assert_eq!(snapshot.reason, None); + } + + #[test] + fn repair_snapshot_is_unknown_before_heal_manager_initializes() { + let snapshot = repair_workload_admission_snapshot_from_counts(false, 0, 0); + + assert_eq!(snapshot.class, WorkloadClass::Repair); + assert_eq!(snapshot.state, AdmissionState::Unknown); + assert_eq!(snapshot.active, Some(0)); + assert_eq!(snapshot.queued, Some(0)); + assert_eq!(snapshot.reason.as_deref(), Some(HEAL_MANAGER_NOT_INITIALIZED)); + } + + #[test] + fn provider_covers_required_classes() { + let provider = RustFsWorkloadAdmissionSnapshotProvider; + let registry = provider.workload_admission_snapshot(); + + assert_eq!(registry.entries().len(), WorkloadClass::REQUIRED.len()); + for class in WorkloadClass::REQUIRED { + assert!(registry.get(class).is_some(), "missing workload class {class}"); + } + + assert!(registry.get(WorkloadClass::Repair).is_some()); + assert_eq!( + registry.get(WorkloadClass::ForegroundRead).map(|snapshot| snapshot.state), + Some(AdmissionState::Unknown) + ); + } +} diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index d0794c9be..bcf4ab075 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -284,6 +284,10 @@ require_source_line \ "rustfs/src/lib.rs" \ "pub mod runtime_capabilities;" \ "RustFS runtime capability provider module" +require_source_line \ + "rustfs/src/lib.rs" \ + "pub mod workload_admission;" \ + "RustFS workload admission provider module" require_source_contains \ "rustfs/src/runtime_capabilities.rs" \ "impl ObservabilitySnapshotProvider for RustFsObservabilitySnapshotProvider" \ @@ -296,6 +300,18 @@ require_source_contains \ "rustfs/src/runtime_capabilities.rs" \ "pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> TopologySnapshot" \ "RustFS endpoint topology snapshot helper" +require_source_contains \ + "rustfs/src/workload_admission.rs" \ + "impl WorkloadAdmissionSnapshotProvider for RustFsWorkloadAdmissionSnapshotProvider" \ + "RustFS workload admission snapshot provider implementation" +require_source_contains \ + "rustfs/src/workload_admission.rs" \ + "pub fn repair_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \ + "RustFS repair workload admission snapshot helper" +require_source_contains \ + "rustfs/src/workload_admission.rs" \ + "WorkloadClass::Repair => repair_workload_admission_snapshot()" \ + "RustFS repair workload admission class mapping" ( cd "$ROOT_DIR"