refactor(iam): hand the built IAM system to the AppContext explicitly (#4624)

backlog#1052 S3, fourth slice. The AppContext's IamHandle already owned an
Arc<IamSys>, but the value it held was read back from the process
singleton after init — so a future second server's context would silently
bind to the first server's IAM domain (credentials, policies, users all
belong to a specific store's .rustfs.sys).

- rustfs_iam gains build_iam_sys(store): construct an IAM system bound to
  a store without touching the singleton. init_iam_sys wraps it (build,
  publish first-wins, return the handle — previously Result<()>).
- Startup threads the built handle: the inline bootstrap path passes the
  Arc it just created into ensure_startup_after_iam, which constructs the
  AppContext from the passed value instead of re-resolving the global.
  The deferred-recovery path resolves the freshly published global
  directly (context-first resolution cannot be used there: the AppContext
  does not exist yet — creating it is the finalizer's job; caught by the
  embedded deferred-IAM e2e).
- IamHandle::is_ready() reports the held system's readiness instead of
  consulting the process singleton; the dead global re-resolver is
  removed. token_signing_key stays on the credentials global until S4.

157 iam tests + embedded e2e (basic + deferred recovery) green.

crates/iam/src/lib.rs

@@ -163,9 +163,25 @@ pub(crate) async fn notify_iam_load_policy_mapping(
 static IAM_SYS: OnceLock<Arc<IamSys<ObjectStore>>> = OnceLock::new();
 static OIDC_SYS: OnceLock<Arc<OidcSys>> = OnceLock::new();
 
+/// Build an IAM system bound to the given store without touching the process
+/// singleton (backlog#1052 S3): a per-server context can own the returned
+/// handle while the singleton keeps serving ambient readers.
 #[instrument(skip(ecstore))]
-pub async fn init_iam_sys(ecstore: Arc<IamStore>) -> Result<()> {
-    if IAM_SYS.get().is_some() {
+pub async fn build_iam_sys(ecstore: Arc<IamStore>) -> Result<Arc<IamSys<ObjectStore>>> {
+    // 1. Create the persistent storage adapter
+    let storage_adapter = ObjectStore::new(ecstore);
+
+    // 2. Create the cache manager.
+    // The `new` method now performs a blocking initial load from disk.
+    let cache_manager = IamCache::new(storage_adapter).await?;
+
+    // 3. Construct the system interface
+    Ok(Arc::new(IamSys::new(cache_manager)))
+}
+
+#[instrument(skip(ecstore))]
+pub async fn init_iam_sys(ecstore: Arc<IamStore>) -> Result<Arc<IamSys<ObjectStore>>> {
+    if let Some(existing) = IAM_SYS.get() {
         info!(
             event = EVENT_IAM_STATE,
             component = LOG_COMPONENT_IAM,
@@ -173,7 +189,7 @@ pub async fn init_iam_sys(ecstore: Arc<IamStore>) -> Result<()> {
             state = "already_initialized",
             "IAM runtime already initialized"
         );
-        return Ok(());
+        return Ok(existing.clone());
     }
 
     info!(
@@ -184,18 +200,10 @@ pub async fn init_iam_sys(ecstore: Arc<IamStore>) -> Result<()> {
         "IAM runtime starting"
     );
 
-    // 1. Create the persistent storage adapter
-    let storage_adapter = ObjectStore::new(ecstore);
+    let iam_instance = build_iam_sys(ecstore).await?;
 
-    // 2. Create the cache manager.
-    // The `new` method now performs a blocking initial load from disk.
-    let cache_manager = IamCache::new(storage_adapter).await?;
-
-    // 3. Construct the system interface
-    let iam_instance = Arc::new(IamSys::new(cache_manager));
-
-    // 4. Securely set the global singleton
-    if IAM_SYS.set(iam_instance).is_err() {
+    // Securely set the global singleton
+    if IAM_SYS.set(iam_instance.clone()).is_err() {
         error!(
             event = EVENT_IAM_STATE,
             component = LOG_COMPONENT_IAM,
@@ -213,7 +221,7 @@ pub async fn init_iam_sys(ecstore: Arc<IamStore>) -> Result<()> {
         state = "ready",
         "IAM runtime ready"
     );
-    Ok(())
+    Ok(iam_instance)
 }
 
 #[inline]

rustfs/src/app/context/handles.rs

@@ -65,7 +65,10 @@ impl IamInterface for IamHandle {
     }
 
     fn is_ready(&self) -> bool {
-        runtime_sources::ready_iam_handle().is_ok()
+        // The handle owns a fully initialized system (contexts are only built
+        // after IAM bootstrap), so readiness is a property of the held Arc —
+        // not of the process singleton (backlog#1052 S3).
+        self.iam.is_ready()
     }
 
     fn token_signing_key(&self) -> Option<String> {

rustfs/src/app/context/runtime_sources.rs

@@ -25,7 +25,7 @@ use super::super::storage_api::context::runtime::{
 use crate::config::{RustFSBufferConfig, get_global_buffer_config};
 use rustfs_config::server_config::{Config, get_global_server_config, set_global_server_config};
 use rustfs_credentials::{Credentials, get_global_action_cred};
-use rustfs_iam::{error::Result as IamResult, oidc::OidcSys, store::object::ObjectStore, sys::IamSys};
+use rustfs_iam::oidc::OidcSys;
 use rustfs_io_metrics::{
     PerformanceMetrics,
     global_metrics::get_global_metrics,
@@ -64,10 +64,6 @@ pub async fn outbound_tls_state() -> GlobalPublishedOutboundTlsState {
     load_global_outbound_tls_state().await
 }
 
-pub fn ready_iam_handle() -> IamResult<Arc<IamSys<ObjectStore>>> {
-    rustfs_iam::get()
-}
-
 pub fn oidc_handle() -> Option<Arc<OidcSys>> {
     rustfs_iam::get_oidc()
 }

rustfs/src/app/context/startup.rs

@@ -17,7 +17,7 @@ use super::global::{AppContext, get_global_app_context, init_global_app_context}
 use super::runtime_sources;
 use super::server_slot::ServerContextSlot;
 use rustfs_kms::KmsServiceManager;
-use std::io::{Error, Result};
+use std::io::Result;
 use std::sync::Arc;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -35,13 +35,15 @@ impl AppContext {
         store: Arc<ECStore>,
         kms_interface: Arc<KmsServiceManager>,
         server_ctx: &ServerContextSlot,
+        iam: Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>>,
     ) -> Result<()> {
         ensure_startup_app_context_after_iam_with(
             || get_global_app_context().is_some(),
             || {
-                let iam_interface =
-                    runtime_sources::ready_iam_handle().map_err(|_| Error::other("IAM is initialized but unavailable"))?;
-                init_global_app_context(AppContext::with_default_interfaces(store, iam_interface, kms_interface));
+                // The caller hands over the IAM system it just built for this
+                // server (backlog#1052 S3) — no read-back through the process
+                // singleton, so a future second server's context owns its own.
+                init_global_app_context(AppContext::with_default_interfaces(store, iam, kms_interface));
                 Ok(())
             },
         )?;

rustfs/src/startup_iam.rs

@@ -78,7 +78,10 @@ async fn finalize_iam_recovery(
     state_manager: Option<Arc<ServiceStateManager>>,
     server_ctx: Arc<ServerContextSlot>,
 ) -> Result<()> {
-    AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx)?;
+    // Resolve the globally published system directly (not context-first —
+    // the AppContext does not exist yet; creating it is this function's job).
+    let iam = rustfs_iam::get().map_err(|_| std::io::Error::other("IAM recovered but unavailable"))?;
+    AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx, iam)?;
 
     readiness.mark_stage(SystemStage::IamReady);
     publish_ready_when_runtime_ready(readiness.as_ref(), state_manager.as_deref()).await
@@ -115,7 +118,7 @@ fn spawn_iam_recovery_task(
             shutdown_token,
             move || {
                 let store = init_store.clone();
-                Box::pin(async move { attempt_init_iam_sys(store).await })
+                Box::pin(async move { attempt_init_iam_sys(store).await.map(|_| ()) })
             },
             move || {
                 let store = finalize_store.clone();
@@ -313,7 +316,9 @@ pub(crate) fn reset_test_failure_counter() {
     TEST_REMAINING_FAILURES.store(u64::MAX, Ordering::SeqCst);
 }
 
-async fn attempt_init_iam_sys(store: Arc<ECStore>) -> std::result::Result<(), std::io::Error> {
+async fn attempt_init_iam_sys(
+    store: Arc<ECStore>,
+) -> std::result::Result<Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>>, std::io::Error> {
     if should_fail_test_init_attempt() {
         return Err(std::io::Error::other("forced test IAM bootstrap failure"));
     }
@@ -341,8 +346,8 @@ pub(crate) async fn bootstrap_or_defer_iam_init(
     server_ctx: Arc<ServerContextSlot>,
 ) -> Result<IamBootstrapDisposition> {
     match attempt_init_iam_sys(store.clone()).await {
-        Ok(()) => {
-            AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx)?;
+        Ok(iam) => {
+            AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx, iam)?;
             readiness.mark_stage(SystemStage::IamReady);
             return Ok(IamBootstrapDisposition::ReadyInline);
         }