diff --git a/crates/config/src/constants/runtime.rs b/crates/config/src/constants/runtime.rs index 7bfc96ac4..63c59563b 100644 --- a/crates/config/src/constants/runtime.rs +++ b/crates/config/src/constants/runtime.rs @@ -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. diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index e02276d55..5b3984c20 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -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; } diff --git a/crates/ecstore/src/rpc/peer_s3_client.rs b/crates/ecstore/src/rpc/peer_s3_client.rs index 3914b99cc..7c906f361 100644 --- a/crates/ecstore/src/rpc/peer_s3_client.rs +++ b/crates/ecstore/src/rpc/peer_s3_client.rs @@ -1047,6 +1047,44 @@ async fn clone_drives() -> Vec> { #[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> { + Poll::Pending + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + 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![ diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index c47a2ec0b..d873dc7b5 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -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> { + Poll::Pending + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + 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::(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> { + Poll::Pending + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + 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::(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 diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index f1feba9ac..500f39c39 100644 --- a/rustfs/src/admin/console.rs +++ b/rustfs/src/admin/console.rs @@ -14,6 +14,7 @@ use crate::admin::handlers::health::{HealthProbe, build_health_response_parts, collect_dependency_readiness}; use crate::license::has_valid_license; +use crate::server::has_path_prefix; use crate::server::{CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE, RUSTFS_ADMIN_PREFIX, VERSION}; use crate::version::build; use axum::{ @@ -436,7 +437,7 @@ fn get_console_config_from_env() -> (bool, u32, u64, String) { /// # Returns: /// - `true` if the path is for console access, `false` otherwise. pub fn is_console_path(path: &str) -> bool { - path == FAVICON_PATH || path.starts_with(CONSOLE_PREFIX) + path == FAVICON_PATH || has_path_prefix(path, CONSOLE_PREFIX) } /// Setup comprehensive middleware stack with tower-http features diff --git a/rustfs/src/admin/utils.rs b/rustfs/src/admin/utils.rs index ac5768e09..0da40d04b 100644 --- a/rustfs/src/admin/utils.rs +++ b/rustfs/src/admin/utils.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::server::MINIO_ADMIN_PREFIX; +use crate::server::{MINIO_ADMIN_PREFIX, has_path_prefix}; use rustfs_crypto::{decrypt_data, decrypt_stream_io, encrypt_stream_io}; use s3s::{Body, S3Result, s3_error}; @@ -21,7 +21,7 @@ pub(crate) fn has_space_be(s: &str) -> bool { } pub(crate) fn is_compat_admin_request(path: &str) -> bool { - path.starts_with(MINIO_ADMIN_PREFIX) + has_path_prefix(path, MINIO_ADMIN_PREFIX) } pub(crate) async fn read_compatible_admin_body( @@ -63,6 +63,7 @@ mod tests { #[test] fn detects_compat_admin_paths_only_for_external_prefix() { assert!(is_compat_admin_request("/minio/admin/v3/list-users")); + assert!(!is_compat_admin_request("/minio/adminx/list-users")); assert!(!is_compat_admin_request("/rustfs/admin/v3/list-users")); } diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs index 494759fb3..31b5902ef 100644 --- a/rustfs/src/embedded.rs +++ b/rustfs/src/embedded.rs @@ -46,14 +46,13 @@ //! storage engine uses process-global singletons (`OnceLock`). Attempting to //! start a second server will return an error. -use crate::app::context::{AppContext, init_global_app_context}; use crate::config::Config; use crate::init::{add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system}; use crate::server::{ - ShutdownHandle, init_event_notifier, publish_ready_when_runtime_ready, shutdown_event_notifier, start_audit_system, - start_http_server, stop_audit_system, + ShutdownHandle, init_event_notifier, shutdown_event_notifier, start_audit_system, start_http_server, stop_audit_system, }; use crate::startup_fs_guard::enforce_unsupported_fs_policy; +use crate::startup_iam::{IamBootstrapDisposition, bootstrap_or_defer_iam_init}; use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr}; use rustfs_credentials::init_global_action_credentials; use rustfs_ecstore::store::init_lock_clients; @@ -74,7 +73,6 @@ use rustfs_ecstore::{ store_api::BucketOptions, update_erasure_type, }; -use rustfs_iam::init_iam_sys; use rustfs_obs::{init_obs, set_global_guard}; use rustfs_utils::net::parse_and_resolve_address; use rustls::crypto::aws_lc_rs::default_provider; @@ -433,21 +431,14 @@ impl RustFSServerBuilder { try_migrate_iam_config(store.clone()).await; // IAM. - init_iam_sys(store.clone()).await.map_err(|e| { - shutdown_embedded_server(); - ServerError::Init(format!("IAM: {e}")) - })?; - readiness.mark_stage(SystemStage::IamReady); - - // App context. - let iam_interface = rustfs_iam::get().map_err(|e| { - shutdown_embedded_server(); - ServerError::Init(format!("IAM get: {e}")) - })?; let kms_interface = rustfs_kms::get_global_kms_service_manager().unwrap_or_else(rustfs_kms::init_global_kms_service_manager); - let _app_context = - init_global_app_context(AppContext::with_default_interfaces(store.clone(), iam_interface, kms_interface)); + let iam_bootstrap = bootstrap_or_defer_iam_init(store.clone(), kms_interface, readiness.clone(), None, Some(ctx.clone())) + .await + .map_err(|e| { + shutdown_embedded_server(); + ServerError::Init(format!("IAM bootstrap setup: {e}")) + })?; // Bucket notifications. add_bucket_notification_configuration(buckets.clone()).await; @@ -457,12 +448,15 @@ impl RustFSServerBuilder { warn!("notification system: {e}"); } - publish_ready_when_runtime_ready(readiness.as_ref(), None) - .await - .map_err(|e| { - shutdown_embedded_server(); - ServerError::Init(format!("runtime readiness: {e}")) - })?; + if iam_bootstrap == IamBootstrapDisposition::ReadyInline { + crate::server::publish_ready_when_runtime_ready(readiness.as_ref(), None) + .await + .map_err(|e| { + shutdown_embedded_server(); + ServerError::Init(format!("runtime readiness: {e}")) + })?; + } + rustfs_common::set_global_init_time_now().await; let server = RustFSServer { diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs index 020f45c48..ece2610b5 100644 --- a/rustfs/src/lib.rs +++ b/rustfs/src/lib.rs @@ -68,6 +68,7 @@ pub mod profiling; pub mod protocols; pub mod server; pub mod startup_fs_guard; +pub mod startup_iam; pub mod storage; pub(crate) mod table_catalog; pub mod tls; diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index fca7c75a4..c6993c8d0 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -13,7 +13,6 @@ // limitations under the License. // Ensure the correct path for parse_license is imported -use rustfs::app::context::{AppContext, init_global_app_context}; use rustfs::init::{ add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system, init_update_check, print_server_info, }; @@ -31,10 +30,11 @@ use futures_util::future::join_all; use rustfs::capacity::capacity_integration::init_capacity_management; use rustfs::license::{current_license, init_license, license_status}; use rustfs::server::{ - ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, init_event_notifier, publish_ready_when_runtime_ready, - shutdown_event_notifier, start_audit_system, start_http_server, stop_audit_system, wait_for_shutdown, + ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, init_event_notifier, shutdown_event_notifier, + start_audit_system, start_http_server, stop_audit_system, wait_for_shutdown, }; use rustfs::startup_fs_guard::enforce_unsupported_fs_policy; +use rustfs::startup_iam::{IamBootstrapDisposition, bootstrap_or_defer_iam_init}; use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr}; use rustfs_credentials::init_global_action_credentials; use rustfs_ecstore::store::init_lock_clients; @@ -57,7 +57,7 @@ use rustfs_ecstore::{ use rustfs_heal::{ create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager, shutdown_ahm_services, }; -use rustfs_iam::{init_iam_sys, init_oidc_sys}; +use rustfs_iam::init_oidc_sys; use rustfs_obs::{init_metrics_runtime, init_obs, set_global_guard}; use rustfs_scanner::init_data_scanner; use rustfs_utils::{ @@ -349,7 +349,7 @@ async fn run(config: rustfs::config::Config) -> Result<()> { } // Initialize capacity management system init_capacity_management().await; - let state_manager = ServiceStateManager::new(); + let state_manager = Arc::new(ServiceStateManager::new()); // Update service status to Starting state_manager.update(ServiceState::Starting); @@ -522,8 +522,15 @@ async fn run(config: rustfs::config::Config) -> Result<()> { // 3. Initialize IAM System (Blocking load) // This ensures data is in memory before moving forward - init_iam_sys(store.clone()).await.map_err(Error::other)?; - readiness.mark_stage(SystemStage::IamReady); + let kms_interface = rustfs_kms::get_global_kms_service_manager().unwrap_or_else(rustfs_kms::init_global_kms_service_manager); + let iam_bootstrap = bootstrap_or_defer_iam_init( + store.clone(), + kms_interface, + readiness.clone(), + Some(state_manager.clone()), + Some(ctx.clone()), + ) + .await?; // 3a. Initialize Keystone authentication if enabled let keystone_config = rustfs_keystone::KeystoneConfig::from_env().map_err(Error::other)?; @@ -542,11 +549,6 @@ async fn run(config: rustfs::config::Config) -> Result<()> { warn!("OIDC initialization failed (non-fatal): {}", e); } - let iam_interface = - rustfs_iam::get().map_err(|e| Error::other(format!("initialize app context IAM dependency failed: {e}")))?; - let kms_interface = rustfs_kms::get_global_kms_service_manager().unwrap_or_else(rustfs_kms::init_global_kms_service_manager); - let _app_context = init_global_app_context(AppContext::with_default_interfaces(store.clone(), iam_interface, kms_interface)); - add_bucket_notification_configuration(buckets.clone()).await; // Initialize the global notification system @@ -601,8 +603,9 @@ async fn run(config: rustfs::config::Config) -> Result<()> { &server_address, jiff::Zoned::now() ); - publish_ready_when_runtime_ready(readiness.as_ref(), Some(&state_manager)).await?; - + if iam_bootstrap == IamBootstrapDisposition::ReadyInline { + rustfs::server::publish_ready_when_runtime_ready(readiness.as_ref(), Some(state_manager.as_ref())).await?; + } // Set the global RustFS initialization time to now rustfs_common::set_global_init_time_now().await; diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index 19b9e5202..ef6085d16 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -20,7 +20,7 @@ use crate::server::hybrid::HybridBody; use crate::server::{ ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, active_http_requests, collect_dependency_readiness_report, - is_admin_path, + has_path_prefix, is_admin_path, }; use crate::storage::apply_cors_headers; use crate::storage::request_context::{RequestContext, extract_request_id_from_headers}; @@ -852,12 +852,12 @@ fn is_object_attributes_request(req: &HttpRequest) -> bool { } let path = req.uri().path(); - if path.starts_with(ADMIN_PREFIX) - || path.starts_with(MINIO_ADMIN_PREFIX) - || path.starts_with(RUSTFS_ADMIN_PREFIX) - || path.starts_with(MINIO_ADMIN_V3_PREFIX) - || path.starts_with(CONSOLE_PREFIX) - || path.starts_with(RPC_PREFIX) + if has_path_prefix(path, ADMIN_PREFIX) + || has_path_prefix(path, MINIO_ADMIN_PREFIX) + || has_path_prefix(path, RUSTFS_ADMIN_PREFIX) + || has_path_prefix(path, MINIO_ADMIN_V3_PREFIX) + || has_path_prefix(path, CONSOLE_PREFIX) + || has_path_prefix(path, RPC_PREFIX) { return false; } @@ -898,9 +898,9 @@ impl ConditionalCorsLayer { fn is_s3_path(path: &str) -> bool { // Exclude Admin, Console, RPC, and configured special paths - !path.starts_with(ADMIN_PREFIX) - && !path.starts_with(MINIO_ADMIN_PREFIX) - && !path.starts_with(RPC_PREFIX) + !has_path_prefix(path, ADMIN_PREFIX) + && !has_path_prefix(path, MINIO_ADMIN_PREFIX) + && !has_path_prefix(path, RPC_PREFIX) && !is_console_path(path) && !Self::EXCLUDED_EXACT_PATHS.contains(&path) } @@ -1819,6 +1819,7 @@ mod tests { assert!(ConditionalCorsLayer::is_s3_path("/")); assert!(!ConditionalCorsLayer::is_s3_path("/rustfs/admin/v3/info")); assert!(!ConditionalCorsLayer::is_s3_path("/minio/admin/v3/info")); + assert!(ConditionalCorsLayer::is_s3_path("/minio/adminx/object")); assert!(!ConditionalCorsLayer::is_s3_path("/health")); assert!(!ConditionalCorsLayer::is_s3_path("/health/ready")); } diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index 345f75ccf..f569a2225 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -49,7 +49,7 @@ pub(crate) use module_switch::{ pub(crate) use prefix::{ ADMIN_PREFIX, CONSOLE_PREFIX, FAVICON_PATH, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, - TONIC_PREFIX, VERSION, is_admin_path, + TONIC_PREFIX, VERSION, has_path_prefix, is_admin_path, }; pub(crate) use readiness::DependencyReadiness; pub(crate) use readiness::DependencyReadinessReport; diff --git a/rustfs/src/server/prefix.rs b/rustfs/src/server/prefix.rs index 077e126e9..1d875d15b 100644 --- a/rustfs/src/server/prefix.rs +++ b/rustfs/src/server/prefix.rs @@ -48,7 +48,7 @@ pub(crate) fn is_admin_path(path: &str) -> bool { has_path_prefix(path, ADMIN_PREFIX) || has_path_prefix(path, MINIO_ADMIN_PREFIX) } -fn has_path_prefix(path: &str, prefix: &str) -> bool { +pub(crate) fn has_path_prefix(path: &str, prefix: &str) -> bool { path == prefix || path.strip_prefix(prefix).is_some_and(|suffix| suffix.starts_with('/')) } diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index 663e21127..48813e925 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::server::has_path_prefix; use crate::server::{ServiceState, ServiceStateManager}; use bytes::Bytes; use http::{Request as HttpRequest, Response, StatusCode}; @@ -126,6 +127,28 @@ pub struct ReadinessGateService { readiness: Arc, } +fn is_probe_path(path: &str) -> bool { + let is_exact_probe = matches!( + path, + crate::server::PROFILE_MEMORY_PATH + | crate::server::PROFILE_CPU_PATH + | crate::server::HEALTH_PREFIX + | crate::server::HEALTH_COMPAT_LIVE_PATH + | crate::server::HEALTH_READY_PATH + | crate::server::FAVICON_PATH + ); + + let is_prefix_probe = has_path_prefix(path, crate::server::RUSTFS_ADMIN_PREFIX) + || has_path_prefix(path, crate::server::MINIO_ADMIN_V3_PREFIX) + || has_path_prefix(path, crate::server::CONSOLE_PREFIX) + || has_path_prefix(path, crate::server::RPC_PREFIX) + || has_path_prefix(path, crate::server::ADMIN_PREFIX) + || has_path_prefix(path, crate::server::MINIO_ADMIN_PREFIX) + || has_path_prefix(path, crate::server::TONIC_PREFIX); + + is_exact_probe || is_prefix_probe +} + type BoxError = Box; type BoxBody = http_body_util::combinators::UnsyncBoxBody; impl Service> for ReadinessGateService @@ -150,27 +173,7 @@ where Box::pin(async move { let path = req.uri().path(); debug!("ReadinessGateService: Received request for path: {}", path); - // 1) Exact match: fixed probe/resource path - let is_exact_probe = matches!( - path, - crate::server::PROFILE_MEMORY_PATH - | crate::server::PROFILE_CPU_PATH - | crate::server::HEALTH_PREFIX - | crate::server::HEALTH_COMPAT_LIVE_PATH - | crate::server::HEALTH_READY_PATH - | crate::server::FAVICON_PATH - ); - - // 2) Prefix matching: the entire set of route prefixes (including their subpaths) - let is_prefix_probe = path.starts_with(crate::server::RUSTFS_ADMIN_PREFIX) - || path.starts_with(crate::server::MINIO_ADMIN_V3_PREFIX) - || path.starts_with(crate::server::CONSOLE_PREFIX) - || path.starts_with(crate::server::RPC_PREFIX) - || path.starts_with(crate::server::ADMIN_PREFIX) - || path.starts_with(crate::server::MINIO_ADMIN_PREFIX) - || path.starts_with(crate::server::TONIC_PREFIX); - - let is_probe = is_exact_probe || is_prefix_probe; + let is_probe = is_probe_path(path); if !is_probe && !readiness.is_ready() { let body: BoxBody = Full::new(Bytes::from_static(b"Service not ready")) .map_err(|e| -> BoxError { Box::new(e) }) @@ -712,6 +715,16 @@ mod tests { assert_eq!(state_manager.current_state(), ServiceState::Starting); } + #[test] + fn probe_path_checks_admin_boundaries() { + assert!(is_probe_path("/minio/admin/v3/info")); + assert!(is_probe_path("/rustfs/admin/v3/info")); + assert!(is_probe_path("/rustfs/console/")); + assert!(!is_probe_path("/minio/adminx/object")); + assert!(!is_probe_path("/rustfs/adminx/object")); + assert!(!is_probe_path("/bucket/object")); + } + #[test] fn storage_ready_from_runtime_state_returns_false_when_all_disks_faulty() { let info = StorageInfo { diff --git a/rustfs/src/startup_iam.rs b/rustfs/src/startup_iam.rs new file mode 100644 index 000000000..841b94052 --- /dev/null +++ b/rustfs/src/startup_iam.rs @@ -0,0 +1,424 @@ +// 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 crate::app::context::{AppContext, get_global_app_context, init_global_app_context}; +use crate::server::{ServiceStateManager, publish_ready_when_runtime_ready}; +use rustfs_common::{GlobalReadiness, SystemStage}; +use rustfs_ecstore::store::ECStore; +use rustfs_iam::init_iam_sys; +use rustfs_kms::KmsServiceManager; +use std::future::Future; +use std::io::Result; +use std::pin::Pin; +use std::sync::Arc; +use std::time::Duration; +use tracing::{error, info, warn}; + +const IAM_RETRY_INITIAL_INTERVAL: Duration = Duration::from_secs(5); +const IAM_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(30); +/// After this many retries (~5 min at initial interval), escalate log level to ERROR. +const IAM_RETRY_ESCALATION_THRESHOLD: u64 = 12; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum IamBootstrapDisposition { + ReadyInline, + Deferred, +} + +fn init_app_context_if_needed(store: Arc, kms_interface: Arc) -> bool { + if get_global_app_context().is_some() { + return false; + } + + let Ok(iam_interface) = rustfs_iam::get() else { + return false; + }; + + init_global_app_context(AppContext::with_default_interfaces(store, iam_interface, kms_interface)); + true +} + +async fn finalize_iam_recovery( + store: Arc, + kms_interface: Arc, + readiness: Arc, + state_manager: Option>, +) -> Result<()> { + if !init_app_context_if_needed(store, kms_interface) && get_global_app_context().is_none() { + return Err(std::io::Error::other("IAM recovered but app context is still unavailable")); + } + + readiness.mark_stage(SystemStage::IamReady); + publish_ready_when_runtime_ready(readiness.as_ref(), state_manager.as_deref()).await +} + +fn compute_backoff_interval(attempt: u64, initial: Duration, max: Duration) -> Duration { + let exponent = u32::try_from(attempt.saturating_sub(1)).unwrap_or(u32::MAX); + let multiplier = 2u32.saturating_pow(exponent); + let backoff = initial.saturating_mul(multiplier); + if backoff > max { max } else { backoff } +} + +fn spawn_iam_recovery_task( + initial_interval: Duration, + max_interval: Duration, + shutdown_token: Option, + store: Arc, + kms_interface: Arc, + readiness: Arc, + state_manager: Option>, +) { + let init_store = store.clone(); + let finalize_store = store; + let finalize_kms_interface = kms_interface; + let finalize_readiness = readiness; + let finalize_state_manager = state_manager; + tokio::spawn(async move { + run_iam_recovery_loop( + initial_interval, + max_interval, + shutdown_token, + move || { + let store = init_store.clone(); + Box::pin(async move { attempt_init_iam_sys(store).await }) + }, + move || { + let store = finalize_store.clone(); + let kms_interface = finalize_kms_interface.clone(); + let readiness = finalize_readiness.clone(); + let state_manager = finalize_state_manager.clone(); + Box::pin(async move { finalize_iam_recovery(store, kms_interface, readiness, state_manager).await }) + }, + ) + .await; + }); +} + +type RecoveryFuture = Pin> + Send>>; + +async fn run_iam_recovery_loop( + initial_interval: Duration, + max_interval: Duration, + shutdown_token: Option, + mut init_fn: InitFn, + mut finalize_fn: FinalizeFn, +) where + InitFn: FnMut() -> RecoveryFuture, + FinalizeFn: FnMut() -> RecoveryFuture, +{ + let mut attempts: u64 = 0; + let degraded_since = std::time::Instant::now(); + + // Phase 1: retry init until it succeeds + loop { + attempts += 1; + let sleep_duration = compute_backoff_interval(attempts, initial_interval, max_interval); + if let Some(token) = shutdown_token.as_ref() { + tokio::select! { + _ = token.cancelled() => return, + _ = tokio::time::sleep(sleep_duration) => {} + } + } else { + tokio::time::sleep(sleep_duration).await; + } + + match init_fn().await { + Ok(()) => { + let degraded_secs = degraded_since.elapsed().as_secs(); + info!( + attempts, + degraded_duration_secs = degraded_secs, + "IAM bootstrap recovered after startup; publishing IAM readiness" + ); + break; + } + Err(err) => { + let next_interval = compute_backoff_interval(attempts + 1, initial_interval, max_interval); + if attempts >= IAM_RETRY_ESCALATION_THRESHOLD { + error!( + attempts, + next_retry_secs = next_interval.as_secs(), + degraded_duration_secs = degraded_since.elapsed().as_secs(), + error = %err, + "IAM bootstrap retry failed after {} attempts; service remains degraded", + attempts + ); + } else { + warn!( + attempts, + next_retry_secs = next_interval.as_secs(), + error = %err, + "IAM bootstrap retry failed; service remains degraded" + ); + } + } + } + } + + // Phase 2: retry finalize until readiness is published + let mut finalize_attempts: u64 = 0; + loop { + finalize_attempts += 1; + match finalize_fn().await { + Ok(()) => { + info!( + init_attempts = attempts, + finalize_attempts, + degraded_duration_secs = degraded_since.elapsed().as_secs(), + "IAM readiness published successfully" + ); + break; + } + Err(err) => { + let retry_interval = compute_backoff_interval(finalize_attempts, initial_interval, max_interval); + warn!( + finalize_attempts, + retry_secs = retry_interval.as_secs(), + error = %err, + "IAM recovered, but readiness publication failed; retrying" + ); + if let Some(token) = shutdown_token.as_ref() { + tokio::select! { + _ = token.cancelled() => return, + _ = tokio::time::sleep(retry_interval) => {} + } + } else { + tokio::time::sleep(retry_interval).await; + } + } + } + } +} + +fn initial_retry_interval() -> Duration { + // Only honor the test override in debug builds to prevent accidental + // production use via environment configuration. + #[cfg(debug_assertions)] + if let Some(ms) = rustfs_utils::get_env_opt_u64(rustfs_config::ENV_TEST_IAM_RETRY_INTERVAL_MS) { + return Duration::from_millis(ms); + } + IAM_RETRY_INITIAL_INTERVAL +} + +// Sentinel marks "not yet initialized"; env var is read on first call. +static TEST_REMAINING_FAILURES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(u64::MAX); + +fn should_fail_test_init_attempt() -> bool { + // Only honor the test hook in debug builds to prevent accidental + // production use via environment configuration. + #[cfg(not(debug_assertions))] + return false; + + #[cfg(debug_assertions)] + { + use std::sync::atomic::Ordering; + + let mut current = TEST_REMAINING_FAILURES.load(Ordering::SeqCst); + if current == u64::MAX { + let configured = rustfs_utils::get_env_opt_u64(rustfs_config::ENV_TEST_IAM_FAIL_INIT_ATTEMPTS).unwrap_or(0); + match TEST_REMAINING_FAILURES.compare_exchange(u64::MAX, configured, Ordering::SeqCst, Ordering::SeqCst) { + Ok(_) => current = configured, + Err(actual) => current = actual, + } + } + + while current > 0 { + match TEST_REMAINING_FAILURES.compare_exchange(current, current - 1, Ordering::SeqCst, Ordering::SeqCst) { + Ok(_) => return true, + Err(actual) => current = actual, + } + } + + false + } +} + +/// Reset the test failure counter so the next `should_fail_test_init_attempt` +/// call re-reads the environment variable by restoring the sentinel value. +/// Intended for use in integration tests that share a process. +#[doc(hidden)] +pub fn reset_test_failure_counter() { + use std::sync::atomic::Ordering; + TEST_REMAINING_FAILURES.store(u64::MAX, Ordering::SeqCst); +} + +async fn attempt_init_iam_sys(store: Arc) -> std::result::Result<(), std::io::Error> { + if should_fail_test_init_attempt() { + return Err(std::io::Error::other("forced test IAM bootstrap failure")); + } + + init_iam_sys(store).await.map_err(std::io::Error::other) +} + +/// Attempt IAM bootstrap at startup. If it fails, enter degraded mode and +/// spawn a background recovery task with exponential backoff. +/// +/// A failure to initialize IAM is not fatal — the process continues in degraded mode where: +/// - `/health/ready` returns 503 +/// - IAM-dependent operations return `IamSysNotInitialized` +/// - A background task retries IAM initialization until it succeeds +/// +/// Returns `Ok(ReadyInline)` if IAM initialized immediately, `Ok(Deferred)` if +/// recovery is happening in the background, or `Err` if IAM succeeded but +/// app context initialization failed (unexpected, indicates a bug). +pub async fn bootstrap_or_defer_iam_init( + store: Arc, + kms_interface: Arc, + readiness: Arc, + state_manager: Option>, + shutdown_token: Option, +) -> Result { + match attempt_init_iam_sys(store.clone()).await { + Ok(()) => { + if !init_app_context_if_needed(store, kms_interface) && get_global_app_context().is_none() { + return Err(std::io::Error::other("IAM bootstrap succeeded but app context is still unavailable")); + } + readiness.mark_stage(SystemStage::IamReady); + return Ok(IamBootstrapDisposition::ReadyInline); + } + Err(err) => { + let interval = initial_retry_interval(); + warn!( + error = %err, + initial_retry_secs = interval.as_secs(), + max_retry_secs = IAM_RETRY_MAX_INTERVAL.as_secs(), + escalation_threshold = IAM_RETRY_ESCALATION_THRESHOLD, + "Initial IAM bootstrap failed; continuing startup in degraded mode until IAM recovers. \ + Health endpoint will report 503 until IAM is ready." + ); + + spawn_iam_recovery_task( + interval, + IAM_RETRY_MAX_INTERVAL, + shutdown_token, + store, + kms_interface, + readiness, + state_manager, + ); + } + } + + Ok(IamBootstrapDisposition::Deferred) +} + +#[cfg(test)] +mod tests { + use super::{ + IAM_RETRY_ESCALATION_THRESHOLD, IAM_RETRY_INITIAL_INTERVAL, IAM_RETRY_MAX_INTERVAL, compute_backoff_interval, + run_iam_recovery_loop, + }; + use std::io::Error; + use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }; + use std::time::Duration; + + #[test] + fn iam_bootstrap_retry_intervals_are_non_zero_and_reasonable() { + assert!(IAM_RETRY_INITIAL_INTERVAL > Duration::ZERO); + assert_eq!(IAM_RETRY_INITIAL_INTERVAL.as_secs(), 5); + assert!(IAM_RETRY_MAX_INTERVAL > IAM_RETRY_INITIAL_INTERVAL); + assert_eq!(IAM_RETRY_MAX_INTERVAL.as_secs(), 30); + const { + assert!(IAM_RETRY_ESCALATION_THRESHOLD > 0); + assert!(IAM_RETRY_ESCALATION_THRESHOLD == 12); + } + } + + #[test] + fn compute_backoff_doubles_until_max() { + let initial = Duration::from_secs(5); + let max = Duration::from_secs(30); + + assert_eq!(compute_backoff_interval(1, initial, max), Duration::from_secs(5)); + assert_eq!(compute_backoff_interval(2, initial, max), Duration::from_secs(10)); + assert_eq!(compute_backoff_interval(3, initial, max), Duration::from_secs(20)); + assert_eq!(compute_backoff_interval(4, initial, max), Duration::from_secs(30)); + assert_eq!(compute_backoff_interval(5, initial, max), Duration::from_secs(30)); + assert_eq!(compute_backoff_interval(100, initial, max), Duration::from_secs(30)); + } + + #[tokio::test(start_paused = true)] + async fn recovery_loop_retries_finalize_until_success() { + let init_calls = Arc::new(AtomicUsize::new(0)); + let finalize_calls = Arc::new(AtomicUsize::new(0)); + + let init_calls_for_assert = init_calls.clone(); + let finalize_calls_for_assert = finalize_calls.clone(); + + run_iam_recovery_loop( + Duration::from_secs(5), + Duration::from_secs(30), + None, + move || { + let init_calls = init_calls.clone(); + Box::pin(async move { + init_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + }) + }, + move || { + let finalize_calls = finalize_calls.clone(); + Box::pin(async move { + let call = finalize_calls.fetch_add(1, Ordering::SeqCst) + 1; + if call < 3 { + Err(Error::other("finalize failed")) + } else { + Ok(()) + } + }) + }, + ) + .await; + + assert_eq!(init_calls_for_assert.load(Ordering::SeqCst), 1); + assert_eq!(finalize_calls_for_assert.load(Ordering::SeqCst), 3); + } + + #[tokio::test(start_paused = true)] + async fn recovery_loop_stops_after_shutdown_cancellation() { + let init_calls = Arc::new(AtomicUsize::new(0)); + let cancel = tokio_util::sync::CancellationToken::new(); + + let task = tokio::spawn({ + let init_calls = init_calls.clone(); + let cancel = cancel.clone(); + async move { + run_iam_recovery_loop( + Duration::from_secs(5), + Duration::from_secs(30), + Some(cancel), + move || { + let init_calls = init_calls.clone(); + Box::pin(async move { + init_calls.fetch_add(1, Ordering::SeqCst); + Err(Error::other("keep retrying")) + }) + }, + move || Box::pin(async { Ok(()) }), + ) + .await; + } + }); + + tokio::task::yield_now().await; + cancel.cancel(); + tokio::task::yield_now().await; + task.await.expect("recovery task should exit after cancellation"); + + assert_eq!(init_calls.load(Ordering::SeqCst), 0); + } +} diff --git a/rustfs/tests/embedded_deferred_iam_test.rs b/rustfs/tests/embedded_deferred_iam_test.rs new file mode 100644 index 000000000..742b9edd5 --- /dev/null +++ b/rustfs/tests/embedded_deferred_iam_test.rs @@ -0,0 +1,93 @@ +// 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 aws_sdk_s3::config::{Credentials, Region}; +use aws_sdk_s3::{Client, Config}; +use reqwest::StatusCode; +use rustfs::embedded::{RustFSServerBuilder, find_available_port}; +use rustfs_config::{ENV_TEST_IAM_FAIL_INIT_ATTEMPTS, ENV_TEST_IAM_RETRY_INTERVAL_MS}; +use std::time::Duration; +use temp_env::async_with_vars; + +fn s3_client(endpoint: &str, access_key: &str, secret_key: &str) -> Client { + let creds = Credentials::new(access_key, secret_key, None, None, "test"); + let config = Config::builder() + .credentials_provider(creds) + .region(Region::new("us-east-1")) + .endpoint_url(endpoint) + .force_path_style(true) + .behavior_version_latest() + .build(); + Client::from_conf(config) +} + +#[cfg(debug_assertions)] +#[tokio::test] +async fn test_embedded_server_recovers_after_deferred_iam_bootstrap() { + async_with_vars( + [ + (ENV_TEST_IAM_FAIL_INIT_ATTEMPTS, Some("1")), + (ENV_TEST_IAM_RETRY_INTERVAL_MS, Some("500")), + ], + async { + let port = find_available_port().expect("find free port"); + let server = RustFSServerBuilder::new() + .address(format!("127.0.0.1:{port}")) + .access_key("testaccesskey") + .secret_key("testsecretkey") + .build() + .await + .expect("start embedded server with deferred IAM bootstrap"); + + let endpoint = server.endpoint(); + let http = reqwest::Client::new(); + let ready_url = format!("{endpoint}/health/ready"); + + let initial_ready = http + .get(&ready_url) + .send() + .await + .expect("readiness probe should respond during deferred bootstrap"); + assert_eq!(initial_ready.status(), StatusCode::SERVICE_UNAVAILABLE); + + let recovered = tokio::time::timeout(Duration::from_secs(10), async { + loop { + tokio::time::sleep(Duration::from_millis(100)).await; + let response = http + .get(&ready_url) + .send() + .await + .expect("readiness probe should keep responding"); + if response.status() == StatusCode::OK { + return true; + } + } + }) + .await + .unwrap_or(false); + assert!(recovered, "readiness should recover after deferred IAM bootstrap succeeds"); + + let client = s3_client(&endpoint, server.access_key(), server.secret_key()); + client + .create_bucket() + .bucket("deferred-bucket") + .send() + .await + .expect("create bucket after readiness recovery"); + + server.shutdown().await; + }, + ) + .await; +} diff --git a/rustfs/tests/embedded_test.rs b/rustfs/tests/embedded_test.rs index 93ff183df..3188ae1ca 100644 --- a/rustfs/tests/embedded_test.rs +++ b/rustfs/tests/embedded_test.rs @@ -1,3 +1,17 @@ +// 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. + // Integration test demonstrating the embedded RustFS server API. // // This test starts a RustFS server in-process and exercises it via the @@ -39,7 +53,6 @@ async fn test_embedded_server_basic_s3_operations() { // 2. Create an S3 client and perform basic operations. let client = s3_client(&endpoint, server.access_key(), server.secret_key()); - // Create bucket client .create_bucket() .bucket("test-bucket") @@ -47,7 +60,6 @@ async fn test_embedded_server_basic_s3_operations() { .await .expect("create bucket"); - // Put object let body = ByteStream::from_static(b"hello rustfs embedded!"); client .put_object() @@ -58,7 +70,6 @@ async fn test_embedded_server_basic_s3_operations() { .await .expect("put object"); - // Get object let resp = client .get_object() .bucket("test-bucket") @@ -70,7 +81,6 @@ async fn test_embedded_server_basic_s3_operations() { let data = resp.body.collect().await.expect("read body").into_bytes(); assert_eq!(data.as_ref(), b"hello rustfs embedded!"); - // List objects let list = client .list_objects_v2() .bucket("test-bucket") @@ -79,7 +89,6 @@ async fn test_embedded_server_basic_s3_operations() { .expect("list objects"); assert_eq!(list.key_count(), Some(1)); - // Delete object client .delete_object() .bucket("test-bucket") @@ -88,7 +97,6 @@ async fn test_embedded_server_basic_s3_operations() { .await .expect("delete object"); - // Delete bucket client .delete_bucket() .bucket("test-bucket") @@ -96,6 +104,5 @@ async fn test_embedded_server_basic_s3_operations() { .await .expect("delete bucket"); - // 3. Shut down. server.shutdown().await; } diff --git a/rustfs/tests/manual/test_dial9.rs b/rustfs/tests/manual/test_dial9.rs index b2303995f..bdfeda2c3 100644 --- a/rustfs/tests/manual/test_dial9.rs +++ b/rustfs/tests/manual/test_dial9.rs @@ -1,3 +1,17 @@ +// 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. + // Manual Dial9 integration runner. // // Run with: