fix(iam): keep error state on initial load failure (#2846)

Signed-off-by: houseme <housemecn@gmail.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: loverustfs <hello@rustfs.com>
Co-authored-by: 安正超 <anzhengchao@gmail.com>

crates/iam/src/lib.rs

@@ -48,7 +48,7 @@ pub async fn init_iam_sys(ecstore: Arc<ECStore>) -> Result<()> {
 
     // 2. Create the cache manager.
     // The `new` method now performs a blocking initial load from disk.
-    let cache_manager = IamCache::new(storage_adapter).await;
+    let cache_manager = IamCache::new(storage_adapter).await?;
 
     // 3. Construct the system interface
     let iam_instance = Arc::new(IamSys::new(cache_manager));

crates/iam/src/manager.rs

@@ -58,6 +58,10 @@ use tracing::{error, info};
 
 const IAM_FORMAT_FILE: &str = "format.json";
 const IAM_FORMAT_VERSION_1: i32 = 1;
+#[cfg(not(test))]
+const INITIAL_LOAD_RETRY_DELAY: Duration = Duration::from_secs(1);
+#[cfg(test)]
+const INITIAL_LOAD_RETRY_DELAY: Duration = Duration::from_millis(1);
 
 #[derive(Serialize, Deserialize)]
 struct IAMFormat {
@@ -116,7 +120,7 @@ where
     ///
     /// # Returns
     /// An Arc-wrapped instance of IamSystem
-    pub(crate) async fn new(api: T) -> Arc<Self> {
+    pub(crate) async fn new(api: T) -> Result<Arc<Self>> {
         let (sender, receiver) = mpsc::channel::<i64>(100);
 
         let sys = Arc::new(Self {
@@ -132,8 +136,8 @@ where
             last_sync_duration_millis: AtomicU64::new(0),
         });
 
-        sys.clone().init(receiver).await.unwrap();
+        sys.clone().init(receiver).await?;
-        sys
+        Ok(sys)
     }
 
     /// Initialize the IAM system
@@ -144,19 +148,26 @@ where
 
         // Critical: Load all existing users/policies into memory cache
         const MAX_RETRIES: usize = 3;
+        let mut load_error = None;
         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);
                     warn!("IAM failed to load initial data after {} attempts: {:?}", MAX_RETRIES, e);
+                    load_error = Some(e);
                 } else {
                     warn!("IAM load failed, retrying... attempt {}", attempt + 1);
-                    tokio::time::sleep(Duration::from_secs(1)).await;
+                    tokio::time::sleep(INITIAL_LOAD_RETRY_DELAY).await;
                 }
             } else {
                 break;
             }
         }
+
+        if let Some(err) = load_error {
+            return Err(err);
+        }
+
         self.state.store(IamState::Ready as u8, Ordering::SeqCst);
         info!("IAM System successfully initialized and marked as READY");
 
