Compare commits

...

2 Commits

Author SHA1 Message Date
Zhengchao An 4d0af8a330 refactor(ecstore): migrate the bucket bandwidth monitor into InstanceContext (Phase 5 Slice 10) (#4489) 2026-07-08 22:44:52 +08:00
overtrue 5a46819589 refactor(ecstore): migrate lifecycle expiry/transition state into InstanceContext (Phase 5 Slice 9)
Phase 5 Slice 9 (backlog#939): move the lifecycle expiry state and transition
state — two stateful service handles defined in bucket_lifecycle_ops — into the
per-instance InstanceContext, following the lazy-materialization pattern from
Slices 7-8.

- InstanceContext gains `expiry_state: OnceLock<Arc<RwLock<ExpiryState>>>` and
  `transition_state: OnceLock<Arc<TransitionState>>`, each materialized lazily
  on first access (both constructors are cheap — empty worker vecs/channels,
  workers started later — so lazy init reproduces the eager static's
  create-once-then-share).
- The lifecycle owner helpers `get_global_expiry_state` / `get_global_transition_state`
  and the runtime-source handles `expiry_state_handle` / `transition_state_handle`
  keep their signatures and route through the current instance's context; the
  two statics (and the now-unused lazy_static import) are removed. The scanner
  boundary and lifecycle-state arch guards still pass.

Tests: expiry/transition state are stable within an instance (same Arc) and
distinct across instances.

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

Refs: backlog#939 (Phase 5, Slice 9).
2026-07-08 22:16:11 +08:00
4 changed files with 87 additions and 15 deletions
@@ -51,7 +51,6 @@ use crate::store::ECStore;
use async_channel::{Receiver as A_Receiver, Sender as A_Sender, bounded};
use futures::Future;
use http::HeaderMap;
use lazy_static::lazy_static;
use rustfs_common::metrics::{
IlmAction, Metrics, ScannerLifecycleExpiryStateUpdate, ScannerLifecycleTransitionStateUpdate, global_metrics,
};
@@ -119,17 +118,15 @@ const DEFAULT_STALE_UPLOADS_CLEANUP_INTERVAL: StdDuration = StdDuration::from_se
const DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS: i64 = 5;
const EXPIRY_WORKER_QUEUE_CAPACITY: usize = 1000;
lazy_static! {
pub static ref GLOBAL_EXPIRY_STATE: Arc<RwLock<ExpiryState>> = ExpiryState::new();
pub static ref GLOBAL_TRANSITION_STATE: Arc<TransitionState> = TransitionState::new();
}
// Phase 5 (backlog#939): lifecycle expiry/transition state moved into the
// per-instance `InstanceContext`; these owner helpers forward to the current
// instance's context (lazily materialized, shared for single-instance).
pub fn get_global_expiry_state() -> Arc<RwLock<ExpiryState>> {
GLOBAL_EXPIRY_STATE.clone()
crate::runtime::global::current_ctx().expiry_state()
}
pub fn get_global_transition_state() -> Arc<TransitionState> {
GLOBAL_TRANSITION_STATE.clone()
crate::runtime::global::current_ctx().transition_state()
}
fn resolve_transition_worker_count() -> (i64, i64, i64) {
+3 -4
View File
@@ -64,20 +64,19 @@ lazy_static! {
rustfs_utils::crypto::hex(GLOBAL_LOCAL_NODE_NAME_FALLBACK.as_bytes());
pub static ref GLOBAL_LOCAL_LOCK_CLIENT: OnceLock<Arc<dyn LockClient>> = OnceLock::new();
pub static ref GLOBAL_LOCK_CLIENTS: OnceLock<HashMap<String, Arc<dyn LockClient>>> = OnceLock::new();
pub static ref GLOBAL_BUCKET_MONITOR: OnceLock<Arc<Monitor>> = OnceLock::new();
}
pub fn init_global_bucket_monitor(num_nodes: u64) {
if GLOBAL_BUCKET_MONITOR.set(Monitor::new(num_nodes)).is_err() {
if !current_ctx().init_bucket_monitor(num_nodes) {
warn!(
"global bucket monitor already initialized, ignoring re-initialization with num_nodes={}",
"bucket monitor already initialized, ignoring re-initialization with num_nodes={}",
num_nodes
);
}
}
pub fn get_global_bucket_monitor() -> Option<Arc<Monitor>> {
GLOBAL_BUCKET_MONITOR.get().cloned()
current_ctx().bucket_monitor()
}
// Startup-owned process globals intentionally fail fast on duplicate writes.
+76
View File
@@ -39,6 +39,8 @@
//! startup writes and post-construction reads hit the same cell and
//! single-instance behavior is byte-for-byte unchanged.
use crate::bucket::bandwidth::monitor::Monitor;
use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState};
use crate::layout::endpoints::{EndpointServerPools, SetupType};
use crate::services::event_notification::EventNotifier;
use crate::services::tier::tier::TierConfigMgr;
@@ -96,6 +98,22 @@ pub struct InstanceContext {
/// Lazily materialized like the tier config manager; holds the instance's
/// notification target list. Replaces the eager process static.
event_notifier: OnceLock<Arc<RwLock<EventNotifier>>>,
/// This instance's lifecycle expiry state (Phase 5 Slice 9, backlog#939).
///
/// Lazily materialized; holds the instance's expiry worker channels/stats.
/// Replaces the eager process static.
expiry_state: OnceLock<Arc<RwLock<ExpiryState>>>,
/// This instance's lifecycle transition state (Phase 5 Slice 9, backlog#939).
///
/// Lazily materialized; holds the instance's transition workers/stats.
/// Replaces the eager process static.
transition_state: OnceLock<Arc<TransitionState>>,
/// This instance's bucket bandwidth monitor (Phase 5 Slice 10, backlog#939).
///
/// Set once at startup with the cluster node count (see
/// [`InstanceContext::init_bucket_monitor`]); read as `None` until then.
/// Replaces the process-global bucket-monitor static.
bucket_monitor: OnceLock<Arc<Monitor>>,
}
impl InstanceContext {
@@ -120,6 +138,9 @@ impl InstanceContext {
endpoints: OnceLock::new(),
tier_config_mgr: OnceLock::new(),
event_notifier: OnceLock::new(),
expiry_state: OnceLock::new(),
transition_state: OnceLock::new(),
bucket_monitor: OnceLock::new(),
}
}
@@ -185,6 +206,30 @@ impl InstanceContext {
self.event_notifier.get_or_init(EventNotifier::new).clone()
}
/// This instance's lifecycle expiry state, materializing it on first
/// access. Shared for the lifetime of the instance.
pub fn expiry_state(&self) -> Arc<RwLock<ExpiryState>> {
self.expiry_state.get_or_init(ExpiryState::new).clone()
}
/// This instance's lifecycle transition state, materializing it on first
/// access. Shared for the lifetime of the instance.
pub fn transition_state(&self) -> Arc<TransitionState> {
self.transition_state.get_or_init(TransitionState::new).clone()
}
/// Initialize this instance's bucket bandwidth monitor with the cluster
/// node count. Returns `false` if it was already initialized (the caller
/// logs; matches the process global's ignore-and-warn behavior).
pub fn init_bucket_monitor(&self, num_nodes: u64) -> bool {
self.bucket_monitor.set(Monitor::new(num_nodes)).is_ok()
}
/// This instance's bucket bandwidth monitor, if it has been initialized.
pub fn bucket_monitor(&self) -> Option<Arc<Monitor>> {
self.bucket_monitor.get().cloned()
}
/// Update this instance's erasure setup type.
pub async fn update_erasure_type(&self, setup_type: SetupType) {
*self.erasure_kind.write().await = setup_type;
@@ -228,6 +273,9 @@ impl std::fmt::Debug for InstanceContext {
.field("endpoints", &self.endpoints)
.field("tier_config_mgr_set", &self.tier_config_mgr.get().is_some())
.field("event_notifier_set", &self.event_notifier.get().is_some())
.field("expiry_state_set", &self.expiry_state.get().is_some())
.field("transition_state_set", &self.transition_state.get().is_some())
.field("bucket_monitor_set", &self.bucket_monitor.get().is_some())
.finish_non_exhaustive()
}
}
@@ -489,4 +537,32 @@ mod tests {
"fresh contexts must not share an event notifier"
);
}
// Lifecycle expiry/transition state are each stable within an instance and
// distinct across instances.
#[tokio::test]
async fn lifecycle_state_is_stable_within_and_distinct_across_instances() {
let ctx_a = InstanceContext::new();
let ctx_b = InstanceContext::new();
assert!(Arc::ptr_eq(&ctx_a.expiry_state(), &ctx_a.expiry_state()));
assert!(Arc::ptr_eq(&ctx_a.transition_state(), &ctx_a.transition_state()));
assert!(!Arc::ptr_eq(&ctx_a.expiry_state(), &ctx_b.expiry_state()));
assert!(!Arc::ptr_eq(&ctx_a.transition_state(), &ctx_b.transition_state()));
}
// The bucket monitor is None until initialized, set-once (second init is a
// no-op returning false), and independent across instances.
#[tokio::test]
async fn bucket_monitor_init_once_and_isolated() {
let ctx_a = InstanceContext::new();
assert!(ctx_a.bucket_monitor().is_none());
assert!(ctx_a.init_bucket_monitor(4));
assert!(ctx_a.bucket_monitor().is_some());
assert!(!ctx_a.init_bucket_monitor(8), "second init must be ignored");
let ctx_b = InstanceContext::new();
assert!(ctx_b.bucket_monitor().is_none(), "a fresh instance has its own uninitialized monitor");
}
}
+3 -3
View File
@@ -21,7 +21,7 @@ use std::{
use crate::bucket::bandwidth::monitor::Monitor;
use crate::disk::endpoint::Endpoint;
use crate::{
bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, GLOBAL_EXPIRY_STATE, GLOBAL_TRANSITION_STATE, TransitionState},
bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, TransitionState},
bucket::metadata_sys::{BucketMetadataSys, get_global_bucket_metadata_sys},
bucket::replication::{
DynReplicationPool, ReplicationStats,
@@ -391,11 +391,11 @@ pub(crate) fn tier_config_mgr_handle() -> Arc<RwLock<TierConfigMgr>> {
}
pub fn expiry_state_handle() -> Arc<RwLock<ExpiryState>> {
GLOBAL_EXPIRY_STATE.clone()
crate::runtime::global::current_ctx().expiry_state()
}
pub fn transition_state_handle() -> Arc<TransitionState> {
GLOBAL_TRANSITION_STATE.clone()
crate::runtime::global::current_ctx().transition_state()
}
pub(crate) fn event_notifier_handle() -> Arc<RwLock<EventNotifier>> {