fix(iam): prevent transient IAM walk timeout from crashing startup (#3188)

* fix(iam): prevent transient IAM walk timeout from crashing startup IAM startup performs a blocking full metadata walk on `.rustfs.sys/config/iam/`. When that distributed walk times out (e.g. disk pressure after cluster reboot), the old code treated the failure as fatal and exited the process, causing a systemd restart loop. Changes: - Add `startup_iam.rs`: attempt IAM init, enter degraded mode on failure, spawn background retry task with exponential backoff (5s→10s→20s→30s cap) - Log level escalates to ERROR after 12 retries (~5 min) to aid diagnosis - `/health/ready` returns 503 until IAM recovers; IAM-dependent ops return `IamSysNotInitialized` (existing fail-closed behavior preserved) - Fix admin path boundary matching: `/minio/administrator` no longer falsely matches as admin prefix - Normalize Content-Length: 0 for admin GET requests with empty body Fixes #3175 * fix(iam): move constant assertion into const block Fixes clippy::assertions-on-constants warning on IAM_RETRY_ESCALATION_THRESHOLD assertion. * fix(iam): address PR review comments - Replace OnceLock with AtomicU64 sentinel for test isolation; add reset_test_failure_counter() for integration tests - Use u32::try_from() instead of `as u32` narrowing cast in compute_backoff_interval - Rename misleading test; update to verify finalize retry behavior - Restructure spawn_iam_recovery_task into init-retry and finalize-retry phases so transient readiness failures are retried instead of leaving the server permanently degraded * fix(iam): gate test hooks behind debug_assertions - reset_test_failure_counter() now stores sentinel (u64::MAX) to correctly trigger env var re-read on next call - RUSTFS_TEST_IAM_FAIL_INIT_ATTEMPTS only honored in debug builds - RUSTFS_TEST_IAM_RETRY_INTERVAL_MS only honored in debug builds * test(iam): cover deferred bootstrap recovery Add a dedicated embedded deferred-IAM integration test in a separate test binary to avoid process-global startup collisions. Strengthen startup IAM recovery coverage with focused unit tests and keep the existing embedded smoke test isolated while carrying the manual test license header update in the same change set. * fix(startup): tighten deferred IAM recovery path Adopt follow-up review feedback by silencing misleading app-context warnings after IAM recovery, reusing boundary-aware path prefix checks in the readiness gate, and tying deferred IAM recovery retries to server shutdown tokens. Keep the deferred IAM embedded integration coverage and startup recovery unit coverage green after the follow-up hardening. * refactor(startup): simplify IAM recovery task Collapse the deferred IAM recovery implementation back to a concrete production flow instead of keeping boxed callback seams in the runtime path. Keep only stable backoff unit coverage in startup_iam and rely on the embedded deferred bootstrap integration test for end-to-end recovery behavior. * refactor(startup): trim IAM recovery test scaffolding Keep the concrete deferred IAM recovery path intact while removing bulky test-only async loop scaffolding from startup_iam. Retain the stable backoff unit checks and rely on the embedded deferred bootstrap integration test for end-to-end recovery coverage. * fix: apply code review improvements from PR #3188 review - Simplify RecoveryFuture type alias by removing unnecessary lifetime - Fix finalize_iam_recovery to return Err if app context unavailable - Update bootstrap_or_defer_iam_init doc comment to reflect Err case - Use boundary-aware has_path_prefix for admin path matching in utils.rs - Add test for adminx boundary rejection in utils.rs and layer.rs - Improve embedded deferred IAM test with timeout wrapper * style: merge has_path_prefix import into existing use block * fix(iam): address final review follow-ups - fix main startup readiness publication to pass ServiceStateManager correctly - centralize IAM test env keys in rustfs_config and reuse them in runtime/tests - keep deferred IAM bootstrap validation aligned with the final review fixes * fix: isolate listing timeouts from drive health Keep walk_dir scanner timeouts request-scoped instead of marking local drives faulty. Add regression coverage for follow-up bucket info, set-level list_path, and system-prefix listings after prior walk timeouts. * test(iam): gate deferred bootstrap test to debug Align the deferred IAM embedded integration test with debug-only IAM fault injection hooks so release-profile runs do not assert deferred bootstrap behavior that cannot be triggered. * test(ecstore): bound prior walk timeout regressions - set walk_dir stall timeout explicitly in prior-timeout listing tests - keep the system-prefix follow-up listing scoped to the same base dir - assert the expected directory entry so the timeout regression test stays fast and stable * fmt
2026-08-12 16:16:55 +00:00 · 2026-06-03 22:37:25 +08:00
parent 0b69f363d6
commit f49827fc58
17 changed files with 993 additions and 83 deletions
@@ -69,6 +69,10 @@ pub const DEFAULT_TRANSITION_QUEUE_CAPACITY: usize = 1000;
 pub const DEFAULT_TRANSITION_QUEUE_SEND_TIMEOUT_MS: usize = 100;
 /// Test-only fault injection env var that forces the immediate transition enqueue timeout path.
 pub const ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT: &str = "RUSTFS_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT";
+/// Test-only fault injection env var that forces a number of IAM bootstrap failures.
+pub const ENV_TEST_IAM_FAIL_INIT_ATTEMPTS: &str = "RUSTFS_TEST_IAM_FAIL_INIT_ATTEMPTS";
+/// Test-only env var that overrides the deferred IAM retry interval in debug builds.
+pub const ENV_TEST_IAM_RETRY_INTERVAL_MS: &str = "RUSTFS_TEST_IAM_RETRY_INTERVAL_MS";
 /// Runtime env var controlling the transition worker count.
 pub const ENV_TRANSITION_WORKERS: &str = "RUSTFS_MAX_TRANSITION_WORKERS";
 /// Runtime env var controlling the absolute maximum transition workers.
@@ -1140,7 +1140,8 @@ impl DiskAPI for LocalDiskWrapper {
            "walk_dir",
            || async { self.disk.walk_dir(opts, wr).await },
            get_drive_walkdir_timeout(),
-            self.scanner_timeout_health_action(),
+            // Listing/scanner backpressure should fail only the current walk, not poison drive health.
+            TimeoutHealthAction::IgnoreFailure,
        )
        .await
    }
