refactor(ecstore): migrate the local disk registry into InstanceContext (Phase 5 Slice 12) (#4518)

Phase 5 Slice 12 (backlog#939): move the local disk registry — the
path->disk map, the disk-id->endpoint map, and the pool/set/drive layout —
out of the three GLOBAL_LOCAL_DISK_* process statics into the per-instance
InstanceContext.

This is the migration the owned-guard prerequisite (#4501) unblocked.

- InstanceContext gains three `Arc<RwLock<..>>` registry fields, constructed
  eagerly (empty), plus pub(crate) handle accessors.
- The three runtime-source handle functions (`local_disk_map_handle` etc.) now
  return the current instance's registry via `current_ctx()`; every direct
  `GLOBAL_LOCAL_DISK_*` access in runtime::sources (~15 sites) routes through
  those handles instead. Accessors that hold a write guard across `.await`
  (`record_local_disks`, `initialize_local_disk_maps`, `replace_local_disk_id`)
  bind the handle first, so the same locks are held across the same awaits.
- The three statics are removed; the test-reset helper clears the current
  context's registry.

Construction window: the registry is populated during storage startup, before
the ECStore exists, via the bootstrap context that `ECStore::new` then adopts —
so startup writes and post-construction reads share one registry (no
split-brain). Single-instance behavior is byte-for-byte unchanged; the
GLOBAL_LOCAL_DISK_* boundary arch guard passes with the statics gone.

Verification: cargo test -p rustfs-ecstore (32 disk/instance tests green) and
-p rustfs-heal (201 green), cargo clippy -p rustfs-ecstore -p rustfs-heal
--all-targets (clean), make pre-commit (pass).

Refs: backlog#939 (Phase 5, Slice 12; disk-registry migration)

crates/ecstore/src/runtime/global.rs

@@ -53,9 +53,6 @@ pub const DISK_RESERVE_FRACTION: f64 = 0.15;
 lazy_static! {
     static ref GLOBAL_RUSTFS_PORT: OnceLock<u16> = OnceLock::new();
     pub static ref GLOBAL_OBJECT_API: OnceLock<Arc<ECStore>> = OnceLock::new();
-    pub static ref GLOBAL_LOCAL_DISK_MAP: Arc<RwLock<HashMap<String, Option<DiskStore>>>> = Arc::new(RwLock::new(HashMap::new()));
-    pub static ref GLOBAL_LOCAL_DISK_ID_MAP: Arc<RwLock<HashMap<Uuid, String>>> = Arc::new(RwLock::new(HashMap::new()));
-    pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc<RwLock<TypeLocalDiskSetDrives>> = Arc::new(RwLock::new(Vec::new()));
     pub static ref GLOBAL_ROOT_DISK_THRESHOLD: RwLock<u64> = RwLock::new(0);
     pub static ref GLOBAL_LIFECYCLE_SYS: Arc<LifecycleSys> = LifecycleSys::new();
     pub static ref GLOBAL_BOOT_TIME: OnceCell<SystemTime> = OnceCell::new();
@@ -158,9 +155,10 @@ pub fn get_global_endpoints_opt() -> Option<EndpointServerPools> {
 
 #[cfg(test)]
 pub async fn reset_local_disk_test_state() {
-    GLOBAL_LOCAL_DISK_MAP.write().await.clear();
-    GLOBAL_LOCAL_DISK_ID_MAP.write().await.clear();
-    GLOBAL_LOCAL_DISK_SET_DRIVES.write().await.clear();
+    let ctx = current_ctx();
+    ctx.local_disk_map().write().await.clear();
+    ctx.local_disk_id_map().write().await.clear();
+    ctx.local_disk_set_drives().write().await.clear();
 }
 
 pub async fn is_first_cluster_node_local() -> bool {

crates/ecstore/src/runtime/instance.rs

@@ -39,14 +39,17 @@
 //! startup writes and post-construction reads hit the same cell and
 //! single-instance behavior is byte-for-byte unchanged.
 
+use super::global::TypeLocalDiskSetDrives;
 use crate::bucket::bandwidth::monitor::Monitor;
 use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState};
 use crate::bucket::replication::{DynReplicationPool, ReplicationStats};
+use crate::disk::DiskStore;
 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::collections::HashMap;
 use std::sync::{Arc, OnceLock};
 use tokio::sync::{OnceCell, RwLock};
 use uuid::Uuid;
@@ -125,6 +128,16 @@ pub struct InstanceContext {
     /// Async set-once, built with the live storage during init. Replaces the
     /// process-global static.
     replication_pool: OnceCell<Arc<DynReplicationPool>>,
+    /// This instance's local disk registry (Phase 5 Slice 12, backlog#939).
+    ///
+    /// Populated during storage startup (before the `ECStore` exists) via the
+    /// bootstrap context that the store then adopts, so startup writes and later
+    /// reads share one registry. Replaces the three `GLOBAL_LOCAL_DISK_*`
+    /// statics: the path→disk map, the disk-id→endpoint map, and the
+    /// pool/set/drive layout.
+    local_disk_map: Arc<RwLock<HashMap<String, Option<DiskStore>>>>,
+    local_disk_id_map: Arc<RwLock<HashMap<Uuid, String>>>,
+    local_disk_set_drives: Arc<RwLock<TypeLocalDiskSetDrives>>,
 }
 
 impl InstanceContext {
@@ -154,6 +167,9 @@ impl InstanceContext {
             bucket_monitor: OnceLock::new(),
             replication_stats: OnceCell::new(),
             replication_pool: OnceCell::new(),
+            local_disk_map: Arc::new(RwLock::new(HashMap::new())),
+            local_disk_id_map: Arc::new(RwLock::new(HashMap::new())),
+            local_disk_set_drives: Arc::new(RwLock::new(Vec::new())),
         }
     }
 
@@ -270,6 +286,21 @@ impl InstanceContext {
         &self.replication_pool
     }
 
+    /// Handle to this instance's local path→disk map.
+    pub(crate) fn local_disk_map(&self) -> Arc<RwLock<HashMap<String, Option<DiskStore>>>> {
+        self.local_disk_map.clone()
+    }
+
+    /// Handle to this instance's local disk-id→endpoint map.
+    pub(crate) fn local_disk_id_map(&self) -> Arc<RwLock<HashMap<Uuid, String>>> {
+        self.local_disk_id_map.clone()
+    }
+
+    /// Handle to this instance's local pool/set/drive layout.
+    pub(crate) fn local_disk_set_drives(&self) -> Arc<RwLock<TypeLocalDiskSetDrives>> {
+        self.local_disk_set_drives.clone()
+    }
+
     /// Update this instance's erasure setup type.
     pub async fn update_erasure_type(&self, setup_type: SetupType) {
         *self.erasure_kind.write().await = setup_type;

crates/ecstore/src/runtime/sources.rs

@@ -29,13 +29,12 @@ use crate::{
     error::Result,
     layout::endpoints::{EndpointServerPools, SetupType},
     runtime::global::{
-        GLOBAL_BOOT_TIME, GLOBAL_LIFECYCLE_SYS, GLOBAL_LOCAL_DISK_ID_MAP, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES,
-        GLOBAL_LOCAL_NODE_NAME_FALLBACK, GLOBAL_ROOT_DISK_THRESHOLD, TypeLocalDiskSetDrives,
-        get_background_services_cancel_token, get_global_bucket_monitor, get_global_deployment_id, get_global_endpoints,
-        get_global_endpoints_opt, get_global_lock_client, get_global_lock_clients, get_global_region, get_global_tier_config_mgr,
-        global_rustfs_port, init_global_bucket_monitor, is_dist_erasure, is_erasure, is_erasure_sd, is_first_cluster_node_local,
-        resolve_object_store_handle, set_global_deployment_id, set_global_lock_client, set_global_lock_clients, set_object_layer,
-        update_erasure_type,
+        GLOBAL_BOOT_TIME, GLOBAL_LIFECYCLE_SYS, GLOBAL_LOCAL_NODE_NAME_FALLBACK, GLOBAL_ROOT_DISK_THRESHOLD,
+        TypeLocalDiskSetDrives, get_background_services_cancel_token, get_global_bucket_monitor, get_global_deployment_id,
+        get_global_endpoints, get_global_endpoints_opt, get_global_lock_client, get_global_lock_clients, get_global_region,
+        get_global_tier_config_mgr, global_rustfs_port, init_global_bucket_monitor, is_dist_erasure, is_erasure, is_erasure_sd,
+        is_first_cluster_node_local, resolve_object_store_handle, set_global_deployment_id, set_global_lock_client,
+        set_global_lock_clients, set_object_layer, update_erasure_type,
     },
     services::batch_processor::{GlobalBatchProcessors, get_global_processors},
     services::event_notification::EventNotifier,
@@ -371,7 +370,7 @@ pub fn bucket_monitor() -> Option<Arc<Monitor>> {
 /// `InstanceContext` (Phase 5 disk-registry migration, backlog#939) without a
 /// cross-crate signature change. Single-instance behavior is unchanged.
 pub async fn local_disk_map_read() -> OwnedRwLockReadGuard<HashMap<String, Option<DiskStore>>> {
-    GLOBAL_LOCAL_DISK_MAP.clone().read_owned().await
+    local_disk_map_handle().read_owned().await
 }
 
 pub(crate) fn init_bucket_monitor_for_current_endpoints() {
@@ -380,15 +379,15 @@ pub(crate) fn init_bucket_monitor_for_current_endpoints() {
 }
 
 pub(crate) fn local_disk_map_handle() -> Arc<RwLock<HashMap<String, Option<DiskStore>>>> {
-    GLOBAL_LOCAL_DISK_MAP.clone()
+    crate::runtime::global::current_ctx().local_disk_map()
 }
 
 pub(crate) fn local_disk_id_map_handle() -> Arc<RwLock<HashMap<Uuid, String>>> {
-    GLOBAL_LOCAL_DISK_ID_MAP.clone()
+    crate::runtime::global::current_ctx().local_disk_id_map()
 }
 
 pub(crate) fn local_disk_set_drives_handle() -> Arc<RwLock<TypeLocalDiskSetDrives>> {
-    GLOBAL_LOCAL_DISK_SET_DRIVES.clone()
+    crate::runtime::global::current_ctx().local_disk_set_drives()
 }
 
 pub(crate) fn tier_config_mgr_handle() -> Arc<RwLock<TierConfigMgr>> {
@@ -408,24 +407,25 @@ pub(crate) fn event_notifier_handle() -> Arc<RwLock<EventNotifier>> {
 }
 
 pub(crate) async fn local_disk_by_path(path: &str) -> Option<DiskStore> {
-    GLOBAL_LOCAL_DISK_MAP.read().await.get(path).cloned().flatten()
+    local_disk_map_handle().read().await.get(path).cloned().flatten()
 }
 
 pub(crate) async fn local_disk_path_by_id(disk_id: &Uuid) -> Option<String> {
-    GLOBAL_LOCAL_DISK_ID_MAP.read().await.get(disk_id).cloned()
+    local_disk_id_map_handle().read().await.get(disk_id).cloned()
 }
 
 #[cfg(test)]
 pub(crate) async fn clear_local_disk_id_map_for_test() {
-    GLOBAL_LOCAL_DISK_ID_MAP.write().await.clear();
+    local_disk_id_map_handle().write().await.clear();
 }
 
 pub(crate) async fn record_local_disk_id(disk_id: Uuid, endpoint: String) {
-    GLOBAL_LOCAL_DISK_ID_MAP.write().await.insert(disk_id, endpoint);
+    local_disk_id_map_handle().write().await.insert(disk_id, endpoint);
 }
 
 pub(crate) async fn replace_local_disk_id(previous: Option<Uuid>, current: Option<Uuid>, endpoint: String) {
-    let mut disk_id_map = GLOBAL_LOCAL_DISK_ID_MAP.write().await;
+    let id_map = local_disk_id_map_handle();
+    let mut disk_id_map = id_map.write().await;
     if let Some(previous_id) = previous {
         disk_id_map.remove(&previous_id);
     }
@@ -435,7 +435,8 @@ pub(crate) async fn replace_local_disk_id(previous: Option<Uuid>, current: Optio
 }
 
 pub(crate) async fn record_local_disks(disks: Vec<DiskStore>) {
-    let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await;
+    let map = local_disk_map_handle();
+    let mut global_local_disk_map = map.write().await;
     for disk in disks {
         let path = disk.endpoint().to_string();
         global_local_disk_map.insert(path, Some(disk.clone()));
@@ -443,13 +444,14 @@ pub(crate) async fn record_local_disks(disks: Vec<DiskStore>) {
 }
 
 pub(crate) async fn local_disk_set_drive(pool_idx: usize, set_idx: usize, disk_idx: usize) -> Option<DiskStore> {
-    GLOBAL_LOCAL_DISK_SET_DRIVES.read().await[pool_idx][set_idx][disk_idx].clone()
+    local_disk_set_drives_handle().read().await[pool_idx][set_idx][disk_idx].clone()
 }
 
 pub(crate) async fn local_disk_for_endpoint(endpoint: &Endpoint) -> Option<DiskStore> {
-    let global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.read().await;
+    let set_drives = local_disk_set_drives_handle();
+    let global_set_drives = set_drives.read().await;
     if global_set_drives.is_empty() {
-        return GLOBAL_LOCAL_DISK_MAP
+        return local_disk_map_handle()
             .read()
             .await
             .get(&endpoint.to_string())
@@ -470,11 +472,11 @@ pub(crate) async fn local_disk_for_endpoint(endpoint: &Endpoint) -> Option<DiskS
 }
 
 pub(crate) async fn local_disk_paths() -> Vec<String> {
-    GLOBAL_LOCAL_DISK_MAP.read().await.keys().cloned().collect()
+    local_disk_map_handle().read().await.keys().cloned().collect()
 }
 
 pub(crate) async fn local_disks() -> Vec<DiskStore> {
-    GLOBAL_LOCAL_DISK_MAP
+    local_disk_map_handle()
         .read()
         .await
         .values()
@@ -483,11 +485,12 @@ pub(crate) async fn local_disks() -> Vec<DiskStore> {
 }
 
 pub(crate) async fn local_disk_entries() -> Vec<Option<DiskStore>> {
-    GLOBAL_LOCAL_DISK_MAP.read().await.values().cloned().collect()
+    local_disk_map_handle().read().await.values().cloned().collect()
 }
 
 pub(crate) async fn initialize_local_disk_maps(endpoint_pools: EndpointServerPools, opt: &DiskOption) -> Result<()> {
-    let mut global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.write().await;
+    let set_drives = local_disk_set_drives_handle();
+    let mut global_set_drives = set_drives.write().await;
     for pool_eps in endpoint_pools.as_ref().iter() {
         let mut set_count_drives = Vec::with_capacity(pool_eps.set_count);
         for _ in 0..pool_eps.set_count {
@@ -497,7 +500,8 @@ pub(crate) async fn initialize_local_disk_maps(endpoint_pools: EndpointServerPoo
         global_set_drives.push(set_count_drives);
     }
 
-    let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await;
+    let map = local_disk_map_handle();
+    let mut global_local_disk_map = map.write().await;
 
     for pool_eps in endpoint_pools.as_ref().iter() {
         for ep in pool_eps.endpoints.as_ref().iter() {