diff --git a/crates/concurrency/src/backpressure.rs b/crates/concurrency/src/backpressure.rs index fd0cecbc4..f846b64d4 100644 --- a/crates/concurrency/src/backpressure.rs +++ b/crates/concurrency/src/backpressure.rs @@ -242,4 +242,19 @@ mod tests { assert_eq!(pipe.state(), BackpressureState::Normal); assert_eq!(pipe.meta().buffer_capacity, 1024); } + + #[test] + fn test_backpressure_pipe_meta_is_read_only() { + let manager = BackpressureManager::new(2048, 80, 50); + let pipe = manager.create_pipe(); + let first = pipe.meta(); + let second = pipe.meta(); + + assert_eq!(first.buffer_capacity, 2048); + assert_eq!(first.state, BackpressureState::Normal); + assert_eq!(second.buffer_capacity, first.buffer_capacity); + assert_eq!(second.state, first.state); + assert!(second.age >= first.age); + assert_eq!(manager.state(), BackpressureState::Normal); + } } diff --git a/crates/concurrency/src/lib.rs b/crates/concurrency/src/lib.rs index cefe3ecba..e4bca7a98 100644 --- a/crates/concurrency/src/lib.rs +++ b/crates/concurrency/src/lib.rs @@ -125,6 +125,7 @@ mod backpressure; mod scheduler; pub mod workers; +pub mod workload; // Public module exports with feature gates #[cfg(feature = "timeout")] @@ -149,6 +150,10 @@ pub use config::{ConcurrencyConfig, ConcurrencyFeatures}; // Manager mod manager; pub use manager::{ConcurrencyManager, GetObjectQueueSnapshot}; +pub use workload::{ + AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadAdmissionSnapshotProvider, + WorkloadClass, +}; // Prelude for convenient imports pub mod prelude { @@ -169,5 +174,5 @@ pub mod prelude { #[cfg(feature = "scheduler")] pub use crate::scheduler::{IoStrategy, SchedulerManager, SchedulerPolicy}; - pub use crate::{ConcurrencyConfig, ConcurrencyFeatures, ConcurrencyManager}; + pub use crate::{AdmissionState, ConcurrencyConfig, ConcurrencyFeatures, ConcurrencyManager, WorkloadAdmissionSnapshot}; } diff --git a/crates/concurrency/src/manager.rs b/crates/concurrency/src/manager.rs index 423f6686a..e005d9fc0 100644 --- a/crates/concurrency/src/manager.rs +++ b/crates/concurrency/src/manager.rs @@ -281,6 +281,24 @@ mod tests { assert!(snapshot.is_congested(70.0)); } + #[test] + fn test_queue_snapshot_clamps_over_available_permits() { + let snapshot = GetObjectQueueSnapshot::from_available_permits(64, 96); + assert_eq!(snapshot.permits_in_use, 0); + assert_eq!(snapshot.permits_available(), 64); + assert_eq!(snapshot.utilization_percent(), 0.0); + assert!(!snapshot.is_congested(0.0)); + } + + #[test] + fn test_queue_snapshot_handles_zero_total_permits() { + let snapshot = GetObjectQueueSnapshot::from_available_permits(0, 0); + assert_eq!(snapshot.permits_in_use, 0); + assert_eq!(snapshot.permits_available(), 0); + assert_eq!(snapshot.utilization_percent(), 0.0); + assert!(!snapshot.is_congested(0.0)); + } + #[test] fn test_manager_creation() { let manager = ConcurrencyManager::with_defaults(); diff --git a/crates/concurrency/src/scheduler.rs b/crates/concurrency/src/scheduler.rs index 86c5b3dce..34e989d8a 100644 --- a/crates/concurrency/src/scheduler.rs +++ b/crates/concurrency/src/scheduler.rs @@ -239,6 +239,26 @@ mod tests { assert_eq!(core.low_priority_size_threshold, policy.low_priority_threshold); } + #[test] + fn test_scheduler_default_thresholds_remain_stable() { + let policy = SchedulerPolicy::default(); + assert_eq!(policy.base_buffer_size, 64 * 1024); + assert_eq!(policy.max_buffer_size, 4 * 1024 * 1024); + assert_eq!(policy.high_priority_threshold, 1024 * 1024); + assert_eq!(policy.low_priority_threshold, 10 * 1024 * 1024); + } + + #[test] + fn test_scheduler_priority_boundaries_remain_stable() { + let policy = SchedulerPolicy::default(); + let manager = SchedulerManager::from_policy(policy); + + assert_eq!(manager.get_priority(1_048_575), IoPriority::High); + assert_eq!(manager.get_priority(1_048_576), IoPriority::Normal); + assert_eq!(manager.get_priority(10_485_760), IoPriority::Normal); + assert_eq!(manager.get_priority(10_485_761), IoPriority::Low); + } + #[test] fn test_scheduler_manager() { let manager = SchedulerManager::new(1024, 4096, 512, 2048); diff --git a/crates/concurrency/src/workload.rs b/crates/concurrency/src/workload.rs new file mode 100644 index 000000000..f9aee7109 --- /dev/null +++ b/crates/concurrency/src/workload.rs @@ -0,0 +1,209 @@ +// 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. + +//! Runtime workload admission contract types. + +/// Stable workload classes used by future runtime admission snapshots. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum WorkloadClass { + /// Foreground read requests. + ForegroundRead, + /// Foreground write requests. + ForegroundWrite, + /// Metadata and namespace operations. + Metadata, + /// Background scanner work. + Scanner, + /// Repair and heal work. + Repair, + /// Replication work. + Replication, +} + +impl WorkloadClass { + /// Required workload classes covered by the runtime admission contract. + pub const REQUIRED: [Self; 6] = [ + Self::ForegroundRead, + Self::ForegroundWrite, + Self::Metadata, + Self::Scanner, + Self::Repair, + Self::Replication, + ]; + + /// Return a stable label for logs, metrics, and snapshots. + pub const fn as_str(self) -> &'static str { + match self { + Self::ForegroundRead => "foreground_read", + Self::ForegroundWrite => "foreground_write", + Self::Metadata => "metadata", + Self::Scanner => "scanner", + Self::Repair => "repair", + Self::Replication => "replication", + } + } +} + +impl std::fmt::Display for WorkloadClass { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Read-only admission state for a workload class. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)] +pub enum AdmissionState { + /// Admission is open for this workload class. + Open, + /// Admission is currently throttled. + Throttled, + /// Admission is saturated or queue-limited. + Saturated, + /// Admission is disabled by configuration or capability. + Disabled, + /// Admission state is not currently known. + #[default] + Unknown, +} + +impl AdmissionState { + /// Return whether the state allows admission without throttling. + pub const fn is_open(self) -> bool { + matches!(self, Self::Open) + } +} + +/// Read-only admission snapshot for one workload class. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkloadAdmissionSnapshot { + /// Workload class described by this snapshot. + pub class: WorkloadClass, + /// Current admission state. + pub state: AdmissionState, + /// Active work count when the owner exposes it. + pub active: Option, + /// Queued work count when the owner exposes it. + pub queued: Option, + /// Admission limit when the owner exposes it. + pub limit: Option, + /// Optional state reason for disabled, throttled, saturated, or unknown states. + pub reason: Option, +} + +impl WorkloadAdmissionSnapshot { + /// Create a snapshot for a workload class and admission state. + pub const fn new(class: WorkloadClass, state: AdmissionState) -> Self { + Self { + class, + state, + active: None, + queued: None, + limit: None, + reason: None, + } + } + + /// Add optional active, queued, and limit counters. + pub const fn with_counts(mut self, active: Option, queued: Option, limit: Option) -> Self { + self.active = active; + self.queued = queued; + self.limit = limit; + self + } + + /// Add a human-readable reason. + pub fn with_reason(mut self, reason: impl Into) -> Self { + self.reason = Some(reason.into()); + self + } +} + +/// Read-only admission registry snapshot. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct WorkloadAdmissionRegistrySnapshot { + entries: Vec, +} + +impl WorkloadAdmissionRegistrySnapshot { + /// Create a registry snapshot from per-class entries. + pub fn new(entries: Vec) -> Self { + Self { entries } + } + + /// Borrow all entries in snapshot order. + pub fn entries(&self) -> &[WorkloadAdmissionSnapshot] { + &self.entries + } + + /// Find the admission snapshot for a workload class. + pub fn get(&self, class: WorkloadClass) -> Option<&WorkloadAdmissionSnapshot> { + self.entries.iter().find(|entry| entry.class == class) + } +} + +/// Provider boundary for future read-only workload admission snapshots. +pub trait WorkloadAdmissionSnapshotProvider { + /// Return the current workload admission registry snapshot. + fn workload_admission_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot; +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn workload_contract_covers_required_classes() { + let labels: Vec<_> = WorkloadClass::REQUIRED.iter().map(|class| class.as_str()).collect(); + + assert_eq!( + labels, + vec![ + "foreground_read", + "foreground_write", + "metadata", + "scanner", + "repair", + "replication" + ] + ); + assert_eq!(WorkloadClass::Repair.to_string(), "repair"); + } + + #[test] + fn admission_snapshot_preserves_unknown_and_counted_states() { + let unknown = WorkloadAdmissionSnapshot::new(WorkloadClass::Scanner, AdmissionState::Unknown); + let open = WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, AdmissionState::Open).with_counts( + Some(3), + Some(1), + Some(8), + ); + let registry = WorkloadAdmissionRegistrySnapshot::new(vec![unknown.clone(), open.clone()]); + + assert_eq!(registry.entries(), &[unknown, open]); + assert_eq!( + registry.get(WorkloadClass::Scanner).map(|snapshot| snapshot.state), + Some(AdmissionState::Unknown) + ); + assert!(matches!( + registry.get(WorkloadClass::ForegroundRead).map(|snapshot| snapshot.state), + Some(state) if state.is_open() + )); + assert_eq!( + registry + .get(WorkloadClass::ForegroundRead) + .and_then(|snapshot| snapshot.limit), + Some(8) + ); + } +} diff --git a/docs/architecture/crate-boundaries.md b/docs/architecture/crate-boundaries.md index 9149fbe2e..79eb2e480 100644 --- a/docs/architecture/crate-boundaries.md +++ b/docs/architecture/crate-boundaries.md @@ -104,6 +104,10 @@ Required `rustfs-storage-api` public re-exports: - `pub use object::{VersionMarker, WalkOptions, WalkVersionsSortOrder};` - `pub use topology::{DiskCapabilities, TopologyCapabilities, TopologyDisk, TopologyLabels, TopologyPool, TopologySet, TopologySnapshot, TopologySnapshotProvider};` +Required `rustfs-concurrency` public workload admission contract re-exports: + +- `pub use workload::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadAdmissionSnapshotProvider, WorkloadClass};` + ECStore must keep compile-time coverage for `StorageAdminApi`, `HealOperations`, and the separate `NamespaceLocking` operation group. diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index c0a198df1..308496a8a 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,19 +5,19 @@ 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-observability-topology-contracts` +- Branch: `overtrue/arch-scheduler-workload-contracts` - Baseline: stacked on `origin/overtrue/arch-test-harness-compat-aliases` - after `rustfs/rustfs#3600` - (`ae6a5befcf60f2f2ebce9799ba93649032234273`). -- PR type for this branch: `contract` + after `rustfs/rustfs#3601` + (`dc099f415670d94c0f79a7f9015653a08cdf458f`). +- PR type for this branch: `test-only` plus `contract` - Runtime behavior changes: none. -- Rust code changes: add observability and topology capability DTO/trait - contracts to `rustfs-storage-api`. -- CI/script changes: extend migration re-export guard coverage for the new - contract exports. +- Rust code changes: add scheduler preservation tests and workload admission + contract types to `rustfs-concurrency`. +- CI/script changes: extend migration re-export guard coverage for workload + admission contract exports. - Docs changes: add - [`runtime-capability-contracts.md`](runtime-capability-contracts.md) and - record the combined PR-08/API-013 plus PR-09/API-014 contract slice. + [`workload-admission-contracts.md`](workload-admission-contracts.md) and + record the combined PR-05/TEST-SCH-001 plus PR-07/R-015 slice. ## Phase 0 Tasks @@ -119,6 +119,31 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block plus supported, unsupported, unknown, and disabled capability states; focused storage-api check; migration guard; formatting; diff hygiene; and three-expert review. + +- [x] `PR-05/TEST-SCH-001` Add scheduler preservation tests. + - Completed slice: pin worker over-release clamping, reusable scheduler + default thresholds and priority boundaries, backpressure pipe metadata + reads, and get-object queue snapshot saturation/zero-total semantics. + - Acceptance: current reusable scheduling and admission-facing behavior is + covered before later read-only snapshot extraction. + - Must preserve: scheduler algorithm, queue capacity, threshold defaults, + Tokio runtime settings, request admission, scanner admission, heal + admission, replication admission, and background task admission behavior. + - Verification: focused concurrency tests, focused concurrency check, + migration guard, formatting, diff hygiene, and three-expert review. + +- [x] `PR-07/R-015` Add runtime workload class contract. + - Completed slice: add `WorkloadClass`, `AdmissionState`, + `WorkloadAdmissionSnapshot`, `WorkloadAdmissionRegistrySnapshot`, and + `WorkloadAdmissionSnapshotProvider` to `rustfs-concurrency`. + - Acceptance: foreground read, foreground write, metadata, scanner, repair, + and replication workload classes are representable through read-only + admission registry snapshots without ECStore dependency. + - Must preserve: no SchedulerManager decision logic, Tokio worker defaults, + scanner/heal admission behavior, replication admission behavior, cluster + scheduling, placement, membership, or business call-site migration. + - Verification: workload contract unit tests, focused concurrency check, + migration guard, formatting, diff hygiene, 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 @@ -1704,7 +1729,9 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block 1. `contract`/`consumer-migration`: wire read-only observability and topology snapshots to implementation owners without changing runtime, profiling, placement, or admin route behavior. -2. `pure-move`/`consumer-migration`: continue larger cleanup slices with the +2. `api-extraction`: expose the set-local scheduler snapshot after the + TEST-SCH-001 and R-015 protection/contract slice. +3. `pure-move`/`consumer-migration`: continue larger cleanup slices with the loss-prevention guards active for remaining ECStore compatibility contracts now that broad compatibility passthroughs are fully closed. @@ -1712,14 +1739,24 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | S-015 removes obsolete KMS admin policy action variants after the handler fallback cleanup; API-042/API-043/API-044/API-045/API-046/API-047/API-048/API-049/API-050/API-051/API-052/API-053/API-054 narrow notify, S3 Select, OBS, IAM, Swift, heal, scanner, RustFS runtime, test, fuzz, lifecycle helper, harness, and RustFS runtime compatibility contracts without moving ECStore storage metadata ownership; G-011/G-012/G-013 add docs-only baselines for scheduler, placement/repair, and profiling/NUMA work; Issue #660 PR-08/PR-09 add read-only observability and topology contracts in rustfs-storage-api only. | -| Migration preservation | passed | KMS endpoint URLs, query aliases, request bodies, response contracts, and dedicated `kms:*` authorization behavior are preserved; event builder call sites, ECStore event bridge conversion, restore event data, version IDs, metadata filtering, config read/save semantics, S3 Select store/error/buffer semantics, OBS metrics state reads, IAM config/notification/error semantics, Swift bucket metadata access, heal disk/resume/task behavior, scanner lifecycle/replication/data-usage behavior, RustFS startup/admin/app/storage runtime access, e2e/test/fuzz import behavior, lifecycle expiration/transition helper DTO field contracts, flattened harness and RustFS runtime scalar/secondary alias behavior, unchanged no-op handling, remove-event behavior, scheduler/readiness/placement/profiling runtime behavior, platform gates, missing/unknown capability states, and placement/topology labels are preserved. | -| Testing/verification | passed | Focused compiles/tests, fuzz target compile, guards, formatting, diff hygiene, risk scan, and full `make pre-commit` passed for prior code slices; current Issue #660 PR-08/PR-09 contract slice uses storage-api tests/checks, migration guard, formatting, diff hygiene, and three-expert review. | +| Quality/architecture | passed | S-015 removes obsolete KMS admin policy action variants after the handler fallback cleanup; API-042/API-043/API-044/API-045/API-046/API-047/API-048/API-049/API-050/API-051/API-052/API-053/API-054 narrow notify, S3 Select, OBS, IAM, Swift, heal, scanner, RustFS runtime, test, fuzz, lifecycle helper, harness, and RustFS runtime compatibility contracts without moving ECStore storage metadata ownership; G-011/G-012/G-013 add docs-only baselines for scheduler, placement/repair, and profiling/NUMA work; Issue #660 PR-08/PR-09 add read-only observability and topology contracts in storage-api only; current PR-05/PR-07 slice adds scheduler preservation tests and a small workload admission contract in concurrency only. | +| Migration preservation | passed | KMS endpoint URLs, query aliases, request bodies, response contracts, and dedicated `kms:*` authorization behavior are preserved; event builder call sites, ECStore event bridge conversion, restore event data, version IDs, metadata filtering, config read/save semantics, S3 Select store/error/buffer semantics, OBS metrics state reads, IAM config/notification/error semantics, Swift bucket metadata access, heal disk/resume/task behavior, scanner lifecycle/replication/data-usage behavior, RustFS startup/admin/app/storage runtime access, e2e/test/fuzz import behavior, lifecycle expiration/transition helper DTO field contracts, flattened harness and RustFS runtime scalar/secondary alias behavior, unchanged no-op handling, remove-event behavior, scheduler/readiness/placement/profiling runtime behavior, platform gates, missing/unknown capability states, placement/topology labels, scheduler thresholds, queue snapshot semantics, and admission behavior are preserved. | +| Testing/verification | passed | Focused compiles/tests, fuzz target compile, guards, formatting, diff hygiene, risk scan, and full `make pre-commit` passed for prior code slices; current Issue #660 PR-05/PR-07 slice uses concurrency tests/checks, migration guard, formatting, diff hygiene, and three-expert review. | ## Verification Notes Passed before push: +- Issue #660 PR-05/PR-07 current slice: + - `cargo test -p rustfs-concurrency --no-fail-fast`: passed. + - `cargo check -p rustfs-concurrency`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - `make pre-commit`: passed. + - Three-expert review: passed. + - Issue #660 PR-08/PR-09 current slice: - `cargo test -p rustfs-storage-api`: passed. - `cargo check -p rustfs-storage-api`: passed. diff --git a/docs/architecture/workload-admission-contracts.md b/docs/architecture/workload-admission-contracts.md new file mode 100644 index 000000000..b0d1a0459 --- /dev/null +++ b/docs/architecture/workload-admission-contracts.md @@ -0,0 +1,43 @@ +# Workload Admission Contracts + +This document records the `rustfs/backlog#660` PR-05 and PR-07 scheduler +preservation and runtime workload-class contract slice. + +## Preservation Coverage + +The `rustfs-concurrency` tests pin the current reusable scheduler and +admission-facing behavior before later snapshot extraction: + +- Worker slot over-release remains clamped by the configured worker limit. +- Scheduler default buffer and priority thresholds remain unchanged. +- Scheduler priority boundaries remain high below the high threshold, normal at + both thresholds, and low above the low threshold. +- Backpressure pipe metadata reads preserve buffer capacity and state without + mutating the manager state. +- `GetObjectQueueSnapshot` preserves saturated, over-available, and zero-total + permit semantics. + +## Workload Class Contract + +`WorkloadClass` defines the required future admission categories: + +- Foreground read. +- Foreground write. +- Metadata. +- Scanner. +- Repair. +- Replication. + +`AdmissionState`, `WorkloadAdmissionSnapshot`, and +`WorkloadAdmissionRegistrySnapshot` define read-only status shapes for later +runtime owners. They do not replace the current scheduler, request guard, +scanner, heal, replication, or ECStore placement behavior. + +## Boundary Rules + +- `rustfs-concurrency` owns this reusable contract surface. +- The contract does not depend on `rustfs-ecstore` or RustFS binary runtime + state. +- No scheduler decision logic, queue capacity, Tokio runtime default, scanner + admission, heal admission, replication admission, placement, membership, or + NUMA behavior changes are part of this slice. diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index a7809cbf8..de09b90cc 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -212,6 +212,18 @@ require_source_line \ "crates/storage-api/src/lib.rs" \ "pub use capability::{CapabilitySnapshotError, CapabilityState, CapabilityStatus};" \ "storage-api public capability contract re-export" +require_source_contains \ + "crates/concurrency/src/lib.rs" \ + "pub use workload::{" \ + "concurrency public workload admission contract re-export" +require_source_contains \ + "crates/concurrency/src/lib.rs" \ + "AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadAdmissionSnapshotProvider," \ + "concurrency public workload admission contract status re-exports" +require_source_contains \ + "crates/concurrency/src/lib.rs" \ + "WorkloadClass," \ + "concurrency public workload class re-export" require_source_line \ "crates/storage-api/src/lib.rs" \ "pub use multipart::{CompletePart, ListMultipartsInfo, ListPartsInfo, MultipartInfo, MultipartUploadResult, PartInfo};" \