refactor(ecstore): migrate the bucket bandwidth monitor into InstanceContext (Phase 5 Slice 10, re-submit) (#4494)

refactor(ecstore): migrate the bucket bandwidth monitor into InstanceContext (Phase 5 Slice 10)

Phase 5 Slice 10 (backlog#939): move the bucket bandwidth monitor out of the
process static into the per-instance InstanceContext.

- InstanceContext gains `bucket_monitor: OnceLock<Arc<Monitor>>` with
  `init_bucket_monitor(num_nodes)` (set-once; returns false if already set) and
  `bucket_monitor() -> Option<..>`.
- global.rs `init_global_bucket_monitor` / `get_global_bucket_monitor` keep
  their signatures and route through the current instance's context; the static
  is removed. The ignore-and-warn-on-reinit behavior is preserved (the warn
  stays in global.rs).

Tests: the monitor is None until initialized, set-once (second init is a no-op),
and independent across instances.

Verification: cargo test -p rustfs-ecstore (21 instance-context tests green),
cargo clippy -p rustfs-ecstore --all-targets (clean), make pre-commit (pass).

Refs: backlog#939 (Phase 5, Slice 10).

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<Arc<dyn LockClient>> = OnceLock::new();
     pub static ref GLOBAL_LOCK_CLIENTS: OnceLock<HashMap<String, Arc<dyn LockClient>>> = OnceLock::new();
-    pub static ref GLOBAL_BUCKET_MONITOR: OnceLock<Arc<Monitor>> = 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<Arc<Monitor>> {
-    GLOBAL_BUCKET_MONITOR.get().cloned()
+    current_ctx().bucket_monitor()
 }
 
 // Startup-owned process globals intentionally fail fast on duplicate writes.

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<Arc<TransitionState>>,
+    /// 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<Arc<Monitor>>,
 }
 
 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<Arc<Monitor>> {
+        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");
+    }
 }