diff --git a/crates/notify/src/event_bridge.rs b/crates/notify/src/event_bridge.rs deleted file mode 100644 index 125f1af23..000000000 --- a/crates/notify/src/event_bridge.rs +++ /dev/null @@ -1,15 +0,0 @@ -// 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. - -pub use crate::pipeline::{LiveEventHistory, NotifyEventBridge}; diff --git a/crates/notify/src/lib.rs b/crates/notify/src/lib.rs index 1c6f5931b..77d8ac5bb 100644 --- a/crates/notify/src/lib.rs +++ b/crates/notify/src/lib.rs @@ -31,7 +31,6 @@ mod bucket_config_manager; mod config_manager; mod error; mod event; -mod event_bridge; pub mod factory; mod global; pub mod integration; @@ -50,13 +49,12 @@ pub use bucket_config_manager::NotifyBucketConfigManager; pub use config_manager::{NotifyConfigManager, runtime_target_id_for_subsystem}; pub use error::{LifecycleError, NotificationError}; pub use event::{Event, EventArgs, EventArgsBuilder, NotifyObjectInfo}; -pub use event_bridge::{LiveEventHistory, NotifyEventBridge}; pub use global::{ initialize, initialize_live_events, is_notification_system_initialized, notification_metrics_snapshot, notification_system, notification_target_metrics, notifier_global, }; pub use integration::{NotificationMetricSnapshot, NotificationSystem, NotificationTargetMetricSnapshot}; -pub use pipeline::NotifyPipeline; +pub use pipeline::{LiveEventHistory, NotifyEventBridge, NotifyPipeline}; pub use rule_engine::NotifyRuleEngine; pub use rules::BucketNotificationConfig; pub use runtime_facade::NotifyRuntimeFacade; diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index a315feae1..81fe9ea10 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,20 +5,22 @@ 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-storage-owner-ecstore-root-imports` -- Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095/API-096/API-097/API-098/API-099/API-100/API-101/API-102/API-103/API-104/API-105/API-106/API-107/API-108/API-109/API-110/API-111/API-112/API-113/API-114/API-115/API-116/API-117/API-118/API-119/API-120/API-121/API-122/API-123/API-124/API-125/API-126/API-127/API-128/API-129/API-130/API-131/API-132/API-133/API-134/API-135/API-136/API-137/API-138/API-139/API-140/API-141/API-142/API-143/API-144/API-145/API-146/API-147/API-148/API-149/API-150/API-151/API-152/API-153`. -- Based on: `overtrue/arch-owner-ecstore-final-bridges` after API-153. -- PR type for this branch: `pure-move` +- Branch: `overtrue/arch-app-context-runtime-resolvers` +- Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095/API-096/API-097/API-098/API-099/API-100/API-101/API-102/API-103/API-104/API-105/API-106/API-107/API-108/API-109/API-110/API-111/API-112/API-113/API-114/API-115/API-116/API-117/API-118/API-119/API-120/API-121/API-122/API-123/API-124/API-125/API-126/API-127/API-128/API-129/API-130/API-131/API-132/API-133/API-134/API-135/API-136/API-137/API-138/API-139/API-140/API-141/API-142/API-143/API-144/API-145/API-146/API-147/API-148/API-149/API-150/API-151/API-152/API-153/API-154/API-155`. +- Based on: `overtrue/arch-app-notify-compat-boundaries` after API-155. +- PR type for this branch: `consumer-migration` - Runtime behavior changes: none. -- Rust code changes: remove the final storage owner ECStore compatibility - bridge file and keep its ECStore source imports in the storage owner root. +- Rust code changes: route selected KMS readiness, notification, and buffer + profile consumers through AppContext resolver helpers with legacy global + fallback. - CI/script changes: lock completed owner and test/fuzz boundaries against bare/glob imports, scattered raw ECStore facade subpaths, and startup runtime/root-server/table/S3/app shared/app bucket/app ECStore/admin facade regressions, plus external runtime, test, fuzz, and storage-owner module ECStore compatibility bypasses, plus runtime crate, owner crate, test/fuzz, - and storage owner thin bridge regressions. -- Docs changes: record the API-136/API-137/API-138/API-139/API-140/API-141/API-142/API-143/API-144/API-145/API-146/API-147/API-148/API-149/API-150/API-151/API-152/API-153/API-154 owner facade cleanup. + and storage owner thin bridge regressions, plus app context and notify + event-bridge thin module regressions. +- Docs changes: record the API-136/API-137/API-138/API-139/API-140/API-141/API-142/API-143/API-144/API-145/API-146/API-147/API-148/API-149/API-150/API-151/API-152/API-153/API-154/API-155/API-156 owner facade cleanup. ## Phase 0 Tasks @@ -4230,9 +4232,39 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block formatting, diff hygiene, Rust risk scan, branch freshness check, pre-commit, and three-expert review. +- [x] `API-155` Collapse app context and notify thin compatibility modules. + - Do: remove `rustfs/src/app/context/compat.rs` by moving resolver helpers + into `rustfs/src/app/context.rs`, and remove the notify event-bridge + re-export module by exporting pipeline symbols directly from the notify + owner root. + - Acceptance: no app context `compat` module or notify `event_bridge` module + remains, while existing `crate::app::context::*` and `rustfs_notify::*` + public symbols keep the same paths. + - Must preserve: AppContext-first resolver precedence, legacy global + fallback behavior, bucket metadata/endpoints/tier/server config handles, + notify live event history, and notify event bridge type aliases. + - Verification: RustFS and notify compile coverage, migration guard, shell + syntax check, formatting, diff hygiene, Rust risk scan, branch freshness + check, pre-commit, and three-expert review. + +- [x] `API-156` Route app runtime consumers through AppContext resolvers. + - Do: add notify and buffer profile resolver helpers, route bucket/object + notification users through the notify resolver, route ECFS buffer sizing + through the buffer resolver, and route public health KMS readiness through + the KMS runtime resolver. + - Acceptance: selected app/server/storage consumers no longer open-code + direct global notifier, buffer config, or KMS service manager fallback when + an AppContext resolver already owns the migration boundary. + - Must preserve: context-first behavior when an AppContext exists, legacy + global fallback when it does not, notification delivery semantics, buffer + opt-in behavior, and public health readiness behavior. + - Verification: RustFS compile coverage, migration guard, shell syntax check, + formatting, diff hygiene, Rust risk scan, branch freshness check, + pre-commit, and three-expert review. + ## Next PRs -1. `pure-move`: continue pruning remaining facade compatibility and owner boundaries. +1. `consumer-migration`: continue reducing direct global reads behind AppContext resolver boundaries. ## Pre-Push Review Log @@ -4247,11 +4279,48 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Quality/architecture | pass | API-154 removes the final storage owner ECStore bridge file and keeps direct imports at the storage owner root. | | Migration preservation | pass | Existing `crate::storage::ecstore_*` modules, constants, wrappers, and downstream call paths keep the same shape. | | Testing/verification | pass | RustFS focused compile, formatting, migration guard, shell syntax, diff hygiene, bridge scan, Rust risk scan, and pre-commit passed for API-154. | +| Quality/architecture | pass | API-155 removes app context and notify thin compatibility modules while keeping owner-root exports. | +| Migration preservation | pass | AppContext resolver precedence and notify pipeline public aliases keep the same public call paths. | +| Testing/verification | pass | RustFS/notify focused compile, targeted tests, formatting, migration guard, shell syntax, diff hygiene, bridge scan, Rust risk scan, and pre-commit passed for API-155. | +| Quality/architecture | pass | API-156 centralizes selected app/server/storage runtime fallbacks behind AppContext resolver helpers without adding new abstractions. | +| Migration preservation | pass | KMS readiness, notification dispatch, and ECFS buffer sizing keep existing global fallback semantics when no AppContext is available. | +| Testing/verification | pass | RustFS focused compile, formatting, migration guard, shell syntax, diff hygiene, Rust risk scan, and pre-commit passed for API-156. | ## Verification Notes Passed before push: +- Issue #660 API-156 current slice: + - `cargo check --tests -p rustfs`: passed. + - `cargo fmt --all`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - `bash -n scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `make pre-commit`: passed. + - AppContext runtime resolver scan: passed; selected bucket/object notify, + ECFS buffer sizing, and public health KMS readiness consumers use resolver + helpers. + - Rust risk scan: no new production unwrap/expect, panic/todo/unsafe, or + cast risks added. + +- Issue #660 API-155 current slice: + - `cargo check --tests -p rustfs -p rustfs-notify`: passed. + - `cargo test -p rustfs resolver_helpers_are_context_first_and_fallback_when_context_is_absent --lib`: + passed. + - `cargo test -p rustfs-notify --lib`: passed. + - `cargo fmt --all`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - `bash -n scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `make pre-commit`: passed. + - App context and notify thin bridge scan: passed; no + `rustfs/src/app/context/compat.rs` or `crates/notify/src/event_bridge.rs` + remains. + - Rust risk scan: no new production unwrap/expect, panic/todo/unsafe, or + cast risks added; changed unwrap/expect matches are moved test setup only. + - Issue #660 API-154 current slice: - `cargo check --tests -p rustfs`: passed. - `cargo fmt --all`: passed. diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 7861d1b09..2684f511f 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -40,7 +40,7 @@ use crate::admin::handlers::site_replication::{ site_replication_bucket_meta_hook, site_replication_delete_bucket_hook, site_replication_make_bucket_hook, }; use crate::app::context::{ - AppContext, default_notify_interface, get_global_app_context, resolve_object_store_handle_for_context, + AppContext, get_global_app_context, resolve_notify_interface_for_context, resolve_object_store_handle_for_context, }; use crate::auth::get_condition_values_with_client_info; use crate::error::ApiError; @@ -1712,11 +1712,7 @@ impl DefaultBucketUsecase { .map_err(ApiError::from)?; let region = resolve_notification_region(self.global_region(), request_region); - let notify = self - .context - .as_ref() - .map(|context| context.notify()) - .unwrap_or_else(default_notify_interface); + let notify = resolve_notify_interface_for_context(self.context.as_deref()); let clear_rules = notify.clear_bucket_notification_rules(&bucket); let parse_rules = async { let mut event_rules = Vec::new(); diff --git a/rustfs/src/app/context.rs b/rustfs/src/app/context.rs index f74872c75..5dfc4a4dc 100644 --- a/rustfs/src/app/context.rs +++ b/rustfs/src/app/context.rs @@ -16,13 +16,374 @@ //! This module introduces explicit dependency injection entry points //! for storage, IAM, and KMS handles. -mod compat; mod global; mod handles; mod interfaces; mod startup; -pub use compat::*; pub use global::*; pub use handles::*; pub use interfaces::*; + +use super::ECStore; +use super::EndpointServerPools; +use super::TierConfigMgr; +use super::metadata_sys::BucketMetadataSys; +use super::new_object_layer_fn; +use crate::config::RustFSBufferConfig; +use rustfs_config::server_config::Config; +use rustfs_kms::KmsServiceManager; +use std::sync::Arc; +use tokio::sync::RwLock; + +/// Resolve KMS runtime service manager using AppContext-first precedence. +pub fn resolve_kms_runtime_service_manager() -> Option> { + 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>> { + resolve_bucket_metadata_handle_with(get_global_app_context(), || default_bucket_metadata_interface().handle()) +} + +/// Resolve object store handle using AppContext-first precedence. +pub fn resolve_object_store_handle() -> Option> { + let context = get_global_app_context(); + resolve_object_store_handle_for_context(context.as_deref()) +} + +/// Resolve object store handle using an explicit AppContext, falling back to the legacy global object layer. +pub fn resolve_object_store_handle_for_context(context: Option<&AppContext>) -> Option> { + context.map(|context| context.object_store()).or_else(new_object_layer_fn) +} + +/// Resolve notify interface using an explicit AppContext, falling back to the legacy global notifier. +pub fn resolve_notify_interface_for_context(context: Option<&AppContext>) -> Arc { + context + .map(|context| context.notify()) + .unwrap_or_else(default_notify_interface) +} + +/// Resolve endpoints using AppContext-first precedence. +pub fn resolve_endpoints_handle() -> Option { + 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> { + 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 { + resolve_server_config_with(get_global_app_context(), || default_server_config_interface().get()) +} + +/// Resolve buffer profile config using AppContext-first precedence. +pub fn resolve_buffer_config() -> RustFSBufferConfig { + resolve_buffer_config_with(get_global_app_context(), || default_buffer_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) +} + +#[cfg(test)] +fn resolve_object_store_handle_with( + context: Option>, + fallback: impl FnOnce() -> Option>, +) -> Option> { + context.map(|context| context.object_store()).or_else(fallback) +} + +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()) +} + +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::super::Endpoint; + use super::super::init_local_disks; + use super::super::new_object_layer_fn; + use super::super::{Endpoints, PoolEndpoints}; + 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_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) { + if let Some(store) = new_object_layer_fn() { + let endpoints = EndpointServerPools(store.pools.iter().map(|pool| pool.endpoints.clone()).collect()); + return (tempfile::tempdir().expect("compat test temp dir"), store, endpoints); + } + + 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]); + + if let Some(store) = new_object_layer_fn() { + return (temp_dir, store, endpoint_pools); + } + + 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()), || None).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!(Arc::ptr_eq( + &resolve_object_store_handle_with(None, || Some(object_store.clone())).expect("fallback object store"), + &object_store + )); + 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/compat.rs b/rustfs/src/app/context/compat.rs deleted file mode 100644 index bd22f8b1c..000000000 --- a/rustfs/src/app/context/compat.rs +++ /dev/null @@ -1,371 +0,0 @@ -// 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 super::super::ECStore; -use super::super::EndpointServerPools; -use super::super::TierConfigMgr; -use super::super::metadata_sys::BucketMetadataSys; -use super::super::new_object_layer_fn; -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_kms::KmsServiceManager; -use std::sync::Arc; -use tokio::sync::RwLock; - -/// Resolve KMS runtime service manager using AppContext-first precedence. -pub fn resolve_kms_runtime_service_manager() -> Option> { - 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>> { - resolve_bucket_metadata_handle_with(get_global_app_context(), || default_bucket_metadata_interface().handle()) -} - -/// Resolve object store handle using AppContext-first precedence. -pub fn resolve_object_store_handle() -> Option> { - let context = get_global_app_context(); - resolve_object_store_handle_for_context(context.as_deref()) -} - -/// Resolve object store handle using an explicit AppContext, falling back to the legacy global object layer. -pub fn resolve_object_store_handle_for_context(context: Option<&AppContext>) -> Option> { - context.map(|context| context.object_store()).or_else(new_object_layer_fn) -} - -/// Resolve endpoints using AppContext-first precedence. -pub fn resolve_endpoints_handle() -> Option { - 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> { - 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 { - 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) -} - -#[cfg(test)] -fn resolve_object_store_handle_with( - context: Option>, - fallback: impl FnOnce() -> Option>, -) -> Option> { - context.map(|context| context.object_store()).or_else(fallback) -} - -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::super::super::Endpoint; - use super::super::super::init_local_disks; - use super::super::super::new_object_layer_fn; - use super::super::super::{Endpoints, PoolEndpoints}; - 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_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) { - if let Some(store) = new_object_layer_fn() { - let endpoints = EndpointServerPools(store.pools.iter().map(|pool| pool.endpoints.clone()).collect()); - return (tempfile::tempdir().expect("compat test temp dir"), store, endpoints); - } - - 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]); - - if let Some(store) = new_object_layer_fn() { - return (temp_dir, store, endpoint_pools); - } - - 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()), || None).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!(Arc::ptr_eq( - &resolve_object_store_handle_with(None, || Some(object_store.clone())).expect("fallback object store"), - &object_store - )); - 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/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 42399d00f..d983fc2d1 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -15,7 +15,7 @@ //! Object application use-case contracts. use crate::app::context::{ - AppContext, default_notify_interface, get_global_app_context, resolve_object_store_handle_for_context, + AppContext, get_global_app_context, resolve_notify_interface_for_context, resolve_object_store_handle_for_context, }; use crate::config::RustFSBufferConfig; use crate::delete_tail_activity::{DeleteTailActivityGuard, DeleteTailStage}; @@ -3707,11 +3707,7 @@ impl DefaultObjectUsecase { } let req_headers = req.headers.clone(); - let notify = self - .context - .as_ref() - .map(|context| context.notify()) - .unwrap_or_else(default_notify_interface); + let notify = resolve_notify_interface_for_context(self.context.as_deref()); let request_context = req.extensions.get::().cloned(); let deleted_any = delete_results.iter().any(|result| result.delete_object.is_some()); let notify_bucket = bucket.clone(); @@ -4712,11 +4708,7 @@ impl DefaultObjectUsecase { None => String::new(), }; - let notify = self - .context - .as_ref() - .map(|context| context.notify()) - .unwrap_or_else(default_notify_interface); + let notify = resolve_notify_interface_for_context(self.context.as_deref()); let req_params = extract_params_header(&req.headers); let host = get_request_host(&req.headers); let port = get_request_port(&req.headers); diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index e30487307..7e446663e 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -14,6 +14,7 @@ use crate::admin::console::is_console_path; use crate::admin::handlers::health::{HealthProbe, build_health_response_parts}; +use crate::app::context::resolve_kms_runtime_service_manager; use crate::error::ApiError; use crate::server::RemoteAddr; use crate::server::cors; @@ -903,7 +904,7 @@ fn is_public_health_endpoint_request(method: &Method, path: &str) -> bool { } async fn health_kms_ready() -> bool { - let Some(service_manager) = rustfs_kms::get_global_kms_service_manager() else { + let Some(service_manager) = resolve_kms_runtime_service_manager() else { return true; }; diff --git a/rustfs/src/storage/ecfs_extend.rs b/rustfs/src/storage/ecfs_extend.rs index 9fa720be7..687590c1e 100644 --- a/rustfs/src/storage/ecfs_extend.rs +++ b/rustfs/src/storage/ecfs_extend.rs @@ -17,7 +17,8 @@ use super::{ StorageError, add_object_lock_years, get_bucket_cors_config, get_bucket_object_lock_config, get_bucket_replication_config, resolve_object_store_handle, }; -use crate::config::{RustFSBufferConfig, WorkloadProfile, get_global_buffer_config, is_buffer_profile_enabled}; +use crate::app::context::resolve_buffer_config; +use crate::config::{RustFSBufferConfig, WorkloadProfile, is_buffer_profile_enabled}; use crate::error::ApiError; use crate::server::cors; use crate::storage::ecfs::ListObjectUnorderedQuery; @@ -261,8 +262,8 @@ pub(crate) fn get_adaptive_buffer_size_with_profile(file_size: i64, profile: Opt /// ``` pub(crate) fn get_buffer_size_opt_in(file_size: i64) -> usize { let buffer_size = if is_buffer_profile_enabled() { - // Use globally configured workload profile (enabled by default in Phase 3) - let config = get_global_buffer_config(); + // Use the AppContext-owned profile when available, with global fallback during migration. + let config = resolve_buffer_config(); config.get_buffer_size(file_size) } else { // Opt-out mode: Use GeneralPurpose profile for consistent behavior diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index c29df942e..79ffc64e4 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -137,6 +137,7 @@ E2E_STORAGE_COMPAT_RPC_PASSTHROUGH_HITS_FILE="${TMP_DIR}/e2e_storage_compat_rpc_ TEST_STORAGE_COMPAT_PASSTHROUGH_HITS_FILE="${TMP_DIR}/test_storage_compat_passthrough_hits.txt" TEST_FUZZ_COMPAT_BRIDGE_HITS_FILE="${TMP_DIR}/test_fuzz_compat_bridge_hits.txt" STANDALONE_THIN_COMPAT_BRIDGE_HITS_FILE="${TMP_DIR}/standalone_thin_compat_bridge_hits.txt" +APP_NOTIFY_THIN_COMPAT_BRIDGE_HITS_FILE="${TMP_DIR}/app_notify_thin_compat_bridge_hits.txt" EXTERNAL_OWNER_COMPAT_BRIDGE_HITS_FILE="${TMP_DIR}/external_owner_compat_bridge_hits.txt" PRODUCTION_UNUSED_COMPAT_ALLOW_HITS_FILE="${TMP_DIR}/production_unused_compat_allow_hits.txt" BROAD_STORE_API_COMPAT_REEXPORT_HITS_FILE="${TMP_DIR}/broad_store_api_compat_reexport_hits.txt" @@ -1653,6 +1654,21 @@ if [[ -s "$STANDALONE_THIN_COMPAT_BRIDGE_HITS_FILE" ]]; then report_failure "storage owner and standalone e2e/IAM/heal/scanner/notify/obs/swift/s3select consumers must import owner APIs directly instead of local thin compatibility bridges: $(paste -sd '; ' "$STANDALONE_THIN_COMPAT_BRIDGE_HITS_FILE")" fi +( + cd "$ROOT_DIR" + { + [[ -e rustfs/src/app/context/compat.rs ]] && printf '%s:1:app context compatibility bridge file exists\n' "rustfs/src/app/context/compat.rs" + [[ -e crates/notify/src/event_bridge.rs ]] && printf '%s:1:notify event bridge re-export file exists\n' "crates/notify/src/event_bridge.rs" + rg -n --with-filename '\bmod\s+compat;|pub\s+use\s+compat::\*|\bmod\s+event_bridge;|pub\s+use\s+event_bridge::' \ + rustfs/src/app/context.rs \ + crates/notify/src/lib.rs || true + } +) >"$APP_NOTIFY_THIN_COMPAT_BRIDGE_HITS_FILE" + +if [[ -s "$APP_NOTIFY_THIN_COMPAT_BRIDGE_HITS_FILE" ]]; then + report_failure "app context and notify event bridge thin compatibility modules must stay collapsed into owner roots: $(paste -sd '; ' "$APP_NOTIFY_THIN_COMPAT_BRIDGE_HITS_FILE")" +fi + ( cd "$ROOT_DIR" {