diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 308496a8a..e5156ca95 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-scheduler-workload-contracts` -- Baseline: stacked on `origin/overtrue/arch-test-harness-compat-aliases` - after `rustfs/rustfs#3601` - (`dc099f415670d94c0f79a7f9015653a08cdf458f`). -- PR type for this branch: `test-only` plus `contract` +- Branch: `overtrue/arch-set-local-scheduler-snapshots` +- Baseline: `origin/main` after `rustfs/rustfs#3602` + (`56c3cf50ae93d6705cf9f615e49b66787cb99954`). +- PR type for this branch: `api-extraction` - Runtime behavior changes: none. -- 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. +- Rust code changes: expose local foreground-read admission snapshots from + RustFS storage concurrency through the `rustfs-concurrency` workload contract. +- CI/script changes: extend migration guard coverage for the storage + `WorkloadAdmissionSnapshotProvider` implementation. - Docs changes: add + set-local snapshot extraction notes to [`workload-admission-contracts.md`](workload-admission-contracts.md) and - record the combined PR-05/TEST-SCH-001 plus PR-07/R-015 slice. + record the API-055/SCH-001 extraction slice. ## Phase 0 Tasks @@ -144,6 +144,19 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block 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] `API-055/SCH-001` Expose set-local scheduler admission snapshot. + - Completed slice: implement `WorkloadAdmissionSnapshotProvider` for the + RustFS storage `ConcurrencyManager` and expose foreground-read disk-read + permit usage through a local read-only workload registry snapshot. + - Acceptance: local foreground read admission reports active permit usage, + configured limit, and open/saturated/disabled state without ECStore, + admin-route, cluster, or scheduler mutation dependencies. + - Must preserve: disk-read semaphore acquisition, priority assignment, + buffer sizing, storage media detection, request guards, and queue behavior. + - Verification: storage concurrency tests, focused RustFS library 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 @@ -1729,8 +1742,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. `api-extraction`: expose the set-local scheduler snapshot after the - TEST-SCH-001 and R-015 protection/contract slice. +2. `consumer-migration`: connect additional scanner, repair, replication, and + metadata admission owners to the workload registry only after each owner has + dedicated preservation coverage. 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. @@ -1739,14 +1753,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 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. | +| 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; PR-05/PR-07 add scheduler preservation tests and workload contracts; current API-055/SCH-001 adds a local storage concurrency provider 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, disk-read semaphore behavior, 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 API-055/SCH-001 slice uses storage concurrency tests, focused RustFS library check, migration guard, formatting, diff hygiene, and three-expert review. | ## Verification Notes Passed before push: +- Issue #660 API-055/SCH-001 current slice: + - `cargo test -p rustfs --lib storage::concurrency::manager::integration_tests -- --nocapture`: passed. + - `cargo check -p rustfs --lib`: 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-05/PR-07 current slice: - `cargo test -p rustfs-concurrency --no-fail-fast`: passed. - `cargo check -p rustfs-concurrency`: passed. diff --git a/docs/architecture/workload-admission-contracts.md b/docs/architecture/workload-admission-contracts.md index b0d1a0459..26a653b89 100644 --- a/docs/architecture/workload-admission-contracts.md +++ b/docs/architecture/workload-admission-contracts.md @@ -41,3 +41,21 @@ scanner, heal, replication, or ECStore placement behavior. - 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. + +## Set-Local Snapshot Extraction + +The RustFS storage `ConcurrencyManager` now implements +`WorkloadAdmissionSnapshotProvider` for local foreground-read admission: + +- `ForegroundRead` reports local disk-read permit usage through + `GetObjectQueueSnapshot`. +- `active` is the number of disk-read permits currently in use. +- `limit` is the configured maximum concurrent disk reads. +- `queued` remains `None` because the current semaphore does not expose waiter + counts. +- Scanner, repair, replication, foreground write, and metadata entries remain + `Unknown` until their owning runtime components expose read-only status. + +This is an observation surface only. Permit acquisition, priority assignment, +buffer sizing, storage media detection, request guards, and queue behavior are +unchanged. diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index c847cf5c5..eb7e62e80 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -19,7 +19,10 @@ use super::io_schedule::{ get_advanced_buffer_size, }; use super::request_guard::GetObjectGuard; -use rustfs_concurrency::GetObjectQueueSnapshot; +use rustfs_concurrency::{ + AdmissionState, GetObjectQueueSnapshot, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, + WorkloadAdmissionSnapshotProvider, WorkloadClass, +}; use rustfs_config::{KI_B, MI_B}; use rustfs_io_core::BytesPool; use rustfs_io_core::io_profile::{AccessPattern, IoPatternDetector, StorageMedia, detect_storage_media}; @@ -521,10 +524,7 @@ impl ConcurrencyManager { /// /// Returns information about permit usage and waiting requests. pub fn io_queue_status(&self) -> IoQueueStatus { - let snapshot = GetObjectQueueSnapshot::from_available_permits( - self.scheduler_config.max_concurrent_reads, - self.disk_read_semaphore.available_permits(), - ); + let snapshot = self.get_object_queue_snapshot(); IoQueueStatus { total_permits: snapshot.total_permits, @@ -539,6 +539,53 @@ impl ConcurrencyManager { } } + /// Get a read-only snapshot of local GetObject disk-read admission. + pub fn get_object_queue_snapshot(&self) -> GetObjectQueueSnapshot { + GetObjectQueueSnapshot::from_available_permits( + self.scheduler_config.max_concurrent_reads, + self.disk_read_semaphore.available_permits(), + ) + } + + /// Get a read-only workload admission snapshot for foreground reads. + pub fn get_object_admission_snapshot(&self) -> WorkloadAdmissionSnapshot { + let snapshot = self.get_object_queue_snapshot(); + let state = if snapshot.total_permits == 0 { + AdmissionState::Disabled + } else if snapshot.permits_available() == 0 { + AdmissionState::Saturated + } else { + AdmissionState::Open + }; + + let admission = WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, state).with_counts( + Some(snapshot.permits_in_use), + None, + Some(snapshot.total_permits), + ); + + match state { + AdmissionState::Disabled => admission.with_reason("disk read permits disabled"), + AdmissionState::Saturated => admission.with_reason("all disk read permits are in use"), + _ => admission, + } + } + + /// Get a read-only workload admission registry snapshot for local storage concurrency. + pub fn workload_admission_registry_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot { + let entries = WorkloadClass::REQUIRED + .iter() + .copied() + .map(|class| match class { + WorkloadClass::ForegroundRead => self.get_object_admission_snapshot(), + class => WorkloadAdmissionSnapshot::new(class, AdmissionState::Unknown) + .with_reason("not exposed by storage concurrency manager"), + }) + .collect(); + + WorkloadAdmissionRegistrySnapshot::new(entries) + } + /// Acquire a disk read permit with priority awareness. /// /// When priority scheduling is enabled, this method logs the priority @@ -573,6 +620,12 @@ impl ConcurrencyManager { } } +impl WorkloadAdmissionSnapshotProvider for ConcurrencyManager { + fn workload_admission_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot { + self.workload_admission_registry_snapshot() + } +} + impl Default for ConcurrencyManager { fn default() -> Self { Self::new() @@ -617,6 +670,54 @@ mod integration_tests { assert_eq!(status.low_priority_waiting, 0); } + #[tokio::test] + #[serial] + async fn test_concurrency_manager_workload_admission_snapshot_tracks_disk_read_permits() { + let manager = ConcurrencyManager::new(); + let initial = manager.get_object_admission_snapshot(); + + assert_eq!(initial.class, WorkloadClass::ForegroundRead); + assert_eq!(initial.state, AdmissionState::Open); + assert_eq!(initial.active, Some(0)); + assert_eq!(initial.queued, None); + assert_eq!(initial.limit, Some(manager.scheduler_config().max_concurrent_reads)); + + let permit = manager.acquire_disk_read_permit().await; + assert!(permit.is_ok()); + let _permit = match permit { + Ok(permit) => permit, + Err(error) => panic!("disk read permit acquisition failed: {error}"), + }; + let snapshot = manager.get_object_admission_snapshot(); + + assert_eq!(snapshot.active, Some(1)); + assert_eq!(snapshot.limit, Some(manager.scheduler_config().max_concurrent_reads)); + } + + #[tokio::test] + #[serial] + async fn test_concurrency_manager_workload_admission_registry_covers_required_classes() { + let manager = ConcurrencyManager::new(); + let registry = manager.workload_admission_registry_snapshot(); + let provider: &dyn WorkloadAdmissionSnapshotProvider = &manager; + let trait_registry = provider.workload_admission_snapshot(); + + assert_eq!(registry.entries(), trait_registry.entries()); + assert_eq!(registry.entries().len(), WorkloadClass::REQUIRED.len()); + for class in WorkloadClass::REQUIRED { + assert!(registry.get(class).is_some(), "missing workload class {class}"); + } + + 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::Unknown) + ); + } + #[tokio::test] #[serial] async fn test_concurrency_manager_disk_read_permit() { diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index de09b90cc..bfcacb7ea 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -224,6 +224,18 @@ require_source_contains \ "crates/concurrency/src/lib.rs" \ "WorkloadClass," \ "concurrency public workload class re-export" +require_source_contains \ + "rustfs/src/storage/concurrency/manager.rs" \ + "impl WorkloadAdmissionSnapshotProvider for ConcurrencyManager" \ + "storage concurrency workload admission snapshot provider" +require_source_contains \ + "rustfs/src/storage/concurrency/manager.rs" \ + "pub fn workload_admission_registry_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot" \ + "storage concurrency workload admission registry snapshot" +require_source_contains \ + "rustfs/src/storage/concurrency/manager.rs" \ + "WorkloadClass::ForegroundRead => self.get_object_admission_snapshot()" \ + "storage concurrency foreground read admission snapshot" require_source_line \ "crates/storage-api/src/lib.rs" \ "pub use multipart::{CompletePart, ListMultipartsInfo, ListPartsInfo, MultipartInfo, MultipartUploadResult, PartInfo};" \