@@ -2024,6 +2035,156 @@ mod tests {
     use serde_json::json;
     use std::collections::HashMap;
 
+    #[derive(Clone)]
+    struct FailingInitialLoadStore;
+
+    #[async_trait::async_trait]
+    impl Store for FailingInitialLoadStore {
+        fn has_watcher(&self) -> bool {
+            false
+        }
+
+        async fn save_iam_config<Item: Serialize + Send>(&self, _item: Item, _path: impl AsRef<str> + Send) -> Result<()> {
+            Ok(())
+        }
+
+        async fn load_iam_config<Item: serde::de::DeserializeOwned>(&self, _path: impl AsRef<str> + Send) -> Result<Item> {
+            Err(Error::ConfigNotFound)
+        }
+
+        async fn delete_iam_config(&self, _path: impl AsRef<str> + Send) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn save_user_identity(
+            &self,
+            _name: &str,
+            _user_type: UserType,
+            _item: UserIdentity,
+            _ttl: Option<usize>,
+        ) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn delete_user_identity(&self, _name: &str, _user_type: UserType) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_user_identity(&self, _name: &str, _user_type: UserType) -> Result<UserIdentity> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_user(&self, _name: &str, _user_type: UserType, _m: &mut HashMap<String, UserIdentity>) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_users(&self, _user_type: UserType, _m: &mut HashMap<String, UserIdentity>) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_secret_key(&self, _name: &str, _user_type: UserType) -> Result<String> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn save_group_info(&self, _name: &str, _item: GroupInfo) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn delete_group_info(&self, _name: &str) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_group(&self, _name: &str, _m: &mut HashMap<String, GroupInfo>) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_groups(&self, _m: &mut HashMap<String, GroupInfo>) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn save_policy_doc(&self, _name: &str, _item: PolicyDoc) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn delete_policy_doc(&self, _name: &str) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_policy(&self, _name: &str) -> Result<PolicyDoc> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_policy_doc(&self, _name: &str, _m: &mut HashMap<String, PolicyDoc>) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_policy_docs(&self, _m: &mut HashMap<String, PolicyDoc>) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn save_mapped_policy(
+            &self,
+            _name: &str,
+            _user_type: UserType,
+            _is_group: bool,
+            _item: MappedPolicy,
+            _ttl: Option<usize>,
+        ) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn delete_mapped_policy(&self, _name: &str, _user_type: UserType, _is_group: bool) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_mapped_policy(
+            &self,
+            _name: &str,
+            _user_type: UserType,
+            _is_group: bool,
+            _m: &mut HashMap<String, MappedPolicy>,
+        ) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_mapped_policies(
+            &self,
+            _user_type: UserType,
+            _is_group: bool,
+            _m: &mut HashMap<String, MappedPolicy>,
+        ) -> Result<()> {
+            Err(Error::InvalidArgument)
+        }
+
+        async fn load_all(&self, _cache: &Cache) -> Result<()> {
+            Err(Error::Io(std::io::Error::other("initial load failed")))
+        }
+    }
+
+    #[tokio::test]
+    async fn test_init_keeps_error_state_when_initial_load_fails() {
+        let (sender, receiver) = mpsc::channel::<i64>(1);
+        let sys = Arc::new(IamCache {
+            api: FailingInitialLoadStore,
+            cache: Cache::default(),
+            state: Arc::new(AtomicU8::new(IamState::Uninitialized as u8)),
+            loading: Arc::new(AtomicBool::new(false)),
+            send_chan: sender,
+            roles: HashMap::new(),
+            last_timestamp: AtomicI64::new(0),
+            sync_failures: AtomicU64::new(0),
+            sync_successes: AtomicU64::new(0),
+            last_sync_duration_millis: AtomicU64::new(0),
+        });
+
+        let result = Arc::clone(&sys).init(receiver).await;
+
+        assert!(matches!(result, Err(Error::Io(_))));
+        assert!(!sys.is_ready());
+        assert_eq!(sys.state.load(Ordering::SeqCst), IamState::Error as u8);
+        assert_eq!(sys.sync_failures.load(Ordering::Relaxed), 3);
+    }
+
     #[test]
     fn test_iam_format_new_version_1() {
         let format = IAMFormat::new_version_1();

crates/iam/src/sys.rs

@@ -1565,7 +1565,7 @@ mod tests {
         ensure_test_global_credentials();
 
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let (cred, _) = iam_sys
@@ -1588,7 +1588,7 @@ mod tests {
         ensure_test_global_credentials();
 
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let initial_expiration = OffsetDateTime::now_utc() + time::Duration::hours(2);
@@ -1641,7 +1641,7 @@ mod tests {
         ensure_test_global_credentials();
 
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let (cred, _) = iam_sys
@@ -1682,7 +1682,7 @@ mod tests {
         ensure_test_global_credentials();
 
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let parent_user = "sts-fallback-test-parent";
@@ -1763,7 +1763,7 @@ mod tests {
         ensure_test_global_credentials();
 
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let parent_user = "sts-fallback-test-parent";
@@ -1867,7 +1867,7 @@ mod tests {
     #[tokio::test]
     async fn test_sts_groups_fallback_temp_creds_receive_parent_group_policies() {
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let parent_user = "sts-fallback-test-parent";
@@ -1898,7 +1898,7 @@ mod tests {
     #[tokio::test]
     async fn test_sts_deny_only_session_policy_deny_blocks_when_iam_policies_empty() {
         let store = StsTestMockStore { empty_policies: true };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let parent_user = "sts-empty-parent-policy-test";
@@ -1938,7 +1938,7 @@ mod tests {
     #[tokio::test]
     async fn test_sts_deny_only_session_policy_allow_when_no_deny_on_action() {
         let store = StsTestMockStore { empty_policies: true };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let parent_user = "sts-empty-parent-policy-test";
@@ -1982,7 +1982,7 @@ mod tests {
     #[tokio::test]
     async fn test_load_user_notification_populates_user_and_policy_caches() {
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         iam_sys.load_user("notify-user", UserType::Reg).await.unwrap();
@@ -2003,7 +2003,7 @@ mod tests {
     #[tokio::test]
     async fn test_check_key_propagates_cache_miss_load_failure() {
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let result = iam_sys.check_key("load-failure-user").await;
@@ -2014,7 +2014,7 @@ mod tests {
     #[tokio::test]
     async fn test_prepare_auth_eval_matches_prepare_sts_auth_for_parent_policy_fallback() {
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let parent_user = "sts-fallback-test-parent";
@@ -2042,7 +2042,7 @@ mod tests {
     #[tokio::test]
     async fn test_prepare_auth_detects_existing_object_tag_in_session_policy() {
         let store = StsTestMockStore { empty_policies: true };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
         let sts_access_key = "sts-session-tag-test-user";
 
@@ -2156,7 +2156,7 @@ mod tests {
     #[tokio::test]
     async fn test_prepare_auth_detects_existing_object_tag_in_encoded_session_policy() {
         let store = StsTestMockStore { empty_policies: true };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
         let sts_access_key = "sts-session-tag-encoded-test-user";
 
@@ -2203,7 +2203,7 @@ mod tests {
     #[tokio::test]
     async fn test_prepare_auth_service_account_inherited_ignores_session_policy_tag_hint() {
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let service_account_access_key = "svc-inherited-tag-hint-test-user";
@@ -2266,7 +2266,7 @@ mod tests {
     #[tokio::test]
     async fn test_policy_db_get_skips_nonexistent_groups() {
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         // "testgroup" exists with "readwrite" policy; "nonexistent-group" does not exist in IAM.
@@ -2287,7 +2287,7 @@ mod tests {
     #[tokio::test]
     async fn test_info_policy_returns_policy_as_json_object() {
         let store = StsTestMockStore { empty_policies: false };
-        let cache_manager = IamCache::new(store).await;
+        let cache_manager = IamCache::new(store).await.unwrap();
         let iam_sys = IamSys::new(cache_manager);
 
         let policy_info = iam_sys