From 476352106ebd7217ab2553dc137c55b01325f3f8 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 9 Jul 2026 20:07:09 +0800 Subject: [PATCH] refactor(app): per-context server-config handle and Arc-owned notification system (#4621) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(app): give each context's server-config handle its own copy backlog#1052 S3, first slice: establish the per-context state pattern on the smallest subsystem. ServerConfigHandle was a unit struct forwarding both get and set to the process-global GLOBAL_SERVER_CONFIG, so every AppContext shared one config regardless of which context a caller held. The handle now keeps its own copy: a set lands on the owning context and on the process global (ambient readers — the config loader path, the scanner — keep working), and a get prefers the owned copy, falling back to the global while the initial load still publishes there. Single instance: the owned copy and the global are written together, so reads are unchanged. Two contexts that both published stay isolated even though the shared global only remembers the last write (covered by a new test). The global write in set() is the transitional bridge for ambient readers; it goes away when those readers migrate and multi-instance flips on (backlog#1052 S5). * refactor(notify): hand out the notification system as an owned Arc backlog#1052 S3, second slice. GLOBAL_NOTIFICATION_SYS stored the NotificationSys by value and every accessor returned Option<&'static NotificationSys> — a process-lifetime borrow that a per-server AppContext can never hold. The global now stores an Arc and the accessor chain (ecstore runtime sources, ECStore::notification_system, the iam forwarders, the rustfs shims, NotificationSystemInterface and the AppContext resolvers) hands out owned Arcs instead. Call sites are unchanged apart from two borrow adjustments in the rebalance admin handler (pass &Arc where a reference is expected; borrow with as_ref() so the handle survives to the second use). Behavior is identical — same single global instance, first init still wins — but the type now permits a future per-server context to own its notification system (backlog#1052 S3 follow-up). --- crates/ecstore/src/runtime/sources.rs | 2 +- .../ecstore/src/services/notification_sys.rs | 13 +++-- crates/ecstore/src/store/mod.rs | 2 +- crates/iam/src/runtime_sources.rs | 2 +- crates/iam/src/storage_api.rs | 2 +- rustfs/src/admin/handlers/rebalance.rs | 4 +- rustfs/src/admin/runtime_sources.rs | 2 +- rustfs/src/app/context.rs | 6 +- rustfs/src/app/context/handles.rs | 57 ++++++++++++++++++- rustfs/src/app/context/interfaces.rs | 2 +- rustfs/src/app/context/runtime_sources.rs | 2 +- rustfs/src/app/storage_api.rs | 2 +- rustfs/src/storage/storage_api.rs | 2 +- 13 files changed, 76 insertions(+), 22 deletions(-) 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() }