mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-13 00:26:53 +00:00
refactor: clean external DTO consumers (#3566)
* refactor: clean external DTO consumers * fix: handle empty erasure shard recovery
This commit is contained in:
@@ -107,6 +107,10 @@ impl LegacyReedSolomonEncoder {
|
||||
}
|
||||
|
||||
fn reconstruct_data(&self, shards: &mut [Option<Vec<u8>>]) -> 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<Vec<u8>>]) -> 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<Vec<u8>>],
|
||||
data_shards: usize,
|
||||
parity_shards: usize,
|
||||
) -> io::Result<bool> {
|
||||
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);
|
||||
|
||||
@@ -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<Option<rustfs_ecstore::store_api::ObjectInfo>> {
|
||||
async fn get_object_meta(&self, _bucket: &str, _object: &str) -> crate::Result<Option<HealObjectInfo>> {
|
||||
Ok(None)
|
||||
}
|
||||
async fn get_object_data(&self, _bucket: &str, _object: &str) -> crate::Result<Option<Vec<u8>>> {
|
||||
|
||||
@@ -2283,7 +2283,7 @@ fn can_schedule_request(request: &HealRequest, running_per_set: &HashMap<String,
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::heal::storage::HealStorageAPI;
|
||||
use crate::heal::storage::{HealObjectInfo, HealStorageAPI};
|
||||
use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealTask, HealType};
|
||||
use rustfs_common::heal_channel::{HealOpts, HealRequestSource};
|
||||
use rustfs_ecstore::disk::{DiskStore, endpoint::Endpoint};
|
||||
@@ -2294,7 +2294,7 @@ mod tests {
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl HealStorageAPI for MockStorage {
|
||||
async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result<Option<rustfs_ecstore::store_api::ObjectInfo>> {
|
||||
async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result<Option<HealObjectInfo>> {
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
|
||||
@@ -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<Option<rustfs_ecstore::store_api::ObjectInfo>>;
|
||||
async fn get_object_meta(&self, bucket: &str, object: &str) -> Result<Option<HealObjectInfo>>;
|
||||
|
||||
/// Get object data
|
||||
async fn get_object_data(&self, bucket: &str, object: &str) -> Result<Option<Vec<u8>>>;
|
||||
@@ -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<Option<rustfs_ecstore::store_api::ObjectInfo>> {
|
||||
async fn get_object_meta(&self, bucket: &str, object: &str) -> Result<Option<HealObjectInfo>> {
|
||||
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()
|
||||
};
|
||||
|
||||
@@ -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<Option<ObjectInfo>> {
|
||||
async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result<Option<HealObjectInfo>> {
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
|
||||
@@ -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<Option<rustfs_ecstore::store_api::ObjectInfo>> {
|
||||
async fn get_object_meta(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Option<HealObjectInfo>> {
|
||||
Ok(None)
|
||||
}
|
||||
async fn get_object_data(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Option<Vec<u8>>> {
|
||||
@@ -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<Option<rustfs_ecstore::store_api::ObjectInfo>> {
|
||||
async fn get_object_meta(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Option<HealObjectInfo>> {
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
|
||||
@@ -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<String, String>, 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<String, String>,
|
||||
pub resp_elements: HashMap<String, String>,
|
||||
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<String, String>,
|
||||
resp_elements: HashMap<String, String>,
|
||||
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<String>, object: rustfs_ecstore::store_api::ObjectInfo) -> Self {
|
||||
pub fn new(event_name: EventName, bucket_name: impl Into<String>, 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 {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<rustfs_storage_api::HTTPRangeSpec>,
|
||||
) -> SwiftResult<rustfs_ecstore::store_api::GetObjectReader> {
|
||||
use rustfs_ecstore::store_api::GetObjectReader;
|
||||
|
||||
) -> SwiftResult<SwiftGetObjectReader> {
|
||||
// 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<rustfs_ecstore::store_api::ObjectInfo> {
|
||||
use rustfs_ecstore::store_api::ObjectInfo;
|
||||
|
||||
) -> SwiftResult<SwiftObjectInfo> {
|
||||
// 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()
|
||||
};
|
||||
|
||||
@@ -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};
|
||||
|
||||
@@ -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<SelectObjectContentInput>,
|
||||
@@ -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<rustfs_ecstore::store_api::ObjectInfo> {
|
||||
async fn object_info(&self, opts: &SelectObjectOptions) -> Result<SelectObjectInfo> {
|
||||
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<HTTPRangeSpec>, opts: &ObjectOptions) -> Result<GetObjectReader> {
|
||||
async fn object_reader(&self, range: Option<HTTPRangeSpec>, opts: &SelectObjectOptions) -> Result<SelectGetObjectReader> {
|
||||
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<u64>, opts: &ObjectOptions) -> Result<Bytes> {
|
||||
async fn read_raw_range_with_opts(&self, range: Range<u64>, opts: &SelectObjectOptions) -> Result<Bytes> {
|
||||
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<Bytes> {
|
||||
async fn read_header_record(&self, object_size: u64, delimiter: &[u8], opts: &SelectObjectOptions) -> Result<Bytes> {
|
||||
if object_size == 0 {
|
||||
return Ok(Bytes::new());
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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<ECStore>,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
timeout: Duration,
|
||||
) -> Option<rustfs_ecstore::store_api::ObjectInfo> {
|
||||
async fn wait_for_transition(ecstore: &Arc<ECStore>, bucket: &str, object: &str, timeout: Duration) -> Option<ObjectInfo> {
|
||||
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) => {
|
||||
|
||||
Reference in New Issue
Block a user