diff --git a/crates/ecstore/src/admin_server_info.rs b/crates/ecstore/src/admin_server_info.rs index b8c7553ad..240347c53 100644 --- a/crates/ecstore/src/admin_server_info.rs +++ b/crates/ecstore/src/admin_server_info.rs @@ -15,14 +15,10 @@ use crate::data_usage::{DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, load_data_usage_from_backend}; use crate::error::{Error, Result}; use crate::rpc::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client}; -use crate::{ - disk::endpoint::Endpoint, - global::{GLOBAL_BOOT_TIME, GLOBAL_Endpoints, get_global_deployment_id, resolve_object_store_handle}, - notification_sys::get_global_notification_sys, -}; +use crate::{disk::endpoint::Endpoint, runtime_sources}; use crate::data_usage::load_data_usage_cache; -use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, heal_channel::DriveState}; +use rustfs_common::heal_channel::DriveState; use rustfs_madmin::{ BackendDisks, Disk, ErasureSetInfo, ITEM_INITIALIZING, ITEM_OFFLINE, ITEM_ONLINE, InfoMessage, ServerProperties, }; @@ -33,7 +29,7 @@ use rustfs_protos::{ use rustfs_storage_api::StorageAdminApi; use std::{ collections::{HashMap, HashSet}, - time::{Duration, SystemTime}, + time::Duration, }; use time::OffsetDateTime; use tokio::time::timeout; @@ -127,11 +123,11 @@ async fn is_server_resolvable(endpoint: &Endpoint) -> Result<()> { } pub async fn get_local_server_property() -> ServerProperties { - let addr = GLOBAL_LOCAL_NODE_NAME.read().await.clone(); + let addr = runtime_sources::local_node_name().await; let mut pool_numbers = HashSet::new(); let mut network = HashMap::new(); - let endpoints = match GLOBAL_Endpoints.get() { + let endpoints = match runtime_sources::endpoint_pools() { Some(eps) => eps, None => return ServerProperties::default(), }; @@ -161,11 +157,7 @@ pub async fn get_local_server_property() -> ServerProperties { let mut props = ServerProperties { endpoint: addr, - uptime: GLOBAL_BOOT_TIME - .get() - .and_then(|boot_time| SystemTime::now().duration_since(*boot_time).ok()) - .unwrap_or_default() - .as_secs(), + uptime: runtime_sources::boot_uptime_secs(), network, version: get_commit_id(), ..Default::default() @@ -184,7 +176,7 @@ pub async fn get_local_server_property() -> ServerProperties { // let mut sensitive = HashSet::new(); // sensitive.insert(rustfs_config::ENV_RUSTFS_ACCESS_KEY.to_string()); // sensitive.insert(rustfs_config::ENV_RUSTFS_SECRET_KEY.to_string()); - if let Some(store) = resolve_object_store_handle() { + if let Some(store) = runtime_sources::object_store_handle() { let storage_info = StorageAdminApi::local_storage_info(store.as_ref()).await; props.state = ITEM_ONLINE.to_string(); props.disks = storage_info.disks; @@ -207,7 +199,7 @@ pub async fn get_server_info(get_pools: bool) -> InfoMessage { warn!("get_local_server_property end {:?}", after1 - nowt); let mut servers = { - if let Some(sys) = get_global_notification_sys() { + if let Some(sys) = runtime_sources::notification_sys() { sys.server_info().await } else { vec![] @@ -228,7 +220,7 @@ pub async fn get_server_info(get_pools: bool) -> InfoMessage { let mut backend = rustfs_madmin::ErasureBackend::default(); let mut pools: HashMap> = HashMap::new(); - if let Some(store) = resolve_object_store_handle() { + if let Some(store) = runtime_sources::object_store_handle() { mode = ITEM_ONLINE; match load_data_usage_from_backend(store.clone()).await { Ok(res) => { @@ -290,7 +282,7 @@ pub async fn get_server_info(get_pools: bool) -> InfoMessage { domain: None, region: None, sqs_arn: None, - deployment_id: get_global_deployment_id(), + deployment_id: runtime_sources::deployment_id(), buckets: Some(buckets), objects: Some(objects), versions: Some(versions), @@ -346,7 +338,7 @@ fn get_online_offline_disks_stats(disks_info: &[Disk]) -> (BackendDisks, Backend } async fn get_pools_info(all_disks: &[Disk]) -> Result>> { - let Some(store) = resolve_object_store_handle() else { + let Some(store) = runtime_sources::object_store_handle() else { return Err(Error::other("ServerNotInitialized")); }; @@ -397,14 +389,14 @@ pub fn get_commit_id() -> String { mod tests { use serial_test::serial; - use crate::global::get_global_deployment_id; + use crate::runtime_sources; use super::get_server_info; #[serial] #[tokio::test] async fn server_info_includes_global_deployment_id() { - let expected_deployment_id = get_global_deployment_id(); + let expected_deployment_id = runtime_sources::deployment_id(); let info = get_server_info(false).await; assert_eq!(info.deployment_id, expected_deployment_id); diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 072cee9f6..06a5b0879 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -28,6 +28,7 @@ use crate::config::com::{read_config, save_config}; use crate::disk::BUCKET_META_PREFIX; use crate::error::Error as EcstoreError; use crate::object_api::{ObjectInfo, ObjectOptions}; +use crate::runtime_sources; use crate::storage_api_contracts::EcstoreObjectIO; use lazy_static::lazy_static; use rustfs_filemeta::MrfOpKind; @@ -1399,12 +1400,11 @@ pub async fn init_background_replication(storage: Arc) }) .await; - assert!(GLOBAL_REPLICATION_STATS.get().is_some()); - assert!(GLOBAL_REPLICATION_POOL.get().is_some()); + assert!(runtime_sources::replication_runtime_initialized()); } pub fn get_global_replication_pool() -> Option> { - GLOBAL_REPLICATION_POOL.get().cloned() + runtime_sources::replication_pool() } pub async fn schedule_replication( @@ -1454,17 +1454,17 @@ pub async fn schedule_replication( } if dsc.is_synchronous() { replicate_object(ri, o).await - } else if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { + } else if let Some(pool) = runtime_sources::replication_pool() { let _ = pool.queue_replica_task(ri).await; } } pub async fn schedule_replication_delete(dv: DeletedObjectReplicationInfo) { - if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { + if let Some(pool) = runtime_sources::replication_pool() { let _ = pool.queue_replica_delete_task(dv.clone()).await; } - if let (Some(rs), Some(stats)) = (dv.delete_object.replication_state, GLOBAL_REPLICATION_STATS.get()) { + if let (Some(rs), Some(stats)) = (dv.delete_object.replication_state, runtime_sources::replication_stats()) { for (k, _v) in rs.targets.iter() { let ri = ReplicatedTargetInfo { arn: k.clone(), @@ -1600,7 +1600,7 @@ pub async fn queue_replication_heal_internal( || roi.version_purge_status == VersionPurgeStatusType::Failed || roi.version_purge_status == VersionPurgeStatusType::Pending { - let admission = if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { + let admission = if let Some(pool) = runtime_sources::replication_pool() { pool.queue_replica_delete_task(dv).await } else { ReplicationQueueAdmission::Missed @@ -1636,7 +1636,7 @@ pub async fn queue_replication_heal_internal( match roi.replication_status { ReplicationStatusType::Pending | ReplicationStatusType::Failed => { roi.event_type = REPLICATE_HEAL.to_string(); - let admission = if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { + let admission = if let Some(pool) = runtime_sources::replication_pool() { pool.queue_replica_task(roi.clone()).await } else { ReplicationQueueAdmission::Missed @@ -1651,7 +1651,7 @@ pub async fn queue_replication_heal_internal( if roi.existing_obj_resync.must_resync() { roi.event_type = REPLICATE_EXISTING.to_string(); - let admission = if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { + let admission = if let Some(pool) = runtime_sources::replication_pool() { pool.queue_replica_task(roi.clone()).await } else { ReplicationQueueAdmission::Missed @@ -1679,7 +1679,7 @@ async fn queue_replicate_deletes_wrapper( let mut dv = doi.clone(); dv.reset_id = v.reset_id.clone(); dv.target_arn = k.clone(); - let target_admission = if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { + let target_admission = if let Some(pool) = runtime_sources::replication_pool() { pool.queue_replica_delete_task(dv).await } else { ReplicationQueueAdmission::Missed diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 1bea41dfb..a6f05bd51 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -19,7 +19,6 @@ use crate::bucket::bucket_target_sys::{ use crate::bucket::metadata_sys; use crate::bucket::msgp_decode::{read_msgp_ext8_time, skip_msgp_value, write_msgp_time}; use crate::bucket::replication::ResyncStatusType; -use crate::bucket::replication::replication_pool::GLOBAL_REPLICATION_STATS; use crate::bucket::replication::{ObjectOpts, ReplicationConfigurationExt as _}; use crate::bucket::tagging::decode_tags_to_map; use crate::bucket::target::BucketTargets; @@ -29,10 +28,10 @@ use crate::config::com::save_config; use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; use crate::error::{Error, Result, is_err_object_not_found, is_err_version_not_found}; use crate::event_notification::{EventArgs, send_event}; -use crate::global::GLOBAL_LocalNodeName; use crate::global::get_global_bucket_monitor; use crate::global::resolve_object_store_handle; use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; +use crate::runtime_sources; use crate::set_disk::get_lock_acquire_timeout; use crate::storage_api_contracts::{EcstoreObjectIO, EcstoreObjectOperations}; use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; @@ -226,7 +225,7 @@ fn is_head_proxy_failure(err: &SdkError) -> bool { } async fn record_proxy_request(bucket: &str, api: &str, is_err: bool) { - if let Some(stats) = GLOBAL_REPLICATION_STATS.get() { + if let Some(stats) = runtime_sources::replication_stats() { stats.inc_proxy(bucket, api, is_err).await; } } @@ -1857,7 +1856,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); @@ -1884,7 +1883,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); return; @@ -1983,7 +1982,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); return; @@ -2016,7 +2015,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); return; @@ -2047,7 +2046,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); return; @@ -2096,7 +2095,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); continue; @@ -2189,7 +2188,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat ) }; - if let Some(stats) = GLOBAL_REPLICATION_STATS.get() { + if let Some(stats) = runtime_sources::replication_stats() { for tgt in rinfos.targets.iter() { if tgt.replication_status != tgt.prev_replication_status { stats @@ -2348,7 +2347,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); return; @@ -2372,7 +2371,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); return; @@ -2404,7 +2403,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); return; @@ -2433,7 +2432,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); return; @@ -2471,7 +2470,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); continue; @@ -2501,7 +2500,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); return; @@ -2544,7 +2543,7 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted ..Default::default() }, user_agent: "Internal: [Replication]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); } @@ -2738,7 +2737,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: roi.to_object_info(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -2758,7 +2757,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: roi.to_object_info(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -2797,7 +2796,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: roi.to_object_info(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -2821,7 +2820,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: roi.to_object_info(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -2846,7 +2845,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: roi.to_object_info(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -2889,7 +2888,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: roi.to_object_info(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -2916,7 +2915,7 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s object_info = u; } - if let Some(stats) = GLOBAL_REPLICATION_STATS.get() { + if let Some(stats) = runtime_sources::replication_stats() { for tgt in &rinfos.targets { if tgt.replication_status != tgt.prev_replication_status { stats @@ -2937,14 +2936,14 @@ pub async fn replicate_object(roi: ReplicateObjectInfo, s event_name, bucket_name: bucket.clone(), object: object_info, - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); if rinfos.replication_status() != ReplicationStatusType::Completed && roi.replication_status_internal == rinfos.replication_status_internal() - && let Some(stats) = GLOBAL_REPLICATION_STATS.get() + && let Some(stats) = runtime_sources::replication_stats() { for tgt in &rinfos.targets { if tgt.replication_status != tgt.prev_replication_status { @@ -3005,7 +3004,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: self.to_object_info(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3045,7 +3044,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: self.to_object_info(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3076,7 +3075,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: object_info, - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3098,7 +3097,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: object_info, - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3193,7 +3192,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: object_info, - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3292,7 +3291,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: self.to_object_info(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3331,7 +3330,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: self.to_object_info(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3371,7 +3370,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: object_info, - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3395,7 +3394,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: object_info, - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3458,7 +3457,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: object_info.clone(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3516,7 +3515,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: object_info, - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3543,7 +3542,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: object_info, - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); @@ -3579,7 +3578,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { event_name: EventName::ObjectReplicationNotTracked.to_string(), bucket_name: bucket.clone(), object: object_info, - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), user_agent: "Internal: [Replication]".to_string(), ..Default::default() }); diff --git a/crates/ecstore/src/data_usage.rs b/crates/ecstore/src/data_usage.rs index 76ec25302..915f42a67 100644 --- a/crates/ecstore/src/data_usage.rs +++ b/crates/ecstore/src/data_usage.rs @@ -19,6 +19,7 @@ use crate::{ config::com::read_config, disk::DiskAPI, error::{Error, classify_system_path_failure_reason}, + runtime_sources, store::ECStore, }; pub use local_snapshot::{LocalUsageSnapshot, read_snapshot as read_local_snapshot, snapshot_path}; @@ -529,7 +530,7 @@ async fn update_usage_cache_if_needed() { let cache_clone = (*memory_cache()).clone(); let updating_clone = (*cache_updating()).clone(); tokio::spawn(async move { - if let Some(store) = crate::global::GLOBAL_OBJECT_API.get() + if let Some(store) = runtime_sources::object_store_handle() && let Ok(data_usage_info) = load_data_usage_from_backend(store.clone()).await { let mut cache = cache_clone.write().await; @@ -560,7 +561,7 @@ async fn update_usage_cache_if_needed() { *updating = true; drop(updating); - if let Some(store) = crate::global::GLOBAL_OBJECT_API.get() + if let Some(store) = runtime_sources::object_store_handle() && let Ok(data_usage_info) = load_data_usage_from_backend(store.clone()).await { let mut cache = memory_cache().write().await; @@ -629,7 +630,7 @@ pub async fn apply_bucket_usage_memory_overlay(data_usage_info: &mut DataUsageIn /// Sync memory cache with backend data (called by scanner) pub async fn sync_memory_cache_with_backend() -> Result<(), Error> { - if let Some(store) = crate::global::GLOBAL_OBJECT_API.get() { + if let Some(store) = runtime_sources::object_store_handle() { match load_data_usage_from_backend(store.clone()).await { Ok(data_usage_info) => { replace_bucket_usage_memory_from_info(&data_usage_info).await; @@ -789,10 +790,9 @@ pub async fn load_data_usage_cache(store: &crate::set_disk::SetDisks, name: &str pub async fn save_data_usage_cache(cache: &DataUsageCache, name: &str) -> crate::error::Result<()> { use crate::config::com::save_config; use crate::disk::BUCKET_META_PREFIX; - use crate::global::resolve_object_store_handle; use std::path::Path; - let Some(store) = resolve_object_store_handle() else { + let Some(store) = runtime_sources::object_store_handle() else { return Err(Error::other("errServerNotInitialized")); }; let buf = cache.marshal_msg().map_err(Error::other)?; diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index def933476..fe7822ae5 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -23,7 +23,7 @@ use crate::disk::{ }, local::{LocalDisk, ScanGuard}, }; -use crate::global::GLOBAL_LOCAL_DISK_ID_MAP; +use crate::runtime_sources; use bytes::Bytes; use metrics::counter; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; @@ -962,13 +962,7 @@ impl LocalDiskWrapper { drop(disk_id); if self.disk.is_local() { - let mut disk_id_map = GLOBAL_LOCAL_DISK_ID_MAP.write().await; - if let Some(previous_id) = previous { - disk_id_map.remove(&previous_id); - } - if let Some(current_id) = id { - disk_id_map.insert(current_id, self.disk.endpoint().to_string()); - } + runtime_sources::replace_local_disk_id(previous, id, self.disk.endpoint().to_string()).await; } Ok(()) } diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 9a118e426..0ce4bd645 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -29,7 +29,7 @@ use crate::disk::{ os::{check_path_length, is_empty_dir, is_root_disk, rename_all, rename_all_ignore_missing_source}, }; use crate::erasure_coding::bitrot_verify; -use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold}; +use crate::runtime_sources; use bytes::Bytes; use metrics::counter; use parking_lot::RwLock as ParkingLotRwLock; @@ -3721,8 +3721,7 @@ async fn get_disk_info(drive_path: PathBuf) -> Result<(rustfs_utils::os::DiskInf let disk_info = get_info(&drive_path).inspect_err(|err| { log_startup_disk_io_error("get_disk_info_stat", Path::new(&drive_path), err); })?; - let root_drive = if !*GLOBAL_IsErasureSD.read().await { - let root_disk_threshold = *GLOBAL_RootDiskThreshold.read().await; + let root_drive = if let Some(root_disk_threshold) = runtime_sources::root_disk_threshold_for_erasure_disk().await { if root_disk_threshold > 0 { disk_info.total <= root_disk_threshold } else { diff --git a/crates/ecstore/src/metrics_realtime.rs b/crates/ecstore/src/metrics_realtime.rs index e496c6ae1..40553915d 100644 --- a/crates/ecstore/src/metrics_realtime.rs +++ b/crates/ecstore/src/metrics_realtime.rs @@ -13,9 +13,9 @@ // limitations under the License. use crate::admin_server_info::get_local_server_property; -use crate::global::resolve_object_store_handle; +use crate::runtime_sources; use chrono::Utc; -use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR, heal_channel::DriveState, metrics::global_metrics}; +use rustfs_common::{heal_channel::DriveState, metrics::global_metrics}; use rustfs_io_metrics::internode_metrics::global_internode_metrics; use rustfs_madmin::metrics::{ DiskIOStats, DiskMetric, LastMinute as MadminLastMinute, NetDevLine, NetMetrics, RPCMetrics, RealtimeMetrics, @@ -364,7 +364,7 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts) return real_time_metrics; } - let mut by_host_name = GLOBAL_RUSTFS_ADDR.read().await.clone(); + let mut by_host_name = runtime_sources::rustfs_addr().await; if !opts.hosts.is_empty() { let server = get_local_server_property().await; if opts.hosts.contains(&server.endpoint) { @@ -373,7 +373,7 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts) return real_time_metrics; } } - let local_node_name = GLOBAL_LOCAL_NODE_NAME.read().await.clone(); + let local_node_name = runtime_sources::local_node_name().await; if by_host_name.starts_with(":") && !local_node_name.starts_with(":") { by_host_name = local_node_name; } @@ -395,7 +395,7 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts) if types.contains(&MetricType::SCANNER) { debug!("start get scanner metrics"); let mut metrics = global_metrics().report().await; - if let Some(init_time) = rustfs_common::get_global_init_time().await { + if let Some(init_time) = runtime_sources::scanner_init_time().await { metrics.current_started = init_time; } real_time_metrics.aggregated.scanner = Some(to_madmin_scanner_metrics(metrics)); @@ -461,7 +461,7 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts) } async fn collect_local_disks_metrics(disks: &HashSet) -> HashMap { - let store = match resolve_object_store_handle() { + let store = match runtime_sources::object_store_handle() { Some(store) => store, None => return HashMap::new(), }; diff --git a/crates/ecstore/src/pools.rs b/crates/ecstore/src/pools.rs index 736a68f1d..cffc5470b 100644 --- a/crates/ecstore/src/pools.rs +++ b/crates/ecstore/src/pools.rs @@ -37,11 +37,11 @@ use crate::error::{ is_err_version_not_found, }; use crate::global::resolve_object_store_handle; -use crate::notification_sys::get_global_notification_sys; use crate::object_api::{GetObjectReader, ObjectOptions}; use crate::rebalance::{REBAL_META_NAME, RebalanceMeta, is_rebalance_conflicting_with_decommission}; +use crate::runtime_sources; use crate::set_disk::{SetDisks, get_lock_acquire_timeout}; -use crate::{global::GLOBAL_LifecycleSys, sets::Sets, store::ECStore}; +use crate::{sets::Sets, store::ECStore}; use byteorder::{ByteOrder, LittleEndian, WriteBytesExt}; use futures::{StreamExt, future::BoxFuture, stream::FuturesUnordered}; use http::HeaderMap; @@ -2051,7 +2051,7 @@ impl ECStore { return Err(err); } - if should_reload_pool_meta && let Some(notification_sys) = get_global_notification_sys() { + if should_reload_pool_meta && let Some(notification_sys) = runtime_sources::notification_sys() { let stage = format!("decommission_cancel for pool {idx}"); resolve_decommission_pool_meta_reload_result(notification_sys.reload_pool_meta().await, stage.as_str())?; } @@ -2083,7 +2083,7 @@ impl ECStore { return Err(err); } - if should_reload_pool_meta && let Some(notification_sys) = get_global_notification_sys() { + if should_reload_pool_meta && let Some(notification_sys) = runtime_sources::notification_sys() { let stage = format!("clear_decommission for pool {idx}"); resolve_decommission_pool_meta_reload_result(notification_sys.reload_pool_meta().await, stage.as_str())?; } @@ -2099,7 +2099,7 @@ impl ECStore { if promoted { self.save_current_pool_meta().await?; - if let Some(notification_sys) = get_global_notification_sys() { + if let Some(notification_sys) = runtime_sources::notification_sys() { let stage = format!("promote_queued_decommission for pool {idx}"); resolve_decommission_pool_meta_reload_result(notification_sys.reload_pool_meta().await, stage.as_str())?; } @@ -2613,7 +2613,7 @@ impl ECStore { } else { let mut pool_meta = self.pool_meta.write().await; pool_meta.mark_decommission_progress_saved(); - if let Some(notification_sys) = get_global_notification_sys() + if let Some(notification_sys) = runtime_sources::notification_sys() && let Err(err) = resolve_decommission_entry_reload_result( notification_sys.reload_pool_meta().await, bucket.as_str(), @@ -2664,7 +2664,7 @@ impl ECStore { "versioning", BucketVersioningSys::get(&bi.name).await, )?; - lifecycle_config = GLOBAL_LifecycleSys.get(&bi.name).await; + lifecycle_config = runtime_sources::bucket_lifecycle_config(&bi.name).await; lock_retention = BucketObjectLockSys::get(&bi.name).await; replication_config = resolve_decommission_optional_bucket_config_result( &bi.name, @@ -3105,7 +3105,7 @@ impl ECStore { let mut pool_meta = self.pool_meta.write().await; pool_meta.mark_decommission_progress_saved(); } - if let Some(notification_sys) = get_global_notification_sys() { + if let Some(notification_sys) = runtime_sources::notification_sys() { let stage = format!("decommission_failed for pool {idx}"); if let Some(err) = observe_decommission_terminal_reload_result( resolve_decommission_pool_meta_reload_result(notification_sys.reload_pool_meta().await, stage.as_str()), @@ -3155,7 +3155,7 @@ impl ECStore { let mut pool_meta = self.pool_meta.write().await; pool_meta.mark_decommission_progress_saved(); } - if let Some(notification_sys) = get_global_notification_sys() { + if let Some(notification_sys) = runtime_sources::notification_sys() { let stage = format!("complete_decommission for pool {idx}"); if let Some(err) = observe_decommission_terminal_reload_result( resolve_decommission_pool_meta_reload_result(notification_sys.reload_pool_meta().await, stage.as_str()), @@ -3339,7 +3339,7 @@ impl ECStore { .save_current_pool_meta_for_decommission_start(&indices, space_infos, decom_buckets) .await?; - if let Some(notification_sys) = get_global_notification_sys() + if let Some(notification_sys) = runtime_sources::notification_sys() && let Err(err) = resolve_start_decommission_pool_meta_reload_result(notification_sys.reload_pool_meta().await) { warn!( @@ -3438,7 +3438,7 @@ impl ECStore { let mut lock_retention = None; let mut replication_config = None; if bucket_info.name != RUSTFS_META_BUCKET { - lifecycle_config = GLOBAL_LifecycleSys.get(&bucket_info.name).await; + lifecycle_config = runtime_sources::bucket_lifecycle_config(&bucket_info.name).await; lock_retention = BucketObjectLockSys::get(&bucket_info.name).await; replication_config = resolve_decommission_optional_bucket_config_result( &bucket_info.name, diff --git a/crates/ecstore/src/rpc/client.rs b/crates/ecstore/src/rpc/client.rs index be121b525..16f4999bb 100644 --- a/crates/ecstore/src/rpc/client.rs +++ b/crates/ecstore/src/rpc/client.rs @@ -14,8 +14,8 @@ use crate::disk::error::{DiskError, Error as DiskErrorType}; use crate::rpc::{TONIC_RPC_PREFIX, gen_signature_headers}; +use crate::runtime_sources; use http::Method; -use rustfs_common::GLOBAL_CONN_MAP; use rustfs_protos::{create_new_channel, proto_gen::node_service::node_service_client::NodeServiceClient}; use std::{error::Error, io::ErrorKind}; use tonic::{service::interceptor::InterceptedService, transport::Channel}; @@ -30,7 +30,7 @@ pub async fn node_service_time_out_client( interceptor: TonicInterceptor, ) -> Result>, Box> { // Try to get cached channel - let cached_channel = { GLOBAL_CONN_MAP.read().await.get(addr).cloned() }; + let cached_channel = runtime_sources::cached_node_channel(addr).await; let channel = match cached_channel { Some(channel) => { diff --git a/crates/ecstore/src/rpc/peer_s3_client.rs b/crates/ecstore/src/rpc/peer_s3_client.rs index e5bae079c..3b9cffe4d 100644 --- a/crates/ecstore/src/rpc/peer_s3_client.rs +++ b/crates/ecstore/src/rpc/peer_s3_client.rs @@ -17,10 +17,10 @@ use crate::disk::error::DiskError; use crate::disk::error::{Error, Result}; use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs}; use crate::disk::{DiskAPI, DiskStore, disk_store::get_max_timeout_duration}; -use crate::global::GLOBAL_LOCAL_DISK_MAP; use crate::rpc::client::{ TonicInterceptor, gen_tonic_signature_interceptor, is_network_like_disk_error, node_service_time_out_client, }; +use crate::runtime_sources; use crate::store::all_local_disk; use crate::store_utils::is_reserved_or_invalid_bucket; use crate::{ @@ -1092,7 +1092,7 @@ pub(crate) async fn heal_bucket_local_on_disks( } async fn clone_drives() -> Vec> { - GLOBAL_LOCAL_DISK_MAP.read().await.values().cloned().collect::>() + runtime_sources::local_disk_entries().await } #[cfg(test)] diff --git a/crates/ecstore/src/runtime_sources.rs b/crates/ecstore/src/runtime_sources.rs index 5ed470386..1e8188946 100644 --- a/crates/ecstore/src/runtime_sources.rs +++ b/crates/ecstore/src/runtime_sources.rs @@ -12,11 +12,34 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::sync::Arc; +use std::{collections::HashMap, sync::Arc, time::SystemTime}; -use crate::{config::get_global_storage_class, global::get_global_deployment_id}; +use crate::bucket::bandwidth::monitor::Monitor; +use crate::disk::endpoint::Endpoint; +use crate::{ + bucket::replication::{DynReplicationPool, GLOBAL_REPLICATION_POOL, GLOBAL_REPLICATION_STATS, ReplicationStats}, + config::get_global_storage_class, + disk::{DiskAPI, DiskOption, DiskStore, new_disk}, + endpoints::EndpointServerPools, + error::Result, + event_notification::EventNotifier, + global::{ + GLOBAL_BOOT_TIME, GLOBAL_EventNotifier, GLOBAL_IsErasureSD, GLOBAL_LOCAL_DISK_ID_MAP, GLOBAL_LOCAL_DISK_MAP, + GLOBAL_LOCAL_DISK_SET_DRIVES, GLOBAL_LifecycleSys, GLOBAL_LocalNodeName, GLOBAL_RootDiskThreshold, GLOBAL_TierConfigMgr, + TypeLocalDiskSetDrives, get_global_bucket_monitor, get_global_deployment_id, get_global_endpoints, + get_global_endpoints_opt, init_global_bucket_monitor, resolve_object_store_handle, set_global_deployment_id, + }, + notification_sys::{NotificationSys, get_global_notification_sys}, + store::ECStore, + tier::tier::TierConfigMgr, +}; +use rustfs_common::{GLOBAL_CONN_MAP, GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR}; use rustfs_io_metrics::internode_metrics::global_internode_metrics; use rustfs_kms::{ObjectEncryptionService, get_global_encryption_service}; +use s3s::dto::BucketLifecycleConfiguration; +use tokio::sync::RwLock; +use tonic::transport::Channel; +use uuid::Uuid; pub(crate) fn record_erasure_write_quorum_failure(stage: &'static str, dominant_error: &'static str) { global_internode_metrics().record_erasure_write_quorum_failure(stage, dominant_error); @@ -26,10 +49,66 @@ pub(crate) async fn object_encryption_service() -> Option Option> { + resolve_object_store_handle() +} + +pub(crate) fn endpoint_pools() -> Option { + get_global_endpoints_opt() +} + +pub(crate) async fn local_node_name() -> String { + GLOBAL_LOCAL_NODE_NAME.read().await.clone() +} + +pub(crate) fn default_local_node_name() -> String { + GLOBAL_LocalNodeName.to_string() +} + +pub(crate) async fn rustfs_addr() -> String { + GLOBAL_RUSTFS_ADDR.read().await.clone() +} + +pub(crate) fn boot_uptime_secs() -> u64 { + GLOBAL_BOOT_TIME + .get() + .and_then(|boot_time| SystemTime::now().duration_since(*boot_time).ok()) + .unwrap_or_default() + .as_secs() +} + +pub(crate) async fn scanner_init_time() -> Option> { + rustfs_common::get_global_init_time().await +} + +pub(crate) async fn root_disk_threshold_for_erasure_disk() -> Option { + if *GLOBAL_IsErasureSD.read().await { + None + } else { + Some(*GLOBAL_RootDiskThreshold.read().await) + } +} + +pub(crate) async fn cached_node_channel(addr: &str) -> Option { + GLOBAL_CONN_MAP.read().await.get(addr).cloned() +} + pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option { get_global_storage_class().and_then(|sc| sc.get_parity_for_sc(storage_class.unwrap_or_default())) } +pub(crate) fn backend_storage_class_parities(default_standard_parity: usize) -> (Option, Option) { + if let Some(sc) = get_global_storage_class() { + let standard = sc + .get_parity_for_sc(crate::config::storageclass::CLASS_STANDARD) + .or(Some(default_standard_parity)); + let reduced_redundancy = sc.get_parity_for_sc(crate::config::storageclass::RRS); + (standard, reduced_redundancy) + } else { + (Some(default_standard_parity), None) + } +} + pub(crate) fn storage_class_should_inline(shard_size: i64, versioned: bool) -> bool { get_global_storage_class().is_some_and(|sc| sc.should_inline(shard_size, versioned)) } @@ -39,6 +118,188 @@ pub(crate) fn deployment_upload_id(upload_id: &str) -> String { .encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_id).as_bytes()) } +pub(crate) fn deployment_id() -> Option { + get_global_deployment_id() +} + +pub(crate) fn replication_pool() -> Option> { + GLOBAL_REPLICATION_POOL.get().cloned() +} + +pub(crate) fn replication_stats() -> Option> { + GLOBAL_REPLICATION_STATS.get().cloned() +} + +pub(crate) fn replication_runtime_initialized() -> bool { + GLOBAL_REPLICATION_STATS.get().is_some() && GLOBAL_REPLICATION_POOL.get().is_some() +} + +pub(crate) fn ensure_deployment_id(deployment_id: Uuid) { + if get_global_deployment_id().is_none() { + set_global_deployment_id(deployment_id); + } +} + pub(crate) fn global_lock_manager() -> Arc { rustfs_lock::get_global_lock_manager() } + +pub(crate) fn notification_sys() -> Option<&'static NotificationSys> { + get_global_notification_sys() +} + +pub(crate) async fn bucket_lifecycle_config(bucket: &str) -> Option { + GLOBAL_LifecycleSys.get(bucket).await +} + +pub(crate) fn delete_bucket_monitor_entry(bucket: &str) { + if let Some(monitor) = get_global_bucket_monitor() { + monitor.delete_bucket(bucket); + } +} + +pub(crate) fn bucket_monitor() -> Option> { + get_global_bucket_monitor() +} + +pub(crate) fn init_bucket_monitor_for_current_endpoints() { + let num_nodes = get_global_endpoints().get_nodes().len().try_into().unwrap_or(u64::MAX); + init_global_bucket_monitor(num_nodes); +} + +pub(crate) fn local_disk_map_handle() -> Arc>>> { + GLOBAL_LOCAL_DISK_MAP.clone() +} + +pub(crate) fn local_disk_id_map_handle() -> Arc>> { + GLOBAL_LOCAL_DISK_ID_MAP.clone() +} + +pub(crate) fn local_disk_set_drives_handle() -> Arc> { + GLOBAL_LOCAL_DISK_SET_DRIVES.clone() +} + +pub(crate) fn tier_config_mgr_handle() -> Arc> { + GLOBAL_TierConfigMgr.clone() +} + +pub(crate) fn event_notifier_handle() -> Arc> { + GLOBAL_EventNotifier.clone() +} + +pub(crate) async fn local_disk_by_path(path: &str) -> Option { + GLOBAL_LOCAL_DISK_MAP.read().await.get(path).cloned().flatten() +} + +pub(crate) async fn local_disk_path_by_id(disk_id: &Uuid) -> Option { + GLOBAL_LOCAL_DISK_ID_MAP.read().await.get(disk_id).cloned() +} + +pub(crate) async fn record_local_disk_id(disk_id: Uuid, endpoint: String) { + GLOBAL_LOCAL_DISK_ID_MAP.write().await.insert(disk_id, endpoint); +} + +pub(crate) async fn replace_local_disk_id(previous: Option, current: Option, endpoint: String) { + let mut disk_id_map = GLOBAL_LOCAL_DISK_ID_MAP.write().await; + if let Some(previous_id) = previous { + disk_id_map.remove(&previous_id); + } + if let Some(current_id) = current { + disk_id_map.insert(current_id, endpoint); + } +} + +pub(crate) async fn record_local_disks(disks: Vec) { + let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await; + for disk in disks { + let path = disk.endpoint().to_string(); + global_local_disk_map.insert(path, Some(disk.clone())); + } +} + +pub(crate) async fn local_disk_set_drive(pool_idx: usize, set_idx: usize, disk_idx: usize) -> Option { + GLOBAL_LOCAL_DISK_SET_DRIVES.read().await[pool_idx][set_idx][disk_idx].clone() +} + +pub(crate) async fn local_disk_for_endpoint(endpoint: &Endpoint) -> Option { + let global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.read().await; + if global_set_drives.is_empty() { + return GLOBAL_LOCAL_DISK_MAP + .read() + .await + .get(&endpoint.to_string()) + .cloned() + .unwrap_or(None); + } + + let pool_idx = usize::try_from(endpoint.pool_idx).ok()?; + let set_idx = usize::try_from(endpoint.set_idx).ok()?; + let disk_idx = usize::try_from(endpoint.disk_idx).ok()?; + + global_set_drives + .get(pool_idx) + .and_then(|sets| sets.get(set_idx)) + .and_then(|disks| disks.get(disk_idx)) + .cloned() + .unwrap_or(None) +} + +pub(crate) async fn local_disk_paths() -> Vec { + GLOBAL_LOCAL_DISK_MAP.read().await.keys().cloned().collect() +} + +pub(crate) async fn local_disks() -> Vec { + GLOBAL_LOCAL_DISK_MAP + .read() + .await + .values() + .filter_map(|v| v.as_ref().cloned()) + .collect() +} + +pub(crate) async fn local_disk_entries() -> Vec> { + GLOBAL_LOCAL_DISK_MAP.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; + 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 { + set_count_drives.push(vec![None; pool_eps.drives_per_set]); + } + + global_set_drives.push(set_count_drives); + } + + let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await; + + for pool_eps in endpoint_pools.as_ref().iter() { + for ep in pool_eps.endpoints.as_ref().iter() { + if !ep.is_local { + continue; + } + + let disk = new_disk(ep, opt).await?; + let path = disk.endpoint().to_string(); + let pool_idx = usize::try_from(ep.pool_idx).map_err(|err| { + crate::error::Error::other(format!("store init failed to convert pool index `{}`: {err}", ep.pool_idx)) + })?; + let set_idx = usize::try_from(ep.set_idx).map_err(|err| { + crate::error::Error::other(format!("store init failed to convert set index `{}`: {err}", ep.set_idx)) + })?; + let disk_idx = usize::try_from(ep.disk_idx).map_err(|err| { + crate::error::Error::other(format!("store init failed to convert disk index `{}`: {err}", ep.disk_idx)) + })?; + + global_local_disk_map.insert(path, Some(disk.clone())); + global_set_drives[pool_idx][set_idx][disk_idx] = Some(disk.clone()); + } + } + + Ok(()) +} + +pub(crate) async fn init_tier_config_mgr(store: Arc) -> Result<()> { + GLOBAL_TierConfigMgr.write().await.init(store).await +} diff --git a/crates/ecstore/src/sets.rs b/crates/ecstore/src/sets.rs index e362ddd52..03b96a92a 100644 --- a/crates/ecstore/src/sets.rs +++ b/crates/ecstore/src/sets.rs @@ -25,8 +25,9 @@ use crate::{ }, endpoints::{Endpoints, PoolEndpoints}, error::StorageError, - global::{GLOBAL_LOCAL_DISK_SET_DRIVES, get_global_lock_clients, is_dist_erasure}, + global::{get_global_lock_clients, is_dist_erasure}, object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}, + runtime_sources, set_disk::SetDisks, store_init::{check_format_erasure_values, get_format_erasure_in_quorum, load_format_erasure_all, save_format_file}, }; @@ -138,10 +139,7 @@ impl Sets { } if disk.as_ref().unwrap().is_local() && is_dist_erasure().await { - let local_disk = { - let local_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.read().await; - local_set_drives[pool_idx][i][j].clone() - }; + let local_disk = runtime_sources::local_disk_set_drive(pool_idx, i, j).await; if local_disk.is_none() { warn!("sets new set_drive {}-{} local_disk is none", i, j); diff --git a/crates/ecstore/src/store.rs b/crates/ecstore/src/store.rs index c52edebf9..9a616d9df 100644 --- a/crates/ecstore/src/store.rs +++ b/crates/ecstore/src/store.rs @@ -44,9 +44,8 @@ use crate::error::{ }; use crate::event_notification::EventNotifier; use crate::global::{ - DISK_RESERVE_FRACTION, GLOBAL_BOOT_TIME, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, TypeLocalDiskSetDrives, - get_global_deployment_id, get_global_endpoints, get_global_region, get_global_tier_config_mgr, init_global_bucket_monitor, - set_global_deployment_id, set_object_layer, + DISK_RESERVE_FRACTION, GLOBAL_BOOT_TIME, TypeLocalDiskSetDrives, get_global_endpoints, get_global_region, + get_global_tier_config_mgr, set_object_layer, }; use crate::notification_sys::get_global_notification_sys; use crate::pools::PoolMeta; @@ -56,7 +55,7 @@ use crate::store_init::{check_disk_fatal_errs, ec_drives_no_config}; use crate::tier::tier::TierConfigMgr; use crate::{ bucket::{lifecycle::bucket_lifecycle_ops::TransitionState, metadata::BucketMetadata}, - disk::{BUCKET_META_PREFIX, DiskOption, DiskStore, RUSTFS_META_BUCKET, new_disk}, + disk::{BUCKET_META_PREFIX, DiskOption, DiskStore, RUSTFS_META_BUCKET}, endpoints::EndpointServerPools, object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}, rpc::S3PeerSys, @@ -805,7 +804,7 @@ mod tests { #[tokio::test] async fn test_find_local_disk() { - let result = peer::find_local_disk(&"/nonexistent/path".to_string()).await; + let result = peer::find_local_disk("/nonexistent/path").await; assert!(result.is_none(), "Should return None for nonexistent path"); } diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index 8dea47bef..0af0a32e0 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -17,7 +17,7 @@ use crate::bucket::{ metadata::{BUCKET_TABLE_RESERVED_PREFIX, table_bucket_catalog_metadata_prefix}, utils::is_meta_bucketname, }; -use crate::global::get_global_bucket_monitor; +use crate::runtime_sources; use crate::set_disk::get_lock_acquire_timeout; use rustfs_storage_api::NamespaceLocking as _; @@ -258,9 +258,7 @@ impl ECStore { for prefix in bucket_delete_metadata_cleanup_prefixes(bucket) { self.delete_all(RUSTFS_META_BUCKET, prefix.as_str()).await?; } - if let Some(monitor) = get_global_bucket_monitor() { - monitor.delete_bucket(bucket); - } + runtime_sources::delete_bucket_monitor_entry(bucket); Ok(()) } } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index b4f02eb92..b9319557e 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -14,11 +14,9 @@ use super::*; use crate::error::is_err_decommission_running; -use crate::global::{ - GLOBAL_EventNotifier, GLOBAL_LOCAL_DISK_ID_MAP, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, GLOBAL_TierConfigMgr, - get_global_bucket_monitor, is_dist_erasure, is_first_cluster_node_local, -}; +use crate::global::{is_dist_erasure, is_first_cluster_node_local}; use crate::pools::local_decommission_queue_prefix; +use crate::runtime_sources; use tracing::{debug, error, info, warn}; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; @@ -302,11 +300,7 @@ impl ECStore { // Replace the local disk if !is_dist_erasure().await { - let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await; - for disk in local_disks { - let path = disk.endpoint().to_string(); - global_local_disk_map.insert(path, Some(disk.clone())); - } + runtime_sources::record_local_disks(local_disks).await; } let peer_sys = S3PeerSys::new(&endpoint_pools); @@ -325,19 +319,17 @@ impl ECStore { start_gate: tokio::sync::Mutex::new(()), pool_meta_save_gate: tokio::sync::Mutex::new(()), - local_disk_map: GLOBAL_LOCAL_DISK_MAP.clone(), - local_disk_id_map: GLOBAL_LOCAL_DISK_ID_MAP.clone(), - local_disk_set_drives: GLOBAL_LOCAL_DISK_SET_DRIVES.clone(), - tier_config_mgr: GLOBAL_TierConfigMgr.clone(), - event_notifier: GLOBAL_EventNotifier.clone(), + local_disk_map: runtime_sources::local_disk_map_handle(), + local_disk_id_map: runtime_sources::local_disk_id_map_handle(), + local_disk_set_drives: runtime_sources::local_disk_set_drives_handle(), + tier_config_mgr: runtime_sources::tier_config_mgr_handle(), + event_notifier: runtime_sources::event_notifier_handle(), bucket_monitor: OnceLock::new(), }); // Only set it when the global deployment ID is not yet configured - if let Some(dep_id) = deployment_id - && get_global_deployment_id().is_none() - { - set_global_deployment_id(dep_id); + if let Some(dep_id) = deployment_id { + runtime_sources::ensure_deployment_id(dep_id); } let wait_sec = 5; @@ -362,7 +354,7 @@ impl ECStore { set_object_layer(ec.clone()).await; - if let Some(monitor) = get_global_bucket_monitor() { + if let Some(monitor) = runtime_sources::bucket_monitor() { let _ = ec.bucket_monitor.set(monitor); } @@ -449,8 +441,7 @@ impl ECStore { }); } - let num_nodes = get_global_endpoints().get_nodes().len() as u64; - init_global_bucket_monitor(num_nodes); + runtime_sources::init_bucket_monitor_for_current_endpoints(); init_background_expiry(self.clone()).await; crate::bucket::lifecycle::bucket_lifecycle_ops::init_background_stale_multipart_upload_cleanup(self.clone()); @@ -458,7 +449,7 @@ impl ECStore { TransitionState::init(self.clone()).await; crate::tier::tier::try_migrate_tiering_config(self.clone()).await; - if let Err(err) = GLOBAL_TierConfigMgr.write().await.init(self.clone()).await { + if let Err(err) = runtime_sources::init_tier_config_mgr(self.clone()).await { info!("TierConfigMgr init error: {}", err); } diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs index 9e9378b9e..4c119d113 100644 --- a/crates/ecstore/src/store/peer.rs +++ b/crates/ecstore/src/store/peer.rs @@ -13,7 +13,7 @@ // limitations under the License. use super::*; -use crate::global::GLOBAL_LOCAL_DISK_ID_MAP; +use crate::runtime_sources; use tracing::{debug, error}; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; @@ -23,25 +23,16 @@ const EVENT_LOCK_CLIENT_INITIALIZATION_FAILED: &str = "lock_client_initializatio async fn remember_local_disk_id(disk: &DiskStore) -> Option { let disk_id = disk.get_disk_id().await.ok().flatten()?; - GLOBAL_LOCAL_DISK_ID_MAP - .write() - .await - .insert(disk_id, disk.endpoint().to_string()); + runtime_sources::record_local_disk_id(disk_id, disk.endpoint().to_string()).await; Some(disk_id) } -pub async fn find_local_disk(disk_path: &String) -> Option { - let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await; - - if let Some(disk) = disk_map.get(disk_path) { - disk.as_ref().cloned() - } else { - None - } +pub async fn find_local_disk(disk_path: &str) -> Option { + runtime_sources::local_disk_by_path(disk_path).await } pub async fn find_local_disk_by_ref(disk_ref: &str) -> Option { - if let Some(disk) = find_local_disk(&disk_ref.to_string()).await { + if let Some(disk) = find_local_disk(disk_ref).await { let _ = remember_local_disk_id(&disk).await; return Some(disk); } @@ -50,7 +41,7 @@ pub async fn find_local_disk_by_ref(disk_ref: &str) -> Option { return None; }; - if let Some(disk_path) = GLOBAL_LOCAL_DISK_ID_MAP.read().await.get(&disk_id).cloned() + if let Some(disk_path) = runtime_sources::local_disk_path_by_id(&disk_id).await && let Some(disk) = find_local_disk(&disk_path).await { return Some(disk); @@ -66,35 +57,15 @@ pub async fn find_local_disk_by_ref(disk_ref: &str) -> Option { } pub async fn get_disk_via_endpoint(endpoint: &Endpoint) -> Option { - let global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.read().await; - if global_set_drives.is_empty() { - return GLOBAL_LOCAL_DISK_MAP - .read() - .await - .get(&endpoint.to_string()) - .cloned() - .unwrap_or(None); - } - global_set_drives - .get(endpoint.pool_idx as usize) - .and_then(|sets| sets.get(endpoint.set_idx as usize)) - .and_then(|disks| disks.get(endpoint.disk_idx as usize)) - .cloned() - .unwrap_or(None) + runtime_sources::local_disk_for_endpoint(endpoint).await } pub async fn all_local_disk_path() -> Vec { - let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await; - disk_map.keys().cloned().collect() + runtime_sources::local_disk_paths().await } pub async fn all_local_disk() -> Vec { - let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await; - disk_map - .values() - .filter(|v| v.is_some()) - .map(|v| v.as_ref().unwrap().clone()) - .collect() + runtime_sources::local_disks().await } pub async fn prewarm_local_disk_id_map() { @@ -121,35 +92,7 @@ pub async fn init_local_disks(endpoint_pools: EndpointServerPools) -> Result<()> health_check: true, }; - let mut global_set_drives = GLOBAL_LOCAL_DISK_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 { - set_count_drives.push(vec![None; pool_eps.drives_per_set]); - } - - global_set_drives.push(set_count_drives); - } - - let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await; - - for pool_eps in endpoint_pools.as_ref().iter() { - for ep in pool_eps.endpoints.as_ref().iter() { - if !ep.is_local { - continue; - } - - let disk = new_disk(ep, opt).await?; - - let path = disk.endpoint().to_string(); - - global_local_disk_map.insert(path, Some(disk.clone())); - - global_set_drives[ep.pool_idx as usize][ep.set_idx as usize][ep.disk_idx as usize] = Some(disk.clone()); - } - } - - Ok(()) + runtime_sources::initialize_local_disk_maps(endpoint_pools, opt).await } pub fn init_lock_clients(endpoint_pools: EndpointServerPools) { diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs index a13b48206..437812f22 100644 --- a/crates/ecstore/src/store/rebalance.rs +++ b/crates/ecstore/src/store/rebalance.rs @@ -13,8 +13,8 @@ // limitations under the License. use super::*; -use crate::config::get_global_storage_class; use crate::layout::pool_space::{ServerPoolsAvailableSpace, build_server_pools_available_space}; +use crate::runtime_sources; use rustfs_storage_api::{NamespaceLocking as _, ObjectOperations as _, StorageAdminApi}; pub(in crate::store) mod support; use support::{ @@ -509,19 +509,8 @@ impl ECStore { #[instrument(skip(self))] pub(super) async fn handle_backend_info(&self) -> rustfs_madmin::BackendInfo { - let (standard_sc_parity, rr_sc_parity) = { - if let Some(sc) = get_global_storage_class() { - let sc_parity = sc - .get_parity_for_sc(storageclass::CLASS_STANDARD) - .or(Some(self.pools[0].default_parity_count)); - - let rrs_sc_parity = sc.get_parity_for_sc(storageclass::RRS); - - (sc_parity, rrs_sc_parity) - } else { - (Some(self.pools[0].default_parity_count), None) - } - }; + let (standard_sc_parity, rr_sc_parity) = + runtime_sources::backend_storage_class_parities(self.pools[0].default_parity_count); let mut standard_sc_data = Vec::new(); let mut rr_sc_data = Vec::new(); @@ -555,7 +544,7 @@ impl ECStore { #[instrument(skip(self))] pub(super) async fn handle_storage_info(&self) -> rustfs_madmin::StorageInfo { - let Some(notification_sy) = get_global_notification_sys() else { + let Some(notification_sy) = runtime_sources::notification_sys() else { return rustfs_madmin::StorageInfo::default(); }; diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index d83be5ccb..b05b0bf2f 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,9 +5,9 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Current Context - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) -- Branch: `overtrue/arch-ecstore-data-plane-runtime-sources` -- Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095/API-096/API-097/API-098/API-099/API-100/API-101/API-102/API-103/API-104/API-105/API-106/API-107/API-108/API-109/API-110/API-111/API-112/API-113/API-114/API-115/API-116/API-117/API-118/API-119/API-120/API-121/API-122/API-123/API-124/API-125/API-126/API-127/API-128/API-129/API-130/API-131/API-132/API-133/API-134/API-135/API-136/API-137/API-138/API-139/API-140/API-141/API-142/API-143/API-144/API-145/API-146/API-147/API-148/API-149/API-150/API-151/API-152/API-153/API-154/API-155/API-156/API-157/API-158/API-159/API-160/API-161/API-162/API-163/API-164/API-165/API-166/API-167/API-168/API-169/API-170/API-171/API-172/API-173/API-174/API-175/API-176/API-177/API-178/API-179/API-180/API-181/API-182/API-183/API-184/API-185/API-186`. -- Based on: latest `origin/main` after PR #3796 merged API-185. +- Branch: `overtrue/arch-ecstore-replication-runtime-sources` +- Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095/API-096/API-097/API-098/API-099/API-100/API-101/API-102/API-103/API-104/API-105/API-106/API-107/API-108/API-109/API-110/API-111/API-112/API-113/API-114/API-115/API-116/API-117/API-118/API-119/API-120/API-121/API-122/API-123/API-124/API-125/API-126/API-127/API-128/API-129/API-130/API-131/API-132/API-133/API-134/API-135/API-136/API-137/API-138/API-139/API-140/API-141/API-142/API-143/API-144/API-145/API-146/API-147/API-148/API-149/API-150/API-151/API-152/API-153/API-154/API-155/API-156/API-157/API-158/API-159/API-160/API-161/API-162/API-163/API-164/API-165/API-166/API-167/API-168/API-169/API-170/API-171/API-172/API-173/API-174/API-175/API-176/API-177/API-178/API-179/API-180/API-181/API-182/API-183/API-184/API-185/API-186/API-187/API-188/API-189`. +- Based on: stacked on API-188 branch while PR #3799 is pending. - PR type for this branch: `consumer-migration` - Runtime behavior changes: none. - Rust code changes: route replication pool, outbound TLS generation, runtime @@ -18,6 +18,9 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block runtime source reads, RIO HTTP reader TLS/metrics runtime source reads, and gRPC/transition network client TLS/metrics runtime source reads, plus ECStore data-plane KMS/storage-class/deployment-id/lock-manager/erasure metric reads, + plus ECStore observability/status object-store, endpoint, node-name, + boot-time, init-time, root-disk threshold, and cached RPC channel reads, + plus ECStore replication pool, replication stats, and event-host reads, through AppContext-first or owner-crate resolver boundaries. - CI/script changes: lock completed owner and test/fuzz boundaries against bare/glob imports, scattered raw ECStore facade subpaths, and startup @@ -27,7 +30,7 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block and storage owner thin bridge regressions, plus app context and notify event-bridge thin module regressions; accept the reviewed AppContext resolver reverse dependencies in the layer baseline. -- Docs changes: record the API-136 through API-186 owner facade cleanup. +- Docs changes: record the API-136 through API-189 owner facade cleanup. ## Phase 0 Tasks @@ -4700,6 +4703,50 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block source scan, Rust risk scan, branch freshness check, pre-commit quality gate, and three-expert review. +- [x] `API-187` Centralize ECStore runtime owner source reads. + - Do: expand the ECStore runtime-source boundary across rebalance storage + class and notification reads, bucket monitor cleanup, lifecycle config + lookups, local disk id/path/set-drive maps, peer disk discovery, and store + init runtime handles. + - Acceptance: ECStore data-plane, rebalance, lifecycle, bucket, peer, and + startup paths route those runtime globals through the ECStore-owned runtime + source module instead of importing them directly. + - Must preserve: rebalance parity selection and notifications, lifecycle + config lookup semantics, bucket monitor deletion, local disk id backfill, + endpoint disk lookup fallback, local disk map initialization, deployment id + publication, and tier config initialization. + - Verification: ECStore compile coverage, focused store/pools/set-disk tests, + formatting, migration guard, layer guard, diff hygiene, residual runtime + source scan, Rust risk scan, branch freshness check, pre-commit quality + gate, and three-expert review. + +- [x] `API-188` Centralize ECStore observability runtime source reads. + - Do: route ECStore server-info, realtime metrics, data-usage cache, + local-disk root detection, and RPC cached-channel runtime reads through the + ECStore-owned runtime-source module. + - Acceptance: ECStore observability/status paths no longer import those + runtime globals directly outside the owner runtime-source boundary. + - Must preserve: server info endpoint/uptime/deployment-id fields, storage + info and backend summary collection, realtime metrics host fallback, + scanner init-time override, data-usage cache refresh behavior, root-disk + threshold checks, and cached gRPC channel reuse. + - Verification: ECStore compile coverage, formatting, diff hygiene, residual + runtime source scan, Rust risk scan, focused tests, migration/layer guards, + PR-before-push pre-commit quality gate, and three-expert review. + +- [x] `API-189` Centralize ECStore replication runtime source reads. + - Do: route replication pool, replication stats, and replication event-host + runtime reads through the ECStore-owned runtime-source module. + - Acceptance: replication pool/resyncer code no longer reads those runtime + globals directly outside the owner runtime-source boundary. + - Must preserve: background replication initialization, async/sync queueing, + delete-task stats updates, proxy request stats, resync status updates, and + emitted replication event host values. + - Verification: ECStore compile coverage, formatting, diff hygiene, residual + replication runtime-source scan, Rust risk scan, focused tests, + migration/layer guards, PR-before-push pre-commit quality gate, and + three-expert review. + ## Next PRs 1. `consumer-migration`: continue reducing direct global reads behind AppContext resolver boundaries. @@ -4814,11 +4861,69 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Quality/architecture | pass | API-186 keeps ECStore data-plane runtime globals behind an ECStore-owned runtime-source module without widening public APIs. | | Migration preservation | pass | Erasure quorum metric labels, managed-KMS fallback, storage-class decisions, multipart upload id encoding, and lock-manager initialization keep existing behavior. | | Testing/verification | pass | ECStore compile/focused tests, formatting, residual data-plane runtime source scan, targeted guard checks, and pre-commit passed for API-186. | +| Quality/architecture | pass | API-187 expands the ECStore runtime-source owner boundary across rebalance, lifecycle, local disk maps, peer lookup, and store init handles without adding public APIs. | +| Migration preservation | pass | Rebalance notifications, lifecycle config reads, bucket monitor cleanup, local disk id/path/set-drive lookups, store init map publication, and deployment id publication keep existing semantics. | +| Testing/verification | pass | ECStore compile/focused tests, formatting, migration/layer guards, diff-only Rust risk scan, and pre-commit passed for API-187. | +| Quality/architecture | pass | API-188 keeps ECStore observability/status runtime reads behind the ECStore runtime-source boundary without adding public APIs. | +| Migration preservation | pass | Server info fields, metrics host fallback, data-usage cache refresh, root-disk checks, and cached RPC channel reuse keep existing semantics. | +| Testing/verification | pass | ECStore compile/focused tests, formatting, migration/layer guards, residual scan, diff-only Rust risk scan, and pre-commit passed for API-188. | +| Quality/architecture | pass | API-189 keeps ECStore replication runtime pool/stats/host reads behind the ECStore runtime-source boundary without adding public APIs. | +| Migration preservation | pass | Replication initialization, queueing, delete stats, proxy stats, resync status updates, and emitted event host values keep existing semantics. | +| Testing/verification | pass | ECStore compile/focused test, formatting, migration/layer guards, diff hygiene, residual scan, diff-only Rust risk scan, and pre-commit passed for API-189. | ## Verification Notes Passed before push: +- Issue #660 API-187 current slice: + - `cargo check -p rustfs-ecstore --tests`: passed. + - `cargo test -p rustfs-ecstore --lib test_find_local_disk_by_ref_backfills_uuid_map -- --test-threads=1`: + passed. + - `cargo test -p rustfs-ecstore --lib should_resume_local_decommission -- --test-threads=1`: + passed. + - `cargo test -p rustfs-ecstore --lib resolve_store_init_stage_result -- --test-threads=1`: + passed. + - `cargo test -p rustfs-ecstore --lib test_find_local_disk -- --test-threads=1`: + passed. + - `cargo fmt --all`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - ECStore runtime source scan: passed for API-187 targets; remaining + `set_disk` tier/local-node/global-map matches are legacy owner boundaries + intentionally left for a later focused slice. + - Rust risk scan: passed; diff adds no new `unwrap`, `expect`, `panic`, + `todo`, `unimplemented`, `unsafe`, production print, boxed public error, + string public error, relaxed ordering, or silent integer cast. + - Branch freshness check: rebased onto latest `origin/main` after PR #3797 + merged API-186. + - `make pre-commit`: passed. +- Issue #660 API-188 current slice: + - `cargo check -p rustfs-ecstore --tests`: passed. + - `cargo test -p rustfs-ecstore --lib server_info_includes_global_deployment_id -- --test-threads=1`: + passed. + - `cargo fmt --all`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - ECStore runtime source scan: passed for API-188 targets. + - Rust risk scan: passed. + - `make pre-commit`: passed. +- Issue #660 API-189 current slice: + - `cargo check -p rustfs-ecstore --tests`: passed. + - `cargo test -p rustfs-ecstore --lib replication_queue_admission_combines_target_results -- --test-threads=1`: + passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - Replication runtime-source scan: passed for API-189 targets. + - Diff-only Rust risk scan: passed. + - `make pre-commit`: passed, including 6552 nextest tests passed and + doctests passed; the existing OPA policy test took 603s. + - Issue #660 API-186 current slice: - `cargo check -p rustfs-ecstore --tests`: passed. - `cargo test -p rustfs-ecstore --lib erasure_coding -- --test-threads=1`: