diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index 94902c6b0..a8ea5af2b 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -306,7 +306,7 @@ pub(crate) async fn publish_object_store(store: Arc) { set_object_layer(store).await; } -pub(crate) fn notification_sys() -> Option<&'static NotificationSys> { +pub(crate) fn notification_sys() -> Option> { get_global_notification_sys() } diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index 6d2ece230..3555159d5 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -31,7 +31,7 @@ use rustfs_utils::XHost; use std::collections::hash_map::DefaultHasher; use std::future::Future; use std::hash::{Hash, Hasher}; -use std::sync::{Mutex, OnceLock}; +use std::sync::{Arc, Mutex, OnceLock}; use std::time::{Duration, SystemTime}; use tokio::time::timeout; use tracing::{debug, error, info, warn}; @@ -72,18 +72,21 @@ impl PeerAdminCache { const SERVER_INFO_CACHE_MAX_AGE: Duration = Duration::from_secs(60); lazy_static! { - pub static ref GLOBAL_NOTIFICATION_SYS: OnceLock = OnceLock::new(); + pub static ref GLOBAL_NOTIFICATION_SYS: OnceLock> = OnceLock::new(); } pub async fn new_global_notification_sys(eps: EndpointServerPools) -> Result<()> { let _ = GLOBAL_NOTIFICATION_SYS - .set(NotificationSys::new(eps).await) + .set(Arc::new(NotificationSys::new(eps).await)) .map_err(|_| Error::other("init notification_sys fail")); Ok(()) } -pub fn get_global_notification_sys() -> Option<&'static NotificationSys> { - GLOBAL_NOTIFICATION_SYS.get() +// Owned handle rather than `&'static` (backlog#1052 S3): per-server contexts +// need to hold their own notification system, which a process-lifetime +// borrow cannot express. +pub fn get_global_notification_sys() -> Option> { + GLOBAL_NOTIFICATION_SYS.get().cloned() } pub struct NotificationSys { diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index f80859133..3963e6514 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -242,7 +242,7 @@ impl ECStore { /// service singletons. The globals remain the source of truth. impl ECStore { /// Get the notification system - pub fn notification_system(&self) -> Option<&'static crate::services::notification_sys::NotificationSys> { + pub fn notification_system(&self) -> Option> { runtime_sources::notification_sys() } diff --git a/crates/iam/src/runtime_sources.rs b/crates/iam/src/runtime_sources.rs index ef6db3211..f9a8925ee 100644 --- a/crates/iam/src/runtime_sources.rs +++ b/crates/iam/src/runtime_sources.rs @@ -24,6 +24,6 @@ pub(crate) fn current_server_config() -> Option { get_global_server_config() } -pub(crate) fn notification_sys() -> Option<&'static IamNotificationSys> { +pub(crate) fn notification_sys() -> Option> { ecstore_notification_sys() } diff --git a/crates/iam/src/storage_api.rs b/crates/iam/src/storage_api.rs index e0fbf60f5..ef4a5c716 100644 --- a/crates/iam/src/storage_api.rs +++ b/crates/iam/src/storage_api.rs @@ -77,7 +77,7 @@ pub(crate) async fn is_iam_first_cluster_node_local() -> bool { ecstore_first_cluster_node_is_local().await } -pub(crate) fn notification_sys() -> Option<&'static IamNotificationSys> { +pub(crate) fn notification_sys() -> Option> { get_global_notification_sys() } diff --git a/rustfs/src/admin/handlers/rebalance.rs b/rustfs/src/admin/handlers/rebalance.rs index d7664a663..5685d3f17 100644 --- a/rustfs/src/admin/handlers/rebalance.rs +++ b/rustfs/src/admin/handlers/rebalance.rs @@ -529,7 +529,7 @@ impl Operation for RebalanceStart { ); let start_err = err.to_string(); - let rollback_result = rollback_cluster_rebalance_start(&store, Some(notification_sys), &id).await; + let rollback_result = rollback_cluster_rebalance_start(&store, Some(¬ification_sys), &id).await; let rollback_label = rollback_result_label(&rollback_result); match &rollback_result { Ok(_) => info!( @@ -754,7 +754,7 @@ impl Operation for RebalanceStop { let notification_sys = current_notification_system(); let stop_attempt_at = OffsetDateTime::now_utc(); let mut stop_failures = Vec::new(); - if let Some(notification_sys) = notification_sys { + if let Some(notification_sys) = notification_sys.as_ref() { stop_failures = notification_sys .stop_rebalance_failures(expected_rebalance_id.as_deref()) .await diff --git a/rustfs/src/admin/runtime_sources.rs b/rustfs/src/admin/runtime_sources.rs index 028a24a51..c3e552924 100644 --- a/rustfs/src/admin/runtime_sources.rs +++ b/rustfs/src/admin/runtime_sources.rs @@ -69,7 +69,7 @@ pub(crate) fn object_store_from_extensions(extensions: &http::Extensions) -> Opt .or_else(current_object_store_handle) } -pub(crate) fn current_notification_system() -> Option<&'static NotificationSys> { +pub(crate) fn current_notification_system() -> Option> { let context = current_app_context(); current_notification_system_for_context(context.as_deref()) } diff --git a/rustfs/src/app/context.rs b/rustfs/src/app/context.rs index ae32edd80..c919ab0b3 100644 --- a/rustfs/src/app/context.rs +++ b/rustfs/src/app/context.rs @@ -155,12 +155,12 @@ pub fn resolve_notify_interface_for_context(context: Option<&AppContext>) -> Opt } /// Resolve notification system handle using AppContext-first precedence. -pub fn resolve_notification_system() -> Option<&'static NotificationSys> { +pub fn resolve_notification_system() -> Option> { resolve_notification_system_with(get_global_app_context()) } /// Resolve notification system handle using an explicit AppContext. -pub fn resolve_notification_system_for_context(context: Option<&AppContext>) -> Option<&'static NotificationSys> { +pub fn resolve_notification_system_for_context(context: Option<&AppContext>) -> Option> { context.and_then(|context| context.notification_system().handle()) } @@ -353,7 +353,7 @@ fn resolve_bucket_metadata_handle_with(context: Option>) -> Opti context.and_then(|context| context.bucket_metadata().handle()) } -fn resolve_notification_system_with(context: Option>) -> Option<&'static NotificationSys> { +fn resolve_notification_system_with(context: Option>) -> Option> { context.and_then(|context| context.notification_system().handle()) } diff --git a/rustfs/src/app/context/handles.rs b/rustfs/src/app/context/handles.rs index addfd9ae8..bd514cd67 100644 --- a/rustfs/src/app/context/handles.rs +++ b/rustfs/src/app/context/handles.rs @@ -178,7 +178,7 @@ impl NotifyInterface for NotifyHandle { pub struct NotificationSystemHandle; impl NotificationSystemInterface for NotificationSystemHandle { - fn handle(&self) -> Option<&'static NotificationSys> { + fn handle(&self) -> Option> { runtime_sources::notification_system() } } @@ -395,15 +395,31 @@ impl TransitionStateInterface for TransitionStateHandle { } /// Default server config interface adapter. +/// +/// Holds this context's own copy of the server config (backlog#1052 S3): a +/// `set` lands on the owning context *and* on the process global, so ambient +/// readers (the config loader path, the scanner) keep working; a `get` +/// prefers the owned copy and falls back to the process global while the +/// initial load still publishes there — the single-instance legacy default. #[derive(Default)] -pub struct ServerConfigHandle; +pub struct ServerConfigHandle { + config: StdRwLock>, +} impl ServerConfigInterface for ServerConfigHandle { fn get(&self) -> Option { + if let Ok(guard) = self.config.read() + && guard.is_some() + { + return guard.clone(); + } runtime_sources::server_config() } fn set(&self, config: Config) { + if let Ok(mut guard) = self.config.write() { + *guard = Some(config.clone()); + } runtime_sources::set_server_config(config); } } @@ -537,7 +553,7 @@ pub fn default_transition_state_interface() -> Arc } pub fn default_server_config_interface() -> Arc { - Arc::new(ServerConfigHandle) + Arc::new(ServerConfigHandle::default()) } pub fn default_storage_class_interface() -> Arc { @@ -547,3 +563,38 @@ pub fn default_storage_class_interface() -> Arc { pub fn default_buffer_config_interface() -> Arc { Arc::new(BufferConfigHandle) } + +#[cfg(test)] +mod tests { + use super::{ServerConfigHandle, runtime_sources}; + use crate::app::context::interfaces::ServerConfigInterface; + use rustfs_config::server_config::Config; + use std::collections::HashMap; + + // backlog#1052 S3: each context's server-config handle keeps its own copy, + // so two handles that both published stay isolated even though the shared + // process global only remembers the last write. + #[test] + fn server_config_handle_prefers_its_own_copy_over_the_shared_global() { + let mut config_a = Config::new(); + config_a.0.insert("handle-a-marker".to_string(), HashMap::new()); + let mut config_b = Config::new(); + config_b.0.insert("handle-b-marker".to_string(), HashMap::new()); + + let handle_a = ServerConfigHandle::default(); + let handle_b = ServerConfigHandle::default(); + handle_a.set(config_a.clone()); + handle_b.set(config_b.clone()); + + assert_eq!(handle_a.get(), Some(config_a), "handle A must serve its own copy"); + assert_eq!(handle_b.get(), Some(config_b), "handle B must serve its own copy"); + } + + // Before anything is published through the handle, it answers exactly like + // the ambient global — the single-instance legacy default. + #[test] + fn unset_server_config_handle_falls_back_to_the_ambient_global() { + let handle = ServerConfigHandle::default(); + assert_eq!(handle.get(), runtime_sources::server_config()); + } +} diff --git a/rustfs/src/app/context/interfaces.rs b/rustfs/src/app/context/interfaces.rs index 281b5c5dc..6c580f522 100644 --- a/rustfs/src/app/context/interfaces.rs +++ b/rustfs/src/app/context/interfaces.rs @@ -87,7 +87,7 @@ pub trait NotifyInterface: Send + Sync { /// Notification system handle interface for admin peer orchestration. pub trait NotificationSystemInterface: Send + Sync { - fn handle(&self) -> Option<&'static NotificationSys>; + fn handle(&self) -> Option>; } /// Bucket metadata interface for application-layer use-cases. diff --git a/rustfs/src/app/context/runtime_sources.rs b/rustfs/src/app/context/runtime_sources.rs index 54c0ff875..23332e07d 100644 --- a/rustfs/src/app/context/runtime_sources.rs +++ b/rustfs/src/app/context/runtime_sources.rs @@ -92,7 +92,7 @@ pub async fn clear_bucket_notification_rules(bucket_name: &str) -> Result<(), No notifier_global::clear_bucket_notification_rules(bucket_name).await } -pub fn notification_system() -> Option<&'static NotificationSys> { +pub fn notification_system() -> Option> { get_global_notification_sys() } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 084b5a64b..ea61c2ef6 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -206,7 +206,7 @@ pub(crate) mod runtime { crate::storage::storage_api::set_object_store_resolver(resolver) } - pub(crate) fn get_global_notification_sys() -> Option<&'static NotificationSys> { + pub(crate) fn get_global_notification_sys() -> Option> { crate::storage::storage_api::get_global_notification_sys() } diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 3cd5dd530..df7b6a167 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -1305,7 +1305,7 @@ pub(crate) fn set_workload_admission_snapshot_provider( rustfs_ecstore::set_workload_admission_snapshot_provider(provider) } -pub(crate) fn get_global_notification_sys() -> Option<&'static NotificationSys> { +pub(crate) fn get_global_notification_sys() -> Option> { ecstore_notification::get_global_notification_sys() }