refactor(app): per-context server-config handle and Arc-owned notification system (#4621)

* 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

@@ -306,7 +306,7 @@ pub(crate) async fn publish_object_store(store: Arc<ECStore>) {
     set_object_layer(store).await;
 }
 
-pub(crate) fn notification_sys() -> Option<&'static NotificationSys> {
+pub(crate) fn notification_sys() -> Option<Arc<NotificationSys>> {
     get_global_notification_sys()
 }
 

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<NotificationSys> = OnceLock::new();
+    pub static ref GLOBAL_NOTIFICATION_SYS: OnceLock<Arc<NotificationSys>> = 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<Arc<NotificationSys>> {
+    GLOBAL_NOTIFICATION_SYS.get().cloned()
 }
 
 pub struct NotificationSys {

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<std::sync::Arc<crate::services::notification_sys::NotificationSys>> {
         runtime_sources::notification_sys()
     }
 

crates/iam/src/runtime_sources.rs

@@ -24,6 +24,6 @@ pub(crate) fn current_server_config() -> Option<ServerConfig> {
     get_global_server_config()
 }
 
-pub(crate) fn notification_sys() -> Option<&'static IamNotificationSys> {
+pub(crate) fn notification_sys() -> Option<std::sync::Arc<IamNotificationSys>> {
     ecstore_notification_sys()
 }

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<std::sync::Arc<IamNotificationSys>> {
     get_global_notification_sys()
 }
 

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(&notification_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

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<Arc<NotificationSys>> {
     let context = current_app_context();
     current_notification_system_for_context(context.as_deref())
 }

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<Arc<NotificationSys>> {
     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<Arc<NotificationSys>> {
     context.and_then(|context| context.notification_system().handle())
 }
 
@@ -353,7 +353,7 @@ fn resolve_bucket_metadata_handle_with(context: Option<Arc<AppContext>>) -> Opti
     context.and_then(|context| context.bucket_metadata().handle())
 }
 
-fn resolve_notification_system_with(context: Option<Arc<AppContext>>) -> Option<&'static NotificationSys> {
+fn resolve_notification_system_with(context: Option<Arc<AppContext>>) -> Option<Arc<NotificationSys>> {
     context.and_then(|context| context.notification_system().handle())
 }
 

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<Arc<NotificationSys>> {
         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<Option<Config>>,
+}
 
 impl ServerConfigInterface for ServerConfigHandle {
     fn get(&self) -> Option<Config> {
+        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<dyn TransitionStateInterface>
 }
 
 pub fn default_server_config_interface() -> Arc<dyn ServerConfigInterface> {
-    Arc::new(ServerConfigHandle)
+    Arc::new(ServerConfigHandle::default())
 }
 
 pub fn default_storage_class_interface() -> Arc<dyn StorageClassInterface> {
@@ -547,3 +563,38 @@ pub fn default_storage_class_interface() -> Arc<dyn StorageClassInterface> {
 pub fn default_buffer_config_interface() -> Arc<dyn BufferConfigInterface> {
     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());
+    }
+}

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<Arc<NotificationSys>>;
 }
 
 /// Bucket metadata interface for application-layer use-cases.

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<Arc<NotificationSys>> {
     get_global_notification_sys()
 }
 

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<std::sync::Arc<NotificationSys>> {
         crate::storage::storage_api::get_global_notification_sys()
     }
 

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<Arc<NotificationSys>> {
     ecstore_notification::get_global_notification_sys()
 }