@@ -1604,7 +1605,7 @@ mod tests {
    }

    #[tokio::test]
-    async fn walk_dir_writer_backpressure_timeout_marks_drive_failure_by_default() {
+    async fn walk_dir_writer_backpressure_timeout_does_not_mark_drive_failure_by_default() {
        temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("1"))], async {
            let dir = tempfile::tempdir().expect("temp dir should be created");
            let endpoint =
@@ -1640,7 +1641,60 @@ mod tests {
                .await;

            assert_eq!(result.expect_err("walk_dir should time out"), DiskError::Timeout);
-            assert_eq!(wrapper.runtime_state(), RuntimeDriveHealthState::Suspect);
+            assert_eq!(wrapper.runtime_state(), RuntimeDriveHealthState::Online);
+            assert!(!wrapper.health.is_faulty());
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    async fn walk_dir_timeout_does_not_break_followup_stat_volume() {
+        temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("1"))], async {
+            let dir = tempfile::tempdir().expect("temp dir should be created");
+            let endpoint =
+                Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8")).expect("endpoint should parse");
+            let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
+            let wrapper = LocalDiskWrapper::new(disk, false);
+            let bucket = "test-bucket";
+            let object = "test-object";
+
+            wrapper.make_volume(bucket).await.expect("bucket should be created");
+
+            let mut file_info = FileInfo::new(&format!("{bucket}/{object}"), 1, 0);
+            file_info.volume = bucket.to_string();
+            file_info.name = object.to_string();
+            file_info.mod_time = Some(::time::OffsetDateTime::now_utc());
+            file_info.erasure.index = 1;
+
+            wrapper
+                .write_metadata("", bucket, object, file_info)
+                .await
+                .expect("object metadata should be written");
+
+            let mut writer = PendingWriter;
+            let walk_err = wrapper
+                .walk_dir(
+                    WalkDirOptions {
+                        bucket: bucket.to_string(),
+                        recursive: true,
+                        ..Default::default()
+                    },
+                    &mut writer,
+                )
+                .await
+                .expect_err("walk_dir should time out");
+
+            assert_eq!(walk_err, DiskError::Timeout);
+            assert_eq!(wrapper.runtime_state(), RuntimeDriveHealthState::Online);
+            assert!(!wrapper.health.is_faulty());
+
+            let info = wrapper
+                .stat_volume(bucket)
+                .await
+                .expect("follow-up bucket stat should still succeed after walk timeout");
+            assert_eq!(info.name, bucket);
+            assert_eq!(wrapper.runtime_state(), RuntimeDriveHealthState::Online);
+            assert!(!wrapper.health.is_faulty());
        })
        .await;
    }
