diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 3f9547788..82507e859 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,18 +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-scanner-replication-admission-snapshots` -- Baseline: `origin/main` after `rustfs/rustfs#3605` - (`00ca3b7c1c4ffbe98c0dd8530cb26b63e94c586a`). +- Branch: `overtrue/arch-foreground-scanner-admission-snapshots` +- Baseline: `origin/main` after `rustfs/rustfs#3606` + (`7cb7aefc3bfd33cfaa46c718db15f00b0471fe88`). - PR type for this branch: `consumer-migration` - Runtime behavior changes: none. -- Rust code changes: extend read-only workload admission snapshots from heal - repair counters to replication runtime worker and queue counters. +- Rust code changes: extend the RustFS workload admission registry from + repair/replication snapshots to foreground-read, scanner, and metadata owner + snapshots. - CI/script changes: extend migration guard coverage for RustFS workload admission provider implementations. -- Docs changes: add RustFS replication provider notes to +- Docs changes: add RustFS runtime owner provider notes to [`workload-admission-contracts.md`](workload-admission-contracts.md) and - record the API-058/R-018 provider slice. + record the API-059/R-019 provider slice. ## Phase 0 Tasks @@ -198,6 +199,21 @@ 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-059/R-019` Expose RustFS runtime owner admission snapshots. + - Completed slice: extend the RustFS workload admission provider to map + foreground-read disk permit state, scanner active work units, and bucket + metadata runtime initialization into the workload registry. + - Acceptance: RustFS-level workload admission snapshots expose existing + foreground-read, scanner, and metadata owner state without changing + admission, queueing, scanner scheduling, metadata loading, metadata locks, + or object write behavior. + - Must preserve: disk-read semaphore acquisition, scanner cycle scheduling, + bucket metadata initialization and loading, object write paths, request + guards, and queue behavior. + - 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 86fcf3049..d8e948618 100644 --- a/docs/architecture/workload-admission-contracts.md +++ b/docs/architecture/workload-admission-contracts.md @@ -92,3 +92,23 @@ snapshot from the existing replication pool and queue statistics: This is an observation surface only. Replication admission, queue channel capacity, worker resize behavior, MRF handling, target dispatch, and resync behavior are unchanged. + +## RustFS Runtime Owner Snapshot Extraction + +The RustFS integration layer now extends the workload admission registry with +additional read-only owner mappings: + +- `ForegroundRead` reuses the storage `ConcurrencyManager` disk-read permit + snapshot so the RustFS-level provider exposes the same active and limit + counts as the storage-local provider. +- `Scanner` reports the existing scanner active work-unit counter. When the + counter is zero, the snapshot remains `Unknown` because the current counter + cannot distinguish an idle scanner from a scanner that has not initialized. +- `Metadata` reports `Open` once the bucket metadata runtime handle is + available, and `Unknown` before initialization. +- `ForegroundWrite` remains `Unknown` until a write-specific admission owner + exposes a read-only surface. + +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. diff --git a/rustfs/src/storage_compat.rs b/rustfs/src/storage_compat.rs index 34d1b6ac8..2e66cd037 100644 --- a/rustfs/src/storage_compat.rs +++ b/rustfs/src/storage_compat.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub(crate) use rustfs_ecstore::bucket::metadata_sys::init_bucket_metadata_sys; +pub(crate) use rustfs_ecstore::bucket::metadata_sys::{get_global_bucket_metadata_sys, init_bucket_metadata_sys}; pub(crate) use rustfs_ecstore::bucket::migration::{try_migrate_bucket_metadata, try_migrate_iam_config}; pub(crate) use rustfs_ecstore::bucket::replication::{ GLOBAL_REPLICATION_STATS, get_global_replication_pool, init_background_replication, diff --git a/rustfs/src/workload_admission.rs b/rustfs/src/workload_admission.rs index 97895f937..e3fc4ba9f 100644 --- a/rustfs/src/workload_admission.rs +++ b/rustfs/src/workload_admission.rs @@ -17,12 +17,15 @@ use rustfs_concurrency::{ WorkloadClass, }; -use crate::storage_compat::{GLOBAL_REPLICATION_STATS, get_global_replication_pool}; +use crate::storage::concurrency::get_concurrency_manager; +use crate::storage_compat::{GLOBAL_REPLICATION_STATS, get_global_bucket_metadata_sys, get_global_replication_pool}; +const BUCKET_METADATA_RUNTIME_NOT_INITIALIZED: &str = "bucket metadata runtime not initialized"; const HEAL_MANAGER_NOT_INITIALIZED: &str = "heal manager not initialized"; const NOT_EXPOSED_BY_PROVIDER: &str = "not exposed by RustFS workload admission provider"; 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"; #[derive(Debug, Default, Clone, Copy)] pub struct RustFsWorkloadAdmissionSnapshotProvider; @@ -38,6 +41,9 @@ pub fn workload_admission_registry_snapshot() -> WorkloadAdmissionRegistrySnapsh .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), @@ -47,6 +53,54 @@ pub fn workload_admission_registry_snapshot() -> WorkloadAdmissionRegistrySnapsh WorkloadAdmissionRegistrySnapshot::new(entries) } +pub fn foreground_read_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { + get_concurrency_manager().get_object_admission_snapshot() +} + +pub fn metadata_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { + metadata_workload_admission_snapshot_from_initialized(get_global_bucket_metadata_sys().is_some()) +} + +fn metadata_workload_admission_snapshot_from_initialized(runtime_initialized: bool) -> WorkloadAdmissionSnapshot { + let state = if runtime_initialized { + AdmissionState::Open + } else { + AdmissionState::Unknown + }; + + let snapshot = WorkloadAdmissionSnapshot::new(WorkloadClass::Metadata, state); + + if runtime_initialized { + snapshot + } else { + snapshot.with_reason(BUCKET_METADATA_RUNTIME_NOT_INITIALIZED) + } +} + +pub fn scanner_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { + scanner_workload_admission_snapshot_from_activity(rustfs_scanner::current_scanner_activity()) +} + +fn scanner_workload_admission_snapshot_from_activity(active: u64) -> WorkloadAdmissionSnapshot { + let state = if active > 0 { + AdmissionState::Open + } else { + AdmissionState::Unknown + }; + + let snapshot = WorkloadAdmissionSnapshot::new(WorkloadClass::Scanner, state).with_counts( + Some(u64_to_usize_saturated(active)), + None, + None, + ); + + if state == AdmissionState::Unknown { + snapshot.with_reason(SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED) + } else { + snapshot + } +} + pub fn repair_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { repair_workload_admission_snapshot_from_counts( rustfs_heal::get_heal_manager().is_some(), @@ -137,6 +191,59 @@ fn i32_to_usize_saturated(value: i32) -> usize { mod tests { use super::*; + #[test] + fn foreground_read_snapshot_uses_storage_concurrency_admission() { + let snapshot = foreground_read_workload_admission_snapshot(); + + assert_eq!(snapshot.class, WorkloadClass::ForegroundRead); + assert_ne!(snapshot.state, AdmissionState::Unknown); + assert!(snapshot.active.is_some()); + assert!(snapshot.limit.is_some()); + } + + #[test] + fn metadata_snapshot_reports_initialized_runtime() { + let snapshot = metadata_workload_admission_snapshot_from_initialized(true); + + assert_eq!(snapshot.class, WorkloadClass::Metadata); + assert_eq!(snapshot.state, AdmissionState::Open); + assert_eq!(snapshot.active, None); + assert_eq!(snapshot.queued, None); + assert_eq!(snapshot.limit, None); + assert_eq!(snapshot.reason, None); + } + + #[test] + fn metadata_snapshot_is_unknown_before_runtime_initializes() { + let snapshot = metadata_workload_admission_snapshot_from_initialized(false); + + assert_eq!(snapshot.class, WorkloadClass::Metadata); + assert_eq!(snapshot.state, AdmissionState::Unknown); + assert_eq!(snapshot.reason.as_deref(), Some(BUCKET_METADATA_RUNTIME_NOT_INITIALIZED)); + } + + #[test] + fn scanner_snapshot_reports_active_work_units() { + let snapshot = scanner_workload_admission_snapshot_from_activity(5); + + assert_eq!(snapshot.class, WorkloadClass::Scanner); + assert_eq!(snapshot.state, AdmissionState::Open); + assert_eq!(snapshot.active, Some(5)); + assert_eq!(snapshot.queued, None); + assert_eq!(snapshot.limit, None); + assert_eq!(snapshot.reason, None); + } + + #[test] + fn scanner_snapshot_is_unknown_when_idle_or_uninitialized() { + let snapshot = scanner_workload_admission_snapshot_from_activity(0); + + assert_eq!(snapshot.class, WorkloadClass::Scanner); + assert_eq!(snapshot.state, AdmissionState::Unknown); + assert_eq!(snapshot.active, Some(0)); + assert_eq!(snapshot.reason.as_deref(), Some(SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED)); + } + #[test] fn repair_snapshot_reports_heal_counters() { let snapshot = repair_workload_admission_snapshot_from_counts(true, 2, 3); @@ -206,9 +313,10 @@ mod tests { assert!(registry.get(WorkloadClass::Repair).is_some()); assert!(registry.get(WorkloadClass::Replication).is_some()); - assert_eq!( + assert_ne!( registry.get(WorkloadClass::ForegroundRead).map(|snapshot| snapshot.state), Some(AdmissionState::Unknown) ); + assert_eq!(registry.get(WorkloadClass::Scanner).and_then(|snapshot| snapshot.active), Some(0)); } } diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 8ceb35bd6..c2816c6ed 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -304,6 +304,30 @@ 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 foreground_read_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \ + "RustFS foreground-read workload admission snapshot helper" +require_source_contains \ + "rustfs/src/workload_admission.rs" \ + "WorkloadClass::ForegroundRead => foreground_read_workload_admission_snapshot()" \ + "RustFS foreground-read workload admission class mapping" +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()" \ + "RustFS metadata workload admission class mapping" +require_source_contains \ + "rustfs/src/workload_admission.rs" \ + "pub fn scanner_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \ + "RustFS scanner workload admission snapshot helper" +require_source_contains \ + "rustfs/src/workload_admission.rs" \ + "WorkloadClass::Scanner => scanner_workload_admission_snapshot()" \ + "RustFS scanner workload admission class mapping" require_source_contains \ "rustfs/src/workload_admission.rs" \ "pub fn repair_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \