From 572a001f9315ec365ed7da76dfd927f5d72896ae Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 2 Jul 2026 00:40:42 +0800 Subject: [PATCH] refactor(storage): hide replication runtime handles (#4152) --- .../ecstore-api-facade-inventory.md | 2 +- .../architecture/ecstore-module-split-plan.md | 7 + .../global-state-crate-split-plan.md | 6 +- rustfs/src/admin/handlers/site_replication.rs | 26 ++-- rustfs/src/storage/storage_api.rs | 143 ++++++++++++++++-- scripts/check_architecture_migration_rules.sh | 19 +++ 6 files changed, 175 insertions(+), 28 deletions(-) diff --git a/docs/architecture/ecstore-api-facade-inventory.md b/docs/architecture/ecstore-api-facade-inventory.md index db63d67ad..4532c1ab6 100644 --- a/docs/architecture/ecstore-api-facade-inventory.md +++ b/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. | diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index 4b7a9b0b5..6331e69c7 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/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`. diff --git a/docs/architecture/global-state-crate-split-plan.md b/docs/architecture/global-state-crate-split-plan.md index 2db6a811c..b60ef7b63 100644 --- a/docs/architecture/global-state-crate-split-plan.md +++ b/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: diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 4cfcb37c6..6a397552a 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/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, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 0f376dcd7..5168e8c5a 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/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 = ecstore_disk::error::Result; 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 = ecstore_rio::EncryptReader; pub(crate) type HardLimitReader = ecstore_rio::HardLimitReader; pub(crate) type NotificationSys = ecstore_notification::NotificationSys; +#[derive(Debug, Clone)] +pub struct ReplicationPoolHandle { + inner: Arc, +} + +impl ReplicationPoolHandle { + fn new(inner: Arc) -> Arc { + 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 { + 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, ctx: CancellationToken, buckets: Vec) -> Result<()> { + self.inner.clone().init_resync(ctx, buckets).await + } +} + +#[derive(Debug, Clone)] +pub struct ReplicationStatsHandle { + inner: Arc, +} + +impl ReplicationStatsHandle { + #[cfg(test)] + pub(crate) fn new() -> Self { + Self { + inner: Arc::new(EcstoreReplicationStats::new()), + } + } + + fn from_ecstore(inner: Arc) -> Arc { + 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 { + 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> { - 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> { - 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 { @@ -620,13 +745,7 @@ pub(crate) async fn prewarm_local_disk_id_map() { } pub(crate) fn replication_queue_current_count() -> Option { - 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, file: &str, data: Vec) -> 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; } } diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 181a648bb..0f1abb04e 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/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" {