From 7cb7aefc3bfd33cfaa46c718db15f00b0471fe88 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Fri, 19 Jun 2026 11:13:17 +0800 Subject: [PATCH] feat: expose replication workload admission snapshot (#3606) --- docs/architecture/migration-progress.md | 29 ++++-- .../workload-admission-contracts.md | 17 ++++ rustfs/src/storage_compat.rs | 4 +- rustfs/src/workload_admission.rs | 90 +++++++++++++++++++ scripts/check_architecture_migration_rules.sh | 8 ++ 5 files changed, 140 insertions(+), 8 deletions(-) diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index c4356781f..3f9547788 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-workload-admission-owner-snapshots` -- Baseline: `origin/main` after `rustfs/rustfs#3590` - (`8cf3c0bf0100d9d2234f08b48df31fd73b92ed43`). +- Branch: `overtrue/arch-scanner-replication-admission-snapshots` +- Baseline: `origin/main` after `rustfs/rustfs#3605` + (`00ca3b7c1c4ffbe98c0dd8530cb26b63e94c586a`). - PR type for this branch: `consumer-migration` - Runtime behavior changes: none. -- Rust code changes: expose read-only repair workload admission snapshots from - existing heal runtime counters. +- Rust code changes: extend read-only workload admission snapshots from heal + repair counters to replication runtime worker and queue counters. - CI/script changes: extend migration guard coverage for RustFS workload admission provider implementations. -- Docs changes: add RustFS heal provider notes to +- Docs changes: add RustFS replication provider notes to [`workload-admission-contracts.md`](workload-admission-contracts.md) and - record the API-057/R-017 provider slice. + record the API-058/R-018 provider slice. ## Phase 0 Tasks @@ -183,6 +183,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-058/R-018` Expose replication admission snapshot. + - Completed slice: extend the RustFS workload admission provider to map + existing replication worker and site queue counters to the `Replication` + workload class. + - Acceptance: replication admission pressure is observable through the + `rustfs-concurrency` workload snapshot contract without changing + replication queueing, channel capacity, worker resize, MRF, target dispatch, + or resync behavior. + - Must preserve: replication admission, queue channel capacity, worker resize + policy, MRF handling, target dispatch, resync behavior, and queue stats + accounting. + - 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 bacef374d..86fcf3049 100644 --- a/docs/architecture/workload-admission-contracts.md +++ b/docs/architecture/workload-admission-contracts.md @@ -75,3 +75,20 @@ from the heal runtime counters: This is an observation surface only. Heal request admission, queue capacity, priority merge/drop policy, task scheduling, retry handling, and repair behavior are unchanged. + +## Replication Snapshot Extraction + +The RustFS integration layer now exposes a read-only replication admission +snapshot from the existing replication pool and queue statistics: + +- `Replication` reports active regular, large-object, and MRF worker counts. +- `queued` reports the current site replication queue count when queue stats + are immediately observable. +- `limit` remains `None` because replication worker limits remain owned by the + async replication pool and resize policy. +- If the replication runtime has not initialized, or queue stats are currently + locked, the snapshot reports `Unknown` instead of blocking or guessing. + +This is an observation surface only. Replication admission, queue channel +capacity, worker resize behavior, MRF handling, target dispatch, and resync +behavior are unchanged. diff --git a/rustfs/src/storage_compat.rs b/rustfs/src/storage_compat.rs index fe4272681..34d1b6ac8 100644 --- a/rustfs/src/storage_compat.rs +++ b/rustfs/src/storage_compat.rs @@ -14,7 +14,9 @@ pub(crate) use rustfs_ecstore::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::{get_global_replication_pool, init_background_replication}; +pub(crate) use rustfs_ecstore::bucket::replication::{ + GLOBAL_REPLICATION_STATS, get_global_replication_pool, init_background_replication, +}; pub(crate) use rustfs_ecstore::bucket::{metadata, metadata_sys, quota}; pub(crate) use rustfs_ecstore::config::{com, init, init_global_config_sys, try_migrate_server_config}; pub(crate) use rustfs_ecstore::disk::{DiskAPI, RUSTFS_META_BUCKET, endpoint::Endpoint}; diff --git a/rustfs/src/workload_admission.rs b/rustfs/src/workload_admission.rs index cca7248e7..97895f937 100644 --- a/rustfs/src/workload_admission.rs +++ b/rustfs/src/workload_admission.rs @@ -17,8 +17,12 @@ use rustfs_concurrency::{ WorkloadClass, }; +use crate::storage_compat::{GLOBAL_REPLICATION_STATS, get_global_replication_pool}; + 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"; #[derive(Debug, Default, Clone, Copy)] pub struct RustFsWorkloadAdmissionSnapshotProvider; @@ -35,6 +39,7 @@ pub fn workload_admission_registry_snapshot() -> WorkloadAdmissionRegistrySnapsh .copied() .map(|class| match class { WorkloadClass::Repair => repair_workload_admission_snapshot(), + WorkloadClass::Replication => replication_workload_admission_snapshot(), class => WorkloadAdmissionSnapshot::new(class, AdmissionState::Unknown).with_reason(NOT_EXPOSED_BY_PROVIDER), }) .collect(); @@ -74,10 +79,60 @@ fn repair_workload_admission_snapshot_from_counts( } } +pub fn replication_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { + let Some(pool) = get_global_replication_pool() else { + return replication_workload_admission_snapshot_from_counts(false, None, None); + }; + + let active = pool + .active_workers() + .saturating_add(pool.active_lrg_workers()) + .saturating_add(pool.active_mrf_workers()); + let queued = GLOBAL_REPLICATION_STATS.get().and_then(|stats| { + stats + .q_cache + .try_lock() + .ok() + .map(|cache| i64_to_usize_saturated(cache.sr_queue_stats.curr.get_current_count())) + }); + + replication_workload_admission_snapshot_from_counts(true, Some(i32_to_usize_saturated(active)), queued) +} + +fn replication_workload_admission_snapshot_from_counts( + runtime_initialized: bool, + active: Option, + queued: Option, +) -> WorkloadAdmissionSnapshot { + let state = if runtime_initialized && queued.is_some() { + AdmissionState::Open + } else { + AdmissionState::Unknown + }; + + let snapshot = WorkloadAdmissionSnapshot::new(WorkloadClass::Replication, state).with_counts(active, queued, None); + + if !runtime_initialized { + snapshot.with_reason(REPLICATION_RUNTIME_NOT_INITIALIZED) + } else if queued.is_none() { + snapshot.with_reason(REPLICATION_QUEUE_STATS_UNAVAILABLE) + } else { + snapshot + } +} + fn u64_to_usize_saturated(value: u64) -> usize { usize::try_from(value).unwrap_or(usize::MAX) } +fn i64_to_usize_saturated(value: i64) -> usize { + usize::try_from(value.max(0)).unwrap_or(usize::MAX) +} + +fn i32_to_usize_saturated(value: i32) -> usize { + usize::try_from(value.max(0)).unwrap_or(usize::MAX) +} + #[cfg(test)] mod tests { use super::*; @@ -105,6 +160,40 @@ mod tests { assert_eq!(snapshot.reason.as_deref(), Some(HEAL_MANAGER_NOT_INITIALIZED)); } + #[test] + fn replication_snapshot_reports_worker_and_queue_counts() { + let snapshot = replication_workload_admission_snapshot_from_counts(true, Some(4), Some(9)); + + assert_eq!(snapshot.class, WorkloadClass::Replication); + assert_eq!(snapshot.state, AdmissionState::Open); + assert_eq!(snapshot.active, Some(4)); + assert_eq!(snapshot.queued, Some(9)); + assert_eq!(snapshot.limit, None); + assert_eq!(snapshot.reason, None); + } + + #[test] + fn replication_snapshot_is_unknown_before_runtime_initializes() { + let snapshot = replication_workload_admission_snapshot_from_counts(false, None, None); + + assert_eq!(snapshot.class, WorkloadClass::Replication); + assert_eq!(snapshot.state, AdmissionState::Unknown); + assert_eq!(snapshot.active, None); + assert_eq!(snapshot.queued, None); + assert_eq!(snapshot.reason.as_deref(), Some(REPLICATION_RUNTIME_NOT_INITIALIZED)); + } + + #[test] + fn replication_snapshot_is_unknown_when_queue_stats_are_unavailable() { + let snapshot = replication_workload_admission_snapshot_from_counts(true, Some(2), None); + + assert_eq!(snapshot.class, WorkloadClass::Replication); + assert_eq!(snapshot.state, AdmissionState::Unknown); + assert_eq!(snapshot.active, Some(2)); + assert_eq!(snapshot.queued, None); + assert_eq!(snapshot.reason.as_deref(), Some(REPLICATION_QUEUE_STATS_UNAVAILABLE)); + } + #[test] fn provider_covers_required_classes() { let provider = RustFsWorkloadAdmissionSnapshotProvider; @@ -116,6 +205,7 @@ mod tests { } assert!(registry.get(WorkloadClass::Repair).is_some()); + assert!(registry.get(WorkloadClass::Replication).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 bcf4ab075..8ceb35bd6 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -312,6 +312,14 @@ require_source_contains \ "rustfs/src/workload_admission.rs" \ "WorkloadClass::Repair => repair_workload_admission_snapshot()" \ "RustFS repair workload admission class mapping" +require_source_contains \ + "rustfs/src/workload_admission.rs" \ + "pub fn replication_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \ + "RustFS replication workload admission snapshot helper" +require_source_contains \ + "rustfs/src/workload_admission.rs" \ + "WorkloadClass::Replication => replication_workload_admission_snapshot()" \ + "RustFS replication workload admission class mapping" ( cd "$ROOT_DIR"