diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 50a9db555..8bfb9af2c 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -618,7 +618,7 @@ impl ExpiryState { async fn worker(rx: &mut Receiver>, api: Arc, stats: Arc) { let cancel_token = runtime_sources::background_services_cancel_token().unwrap_or_else(|| { static FALLBACK: std::sync::OnceLock = std::sync::OnceLock::new(); - FALLBACK.get_or_init(tokio_util::sync::CancellationToken::new) + FALLBACK.get_or_init(tokio_util::sync::CancellationToken::new).clone() }); loop { @@ -1352,9 +1352,7 @@ fn spawn_tier_free_version_recovery_once(api: Arc) { } tokio::spawn(async move { - let cancel_token = runtime_sources::background_services_cancel_token() - .cloned() - .unwrap_or_else(CancellationToken::new); + let cancel_token = runtime_sources::background_services_cancel_token().unwrap_or_default(); let mut interval = tokio::time::interval(StdDuration::from_secs(60)); let mut bucket_marker: Option = None; let mut object_marker: Option = None; @@ -1427,9 +1425,7 @@ fn spawn_tier_delete_journal_recovery_once(api: Arc) { } tokio::spawn(async move { - let cancel_token = runtime_sources::background_services_cancel_token() - .cloned() - .unwrap_or_else(CancellationToken::new); + let cancel_token = runtime_sources::background_services_cancel_token().unwrap_or_default(); run_tier_delete_journal_recovery_loop(api, cancel_token).await; }); } diff --git a/crates/ecstore/src/bucket/lifecycle/runtime_boundary.rs b/crates/ecstore/src/bucket/lifecycle/runtime_boundary.rs index afadcb39d..fcd886ca5 100644 --- a/crates/ecstore/src/bucket/lifecycle/runtime_boundary.rs +++ b/crates/ecstore/src/bucket/lifecycle/runtime_boundary.rs @@ -35,7 +35,7 @@ pub(crate) fn tier_config_mgr_handle() -> Arc> { sources::tier_config_mgr_handle() } -pub(crate) fn background_services_cancel_token() -> Option<&'static CancellationToken> { +pub(crate) fn background_services_cancel_token() -> Option { sources::background_services_cancel_token() } diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 073059069..89954820d 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -93,7 +93,7 @@ pub async fn remove_bucket_metadata(bucket: &str) -> Result { } fn start_refresh_buckets_metadata_loop(sys: Arc>) { - let Some(cancel_token) = runtime_sources::background_services_cancel_token().cloned() else { + let Some(cancel_token) = runtime_sources::background_services_cancel_token() else { warn!("bucket metadata refresh loop skipped because background cancellation token is not initialized"); return; }; diff --git a/crates/ecstore/src/runtime/global.rs b/crates/ecstore/src/runtime/global.rs index 40bd14f50..d88de0727 100644 --- a/crates/ecstore/src/runtime/global.rs +++ b/crates/ecstore/src/runtime/global.rs @@ -76,12 +76,9 @@ pub fn get_global_bucket_monitor() -> Option> { current_ctx().bucket_monitor() } -// Startup-owned process globals intentionally fail fast on duplicate writes. +// Startup-owned instance-context state intentionally fails fast on duplicate writes. // A second write means startup published conflicting runtime scalar state. -/// Global cancellation token for background services (data scanner and auto heal) -static GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN: OnceLock = OnceLock::new(); - /// Get the global rustfs port /// /// # Returns @@ -297,7 +294,7 @@ pub fn get_global_region() -> Option { /// * `Err(CancellationToken)` if setting fails /// pub fn init_background_services_cancel_token(cancel_token: CancellationToken) -> Result<(), CancellationToken> { - GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN.set(cancel_token) + current_ctx().init_background_cancel_token(cancel_token) } /// Get the global background services cancellation token @@ -305,8 +302,8 @@ pub fn init_background_services_cancel_token(cancel_token: CancellationToken) -> /// # Returns /// * `Option<&'static CancellationToken>` - The global cancellation token, if set /// -pub fn get_background_services_cancel_token() -> Option<&'static CancellationToken> { - GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN.get() +pub fn get_background_services_cancel_token() -> Option { + current_ctx().background_cancel_token() } /// Create and initialize the global background services cancellation token @@ -326,7 +323,7 @@ pub fn create_background_services_cancel_token() -> CancellationToken { /// # Returns /// * None pub fn shutdown_background_services() { - if let Some(cancel_token) = GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN.get() { + if let Some(cancel_token) = get_background_services_cancel_token() { cancel_token.cancel(); } } diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index 68d60a4ce..8a0af712d 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -52,6 +52,7 @@ use s3s::region::Region; use std::collections::HashMap; use std::sync::{Arc, OnceLock}; use tokio::sync::{OnceCell, RwLock}; +use tokio_util::sync::CancellationToken; use uuid::Uuid; /// Runtime state owned by a single `ECStore` instance. @@ -138,6 +139,13 @@ pub struct InstanceContext { local_disk_map: Arc>>>, local_disk_id_map: Arc>>, local_disk_set_drives: Arc>, + /// This instance's background-services cancellation token (Phase 5 Slice 13, + /// backlog#939). + /// + /// Set once at startup; cancelling it stops this instance's background + /// workers (scanner/heal/tier/lifecycle) without touching another instance. + /// Replaces the process-global cancel-token static. + background_cancel_token: OnceLock, } impl InstanceContext { @@ -170,6 +178,7 @@ impl InstanceContext { local_disk_map: Arc::new(RwLock::new(HashMap::new())), local_disk_id_map: Arc::new(RwLock::new(HashMap::new())), local_disk_set_drives: Arc::new(RwLock::new(Vec::new())), + background_cancel_token: OnceLock::new(), } } @@ -301,6 +310,16 @@ impl InstanceContext { self.local_disk_set_drives.clone() } + /// Set this instance's background-services cancellation token (once). + pub fn init_background_cancel_token(&self, token: CancellationToken) -> Result<(), CancellationToken> { + self.background_cancel_token.set(token) + } + + /// This instance's background-services cancellation token, if initialized. + pub fn background_cancel_token(&self) -> Option { + self.background_cancel_token.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; @@ -349,6 +368,7 @@ impl std::fmt::Debug for InstanceContext { .field("bucket_monitor_set", &self.bucket_monitor.get().is_some()) .field("replication_stats_set", &self.replication_stats.get().is_some()) .field("replication_pool_set", &self.replication_pool.get().is_some()) + .field("background_cancel_token_set", &self.background_cancel_token.get().is_some()) .finish_non_exhaustive() } } @@ -660,4 +680,80 @@ mod tests { "a distinct instance has independent replication state" ); } + + // The background cancel token is set-once, and cancelling one instance's + // token leaves a distinct instance's uninitialized. + #[tokio::test] + async fn background_cancel_token_init_once_and_isolated() { + let ctx_a = InstanceContext::new(); + assert!(ctx_a.background_cancel_token().is_none()); + + let token = CancellationToken::new(); + ctx_a.init_background_cancel_token(token.clone()).expect("init once"); + assert!(ctx_a.background_cancel_token().is_some()); + assert!( + ctx_a.init_background_cancel_token(CancellationToken::new()).is_err(), + "second init rejected" + ); + + let ctx_b = InstanceContext::new(); + assert!(ctx_b.background_cancel_token().is_none(), "distinct instance is independent"); + + token.cancel(); + assert!(ctx_a.background_cancel_token().unwrap().is_cancelled()); + } + + // Phase 5 acceptance (backlog#939): two independent instance contexts share + // NONE of the runtime state that used to live in process globals. This is + // the end-to-end proof that the object-graph isolation carrier works — every + // field migrated across Slices 1-13 is verified independent here. + #[tokio::test] + async fn two_instances_isolate_all_migrated_state() { + let a = InstanceContext::new(); + let b = InstanceContext::new(); + + // Erasure setup type (Slice 1). + a.update_erasure_type(SetupType::DistErasure).await; + b.update_erasure_type(SetupType::ErasureSD).await; + assert!(a.is_dist_erasure().await && !b.is_dist_erasure().await); + assert!(b.is_erasure_sd().await && !a.is_erasure_sd().await); + + // Lock manager (Slice 3). + assert!(!Arc::ptr_eq(&a.lock_manager(), &b.lock_manager())); + + // Region (Slice 4). + a.set_region("us-east-1".parse().expect("region")); + b.set_region("eu-west-1".parse().expect("region")); + assert_eq!(a.region().expect("a region"), "us-east-1".parse().expect("region")); + assert_eq!(b.region().expect("b region"), "eu-west-1".parse().expect("region")); + + // Deployment id (Slice 5). + let (id_a, id_b) = (Uuid::new_v4(), Uuid::new_v4()); + a.set_deployment_id(id_a); + b.set_deployment_id(id_b); + assert_eq!(a.deployment_id(), Some(id_a)); + assert_eq!(b.deployment_id(), Some(id_b)); + + // Service handles (Slices 7-11): each instance materializes its own. + assert!(!Arc::ptr_eq(&a.tier_config_mgr(), &b.tier_config_mgr())); + assert!(!Arc::ptr_eq(&a.event_notifier(), &b.event_notifier())); + assert!(!Arc::ptr_eq(&a.expiry_state(), &b.expiry_state())); + assert!(!Arc::ptr_eq(&a.transition_state(), &b.transition_state())); + + // Local disk registry (Slice 12): a write to one is invisible to the other. + a.local_disk_map().write().await.insert("/disk-a".to_string(), None); + assert_eq!(a.local_disk_map().read().await.len(), 1); + assert_eq!(b.local_disk_map().read().await.len(), 0); + + // Bucket monitor (Slice 10): set-once per instance. + assert!(a.init_bucket_monitor(4)); + assert!(a.bucket_monitor().is_some() && b.bucket_monitor().is_none()); + + // Background cancel token (Slice 13): cancelling one doesn't touch the other. + let token_a = CancellationToken::new(); + a.init_background_cancel_token(token_a.clone()).expect("init a"); + token_a.cancel(); + assert!(a.background_cancel_token().expect("a token").is_cancelled()); + assert!(b.background_cancel_token().is_none()); + } } diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index dc1137427..b249ad322 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -179,7 +179,7 @@ pub fn rustfs_port() -> u16 { global_rustfs_port() } -pub(crate) fn background_services_cancel_token() -> Option<&'static CancellationToken> { +pub(crate) fn background_services_cancel_token() -> Option { get_background_services_cancel_token() } diff --git a/rustfs/src/startup_lifecycle.rs b/rustfs/src/startup_lifecycle.rs index 21ac47fa2..f66647509 100644 --- a/rustfs/src/startup_lifecycle.rs +++ b/rustfs/src/startup_lifecycle.rs @@ -41,6 +41,18 @@ const EVENT_SERVER_READY: &str = "server_ready"; const EVENT_SERVER_SHUTDOWN_STATE: &str = "server_shutdown_state"; const EVENT_EMBEDDED_SERVER_STATE: &str = "embedded_server_state"; +/// Guards against a second embedded server starting in the same process. +/// +/// Phase 5 (backlog#939) moved per-instance runtime state (erasure setup, +/// region, deployment id, endpoints, service handles, disk registry, cancel +/// token) into `ECStore`'s `InstanceContext`, so two `ECStore` object graphs +/// can now stay isolated. This guard is intentionally **retained**: the startup +/// path still publishes into the process-level *bootstrap* context (write-once +/// region/endpoints/deployment id) and the single `GLOBAL_OBJECT_API` handle, +/// so a second startup would fail-fast on that shared state. Lifting the guard +/// requires threading a per-instance context through storage startup (so each +/// server constructs its own context instead of sharing the bootstrap); until +/// then, rejecting the second start is safer than the panic it would become. static EMBEDDED_SERVER_STARTED: AtomicBool = AtomicBool::new(false); #[derive(Debug, Clone, Copy, PartialEq, Eq)]