From 03045ff2e6d9dc0635a7ae73d4c72088ccfae8c0 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Fri, 8 May 2026 16:26:01 +0800 Subject: [PATCH] fix(iam): keep error state on initial load failure (#2846) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: houseme Co-authored-by: houseme Co-authored-by: Claude Sonnet 4.6 Co-authored-by: loverustfs Co-authored-by: 安正超 --- crates/iam/src/lib.rs | 2 +- crates/iam/src/manager.rs | 169 +++++++++++++++++++++++++++++++++++++- crates/iam/src/sys.rs | 32 ++++---- 3 files changed, 182 insertions(+), 21 deletions(-) diff --git a/crates/iam/src/lib.rs b/crates/iam/src/lib.rs index 96a2763f9..31db6cc25 100644 --- a/crates/iam/src/lib.rs +++ b/crates/iam/src/lib.rs @@ -48,7 +48,7 @@ pub async fn init_iam_sys(ecstore: Arc) -> 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)); diff --git a/crates/iam/src/manager.rs b/crates/iam/src/manager.rs index c31448355..841230b28 100644 --- a/crates/iam/src/manager.rs +++ b/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 { + pub(crate) async fn new(api: T) -> Result> { let (sender, receiver) = mpsc::channel::(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 + sys.clone().init(receiver).await?; + 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(&self, _item: Item, _path: impl AsRef + Send) -> Result<()> { + Ok(()) + } + + async fn load_iam_config(&self, _path: impl AsRef + Send) -> Result { + Err(Error::ConfigNotFound) + } + + async fn delete_iam_config(&self, _path: impl AsRef + Send) -> Result<()> { + Err(Error::InvalidArgument) + } + + async fn save_user_identity( + &self, + _name: &str, + _user_type: UserType, + _item: UserIdentity, + _ttl: Option, + ) -> 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 { + Err(Error::InvalidArgument) + } + + async fn load_user(&self, _name: &str, _user_type: UserType, _m: &mut HashMap) -> Result<()> { + Err(Error::InvalidArgument) + } + + async fn load_users(&self, _user_type: UserType, _m: &mut HashMap) -> Result<()> { + Err(Error::InvalidArgument) + } + + async fn load_secret_key(&self, _name: &str, _user_type: UserType) -> Result { + 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) -> Result<()> { + Err(Error::InvalidArgument) + } + + async fn load_groups(&self, _m: &mut HashMap) -> 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 { + Err(Error::InvalidArgument) + } + + async fn load_policy_doc(&self, _name: &str, _m: &mut HashMap) -> Result<()> { + Err(Error::InvalidArgument) + } + + async fn load_policy_docs(&self, _m: &mut HashMap) -> Result<()> { + Err(Error::InvalidArgument) + } + + async fn save_mapped_policy( + &self, + _name: &str, + _user_type: UserType, + _is_group: bool, + _item: MappedPolicy, + _ttl: Option, + ) -> 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, + ) -> Result<()> { + Err(Error::InvalidArgument) + } + + async fn load_mapped_policies( + &self, + _user_type: UserType, + _is_group: bool, + _m: &mut HashMap, + ) -> 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::(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(); diff --git a/crates/iam/src/sys.rs b/crates/iam/src/sys.rs index 7285bd947..2a084df40 100644 --- a/crates/iam/src/sys.rs +++ b/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