From c13f42e9a021972af1fe6d20488fad72c3ab74b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Mon, 22 Jun 2026 06:31:24 +0800 Subject: [PATCH] refactor: wrap disk rpc compat trait methods (#3707) --- crates/ecstore/src/api/mod.rs | 8 +- crates/heal/src/heal/manager.rs | 3 +- crates/heal/src/heal/resume.rs | 3 +- crates/heal/src/heal/storage_compat.rs | 46 +++ crates/scanner/src/scanner_folder.rs | 9 +- crates/scanner/src/scanner_io.rs | 8 +- crates/scanner/src/storage_compat.rs | 37 +++ crates/scanner/tests/common/storage_compat.rs | 14 + .../tests/lifecycle_integration_test.rs | 12 +- docs/architecture/crate-boundaries.md | 7 +- docs/architecture/migration-progress.md | 47 ++- .../src/app/lifecycle_transition_api_test.rs | 5 +- rustfs/src/app/storage_compat.rs | 2 + rustfs/src/storage/rpc/bucket.rs | 1 - rustfs/src/storage/rpc/http_service.rs | 2 +- rustfs/src/storage/rpc/node_service.rs | 7 +- rustfs/src/storage/storage_compat.rs | 294 ++++++++++++++++++ scripts/check_architecture_migration_rules.sh | 7 +- 18 files changed, 460 insertions(+), 52 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index d6ace40ed..9f7bab95f 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -77,12 +77,14 @@ pub mod disk { pub use crate::disk::endpoint::Endpoint; pub use crate::disk::error::DiskError; pub use crate::disk::error_reduce::is_all_buckets_not_found; + pub use crate::disk::local::ScanGuard; pub use crate::disk::{ - BUCKET_META_PREFIX, DeleteOptions, Disk, DiskAPI, DiskInfoOptions, DiskOption, DiskStore, FileInfoVersions, - RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, - WalkDirOptions, new_disk, + BUCKET_META_PREFIX, CheckPartsResp, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, + DiskStore, FileInfoVersions, FileReader, FileWriter, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, + RenameDataResp, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk, }; pub use crate::disk::{endpoint, error, error_reduce}; + pub use bytes::Bytes; } pub mod error { diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 50cac731c..f10a03a30 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -20,7 +20,6 @@ use crate::heal::{ use crate::{Error, Result}; use metrics::{counter, gauge}; use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult, HealRequestSource}; -use rustfs_ecstore::api::disk::DiskAPI as _; use rustfs_madmin::heal_commands::HealResultItem; use std::{ collections::{BinaryHeap, HashMap}, @@ -34,7 +33,7 @@ use tokio::{ use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; -use super::storage_compat::{DiskError, GLOBAL_LOCAL_DISK_MAP}; +use super::storage_compat::{DiskError, GLOBAL_LOCAL_DISK_MAP, HealDiskExt as _}; const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60); const LOG_COMPONENT_HEAL: &str = "heal"; diff --git a/crates/heal/src/heal/resume.rs b/crates/heal/src/heal/resume.rs index 9dac25a58..88d2ee79b 100644 --- a/crates/heal/src/heal/resume.rs +++ b/crates/heal/src/heal/resume.rs @@ -13,7 +13,6 @@ // limitations under the License. use crate::{Error, Result}; -use rustfs_ecstore::api::disk::DiskAPI as _; use serde::{Deserialize, Serialize}; use std::path::Path; use std::sync::Arc; @@ -22,7 +21,7 @@ use tokio::sync::RwLock; use tracing::{debug, warn}; use uuid::Uuid; -use super::storage_compat::{BUCKET_META_PREFIX, DiskError, DiskStore, RUSTFS_META_BUCKET}; +use super::storage_compat::{BUCKET_META_PREFIX, DiskError, DiskStore, HealDiskExt as _, RUSTFS_META_BUCKET}; const LOG_COMPONENT_HEAL: &str = "heal"; const LOG_SUBSYSTEM_RESUME: &str = "resume"; diff --git a/crates/heal/src/heal/storage_compat.rs b/crates/heal/src/heal/storage_compat.rs index b02eeb5b2..ccbd556de 100644 --- a/crates/heal/src/heal/storage_compat.rs +++ b/crates/heal/src/heal/storage_compat.rs @@ -22,6 +22,7 @@ pub(crate) const BUCKET_META_PREFIX: &str = ecstore_disk::BUCKET_META_PREFIX; pub(crate) const RUSTFS_META_BUCKET: &str = ecstore_disk::RUSTFS_META_BUCKET; pub(crate) type DiskError = ecstore_disk::error::DiskError; +pub(crate) type DiskResult = ecstore_disk::error::Result; pub(crate) type DiskStore = ecstore_disk::DiskStore; pub(crate) type ECStore = ecstore_storage::ECStore; pub(crate) type EcstoreError = ecstore_error::Error; @@ -47,6 +48,51 @@ pub(crate) async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> ecstore_disk::e ecstore_disk::new_disk(ep, opt).await } +pub(crate) trait HealDiskExt { + fn endpoint(&self) -> Endpoint; + async fn get_disk_id(&self) -> DiskResult>; + async fn read_all(&self, volume: &str, path: &str) -> DiskResult; + async fn write_all(&self, volume: &str, path: &str, data: ecstore_disk::Bytes) -> DiskResult<()>; + async fn delete(&self, volume: &str, path: &str, options: ecstore_disk::DeleteOptions) -> DiskResult<()>; + async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult>; + #[cfg(test)] + async fn make_volume(&self, volume: &str) -> DiskResult<()>; +} + +impl HealDiskExt for T +where + T: ecstore_disk::DiskAPI, +{ + fn endpoint(&self) -> Endpoint { + ecstore_disk::DiskAPI::endpoint(self) + } + + async fn get_disk_id(&self) -> DiskResult> { + ecstore_disk::DiskAPI::get_disk_id(self).await + } + + async fn read_all(&self, volume: &str, path: &str) -> DiskResult { + ecstore_disk::DiskAPI::read_all(self, volume, path).await + } + + async fn write_all(&self, volume: &str, path: &str, data: ecstore_disk::Bytes) -> DiskResult<()> { + ecstore_disk::DiskAPI::write_all(self, volume, path, data).await + } + + async fn delete(&self, volume: &str, path: &str, options: ecstore_disk::DeleteOptions) -> DiskResult<()> { + ecstore_disk::DiskAPI::delete(self, volume, path, options).await + } + + async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult> { + ecstore_disk::DiskAPI::list_dir(self, origvolume, volume, dir_path, count).await + } + + #[cfg(test)] + async fn make_volume(&self, volume: &str) -> DiskResult<()> { + ecstore_disk::DiskAPI::make_volume(self, volume).await + } +} + pub type HealObjectInfo = ::ObjectInfo; pub type HealObjectOptions = ::ObjectOptions; pub type HealPutObjReader = ::PutObjectReader; diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 6ae2f9d34..d67bc89df 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -39,7 +39,6 @@ use rustfs_common::metrics::{ IlmAction, Metric, Metrics, ScannerReplicationRepairKind, ScannerSourceWorkUpdate, ScannerWorkSource, UpdateCurrentPathFn, current_path_updater, global_metrics, }; -use rustfs_ecstore::api::disk::DiskAPI as _; use rustfs_filemeta::{ MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicateObjectInfo, ReplicationStatusType, ReplicationType, }; @@ -53,10 +52,10 @@ use tracing::{debug, error, warn}; use crate::storage_compat::{ BucketVersioningSys, Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts, - ReplicationConfig, ReplicationQueueAdmission, ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _, - ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule, enqueue_global_newer_noncurrent, - is_erasure, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object, path2_bucket_object_with_base_path, - queue_replication_heal_internal, + ReplicationConfig, ReplicationQueueAdmission, ScannerDiskExt as _, ScannerLifecycleConfigExt as _, + ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule, + enqueue_global_newer_noncurrent, is_erasure, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object, + path2_bucket_object_with_base_path, queue_replication_heal_internal, }; use crate::{ScannerObjectInfo as ObjectInfo, ScannerObjectToDelete as ObjectToDelete}; diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 0aa4cc7d6..5dd9d5ecd 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -26,7 +26,6 @@ use rustfs_common::heal_channel::HealScanMode; use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete, emit_scan_bucket_drive_partial, global_metrics}; #[cfg(test)] use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS}; -use rustfs_ecstore::api::disk::DiskAPI as _; use rustfs_filemeta::FileMeta; use rustfs_storage_api::{BucketInfo, BucketOperations, BucketOptions, DiskSetSelector, StorageAdminApi}; use rustfs_utils::path::path_join_buf; @@ -46,9 +45,10 @@ use tracing::{debug, error, warn}; use crate::ScannerObjectInfo as ObjectInfo; use crate::storage_compat::{ BucketTargetSys, BucketVersioningSys, Disk, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result, - ReplicationConfig, STORAGE_FORMAT_FILE, ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _, - ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_global_free_version, get_lifecycle_config, - get_object_lock_config, get_replication_config, list_global_tiers, resolve_scanner_object_store_handle, storageclass, + ReplicationConfig, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _, + ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_global_free_version, + get_lifecycle_config, get_object_lock_config, get_replication_config, list_global_tiers, resolve_scanner_object_store_handle, + storageclass, }; pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file"; diff --git a/crates/scanner/src/storage_compat.rs b/crates/scanner/src/storage_compat.rs index e07b709e1..cb7976a56 100644 --- a/crates/scanner/src/storage_compat.rs +++ b/crates/scanner/src/storage_compat.rs @@ -19,6 +19,7 @@ use rustfs_ecstore::api::{ set_disk as ecstore_set_disk, storage as ecstore_storage, tier as ecstore_tier, }; use rustfs_storage_api::{HTTPRangeSpec, ObjectIO, ObjectToDelete}; +use std::path::PathBuf; use std::sync::Arc; use tokio_util::sync::CancellationToken; @@ -30,7 +31,9 @@ pub(crate) const TRANSITION_COMPLETE: &str = ecstore_bucket::lifecycle::lifecycl pub(crate) type Disk = ecstore_disk::Disk; #[cfg(test)] pub(crate) type DiskStore = ecstore_disk::DiskStore; +pub(crate) type DiskLocation = ecstore_disk::DiskLocation; pub(crate) type DiskError = ecstore_disk::error::DiskError; +pub(crate) type DiskResult = ecstore_disk::error::Result; pub(crate) type ECStore = ecstore_storage::ECStore; pub(crate) type EcstoreError = ecstore_error::Error; pub(crate) type EcstoreResult = ecstore_error::Result; @@ -45,6 +48,7 @@ pub(crate) type ObjectOpts = ecstore_bucket::lifecycle::lifecycle::ObjectOpts; pub(crate) type ReplicationConfig = ecstore_bucket::replication::ReplicationConfig; pub(crate) type ReplicationHealQueueResult = ecstore_bucket::replication::ReplicationHealQueueResult; pub(crate) type ReplicationQueueAdmission = ecstore_bucket::replication::ReplicationQueueAdmission; +pub(crate) type ScanGuard = ecstore_disk::ScanGuard; pub(crate) type SetDisks = ecstore_set_disk::SetDisks; pub(crate) type StorageError = ecstore_error::StorageError; @@ -133,6 +137,39 @@ impl ScannerVersioningConfigExt for s3s::dto::VersioningConfiguration { } } +pub(crate) trait ScannerDiskExt { + async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult; + async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult; + fn path(&self) -> PathBuf; + fn get_disk_location(&self) -> DiskLocation; + fn start_scan(&self) -> ScanGuard; +} + +impl ScannerDiskExt for T +where + T: ecstore_disk::DiskAPI, +{ + async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult { + ecstore_disk::DiskAPI::disk_info(self, opts).await + } + + async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult { + ecstore_disk::DiskAPI::read_metadata(self, volume, path).await + } + + fn path(&self) -> PathBuf { + ecstore_disk::DiskAPI::path(self) + } + + fn get_disk_location(&self) -> DiskLocation { + ecstore_disk::DiskAPI::get_disk_location(self) + } + + fn start_scan(&self) -> ScanGuard { + ecstore_disk::DiskAPI::start_scan(self) + } +} + pub(crate) async fn apply_transition_rule(event: &Event, src: &LcEventSrc, oi: &ScannerObjectInfo) -> bool { ecstore_bucket::lifecycle::bucket_lifecycle_ops::apply_transition_rule(event, src, oi).await } diff --git a/crates/scanner/tests/common/storage_compat.rs b/crates/scanner/tests/common/storage_compat.rs index a78b58b4c..1dc9a227e 100644 --- a/crates/scanner/tests/common/storage_compat.rs +++ b/crates/scanner/tests/common/storage_compat.rs @@ -41,8 +41,22 @@ pub(crate) type TierConfigMgr = ecstore_tier::tier::TierConfigMgr; pub(crate) type TierMinIO = ecstore_tier::tier_config::TierMinIO; pub(crate) type TierType = ecstore_tier::tier_config::TierType; pub(crate) type TransitionOptions = ecstore_bucket::lifecycle::lifecycle::TransitionOptions; +pub(crate) use ecstore_tier::warm_backend::WarmBackend as ScannerWarmBackend; pub(crate) type WarmBackendGetOpts = ecstore_tier::warm_backend::WarmBackendGetOpts; +pub(crate) trait ScannerTestDiskExt { + async fn read_metadata(&self, volume: &str, path: &str) -> ecstore_disk::error::Result; +} + +impl ScannerTestDiskExt for T +where + T: ecstore_disk::DiskAPI, +{ + async fn read_metadata(&self, volume: &str, path: &str) -> ecstore_disk::error::Result { + ecstore_disk::DiskAPI::read_metadata(self, volume, path).await + } +} + #[allow(non_snake_case)] pub(crate) fn EndpointServerPools(pools: Vec) -> EndpointServerPools { ecstore_layout::EndpointServerPools::from(pools) diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index 2b10f2d6f..92fdc2ec8 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/crates/scanner/tests/lifecycle_integration_test.rs @@ -16,15 +16,13 @@ mod common; use crate::common::storage_compat::{ BUCKET_LIFECYCLE_CONFIG, BucketVersioningSys, DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints, - GLOBAL_TierConfigMgr, PoolEndpoints, ReadCloser, ReaderImpl, STORAGE_FORMAT_FILE, TierConfig, TierMinIO, TierType, - TransitionOptions, WarmBackendGetOpts, build_transition_put_options, enqueue_transition_for_existing_objects, - get_bucket_metadata, init_background_expiry, init_bucket_metadata_sys, init_local_disks, new_disk, - path2_bucket_object_with_base_path, update_bucket_metadata, + GLOBAL_TierConfigMgr, PoolEndpoints, ReadCloser, ReaderImpl, STORAGE_FORMAT_FILE, ScannerTestDiskExt as _, + ScannerWarmBackend, TierConfig, TierMinIO, TierType, TransitionOptions, WarmBackendGetOpts, build_transition_put_options, + enqueue_transition_for_existing_objects, get_bucket_metadata, init_background_expiry, init_bucket_metadata_sys, + init_local_disks, new_disk, path2_bucket_object_with_base_path, update_bucket_metadata, }; use futures::FutureExt; use rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT; -use rustfs_ecstore::api::disk::DiskAPI as _; -use rustfs_ecstore::api::tier::warm_backend::WarmBackend; use rustfs_filemeta::FileMeta; use rustfs_scanner::scanner_folder::ScannerItem; use rustfs_scanner::scanner_io::ScannerIODisk; @@ -687,7 +685,7 @@ impl MockWarmBackend { } #[async_trait::async_trait] -impl WarmBackend for MockWarmBackend { +impl ScannerWarmBackend for MockWarmBackend { async fn put(&self, object: &str, r: ReaderImpl, _length: i64) -> Result { let bytes = self.read_bytes(r).await?; Ok(self.put_bytes(object, bytes, HashMap::new()).await) diff --git a/docs/architecture/crate-boundaries.md b/docs/architecture/crate-boundaries.md index 49a93a920..39f35749a 100644 --- a/docs/architecture/crate-boundaries.md +++ b/docs/architecture/crate-boundaries.md @@ -200,9 +200,10 @@ same pattern, keeping raw ECStore facade access centralized behind local Outer bucket lifecycle, replication, versioning, object-lock, and restore-request trait method access must stay behind local compatibility traits or wrapper functions. Non-compat sources must not import those ECStore bucket -API traits directly after the wrapper boundary is established. Remaining -temporary direct ECStore method-resolution imports are limited to disk, RPC peer -client, and warm-backend traits until their hot-path wrappers are isolated. +API traits directly after the wrapper boundary is established. Disk, RPC peer +client, and warm-backend method-resolution access must follow the same pattern: +non-compat sources use owner-local compatibility traits or test aliases instead +of importing ECStore traits directly. Scanner, notify, observability, and e2e `storage_compat.rs` boundaries must also stay narrow. Scanner must not restore grouped bucket compatibility exports for target, lifecycle, metadata, replication, or versioning modules. Notify diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 825110276..aa3c29e5e 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,17 +5,17 @@ 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-compat-trait-method-wrappers` -- 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`. -- Stacked on: `overtrue/arch-root-e2e-compat-raw-facade-prune` pending API-095 merge. +- Branch: `overtrue/arch-disk-rpc-method-wrappers` +- 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`. +- Stacked on: `overtrue/arch-compat-trait-method-wrappers` pending API-096 merge. - PR type for this branch: `pure-move` - Runtime behavior changes: none. -- Rust code changes: move remaining outer bucket lifecycle, replication, - versioning, object-lock, and restore-request trait method access behind - local compatibility traits and wrapper functions. -- CI/script changes: stop allowing those bucket trait imports as direct ECStore +- Rust code changes: move remaining disk RPC, peer S3 RPC, heal/scanner disk, + and warm-backend test method access behind local compatibility traits or + aliases. +- CI/script changes: remove the final direct ECStore method-resolution import exceptions outside compatibility boundaries. -- Docs changes: record the API-096 trait-method wrapper cleanup. +- Docs changes: record the API-097 disk/RPC/warm-backend wrapper cleanup. ## Phase 0 Tasks @@ -376,6 +376,20 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block - Verification: RustFS/scanner/heal compile coverage, direct bucket trait import residual scan, migration guard, formatting, diff hygiene, Rust risk scan, pre-commit quality gate, and three-expert review. +- [x] `API-097` Prune disk/RPC/warm-backend method imports. + - Current slice: move disk RPC, peer S3 RPC, heal/scanner disk, and + warm-backend test method access behind local compatibility traits or + aliases in the owning boundaries. + - Acceptance: non-compat RustFS, scanner, heal, and test sources no longer + import ECStore `DiskAPI`, `PeerS3Client`, or `WarmBackend` traits directly; + the migration guard no longer allowlists those direct imports. + - Must preserve: disk RPC request/response behavior, internode HTTP file and + walk streams, heal resume and auto-scan disk handling, scanner disk scan + behavior, and transition warm-backend test harness behavior. + - Verification: RustFS/scanner/heal/e2e compile coverage, direct + disk/RPC/warm-backend trait import residual scan, migration guard, + formatting, diff hygiene, Rust risk scan, pre-commit quality gate, and + three-expert review. - [x] `G-012` Inventory placement and repair invariants. - Acceptance: [`placement-repair-invariants.md`](placement-repair-invariants.md) records @@ -3409,14 +3423,25 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | pass | API-096 localizes bucket trait method access behind owner-local compatibility traits/wrappers without re-exporting ECStore traits or widening the facade. | -| Migration preservation | pass | App/admin/storage/scanner call sites keep the same method names or direct wrapper calls; disk/RPC/warm-backend exceptions are unchanged. | -| Testing/verification | pass | Focused compile, residual scan, migration guard, formatting, diff hygiene, added-line risk scan, and full pre-commit passed. | +| Quality/architecture | pass | API-097 localizes disk/RPC/warm-backend method access behind owner-local compatibility traits and aliases without widening non-compat imports. | +| Migration preservation | pass | RPC, heal, scanner, and transition-test call sites keep existing behavior while losing direct ECStore trait imports. | +| Testing/verification | pass | Focused compile, residual scan, migration guard, layer guard, formatting, diff hygiene, risk scan, and full pre-commit passed. | ## Verification Notes Passed before push: +- Issue #660 API-097 current slice: + - `cargo check -p rustfs -p rustfs-scanner -p rustfs-heal -p e2e_test --tests`: 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. + - `./scripts/check_layer_dependencies.sh`: passed. + - Direct non-compat disk/RPC/warm-backend trait import residual scan: passed. + - Rust risk scan on changed Rust files and guard script: passed. + - `make pre-commit`: passed. + - Issue #660 API-096 current slice: - `cargo check -p rustfs -p rustfs-scanner -p rustfs-heal`: passed. - `cargo fmt --all --check`: passed. diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 3d5ccd361..3c5ed5cf5 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -15,7 +15,7 @@ use super::{multipart_usecase::DefaultMultipartUsecase, object_usecase::DefaultObjectUsecase}; use crate::app::bucket_usecase::DefaultBucketUsecase; use crate::app::storage_compat::{ - ECStore, Endpoint, EndpointServerPools, Endpoints, GLOBAL_TierConfigMgr, PoolEndpoints, TierConfig, TierType, + AppWarmBackend, ECStore, Endpoint, EndpointServerPools, Endpoints, GLOBAL_TierConfigMgr, PoolEndpoints, TierConfig, TierType, WarmBackendGetOpts, metadata::{BUCKET_LIFECYCLE_CONFIG, OBJECT_LOCK_CONFIG}, metadata_sys, @@ -31,7 +31,6 @@ use futures::FutureExt; use futures::stream; use http::{Extensions, HeaderMap, HeaderValue, Method, Uri, header::IF_NONE_MATCH}; use rustfs_config::{ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT}; -use rustfs_ecstore::api::tier::warm_backend::WarmBackend; use rustfs_object_capacity::capacity_manager::{HybridStrategyConfig, create_isolated_manager}; use rustfs_storage_api::{ BucketOperations, BucketOptions, ListOperations as _, MakeBucketOptions, MultipartOperations as _, ObjectIO as _, @@ -295,7 +294,7 @@ impl MockWarmBackend { } #[async_trait::async_trait] -impl WarmBackend for MockWarmBackend { +impl AppWarmBackend for MockWarmBackend { async fn put(&self, object: &str, r: ReaderImpl, _length: i64) -> Result { let bytes = self.read_bytes(r).await?; Ok(self.put_bytes(object, bytes).await) diff --git a/rustfs/src/app/storage_compat.rs b/rustfs/src/app/storage_compat.rs index 8ad90e3bb..3e5efa285 100644 --- a/rustfs/src/app/storage_compat.rs +++ b/rustfs/src/app/storage_compat.rs @@ -57,6 +57,8 @@ pub(crate) type TierConfig = crate::app::storage_compat::ecstore_tier::tier_conf #[cfg(test)] pub(crate) type TierType = crate::app::storage_compat::ecstore_tier::tier_config::TierType; #[cfg(test)] +pub(crate) use crate::app::storage_compat::ecstore_tier::warm_backend::WarmBackend as AppWarmBackend; +#[cfg(test)] pub(crate) type WarmBackendGetOpts = crate::app::storage_compat::ecstore_tier::warm_backend::WarmBackendGetOpts; #[cfg(test)] diff --git a/rustfs/src/storage/rpc/bucket.rs b/rustfs/src/storage/rpc/bucket.rs index ee868d5fe..6083157ae 100644 --- a/rustfs/src/storage/rpc/bucket.rs +++ b/rustfs/src/storage/rpc/bucket.rs @@ -13,7 +13,6 @@ // limitations under the License. use super::*; -use rustfs_ecstore::api::rpc::PeerS3Client as _; impl NodeService { pub(super) async fn handle_delete_bucket_metadata( diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index 18bc92167..3ae40d20c 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -15,6 +15,7 @@ use crate::server::RPC_PREFIX; use crate::storage::request_context::spawn_traced; use crate::storage::storage_compat::DEFAULT_READ_BUFFER_SIZE; +use crate::storage::storage_compat::StorageDiskRpcExt as _; use crate::storage::storage_compat::WalkDirOptions; use crate::storage::storage_compat::find_local_disk_by_ref; use crate::storage::storage_compat::verify_rpc_signature; @@ -24,7 +25,6 @@ use http::{HeaderMap, Method, Request, Response, StatusCode, Uri}; use http_body_util::{BodyExt, Limited}; use hyper::body::Incoming; use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; -use rustfs_ecstore::api::disk::DiskAPI as _; use rustfs_io_metrics::internode_metrics::{ INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index b430e8c78..7fe5b9095 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -19,16 +19,15 @@ use crate::admin::service::{ use crate::storage::storage_compat::{ CollectMetricsOpts, DeleteOptions, DiskError, DiskInfoOptions, DiskStore, FileInfoVersions, LocalPeerS3Client, MetricType, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, ReadMultipleReq, ReadMultipleResp, ReadOptions, SERVICE_SIGNAL_REFRESH_CONFIG, - SERVICE_SIGNAL_RELOAD_DYNAMIC, UpdateMetadataOpts, all_local_disk_path, collect_local_metrics, find_local_disk_by_ref, - get_global_lock_client, get_local_server_property, load_bucket_metadata, reload_transition_tier_config, - resolve_object_store_handle, set_bucket_metadata, + SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt as _, StoragePeerS3ClientExt as _, UpdateMetadataOpts, all_local_disk_path, + collect_local_metrics, find_local_disk_by_ref, get_global_lock_client, get_local_server_property, load_bucket_metadata, + reload_transition_tier_config, resolve_object_store_handle, set_bucket_metadata, }; use bytes::Bytes; use futures::Stream; use futures_util::future::join_all; use rmp_serde::Deserializer; use rustfs_common::{get_global_local_node_name, heal_channel::HealOpts}; -use rustfs_ecstore::api::disk::DiskAPI as _; use rustfs_filemeta::{FileInfo, MetacacheReader}; use rustfs_iam::{get_global_iam_sys, store::UserType}; use rustfs_lock::{LockClient, LockRequest}; diff --git a/rustfs/src/storage/storage_compat.rs b/rustfs/src/storage/storage_compat.rs index 123c6dcb7..0c7bdcabd 100644 --- a/rustfs/src/storage/storage_compat.rs +++ b/rustfs/src/storage/storage_compat.rs @@ -40,24 +40,33 @@ pub(crate) type BucketMetadata = ecstore_bucket::metadata::BucketMetadata; #[cfg(test)] pub(crate) type BucketMetadataSys = ecstore_bucket::metadata_sys::BucketMetadataSys; pub(crate) type BucketVersioningSys = ecstore_bucket::versioning_sys::BucketVersioningSys; +pub(crate) type CheckPartsResp = ecstore_disk::CheckPartsResp; pub(crate) type CollectMetricsOpts = ecstore_metrics::CollectMetricsOpts; pub(crate) type DeleteOptions = ecstore_disk::DeleteOptions; pub(crate) type DiskError = ecstore_disk::error::DiskError; +pub(crate) type DiskInfo = ecstore_disk::DiskInfo; pub(crate) type DiskInfoOptions = ecstore_disk::DiskInfoOptions; +pub(crate) type DiskResult = ecstore_disk::error::Result; pub(crate) type DiskStore = ecstore_disk::DiskStore; pub(crate) type ECStore = ecstore_storage::ECStore; pub(crate) type FileInfoVersions = ecstore_disk::FileInfoVersions; +pub(crate) type FileReader = ecstore_disk::FileReader; +pub(crate) type FileWriter = ecstore_disk::FileWriter; pub(crate) type LocalPeerS3Client = ecstore_rpc::LocalPeerS3Client; pub(crate) type MetricType = ecstore_metrics::MetricType; +pub(crate) type ObjectPartInfo = rustfs_filemeta::ObjectPartInfo; pub(crate) type ObjectLockBlockReason = ecstore_bucket::object_lock::objectlock_sys::ObjectLockBlockReason; pub(crate) type PolicySys = ecstore_bucket::policy_sys::PolicySys; +pub(crate) type RawFileInfo = rustfs_filemeta::RawFileInfo; pub(crate) type ReadMultipleReq = ecstore_disk::ReadMultipleReq; pub(crate) type ReadMultipleResp = ecstore_disk::ReadMultipleResp; pub(crate) type ReadOptions = ecstore_disk::ReadOptions; +pub(crate) type RenameDataResp = ecstore_disk::RenameDataResp; pub(crate) type StorageError = ecstore_error::StorageError; pub(crate) type Error = StorageError; pub(crate) type Result = core::result::Result; pub(crate) type UpdateMetadataOpts = ecstore_disk::UpdateMetadataOpts; +pub(crate) type VolumeInfo = ecstore_disk::VolumeInfo; pub(crate) type WalkDirOptions = ecstore_disk::WalkDirOptions; pub(crate) type WriteEncryption = ecstore_rio::WriteEncryption; @@ -65,6 +74,291 @@ pub(crate) async fn get_local_server_property() -> rustfs_madmin::ServerProperti ecstore_admin::get_local_server_property().await } +pub(crate) trait StorageDiskRpcExt { + async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult; + async fn delete_volume(&self, volume: &str) -> DiskResult<()>; + async fn read_multiple(&self, req: ReadMultipleReq) -> DiskResult>; + async fn delete_versions(&self, volume: &str, versions: Vec, opts: DeleteOptions) + -> Vec>; + async fn delete_version( + &self, + volume: &str, + path: &str, + file_info: rustfs_filemeta::FileInfo, + force_del_marker: bool, + opts: DeleteOptions, + ) -> DiskResult<()>; + async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> DiskResult; + async fn read_version( + &self, + org_volume: &str, + volume: &str, + path: &str, + version_id: &str, + opts: &ReadOptions, + ) -> DiskResult; + async fn write_metadata( + &self, + org_volume: &str, + volume: &str, + path: &str, + file_info: rustfs_filemeta::FileInfo, + ) -> DiskResult<()>; + async fn update_metadata( + &self, + volume: &str, + path: &str, + file_info: rustfs_filemeta::FileInfo, + opts: &UpdateMetadataOpts, + ) -> DiskResult<()>; + async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult; + async fn delete_paths(&self, volume: &str, paths: &[String]) -> DiskResult<()>; + async fn stat_volume(&self, volume: &str) -> DiskResult; + async fn list_volumes(&self) -> DiskResult>; + async fn make_volume(&self, volume: &str) -> DiskResult<()>; + async fn make_volumes(&self, volume: Vec<&str>) -> DiskResult<()>; + async fn rename_data( + &self, + src_volume: &str, + src_path: &str, + file_info: rustfs_filemeta::FileInfo, + dst_volume: &str, + dst_path: &str, + ) -> DiskResult; + async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult>; + async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> DiskResult; + async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> DiskResult<()>; + async fn rename_part( + &self, + src_volume: &str, + src_path: &str, + dst_volume: &str, + dst_path: &str, + meta: bytes::Bytes, + ) -> DiskResult<()>; + async fn delete(&self, volume: &str, path: &str, options: DeleteOptions) -> DiskResult<()>; + async fn verify_file(&self, volume: &str, path: &str, file_info: &rustfs_filemeta::FileInfo) -> DiskResult; + async fn check_parts(&self, volume: &str, path: &str, file_info: &rustfs_filemeta::FileInfo) -> DiskResult; + async fn read_parts(&self, bucket: &str, paths: &[String]) -> DiskResult>; + async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> DiskResult<()>; + async fn write_all(&self, volume: &str, path: &str, data: bytes::Bytes) -> DiskResult<()>; + async fn read_all(&self, volume: &str, path: &str) -> DiskResult; + async fn append_file(&self, volume: &str, path: &str) -> DiskResult; + async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: i64) -> DiskResult; +} + +impl StorageDiskRpcExt for T +where + T: ecstore_disk::DiskAPI, +{ + async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult { + ecstore_disk::DiskAPI::disk_info(self, opts).await + } + + async fn delete_volume(&self, volume: &str) -> DiskResult<()> { + ecstore_disk::DiskAPI::delete_volume(self, volume).await + } + + async fn read_multiple(&self, req: ReadMultipleReq) -> DiskResult> { + ecstore_disk::DiskAPI::read_multiple(self, req).await + } + + async fn delete_versions( + &self, + volume: &str, + versions: Vec, + opts: DeleteOptions, + ) -> Vec> { + ecstore_disk::DiskAPI::delete_versions(self, volume, versions, opts).await + } + + async fn delete_version( + &self, + volume: &str, + path: &str, + file_info: rustfs_filemeta::FileInfo, + force_del_marker: bool, + opts: DeleteOptions, + ) -> DiskResult<()> { + ecstore_disk::DiskAPI::delete_version(self, volume, path, file_info, force_del_marker, opts).await + } + + async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> DiskResult { + ecstore_disk::DiskAPI::read_xl(self, volume, path, read_data).await + } + + async fn read_version( + &self, + org_volume: &str, + volume: &str, + path: &str, + version_id: &str, + opts: &ReadOptions, + ) -> DiskResult { + ecstore_disk::DiskAPI::read_version(self, org_volume, volume, path, version_id, opts).await + } + + async fn write_metadata( + &self, + org_volume: &str, + volume: &str, + path: &str, + file_info: rustfs_filemeta::FileInfo, + ) -> DiskResult<()> { + ecstore_disk::DiskAPI::write_metadata(self, org_volume, volume, path, file_info).await + } + + async fn update_metadata( + &self, + volume: &str, + path: &str, + file_info: rustfs_filemeta::FileInfo, + opts: &UpdateMetadataOpts, + ) -> DiskResult<()> { + ecstore_disk::DiskAPI::update_metadata(self, volume, path, file_info, opts).await + } + + async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult { + ecstore_disk::DiskAPI::read_metadata(self, volume, path).await + } + + async fn delete_paths(&self, volume: &str, paths: &[String]) -> DiskResult<()> { + ecstore_disk::DiskAPI::delete_paths(self, volume, paths).await + } + + async fn stat_volume(&self, volume: &str) -> DiskResult { + ecstore_disk::DiskAPI::stat_volume(self, volume).await + } + + async fn list_volumes(&self) -> DiskResult> { + ecstore_disk::DiskAPI::list_volumes(self).await + } + + async fn make_volume(&self, volume: &str) -> DiskResult<()> { + ecstore_disk::DiskAPI::make_volume(self, volume).await + } + + async fn make_volumes(&self, volume: Vec<&str>) -> DiskResult<()> { + ecstore_disk::DiskAPI::make_volumes(self, volume).await + } + + async fn rename_data( + &self, + src_volume: &str, + src_path: &str, + file_info: rustfs_filemeta::FileInfo, + dst_volume: &str, + dst_path: &str, + ) -> DiskResult { + ecstore_disk::DiskAPI::rename_data(self, src_volume, src_path, file_info, dst_volume, dst_path).await + } + + async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult> { + ecstore_disk::DiskAPI::list_dir(self, origvolume, volume, dir_path, count).await + } + + async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> DiskResult { + ecstore_disk::DiskAPI::read_file_stream(self, volume, path, offset, length).await + } + + async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> DiskResult<()> { + ecstore_disk::DiskAPI::rename_file(self, src_volume, src_path, dst_volume, dst_path).await + } + + async fn rename_part( + &self, + src_volume: &str, + src_path: &str, + dst_volume: &str, + dst_path: &str, + meta: bytes::Bytes, + ) -> DiskResult<()> { + ecstore_disk::DiskAPI::rename_part(self, src_volume, src_path, dst_volume, dst_path, meta).await + } + + async fn delete(&self, volume: &str, path: &str, options: DeleteOptions) -> DiskResult<()> { + ecstore_disk::DiskAPI::delete(self, volume, path, options).await + } + + async fn verify_file(&self, volume: &str, path: &str, file_info: &rustfs_filemeta::FileInfo) -> DiskResult { + ecstore_disk::DiskAPI::verify_file(self, volume, path, file_info).await + } + + async fn check_parts(&self, volume: &str, path: &str, file_info: &rustfs_filemeta::FileInfo) -> DiskResult { + ecstore_disk::DiskAPI::check_parts(self, volume, path, file_info).await + } + + async fn read_parts(&self, bucket: &str, paths: &[String]) -> DiskResult> { + ecstore_disk::DiskAPI::read_parts(self, bucket, paths).await + } + + async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> DiskResult<()> { + ecstore_disk::DiskAPI::walk_dir(self, opts, wr).await + } + + async fn write_all(&self, volume: &str, path: &str, data: bytes::Bytes) -> DiskResult<()> { + ecstore_disk::DiskAPI::write_all(self, volume, path, data).await + } + + async fn read_all(&self, volume: &str, path: &str) -> DiskResult { + ecstore_disk::DiskAPI::read_all(self, volume, path).await + } + + async fn append_file(&self, volume: &str, path: &str) -> DiskResult { + ecstore_disk::DiskAPI::append_file(self, volume, path).await + } + + async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: i64) -> DiskResult { + ecstore_disk::DiskAPI::create_file(self, origvolume, volume, path, file_size).await + } +} + +pub(crate) trait StoragePeerS3ClientExt { + async fn heal_bucket( + &self, + bucket: &str, + opts: &rustfs_common::heal_channel::HealOpts, + ) -> DiskResult; + async fn make_bucket(&self, bucket: &str, opts: &rustfs_storage_api::MakeBucketOptions) -> DiskResult<()>; + async fn list_bucket(&self, opts: &rustfs_storage_api::BucketOptions) -> DiskResult>; + async fn delete_bucket(&self, bucket: &str, opts: &rustfs_storage_api::DeleteBucketOptions) -> DiskResult<()>; + async fn get_bucket_info( + &self, + bucket: &str, + opts: &rustfs_storage_api::BucketOptions, + ) -> DiskResult; +} + +impl StoragePeerS3ClientExt for LocalPeerS3Client { + async fn heal_bucket( + &self, + bucket: &str, + opts: &rustfs_common::heal_channel::HealOpts, + ) -> DiskResult { + ecstore_rpc::PeerS3Client::heal_bucket(self, bucket, opts).await + } + + async fn make_bucket(&self, bucket: &str, opts: &rustfs_storage_api::MakeBucketOptions) -> DiskResult<()> { + ecstore_rpc::PeerS3Client::make_bucket(self, bucket, opts).await + } + + async fn list_bucket(&self, opts: &rustfs_storage_api::BucketOptions) -> DiskResult> { + ecstore_rpc::PeerS3Client::list_bucket(self, opts).await + } + + async fn delete_bucket(&self, bucket: &str, opts: &rustfs_storage_api::DeleteBucketOptions) -> DiskResult<()> { + ecstore_rpc::PeerS3Client::delete_bucket(self, bucket, opts).await + } + + async fn get_bucket_info( + &self, + bucket: &str, + opts: &rustfs_storage_api::BucketOptions, + ) -> DiskResult { + ecstore_rpc::PeerS3Client::get_bucket_info(self, bucket, opts).await + } +} + pub(crate) async fn load_bucket_metadata(api: Arc, bucket: &str) -> Result { ecstore_bucket::metadata::load_bucket_metadata(api, bucket).await } diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 738da684b..5a237f32f 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -691,12 +691,7 @@ fi --glob '!**/storage_compat.rs' \ --glob '!target/**' || true ) | - perl -ne ' - next if /\buse rustfs_ecstore::api::disk::DiskAPI(?:\s+as\s+_)?;/; - next if /\buse rustfs_ecstore::api::rpc::PeerS3Client(?:\s+as\s+_)?;/; - next if /\buse rustfs_ecstore::api::tier::warm_backend::WarmBackend(?:\s+as\s+_)?;/; - print; - ' >"$DIRECT_ECSTORE_IMPORT_HITS_FILE" + cat >"$DIRECT_ECSTORE_IMPORT_HITS_FILE" if [[ -s "$DIRECT_ECSTORE_IMPORT_HITS_FILE" ]]; then report_failure "direct rustfs_ecstore imports outside compatibility boundaries are forbidden: $(paste -sd '; ' "$DIRECT_ECSTORE_IMPORT_HITS_FILE")"