@@ -1047,6 +1047,44 @@ async fn clone_drives() -> Vec<Option<DiskStore>> {
 #[cfg(test)]
 mod tests {
    use super::*;
+    use crate::disk::WalkDirOptions;
+    use crate::disk::disk_store::LocalDiskWrapper;
+    use crate::disk::endpoint::Endpoint;
+    use crate::disk::local::LocalDisk;
+    use crate::endpoints::{Endpoints, PoolEndpoints};
+    use crate::global::{GLOBAL_LOCAL_DISK_ID_MAP, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES};
+    use crate::store::init_local_disks;
+    use rustfs_filemeta::FileInfo;
+    use serial_test::serial;
+    use std::{
+        io,
+        pin::Pin,
+        task::{Context, Poll},
+    };
+    use tempfile::TempDir;
+    use tokio::io::AsyncWrite;
+
+    struct PendingWriter;
+
+    impl AsyncWrite for PendingWriter {
+        fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll<io::Result<usize>> {
+            Poll::Pending
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    async fn reset_local_disk_globals() {
+        GLOBAL_LOCAL_DISK_MAP.write().await.clear();
+        GLOBAL_LOCAL_DISK_ID_MAP.write().await.clear();
+        GLOBAL_LOCAL_DISK_SET_DRIVES.write().await.clear();
+    }

    #[derive(Debug)]
    struct TestPeerS3Client {
@@ -1153,6 +1191,77 @@ mod tests {
        client.cancel_token.cancel();
    }

+    #[tokio::test]
+    #[serial]
+    async fn local_get_bucket_info_survives_prior_walk_timeout() {
+        reset_local_disk_globals().await;
+
+        let temp_dir = TempDir::new().expect("create temp dir for local peer listing regression");
+        let disk_path = temp_dir.path().join("disk1");
+        std::fs::create_dir_all(&disk_path).expect("create disk path");
+
+        let mut endpoint = Endpoint::try_from(disk_path.to_str().expect("disk path to str")).expect("endpoint");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(0);
+
+        let endpoint_pools = EndpointServerPools(vec![PoolEndpoints {
+            legacy: false,
+            set_count: 1,
+            drives_per_set: 1,
+            endpoints: Endpoints::from(vec![endpoint.clone()]),
+            cmd_line: "local-get-bucket-info-survives-prior-walk-timeout".to_string(),
+            platform: "test".to_string(),
+        }]);
+
+        init_local_disks(endpoint_pools).await.expect("init local disks");
+
+        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
+        let wrapper = crate::disk::Disk::Local(Box::new(LocalDiskWrapper::new(disk, false)));
+        let disk_store: DiskStore = Arc::new(wrapper);
+        let bucket = "test-bucket";
+        let object = "test-object";
+
+        disk_store.make_volume(bucket).await.expect("bucket should be created");
+
+        let mut file_info = FileInfo::new(&format!("{bucket}/{object}"), 1, 0);
+        file_info.volume = bucket.to_string();
+        file_info.name = object.to_string();
+        file_info.mod_time = Some(::time::OffsetDateTime::now_utc());
+        file_info.erasure.index = 1;
+
+        disk_store
+            .write_metadata("", bucket, object, file_info)
+            .await
+            .expect("object metadata should be written");
+
+        temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("1"))], async {
+            let mut writer = PendingWriter;
+            let walk_err = disk_store
+                .walk_dir(
+                    WalkDirOptions {
+                        bucket: bucket.to_string(),
+                        recursive: true,
+                        ..Default::default()
+                    },
+                    &mut writer,
+                )
+                .await
+                .expect_err("walk_dir should time out against a non-draining writer");
+
+            assert_eq!(walk_err, DiskError::Timeout);
+
+            let info = LocalPeerS3Client::new(None, Some(vec![0]))
+                .get_bucket_info(bucket, &BucketOptions::default())
+                .await
+                .expect("bucket info should still succeed after prior walk timeout");
+            assert_eq!(info.name, bucket);
+        })
+        .await;
+
+        reset_local_disk_globals().await;
+    }
+
    #[test]
    fn test_reduce_pool_write_quorum_uses_only_pool_participants() {
        let clients = vec![
@@ -4649,6 +4649,9 @@ mod tests {
    use super::*;
    use crate::disk::CHECK_PART_UNKNOWN;
    use crate::disk::CHECK_PART_VOLUME_NOT_FOUND;
+    use crate::disk::RUSTFS_META_BUCKET;
+    use crate::disk::STORAGE_FORMAT_FILE;
+    use crate::disk::WalkDirOptions;
    use crate::disk::endpoint::Endpoint;
    use crate::disk::error::DiskError;
    use crate::disk::health_state::RuntimeDriveHealthState;
@@ -4656,7 +4659,9 @@ mod tests {
    use crate::global::{is_dist_erasure, is_erasure, is_erasure_sd, update_erasure_type};
    use crate::store_api::{CompletePart, ObjectInfo};
    use crate::store_init::save_format_file;
+    use crate::store_list_objects::ListPathOptions;
    use rustfs_filemeta::ErasureInfo;
+    use rustfs_filemeta::MetaCacheEntry;
    use rustfs_filemeta::ReplicationState;
    use rustfs_lock::client::local::LocalClient;
    use rustfs_lock::{LockError, LockInfo, LockResponse, LockStats};
@@ -5935,6 +5940,192 @@ mod tests {
        drop(temp_dirs);
    }

+    #[tokio::test]
+    async fn list_path_still_uses_disk_after_prior_walk_timeout() {
+        use std::pin::Pin;
+        use std::task::{Context, Poll};
+        use tokio::io::AsyncWrite;
+
+        struct PendingWriter;
+
+        impl AsyncWrite for PendingWriter {
+            fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll<std::io::Result<usize>> {
+                Poll::Pending
+            }
+
+            fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+                Poll::Ready(Ok(()))
+            }
+
+            fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+                Poll::Ready(Ok(()))
+            }
+        }
+
+        let format = FormatV3::new(1, 1);
+        let (temp_dir, endpoint, disk) = make_formatted_local_disk_for_info_test(0, &format).await;
+        let bucket = "bucket";
+        let object = "obj";
+
+        disk.make_volume(bucket).await.expect("bucket should be created");
+        let metadata_path = format!("{object}/{STORAGE_FORMAT_FILE}");
+        disk.write_all(bucket, &metadata_path, bytes::Bytes::from_static(b"not-an-xl-meta"))
+            .await
+            .expect("metadata file should be created");
+
+        let set_disks = SetDisks::new(
+            "test-owner".to_string(),
+            Arc::new(RwLock::new(vec![Some(disk.clone())])),
+            1,
+            0,
+            0,
+            0,
+            vec![endpoint],
+            format,
+            Vec::new(),
+        )
+        .await;
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("1")),
+                (rustfs_config::ENV_DRIVE_WALKDIR_STALL_TIMEOUT_SECS, Some("1")),
+            ],
+            async {
+                let mut writer = PendingWriter;
+                let walk_err = disk
+                    .walk_dir(
+                        WalkDirOptions {
+                            bucket: bucket.to_string(),
+                            recursive: true,
+                            ..Default::default()
+                        },
+                        &mut writer,
+                    )
+                    .await
+                    .expect_err("walk_dir should time out");
+                assert_eq!(walk_err, DiskError::Timeout);
+                assert_eq!(disk.runtime_state(), RuntimeDriveHealthState::Online);
+
+                let (tx, mut rx) = tokio::sync::mpsc::channel::<MetaCacheEntry>(4);
+                set_disks
+                    .list_path(
+                        CancellationToken::new(),
+                        ListPathOptions {
+                            bucket: bucket.to_string(),
+                            recursive: true,
+                            ..Default::default()
+                        },
+                        tx,
+                    )
+                    .await
+                    .expect("list_path should still succeed after prior walk timeout");
+
+                let entry = rx.recv().await.expect("listing should yield the object entry");
+                assert_eq!(entry.name, object);
+                assert_eq!(disk.runtime_state(), RuntimeDriveHealthState::Online);
+            },
+        )
+        .await;
+
+        drop(temp_dir);
+    }
+
+    #[tokio::test]
+    async fn list_path_system_prefix_survives_prior_walk_timeout() {
+        use std::pin::Pin;
+        use std::task::{Context, Poll};
+        use tokio::io::AsyncWrite;
+
+        struct PendingWriter;
+
+        impl AsyncWrite for PendingWriter {
+            fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll<std::io::Result<usize>> {
+                Poll::Pending
+            }
+
+            fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+                Poll::Ready(Ok(()))
+            }
+
+            fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+                Poll::Ready(Ok(()))
+            }
+        }
+
+        let format = FormatV3::new(1, 1);
+        let (temp_dir, endpoint, disk) = make_formatted_local_disk_for_info_test(0, &format).await;
+        let object = "config/iam/sts/test/identity.json";
+
+        let metadata_path = format!("{object}/{STORAGE_FORMAT_FILE}");
+        disk.write_all(RUSTFS_META_BUCKET, &metadata_path, bytes::Bytes::from_static(b"not-an-xl-meta"))
+            .await
+            .expect("system path metadata file should be created");
+
+        let set_disks = SetDisks::new(
+            "test-owner".to_string(),
+            Arc::new(RwLock::new(vec![Some(disk.clone())])),
+            1,
+            0,
+            0,
+            0,
+            vec![endpoint],
+            format,
+            Vec::new(),
+        )
+        .await;
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("1")),
+                (rustfs_config::ENV_DRIVE_WALKDIR_STALL_TIMEOUT_SECS, Some("1")),
+            ],
+            async {
+                let mut writer = PendingWriter;
+                let walk_err = disk
+                    .walk_dir(
+                        WalkDirOptions {
+                            bucket: RUSTFS_META_BUCKET.to_string(),
+                            base_dir: "config/iam/".to_string(),
+                            recursive: true,
+                            ..Default::default()
+                        },
+                        &mut writer,
+                    )
+                    .await
+                    .expect_err("walk_dir should time out");
+                assert_eq!(walk_err, DiskError::Timeout);
+                assert_eq!(disk.runtime_state(), RuntimeDriveHealthState::Online);
+
+                let (tx, mut rx) = tokio::sync::mpsc::channel::<MetaCacheEntry>(4);
+                set_disks
+                    .list_path(
+                        CancellationToken::new(),
+                        ListPathOptions {
+                            bucket: RUSTFS_META_BUCKET.to_string(),
+                            base_dir: "config/iam/".to_string(),
+                            recursive: true,
+                            ..Default::default()
+                        },
+                        tx,
+                    )
+                    .await
+                    .expect("system prefix list_path should still succeed after prior walk timeout");
+
+                let entry = rx.recv().await.expect("listing should yield the system-path entry");
+                assert_eq!(entry.name, "config/iam/sts/");
+                assert!(
+                    entry.is_dir(),
+                    "system prefix listing should still yield a directory entry after timeout recovery"
+                );
+                assert_eq!(disk.runtime_state(), RuntimeDriveHealthState::Online);
+            },
+        )
+        .await;
+
+        drop(temp_dir);
+    }
+
    #[test]
    fn test_dangling_meta_errs_count() {
        // Test counting dangling metadata errors