refactor(storage): hide replication runtime handles (#4152)

docs/architecture/ecstore-api-facade-inventory.md

@@ -20,7 +20,7 @@ External `rustfs_ecstore::api` imports must stay in these local boundary files:
 
 | Boundary file | Current facade families |
 |---|---|
-| `rustfs/src/storage/storage_api.rs` | Broad RustFS storage owner bridge for admin, bucket, capacity, client, compression, cluster, config, data usage, disk, error, event, global bootstrap controls, runtime-source getters, layout, metrics, notification, rebalance, rio, rpc, set disk, storage, and tier. |
+| `rustfs/src/storage/storage_api.rs` | Broad RustFS storage owner bridge for admin, bucket, capacity, client, compression, cluster, config, data usage, disk, error, event, global bootstrap controls, runtime-source getters, layout, metrics, notification, rebalance, rio, rpc, set disk, storage, and tier. Replication pool/stat handles are projected into RustFS-local wrapper types here. |
 | `crates/scanner/src/storage_api.rs` | Scanner bridge for bucket lifecycle, replication, metadata, capacity, config, data usage, disk, error, runtime, set disk, storage, and tier. Replication queue config, admission, and heal object DTOs are projected into scanner-local types here. |
 | `crates/obs/src/metrics/storage_api.rs` | Metrics bridge for bucket bandwidth, lifecycle, replication, quota, capacity, data usage, error, runtime, and storage. |
 | `crates/iam/src/storage_api.rs` | IAM bridge for config, error, notification, runtime, and storage. |

docs/architecture/ecstore-module-split-plan.md

@@ -126,6 +126,9 @@ Current coupling:
 - app object and multipart writes call object-replication boundary helpers
   instead of constructing replication work DTOs or choosing object replication
   operation types at the use-case layer;
+- RustFS runtime consumers receive replication pool/stat handles through
+  storage-owner wrapper types instead of carrying ECStore replication handles
+  through app, admin, startup, or workload-admission layers;
 - global replication pool/stat initialization still lives with ECStore runtime
   compatibility state;
 - modules inside `bucket/replication` use local relative paths rather than the
@@ -193,6 +196,10 @@ Required contracts before crate movement:
   metrics through obs-local snapshot DTOs in
   `crates/obs/src/metrics/storage_api.rs` instead of carrying the ECStore
   replication stats handle through collectors.
+- `ReplicationPoolHandle` / `ReplicationStatsHandle`: RustFS app, admin,
+  startup, and workload-admission code use storage-owner wrapper types from
+  `rustfs/src/storage/storage_api.rs` for pool activity, resync, queue counts,
+  proxy stats, and site metrics snapshots.
 - `ReplicationScannerBridge`: scanner-originated replication heal scheduling is
   exposed through the contract type in
   `crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs`.

docs/architecture/global-state-crate-split-plan.md

@@ -174,8 +174,10 @@ Required evidence before proposing the split:
 
 Decision: do not split in code yet. Replication remains coupled to ECStore
 object APIs, bucket target clients, metadata systems, file metadata replication
-state, runtime replication pool/stat handles, bucket monitor state, scanner
-repair classification, lifecycle-originated deletes, and notification events.
+state, ECStore-owned runtime replication pool/stat handles, bucket monitor
+state, scanner repair classification, lifecycle-originated deletes, and
+notification events. RustFS-facing runtime consumers should use storage-owner
+wrapper handles while that state remains in ECStore.
 
 Required evidence before proposing the split:
 

rustfs/src/admin/handlers/site_replication.rs

@@ -2395,7 +2395,7 @@ async fn build_metrics_summary(local_peer: &PeerInfo) -> SRMetricsSummary {
         return SRMetricsSummary::default();
     };
 
-    let node = stats.get_sr_metrics_for_node().await;
+    let node = stats.site_metrics_snapshot().await;
     let mut metrics = BTreeMap::new();
     metrics.insert(
         local_peer.deployment_id.clone(),
@@ -2412,25 +2412,25 @@ async fn build_metrics_summary(local_peer: &PeerInfo) -> SRMetricsSummary {
 
     SRMetricsSummary {
         active_workers: WorkerStat {
-            curr: node.active_workers.curr,
-            avg: node.active_workers.avg,
-            max: node.active_workers.max,
+            curr: node.active_workers_curr,
+            avg: node.active_workers_avg,
+            max: node.active_workers_max,
         },
         replica_size: node.replica_size,
         replica_count: node.replica_count,
         queued: InQueueMetric {
-            curr: qstat(node.queued.curr.count, node.queued.curr.bytes),
-            avg: qstat(node.queued.avg.count, node.queued.avg.bytes),
-            max: qstat(node.queued.max.count, node.queued.max.bytes),
+            curr: qstat(node.queued_curr_count, node.queued_curr_bytes),
+            avg: qstat(node.queued_avg_count, node.queued_avg_bytes),
+            max: qstat(node.queued_max_count, node.queued_max_bytes),
         },
         in_progress: InProgressMetric::default(),
         proxied: ReplProxyMetric {
-            get_total: non_negative_u64(node.proxied.get_total),
-            head_total: non_negative_u64(node.proxied.head_total),
-            get_failed_total: non_negative_u64(node.proxied.get_failed),
-            head_failed_total: non_negative_u64(node.proxied.head_failed),
-            put_tag_total: non_negative_u64(node.proxied.put_tag_total),
-            put_tag_failed_total: non_negative_u64(node.proxied.put_tag_failed),
+            get_total: non_negative_u64(node.proxy_get_total),
+            head_total: non_negative_u64(node.proxy_head_total),
+            get_failed_total: non_negative_u64(node.proxy_get_failed),
+            head_failed_total: non_negative_u64(node.proxy_head_failed),
+            put_tag_total: non_negative_u64(node.proxy_put_tag_total),
+            put_tag_failed_total: non_negative_u64(node.proxy_put_tag_failed),
             ..Default::default()
         },
         metrics,

rustfs/src/storage/storage_api.rs

@@ -17,6 +17,7 @@
 use std::sync::Arc;
 
 use rustfs_storage_api as storage_contracts;
+use tokio_util::sync::CancellationToken;
 
 pub(crate) mod contract {
     pub(crate) mod admin {
@@ -483,7 +484,9 @@ pub(crate) type DiskResult<T> = ecstore_disk::error::Result<T>;
 pub(crate) type DiskStore = ecstore_disk::DiskStore;
 #[cfg(test)]
 pub(crate) type DisksLayout = ecstore_layout::DisksLayout;
-pub(crate) type DynReplicationPool = ecstore_bucket::replication::DynReplicationPool;
+type EcstoreDynReplicationPool = ecstore_bucket::replication::DynReplicationPool;
+type EcstoreReplicationStats = ecstore_bucket::replication::ReplicationStats;
+pub(crate) type DynReplicationPool = ReplicationPoolHandle;
 pub(crate) type DynReader = ecstore_rio::DynReader;
 pub(crate) type ECStore = ecstore_storage::ECStore;
 pub(crate) type Endpoint = ecstore_disk::endpoint::Endpoint;
@@ -511,7 +514,7 @@ pub(crate) type ReadMultipleReq = ecstore_disk::ReadMultipleReq;
 pub(crate) type ReadMultipleResp = ecstore_disk::ReadMultipleResp;
 pub(crate) type ReadOptions = ecstore_disk::ReadOptions;
 pub(crate) type RenameDataResp = ecstore_disk::RenameDataResp;
-pub(crate) type ReplicationStats = ecstore_bucket::replication::ReplicationStats;
+pub(crate) type ReplicationStats = ReplicationStatsHandle;
 pub(crate) type SetupType = ecstore_layout::SetupType;
 pub(crate) type StorageError = ecstore_error::StorageError;
 pub(crate) type TierConfigMgr = ecstore_tier::TierConfigMgr;
@@ -531,6 +534,128 @@ pub(crate) type EncryptReader<R> = ecstore_rio::EncryptReader<R>;
 pub(crate) type HardLimitReader<R> = ecstore_rio::HardLimitReader<R>;
 pub(crate) type NotificationSys = ecstore_notification::NotificationSys;
 
+#[derive(Debug, Clone)]
+pub struct ReplicationPoolHandle {
+    inner: Arc<EcstoreDynReplicationPool>,
+}
+
+impl ReplicationPoolHandle {
+    fn new(inner: Arc<EcstoreDynReplicationPool>) -> Arc<Self> {
+        Arc::new(Self { inner })
+    }
+
+    pub(crate) fn active_workers(&self) -> i32 {
+        self.inner.active_workers()
+    }
+
+    pub(crate) fn active_mrf_workers(&self) -> i32 {
+        self.inner.active_mrf_workers()
+    }
+
+    pub(crate) fn active_lrg_workers(&self) -> i32 {
+        self.inner.active_lrg_workers()
+    }
+
+    pub(crate) async fn get_bucket_resync_status(
+        &self,
+        bucket: &str,
+    ) -> Result<ecstore_bucket::replication::BucketReplicationResyncStatus> {
+        self.inner.get_bucket_resync_status(bucket).await
+    }
+
+    pub(crate) async fn cancel_bucket_resync(&self, opts: ecstore_bucket::replication::ResyncOpts) -> Result<()> {
+        self.inner.clone().cancel_bucket_resync(opts).await
+    }
+
+    pub(crate) async fn start_bucket_resync(&self, opts: ecstore_bucket::replication::ResyncOpts) -> Result<()> {
+        self.inner.clone().start_bucket_resync(opts).await
+    }
+
+    pub(crate) async fn init_resync(self: Arc<Self>, ctx: CancellationToken, buckets: Vec<String>) -> Result<()> {
+        self.inner.clone().init_resync(ctx, buckets).await
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct ReplicationStatsHandle {
+    inner: Arc<EcstoreReplicationStats>,
+}
+
+impl ReplicationStatsHandle {
+    #[cfg(test)]
+    pub(crate) fn new() -> Self {
+        Self {
+            inner: Arc::new(EcstoreReplicationStats::new()),
+        }
+    }
+
+    fn from_ecstore(inner: Arc<EcstoreReplicationStats>) -> Arc<Self> {
+        Arc::new(Self { inner })
+    }
+
+    pub(crate) async fn get_latest_replication_stats(&self, bucket: &str) -> ecstore_bucket::replication::BucketStats {
+        self.inner.get_latest_replication_stats(bucket).await
+    }
+
+    pub(crate) async fn site_metrics_snapshot(&self) -> ReplicationSiteMetricsSnapshot {
+        let metrics = self.inner.get_sr_metrics_for_node().await;
+        ReplicationSiteMetricsSnapshot {
+            uptime: metrics.uptime,
+            queued_curr_count: metrics.queued.curr.count,
+            queued_curr_bytes: metrics.queued.curr.bytes,
+            queued_avg_count: metrics.queued.avg.count,
+            queued_avg_bytes: metrics.queued.avg.bytes,
+            queued_max_count: metrics.queued.max.count,
+            queued_max_bytes: metrics.queued.max.bytes,
+            active_workers_curr: metrics.active_workers.curr,
+            active_workers_avg: metrics.active_workers.avg,
+            active_workers_max: metrics.active_workers.max,
+            proxy_get_total: metrics.proxied.get_total,
+            proxy_head_total: metrics.proxied.head_total,
+            proxy_get_failed: metrics.proxied.get_failed,
+            proxy_head_failed: metrics.proxied.head_failed,
+            proxy_put_tag_total: metrics.proxied.put_tag_total,
+            proxy_put_tag_failed: metrics.proxied.put_tag_failed,
+            replica_size: metrics.replica_size,
+            replica_count: metrics.replica_count,
+        }
+    }
+
+    fn queue_current_count(&self) -> Option<i64> {
+        self.inner
+            .q_cache
+            .try_lock()
+            .ok()
+            .map(|cache| cache.sr_queue_stats.curr.get_current_count())
+    }
+
+    async fn record_proxy(&self, bucket: &str, api: &str, is_err: bool) {
+        self.inner.inc_proxy(bucket, api, is_err).await;
+    }
+}
+
+#[derive(Debug, Clone, Copy, Default)]
+pub(crate) struct ReplicationSiteMetricsSnapshot {
+    pub(crate) uptime: i64,
+    pub(crate) queued_curr_count: i64,
+    pub(crate) queued_curr_bytes: i64,
+    pub(crate) queued_avg_count: i64,
+    pub(crate) queued_avg_bytes: i64,
+    pub(crate) queued_max_count: i64,
+    pub(crate) queued_max_bytes: i64,
+    pub(crate) active_workers_curr: i32,
+    pub(crate) active_workers_avg: f64,
+    pub(crate) active_workers_max: i32,
+    pub(crate) proxy_get_total: i64,
+    pub(crate) proxy_head_total: i64,
+    pub(crate) proxy_get_failed: i64,
+    pub(crate) proxy_head_failed: i64,
+    pub(crate) proxy_put_tag_total: i64,
+    pub(crate) proxy_put_tag_failed: i64,
+    pub(crate) replica_size: i64,
+    pub(crate) replica_count: i64,
+}
+
 pub(crate) async fn get_local_server_property() -> rustfs_madmin::ServerProperties {
     ecstore_admin::get_local_server_property().await
 }
@@ -564,11 +689,11 @@ pub(crate) fn disk_endpoint(disk: &DiskStore) -> String {
 }
 
 pub(crate) fn get_global_replication_pool() -> Option<Arc<DynReplicationPool>> {
-    ecstore_bucket::replication::get_global_replication_pool()
+    ecstore_bucket::replication::get_global_replication_pool().map(ReplicationPoolHandle::new)
 }
 
 pub(crate) fn get_global_replication_stats() -> Option<Arc<ReplicationStats>> {
-    ecstore_bucket::replication::get_global_replication_stats()
+    ecstore_bucket::replication::get_global_replication_stats().map(ReplicationStatsHandle::from_ecstore)
 }
 
 pub(crate) fn get_global_boot_time() -> Option<std::time::SystemTime> {
@@ -620,13 +745,7 @@ pub(crate) async fn prewarm_local_disk_id_map() {
 }
 
 pub(crate) fn replication_queue_current_count() -> Option<i64> {
-    get_global_replication_stats().and_then(|stats| {
-        stats
-            .q_cache
-            .try_lock()
-            .ok()
-            .map(|cache| cache.sr_queue_stats.curr.get_current_count())
-    })
+    get_global_replication_stats().and_then(|stats| stats.queue_current_count())
 }
 
 pub(crate) async fn save_config(api: Arc<ECStore>, file: &str, data: Vec<u8>) -> Result<()> {
@@ -1048,7 +1167,7 @@ pub(crate) fn check_retention_for_modification(
 
 pub(crate) async fn record_replication_proxy(bucket: &str, api: &str, is_err: bool) {
     if let Some(stats) = get_global_replication_stats() {
-        stats.inc_proxy(bucket, api, is_err).await;
+        stats.record_proxy(bucket, api, is_err).await;
     }
 }
 

scripts/check_architecture_migration_rules.sh

@@ -199,6 +199,7 @@ FUZZ_ECSTORE_COMPAT_BYPASS_HITS_FILE="${TMP_DIR}/fuzz_ecstore_compat_bypass_hits
 EXTERNAL_ECSTORE_API_BOUNDARY_HITS_FILE="${TMP_DIR}/external_ecstore_api_boundary_hits.txt"
 REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/replication_facade_bypass_hits.txt"
 REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE="${TMP_DIR}/replication_facade_wildcard_export_hits.txt"
+STORAGE_REPLICATION_HANDLE_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/storage_replication_handle_boundary_bypass_hits.txt"
 ADMIN_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/admin_replication_dto_boundary_bypass_hits.txt"
 APP_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/app_replication_dto_boundary_bypass_hits.txt"
 SCANNER_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/scanner_replication_dto_boundary_bypass_hits.txt"
@@ -2539,6 +2540,24 @@ if [[ -s "$REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE" ]]; then
   report_failure "replication facade must use explicit compatibility exports instead of wildcard re-exports: $(paste -sd '; ' "$REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE")"
 fi
 
+(
+  cd "$ROOT_DIR"
+  {
+    rg -n --with-filename 'ecstore_bucket::replication::(DynReplicationPool|ReplicationStats|get_global_replication_pool|get_global_replication_stats|init_background_replication)' \
+      rustfs/src \
+      --glob '*.rs' \
+      --glob '!rustfs/src/storage/storage_api.rs' || true
+    rg -n --with-filename '\b(get_sr_metrics_for_node|q_cache|inc_proxy)\b' \
+      rustfs/src \
+      --glob '*.rs' \
+      --glob '!rustfs/src/storage/storage_api.rs' || true
+  }
+) >"$STORAGE_REPLICATION_HANDLE_BOUNDARY_BYPASS_HITS_FILE"
+
+if [[ -s "$STORAGE_REPLICATION_HANDLE_BOUNDARY_BYPASS_HITS_FILE" ]]; then
+  report_failure "RustFS replication pool/stat handles must stay behind rustfs/src/storage/storage_api.rs: $(paste -sd '; ' "$STORAGE_REPLICATION_HANDLE_BOUNDARY_BYPASS_HITS_FILE")"
+fi
+
 (
   cd "$ROOT_DIR"
   {