diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 843e95a4a..8492a9c30 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,16 +5,15 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Current Context - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) -- Branch: `overtrue/arch-app-context-split` -- Baseline: `origin/main` at `af291afb930e072d162ff10c63e33c091b706e58` -- PR type for this branch: `pure-move` -- Runtime behavior changes: none; `crate::app::context::*` remains the public - import path and resolver/global fallback semantics are unchanged. -- Rust code changes: split `rustfs/src/app/context.rs` into module-local - `interfaces`, `handles`, `global`, and `compat` files behind the existing - `app::context` module. +- Branch: `overtrue/arch-app-context-tests` +- Baseline: `origin/main` at `cb37a64a81bb857ea22efd2ebb015fbd3b64bef0` +- PR type for this branch: `test-only` +- Runtime behavior changes: none; resolver/global fallback semantics and IAM + degraded recovery behavior are unchanged. +- Rust code changes: add AppContext resolver compatibility tests and IAM + deferred recovery readiness coverage. - CI/script changes: none. -- Docs changes: record `CTX-001` AppContext file split scope. +- Docs changes: record `CTX-002`/`CTX-003` completion and verification scope. ## Phase 0 Tasks @@ -144,16 +143,22 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block initialization, resolver fallback order, and all consumer import paths. - Verification: formatting, compile checks, migration guards, diff hygiene, Rust risk scan, and full `make pre-commit`. -- [ ] `CTX-002` Add resolver compatibility tests. +- [x] `CTX-002` Add resolver compatibility tests. - Do: test context-first and global fallback for KMS runtime, bucket metadata, object store, endpoints, tier config, server config, and buffer config. - Acceptance: context wins when present and global fallback works when absent. -- [ ] `CTX-003` Add IAM deferred recovery readiness test. + - Verification: focused resolver compatibility test, formatting, compile + checks, migration guards, diff hygiene, Rust risk scan, and full + `make pre-commit`. +- [x] `CTX-003` Add IAM deferred recovery readiness test. - Do: verify IAM degraded recovery can still publish `IamReady` and `FullReady`. - Acceptance: boot/lifecycle changes cannot lose deferred readiness publication. + - Verification: focused IAM recovery test, formatting, compile checks, + migration guards, diff hygiene, Rust risk scan, and full + `make pre-commit`. ## Phase 1 Security Governance Tasks @@ -460,40 +465,43 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Next PRs -1. `test-only`: add CTX-002 resolver compatibility tests before migrating more - consumers to AppContext-first access. -2. `test-only`: add CTX-003 IAM deferred recovery readiness coverage before - boot phase extraction. +1. `pure-move`: start `R-009` boot wrapper with the IAM degraded readiness + contract covered. +2. `consumer-migration`: migrate a small consumer group to AppContext-first + access with resolver fallback tests in place. ## Pre-Push Review Log | Expert | Status | Notes | |---|---|---| -| Quality/architecture | pass | Pure module split behind the existing `app::context` root; no service container, public API, or runtime behavior added. | -| Migration preservation | pass | Re-exports preserve `crate::app::context::*`; AppContext construction, default handles, singleton initialization, and resolver fallback order are unchanged. | -| Testing/verification | pass | Focused compile, formatting, diff hygiene, migration guards, Rust risk scan, and full `make pre-commit` passed. | +| Quality/architecture | pass | Test-focused slice; resolver helpers stay private, AppContext test constructor is `#[cfg(test)]`, and no service container or registry is added. | +| Migration preservation | pass | Context-first plus fallback semantics are asserted, object-store resolution remains AppContext-only, and IAM recovery still publishes `IamReady` before `FullReady`. | +| Testing/verification | pass | Focused resolver/IAM tests, formatting, compile check, diff hygiene, migration guards, Rust risk scan, and full `make pre-commit` passed. | ## Verification Notes -Passed on `af291afb930e072d162ff10c63e33c091b706e58`: +Passed on `cb37a64a81bb857ea22efd2ebb015fbd3b64bef0`: - `cargo fmt --all`. - `cargo fmt --all --check`. +- `cargo test -p rustfs app::context::compat::tests::resolver_helpers_are_context_first_and_fallback_when_context_is_absent -- --nocapture`. +- `cargo test -p rustfs startup_iam::tests::recovery_loop_can_publish_iam_and_full_ready_after_degraded_init -- --nocapture`. - `cargo check -p rustfs --lib`. -- `cargo check -p rustfs --bin rustfs`. - `git diff --check`. - `./scripts/check_architecture_migration_rules.sh`. - `./scripts/check_layer_dependencies.sh`. -- Rust risk scan for changed Rust files; no matches. -- `make pre-commit`; all checks passed, including nextest 5883 passed and 111 - skipped. +- Rust risk scan for changed Rust files; matches are test-only `expect` + assertions, with no production `unwrap`/`expect`, numeric cast, string error, + boxed error, print macro, or relaxed-ordering match. +- `make pre-commit`; all checks passed, including nextest 5891 passed and + 111 skipped, plus doctests. Notes: -- This slice moves code between files only; it keeps `crate::app::context::*` - imports working through the module root. -- Resolver bodies still use the same AppContext-first and global fallback order. +- This slice adds coverage before further consumer migration. +- Resolver helpers preserve the same AppContext-first and fallback order. +- IAM degraded recovery keeps `IamReady` publication before `FullReady`. ## Handoff Notes -- CTX-002 should add resolver compatibility tests before moving consumers. -- CTX-003 should protect IAM degraded-readiness publication before boot - extraction. +- CTX-002 and CTX-003 are complete. +- Keep the next consumer or boot wrapper PR small and preserve global fallback + until the planned cleanup phase. diff --git a/rustfs/src/app/context/compat.rs b/rustfs/src/app/context/compat.rs index b511311d2..2c447242e 100644 --- a/rustfs/src/app/context/compat.rs +++ b/rustfs/src/app/context/compat.rs @@ -12,11 +12,13 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::global::get_global_app_context; +use super::global::{AppContext, get_global_app_context}; use super::handles::{ default_bucket_metadata_interface, default_endpoints_interface, default_kms_runtime_interface, default_server_config_interface, default_tier_config_interface, }; +#[cfg(test)] +use crate::config::RustFSBufferConfig; use rustfs_config::server_config::Config; use rustfs_ecstore::bucket::metadata_sys::BucketMetadataSys; use rustfs_ecstore::endpoints::EndpointServerPools; @@ -28,41 +30,318 @@ use tokio::sync::RwLock; /// Resolve KMS runtime service manager using AppContext-first precedence. pub fn resolve_kms_runtime_service_manager() -> Option> { - get_global_app_context() - .and_then(|context| context.kms_runtime().service_manager()) - .or_else(|| default_kms_runtime_interface().service_manager()) + resolve_kms_runtime_service_manager_with(get_global_app_context(), || default_kms_runtime_interface().service_manager()) } /// Resolve bucket metadata handle using AppContext-first precedence. pub fn resolve_bucket_metadata_handle() -> Option>> { - get_global_app_context() - .and_then(|context| context.bucket_metadata().handle()) - .or_else(|| default_bucket_metadata_interface().handle()) + resolve_bucket_metadata_handle_with(get_global_app_context(), || default_bucket_metadata_interface().handle()) } /// Resolve object store handle from AppContext. pub fn resolve_object_store_handle() -> Option> { - get_global_app_context().map(|context| context.object_store()) + resolve_object_store_handle_with(get_global_app_context()) } /// Resolve endpoints using AppContext-first precedence. pub fn resolve_endpoints_handle() -> Option { - get_global_app_context() - .and_then(|context| context.endpoints().handle()) - .or_else(|| default_endpoints_interface().handle()) + resolve_endpoints_handle_with(get_global_app_context(), || default_endpoints_interface().handle()) } /// Resolve tier config handle using AppContext-first precedence. pub fn resolve_tier_config_handle() -> Arc> { - get_global_app_context() - .map(|context| context.tier_config().handle()) - .unwrap_or_else(|| default_tier_config_interface().handle()) + resolve_tier_config_handle_with(get_global_app_context(), || default_tier_config_interface().handle()) } /// Resolve server config using AppContext-first precedence. pub fn resolve_server_config() -> Option { - match get_global_app_context() { - Some(context) => context.server_config().get(), - None => default_server_config_interface().get(), + resolve_server_config_with(get_global_app_context(), || default_server_config_interface().get()) +} + +fn resolve_kms_runtime_service_manager_with( + context: Option>, + fallback: impl FnOnce() -> Option>, +) -> Option> { + context + .and_then(|context| context.kms_runtime().service_manager()) + .or_else(fallback) +} + +fn resolve_bucket_metadata_handle_with( + context: Option>, + fallback: impl FnOnce() -> Option>>, +) -> Option>> { + context + .and_then(|context| context.bucket_metadata().handle()) + .or_else(fallback) +} + +fn resolve_object_store_handle_with(context: Option>) -> Option> { + context.map(|context| context.object_store()) +} + +fn resolve_endpoints_handle_with( + context: Option>, + fallback: impl FnOnce() -> Option, +) -> Option { + context.and_then(|context| context.endpoints().handle()).or_else(fallback) +} + +fn resolve_tier_config_handle_with( + context: Option>, + fallback: impl FnOnce() -> Arc>, +) -> Arc> { + context.map(|context| context.tier_config().handle()).unwrap_or_else(fallback) +} + +fn resolve_server_config_with(context: Option>, fallback: impl FnOnce() -> Option) -> Option { + context.map_or_else(fallback, |context| context.server_config().get()) +} + +#[cfg(test)] +fn resolve_buffer_config_with( + context: Option>, + fallback: impl FnOnce() -> RustFSBufferConfig, +) -> RustFSBufferConfig { + context.map_or_else(fallback, |context| context.buffer_config().get()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::app::context::global::AppContextTestInterfaces; + use crate::app::context::handles::{default_notify_interface, default_region_interface}; + use crate::app::context::interfaces::{ + BucketMetadataInterface, BufferConfigInterface, EndpointsInterface, IamInterface, KmsInterface, KmsRuntimeInterface, + ServerConfigInterface, TierConfigInterface, + }; + use crate::config::{RustFSBufferConfig, WorkloadProfile}; + use rustfs_ecstore::disk::endpoint::Endpoint; + use rustfs_ecstore::endpoints::{Endpoints, PoolEndpoints}; + use rustfs_ecstore::store::init_local_disks; + use rustfs_iam::{store::object::ObjectStore, sys::IamSys}; + use std::path::PathBuf; + use tempfile::TempDir; + use tokio_util::sync::CancellationToken; + + struct TestIamInterface; + + impl IamInterface for TestIamInterface { + fn handle(&self) -> Arc> { + unreachable!("resolver tests do not need an IAM handle") + } + + fn is_ready(&self) -> bool { + true + } + } + + struct TestKmsInterface { + kms: Arc, + } + + impl KmsInterface for TestKmsInterface { + fn handle(&self) -> Arc { + self.kms.clone() + } + } + + struct TestKmsRuntimeInterface { + kms: Option>, + } + + impl KmsRuntimeInterface for TestKmsRuntimeInterface { + fn service_manager(&self) -> Option> { + self.kms.clone() + } + } + + struct TestBucketMetadataInterface { + metadata: Option>>, + } + + impl BucketMetadataInterface for TestBucketMetadataInterface { + fn handle(&self) -> Option>> { + self.metadata.clone() + } + } + + struct TestEndpointsInterface { + endpoints: Option, + } + + impl EndpointsInterface for TestEndpointsInterface { + fn handle(&self) -> Option { + self.endpoints.clone() + } + } + + struct TestTierConfigInterface { + tier_config: Arc>, + } + + impl TierConfigInterface for TestTierConfigInterface { + fn handle(&self) -> Arc> { + self.tier_config.clone() + } + } + + struct TestServerConfigInterface { + config: Option, + } + + impl ServerConfigInterface for TestServerConfigInterface { + fn get(&self) -> Option { + self.config.clone() + } + } + + struct TestBufferConfigInterface { + config: RustFSBufferConfig, + } + + impl BufferConfigInterface for TestBufferConfigInterface { + fn get(&self) -> RustFSBufferConfig { + self.config.clone() + } + } + + async fn test_store() -> (TempDir, Arc, EndpointServerPools) { + let temp_dir = tempfile::tempdir().expect("test temp dir"); + let disk_paths = (0..4) + .map(|index| temp_dir.path().join(format!("disk{index}"))) + .collect::>(); + + for disk_path in &disk_paths { + tokio::fs::create_dir_all(disk_path).await.expect("test disk dir"); + } + + let mut endpoints = Vec::with_capacity(disk_paths.len()); + for (index, disk_path) in disk_paths.iter().enumerate() { + let mut endpoint = Endpoint::try_from(disk_path.to_str().expect("utf-8 test path")).expect("test endpoint"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(index); + endpoints.push(endpoint); + } + + let pool_endpoints = PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 4, + endpoints: Endpoints::from(endpoints), + cmd_line: "test".to_string(), + platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), + }; + let endpoint_pools = EndpointServerPools(vec![pool_endpoints]); + + init_local_disks(endpoint_pools.clone()).await.expect("test local disks"); + let store = ECStore::new( + "127.0.0.1:0".parse().expect("test addr"), + endpoint_pools.clone(), + CancellationToken::new(), + ) + .await + .expect("test ecstore"); + + (temp_dir, store, endpoint_pools) + } + + #[tokio::test] + async fn resolver_helpers_are_context_first_and_fallback_when_context_is_absent() { + let (_temp_dir, object_store, endpoints) = test_store().await; + let context_kms = Arc::new(KmsServiceManager::new()); + let fallback_kms = Arc::new(KmsServiceManager::new()); + let bucket_metadata = Arc::new(RwLock::new(BucketMetadataSys::new(object_store.clone()))); + let tier_config = TierConfigMgr::new(); + let server_config = Config::new(); + let buffer_config = RustFSBufferConfig::new(WorkloadProfile::AiTraining); + + let context = Arc::new(AppContext::with_test_interfaces( + object_store.clone(), + AppContextTestInterfaces { + iam: Arc::new(TestIamInterface), + kms: Arc::new(TestKmsInterface { + kms: context_kms.clone(), + }), + kms_runtime: Arc::new(TestKmsRuntimeInterface { + kms: Some(context_kms.clone()), + }), + notify: default_notify_interface(), + bucket_metadata: Arc::new(TestBucketMetadataInterface { + metadata: Some(bucket_metadata.clone()), + }), + endpoints: Arc::new(TestEndpointsInterface { + endpoints: Some(endpoints.clone()), + }), + region: default_region_interface(), + tier_config: Arc::new(TestTierConfigInterface { + tier_config: tier_config.clone(), + }), + server_config: Arc::new(TestServerConfigInterface { + config: Some(server_config.clone()), + }), + buffer_config: Arc::new(TestBufferConfigInterface { config: buffer_config }), + }, + )); + + assert!(Arc::ptr_eq( + &resolve_kms_runtime_service_manager_with(Some(context.clone()), || Some(fallback_kms.clone())) + .expect("context KMS runtime"), + &context_kms + )); + assert!(Arc::ptr_eq( + &resolve_bucket_metadata_handle_with(Some(context.clone()), || None).expect("context bucket metadata"), + &bucket_metadata + )); + assert!(Arc::ptr_eq( + &resolve_object_store_handle_with(Some(context.clone())).expect("context object store"), + &object_store + )); + assert_eq!( + resolve_endpoints_handle_with(Some(context.clone()), || None) + .expect("context endpoints") + .as_ref()[0] + .drives_per_set, + endpoints.as_ref()[0].drives_per_set + ); + assert!(Arc::ptr_eq( + &resolve_tier_config_handle_with(Some(context.clone()), TierConfigMgr::new), + &tier_config + )); + assert_eq!( + resolve_server_config_with(Some(context.clone()), || None).expect("context server config"), + server_config + ); + assert_eq!( + resolve_buffer_config_with(Some(context), || RustFSBufferConfig::new(WorkloadProfile::GeneralPurpose)).workload, + WorkloadProfile::AiTraining + ); + + assert!(Arc::ptr_eq( + &resolve_kms_runtime_service_manager_with(None, || Some(fallback_kms.clone())).expect("fallback KMS runtime"), + &fallback_kms + )); + assert!(Arc::ptr_eq( + &resolve_bucket_metadata_handle_with(None, || Some(bucket_metadata.clone())).expect("fallback bucket metadata"), + &bucket_metadata + )); + assert!(resolve_object_store_handle_with(None).is_none()); + assert_eq!( + resolve_endpoints_handle_with(None, || Some(endpoints.clone())) + .expect("fallback endpoints") + .as_ref()[0] + .drives_per_set, + endpoints.as_ref()[0].drives_per_set + ); + assert!(Arc::ptr_eq(&resolve_tier_config_handle_with(None, || tier_config.clone()), &tier_config)); + assert_eq!( + resolve_server_config_with(None, || Some(server_config.clone())).expect("fallback server config"), + server_config + ); + assert_eq!( + resolve_buffer_config_with(None, || RustFSBufferConfig::new(WorkloadProfile::DataAnalytics)).workload, + WorkloadProfile::DataAnalytics + ); } } diff --git a/rustfs/src/app/context/global.rs b/rustfs/src/app/context/global.rs index 66bc1f052..cd0c2eb33 100644 --- a/rustfs/src/app/context/global.rs +++ b/rustfs/src/app/context/global.rs @@ -114,6 +114,39 @@ impl AppContext { } } +#[cfg(test)] +pub(super) struct AppContextTestInterfaces { + pub(super) iam: Arc, + pub(super) kms: Arc, + pub(super) kms_runtime: Arc, + pub(super) notify: Arc, + pub(super) bucket_metadata: Arc, + pub(super) endpoints: Arc, + pub(super) region: Arc, + pub(super) tier_config: Arc, + pub(super) server_config: Arc, + pub(super) buffer_config: Arc, +} + +#[cfg(test)] +impl AppContext { + pub(super) fn with_test_interfaces(object_store: Arc, interfaces: AppContextTestInterfaces) -> Self { + Self { + object_store, + iam: interfaces.iam, + kms: interfaces.kms, + kms_runtime: interfaces.kms_runtime, + notify: interfaces.notify, + bucket_metadata: interfaces.bucket_metadata, + endpoints: interfaces.endpoints, + region: interfaces.region, + tier_config: interfaces.tier_config, + server_config: interfaces.server_config, + buffer_config: interfaces.buffer_config, + } + } +} + static APP_CONTEXT_SINGLETON: OnceLock> = OnceLock::new(); /// Initialize global application context once and return the canonical instance. diff --git a/rustfs/src/startup_iam.rs b/rustfs/src/startup_iam.rs index ae95cef1c..f46e5d631 100644 --- a/rustfs/src/startup_iam.rs +++ b/rustfs/src/startup_iam.rs @@ -344,10 +344,11 @@ mod tests { IAM_RETRY_ESCALATION_THRESHOLD, IAM_RETRY_INITIAL_INTERVAL, IAM_RETRY_MAX_INTERVAL, compute_backoff_interval, run_iam_recovery_loop, }; + use rustfs_common::{GlobalReadiness, SystemStage}; use std::io::Error; use std::sync::{ Arc, - atomic::{AtomicUsize, Ordering}, + atomic::{AtomicBool, AtomicUsize, Ordering}, }; use std::time::Duration; @@ -413,6 +414,51 @@ mod tests { assert_eq!(finalize_calls_for_assert.load(Ordering::SeqCst), 3); } + #[tokio::test(start_paused = true)] + async fn recovery_loop_can_publish_iam_and_full_ready_after_degraded_init() { + let init_calls = Arc::new(AtomicUsize::new(0)); + let readiness = Arc::new(GlobalReadiness::new()); + let observed_iam_ready = Arc::new(AtomicBool::new(false)); + + let init_calls_for_assert = init_calls.clone(); + let readiness_for_finalize = readiness.clone(); + let observed_iam_ready_for_finalize = observed_iam_ready.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 { + let call = init_calls.fetch_add(1, Ordering::SeqCst) + 1; + if call == 1 { + Err(Error::other("degraded init")) + } else { + Ok(()) + } + }) + }, + move || { + let readiness = readiness_for_finalize.clone(); + let observed_iam_ready = observed_iam_ready_for_finalize.clone(); + Box::pin(async move { + readiness.mark_stage(SystemStage::IamReady); + if matches!(readiness.current_stage(), SystemStage::IamReady) { + observed_iam_ready.store(true, Ordering::SeqCst); + } + readiness.mark_stage(SystemStage::FullReady); + Ok(()) + }) + }, + ) + .await; + + assert_eq!(init_calls_for_assert.load(Ordering::SeqCst), 2); + assert!(observed_iam_ready.load(Ordering::SeqCst)); + assert!(readiness.is_ready()); + } + #[tokio::test(start_paused = true)] async fn recovery_loop_stops_after_shutdown_cancellation() { let init_calls = Arc::new(AtomicUsize::new(0));