Inject GlobalReadiness into HTTP server pipeline and gate traffic until FullReady (#1255)

crates/common/src/lib.rs

@@ -19,6 +19,10 @@ pub mod globals;
 pub mod heal_channel;
 pub mod last_minute;
 pub mod metrics;
+mod readiness;
+
+pub use globals::*;
+pub use readiness::{GlobalReadiness, SystemStage};
 
 // is ','
 pub static DEFAULT_DELIMITER: u8 = 44;

crates/common/src/readiness.rs

@@ -0,0 +1,136 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::atomic::{AtomicU8, Ordering};
+
+/// Represents the various stages of system startup
+#[repr(u8)]
+pub enum SystemStage {
+    Booting = 0,
+    StorageReady = 1, // Disks online, Quorum met
+    IamReady = 2,     // Users and Policies loaded into cache
+    FullReady = 3,    // System ready to serve all traffic
+}
+
+/// Global readiness tracker for the service
+/// This struct uses atomic operations to track the readiness status of various components
+/// of the service in a thread-safe manner.
+pub struct GlobalReadiness {
+    status: AtomicU8,
+}
+
+impl Default for GlobalReadiness {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl GlobalReadiness {
+    /// Create a new GlobalReadiness instance with initial status as Starting
+    /// # Returns
+    /// A new instance of GlobalReadiness
+    pub fn new() -> Self {
+        Self {
+            status: AtomicU8::new(SystemStage::Booting as u8),
+        }
+    }
+
+    /// Update the system to a new stage
+    ///
+    /// # Arguments
+    /// * `step` - The SystemStage step to mark as ready
+    pub fn mark_stage(&self, step: SystemStage) {
+        self.status.fetch_max(step as u8, Ordering::SeqCst);
+    }
+
+    /// Check if the service is fully ready
+    /// # Returns
+    /// `true` if the service is fully ready, `false` otherwise
+    pub fn is_ready(&self) -> bool {
+        self.status.load(Ordering::SeqCst) == SystemStage::FullReady as u8
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+    use std::thread;
+
+    #[test]
+    fn test_initial_state() {
+        let readiness = GlobalReadiness::new();
+        assert!(!readiness.is_ready());
+        assert_eq!(readiness.status.load(Ordering::SeqCst), SystemStage::Booting as u8);
+    }
+
+    #[test]
+    fn test_mark_stage_progression() {
+        let readiness = GlobalReadiness::new();
+        readiness.mark_stage(SystemStage::StorageReady);
+        assert!(!readiness.is_ready());
+        assert_eq!(readiness.status.load(Ordering::SeqCst), SystemStage::StorageReady as u8);
+
+        readiness.mark_stage(SystemStage::IamReady);
+        assert!(!readiness.is_ready());
+        assert_eq!(readiness.status.load(Ordering::SeqCst), SystemStage::IamReady as u8);
+
+        readiness.mark_stage(SystemStage::FullReady);
+        assert!(readiness.is_ready());
+    }
+
+    #[test]
+    fn test_no_regression() {
+        let readiness = GlobalReadiness::new();
+        readiness.mark_stage(SystemStage::FullReady);
+        readiness.mark_stage(SystemStage::IamReady); // Should not regress
+        assert!(readiness.is_ready());
+    }
+
+    #[test]
+    fn test_concurrent_marking() {
+        let readiness = Arc::new(GlobalReadiness::new());
+        let mut handles = vec![];
+
+        for _ in 0..10 {
+            let r = Arc::clone(&readiness);
+            handles.push(thread::spawn(move || {
+                r.mark_stage(SystemStage::StorageReady);
+                r.mark_stage(SystemStage::IamReady);
+                r.mark_stage(SystemStage::FullReady);
+            }));
+        }
+
+        for h in handles {
+            h.join().unwrap();
+        }
+
+        assert!(readiness.is_ready());
+    }
+
+    #[test]
+    fn test_is_ready_only_at_full_ready() {
+        let readiness = GlobalReadiness::new();
+        assert!(!readiness.is_ready());
+
+        readiness.mark_stage(SystemStage::StorageReady);
+        assert!(!readiness.is_ready());
+
+        readiness.mark_stage(SystemStage::IamReady);
+        assert!(!readiness.is_ready());
+
+        readiness.mark_stage(SystemStage::FullReady);
+        assert!(readiness.is_ready());
+    }
+}

crates/ecstore/src/admin_server_info.rs

@@ -23,7 +23,7 @@ use crate::{
 };
 
 use crate::data_usage::load_data_usage_cache;
-use rustfs_common::{globals::GLOBAL_LOCAL_NODE_NAME, heal_channel::DriveState};
+use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, heal_channel::DriveState};
 use rustfs_madmin::{
     BackendDisks, Disk, ErasureSetInfo, ITEM_INITIALIZING, ITEM_OFFLINE, ITEM_ONLINE, InfoMessage, ServerProperties,
 };

crates/ecstore/src/metrics_realtime.rs

@@ -19,11 +19,7 @@ use crate::{
     // utils::os::get_drive_stats,
 };
 use chrono::Utc;
-use rustfs_common::{
-    globals::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR},
-    heal_channel::DriveState,
-    metrics::global_metrics,
-};
+use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR, heal_channel::DriveState, metrics::global_metrics};
 use rustfs_madmin::metrics::{DiskIOStats, DiskMetric, RealtimeMetrics};
 use rustfs_utils::os::get_drive_stats;
 use serde::{Deserialize, Serialize};

