refactor(ecstore): migrate background replication pool/stats into InstanceContext (Phase 5 Slice 11) (#4495)

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<Arc<ReplicationStats>>` and
  `replication_pool: OnceCell<Arc<DynReplicationPool>>` (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).

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<S: ReplicationStorage> ReplicationPoolTrait for ReplicationPool<S> {
     }
 }
 
-lazy_static! {
-    pub static ref GLOBAL_REPLICATION_POOL: tokio::sync::OnceCell<Arc<DynReplicationPool>> = tokio::sync::OnceCell::new();
-    pub static ref GLOBAL_REPLICATION_STATS: tokio::sync::OnceCell<Arc<ReplicationStats>> = 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<S: ReplicationStorage>(storage: Arc<S>) {
-    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<S: ReplicationStorage>(storage: Arc<S>)
         })
         .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<DynReplicationPool>

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<Arc<Monitor>>,
+    /// 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<Arc<ReplicationStats>>,
+    /// 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<Arc<DynReplicationPool>>,
 }
 
 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<Arc<ReplicationStats>> {
+        self.replication_stats.get().cloned()
+    }
+
+    /// This instance's background replication pool, if initialized.
+    pub fn replication_pool(&self) -> Option<Arc<DynReplicationPool>> {
+        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<Arc<ReplicationStats>> {
+        &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<Arc<DynReplicationPool>> {
+        &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"
+        );
+    }
 }

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<String> {
 }
 
 pub(crate) fn replication_pool() -> Option<Arc<DynReplicationPool>> {
-    GLOBAL_REPLICATION_POOL.get().cloned()
+    crate::runtime::global::current_ctx().replication_pool()
 }
 
 pub(crate) fn replication_stats() -> Option<Arc<ReplicationStats>> {
-    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) {