diff --git a/crates/ecstore/src/runtime/global.rs b/crates/ecstore/src/runtime/global.rs index 0387c7257..ba43afac3 100644 --- a/crates/ecstore/src/runtime/global.rs +++ b/crates/ecstore/src/runtime/global.rs @@ -64,20 +64,19 @@ lazy_static! { rustfs_utils::crypto::hex(GLOBAL_LOCAL_NODE_NAME_FALLBACK.as_bytes()); pub static ref GLOBAL_LOCAL_LOCK_CLIENT: OnceLock> = OnceLock::new(); pub static ref GLOBAL_LOCK_CLIENTS: OnceLock>> = OnceLock::new(); - pub static ref GLOBAL_BUCKET_MONITOR: OnceLock> = OnceLock::new(); } pub fn init_global_bucket_monitor(num_nodes: u64) { - if GLOBAL_BUCKET_MONITOR.set(Monitor::new(num_nodes)).is_err() { + if !current_ctx().init_bucket_monitor(num_nodes) { warn!( - "global bucket monitor already initialized, ignoring re-initialization with num_nodes={}", + "bucket monitor already initialized, ignoring re-initialization with num_nodes={}", num_nodes ); } } pub fn get_global_bucket_monitor() -> Option> { - GLOBAL_BUCKET_MONITOR.get().cloned() + current_ctx().bucket_monitor() } // Startup-owned process globals intentionally fail fast on duplicate writes. diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index 95981b7f5..1e75fd068 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -39,6 +39,7 @@ //! startup writes and post-construction reads hit the same cell and //! single-instance behavior is byte-for-byte unchanged. +use crate::bucket::bandwidth::monitor::Monitor; use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState}; use crate::layout::endpoints::{EndpointServerPools, SetupType}; use crate::services::event_notification::EventNotifier; @@ -107,6 +108,12 @@ pub struct InstanceContext { /// Lazily materialized; holds the instance's transition workers/stats. /// Replaces the eager process static. transition_state: OnceLock>, + /// This instance's bucket bandwidth monitor (Phase 5 Slice 10, backlog#939). + /// + /// Set once at startup with the cluster node count (see + /// [`InstanceContext::init_bucket_monitor`]); read as `None` until then. + /// Replaces the process-global bucket-monitor static. + bucket_monitor: OnceLock>, } impl InstanceContext { @@ -133,6 +140,7 @@ impl InstanceContext { event_notifier: OnceLock::new(), expiry_state: OnceLock::new(), transition_state: OnceLock::new(), + bucket_monitor: OnceLock::new(), } } @@ -210,6 +218,18 @@ impl InstanceContext { self.transition_state.get_or_init(TransitionState::new).clone() } + /// Initialize this instance's bucket bandwidth monitor with the cluster + /// node count. Returns `false` if it was already initialized (the caller + /// logs; matches the process global's ignore-and-warn behavior). + pub fn init_bucket_monitor(&self, num_nodes: u64) -> bool { + self.bucket_monitor.set(Monitor::new(num_nodes)).is_ok() + } + + /// This instance's bucket bandwidth monitor, if it has been initialized. + pub fn bucket_monitor(&self) -> Option> { + self.bucket_monitor.get().cloned() + } + /// Update this instance's erasure setup type. pub async fn update_erasure_type(&self, setup_type: SetupType) { *self.erasure_kind.write().await = setup_type; @@ -255,6 +275,7 @@ impl std::fmt::Debug for InstanceContext { .field("event_notifier_set", &self.event_notifier.get().is_some()) .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()) .finish_non_exhaustive() } } @@ -530,4 +551,18 @@ mod tests { assert!(!Arc::ptr_eq(&ctx_a.expiry_state(), &ctx_b.expiry_state())); assert!(!Arc::ptr_eq(&ctx_a.transition_state(), &ctx_b.transition_state())); } + + // The bucket monitor is None until initialized, set-once (second init is a + // no-op returning false), and independent across instances. + #[tokio::test] + async fn bucket_monitor_init_once_and_isolated() { + let ctx_a = InstanceContext::new(); + assert!(ctx_a.bucket_monitor().is_none()); + assert!(ctx_a.init_bucket_monitor(4)); + assert!(ctx_a.bucket_monitor().is_some()); + assert!(!ctx_a.init_bucket_monitor(8), "second init must be ignored"); + + let ctx_b = InstanceContext::new(); + assert!(ctx_b.bucket_monitor().is_none(), "a fresh instance has its own uninitialized monitor"); + } }