feat: expose replication workload admission snapshot (#3606)

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

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.

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};

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<usize>,
+    queued: Option<usize>,
+) -> 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)

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"