From acdf43937162b247619c6a32a5fe079146ca794d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Thu, 18 Jun 2026 14:22:44 +0800 Subject: [PATCH] refactor: clean external DTO consumers (#3566) * refactor: clean external DTO consumers * fix: handle empty erasure shard recovery --- crates/ecstore/src/erasure_coding/erasure.rs | 78 +++++++++++ crates/heal/src/heal/channel.rs | 8 +- crates/heal/src/heal/manager.rs | 4 +- crates/heal/src/heal/storage.rs | 14 +- crates/heal/src/heal/task.rs | 5 +- crates/heal/tests/heal_bug_fixes_test.rs | 16 +-- crates/notify/src/event.rs | 17 +-- crates/notify/src/lib.rs | 2 +- crates/protocols/src/swift/object.rs | 44 +++--- crates/protocols/src/swift/versioning.rs | 3 +- crates/s3select-api/src/object_store.rs | 20 ++- crates/scanner/src/data_usage_define.rs | 8 +- crates/scanner/src/scanner.rs | 5 +- crates/scanner/src/scanner_folder.rs | 3 +- crates/scanner/src/scanner_io.rs | 3 +- .../tests/lifecycle_integration_test.rs | 15 +-- docs/architecture/migration-progress.md | 127 ++++++++++++------ rustfs/src/app/bucket_usecase.rs | 2 +- .../src/app/lifecycle_transition_api_test.rs | 18 +-- rustfs/src/app/multipart_usecase.rs | 2 +- rustfs/src/app/object_usecase.rs | 16 ++- rustfs/src/storage/ecfs.rs | 3 +- rustfs/src/storage/ecfs_extend.rs | 3 +- rustfs/src/storage/ecfs_test.rs | 11 +- rustfs/src/storage/helper.rs | 3 +- rustfs/src/storage/mod.rs | 6 + rustfs/src/storage/options.rs | 2 +- rustfs/src/storage/s3_api/bucket.rs | 5 +- 28 files changed, 292 insertions(+), 151 deletions(-) diff --git a/crates/ecstore/src/erasure_coding/erasure.rs b/crates/ecstore/src/erasure_coding/erasure.rs index f185b6a13..21cd5131a 100644 --- a/crates/ecstore/src/erasure_coding/erasure.rs +++ b/crates/ecstore/src/erasure_coding/erasure.rs @@ -107,6 +107,10 @@ impl LegacyReedSolomonEncoder { } fn reconstruct_data(&self, shards: &mut [Option>]) -> io::Result<()> { + if recover_empty_payload_data_shards(shards, self.data_shards, self.parity_shards)? { + return Ok(()); + } + let shard_len = shards .iter() .find_map(|s| s.as_ref().map(|v| v.len())) @@ -235,6 +239,10 @@ impl ReedSolomonEncoder { /// Reconstruct missing data shards. pub fn reconstruct_data(&self, shards: &mut [Option>]) -> io::Result<()> { + if recover_empty_payload_data_shards(shards, self.data_shards, self.parity_shards)? { + return Ok(()); + } + if let Some(ref rs) = self.encoder { rs.reconstruct_data(shards) .map_err(|e| io::Error::other(format!("Reed-Solomon reconstruct failed: {e:?}"))) @@ -278,6 +286,22 @@ where } } + if shard_len == 0 { + for (index, shard) in shards.iter().enumerate() { + let shard = shard + .as_ref() + .ok_or_else(|| io::Error::other(format!("missing shard {index} after data reconstruction")))?; + if !shard.is_empty() { + return Err(io::Error::other(format!( + "inconsistent shard length at index {index}: got {}, expected {}", + shard.len(), + shard_len + ))); + } + } + return Ok(()); + } + let mut shard_refs: SmallVec<[&mut [u8]; 16]> = SmallVec::new(); for (index, shard) in shards.iter_mut().enumerate() { let shard = shard @@ -296,6 +320,39 @@ where encode(shard_refs) } +fn recover_empty_payload_data_shards( + shards: &mut [Option>], + data_shards: usize, + parity_shards: usize, +) -> io::Result { + let expected_shards = data_shards + parity_shards; + if shards.len() != expected_shards { + return Err(io::Error::other(format!( + "invalid shard count: got {}, expected {}", + shards.len(), + expected_shards + ))); + } + + let mut has_present_shard = false; + for shard in shards.iter().filter_map(|shard| shard.as_ref()) { + has_present_shard = true; + if !shard.is_empty() { + return Ok(false); + } + } + if !has_present_shard { + return Ok(false); + } + + for shard in shards.iter_mut().take(data_shards) { + if shard.is_none() { + *shard = Some(Vec::new()); + } + } + Ok(true) +} + /// Erasure coding utility for data reliability using Reed-Solomon codes. /// /// This struct provides encoding and decoding of data into data and parity shards. @@ -863,6 +920,27 @@ mod tests { } } + #[test] + fn legacy_decode_data_and_parity_reconstructs_empty_object_shards() { + let erasure = Erasure::new_with_options(3, 3, 64, true); + let encoded = erasure.encode_data(&[]).expect("empty encode should succeed"); + let mut shards = optional_shards(&encoded); + shards[1] = None; + shards[4] = None; + + erasure + .decode_data_and_parity(&mut shards) + .expect("empty decode should rebuild missing shards without SIMD"); + + for (index, shard) in shards.iter().enumerate() { + assert_eq!( + shard.as_deref(), + Some(encoded[index].as_ref()), + "empty shard {index} should match encoded source" + ); + } + } + #[test] fn test_shard_file_size_cases2() { let erasure = Erasure::new(12, 4, 1024 * 1024); diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index a06c1754e..a1cc130fc 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -488,7 +488,7 @@ impl HealChannelProcessor { mod tests { use super::*; use crate::heal::manager::HealConfig; - use crate::heal::storage::HealStorageAPI; + use crate::heal::storage::{HealObjectInfo, HealStorageAPI}; use rustfs_common::heal_channel::{ HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealRequestSource, HealScanMode, }; @@ -498,11 +498,7 @@ mod tests { struct MockStorage; #[async_trait::async_trait] impl HealStorageAPI for MockStorage { - async fn get_object_meta( - &self, - _bucket: &str, - _object: &str, - ) -> crate::Result> { + async fn get_object_meta(&self, _bucket: &str, _object: &str) -> crate::Result> { Ok(None) } async fn get_object_data(&self, _bucket: &str, _object: &str) -> crate::Result>> { diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 985965fd8..55085bfe6 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -2283,7 +2283,7 @@ fn can_schedule_request(request: &HealRequest, running_per_set: &HashMap Result> { + async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result> { Ok(None) } diff --git a/crates/heal/src/heal/storage.rs b/crates/heal/src/heal/storage.rs index b005770cd..c6a92e09c 100644 --- a/crates/heal/src/heal/storage.rs +++ b/crates/heal/src/heal/storage.rs @@ -19,7 +19,7 @@ use rustfs_ecstore::{ disk::{DiskStore, endpoint::Endpoint}, error::StorageError, store::ECStore, - store_api::ObjectOptions, + store_api::{ObjectInfo as EcstoreObjectInfo, ObjectOptions as EcstoreObjectOptions, PutObjReader as EcstorePutObjReader}, }; use rustfs_madmin::heal_commands::HealResultItem; use rustfs_storage_api::{ @@ -37,6 +37,10 @@ const EVENT_HEAL_STORAGE_OBJECT_VERIFY: &str = "heal_storage_object_verify"; const EVENT_HEAL_STORAGE_ADMIN_OP: &str = "heal_storage_admin_op"; const EVENT_HEAL_STORAGE_REPAIR_OP: &str = "heal_storage_repair_op"; +pub type HealObjectInfo = EcstoreObjectInfo; +pub type HealObjectOptions = EcstoreObjectOptions; +pub type HealPutObjReader = EcstorePutObjReader; + /// Disk status for heal operations #[derive(Debug, Clone, PartialEq, Eq)] pub enum DiskStatus { @@ -64,7 +68,7 @@ pub enum DiskStatus { #[async_trait] pub trait HealStorageAPI: Send + Sync { /// Get object meta - async fn get_object_meta(&self, bucket: &str, object: &str) -> Result>; + async fn get_object_meta(&self, bucket: &str, object: &str) -> Result>; /// Get object data async fn get_object_data(&self, bucket: &str, object: &str) -> Result>>; @@ -183,7 +187,7 @@ fn is_transient_object_exists_error(err: &StorageError) -> bool { #[async_trait] impl HealStorageAPI for ECStoreHealStorage { - async fn get_object_meta(&self, bucket: &str, object: &str) -> Result> { + async fn get_object_meta(&self, bucket: &str, object: &str) -> Result> { debug!( target: "rustfs::heal::storage", event = EVENT_HEAL_STORAGE_OBJECT_IO, @@ -330,7 +334,7 @@ impl HealStorageAPI for ECStoreHealStorage { "Heal storage request started" ); - let mut reader = rustfs_ecstore::store_api::PutObjReader::from_vec(data.to_vec()); + let mut reader = HealPutObjReader::from_vec(data.to_vec()); match (*self.ecstore) .put_object(bucket, object, &mut reader, &Default::default()) .await @@ -801,7 +805,7 @@ impl HealStorageAPI for ECStoreHealStorage { // Existence checks are best-effort for background heal scheduling, so avoid // acquiring an extra namespace read lock here. - let opts = ObjectOptions { + let opts = HealObjectOptions { no_lock: true, ..Default::default() }; diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 2b2cd7b65..1da9c10c6 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -2050,11 +2050,10 @@ impl std::fmt::Debug for HealTask { #[cfg(test)] mod tests { use super::*; - use crate::heal::storage::DiskStatus; + use crate::heal::storage::{DiskStatus, HealObjectInfo}; use rustfs_ecstore::{ data_usage::DATA_USAGE_CACHE_NAME, disk::{BUCKET_META_PREFIX, DiskStore, RUSTFS_META_BUCKET, endpoint::Endpoint}, - store_api::ObjectInfo, }; use rustfs_madmin::heal_commands::HealResultItem; use rustfs_storage_api::BucketInfo; @@ -2070,7 +2069,7 @@ mod tests { #[async_trait::async_trait] impl HealStorageAPI for MockStorage { - async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result> { + async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result> { Ok(None) } diff --git a/crates/heal/tests/heal_bug_fixes_test.rs b/crates/heal/tests/heal_bug_fixes_test.rs index 79f006b2c..1875b7d73 100644 --- a/crates/heal/tests/heal_bug_fixes_test.rs +++ b/crates/heal/tests/heal_bug_fixes_test.rs @@ -162,7 +162,7 @@ fn test_path_to_str_helper() { #[test] fn test_heal_task_status_atomic_update() { - use rustfs_heal::heal::storage::HealStorageAPI; + use rustfs_heal::heal::storage::{HealObjectInfo, HealStorageAPI}; use rustfs_heal::heal::task::{HealOptions, HealRequest, HealTask, HealTaskStatus}; use std::sync::Arc; @@ -170,11 +170,7 @@ fn test_heal_task_status_atomic_update() { struct MockStorage; #[async_trait::async_trait] impl HealStorageAPI for MockStorage { - async fn get_object_meta( - &self, - _bucket: &str, - _object: &str, - ) -> rustfs_heal::Result> { + async fn get_object_meta(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { Ok(None) } async fn get_object_data(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result>> { @@ -284,7 +280,7 @@ fn test_heal_task_status_atomic_update() { #[tokio::test] async fn test_heal_task_transient_object_exists_skip_avoids_recreate() { - use rustfs_heal::heal::storage::{DiskStatus, HealStorageAPI}; + use rustfs_heal::heal::storage::{DiskStatus, HealObjectInfo, HealStorageAPI}; use rustfs_heal::heal::task::{HealOptions, HealPriority, HealRequest, HealTask, HealTaskStatus, HealType}; use std::sync::{ Arc, @@ -298,11 +294,7 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() { #[async_trait::async_trait] impl HealStorageAPI for MockStorage { - async fn get_object_meta( - &self, - _bucket: &str, - _object: &str, - ) -> rustfs_heal::Result> { + async fn get_object_meta(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result> { Ok(None) } diff --git a/crates/notify/src/event.rs b/crates/notify/src/event.rs index 2369cceb5..527487c97 100644 --- a/crates/notify/src/event.rs +++ b/crates/notify/src/event.rs @@ -19,6 +19,8 @@ use rustfs_s3_types::{EventName, event_schema_version}; use serde::{Deserialize, Serialize}; use url::form_urlencoded; +pub type NotifyObjectInfo = rustfs_ecstore::store_api::ObjectInfo; + /// Represents the identity of the user who triggered the event #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -305,7 +307,7 @@ fn initialize_response_elements(elements: &mut HashMap, keys: &[ pub struct EventArgs { pub event_name: EventName, pub bucket_name: String, - pub object: rustfs_ecstore::store_api::ObjectInfo, + pub object: NotifyObjectInfo, pub req_params: HashMap, pub resp_elements: HashMap, pub version_id: String, @@ -354,7 +356,7 @@ impl EventArgs { pub struct EventArgsBuilder { event_name: EventName, bucket_name: String, - object: rustfs_ecstore::store_api::ObjectInfo, + object: NotifyObjectInfo, req_params: HashMap, resp_elements: HashMap, version_id: String, @@ -365,7 +367,7 @@ pub struct EventArgsBuilder { impl EventArgsBuilder { /// Creates a new builder with the required fields. - pub fn new(event_name: EventName, bucket_name: impl Into, object: rustfs_ecstore::store_api::ObjectInfo) -> Self { + pub fn new(event_name: EventName, bucket_name: impl Into, object: NotifyObjectInfo) -> Self { Self { event_name, bucket_name: bucket_name.into(), @@ -387,7 +389,7 @@ impl EventArgsBuilder { } /// Sets the object information. - pub fn object(mut self, object: rustfs_ecstore::store_api::ObjectInfo) -> Self { + pub fn object(mut self, object: NotifyObjectInfo) -> Self { self.object = object; self } @@ -482,7 +484,7 @@ mod tests { let args = EventArgsBuilder::new( EventName::LifecycleTransition, "bucket", - rustfs_ecstore::store_api::ObjectInfo { + NotifyObjectInfo { bucket: "bucket".to_string(), name: "key".to_string(), ..Default::default() @@ -498,7 +500,7 @@ mod tests { let args = EventArgsBuilder::new( EventName::ObjectRestoreCompleted, "bucket", - rustfs_ecstore::store_api::ObjectInfo { + NotifyObjectInfo { bucket: "bucket".to_string(), name: "key".to_string(), restore_expires: Some(time::OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap()), @@ -518,9 +520,8 @@ mod tests { #[cfg(test)] mod event_args_tests { - use super::EventArgs; + use super::{EventArgs, NotifyObjectInfo as ObjectInfo}; use hashbrown::HashMap; - use rustfs_ecstore::store_api::ObjectInfo; use rustfs_s3_types::EventName; fn args_with_headers(pairs: &[(&str, &str)]) -> EventArgs { diff --git a/crates/notify/src/lib.rs b/crates/notify/src/lib.rs index b1f85587c..adb72e4a4 100644 --- a/crates/notify/src/lib.rs +++ b/crates/notify/src/lib.rs @@ -40,7 +40,7 @@ mod status_view; 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}; +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, diff --git a/crates/protocols/src/swift/object.rs b/crates/protocols/src/swift/object.rs index df0cdd539..3224391f5 100644 --- a/crates/protocols/src/swift/object.rs +++ b/crates/protocols/src/swift/object.rs @@ -55,7 +55,10 @@ use super::{SwiftError, SwiftResult}; use axum::http::HeaderMap; use rustfs_credentials::Credentials; use rustfs_ecstore::resolve_object_store_handle; -use rustfs_ecstore::store_api::{ObjectOptions, PutObjReader}; +use rustfs_ecstore::store_api::{ + GetObjectReader as EcstoreGetObjectReader, ObjectInfo as EcstoreObjectInfo, ObjectOptions as EcstoreObjectOptions, + PutObjReader as EcstorePutObjReader, +}; use rustfs_rio::HashReader; use rustfs_storage_api::{BucketOperations, BucketOptions, ObjectIO as _, ObjectOperations as _}; use std::collections::HashMap; @@ -66,6 +69,11 @@ const LOG_COMPONENT_PROTOCOLS: &str = "protocols"; const LOG_SUBSYSTEM_SWIFT_OBJECT: &str = "swift_object"; const EVENT_SWIFT_OBJECT_STORAGE_STATE: &str = "swift_object_storage_state"; +pub type SwiftGetObjectReader = EcstoreGetObjectReader; +pub type SwiftObjectInfo = EcstoreObjectInfo; +pub type SwiftObjectOptions = EcstoreObjectOptions; +pub type SwiftPutObjReader = EcstorePutObjReader; + /// Maximum number of metadata headers allowed per object (Swift standard) const MAX_METADATA_COUNT: usize = 90; @@ -397,7 +405,7 @@ where })?; // 12. Prepare object options with metadata - let opts = ObjectOptions { + let opts = SwiftObjectOptions { user_defined: user_metadata, ..Default::default() }; @@ -410,7 +418,7 @@ where .map_err(|e| sanitize_storage_error("Hash reader creation", e))?; // 15. Wrap in PutObjReader as expected by storage layer - let mut put_reader = PutObjReader::new(hash_reader); + let mut put_reader = SwiftPutObjReader::new(hash_reader); // 16. Upload object to storage let obj_info = store @@ -477,7 +485,7 @@ where })?; // Prepare object options with metadata - let opts = ObjectOptions { + let opts = SwiftObjectOptions { user_defined: metadata.clone(), ..Default::default() }; @@ -493,7 +501,7 @@ where .map_err(|e| sanitize_storage_error("Hash reader creation", e))?; // Wrap in PutObjReader - let mut put_reader = PutObjReader::new(hash_reader); + let mut put_reader = SwiftPutObjReader::new(hash_reader); // Upload object to storage let obj_info = store @@ -534,9 +542,7 @@ pub async fn get_object( object: &str, credentials: &Credentials, range: Option, -) -> SwiftResult { - use rustfs_ecstore::store_api::GetObjectReader; - +) -> SwiftResult { // 1. Validate account access and get project_id let project_id = validate_account_access(account, credentials)?; @@ -556,10 +562,10 @@ pub async fn get_object( }; // 6. Prepare object options - let opts = ObjectOptions::default(); + let opts = SwiftObjectOptions::default(); // 7. Get object reader from storage with range support - let reader: GetObjectReader = store + let reader: SwiftGetObjectReader = store .get_object_reader(&bucket, &s3_key, range, HeaderMap::new(), &opts) .await .map_err(|e| { @@ -594,9 +600,7 @@ pub async fn head_object( container: &str, object: &str, credentials: &Credentials, -) -> SwiftResult { - use rustfs_ecstore::store_api::ObjectInfo; - +) -> SwiftResult { // 1. Validate account access and get project_id let project_id = validate_account_access(account, credentials)?; @@ -616,10 +620,10 @@ pub async fn head_object( }; // 6. Prepare object options - let opts = ObjectOptions::default(); + let opts = SwiftObjectOptions::default(); // 7. Get object info (metadata only) from storage - let info: ObjectInfo = store.get_object_info(&bucket, &s3_key, &opts).await.map_err(|e| { + let info: SwiftObjectInfo = store.get_object_info(&bucket, &s3_key, &opts).await.map_err(|e| { let err_str = e.to_string(); if err_str.contains("does not exist") || err_str.contains("not found") { SwiftError::NotFound(format!("Object '{}' not found in container '{}'", object, container)) @@ -674,7 +678,7 @@ pub async fn delete_object(account: &str, container: &str, object: &str, credent }; // 6. Prepare object options for deletion - let opts = ObjectOptions::default(); + let opts = SwiftObjectOptions::default(); // 7. Delete object from storage // Swift DELETE is idempotent - returns success even if object doesn't exist @@ -737,7 +741,7 @@ pub async fn update_object_metadata( }; // 6. First, get the existing object info to verify it exists - let opts = ObjectOptions::default(); + let opts = SwiftObjectOptions::default(); let existing_info = store.get_object_info(&bucket, &s3_key, &opts).await.map_err(|e| { let err_str = e.to_string(); if err_str.contains("does not exist") || err_str.contains("not found") { @@ -778,7 +782,7 @@ pub async fn update_object_metadata( // 11. Prepare options for metadata update // Swift POST replaces all custom metadata, not merges - let update_opts = ObjectOptions { + let update_opts = SwiftObjectOptions { user_defined: new_metadata, mod_time: existing_info.mod_time, version_id: existing_info.version_id.map(|v| v.to_string()), @@ -861,7 +865,7 @@ pub async fn copy_object( }; // 7. First, verify source object exists and get its info - let src_opts = ObjectOptions::default(); + let src_opts = SwiftObjectOptions::default(); let mut src_info = store .get_object_info(&src_bucket, &src_s3_key, &src_opts) .await @@ -928,7 +932,7 @@ pub async fn copy_object( validate_metadata(&new_metadata)?; // 14. Prepare destination options - let dst_opts = ObjectOptions { + let dst_opts = SwiftObjectOptions { user_defined: new_metadata, ..Default::default() }; diff --git a/crates/protocols/src/swift/versioning.rs b/crates/protocols/src/swift/versioning.rs index 7d5553fb1..85a648f0e 100644 --- a/crates/protocols/src/swift/versioning.rs +++ b/crates/protocols/src/swift/versioning.rs @@ -53,11 +53,10 @@ use super::account::validate_account_access; use super::container::ContainerMapper; -use super::object::{ObjectKeyMapper, head_object}; +use super::object::{ObjectKeyMapper, SwiftObjectOptions as ObjectOptions, head_object}; use super::{SwiftError, SwiftResult}; use rustfs_credentials::Credentials; use rustfs_ecstore::resolve_object_store_handle; -use rustfs_ecstore::store_api::ObjectOptions; use rustfs_storage_api::{ListOperations as _, ObjectOperations as _}; use std::time::{SystemTime, UNIX_EPOCH}; use tracing::{debug, error}; diff --git a/crates/s3select-api/src/object_store.rs b/crates/s3select-api/src/object_store.rs index e2366427c..f01836475 100644 --- a/crates/s3select-api/src/object_store.rs +++ b/crates/s3select-api/src/object_store.rs @@ -29,7 +29,9 @@ use rustfs_ecstore::error::{StorageError, is_err_bucket_not_found, is_err_object use rustfs_ecstore::resolve_object_store_handle; use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; use rustfs_ecstore::store::ECStore; -use rustfs_ecstore::store_api::{GetObjectReader, ObjectOptions}; +use rustfs_ecstore::store_api::{ + GetObjectReader as EcstoreGetObjectReader, ObjectInfo as EcstoreObjectInfo, ObjectOptions as EcstoreObjectOptions, +}; use rustfs_storage_api::{HTTPRangeSpec, ObjectIO as _, ObjectOperations as _}; use s3s::S3Result; use s3s::dto::SelectObjectContentInput; @@ -66,6 +68,10 @@ pub const MAX_JSON_DOCUMENT_BYTES: u64 = 128 * 1024 * 1024; pub const INVALID_SCAN_RANGE_MESSAGE: &str = "The value of a parameter in ScanRange element is invalid. Check the service API documentation and try again."; +type SelectGetObjectReader = EcstoreGetObjectReader; +type SelectObjectInfo = EcstoreObjectInfo; +type SelectObjectOptions = EcstoreObjectOptions; + #[derive(Debug)] pub struct EcObjectStore { input: Arc, @@ -155,8 +161,8 @@ impl EcObjectStore { }) } - fn object_options(&self, options: &GetOptions) -> ObjectOptions { - ObjectOptions { + fn object_options(&self, options: &GetOptions) -> SelectObjectOptions { + SelectObjectOptions { version_id: options.version.clone(), ..Default::default() } @@ -194,14 +200,14 @@ impl EcObjectStore { .is_some_and(|info| matches!(info.as_str(), "USE" | "IGNORE")) } - async fn object_info(&self, opts: &ObjectOptions) -> Result { + async fn object_info(&self, opts: &SelectObjectOptions) -> Result { self.store .get_object_info(&self.input.bucket, &self.input.key, opts) .await .map_err(|err| map_storage_error(&self.input.bucket, &self.input.key, err)) } - async fn object_reader(&self, range: Option, opts: &ObjectOptions) -> Result { + async fn object_reader(&self, range: Option, opts: &SelectObjectOptions) -> Result { let h = self.read_headers(); self.store .get_object_reader(&self.input.bucket, &self.input.key, range, h, opts) @@ -209,7 +215,7 @@ impl EcObjectStore { .map_err(|err| map_storage_error(&self.input.bucket, &self.input.key, err)) } - async fn read_raw_range_with_opts(&self, range: Range, opts: &ObjectOptions) -> Result { + async fn read_raw_range_with_opts(&self, range: Range, opts: &SelectObjectOptions) -> Result { if range.is_empty() { return Ok(Bytes::new()); } @@ -228,7 +234,7 @@ impl EcObjectStore { .await } - async fn read_header_record(&self, object_size: u64, delimiter: &[u8], opts: &ObjectOptions) -> Result { + async fn read_header_record(&self, object_size: u64, delimiter: &[u8], opts: &SelectObjectOptions) -> Result { if object_size == 0 { return Ok(Bytes::new()); } diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 2cfd002fe..f1ddccfd3 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -33,7 +33,7 @@ use rustfs_ecstore::{ config::{com::save_config, storageclass}, disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}, error::{Error, Result as StorageResult, StorageError}, - store_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}, + store_api::{GetObjectReader, ObjectInfo, ObjectOptions, ObjectToDelete, PutObjReader}, }; use rustfs_storage_api::{HTTPRangeSpec, ObjectIO}; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; @@ -61,6 +61,12 @@ const EVENT_SCANNER_CACHE_LOAD_STATE: &str = "scanner_cache_load_state"; const EVENT_SCANNER_CACHE_SAVE_STATE: &str = "scanner_cache_save_state"; static CACHE_SAVE_METRICS_ONCE: Once = Once::new(); +pub type ScannerGetObjectReader = GetObjectReader; +pub type ScannerObjectInfo = ObjectInfo; +pub type ScannerObjectOptions = ObjectOptions; +pub type ScannerObjectToDelete = ObjectToDelete; +pub type ScannerPutObjReader = PutObjReader; + pub trait ScannerObjectIO: ObjectIO< Error = Error, diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 496df5ea1..3b300ca8c 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1070,7 +1070,10 @@ pub async fn store_data_usage_in_backend( #[cfg(test)] mod tests { use super::*; - use rustfs_ecstore::store_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; + use crate::{ + ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, + ScannerPutObjReader as PutObjReader, + }; use serial_test::serial; use std::collections::HashMap; use std::io::Cursor; diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 39b631812..553121cb5 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -57,7 +57,6 @@ use rustfs_ecstore::disk::{Disk, DiskAPI as _, DiskInfoOptions}; use rustfs_ecstore::error::StorageError; use rustfs_ecstore::global::is_erasure; use rustfs_ecstore::pools::{path2_bucket_object, path2_bucket_object_with_base_path}; -use rustfs_ecstore::store_api::{ObjectInfo, ObjectToDelete}; use rustfs_ecstore::store_utils::is_reserved_or_invalid_bucket; use rustfs_filemeta::{ MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicateObjectInfo, ReplicationStatusType, ReplicationType, @@ -70,6 +69,8 @@ use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use tracing::{debug, error, warn}; +use crate::{ScannerObjectInfo as ObjectInfo, ScannerObjectToDelete as ObjectToDelete}; + const LOG_COMPONENT_SCANNER: &str = "scanner"; const LOG_SUBSYSTEM_FOLDER: &str = "folder"; const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index dfb619bfe..92dc57bf8 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -41,7 +41,6 @@ use rustfs_ecstore::error::{Error, StorageError}; use rustfs_ecstore::global::GLOBAL_TierConfigMgr; use rustfs_ecstore::resolve_object_store_handle; use rustfs_ecstore::set_disk::SetDisks; -use rustfs_ecstore::store_api::ObjectInfo; use rustfs_ecstore::{error::Result, store::ECStore}; use rustfs_filemeta::FileMeta; use rustfs_storage_api::{BucketInfo, BucketOperations, BucketOptions, DiskSetSelector, StorageAdminApi}; @@ -59,6 +58,8 @@ use tokio::time::Duration; use tokio_util::sync::CancellationToken; use tracing::{debug, error, warn}; +use crate::ScannerObjectInfo as ObjectInfo; + pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file"; const LOG_COMPONENT_SCANNER: &str = "scanner"; const LOG_SUBSYSTEM_IO: &str = "io"; diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index f08a4ad35..f7ed8afd2 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/crates/scanner/tests/lifecycle_integration_test.rs @@ -28,16 +28,18 @@ use rustfs_ecstore::{ global::GLOBAL_TierConfigMgr, pools::path2_bucket_object_with_base_path, store::ECStore, - store_api::{ObjectOptions, PutObjReader}, tier::{ tier_config::{TierConfig, TierMinIO, TierType}, warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options}, }, }; use rustfs_filemeta::FileMeta; -use rustfs_scanner::scanner::init_data_scanner; use rustfs_scanner::scanner_folder::ScannerItem; use rustfs_scanner::scanner_io::ScannerIODisk; +use rustfs_scanner::{ + ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader, + scanner::init_data_scanner, +}; use rustfs_storage_api::{ BucketOperations, ListOperations as _, MakeBucketOptions, MultipartOperations as _, ObjectIO as _, ObjectOperations as _, }; @@ -784,12 +786,7 @@ async fn register_mock_tier(tier_name: &str) -> MockWarmBackend { backend } -async fn wait_for_transition( - ecstore: &Arc, - bucket: &str, - object: &str, - timeout: Duration, -) -> Option { +async fn wait_for_transition(ecstore: &Arc, bucket: &str, object: &str, timeout: Duration) -> Option { let deadline = tokio::time::Instant::now() + timeout; loop { @@ -900,7 +897,7 @@ mod serial_tests { println!("✅ Object was transitioned by lifecycle processing"); // Let's try to get object info to see its details match ecstore - .get_object_info(bucket_name.as_str(), object_name, &rustfs_ecstore::store_api::ObjectOptions::default()) + .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default()) .await { Ok(obj_info) => { diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 6cf3faddc..d81a84188 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-operation-bounds-cleanup` -- Baseline: `main` at `a30cafa73fde3e7e88c160e70c290e74a0c2235f` - after `rustfs/rustfs#3563` merged. +- Branch: `overtrue/arch-storage-dto-consumer-boundaries` +- Baseline: `main` at `99941f7e7c5e0c88532a93cc175a3ea4111d7098` + after `rustfs/rustfs#3565` merged. - PR type for this branch: `consumer-migration` -- Runtime behavior changes: no external behavior change expected. -- Rust code changes: migrate scanner cache persistence, RustFS object - namespace-lock helper, and table catalog storage bounds away from ECStore - compatibility operation traits to `rustfs-storage-api` operation traits with - ECStore concrete associated-type bindings. ECStore-owned `ObjectInfo`, - `ObjectOptions`, readers, delete DTOs, walk filters, lock wrappers, and - implementation behavior stay in ECStore. -- CI/script changes: extend migration guards so outer consumers do not drift - back to ECStore operation trait imports. -- Docs changes: record the larger operation consumer-bound cleanup slice. +- Runtime behavior changes: no migration behavior change expected; CI follow-up + preserves empty-object erasure recovery by avoiding zero-byte SIMD decode. +- Rust code changes: add crate-local semantic aliases for ECStore-owned object + metadata, options, readers, and delete DTOs in scanner, heal, notify, Swift, + S3 Select, and RustFS storage/app consumers so external call sites stop + importing raw `rustfs_ecstore::store_api` DTOs outside deliberate boundary + files. ECStore-owned DTO definitions and runtime behavior stay in ECStore. + CI follow-up handles zero-length shards in erasure reconstruction without + changing non-empty shard behavior. +- CI/script changes: none. +- Docs changes: record the larger DTO consumer-boundary cleanup slice and the + empty-object erasure recovery CI follow-up. ## Phase 0 Tasks @@ -775,6 +777,29 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block guards, formatting, diff hygiene, Rust risk scan, full pre-commit, and required three-expert review passed. +- [x] `API-026` Clean external DTO consumer boundaries. + - Current branch: `overtrue/arch-storage-dto-consumer-boundaries`. + - Completed slice: introduce crate-local semantic aliases for ECStore-owned + object metadata/options/readers/delete DTOs in scanner, heal, notify, Swift, + S3 Select, and RustFS storage/app consumers; update production and affected + test call sites to use those local aliases instead of raw + `rustfs_ecstore::store_api` DTO imports. + - Acceptance: non-ECStore direct `rustfs_ecstore::store_api` references are + limited to boundary alias definitions, ECStore remains the owner of + `ObjectInfo`, `ObjectOptions`, object readers, delete DTOs, walk filters, + lock wrappers, and implementation behavior, and external consumers express + their local semantic dependency through crate-owned names. + - Must preserve: object metadata shape, object option defaults, reader/writer + behavior, delete replication DTO handling, scanner cache semantics, heal + storage metadata semantics, Swift and S3 Select object reads, notification + event payloads, S3 response DTO mapping, and storage/app test behavior. + - Risk defense: this slice uses type aliases and import-boundary cleanup only; + it does not move DTO definitions, alter serialization, change object-store + implementations, or adjust runtime control flow. + - Verification: focused compile/tests, migration/layer guards, formatting, + diff hygiene, direct import scan, Rust risk scan, full pre-commit, and + required three-expert review passed. + ## Phase 8 Background Controller Tasks - [x] `BGC-001` Inventory background services. @@ -1051,46 +1076,70 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | Outer scanner/RustFS operation consumers now use storage-api operation traits with explicit ECStore concrete associated-type bindings; ECStore keeps compatibility traits only for implementation and downstream compatibility. | -| Migration preservation | passed | Scanner cache persistence, object self-copy namespace locking, and table catalog object storage keep the same ECStore DTOs, readers, lock wrappers, walk filter shape, and storage error conversion behavior. | -| Testing/verification | passed | Focused RustFS/scanner compile/tests, migration/layer guards, formatting, diff hygiene, Rust risk scan, and full `make pre-commit` passed. | +| Quality/architecture | passed | Scanner, heal, notify, Swift, S3 Select, and RustFS storage/app consumers now use crate-local DTO aliases; the erasure CI fix is isolated to zero-length shard reconstruction. | +| Migration preservation | passed | ECStore remains the owner of object DTOs/readers/walk filters/implementation behavior; aliases preserve concrete types and non-empty erasure encode/decode paths still use the existing encoders. | +| Testing/verification | passed | Focused DTO/erasure compile/tests, migration/layer guards, direct import scan, formatting, diff hygiene, Rust risk scan, and full `make pre-commit` passed. | ## Verification Notes Passed before push: -- `cargo check --tests -p rustfs -p rustfs-scanner`: passed. -- `cargo test -p rustfs-scanner`: passed; 151 unit tests passed, lifecycle - focused test passed, 14 lifecycle integration tests ignored by default. -- `cargo test -p rustfs --lib table_catalog`: passed; 168 passed. -- `cargo test -p rustfs --lib app::object_usecase`: passed; 86 passed, 2 - ignored. +- `cargo check --tests -p rustfs -p rustfs-scanner -p rustfs-heal -p + rustfs-protocols -p rustfs-notify -p rustfs-s3select-api`: passed. +- `cargo test -p rustfs-scanner -p rustfs-heal -p rustfs-notify -p + rustfs-protocols -p rustfs-s3select-api`: passed. +- `cargo test -p rustfs --lib app::object_usecase::tests`: passed; 85 passed, + 2 ignored. +- `cargo test -p rustfs --lib app::bucket_usecase::tests`: passed; 64 passed. +- `cargo test -p rustfs --lib app::multipart_usecase::tests`: passed; 24 + passed, 4 ignored. +- `cargo test -p rustfs --lib storage::s3_api::bucket::tests`: passed; 22 + passed. +- `cargo test -p rustfs --lib storage::ecfs_test::tests`: passed; 59 passed, + 14 ignored. +- `rg -n 'rustfs_ecstore::store_api' rustfs/src crates --glob + '!crates/ecstore/**' --glob '*.rs'`: remaining matches are deliberate + boundary alias definitions. +- `cargo test -p rustfs-ecstore erasure_coding::erasure::tests`: passed; 31 + passed. +- `PROPTEST_CASES=1024 cargo test -p rustfs-ecstore + erasure_coding::erasure::tests::decode_data_and_parity_round_trips_bounded_recoverability`: + passed. +- `cargo test -p rustfs-ecstore + rpc::peer_s3_client::tests::local_get_bucket_info_survives_prior_walk_timeout`: + passed after a full-crate `cargo test -p rustfs-ecstore` run exposed this + unrelated ordinary-harness global-state interference. - `./scripts/check_architecture_migration_rules.sh`: passed. - `./scripts/check_layer_dependencies.sh`: passed. - `cargo fmt --all --check`: passed. - `git diff --check`: passed. -- Rust risk scan: no new `unwrap`/`expect`, panic/todo markers, `unsafe`, - process-spawning calls, lossy casts, println/eprintln, or relaxed ordering in +- Rust risk scan: new hits are crate-local alias casts plus the empty-erasure + test expectations; no new production `unwrap`/`expect`, panic/todo markers, + `unsafe`, process-spawning calls, println/eprintln, or relaxed ordering in added Rust lines. -- `make pre-commit`: passed; nextest reported 6207 tests passed and 111 +- `make pre-commit`: passed; nextest reported 6218 tests passed and 111 skipped, and doctests passed. Notes: -- This slice follows the shared list operation consumer cleanup and keeps the - old aggregate facade, DTO/helper, response, and operation contract guards - active. -- Outer scanner/RustFS operation consumers now bind operation traits through - `rustfs-storage-api`; ECStore keeps concrete object metadata, options, - readers, delete DTOs, lock wrappers, filemeta-bound walk filters, and - implementation behavior. -- The slice does not alter scanner cache load/save behavior, object self-copy - lock behavior, table catalog object operations, list/walk implementation, - object reader/writer behavior, or storage error runtime behavior. +- This slice follows the external operation consumer cleanup and keeps the old + aggregate facade, DTO/helper, response, and operation contract guards active. +- External DTO consumers now name ECStore-owned object DTOs through local + semantic aliases in their owning crates or RustFS storage boundary module. +- The slice does not alter object metadata shape, options defaults, + reader/writer behavior, delete replication DTO handling, scanner cache + semantics, heal storage metadata semantics, Swift/S3 Select object reads, + notification event payloads, S3 response DTO mapping, or runtime storage + behavior. +- The CI follow-up preserves empty-object erasure reconstruction by filling + missing zero-length shards before calling encoders that reject zero-byte SIMD + shard sizes. ## Handoff Notes -- External operation consumer cleanup is based on `main` after - `rustfs/rustfs#3563` merged. -- Continue with larger consumer-migration batches; avoid moving ECStore-owned - DTOs/readers/walk filters until their concrete behavior is isolated. +- External DTO consumer cleanup is based on `main` after + `rustfs/rustfs#3565` merged and carries the empty-object erasure recovery + CI follow-up because the merged PR did not include that post-merge fix. +- Continue with larger consumer-migration batches; keep ECStore-owned + DTOs/readers/walk filters in ECStore until their concrete behavior is isolated + enough for a pure-move slice. diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index ee71ce235..347f3e403 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -23,6 +23,7 @@ use crate::app::context::{ use crate::auth::get_condition_values_with_client_info; use crate::error::ApiError; use crate::server::RemoteAddr; +use crate::storage::StorageObjectInfo as ObjectInfo; use crate::storage::access::{ReqInfo, authorize_request, req_info_ref}; use crate::storage::helper::{OperationHelper, spawn_background_with_context}; use crate::storage::s3_api::bucket::{ @@ -57,7 +58,6 @@ use rustfs_ecstore::client::object_api_utils::to_s3s_etag; use rustfs_ecstore::error::StorageError; use rustfs_ecstore::notification_sys::get_global_notification_sys; use rustfs_ecstore::store::ECStore; -use rustfs_ecstore::store_api::ObjectInfo; use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta}; use rustfs_policy::policy::{ action::{Action, S3Action}, diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index d1546e660..314bc7e40 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -15,6 +15,9 @@ use super::{multipart_usecase::DefaultMultipartUsecase, object_usecase::DefaultObjectUsecase}; use crate::app::bucket_usecase::DefaultBucketUsecase; use crate::storage::ecfs::FS; +use crate::storage::{ + StorageObjectInfo as ObjectInfo, StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader, +}; use bytes::Bytes; use futures::FutureExt; use futures::stream; @@ -29,7 +32,6 @@ use rustfs_ecstore::{ endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}, global::GLOBAL_TierConfigMgr, store::ECStore, - store_api::{ObjectOptions, PutObjReader}, tier::{ tier_config::{TierConfig, TierType}, warm_backend::{WarmBackend, WarmBackendGetOpts}, @@ -148,12 +150,7 @@ async fn create_test_bucket(ecstore: &Arc, bucket_name: &str) { .expect("Failed to create test bucket"); } -async fn upload_test_object( - ecstore: &Arc, - bucket: &str, - object: &str, - data: &[u8], -) -> rustfs_ecstore::store_api::ObjectInfo { +async fn upload_test_object(ecstore: &Arc, bucket: &str, object: &str, data: &[u8]) -> ObjectInfo { let mut reader = PutObjReader::from_vec(data.to_vec()); (**ecstore) .put_object(bucket, object, &mut reader, &ObjectOptions::default()) @@ -362,12 +359,7 @@ async fn register_mock_tier(tier_name: &str) -> MockWarmBackend { backend } -async fn wait_for_transition( - ecstore: &Arc, - bucket: &str, - object: &str, - timeout: Duration, -) -> Option { +async fn wait_for_transition(ecstore: &Arc, bucket: &str, object: &str, timeout: Duration) -> Option { let deadline = tokio::time::Instant::now() + timeout; loop { diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 1b850ebc0..5b444c75f 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -33,6 +33,7 @@ use crate::storage::sse::{ extract_ssekms_context_from_headers, map_get_object_reader_error, mark_encrypted_multipart_metadata, }; use crate::storage::*; +use crate::storage::{StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader}; use crate::table_catalog; use bytes::Bytes; use futures::StreamExt; @@ -53,7 +54,6 @@ use rustfs_ecstore::rio::{DecryptReader, EncryptReader, HardLimitReader, boxed_r 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::{ObjectOptions, PutObjReader}; use rustfs_filemeta::{ReplicationStatusType, ReplicationType}; use rustfs_s3_ops::S3Operation; #[cfg(test)] diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index be352d3a6..9b8c43c7e 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -80,7 +80,6 @@ use rustfs_ecstore::error::{ 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::{ObjectInfo, ObjectOptions, ObjectToDelete, PutObjReader}; use rustfs_filemeta::{ REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateTargetDecision, ReplicationState, ReplicationStatusType, ReplicationType, RestoreStatusOps, VersionPurgeStatusType, parse_restore_obj_status, replication_statuses_map, @@ -137,6 +136,11 @@ use tokio_util::io::{ReaderStream, StreamReader}; use tracing::{debug, error, instrument, warn}; use uuid::Uuid; +use crate::storage::{ + StorageDeletedObject, StorageObjectInfo as ObjectInfo, StorageObjectOptions as ObjectOptions, + StorageObjectToDelete as ObjectToDelete, StoragePutObjReader as PutObjReader, +}; + const ACCEPT_RANGES_BYTES: &str = "bytes"; const MAX_GET_OBJECT_MEMORY_BUFFER_BYTES: i64 = 64 * 1024 * 1024; const LOG_COMPONENT_APP: &str = "app"; @@ -778,7 +782,7 @@ fn delete_replication_state_from_config( async fn enrich_delete_replication_state_if_needed( bucket: &str, - delete_object: &mut rustfs_ecstore::store_api::DeletedObject, + delete_object: &mut StorageDeletedObject, obj_info: &ObjectInfo, ) { let Some(replication_state) = delete_object.replication_state.as_ref() else { @@ -3380,7 +3384,7 @@ impl DefaultObjectUsecase { #[derive(Default, Clone)] struct DeleteResult { - delete_object: Option, + delete_object: Option, error: Option, } @@ -3823,7 +3827,7 @@ impl DefaultObjectUsecase { if obj_info.name.is_empty() { if replicate_force_delete { schedule_replication_delete(DeletedObjectReplicationInfo { - delete_object: rustfs_ecstore::store_api::DeletedObject { + delete_object: StorageDeletedObject { object_name: key.clone(), force_delete: true, ..Default::default() @@ -3871,7 +3875,7 @@ impl DefaultObjectUsecase { if schedule_delete_replication { let _activity_guard = DeleteTailActivityGuard::new(DeleteTailStage::Replication); let mut deleted_object = DeletedObjectReplicationInfo { - delete_object: rustfs_ecstore::store_api::DeletedObject { + delete_object: StorageDeletedObject { delete_marker: deleted_object_source.delete_marker && !deleted_delete_marker_version, delete_marker_version_id: if deleted_object_source.delete_marker { deleted_object_source.version_id @@ -6309,7 +6313,7 @@ mod tests { #[test] fn replica_delete_enrichment_must_not_reuse_upstream_targets() { - let delete_object = rustfs_ecstore::store_api::DeletedObject { + let delete_object = StorageDeletedObject { replication_state: Some(ReplicationState { replicate_decision_str: "arn:aws:s3:::upstream=true;false;arn:aws:s3:::upstream;".to_string(), replication_status_internal: Some("arn:aws:s3:::upstream=COMPLETED;".to_string()), diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index f511cc1fc..5b37052ec 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -40,7 +40,6 @@ use rustfs_ecstore::{ versioning_sys::BucketVersioningSys, }, error::{StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found}, - store_api::ObjectOptions, }; use rustfs_io_metrics::record_s3_op; use rustfs_s3_ops::S3Operation; @@ -60,6 +59,8 @@ const LOG_SUBSYSTEM_OBJECT: &str = "object"; const LOG_SUBSYSTEM_OBJECT_LOCK: &str = "object_lock"; const LOG_SUBSYSTEM_TAGGING: &str = "tagging"; +use crate::storage::StorageObjectOptions as ObjectOptions; + #[derive(Debug, Clone)] pub struct FS { // pub store: ECStore, diff --git a/rustfs/src/storage/ecfs_extend.rs b/rustfs/src/storage/ecfs_extend.rs index 8480ac449..ef46ed273 100644 --- a/rustfs/src/storage/ecfs_extend.rs +++ b/rustfs/src/storage/ecfs_extend.rs @@ -25,7 +25,6 @@ use rustfs_ecstore::bucket::object_lock::objectlock_sys; use rustfs_ecstore::bucket::replication::ReplicationConfigurationExt; use rustfs_ecstore::error::StorageError; use rustfs_ecstore::resolve_object_store_handle; -use rustfs_ecstore::store_api::{ObjectInfo, ObjectToDelete}; use rustfs_storage_api::{BucketOperations, BucketOptions}; use rustfs_targets::EventName; use rustfs_targets::arn::{TargetID, TargetIDError}; @@ -48,6 +47,8 @@ use time::format_description::well_known::Rfc3339; use time::{format_description::FormatItem, macros::format_description}; use tracing::{debug, warn}; +use crate::storage::{StorageObjectInfo as ObjectInfo, StorageObjectToDelete as ObjectToDelete}; + const LOG_COMPONENT_STORAGE: &str = "storage"; const LOG_SUBSYSTEM_OBJECT_LOCK: &str = "object_lock"; const LOG_SUBSYSTEM_OBJECT_KEY: &str = "object_key"; diff --git a/rustfs/src/storage/ecfs_test.rs b/rustfs/src/storage/ecfs_test.rs index 10c2718cd..82b2131c8 100644 --- a/rustfs/src/storage/ecfs_test.rs +++ b/rustfs/src/storage/ecfs_test.rs @@ -19,17 +19,16 @@ mod tests { use crate::storage::ecfs::{FS, validate_object_lock_configuration_input}; use crate::storage::s3_api::common::{rustfs_initiator, rustfs_owner}; use crate::storage::{ - apply_cors_headers, apply_default_lock_retention_metadata, check_preconditions, get_adaptive_buffer_size_with_profile, - get_buffer_size_opt_in, is_etag_equal, matches_origin_pattern, parse_etag, parse_object_lock_legal_hold, - parse_object_lock_retention, process_lambda_configurations, process_queue_configurations, process_topic_configurations, - remove_object_lock_metadata_for_copy, remove_object_lock_retention_metadata, validate_bucket_object_lock_enabled, - validate_list_object_unordered_with_delimiter, + StorageObjectInfo as ObjectInfo, apply_cors_headers, apply_default_lock_retention_metadata, check_preconditions, + get_adaptive_buffer_size_with_profile, get_buffer_size_opt_in, is_etag_equal, matches_origin_pattern, parse_etag, + parse_object_lock_legal_hold, parse_object_lock_retention, process_lambda_configurations, process_queue_configurations, + process_topic_configurations, remove_object_lock_metadata_for_copy, remove_object_lock_retention_metadata, + validate_bucket_object_lock_enabled, validate_list_object_unordered_with_delimiter, }; use http::{Extensions, HeaderMap, HeaderValue, Method, StatusCode, Uri}; use rustfs_config::MI_B; use rustfs_ecstore::bucket::{metadata::BucketMetadata, metadata_sys}; use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; - use rustfs_ecstore::store_api::ObjectInfo; use rustfs_utils::http::{ AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, SUFFIX_OBJECTLOCK_LEGALHOLD_TIMESTAMP, SUFFIX_OBJECTLOCK_RETENTION_TIMESTAMP, contains_key_str, get_str, insert_str, diff --git a/rustfs/src/storage/helper.rs b/rustfs/src/storage/helper.rs index 995d7e590..5609ca052 100644 --- a/rustfs/src/storage/helper.rs +++ b/rustfs/src/storage/helper.rs @@ -22,7 +22,6 @@ use rustfs_audit::{ entity::{ApiDetails, ApiDetailsBuilder, AuditEntryBuilder}, global::AuditLogger, }; -use rustfs_ecstore::store_api::ObjectInfo; use rustfs_io_metrics::record_s3_op; use rustfs_notify::{EventArgsBuilder, notifier_global}; use rustfs_s3_ops::{S3Operation, operation_matches_event_name}; @@ -37,6 +36,8 @@ use std::future::Future; use tokio::runtime::{Builder, Handle}; use tracing::{Instrument, info_span, warn}; +use crate::storage::StorageObjectInfo as ObjectInfo; + /// Schedules an asynchronous task on the current runtime; /// if there is no runtime, creates a minimal runtime execution on a new thread. pub(crate) fn spawn_background(fut: F) diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index 2263ed81a..254c3b74c 100644 --- a/rustfs/src/storage/mod.rs +++ b/rustfs/src/storage/mod.rs @@ -27,6 +27,12 @@ pub(crate) mod sse; pub mod timeout_wrapper; pub mod tonic_service; +pub(crate) type StorageDeletedObject = rustfs_ecstore::store_api::DeletedObject; +pub(crate) type StorageObjectInfo = rustfs_ecstore::store_api::ObjectInfo; +pub(crate) type StorageObjectOptions = rustfs_ecstore::store_api::ObjectOptions; +pub(crate) type StorageObjectToDelete = rustfs_ecstore::store_api::ObjectToDelete; +pub(crate) type StoragePutObjReader = rustfs_ecstore::store_api::PutObjReader; + #[cfg(test)] mod concurrent_fix_test; #[cfg(test)] diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index eaecd0601..57a025fcf 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -31,7 +31,6 @@ use s3s::header::X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE; use crate::auth::UNSIGNED_PAYLOAD; use crate::auth::UNSIGNED_PAYLOAD_TRAILER; -use rustfs_ecstore::store_api::ObjectOptions; use rustfs_policy::service_type::ServiceType; use rustfs_storage_api::{HTTPPreconditions, HTTPRangeSpec}; use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH; @@ -47,6 +46,7 @@ use crate::auth::AuthType; use crate::auth::get_query_param; use crate::auth::get_request_auth_type_with_query; use crate::auth::is_request_presigned_signature_v4_with_query; +use crate::storage::StorageObjectOptions as ObjectOptions; #[cfg(test)] use rustfs_utils::http::insert_header; diff --git a/rustfs/src/storage/s3_api/bucket.rs b/rustfs/src/storage/s3_api/bucket.rs index 4cf1dee94..e51b2961f 100644 --- a/rustfs/src/storage/s3_api/bucket.rs +++ b/rustfs/src/storage/s3_api/bucket.rs @@ -15,7 +15,6 @@ use crate::storage::s3_api::common::rustfs_owner; use percent_encoding::percent_decode_str; use rustfs_ecstore::client::object_api_utils::to_s3s_etag; -use rustfs_ecstore::store_api::ObjectInfo; use rustfs_storage_api::{ BucketInfo, ListObjectVersionsInfo as StorageListObjectVersionsInfo, ListObjectsV2Info as StorageListObjectsV2Info, }; @@ -27,6 +26,8 @@ use s3s::{S3Error, S3ErrorCode}; use tracing::debug; use urlencoding::encode; +use crate::storage::StorageObjectInfo as ObjectInfo; + const S3_MAX_KEYS: i32 = 1000; type ListObjectVersionsInfo = StorageListObjectVersionsInfo; @@ -393,8 +394,8 @@ mod tests { build_list_object_versions_output, build_list_objects_output, build_list_objects_v2_output, parse_list_object_versions_params, parse_list_objects_v2_params, }; + use crate::storage::StorageObjectInfo as ObjectInfo; use crate::storage::s3_api::common::rustfs_owner; - use rustfs_ecstore::store_api::ObjectInfo; use rustfs_storage_api::BucketInfo; use s3s::S3ErrorCode; use s3s::dto::{CommonPrefix, EncodingType, ListObjectsV2Output, Object};