From 6fcd62ba5648f90f6a78a0819f54e4a881cc2a87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Sun, 14 Jun 2026 09:38:12 +0800 Subject: [PATCH] refactor: prefer app context object store in usecases (#3425) --- docs/architecture/migration-progress.md | 58 +++++++++++---------- rustfs/src/app/admin_usecase.rs | 16 ++++-- rustfs/src/app/bucket_usecase.rs | 68 ++++++++++++++----------- rustfs/src/app/multipart_usecase.rs | 24 ++++++--- rustfs/src/app/object_usecase.rs | 22 +++++--- 5 files changed, 113 insertions(+), 75 deletions(-) diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 8492a9c30..2bcf6533c 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,15 +5,16 @@ 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-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. +- Branch: `overtrue/arch-app-usecase-object-store-context` +- Baseline: `origin/main` at `bed875d3e0d114398da8b2f94d4473293cbc45a8` +- PR type for this branch: `consumer-migration` +- Runtime behavior changes: no external behavior change expected; app usecases + prefer the AppContext-owned object store and keep the existing global + fallback. +- Rust code changes: migrate admin, bucket, multipart, and object usecase + object-store lookups to AppContext-first resolution. - CI/script changes: none. -- Docs changes: record `CTX-002`/`CTX-003` completion and verification scope. +- Docs changes: record `CTX-004` consumer migration scope and verification. ## Phase 0 Tasks @@ -159,6 +160,13 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block - Verification: focused IAM recovery test, formatting, compile checks, migration guards, diff hygiene, Rust risk scan, and full `make pre-commit`. +- [x] `CTX-004` Migrate app usecase object-store consumers. + - Do: migrate admin, bucket, multipart, and object usecases to resolve the + object store from AppContext first. + - Acceptance: usecase object-store lookups use AppContext when present and + preserve the existing global object-layer fallback when absent. + - Verification: formatting, compile check, migration guards, diff hygiene, + Rust risk scan, and full `make pre-commit`. ## Phase 1 Security Governance Tasks @@ -467,41 +475,39 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block 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. +2. `consumer-migration`: migrate the next low-risk AppContext consumer group + while preserving global fallback. ## Pre-Push Review Log | Expert | Status | Notes | |---|---|---| -| 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. | +| Quality/architecture | pass | Single `consumer-migration` slice; helpers stay private to each usecase and no service container, registry, or cross-layer dependency is added. | +| Migration preservation | pass | AppContext object-store resolution is preferred when present, existing `new_object_layer_fn` fallback and `Not init` error paths are preserved, and method bodies keep their storage behavior. | +| Testing/verification | pass | Formatting, compile check, diff hygiene, migration guards, added-line Rust risk scan, and full `make pre-commit` passed. | ## Verification Notes -Passed on `cb37a64a81bb857ea22efd2ebb015fbd3b64bef0`: +Passed on `bed875d3e0d114398da8b2f94d4473293cbc45a8`: - `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`. - `git diff --check`. - `./scripts/check_architecture_migration_rules.sh`. - `./scripts/check_layer_dependencies.sh`. -- 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 +- Rust risk scan for changed Rust files; full-file matches were existing + code/tests, and the added-line scan returned no `unwrap`/`expect`, numeric + cast, string error, boxed error, print macro, or relaxed-ordering match. +- `make pre-commit`; all checks passed, including nextest 5905 passed and 111 skipped, plus doctests. Notes: -- 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`. +- This slice migrates one coherent app usecase consumer group. +- Global object-layer fallback remains available until the planned cleanup + phase. ## Handoff Notes -- 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. +- CTX-004 is complete. +- Keep the next consumer or boot wrapper PR scoped to one PR type and preserve + global fallback until the planned cleanup phase. diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 13e1abf39..6bfa298f7 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -24,6 +24,7 @@ use rustfs_ecstore::data_usage::{apply_bucket_usage_memory_overlay, load_data_us use rustfs_ecstore::endpoints::EndpointServerPools; use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::pools::{PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free}; +use rustfs_ecstore::store::ECStore; use rustfs_madmin::{InfoMessage, StorageInfo}; use rustfs_storage_api::StorageAdminApi; use s3s::S3ErrorCode; @@ -102,6 +103,13 @@ impl DefaultAdminUsecase { self.context.as_ref().and_then(|context| context.endpoints().handle()) } + fn object_store(&self) -> Option> { + self.context + .as_ref() + .map(|context| context.object_store()) + .or_else(new_object_layer_fn) + } + fn app_error(code: S3ErrorCode, message: impl Into) -> ApiError { ApiError { code, @@ -121,7 +129,7 @@ impl DefaultAdminUsecase { } pub async fn execute_query_storage_info(&self) -> AdminUsecaseResult { - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); }; @@ -129,7 +137,7 @@ impl DefaultAdminUsecase { } pub async fn execute_query_data_usage_info(&self) -> AdminUsecaseResult { - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); }; @@ -208,7 +216,7 @@ impl DefaultAdminUsecase { } pub async fn execute_list_pool_statuses(&self) -> AdminUsecaseResult> { - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); }; @@ -255,7 +263,7 @@ impl DefaultAdminUsecase { return Err(Self::app_error_default(S3ErrorCode::InvalidArgument)); }; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); }; diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index d377c56cb..b92039407 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -55,6 +55,7 @@ use rustfs_ecstore::client::object_api_utils::to_s3s_etag; use rustfs_ecstore::error::StorageError; use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::notification_sys::get_global_notification_sys; +use rustfs_ecstore::store::ECStore; use rustfs_ecstore::store_api::{BucketOperations, ListObjectVersionsInfo, ListObjectsV2Info, ListOperations, ObjectInfo}; use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta}; use rustfs_policy::policy::{ @@ -738,6 +739,13 @@ impl DefaultBucketUsecase { self.context.as_ref().and_then(|context| context.region().get()) } + fn object_store(&self) -> Option> { + self.context + .as_ref() + .map(|context| context.object_store()) + .or_else(new_object_layer_fn) + } + #[instrument( level = "debug", skip(self, req), @@ -758,7 +766,7 @@ impl DefaultBucketUsecase { } = req.input; let lock_enabled = object_lock_enabled_for_bucket.is_some_and(|v| v); - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -801,7 +809,7 @@ impl DefaultBucketUsecase { let helper = OperationHelper::new(&req, EventName::BucketRemoved, S3Operation::DeleteBucket); let input = req.input.clone(); - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -839,7 +847,7 @@ impl DefaultBucketUsecase { pub async fn execute_head_bucket(&self, req: S3Request) -> S3Result> { let input = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -858,7 +866,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let input = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -878,7 +886,7 @@ impl DefaultBucketUsecase { #[instrument(level = "debug", skip(self))] pub async fn execute_list_buckets(&self, req: S3Request) -> S3Result> { - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -936,7 +944,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let DeleteBucketEncryptionInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -964,7 +972,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let DeleteBucketCorsInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -993,7 +1001,7 @@ impl DefaultBucketUsecase { let request_context = req.extensions.get::().cloned(); let DeleteBucketLifecycleInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1023,7 +1031,7 @@ impl DefaultBucketUsecase { let request_context = req.extensions.get::().cloned(); let DeleteBucketPolicyInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1052,7 +1060,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let DeleteBucketReplicationInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1112,7 +1120,7 @@ impl DefaultBucketUsecase { let request_context = req.extensions.get::().cloned(); let DeletePublicAccessBlockInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1136,7 +1144,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let GetBucketEncryptionInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1172,7 +1180,7 @@ impl DefaultBucketUsecase { pub async fn execute_get_bucket_cors(&self, req: S3Request) -> S3Result> { let GetBucketCorsInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1214,7 +1222,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let GetBucketLifecycleConfigurationInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1242,7 +1250,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let GetBucketNotificationConfigurationInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1287,7 +1295,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let GetBucketPolicyInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1317,7 +1325,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let GetBucketPolicyStatusInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1402,7 +1410,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let GetBucketReplicationInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1437,7 +1445,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let GetBucketTaggingInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1474,7 +1482,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let GetPublicAccessBlockInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1507,7 +1515,7 @@ impl DefaultBucketUsecase { req: S3Request, ) -> S3Result> { let GetBucketVersioningInput { bucket, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1555,7 +1563,7 @@ impl DefaultBucketUsecase { info!("sse_config {:?}", &server_side_encryption_configuration); - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1639,7 +1647,7 @@ impl DefaultBucketUsecase { } if lifecycle_has_transition_rules(&input_cfg) - && let Some(store) = new_object_layer_fn() + && let Some(store) = self.object_store() { let bucket_name = bucket.clone(); let request_context = req.extensions.get::().cloned(); @@ -1651,7 +1659,7 @@ impl DefaultBucketUsecase { } if lifecycle_has_expiry_rules(&input_cfg) - && let Some(store) = new_object_layer_fn() + && let Some(store) = self.object_store() { let bucket_name = bucket.clone(); let request_context = req.extensions.get::().cloned(); @@ -1679,7 +1687,7 @@ impl DefaultBucketUsecase { validate_notification_configuration_filters(¬ification_configuration)?; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1747,7 +1755,7 @@ impl DefaultBucketUsecase { let request_context = req.extensions.get::().cloned(); let PutBucketPolicyInput { bucket, policy, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1817,7 +1825,7 @@ impl DefaultBucketUsecase { .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1852,7 +1860,7 @@ impl DefaultBucketUsecase { } = req.input; info!(bucket = %bucket, "updating bucket replication config"); - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1890,7 +1898,7 @@ impl DefaultBucketUsecase { .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1916,7 +1924,7 @@ impl DefaultBucketUsecase { ) -> S3Result> { let PutBucketTaggingInput { bucket, tagging, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 5b6453dc9..d009cd985 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -53,6 +53,7 @@ use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::rio::{DecryptReader, EncryptReader, HardLimitReader, boxed_reader, wrap_reader}; use rustfs_ecstore::rio::{HashReader, WritePlan}; use rustfs_ecstore::set_disk::is_valid_storage_class; +use rustfs_ecstore::store::ECStore; use rustfs_ecstore::store_api::{CompletePart, HTTPRangeSpec, MultipartUploadResult, ObjectIO, ObjectOptions, PutObjReader}; use rustfs_ecstore::store_api::{MultipartOperations, ObjectOperations}; use rustfs_filemeta::{ReplicationStatusType, ReplicationType}; @@ -228,6 +229,13 @@ impl DefaultMultipartUsecase { self.context.as_ref().and_then(|context| context.bucket_metadata().handle()) } + fn object_store(&self) -> Option> { + self.context + .as_ref() + .map(|context| context.object_store()) + .or_else(new_object_layer_fn) + } + #[instrument(level = "debug", skip(self))] pub async fn execute_abort_multipart_upload( &self, @@ -237,7 +245,7 @@ impl DefaultMultipartUsecase { bucket, key, upload_id, .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -288,7 +296,7 @@ impl DefaultMultipartUsecase { validate_table_catalog_object_mutation(&bucket, &key).await?; if if_match.is_some() || if_none_match.is_some() { - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -349,7 +357,7 @@ impl DefaultMultipartUsecase { )); } - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -548,7 +556,7 @@ impl DefaultMultipartUsecase { validate_table_catalog_object_mutation(&bucket, &key).await?; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -731,7 +739,7 @@ impl DefaultMultipartUsecase { } // Get multipart info early to check if managed encryption will be applied - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -949,7 +957,7 @@ impl DefaultMultipartUsecase { max_uploads, } = parse_list_multipart_uploads_params(prefix, key_marker, max_uploads)?; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -973,7 +981,7 @@ impl DefaultMultipartUsecase { let params = parse_list_parts_params(part_number_marker, max_parts)?; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -1035,7 +1043,7 @@ impl DefaultMultipartUsecase { validate_table_catalog_object_mutation(&bucket, &key).await?; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index c8bc05c12..cec12926b 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -76,6 +76,7 @@ use rustfs_ecstore::error::{StorageError, is_err_bucket_not_found, is_err_object use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::rio::{DynReader, HashReader, WritePlan, wrap_reader}; use rustfs_ecstore::set_disk::{get_lock_acquire_timeout, is_valid_storage_class}; +use rustfs_ecstore::store::ECStore; use rustfs_ecstore::store_api::{ HTTPRangeSpec, NamespaceLocking, ObjectIO, ObjectInfo, ObjectOperations, ObjectOptions, ObjectToDelete, PutObjReader, }; @@ -1153,6 +1154,13 @@ impl DefaultObjectUsecase { self.context.as_ref().and_then(|context| context.bucket_metadata().handle()) } + fn object_store(&self) -> Option> { + self.context + .as_ref() + .map(|context| context.object_store()) + .or_else(new_object_layer_fn) + } + fn base_buffer_size(&self) -> usize { self.context .clone() @@ -2501,7 +2509,7 @@ impl DefaultObjectUsecase { .. } = req.input; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -2798,7 +2806,7 @@ impl DefaultObjectUsecase { let cp_src_dst_same = path_join_buf(&[&src_bucket, &src_key]) == path_join_buf(&[&bucket, &key]); - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -3086,7 +3094,7 @@ impl DefaultObjectUsecase { ) .await; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -3439,7 +3447,7 @@ impl DefaultObjectUsecase { // } } - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -3658,7 +3666,7 @@ impl DefaultObjectUsecase { .await .map_err(ApiError::from)?; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; // Modification Points: Explicitly handles get_object_info errors, distinguishing between object absence and other errors @@ -3962,7 +3970,7 @@ impl DefaultObjectUsecase { S3Error::with_message(S3ErrorCode::Custom("ErrValidRestoreObject".into()), "restore request is required") })?; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; @@ -4304,7 +4312,7 @@ impl DefaultObjectUsecase { s3_error!(InvalidArgument, "get entries err") })?; - let Some(store) = new_object_layer_fn() else { + let Some(store) = self.object_store() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); };