From d774d6821b398d89065ad4d87746fe19b25a02e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Wed, 25 Feb 2026 15:07:09 +0800 Subject: [PATCH] refactor(app): route metadata/endpoints access through AppContext (#1949) --- rustfs/src/app/admin_usecase.rs | 14 ++++++-- rustfs/src/app/context.rs | 54 +++++++++++++++++++++++++++++ rustfs/src/app/multipart_usecase.rs | 17 ++++++--- rustfs/src/app/object_usecase.rs | 23 ++++++++---- 4 files changed, 94 insertions(+), 14 deletions(-) diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index bd4709247..770370e5e 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -20,9 +20,10 @@ use crate::error::ApiError; use rustfs_common::data_usage::DataUsageInfo; use rustfs_ecstore::admin_server_info::get_server_info; use rustfs_ecstore::data_usage::load_data_usage_from_backend; +use rustfs_ecstore::endpoints::EndpointServerPools; +use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::pools::{PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free}; use rustfs_ecstore::store_api::StorageAPI; -use rustfs_ecstore::{GLOBAL_Endpoints, new_object_layer_fn}; use rustfs_madmin::{InfoMessage, StorageInfo}; use s3s::S3ErrorCode; use std::sync::Arc; @@ -96,6 +97,13 @@ impl DefaultAdminUsecase { self.context.clone() } + fn endpoints(&self) -> Option { + self.context + .as_ref() + .and_then(|context| context.endpoints().handle()) + .or_else(|| rustfs_ecstore::GLOBAL_Endpoints.get().cloned()) + } + fn app_error(code: S3ErrorCode, message: impl Into) -> ApiError { ApiError { code, @@ -221,7 +229,7 @@ impl DefaultAdminUsecase { return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); }; - let Some(endpoints) = GLOBAL_Endpoints.get() else { + let Some(endpoints) = self.endpoints() else { return Err(Self::app_error_default(S3ErrorCode::NotImplemented)); }; @@ -243,7 +251,7 @@ impl DefaultAdminUsecase { let _ = context.object_store(); } - let Some(endpoints) = GLOBAL_Endpoints.get() else { + let Some(endpoints) = self.endpoints() else { return Err(Self::app_error_default(S3ErrorCode::NotImplemented)); }; diff --git a/rustfs/src/app/context.rs b/rustfs/src/app/context.rs index 0d97b8e26..74d59e6fc 100644 --- a/rustfs/src/app/context.rs +++ b/rustfs/src/app/context.rs @@ -18,12 +18,16 @@ #![allow(dead_code)] use async_trait::async_trait; +use rustfs_ecstore::GLOBAL_Endpoints; +use rustfs_ecstore::bucket::metadata_sys::{BucketMetadataSys, GLOBAL_BucketMetadataSys}; +use rustfs_ecstore::endpoints::EndpointServerPools; use rustfs_ecstore::store::ECStore; use rustfs_iam::{store::object::ObjectStore, sys::IamSys}; use rustfs_kms::KmsServiceManager; use rustfs_notify::{EventArgs, NotificationError, notifier_global}; use rustfs_targets::{EventName, arn::TargetID}; use std::sync::{Arc, OnceLock}; +use tokio::sync::RwLock; /// IAM interface for application-layer use-cases. pub trait IamInterface: Send + Sync { @@ -51,6 +55,16 @@ pub trait NotifyInterface: Send + Sync { async fn clear_bucket_notification_rules(&self, bucket_name: &str) -> Result<(), NotificationError>; } +/// Bucket metadata interface for application-layer use-cases. +pub trait BucketMetadataInterface: Send + Sync { + fn handle(&self) -> Option>>; +} + +/// Endpoints interface for application-layer use-cases. +pub trait EndpointsInterface: Send + Sync { + fn handle(&self) -> Option; +} + /// Default IAM interface adapter. pub struct IamHandle { iam: Arc>, @@ -113,6 +127,26 @@ impl NotifyInterface for NotifyHandle { } } +/// Default bucket metadata interface adapter. +#[derive(Default)] +pub struct BucketMetadataHandle; + +impl BucketMetadataInterface for BucketMetadataHandle { + fn handle(&self) -> Option>> { + GLOBAL_BucketMetadataSys.get().cloned() + } +} + +/// Default endpoints interface adapter. +#[derive(Default)] +pub struct EndpointsHandle; + +impl EndpointsInterface for EndpointsHandle { + fn handle(&self) -> Option { + GLOBAL_Endpoints.get().cloned() + } +} + /// Application-layer context with explicit dependencies. #[derive(Clone)] pub struct AppContext { @@ -120,6 +154,8 @@ pub struct AppContext { iam: Arc, kms: Arc, notify: Arc, + bucket_metadata: Arc, + endpoints: Arc, } impl AppContext { @@ -129,6 +165,8 @@ impl AppContext { iam, kms, notify: default_notify_interface(), + bucket_metadata: default_bucket_metadata_interface(), + endpoints: default_endpoints_interface(), } } @@ -155,12 +193,28 @@ impl AppContext { pub fn notify(&self) -> Arc { self.notify.clone() } + + pub fn bucket_metadata(&self) -> Arc { + self.bucket_metadata.clone() + } + + pub fn endpoints(&self) -> Arc { + self.endpoints.clone() + } } pub fn default_notify_interface() -> Arc { Arc::new(NotifyHandle) } +pub fn default_bucket_metadata_interface() -> Arc { + Arc::new(BucketMetadataHandle) +} + +pub fn default_endpoints_interface() -> Arc { + Arc::new(EndpointsHandle) +} + static GLOBAL_APP_CONTEXT: OnceLock> = OnceLock::new(); /// Initialize global application context once and return the canonical instance. diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 7fd5417a9..ca85a81ba 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -30,6 +30,7 @@ use futures::StreamExt; use rustfs_ecstore::StorageAPI; use rustfs_ecstore::bucket::quota::checker::QuotaChecker; use rustfs_ecstore::bucket::{ + metadata_sys, quota::QuotaOperation, replication::{get_must_replicate_options, must_replicate, schedule_replication}, }; @@ -52,6 +53,7 @@ use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use std::collections::HashMap; use std::str::FromStr; use std::sync::Arc; +use tokio::sync::RwLock; use tokio_util::io::StreamReader; use tracing::{info, instrument, warn}; @@ -158,6 +160,13 @@ impl DefaultMultipartUsecase { self.context.clone() } + fn bucket_metadata_sys(&self) -> Option>> { + self.context + .as_ref() + .and_then(|context| context.bucket_metadata().handle()) + .or_else(|| rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get().cloned()) + } + #[instrument(level = "debug", skip(self))] pub async fn execute_abort_multipart_upload( &self, @@ -337,8 +346,8 @@ impl DefaultMultipartUsecase { .map_err(ApiError::from)?; // check quota after completing multipart upload - if let Some(metadata_sys) = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get() { - let quota_checker = QuotaChecker::new(metadata_sys.clone()); + if let Some(metadata_sys) = self.bucket_metadata_sys() { + let quota_checker = QuotaChecker::new(metadata_sys); match quota_checker .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size as u64) @@ -358,9 +367,7 @@ impl DefaultMultipartUsecase { )); } // Update quota tracking after successful multipart upload - if rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get().is_some() { - rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await; - } + rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, obj_info.size as u64).await; } Err(e) => { warn!("Quota check failed for bucket {}: {}, allowing operation", bucket, e); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 064947d37..59de9ae90 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -101,6 +101,7 @@ use std::ops::Add; use std::str::FromStr; use std::sync::Arc; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; +use tokio::sync::RwLock; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tokio_util::io::{ReaderStream, StreamReader}; @@ -199,6 +200,13 @@ impl DefaultObjectUsecase { self.context.clone() } + fn bucket_metadata_sys(&self) -> Option>> { + self.context + .as_ref() + .and_then(|context| context.bucket_metadata().handle()) + .or_else(|| rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get().cloned()) + } + #[instrument(level = "debug", skip(self, fs, req))] pub async fn execute_put_object(&self, fs: &FS, req: S3Request) -> S3Result> { if let Some(context) = &self.context { @@ -258,9 +266,9 @@ impl DefaultObjectUsecase { // check quota for put operation if let Some(size) = content_length - && let Some(metadata_sys) = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get() + && let Some(metadata_sys) = self.bucket_metadata_sys() { - let quota_checker = QuotaChecker::new(metadata_sys.clone()); + let quota_checker = QuotaChecker::new(metadata_sys); match quota_checker .check_quota(&bucket, QuotaOperation::PutObject, size as u64) @@ -2091,8 +2099,8 @@ impl DefaultObjectUsecase { } // check quota for copy operation - if let Some(metadata_sys) = rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get() { - let quota_checker = QuotaChecker::new(metadata_sys.clone()); + let has_bucket_metadata = if let Some(metadata_sys) = self.bucket_metadata_sys() { + let quota_checker = QuotaChecker::new(metadata_sys); match quota_checker .check_quota(&bucket, QuotaOperation::CopyObject, src_info.size as u64) @@ -2114,7 +2122,10 @@ impl DefaultObjectUsecase { warn!("Quota check failed for bucket {}: {}, allowing operation", bucket, e); } } - } + true + } else { + false + }; let oi = store .copy_object(&src_bucket, &src_key, &bucket, &key, &mut src_info, &src_opts, &dst_opts) @@ -2122,7 +2133,7 @@ impl DefaultObjectUsecase { .map_err(ApiError::from)?; // Update quota tracking after successful copy - if rustfs_ecstore::bucket::metadata_sys::GLOBAL_BucketMetadataSys.get().is_some() { + if has_bucket_metadata { rustfs_ecstore::data_usage::increment_bucket_usage_memory(&bucket, oi.size as u64).await; }