From ca63a6cced2e493cb8a6e571ee929117c2b076d7 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 8 Jul 2026 23:15:01 +0800 Subject: [PATCH] refactor(ecstore): migrate background replication pool/stats into InstanceContext (Phase 5 Slice 11) (#4495) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 5 Slice 11 (backlog#939): move the background replication pool and stats — the last service handles, and the only async ones — out of the process statics into the per-instance InstanceContext. - InstanceContext gains `replication_stats: OnceCell>` and `replication_pool: OnceCell>` (tokio async OnceCell), with sync read accessors (`replication_stats`/`replication_pool`/ `replication_initialized`) and pub(crate) cell accessors for the async init. - `init_background_replication` initializes the current instance's cells via the same `get_or_init(async {…}).await` (workers still spawned once on first init). The lifecycle owner helpers and the runtime-source accessors keep their signatures and route through the current instance's context; the two statics (and the now-unused lazy_static import) are removed. The replication-boundary arch guard still passes. Single-instance: init materializes one shared pool/stats via the bootstrap context — byte-for-byte the same as the eager statics. Tests: replication state is None until set and independent across instances. Verification: cargo test -p rustfs-ecstore (22 instance-context tests green), cargo clippy -p rustfs-ecstore --all-targets (clean), make pre-commit (pass). Refs: backlog#939 (Phase 5, Slice 11). Stacked on Slice 10 (#4494). --- .../bucket/replication/replication_pool.rs | 20 +++--- crates/ecstore/src/runtime/instance.rs | 66 ++++++++++++++++++- crates/ecstore/src/runtime/sources.rs | 11 ++-- 3 files changed, 80 insertions(+), 17 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 68c22e0c2..48e5486bf 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -45,7 +45,6 @@ use super::replication_storage_boundary::{ use super::replication_target_boundary::{ReplicationTargetStore, replication_object_is_ssec_encrypted}; use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; -use lazy_static::lazy_static; use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str}; use std::sync::Arc; use std::sync::atomic::AtomicI32; @@ -1293,14 +1292,16 @@ impl ReplicationPoolTrait for ReplicationPool { } } -lazy_static! { - pub static ref GLOBAL_REPLICATION_POOL: tokio::sync::OnceCell> = tokio::sync::OnceCell::new(); - pub static ref GLOBAL_REPLICATION_STATS: tokio::sync::OnceCell> = tokio::sync::OnceCell::new(); -} - -/// Initializes background replication with the given options +/// Initializes background replication with the given options. +/// +/// Phase 5 (backlog#939): the replication stats/pool moved into the per-instance +/// `InstanceContext`; this owner initializes the current instance's cells +/// (lazily, once — single-instance behavior is unchanged). pub async fn init_background_replication(storage: Arc) { - let stats = GLOBAL_REPLICATION_STATS + let ctx = crate::runtime::global::current_ctx(); + + let stats = ctx + .replication_stats_cell() .get_or_init(|| async { let stats = Arc::new(ReplicationStats::new()); stats.start_background_tasks().await; @@ -1308,7 +1309,8 @@ pub async fn init_background_replication(storage: Arc) }) .await; - let _pool = GLOBAL_REPLICATION_POOL + let _pool = ctx + .replication_pool_cell() .get_or_init(|| async { let pool = ReplicationPool::new(ReplicationPoolOpts::default(), stats.clone(), storage).await; pool as Arc diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index 1e75fd068..d3f950bc9 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -41,13 +41,14 @@ use crate::bucket::bandwidth::monitor::Monitor; use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState}; +use crate::bucket::replication::{DynReplicationPool, ReplicationStats}; use crate::layout::endpoints::{EndpointServerPools, SetupType}; use crate::services::event_notification::EventNotifier; use crate::services::tier::tier::TierConfigMgr; use rustfs_lock::{GlobalLockManager, get_global_lock_manager}; use s3s::region::Region; use std::sync::{Arc, OnceLock}; -use tokio::sync::RwLock; +use tokio::sync::{OnceCell, RwLock}; use uuid::Uuid; /// Runtime state owned by a single `ECStore` instance. @@ -114,6 +115,16 @@ pub struct InstanceContext { /// [`InstanceContext::init_bucket_monitor`]); read as `None` until then. /// Replaces the process-global bucket-monitor static. bucket_monitor: OnceLock>, + /// This instance's background replication stats (Phase 5 Slice 11, backlog#939). + /// + /// Async set-once (workers are spawned during init); read as `None` until + /// `init_background_replication` runs. Replaces the process-global static. + replication_stats: OnceCell>, + /// This instance's background replication pool (Phase 5 Slice 11, backlog#939). + /// + /// Async set-once, built with the live storage during init. Replaces the + /// process-global static. + replication_pool: OnceCell>, } impl InstanceContext { @@ -141,6 +152,8 @@ impl InstanceContext { expiry_state: OnceLock::new(), transition_state: OnceLock::new(), bucket_monitor: OnceLock::new(), + replication_stats: OnceCell::new(), + replication_pool: OnceCell::new(), } } @@ -230,6 +243,33 @@ impl InstanceContext { self.bucket_monitor.get().cloned() } + /// This instance's background replication stats, if initialized. + pub fn replication_stats(&self) -> Option> { + self.replication_stats.get().cloned() + } + + /// This instance's background replication pool, if initialized. + pub fn replication_pool(&self) -> Option> { + self.replication_pool.get().cloned() + } + + /// Whether background replication has been initialized for this instance. + pub fn replication_initialized(&self) -> bool { + self.replication_stats.get().is_some() && self.replication_pool.get().is_some() + } + + /// The replication stats cell, for the async `init_background_replication` + /// owner to `get_or_init`. Exposed to the replication module only. + pub(crate) fn replication_stats_cell(&self) -> &OnceCell> { + &self.replication_stats + } + + /// The replication pool cell, for the async `init_background_replication` + /// owner to `get_or_init`. Exposed to the replication module only. + pub(crate) fn replication_pool_cell(&self) -> &OnceCell> { + &self.replication_pool + } + /// Update this instance's erasure setup type. pub async fn update_erasure_type(&self, setup_type: SetupType) { *self.erasure_kind.write().await = setup_type; @@ -276,6 +316,8 @@ impl std::fmt::Debug for InstanceContext { .field("expiry_state_set", &self.expiry_state.get().is_some()) .field("transition_state_set", &self.transition_state.get().is_some()) .field("bucket_monitor_set", &self.bucket_monitor.get().is_some()) + .field("replication_stats_set", &self.replication_stats.get().is_some()) + .field("replication_pool_set", &self.replication_pool.get().is_some()) .finish_non_exhaustive() } } @@ -565,4 +607,26 @@ mod tests { let ctx_b = InstanceContext::new(); assert!(ctx_b.bucket_monitor().is_none(), "a fresh instance has its own uninitialized monitor"); } + + // Replication state is None until initialized, and independent per instance. + // (The real async init spawns workers and needs live storage; here we only + // exercise the read side and cross-instance isolation.) + #[tokio::test] + async fn replication_state_is_isolated() { + let ctx_a = InstanceContext::new(); + assert!(ctx_a.replication_stats().is_none()); + assert!(ctx_a.replication_pool().is_none()); + assert!(!ctx_a.replication_initialized()); + + ctx_a.replication_stats_cell().set(Arc::new(ReplicationStats::new())).ok(); + assert!(ctx_a.replication_stats().is_some()); + // The pool is still unset, so replication is not fully initialized. + assert!(!ctx_a.replication_initialized()); + + let ctx_b = InstanceContext::new(); + assert!( + ctx_b.replication_stats().is_none(), + "a distinct instance has independent replication state" + ); + } } diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index cba0ff787..ed64ac7fe 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -23,10 +23,7 @@ use crate::disk::endpoint::Endpoint; use crate::{ bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState}, bucket::metadata_sys::{BucketMetadataSys, get_global_bucket_metadata_sys}, - bucket::replication::{ - DynReplicationPool, ReplicationStats, - replication_pool::{GLOBAL_REPLICATION_POOL, GLOBAL_REPLICATION_STATS}, - }, + bucket::replication::{DynReplicationPool, ReplicationStats}, config::{get_global_storage_class, set_global_storage_class, storageclass}, disk::{DiskAPI, DiskOption, DiskStore, new_disk}, error::Result, @@ -271,15 +268,15 @@ pub fn deployment_id() -> Option { } pub(crate) fn replication_pool() -> Option> { - GLOBAL_REPLICATION_POOL.get().cloned() + crate::runtime::global::current_ctx().replication_pool() } pub(crate) fn replication_stats() -> Option> { - GLOBAL_REPLICATION_STATS.get().cloned() + crate::runtime::global::current_ctx().replication_stats() } pub(crate) fn replication_runtime_initialized() -> bool { - GLOBAL_REPLICATION_STATS.get().is_some() && GLOBAL_REPLICATION_POOL.get().is_some() + crate::runtime::global::current_ctx().replication_initialized() } pub(crate) fn ensure_deployment_id(deployment_id: Uuid) {