mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-25 21:46:50 +00:00
test: cover app context resolver compatibility (#3416)
This commit is contained in:
@@ -5,16 +5,15 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
|
|||||||
## Current Context
|
## Current Context
|
||||||
|
|
||||||
- Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
|
- Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
|
||||||
- Branch: `overtrue/arch-app-context-split`
|
- Branch: `overtrue/arch-app-context-tests`
|
||||||
- Baseline: `origin/main` at `af291afb930e072d162ff10c63e33c091b706e58`
|
- Baseline: `origin/main` at `cb37a64a81bb857ea22efd2ebb015fbd3b64bef0`
|
||||||
- PR type for this branch: `pure-move`
|
- PR type for this branch: `test-only`
|
||||||
- Runtime behavior changes: none; `crate::app::context::*` remains the public
|
- Runtime behavior changes: none; resolver/global fallback semantics and IAM
|
||||||
import path and resolver/global fallback semantics are unchanged.
|
degraded recovery behavior are unchanged.
|
||||||
- Rust code changes: split `rustfs/src/app/context.rs` into module-local
|
- Rust code changes: add AppContext resolver compatibility tests and IAM
|
||||||
`interfaces`, `handles`, `global`, and `compat` files behind the existing
|
deferred recovery readiness coverage.
|
||||||
`app::context` module.
|
|
||||||
- CI/script changes: none.
|
- 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
|
## 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.
|
initialization, resolver fallback order, and all consumer import paths.
|
||||||
- Verification: formatting, compile checks, migration guards, diff hygiene,
|
- Verification: formatting, compile checks, migration guards, diff hygiene,
|
||||||
Rust risk scan, and full `make pre-commit`.
|
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
|
- Do: test context-first and global fallback for KMS runtime, bucket
|
||||||
metadata, object store, endpoints, tier config, server config, and buffer
|
metadata, object store, endpoints, tier config, server config, and buffer
|
||||||
config.
|
config.
|
||||||
- Acceptance: context wins when present and global fallback works when absent.
|
- 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
|
- Do: verify IAM degraded recovery can still publish `IamReady` and
|
||||||
`FullReady`.
|
`FullReady`.
|
||||||
- Acceptance: boot/lifecycle changes cannot lose deferred readiness
|
- Acceptance: boot/lifecycle changes cannot lose deferred readiness
|
||||||
publication.
|
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
|
## Phase 1 Security Governance Tasks
|
||||||
|
|
||||||
@@ -460,40 +465,43 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
|
|||||||
|
|
||||||
## Next PRs
|
## Next PRs
|
||||||
|
|
||||||
1. `test-only`: add CTX-002 resolver compatibility tests before migrating more
|
1. `pure-move`: start `R-009` boot wrapper with the IAM degraded readiness
|
||||||
consumers to AppContext-first access.
|
contract covered.
|
||||||
2. `test-only`: add CTX-003 IAM deferred recovery readiness coverage before
|
2. `consumer-migration`: migrate a small consumer group to AppContext-first
|
||||||
boot phase extraction.
|
access with resolver fallback tests in place.
|
||||||
|
|
||||||
## Pre-Push Review Log
|
## Pre-Push Review Log
|
||||||
|
|
||||||
| Expert | Status | Notes |
|
| Expert | Status | Notes |
|
||||||
|---|---|---|
|
|---|---|---|
|
||||||
| Quality/architecture | pass | Pure module split behind the existing `app::context` root; no service container, public API, or runtime behavior added. |
|
| 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 | Re-exports preserve `crate::app::context::*`; AppContext construction, default handles, singleton initialization, and resolver fallback order are unchanged. |
|
| 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 compile, formatting, diff hygiene, migration guards, Rust risk scan, and full `make pre-commit` passed. |
|
| 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
|
## Verification Notes
|
||||||
|
|
||||||
Passed on `af291afb930e072d162ff10c63e33c091b706e58`:
|
Passed on `cb37a64a81bb857ea22efd2ebb015fbd3b64bef0`:
|
||||||
- `cargo fmt --all`.
|
- `cargo fmt --all`.
|
||||||
- `cargo fmt --all --check`.
|
- `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 --lib`.
|
||||||
- `cargo check -p rustfs --bin rustfs`.
|
|
||||||
- `git diff --check`.
|
- `git diff --check`.
|
||||||
- `./scripts/check_architecture_migration_rules.sh`.
|
- `./scripts/check_architecture_migration_rules.sh`.
|
||||||
- `./scripts/check_layer_dependencies.sh`.
|
- `./scripts/check_layer_dependencies.sh`.
|
||||||
- Rust risk scan for changed Rust files; no matches.
|
- Rust risk scan for changed Rust files; matches are test-only `expect`
|
||||||
- `make pre-commit`; all checks passed, including nextest 5883 passed and 111
|
assertions, with no production `unwrap`/`expect`, numeric cast, string error,
|
||||||
skipped.
|
boxed error, print macro, or relaxed-ordering match.
|
||||||
|
- `make pre-commit`; all checks passed, including nextest 5891 passed and
|
||||||
|
111 skipped, plus doctests.
|
||||||
|
|
||||||
Notes:
|
Notes:
|
||||||
- This slice moves code between files only; it keeps `crate::app::context::*`
|
- This slice adds coverage before further consumer migration.
|
||||||
imports working through the module root.
|
- Resolver helpers preserve the same AppContext-first and fallback order.
|
||||||
- Resolver bodies still use the same AppContext-first and global fallback order.
|
- IAM degraded recovery keeps `IamReady` publication before `FullReady`.
|
||||||
|
|
||||||
## Handoff Notes
|
## Handoff Notes
|
||||||
|
|
||||||
- CTX-002 should add resolver compatibility tests before moving consumers.
|
- CTX-002 and CTX-003 are complete.
|
||||||
- CTX-003 should protect IAM degraded-readiness publication before boot
|
- Keep the next consumer or boot wrapper PR small and preserve global fallback
|
||||||
extraction.
|
until the planned cleanup phase.
|
||||||
|
|||||||
@@ -12,11 +12,13 @@
|
|||||||
// See the License for the specific language governing permissions and
|
// See the License for the specific language governing permissions and
|
||||||
// limitations under the License.
|
// limitations under the License.
|
||||||
|
|
||||||
use super::global::get_global_app_context;
|
use super::global::{AppContext, get_global_app_context};
|
||||||
use super::handles::{
|
use super::handles::{
|
||||||
default_bucket_metadata_interface, default_endpoints_interface, default_kms_runtime_interface,
|
default_bucket_metadata_interface, default_endpoints_interface, default_kms_runtime_interface,
|
||||||
default_server_config_interface, default_tier_config_interface,
|
default_server_config_interface, default_tier_config_interface,
|
||||||
};
|
};
|
||||||
|
#[cfg(test)]
|
||||||
|
use crate::config::RustFSBufferConfig;
|
||||||
use rustfs_config::server_config::Config;
|
use rustfs_config::server_config::Config;
|
||||||
use rustfs_ecstore::bucket::metadata_sys::BucketMetadataSys;
|
use rustfs_ecstore::bucket::metadata_sys::BucketMetadataSys;
|
||||||
use rustfs_ecstore::endpoints::EndpointServerPools;
|
use rustfs_ecstore::endpoints::EndpointServerPools;
|
||||||
@@ -28,41 +30,318 @@ use tokio::sync::RwLock;
|
|||||||
|
|
||||||
/// Resolve KMS runtime service manager using AppContext-first precedence.
|
/// Resolve KMS runtime service manager using AppContext-first precedence.
|
||||||
pub fn resolve_kms_runtime_service_manager() -> Option<Arc<KmsServiceManager>> {
|
pub fn resolve_kms_runtime_service_manager() -> Option<Arc<KmsServiceManager>> {
|
||||||
get_global_app_context()
|
resolve_kms_runtime_service_manager_with(get_global_app_context(), || default_kms_runtime_interface().service_manager())
|
||||||
.and_then(|context| context.kms_runtime().service_manager())
|
|
||||||
.or_else(|| default_kms_runtime_interface().service_manager())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Resolve bucket metadata handle using AppContext-first precedence.
|
/// Resolve bucket metadata handle using AppContext-first precedence.
|
||||||
pub fn resolve_bucket_metadata_handle() -> Option<Arc<RwLock<BucketMetadataSys>>> {
|
pub fn resolve_bucket_metadata_handle() -> Option<Arc<RwLock<BucketMetadataSys>>> {
|
||||||
get_global_app_context()
|
resolve_bucket_metadata_handle_with(get_global_app_context(), || default_bucket_metadata_interface().handle())
|
||||||
.and_then(|context| context.bucket_metadata().handle())
|
|
||||||
.or_else(|| default_bucket_metadata_interface().handle())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Resolve object store handle from AppContext.
|
/// Resolve object store handle from AppContext.
|
||||||
pub fn resolve_object_store_handle() -> Option<Arc<ECStore>> {
|
pub fn resolve_object_store_handle() -> Option<Arc<ECStore>> {
|
||||||
get_global_app_context().map(|context| context.object_store())
|
resolve_object_store_handle_with(get_global_app_context())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Resolve endpoints using AppContext-first precedence.
|
/// Resolve endpoints using AppContext-first precedence.
|
||||||
pub fn resolve_endpoints_handle() -> Option<EndpointServerPools> {
|
pub fn resolve_endpoints_handle() -> Option<EndpointServerPools> {
|
||||||
get_global_app_context()
|
resolve_endpoints_handle_with(get_global_app_context(), || default_endpoints_interface().handle())
|
||||||
.and_then(|context| context.endpoints().handle())
|
|
||||||
.or_else(|| default_endpoints_interface().handle())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Resolve tier config handle using AppContext-first precedence.
|
/// Resolve tier config handle using AppContext-first precedence.
|
||||||
pub fn resolve_tier_config_handle() -> Arc<RwLock<TierConfigMgr>> {
|
pub fn resolve_tier_config_handle() -> Arc<RwLock<TierConfigMgr>> {
|
||||||
get_global_app_context()
|
resolve_tier_config_handle_with(get_global_app_context(), || default_tier_config_interface().handle())
|
||||||
.map(|context| context.tier_config().handle())
|
|
||||||
.unwrap_or_else(|| default_tier_config_interface().handle())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Resolve server config using AppContext-first precedence.
|
/// Resolve server config using AppContext-first precedence.
|
||||||
pub fn resolve_server_config() -> Option<Config> {
|
pub fn resolve_server_config() -> Option<Config> {
|
||||||
match get_global_app_context() {
|
resolve_server_config_with(get_global_app_context(), || default_server_config_interface().get())
|
||||||
Some(context) => context.server_config().get(),
|
}
|
||||||
None => default_server_config_interface().get(),
|
|
||||||
|
fn resolve_kms_runtime_service_manager_with(
|
||||||
|
context: Option<Arc<AppContext>>,
|
||||||
|
fallback: impl FnOnce() -> Option<Arc<KmsServiceManager>>,
|
||||||
|
) -> Option<Arc<KmsServiceManager>> {
|
||||||
|
context
|
||||||
|
.and_then(|context| context.kms_runtime().service_manager())
|
||||||
|
.or_else(fallback)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn resolve_bucket_metadata_handle_with(
|
||||||
|
context: Option<Arc<AppContext>>,
|
||||||
|
fallback: impl FnOnce() -> Option<Arc<RwLock<BucketMetadataSys>>>,
|
||||||
|
) -> Option<Arc<RwLock<BucketMetadataSys>>> {
|
||||||
|
context
|
||||||
|
.and_then(|context| context.bucket_metadata().handle())
|
||||||
|
.or_else(fallback)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn resolve_object_store_handle_with(context: Option<Arc<AppContext>>) -> Option<Arc<ECStore>> {
|
||||||
|
context.map(|context| context.object_store())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn resolve_endpoints_handle_with(
|
||||||
|
context: Option<Arc<AppContext>>,
|
||||||
|
fallback: impl FnOnce() -> Option<EndpointServerPools>,
|
||||||
|
) -> Option<EndpointServerPools> {
|
||||||
|
context.and_then(|context| context.endpoints().handle()).or_else(fallback)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn resolve_tier_config_handle_with(
|
||||||
|
context: Option<Arc<AppContext>>,
|
||||||
|
fallback: impl FnOnce() -> Arc<RwLock<TierConfigMgr>>,
|
||||||
|
) -> Arc<RwLock<TierConfigMgr>> {
|
||||||
|
context.map(|context| context.tier_config().handle()).unwrap_or_else(fallback)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn resolve_server_config_with(context: Option<Arc<AppContext>>, fallback: impl FnOnce() -> Option<Config>) -> Option<Config> {
|
||||||
|
context.map_or_else(fallback, |context| context.server_config().get())
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
fn resolve_buffer_config_with(
|
||||||
|
context: Option<Arc<AppContext>>,
|
||||||
|
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<IamSys<ObjectStore>> {
|
||||||
|
unreachable!("resolver tests do not need an IAM handle")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn is_ready(&self) -> bool {
|
||||||
|
true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct TestKmsInterface {
|
||||||
|
kms: Arc<KmsServiceManager>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl KmsInterface for TestKmsInterface {
|
||||||
|
fn handle(&self) -> Arc<KmsServiceManager> {
|
||||||
|
self.kms.clone()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct TestKmsRuntimeInterface {
|
||||||
|
kms: Option<Arc<KmsServiceManager>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl KmsRuntimeInterface for TestKmsRuntimeInterface {
|
||||||
|
fn service_manager(&self) -> Option<Arc<KmsServiceManager>> {
|
||||||
|
self.kms.clone()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct TestBucketMetadataInterface {
|
||||||
|
metadata: Option<Arc<RwLock<BucketMetadataSys>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl BucketMetadataInterface for TestBucketMetadataInterface {
|
||||||
|
fn handle(&self) -> Option<Arc<RwLock<BucketMetadataSys>>> {
|
||||||
|
self.metadata.clone()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct TestEndpointsInterface {
|
||||||
|
endpoints: Option<EndpointServerPools>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl EndpointsInterface for TestEndpointsInterface {
|
||||||
|
fn handle(&self) -> Option<EndpointServerPools> {
|
||||||
|
self.endpoints.clone()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct TestTierConfigInterface {
|
||||||
|
tier_config: Arc<RwLock<TierConfigMgr>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl TierConfigInterface for TestTierConfigInterface {
|
||||||
|
fn handle(&self) -> Arc<RwLock<TierConfigMgr>> {
|
||||||
|
self.tier_config.clone()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct TestServerConfigInterface {
|
||||||
|
config: Option<Config>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ServerConfigInterface for TestServerConfigInterface {
|
||||||
|
fn get(&self) -> Option<Config> {
|
||||||
|
self.config.clone()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct TestBufferConfigInterface {
|
||||||
|
config: RustFSBufferConfig,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl BufferConfigInterface for TestBufferConfigInterface {
|
||||||
|
fn get(&self) -> RustFSBufferConfig {
|
||||||
|
self.config.clone()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn test_store() -> (TempDir, Arc<ECStore>, 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::<Vec<PathBuf>>();
|
||||||
|
|
||||||
|
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
|
||||||
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -114,6 +114,39 @@ impl AppContext {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
pub(super) struct AppContextTestInterfaces {
|
||||||
|
pub(super) iam: Arc<dyn IamInterface>,
|
||||||
|
pub(super) kms: Arc<dyn KmsInterface>,
|
||||||
|
pub(super) kms_runtime: Arc<dyn KmsRuntimeInterface>,
|
||||||
|
pub(super) notify: Arc<dyn NotifyInterface>,
|
||||||
|
pub(super) bucket_metadata: Arc<dyn BucketMetadataInterface>,
|
||||||
|
pub(super) endpoints: Arc<dyn EndpointsInterface>,
|
||||||
|
pub(super) region: Arc<dyn RegionInterface>,
|
||||||
|
pub(super) tier_config: Arc<dyn TierConfigInterface>,
|
||||||
|
pub(super) server_config: Arc<dyn ServerConfigInterface>,
|
||||||
|
pub(super) buffer_config: Arc<dyn BufferConfigInterface>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
impl AppContext {
|
||||||
|
pub(super) fn with_test_interfaces(object_store: Arc<ECStore>, 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<Arc<AppContext>> = OnceLock::new();
|
static APP_CONTEXT_SINGLETON: OnceLock<Arc<AppContext>> = OnceLock::new();
|
||||||
|
|
||||||
/// Initialize global application context once and return the canonical instance.
|
/// Initialize global application context once and return the canonical instance.
|
||||||
|
|||||||
@@ -344,10 +344,11 @@ mod tests {
|
|||||||
IAM_RETRY_ESCALATION_THRESHOLD, IAM_RETRY_INITIAL_INTERVAL, IAM_RETRY_MAX_INTERVAL, compute_backoff_interval,
|
IAM_RETRY_ESCALATION_THRESHOLD, IAM_RETRY_INITIAL_INTERVAL, IAM_RETRY_MAX_INTERVAL, compute_backoff_interval,
|
||||||
run_iam_recovery_loop,
|
run_iam_recovery_loop,
|
||||||
};
|
};
|
||||||
|
use rustfs_common::{GlobalReadiness, SystemStage};
|
||||||
use std::io::Error;
|
use std::io::Error;
|
||||||
use std::sync::{
|
use std::sync::{
|
||||||
Arc,
|
Arc,
|
||||||
atomic::{AtomicUsize, Ordering},
|
atomic::{AtomicBool, AtomicUsize, Ordering},
|
||||||
};
|
};
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
@@ -413,6 +414,51 @@ mod tests {
|
|||||||
assert_eq!(finalize_calls_for_assert.load(Ordering::SeqCst), 3);
|
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)]
|
#[tokio::test(start_paused = true)]
|
||||||
async fn recovery_loop_stops_after_shutdown_cancellation() {
|
async fn recovery_loop_stops_after_shutdown_cancellation() {
|
||||||
let init_calls = Arc::new(AtomicUsize::new(0));
|
let init_calls = Arc::new(AtomicUsize::new(0));
|
||||||
|
|||||||
Reference in New Issue
Block a user