crates/ecstore/src/sets.rs

@@ -40,7 +40,7 @@ use futures::future::join_all;
 use http::HeaderMap;
 use rustfs_common::heal_channel::HealOpts;
 use rustfs_common::{
-    globals::GLOBAL_LOCAL_NODE_NAME,
+    GLOBAL_LOCAL_NODE_NAME,
     heal_channel::{DriveState, HealItemType},
 };
 use rustfs_filemeta::FileInfo;

crates/ecstore/src/store.rs

@@ -55,8 +55,8 @@ use futures::future::join_all;
 use http::HeaderMap;
 use lazy_static::lazy_static;
 use rand::Rng as _;
-use rustfs_common::globals::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_HOST, GLOBAL_RUSTFS_PORT};
 use rustfs_common::heal_channel::{HealItemType, HealOpts};
+use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_HOST, GLOBAL_RUSTFS_PORT};
 use rustfs_filemeta::FileInfo;
 use rustfs_madmin::heal_commands::HealResultItem;
 use rustfs_utils::path::{SLASH_SEPARATOR, decode_dir_object, encode_dir_object, path_join_buf};

crates/iam/src/error.rs

@@ -109,6 +109,9 @@ pub enum Error {
 
     #[error("io error: {0}")]
     Io(std::io::Error),
+
+    #[error("system already initialized")]
+    IamSysAlreadyInitialized,
 }
 
 impl PartialEq for Error {
@@ -162,6 +165,7 @@ impl Clone for Error {
             Error::PolicyTooLarge => Error::PolicyTooLarge,
             Error::ConfigNotFound => Error::ConfigNotFound,
             Error::Io(e) => Error::Io(std::io::Error::new(e.kind(), e.to_string())),
+            Error::IamSysAlreadyInitialized => Error::IamSysAlreadyInitialized,
         }
     }
 }
@@ -226,6 +230,7 @@ impl From<rustfs_policy::error::Error> for Error {
             rustfs_policy::error::Error::StringError(s) => Error::StringError(s),
             rustfs_policy::error::Error::CryptoError(e) => Error::CryptoError(e),
             rustfs_policy::error::Error::ErrCredMalformed => Error::ErrCredMalformed,
+            rustfs_policy::error::Error::IamSysAlreadyInitialized => Error::IamSysAlreadyInitialized,
         }
     }
 }

crates/iam/src/lib.rs

@@ -18,30 +18,58 @@ use rustfs_ecstore::store::ECStore;
 use std::sync::{Arc, OnceLock};
 use store::object::ObjectStore;
 use sys::IamSys;
-use tracing::{debug, instrument};
+use tracing::{error, info, instrument};
 
 pub mod cache;
 pub mod error;
 pub mod manager;
 pub mod store;
-pub mod utils;
-
 pub mod sys;
