diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index a2bc1732d..f0d325207 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -33,9 +33,8 @@ use crate::error::Error; use crate::error::StorageError; use crate::error::{error_resp_to_object_err, is_err_object_not_found, is_err_version_not_found, is_network_or_host_down}; use crate::event_notification::{EventArgs, send_event}; -use crate::global::GLOBAL_LocalNodeName; -use crate::global::{GLOBAL_LifecycleSys, GLOBAL_TierConfigMgr, get_global_deployment_id}; use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions}; +use crate::runtime_sources; use crate::set_disk::{MAX_PARTS_COUNT, RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY, SetDisks}; use crate::store::ECStore; use crate::tier::warm_backend::WarmBackendGetOpts; @@ -577,11 +576,12 @@ impl ExpiryState { } pub async fn resize_workers(n: usize, api: Arc) { - if n == GLOBAL_ExpiryState.read().await.tasks_tx.len() || n < 1 { + let expiry_state = runtime_sources::expiry_state_handle(); + if n == expiry_state.read().await.tasks_tx.len() || n < 1 { return; } - let mut state = GLOBAL_ExpiryState.write().await; + let mut state = expiry_state.write().await; while state.tasks_tx.len() < n { let (tx, rx) = mpsc::channel(EXPIRY_WORKER_QUEUE_CAPACITY); @@ -593,7 +593,7 @@ impl ExpiryState { state.stats.increment_workers(); tokio::spawn(async move { let mut rx = rx.lock().await; - //let mut expiry_state = GLOBAL_ExpiryState.read().await; + //let mut expiry_state = runtime_sources::expiry_state_handle().read().await; ExpiryState::worker(&mut rx, api, stats).await; }); } @@ -789,7 +789,8 @@ async fn enqueue_recovered_free_version_with_state(state: &Arc bool { - enqueue_recovered_free_version_with_state(&GLOBAL_ExpiryState, oi).await + let expiry_state = runtime_sources::expiry_state_handle(); + enqueue_recovered_free_version_with_state(&expiry_state, oi).await } struct TransitionTask { @@ -1114,6 +1115,7 @@ impl TransitionState { pub async fn init(api: Arc) { let (configured, absolute_max, n) = resolve_transition_worker_count(); + let transition_state = runtime_sources::transition_state_handle(); debug!( event = EVENT_LIFECYCLE_WORKER_STATE, component = LOG_COMPONENT_ECSTORE, @@ -1121,8 +1123,8 @@ impl TransitionState { configured_transition_workers = configured, absolute_max_workers = absolute_max, effective_transition_workers = n, - transition_queue_capacity = GLOBAL_TransitionState.transition_queue_capacity, - transition_queue_send_timeout_ms = GLOBAL_TransitionState.transition_queue_send_timeout.as_millis() as u64, + transition_queue_capacity = transition_state.transition_queue_capacity, + transition_queue_send_timeout_ms = transition_state.transition_queue_send_timeout.as_millis() as u64, state = "configured", "Lifecycle worker configuration resolved" ); @@ -1133,9 +1135,8 @@ impl TransitionState { } pub fn pending_tasks(&self) -> usize { - //let transition_rx = GLOBAL_TransitionState.transition_rx.lock().unwrap(); - let transition_rx = &GLOBAL_TransitionState.transition_rx; - transition_rx.len() + //let transition_rx = runtime_sources::transition_state_handle().transition_rx.lock().unwrap(); + self.transition_rx.len() } pub fn active_tasks(&self) -> i64 { @@ -1170,6 +1171,7 @@ impl TransitionState { } async fn worker_with_cancel(api: Arc, cancel_token: CancellationToken) { + let transition_state = runtime_sources::transition_state_handle(); loop { select! { biased; @@ -1177,7 +1179,7 @@ impl TransitionState { _ = cancel_token.cancelled() => { return; } - task = GLOBAL_TransitionState.transition_rx.recv() => { + task = transition_state.transition_rx.recv() => { if task.is_err() { break; } @@ -1191,8 +1193,8 @@ impl TransitionState { if task.as_any().is::() { let task = task.as_any().downcast_ref::().expect("TransitionTask downcast failed"); - TransitionState::inc_counter(&GLOBAL_TransitionState.active_tasks); - GLOBAL_TransitionState.record_scanner_transition_state(); + TransitionState::inc_counter(&transition_state.active_tasks); + transition_state.record_scanner_transition_state(); let obj_info_for_event = ObjectInfo { bucket: task.obj_info.bucket.clone(), @@ -1224,7 +1226,7 @@ impl TransitionState { bucket_name: obj_info_for_event.bucket.clone(), object: obj_info_for_event, user_agent: "Internal: [ILM-Transition]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); } else { @@ -1237,7 +1239,7 @@ impl TransitionState { if task.obj_info.is_latest { ts.num_objects = 1; } - GLOBAL_TransitionState.add_lastday_stats(&task.event.storage_class, ts); + transition_state.add_lastday_stats(&task.event.storage_class, ts); // Send s3:ObjectTransition:Complete event send_event(EventArgs { @@ -1245,12 +1247,12 @@ impl TransitionState { bucket_name: obj_info_for_event.bucket.clone(), object: obj_info_for_event, user_agent: "Internal: [ILM-Transition]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); } - TransitionState::add_counter(&GLOBAL_TransitionState.active_tasks, -1); - GLOBAL_TransitionState.record_scanner_transition_state(); + TransitionState::add_counter(&transition_state.active_tasks, -1); + transition_state.record_scanner_transition_state(); } } else => () @@ -1295,7 +1297,8 @@ impl TransitionState { fn resize_workers_to(api: Arc, n: i64, requested: i64, absolute_max: i64) { let target = n as usize; - let mut workers = GLOBAL_TransitionState.workers.lock().unwrap(); + let transition_state = runtime_sources::transition_state_handle(); + let mut workers = transition_state.workers.lock().unwrap(); let tracked_workers = workers.len(); workers.retain(|worker| !worker.handle.is_finished()); let pruned_finished_workers = tracked_workers.saturating_sub(workers.len()); @@ -1318,8 +1321,8 @@ impl TransitionState { } let current_workers = workers.len() as i64; - GLOBAL_TransitionState.num_workers.store(current_workers, Ordering::SeqCst); - GLOBAL_TransitionState.record_scanner_transition_state(); + transition_state.num_workers.store(current_workers, Ordering::SeqCst); + transition_state.record_scanner_transition_state(); debug!( event = EVENT_LIFECYCLE_WORKER_STATE, @@ -1388,7 +1391,8 @@ fn spawn_tier_free_version_recovery_once(api: Arc) { bucket_marker = stats.next_bucket_marker; object_marker = stats.next_object_marker; let (pending_tasks, active_tasks) = { - let state = GLOBAL_ExpiryState.read().await; + let expiry_state = runtime_sources::expiry_state_handle(); + let state = expiry_state.read().await; (state.pending_tasks(), state.stats.active_tasks()) }; debug!( @@ -1468,7 +1472,7 @@ fn stale_uploads_cleanup_interval() -> StdDuration { fn encode_stale_upload_id(upload_uuid: &str) -> String { base64_simd::URL_SAFE_NO_PAD - .encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_uuid).as_bytes()) + .encode_to_string(format!("{}.{}", runtime_sources::deployment_id().unwrap_or_default(), upload_uuid).as_bytes()) } fn initiated_from_upload_dir(upload_dir: &str, fallback: Option) -> OffsetDateTime { @@ -1859,7 +1863,10 @@ pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Resu if let Some(storage_class) = &transition.storage_class && storage_class.as_str() != "" { - let valid = GLOBAL_TierConfigMgr.read().await.is_tier_valid(storage_class.as_str()); + let valid = runtime_sources::tier_config_mgr_handle() + .read() + .await + .is_tier_valid(storage_class.as_str()); if !valid { return Err(std::io::Error::other(ERR_INVALID_STORAGECLASS)); } @@ -1871,7 +1878,10 @@ pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Resu if let Some(storage_class) = &noncurrent_version_transition.storage_class && storage_class.as_str() != "" { - let valid = GLOBAL_TierConfigMgr.read().await.is_tier_valid(storage_class.as_str()); + let valid = runtime_sources::tier_config_mgr_handle() + .read() + .await + .is_tier_valid(storage_class.as_str()); if !valid { return Err(std::io::Error::other(ERR_INVALID_STORAGECLASS)); } @@ -1900,13 +1910,13 @@ fn transitioned_cleanup_tuple(oi: &ObjectInfo) -> Result<(&str, &str, &str), std } pub async fn enqueue_transition_immediate(oi: &ObjectInfo, src: LcEventSrc) { - if let Some(lc) = GLOBAL_LifecycleSys.get(&oi.bucket).await { + if let Some(lc) = runtime_sources::bucket_lifecycle_config(&oi.bucket).await { enqueue_transition_with_lifecycle(oi, &lc, &src).await; } } pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) { - let Some(lifecycle) = GLOBAL_LifecycleSys.get(&oi.bucket).await else { + let Some(lifecycle) = runtime_sources::bucket_lifecycle_config(&oi.bucket).await else { return; }; let Some(api) = crate::global::resolve_object_store_handle() else { @@ -1995,7 +2005,8 @@ pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) { if !to_delete_objs.is_empty() && let Some(event) = noncurrent_event { - GLOBAL_ExpiryState + let expiry_state = runtime_sources::expiry_state_handle(); + expiry_state .write() .await .enqueue_by_newer_noncurrent(&oi.bucket, to_delete_objs, event, &src) @@ -2004,7 +2015,7 @@ pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) { } pub async fn enqueue_transition_for_existing_objects(api: Arc, bucket: &str) -> Result<(), Error> { - let Some(lc) = GLOBAL_LifecycleSys.get(bucket).await else { + let Some(lc) = runtime_sources::bucket_lifecycle_config(bucket).await else { return Ok(()); }; let mut marker = None; @@ -2117,7 +2128,9 @@ async fn enqueue_transition_with_lifecycle(oi: &ObjectInfo, lc: &BucketLifecycle if oi.delete_marker || oi.is_dir { return; } - GLOBAL_TransitionState.queue_transition_task(oi, &event, src).await; + runtime_sources::transition_state_handle() + .queue_transition_task(oi, &event, src) + .await; } _ => (), } @@ -2198,7 +2211,7 @@ pub async fn expire_transitioned_object( bucket_name: obj_info.bucket.clone(), object: obj_info, user_agent: "Internal: [ILM-Expiry]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); /*let system = match notification_system() { @@ -2218,7 +2231,7 @@ pub async fn expire_transitioned_object( pub fn gen_transition_objname(bucket: &str) -> Result { let us = Uuid::new_v4().to_string(); let mut hasher = Sha256::new(); - hasher.update(format!("{}/{}", get_global_deployment_id().unwrap_or_default(), bucket).as_bytes()); + hasher.update(format!("{}/{}", runtime_sources::deployment_id().unwrap_or_default(), bucket).as_bytes()); let hash = rustfs_utils::crypto::hex(hasher.finalize().as_slice()); let obj = format!("{}/{}/{}/{}", &hash[0..16], &us[0..2], &us[2..4], &us); Ok(obj) @@ -2275,7 +2288,8 @@ pub async fn get_transitioned_object_reader( oi: &ObjectInfo, opts: &ObjectOptions, ) -> Result { - let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await; + let tier_config_mgr = runtime_sources::tier_config_mgr_handle(); + let mut tier_config_mgr = tier_config_mgr.write().await; let tgt_client = match tier_config_mgr.get_driver(&oi.transitioned_object.tier).await { Ok(d) => d, Err(err) => return Err(std::io::Error::other(err)), @@ -2587,7 +2601,9 @@ pub async fn apply_transition_rule(event: &lifecycle::Event, src: &LcEventSrc, o if oi.delete_marker || oi.is_dir { return false; } - GLOBAL_TransitionState.queue_transition_task(oi, event, src).await + runtime_sources::transition_state_handle() + .queue_transition_task(oi, event, src) + .await } pub async fn apply_expiry_on_transitioned_object( @@ -2668,7 +2684,7 @@ pub async fn apply_expiry_on_non_transitioned_objects( bucket_name: dobj.bucket.clone(), object: dobj, user_agent: "Internal: [ILM-Expiry]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); @@ -2684,7 +2700,8 @@ pub async fn apply_expiry_on_non_transitioned_objects( } pub async fn apply_expiry_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool { - let mut expiry_state = GLOBAL_ExpiryState.write().await; + let expiry_state = runtime_sources::expiry_state_handle(); + let mut expiry_state = expiry_state.write().await; expiry_state.enqueue_by_days(oi, event, src).await } @@ -2846,8 +2863,8 @@ pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, mod tests { use super::{ DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS, DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX, - DEFAULT_TRANSITION_WORKERS_CAP, ExpiryState, GLOBAL_TransitionState, StaleMultipartUploadCandidate, TransitionState, - TransitionedObject, cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at, + DEFAULT_TRANSITION_WORKERS_CAP, ExpiryState, StaleMultipartUploadCandidate, TransitionState, TransitionedObject, + cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at, enqueue_recovered_free_version_with_state, lifecycle_deleted_object, lifecycle_rule_has_date_expiration, lifecycle_version_purge_state_from_completed_targets, mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, replication_state_for_delete, resolve_transition_queue_capacity, @@ -2864,6 +2881,7 @@ mod tests { use crate::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; use crate::error::is_err_invalid_upload_id; use crate::object_api::{ObjectInfo, ObjectOptions, PutObjReader}; + use crate::runtime_sources; use crate::set_disk::{RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY}; use crate::store::ECStore; use futures::FutureExt; @@ -3520,10 +3538,11 @@ mod tests { #[serial] async fn transition_state_init_honors_runtime_configured_worker_count() { let (_paths, ecstore) = setup_test_env().await; - let original_workers = GLOBAL_TransitionState.num_workers.load(Ordering::SeqCst); + let transition_state = runtime_sources::transition_state_handle(); + let original_workers = transition_state.num_workers.load(Ordering::SeqCst); with_transition_worker_env_async(Some("3"), Some("8"), || async { TransitionState::update_workers(ecstore.clone(), 0).await; - assert_eq!(GLOBAL_TransitionState.num_workers.load(Ordering::SeqCst), 3); + assert_eq!(transition_state.num_workers.load(Ordering::SeqCst), 3); }) .await; @@ -3535,26 +3554,27 @@ mod tests { #[serial] async fn transition_worker_resize_cancels_removed_workers_directly() { let (_paths, ecstore) = setup_test_env().await; - let original_workers = GLOBAL_TransitionState.num_workers.load(Ordering::SeqCst); + let transition_state = runtime_sources::transition_state_handle(); + let original_workers = transition_state.num_workers.load(Ordering::SeqCst); let absolute_max = resolve_transition_workers_absolute_max(); TransitionState::resize_workers_to(ecstore.clone(), 0, 0, absolute_max); - assert_eq!(GLOBAL_TransitionState.num_workers.load(Ordering::SeqCst), 0); + assert_eq!(transition_state.num_workers.load(Ordering::SeqCst), 0); TransitionState::resize_workers_to(ecstore.clone(), 2, 2, absolute_max); let worker_tokens = { - let workers = GLOBAL_TransitionState.workers.lock().unwrap(); + let workers = transition_state.workers.lock().unwrap(); assert_eq!(workers.len(), 2); workers.iter().map(|worker| worker.cancel.clone()).collect::>() }; TransitionState::resize_workers_to(ecstore.clone(), 1, 1, absolute_max); - assert_eq!(GLOBAL_TransitionState.num_workers.load(Ordering::SeqCst), 1); + assert_eq!(transition_state.num_workers.load(Ordering::SeqCst), 1); assert_eq!(worker_tokens.iter().filter(|token| token.is_cancelled()).count(), 1); let remaining_token = { - let workers = GLOBAL_TransitionState.workers.lock().unwrap(); + let workers = transition_state.workers.lock().unwrap(); assert_eq!(workers.len(), 1); let token = workers[0].cancel.clone(); assert!(!token.is_cancelled()); @@ -3562,7 +3582,7 @@ mod tests { }; TransitionState::resize_workers_to(ecstore.clone(), 0, 0, absolute_max); - assert_eq!(GLOBAL_TransitionState.num_workers.load(Ordering::SeqCst), 0); + assert_eq!(transition_state.num_workers.load(Ordering::SeqCst), 0); assert!(remaining_token.is_cancelled()); TransitionState::resize_workers_to(ecstore, original_workers, original_workers, absolute_max); diff --git a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs index e75bd9170..234e95ebc 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs @@ -18,11 +18,11 @@ #![allow(unused_must_use)] #![allow(clippy::all)] -use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryOp, GLOBAL_ExpiryState}; +use crate::bucket::lifecycle::bucket_lifecycle_ops::ExpiryOp; use crate::bucket::lifecycle::lifecycle::{self, ObjectOpts}; use crate::bucket::lifecycle::tier_delete_journal::persist_tier_delete_journal_entry; use crate::client::signer_error::error_chain_contains_signer_header_marker; -use crate::global::GLOBAL_TierConfigMgr; +use crate::runtime_sources; use crate::store::ECStore; use rustfs_storage_api::TransitionedObject; use rustfs_utils::get_env_usize; @@ -256,20 +256,21 @@ impl ObjSweeper { let Some(je) = self.should_remove_remote_object() else { return; }; + let expiry_state = runtime_sources::expiry_state_handle(); if persist_tier_delete_journal_entry(api, &je).await.is_err() { - GLOBAL_ExpiryState.write().await.increment_missed_tier_journal_tasks(); + expiry_state.write().await.increment_missed_tier_journal_tasks(); return; } let hash = je.op_hash(); // Grab the sender under a short read lock, then release the lock so we // don't hold it across the async send. - let wrkr = GLOBAL_ExpiryState.read().await.get_worker_ch(hash); + let wrkr = expiry_state.read().await.get_worker_ch(hash); let Some(wrkr) = wrkr else { - GLOBAL_ExpiryState.write().await.increment_missed_tier_journal_tasks(); + expiry_state.write().await.increment_missed_tier_journal_tasks(); return; }; if wrkr.send(Some(Box::new(je))).await.is_err() { - GLOBAL_ExpiryState.write().await.increment_missed_tier_journal_tasks(); + expiry_state.write().await.increment_missed_tier_journal_tasks(); } } } @@ -322,7 +323,8 @@ async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_na .map_err(|_| std::io::Error::other(ERR_REMOTE_DELETE_LIMITER_CLOSED))?; let _inflight = RemoteDeleteInflightGuard::new(); - let mut config_mgr = GLOBAL_TierConfigMgr.write().await; + let tier_config_mgr = runtime_sources::tier_config_mgr_handle(); + let mut config_mgr = tier_config_mgr.write().await; let w = match config_mgr.get_driver(tier_name).await { Ok(w) => w, Err(e) => return Err(std::io::Error::other(e)), diff --git a/crates/ecstore/src/rebalance/worker.rs b/crates/ecstore/src/rebalance/worker.rs index dabda47a1..cb80e4ee7 100644 --- a/crates/ecstore/src/rebalance/worker.rs +++ b/crates/ecstore/src/rebalance/worker.rs @@ -10,6 +10,7 @@ use crate::error::{ Error, is_err_object_not_found, is_err_operation_canceled, is_err_version_not_found, is_network_or_host_down, }; use crate::pools::ListCallback; +use crate::runtime_sources; use crate::set_disk::{SetDisks, get_lock_acquire_timeout}; use rand::RngExt as _; use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams}; @@ -382,7 +383,7 @@ pub(super) async fn load_rebalance_bucket_configs(bucket: &str) -> Result Arc> { GLOBAL_TierConfigMgr.clone() } +pub(crate) fn expiry_state_handle() -> Arc> { + GLOBAL_ExpiryState.clone() +} + +pub(crate) fn transition_state_handle() -> Arc { + GLOBAL_TransitionState.clone() +} + pub(crate) fn event_notifier_handle() -> Arc> { GLOBAL_EventNotifier.clone() } diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index f40fa2758..4ce751166 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -35,7 +35,6 @@ use crate::disk::{STORAGE_FORMAT_FILE, count_part_not_success}; use crate::erasure_coding; use crate::error::{Error, Result, is_err_version_not_found}; use crate::error::{GenericError, ObjectApiError, is_err_object_not_found}; -use crate::global::{GLOBAL_LocalNodeName, GLOBAL_TierConfigMgr}; use crate::object_api::ObjectOptions; use crate::rpc::heal_bucket_local_on_disks; use crate::runtime_sources; @@ -2731,7 +2730,8 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { #[tracing::instrument(level = "debug", skip(self))] async fn transition_object(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> { - let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await; + let tier_config_mgr = runtime_sources::tier_config_mgr_handle(); + let mut tier_config_mgr = tier_config_mgr.write().await; let tgt_client = match tier_config_mgr.get_driver(&opts.transition.tier).await { Ok(client) => client, Err(err) => { @@ -2893,7 +2893,7 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { bucket_name: bucket.to_string(), object: obj_info, user_agent: "Internal: [ILM-Transition]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); } @@ -2951,7 +2951,7 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { bucket_name: bucket.to_string(), object: restored_info, user_agent: "Internal: [Restore-Completed]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); Ok(()) @@ -3062,7 +3062,7 @@ impl rustfs_storage_api::ObjectOperations for SetDisks { bucket_name: bucket.to_string(), object: restored_info, user_agent: "Internal: [Restore-Completed]".to_string(), - host: GLOBAL_LocalNodeName.to_string(), + host: runtime_sources::default_local_node_name(), ..Default::default() }); Ok(()) diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index b05b0bf2f..582a7adba 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-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. +- Branch: `overtrue/arch-ecstore-lifecycle-runtime-sources-batch` +- 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/API-190`. +- Based on: latest default branch after merge 3802. - PR type for this branch: `consumer-migration` - Runtime behavior changes: none. - Rust code changes: route replication pool, outbound TLS generation, runtime @@ -21,6 +21,8 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block 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, + plus ECStore lifecycle queue state, tier config, lifecycle config, deployment + id, 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 @@ -30,7 +32,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-189 owner facade cleanup. +- Docs changes: record the API-136 through API-190 owner facade cleanup. ## Phase 0 Tasks @@ -4747,6 +4749,22 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block migration/layer guards, PR-before-push pre-commit quality gate, and three-expert review. +- [x] `API-190` Centralize ECStore lifecycle runtime source reads. + - Do: route lifecycle queue state handles, tier config manager reads, + lifecycle config reads, deployment-id reads, and lifecycle event-host reads + through the ECStore-owned runtime-source module. + - Acceptance: lifecycle ops, tier sweeper, rebalance config loading, and + set-disk transition/restore event paths no longer read those runtime + globals directly outside the owner runtime-source boundary. + - Must preserve: lifecycle worker sizing, expiry and transition queueing, + tier delete journal accounting, rebalance lifecycle config snapshots, + transitioned-object reader driver lookup, and emitted lifecycle transition, + restore, and expiry event host values. + - Verification: ECStore compile coverage, focused lifecycle worker test, + formatting, diff hygiene, residual lifecycle runtime-source scan, Rust risk + scan, 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. @@ -4870,6 +4888,9 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | 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. | +| Quality/architecture | pass | API-190 keeps ECStore lifecycle, tier, rebalance, and set-disk runtime reads behind the ECStore runtime-source boundary without adding public APIs. | +| Migration preservation | pass | Lifecycle worker state, expiry/transition queueing, tier driver lookups, rebalance lifecycle snapshots, and emitted event host values keep existing semantics. | +| Testing/verification | pass | ECStore compile/focused lifecycle test, formatting, migration/layer guards, diff hygiene, residual scan, diff-only Rust risk scan, and pre-commit passed for API-190. | ## Verification Notes @@ -4923,6 +4944,18 @@ Passed before push: - 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-190 current slice: + - `cargo check -p rustfs-ecstore --tests`: passed. + - `cargo test -p rustfs-ecstore --lib transition_worker_resize_cancels_removed_workers_directly -- --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. + - Lifecycle runtime-source scan: passed for API-190 targets. + - Diff-only Rust risk scan: passed. + - `make pre-commit`: passed. - Issue #660 API-186 current slice: - `cargo check -p rustfs-ecstore --tests`: passed.