+pub mod utils;
 
 static IAM_SYS: OnceLock<Arc<IamSys<ObjectStore>>> = OnceLock::new();
 
 #[instrument(skip(ecstore))]
 pub async fn init_iam_sys(ecstore: Arc<ECStore>) -> Result<()> {
-    debug!("init iam system");
-    let s = IamCache::new(ObjectStore::new(ecstore)).await;
+    if IAM_SYS.get().is_some() {
+        info!("IAM system already initialized, skipping.");
+        return Ok(());
+    }
 
-    IAM_SYS.get_or_init(move || IamSys::new(s).into());
+    info!("Starting IAM system initialization sequence...");
+
+    // 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
+    let iam_instance = Arc::new(IamSys::new(cache_manager));
+
+    // 4. Securely set the global singleton
+    if IAM_SYS.set(iam_instance).is_err() {
+        error!("Critical: Race condition detected during IAM initialization!");
+        return Err(Error::IamSysAlreadyInitialized);
+    }
+
+    info!("IAM system initialization completed successfully.");
     Ok(())
 }
 
 #[inline]
 pub fn get() -> Result<Arc<IamSys<ObjectStore>>> {
-    IAM_SYS.get().map(Arc::clone).ok_or(Error::IamSysNotInitialized)
+    let sys = IAM_SYS.get().map(Arc::clone).ok_or(Error::IamSysNotInitialized)?;
+
+    // Double-check the internal readiness state. The OnceLock is only set
+    // after initialization and data loading complete, so this is a defensive
+    // guard to ensure callers never operate on a partially initialized system.
+    if !sys.is_ready() {
+        return Err(Error::IamSysNotInitialized);
+    }
+
+    Ok(sys)
 }
 
 pub fn get_global_iam_sys() -> Option<Arc<IamSys<ObjectStore>>> {

crates/iam/src/manager.rs

@@ -37,6 +37,7 @@ use rustfs_policy::{
 use rustfs_utils::path::path_join_buf;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
+use std::sync::atomic::AtomicU8;
 use std::{
     collections::{HashMap, HashSet},
     sync::{
@@ -76,9 +77,19 @@ fn get_iam_format_file_path() -> String {
     path_join_buf(&[&IAM_CONFIG_PREFIX, IAM_FORMAT_FILE])
 }
 
+#[repr(u8)]
+#[derive(Debug, PartialEq)]
+pub enum IamState {
+    Uninitialized = 0,
+    Loading = 1,
+    Ready = 2,
+    Error = 3,
+}
+
 pub struct IamCache<T> {
     pub cache: Cache,
     pub api: T,
+    pub state: Arc<AtomicU8>,
     pub loading: Arc<AtomicBool>,
     pub roles: HashMap<ARN, Vec<String>>,
     pub send_chan: Sender<i64>,
@@ -89,12 +100,19 @@ impl<T> IamCache<T>
 where
     T: Store,
 {
+    /// Create a new IAM system instance
+    /// # Arguments
+    /// * `api` - The storage backend implementing the Store trait
+    ///
+    /// # Returns
+    /// An Arc-wrapped instance of IamSystem
     pub(crate) async fn new(api: T) -> Arc<Self> {
         let (sender, receiver) = mpsc::channel::<i64>(100);
 
         let sys = Arc::new(Self {
             api,
             cache: Cache::default(),
+            state: Arc::new(AtomicU8::new(IamState::Uninitialized as u8)),
             loading: Arc::new(AtomicBool::new(false)),
             send_chan: sender,
             roles: HashMap::new(),
@@ -105,10 +123,32 @@ where
         sys
     }
 
+    /// Initialize the IAM system
     async fn init(self: Arc<Self>, receiver: Receiver<i64>) -> Result<()> {
+        self.state.store(IamState::Loading as u8, Ordering::SeqCst);
+        // Ensure the IAM format file is persisted first
         self.clone().save_iam_formatter().await?;
-        self.clone().load().await?;
 
+        // Critical: Load all existing users/policies into memory cache
+        const MAX_RETRIES: usize = 3;
+        for attempt in 0..MAX_RETRIES {
+            if let Err(e) = self.clone().load().await {
+                if attempt == MAX_RETRIES - 1 {
+                    self.state.store(IamState::Error as u8, Ordering::SeqCst);
+                    error!("IAM fail to load initial data after {} attempts: {:?}", MAX_RETRIES, e);
+                    return Err(e);
+                } else {
+                    warn!("IAM load failed, retrying... attempt {}", attempt + 1);
+                    tokio::time::sleep(Duration::from_secs(1)).await;
+                }
+            } else {
+                break;
+            }
+        }
+        self.state.store(IamState::Ready as u8, Ordering::SeqCst);
+        info!("IAM System successfully initialized and marked as READY");
+
+        // Background ticker for synchronization
         // Check if environment variable is set
         let skip_background_task = std::env::var("RUSTFS_SKIP_BACKGROUND_TASK").is_ok();
 
@@ -152,6 +192,11 @@ where
         Ok(())
     }
 
+    /// Check if IAM system is ready
+    pub fn is_ready(&self) -> bool {
+        self.state.load(Ordering::SeqCst) == IamState::Ready as u8
+    }
+
     async fn _notify(&self) {
         self.send_chan.send(OffsetDateTime::now_utc().unix_timestamp()).await.unwrap();
     }

crates/iam/src/store/object.rs

@@ -38,7 +38,7 @@ use std::sync::LazyLock;
 use std::{collections::HashMap, sync::Arc};
 use tokio::sync::mpsc::{self, Sender};
 use tokio_util::sync::CancellationToken;
-use tracing::{info, warn};
+use tracing::{debug, error, info, warn};
 
 pub static IAM_CONFIG_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam"));
 pub static IAM_CONFIG_USERS_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam/users/"));
@@ -341,6 +341,27 @@ impl ObjectStore {
         Ok(policies)
     }
 
+    /// Checks if the underlying ECStore is ready for metadata operations.
+    /// This prevents silent failures during the storage boot-up phase.
+    ///
+    /// Performs a lightweight probe by attempting to read a known configuration object.
+    /// If the object is not found, it indicates the storage metadata is not ready.
+    /// The upper-level caller should handle retries if needed.
+    async fn check_storage_readiness(&self) -> Result<()> {
+        // Probe path for a fixed object under the IAM root prefix.
+        // If it doesn't exist, the system bucket or metadata is not ready.
+        let probe_path = format!("{}/format.json", *IAM_CONFIG_PREFIX);
+
+        match read_config(self.object_api.clone(), &probe_path).await {
+            Ok(_) => Ok(()),
+            Err(rustfs_ecstore::error::StorageError::ConfigNotFound) => Err(Error::other(format!(
+                "Storage metadata not ready: probe object '{}' not found (expected IAM config to be initialized)",
+                probe_path
+            ))),
+            Err(e) => Err(e.into()),
+        }
+    }
+
     // async fn load_policy(&self, name: &str) -> Result<PolicyDoc> {
     //     let mut policy = self
     //         .load_iam_config::<PolicyDoc>(&format!("config/iam/policies/{name}/policy.json"))
@@ -398,13 +419,50 @@ impl Store for ObjectStore {
 
         Ok(serde_json::from_slice(&data)?)
     }
+    /// Saves IAM configuration with a retry mechanism on failure.
+    ///
+    /// Attempts to save the IAM configuration up to 5 times if the storage layer is not ready,
+    /// using exponential backoff between attempts (starting at 200ms, doubling each retry).
+    ///
+    /// # Arguments
+    ///
+    /// * `item` - The IAM configuration item to save, must implement `Serialize` and `Send`.
+    /// * `path` - The path where the configuration will be saved.
+    ///
+    /// # Returns
+    ///
+    /// * `Result<()>` - `Ok(())` on success, or an `Error` if all attempts fail.
     #[tracing::instrument(level = "debug", skip(self, item, path))]
     async fn save_iam_config<Item: Serialize + Send>(&self, item: Item, path: impl AsRef<str> + Send) -> Result<()> {
         let mut data = serde_json::to_vec(&item)?;
         data = Self::encrypt_data(&data)?;
 
-        save_config(self.object_api.clone(), path.as_ref(), data).await?;
-        Ok(())
+        let mut attempts = 0;
+        let max_attempts = 5;
+        let path_ref = path.as_ref();
+
+        loop {
+            match save_config(self.object_api.clone(), path_ref, data.clone()).await {
+                Ok(_) => {
+                    debug!("Successfully saved IAM config to {}", path_ref);
+                    return Ok(());
+                }
+                Err(e) if attempts < max_attempts => {
+                    attempts += 1;
+                    // Exponential backoff: 200ms, 400ms, 800ms...
+                    let wait_ms = 200 * (1 << attempts);
+                    warn!(
+                        "Storage layer not ready for IAM write (attempt {}/{}). Retrying in {}ms. Path: {}, Error: {:?}",
+                        attempts, max_attempts, wait_ms, path_ref, e
+                    );
+                    tokio::time::sleep(std::time::Duration::from_millis(wait_ms)).await;
+                }
+                Err(e) => {
+                    error!("Final failure saving IAM config to {}: {:?}", path_ref, e);
+                    return Err(e.into());
+                }
+            }
+        }
     }
     async fn delete_iam_config(&self, path: impl AsRef<str> + Send) -> Result<()> {
         delete_config(self.object_api.clone(), path.as_ref()).await?;
@@ -418,8 +476,16 @@ impl Store for ObjectStore {
         user_identity: UserIdentity,
         _ttl: Option<usize>,
     ) -> Result<()> {
-        self.save_iam_config(user_identity, get_user_identity_path(name, user_type))
-            .await
+        // Pre-check storage health
+        self.check_storage_readiness().await?;
+
+        let path = get_user_identity_path(name, user_type);
+        debug!("Saving IAM identity to path: {}", path);
+
+        self.save_iam_config(user_identity, path).await.map_err(|e| {
+            error!("ObjectStore save failure for {}: {:?}", name, e);
+            e
+        })
     }
     async fn delete_user_identity(&self, name: &str, user_type: UserType) -> Result<()> {
         self.delete_iam_config(get_user_identity_path(name, user_type))

crates/iam/src/sys.rs

@@ -67,6 +67,13 @@ pub struct IamSys<T> {
 }
 
 impl<T: Store> IamSys<T> {
+    /// Create a new IamSys instance with the given IamCache store
+    ///
+    /// # Arguments
+    /// * `store` - An Arc to the IamCache instance
+    ///
+    /// # Returns
+    /// A new instance of IamSys
     pub fn new(store: Arc<IamCache<T>>) -> Self {
         tokio::spawn(async move {
             match opa::lookup_config().await {
@@ -87,6 +94,11 @@ impl<T: Store> IamSys<T> {
             roles_map: HashMap::new(),
         }
     }
+
+    /// Check if the IamSys has a watcher configured
+    ///
+    /// # Returns
+    /// `true` if a watcher is configured, `false` otherwise
     pub fn has_watcher(&self) -> bool {
         self.store.api.has_watcher()
     }
@@ -859,6 +871,11 @@ impl<T: Store> IamSys<T> {
 
         self.get_combined_policy(&policies).await.is_allowed(args).await
     }
+
+    /// Check if the underlying store is ready
+    pub fn is_ready(&self) -> bool {
+        self.store.is_ready()
+    }
 }
 
 fn is_allowed_by_session_policy(args: &Args<'_>) -> (bool, bool) {

crates/policy/src/error.rs

@@ -89,6 +89,7 @@ pub enum Error {
 
     #[error("invalid access_key")]
     InvalidAccessKey,
+
     #[error("action not allowed")]
     IAMActionNotAllowed,
 
@@ -106,6 +107,9 @@ pub enum Error {
 
     #[error("io error: {0}")]
     Io(std::io::Error),
+
+    #[error("system already initialized")]
+    IamSysAlreadyInitialized,
 }
 
 impl Error {

crates/protos/src/lib.rs

@@ -16,7 +16,7 @@
 mod generated;
 
 use proto_gen::node_service::node_service_client::NodeServiceClient;
-use rustfs_common::globals::{GLOBAL_CONN_MAP, GLOBAL_ROOT_CERT, evict_connection};
+use rustfs_common::{GLOBAL_CONN_MAP, GLOBAL_ROOT_CERT, evict_connection};
 use std::{error::Error, time::Duration};
 use tonic::{
     Request, Status,

crates/s3select-api/src/lib.rs

@@ -12,10 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::fmt::Display;
-
 use datafusion::{common::DataFusionError, sql::sqlparser::parser::ParserError};
 use snafu::{Backtrace, Location, Snafu};
+use std::fmt::Display;
 
 pub mod object_store;
 pub mod query;

crates/s3select-api/src/query/mod.rs

@@ -12,9 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
 use s3s::dto::SelectObjectContentInput;
+use std::sync::Arc;
 
 pub mod analyzer;
 pub mod ast;

crates/s3select-api/src/query/session.rs

@@ -12,20 +12,17 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
+use crate::query::Context;
+use crate::{QueryError, QueryResult, object_store::EcObjectStore};
 use datafusion::{
     execution::{SessionStateBuilder, context::SessionState, runtime_env::RuntimeEnvBuilder},
     parquet::data_type::AsBytes,
     prelude::SessionContext,
 };
 use object_store::{ObjectStore, memory::InMemory, path::Path};
+use std::sync::Arc;
 use tracing::error;
 
-use crate::{QueryError, QueryResult, object_store::EcObjectStore};
-
-use super::Context;
-
 #[derive(Clone)]
 pub struct SessionCtx {
     _desc: Arc<SessionCtxDesc>,