gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-06 13:27:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fmt and improve import

Browse Source
This commit is contained in:
houseme
2026-06-28 11:41:11 +08:00
parent 4c83669c30
commit 143491e0af
18 changed files with 269 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+22 -5
View File
@@ -617,7 +617,10 @@ impl ReplicationResyncer {
} else {
let state = TargetReplicationResyncStatus::new();
bucket_status.targets_map.insert(opts.arn.clone(), state);
bucket_status.targets_map.get_mut(&opts.arn).expect("ARN should be in targets map")
bucket_status
.targets_map
.get_mut(&opts.arn)
.expect("ARN should be in targets map")
};
if !resync_state_accepts_update(state, &opts) {
@@ -678,7 +681,10 @@ impl ReplicationResyncer {
} else {
let state = TargetReplicationResyncStatus::new();
bucket_status.targets_map.insert(opts.arn.clone(), state);
bucket_status.targets_map.get_mut(&opts.arn).expect("ARN should be in targets map")
bucket_status
.targets_map
.get_mut(&opts.arn)
.expect("ARN should be in targets map")
};
if !resync_state_accepts_update(state, &opts) {
@@ -2710,7 +2716,13 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
&& !tgt_client.reset_id.is_empty()
&& dobj.op_type == ReplicationType::ExistingObject
{
rinfo.resync_timestamp = format!("{};{}", OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_else(|_| "invalid-time".to_string()), tgt_client.reset_id);
rinfo.resync_timestamp = format!(
"{};{}",
OffsetDateTime::now_utc()
.format(&Rfc3339)
.unwrap_or_else(|_| "invalid-time".to_string()),
tgt_client.reset_id
);
}
rinfo
@@ -3473,8 +3485,13 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
&& self.op_type == ReplicationType::ExistingObject
&& !tgt_client.reset_id.is_empty()
{
rinfo.resync_timestamp =
format!("{};{}", OffsetDateTime::now_utc().format(&Rfc3339).unwrap_or_else(|_| "invalid-time".to_string()), tgt_client.reset_id);
rinfo.resync_timestamp = format!(
"{};{}",
OffsetDateTime::now_utc()
.format(&Rfc3339)
.unwrap_or_else(|_| "invalid-time".to_string()),
tgt_client.reset_id
);
rinfo.replication_resynced = true;
}
crates/ecstore/src/client/api_get_object_attributes.rs
+16 -11
View File
@@ -133,7 +133,8 @@ struct ObjectAttributePart {
impl ObjectAttributes {
pub async fn parse_response(&mut self, h: &HeaderMap, body_vec: Vec<u8>) -> Result<(), std::io::Error> {
let last_modified = h.get("Last-Modified")
let last_modified = h
.get("Last-Modified")
.ok_or_else(|| std::io::Error::other("missing Last-Modified header"))?
.to_str()
.map_err(|e| std::io::Error::other(format!("invalid Last-Modified header: {e}")))?;
@@ -141,14 +142,14 @@ impl ObjectAttributes {
.map_err(|e| std::io::Error::other(format!("invalid Last-Modified date: {e}")))?;
self.last_modified = mod_time;
let version_id = h.get(X_AMZ_VERSION_ID)
let version_id = h
.get(X_AMZ_VERSION_ID)
.ok_or_else(|| std::io::Error::other("missing version ID header"))?
.to_str()
.map_err(|e| std::io::Error::other(format!("invalid version ID header: {e}")))?;
self.version_id = version_id.to_string();
let body_str = String::from_utf8(body_vec)
.map_err(|e| std::io::Error::other(format!("invalid UTF-8 body: {e}")))?;
let body_str = String::from_utf8(body_vec).map_err(|e| std::io::Error::other(format!("invalid UTF-8 body: {e}")))?;
let mut response = match quick_xml::de::from_str::<ObjectAttributesResponse>(&body_str) {
Ok(result) => result,
Err(err) => {
@@ -175,7 +176,10 @@ impl TransitionClient {
}
let mut headers = HeaderMap::new();
headers.insert(X_AMZ_OBJECT_ATTRIBUTES, HeaderValue::from_str(GET_OBJECT_ATTRIBUTES_TAGS).expect("valid header value"));
headers.insert(
X_AMZ_OBJECT_ATTRIBUTES,
HeaderValue::from_str(GET_OBJECT_ATTRIBUTES_TAGS).expect("valid header value"),
);
if opts.part_number_marker > 0 {
headers.insert(
@@ -185,7 +189,10 @@ impl TransitionClient {
}
if opts.max_parts > 0 {
headers.insert(X_AMZ_MAX_PARTS, HeaderValue::from_str(&opts.max_parts.to_string()).expect("valid header value"));
headers.insert(
X_AMZ_MAX_PARTS,
HeaderValue::from_str(&opts.max_parts.to_string()).expect("valid header value"),
);
} else {
headers.insert(
X_AMZ_MAX_PARTS,
@@ -222,9 +229,7 @@ impl TransitionClient {
let resp_status = resp.status();
let h = resp.headers().clone();
let has_etag = h.get("ETag")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
let has_etag = h.get("ETag").and_then(|v| v.to_str().ok()).unwrap_or("");
if !has_etag.is_empty() {
return Err(std::io::Error::other(
"get_object_attributes is not supported by the current endpoint version",
@@ -241,8 +246,8 @@ impl TransitionClient {
}
if resp_status != http::StatusCode::OK {
let err_body = String::from_utf8(body_vec)
.map_err(|e| std::io::Error::other(format!("invalid UTF-8 error body: {e}")))?;
let err_body =
String::from_utf8(body_vec).map_err(|e| std::io::Error::other(format!("invalid UTF-8 error body: {e}")))?;
let mut er = match quick_xml::de::from_str::<AccessControlPolicy>(&err_body) {
Ok(result) => result,
Err(err) => {
crates/ecstore/src/layout/mod.rs
+14
View File
@@ -1,3 +1,17 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Static ECStore layout boundaries.
//!
//! This module owns read-only layout descriptors used to keep static set
crates/ecstore/src/object_api/readers.rs
+27 -21
View File
@@ -1,3 +1,17 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use super::*;
#[cfg(feature = "rio-v2")]
use aes_gcm::aead::Payload;
@@ -122,7 +136,7 @@ fn restore_request_active(opts: &ObjectOptions) -> bool {
restore.type_.is_some() || restore.days.is_some() || restore.output_location.is_some() || restore.select_parameters.is_some()
}
fn decode_compression_index(index: Option<&bytes::Bytes>) -> Option<Index> {
fn decode_compression_index(index: Option<&Bytes>) -> Option<Index> {
crate::io_support::rio::decode_compression_index_bytes(index?)
}
@@ -833,11 +847,7 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> RangedDecompressReader<R> {
}
impl<R: AsyncRead + Unpin + Send + Sync + 'static> AsyncRead for RangedDecompressReader<R> {
fn poll_read(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<std::io::Result<()>> {
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
use std::pin::Pin;
use std::task::Poll;
use tokio::io::ReadBuf;
@@ -985,11 +995,7 @@ impl<R: AsyncRead + Unpin + Send + 'static> StreamConsumer<R> {
}
impl<R: AsyncRead + Unpin + Send + 'static> AsyncRead for StreamConsumer<R> {
fn poll_read(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<std::io::Result<()>> {
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
use std::pin::Pin;
use std::task::Poll;
@@ -1032,7 +1038,7 @@ fn encrypted_plaintext_size(oi: &ObjectInfo, is_multipart: bool, is_compressed:
oi.decrypted_size().map_err(Into::into)
}
fn is_multipart_encrypted_object(parts: &[rustfs_filemeta::ObjectPartInfo], etag: Option<&str>) -> bool {
fn is_multipart_encrypted_object(parts: &[ObjectPartInfo], etag: Option<&str>) -> bool {
if parts.len() > 1 {
return true;
}
@@ -1040,13 +1046,13 @@ fn is_multipart_encrypted_object(parts: &[rustfs_filemeta::ObjectPartInfo], etag
etag.map(|etag| etag.trim_matches('"').len() != 32).unwrap_or(false)
}
fn multipart_plaintext_size(parts: &[rustfs_filemeta::ObjectPartInfo], fallback: i64) -> i64 {
fn multipart_plaintext_size(parts: &[ObjectPartInfo], fallback: i64) -> i64 {
let total: i64 = parts.iter().map(part_plaintext_size).sum();
if total > 0 { total } else { fallback }
}
fn multipart_part_numbers(parts: &[rustfs_filemeta::ObjectPartInfo]) -> Vec<usize> {
fn multipart_part_numbers(parts: &[ObjectPartInfo]) -> Vec<usize> {
parts.iter().map(|part| part.number).collect()
}
@@ -1605,13 +1611,13 @@ mod tests {
fn ssec_headers_from_key(key_bytes: [u8; 32]) -> HeaderMap<HeaderValue> {
let mut headers = HeaderMap::new();
headers.insert(rustfs_utils::http::SSEC_ALGORITHM_HEADER, HeaderValue::from_static("AES256"));
headers.insert(SSEC_ALGORITHM_HEADER, HeaderValue::from_static("AES256"));
headers.insert(
rustfs_utils::http::SSEC_KEY_HEADER,
SSEC_KEY_HEADER,
HeaderValue::from_str(&BASE64_STANDARD.encode(key_bytes)).expect("valid base64 header"),
);
headers.insert(
rustfs_utils::http::SSEC_KEY_MD5_HEADER,
SSEC_KEY_MD5_HEADER,
HeaderValue::from_str(&BASE64_STANDARD.encode(md5_bytes(key_bytes))).expect("valid md5 header"),
);
headers
@@ -2616,7 +2622,7 @@ mod tests {
#[tokio::test]
async fn test_get_object_reader_compressed_range_returns_physical_offset_from_index() {
let mut index = crate::io_support::rio::Index::new();
let mut index = Index::new();
index.add(0, 0).unwrap();
index.add(1_048_576, 2_097_152).unwrap();
@@ -2670,7 +2676,7 @@ mod tests {
#[tokio::test]
async fn test_read_plan_compressed_range_tracks_storage_and_visible_offsets() {
let mut index = crate::io_support::rio::Index::new();
let mut index = Index::new();
index.add(0, 0).unwrap();
index.add(1_048_576, 2_097_152).unwrap();
@@ -2746,7 +2752,7 @@ mod tests {
#[cfg(feature = "rio-v2")]
#[tokio::test]
async fn test_read_plan_accepts_minio_headerless_compression_index() {
let mut index = crate::io_support::rio::Index::new();
let mut index = Index::new();
index.add(0, 0).unwrap();
index.add(1_048_576, 2_097_152).unwrap();
let headerless_index = crate::io_support::rio::compression_index_storage_bytes(&index);
@@ -2800,7 +2806,7 @@ mod tests {
#[cfg(feature = "rio-v2")]
#[test]
fn test_get_compressed_offsets_aligns_encrypted_ranges_to_dare_packages() {
let mut index = crate::io_support::rio::Index::new();
let mut index = Index::new();
index.add(0, 0).unwrap();
index.add(200_000, 2_097_152).unwrap();
let stored_index = crate::io_support::rio::compression_index_storage_bytes(&index);
crates/ecstore/src/object_api/types.rs
+23 -9
View File
@@ -1,3 +1,17 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use super::*;
use crate::storage_api_contracts::{
list::VersionMarker,
@@ -354,14 +368,14 @@ impl ObjectInfo {
// Parse expires from metadata (HTTP date format RFC 7231 or ISO 8601)
let expires = fi.metadata.get("expires").and_then(|s| {
// Try parsing as ISO 8601 first
time::OffsetDateTime::parse(s, &time::format_description::well_known::Iso8601::DEFAULT)
OffsetDateTime::parse(s, &time::format_description::well_known::Iso8601::DEFAULT)
.or_else(|_| {
// Try RFC 2822 format
time::OffsetDateTime::parse(s, &time::format_description::well_known::Rfc2822)
OffsetDateTime::parse(s, &time::format_description::well_known::Rfc2822)
})
.or_else(|_| {
// Try RFC 3339 format
time::OffsetDateTime::parse(s, &time::format_description::well_known::Rfc3339)
OffsetDateTime::parse(s, &time::format_description::well_known::Rfc3339)
})
.ok()
});
@@ -775,7 +789,7 @@ mod tests {
#[tokio::test]
async fn versions_listing_applies_version_marker_only_to_first_entry() {
let metadata = rustfs_filemeta::test_data::create_real_xlmeta().expect("test metadata should be valid");
let entries = rustfs_filemeta::MetaCacheEntriesSorted {
let entries = MetaCacheEntriesSorted {
o: rustfs_filemeta::MetaCacheEntries(vec![
Some(rustfs_filemeta::MetaCacheEntry {
name: "obj-a".to_owned(),
@@ -878,7 +892,7 @@ mod tests {
#[test]
fn from_file_info_preserves_replication_decision() {
let fi = rustfs_filemeta::FileInfo {
let fi = FileInfo {
replication_state_internal: Some(ReplicationState {
replicate_decision_str: "arn=true;false;arn:replication::1:dest;rule-id".to_string(),
..Default::default()
@@ -904,11 +918,11 @@ mod tests {
actual_size: 0,
user_defined: Arc::new(user_defined),
parts: Arc::new(vec![
rustfs_filemeta::ObjectPartInfo {
ObjectPartInfo {
actual_size: 4,
..Default::default()
},
rustfs_filemeta::ObjectPartInfo {
ObjectPartInfo {
actual_size: 5,
..Default::default()
},
@@ -1016,13 +1030,13 @@ mod tests {
user_defined: Arc::new(ud),
user_tags: Arc::new("env=prod&team=storage".to_string()),
parts: Arc::new(vec![
rustfs_filemeta::ObjectPartInfo {
ObjectPartInfo {
number: 1,
size: 1024,
actual_size: 1024,
..Default::default()
},
rustfs_filemeta::ObjectPartInfo {
ObjectPartInfo {
number: 2,
size: 512,
actual_size: 512,
crates/ecstore/src/runtime/global.rs
+2 -4
View File
@@ -338,9 +338,7 @@ pub fn shutdown_background_services() {
/// * `Ok(())` if successful
/// * `Err(Arc<dyn LockClient>)` if setting fails (client already set)
///
pub fn set_global_lock_client(
client: Arc<dyn rustfs_lock::client::LockClient>,
) -> Result<(), Arc<dyn rustfs_lock::client::LockClient>> {
pub fn set_global_lock_client(client: Arc<dyn LockClient>) -> Result<(), Arc<dyn LockClient>> {
GLOBAL_LOCAL_LOCK_CLIENT.set(client)
}
@@ -349,7 +347,7 @@ pub fn set_global_lock_client(
/// # Returns
/// * `Option<Arc<dyn LockClient>>` - The global lock client, if set
///
pub fn get_global_lock_client() -> Option<Arc<dyn rustfs_lock::client::LockClient>> {
pub fn get_global_lock_client() -> Option<Arc<dyn LockClient>> {
GLOBAL_LOCAL_LOCK_CLIENT.get().cloned()
}
crates/ecstore/src/runtime/sources.rs
+2 -2
View File
@@ -237,9 +237,9 @@ pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option<usize>
pub(crate) fn backend_storage_class_parities(default_standard_parity: usize) -> (Option<usize>, Option<usize>) {
if let Some(sc) = get_global_storage_class() {
let standard = sc
.get_parity_for_sc(crate::config::storageclass::CLASS_STANDARD)
.get_parity_for_sc(storageclass::CLASS_STANDARD)
.or(Some(default_standard_parity));
let reduced_redundancy = sc.get_parity_for_sc(crate::config::storageclass::RRS);
let reduced_redundancy = sc.get_parity_for_sc(storageclass::RRS);
(standard, reduced_redundancy)
} else {
(Some(default_standard_parity), None)
crates/ecstore/src/set_disk/heal.rs
+5 -5
View File
@@ -129,14 +129,14 @@ impl SetDisks {
let erasure = if !latest_meta.deleted && !latest_meta.is_remote() {
// Initialize erasure coding; use legacy mode for old-version files
crate::erasure::coding::Erasure::new_with_options(
coding::Erasure::new_with_options(
latest_meta.erasure.data_blocks,
latest_meta.erasure.parity_blocks,
latest_meta.erasure.block_size,
latest_meta.uses_legacy_checksum,
)
} else {
crate::erasure::coding::Erasure::default()
coding::Erasure::default()
};
result.object_size =
@@ -385,9 +385,9 @@ impl SetDisks {
if let (Some(disk), Some(metadata)) = (disk, &copy_parts_metadata[index]) {
let checksum_info = metadata.erasure.get_checksum_info(part.number);
let checksum_algo = if metadata.uses_legacy_checksum
&& checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S
&& checksum_info.algorithm == HashAlgorithm::HighwayHash256S
{
rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
@@ -498,7 +498,7 @@ impl SetDisks {
// parts_metadata[index].data = Some(w.inline_data().to_vec());
// }
parts_metadata[index].data =
Some(writer.into_inline_data().map(bytes::Bytes::from).unwrap_or_default());
Some(writer.into_inline_data().map(Bytes::from).unwrap_or_default());
}
parts_metadata[index].set_inline_data();
} else {
crates/ecstore/src/set_disk/lock.rs
+6 -6
View File
@@ -131,7 +131,7 @@ impl SetDisks {
fn reprobe_runtime_candidates_once(&self, disks: &[DiskStore]) {
for disk in disks {
if disk.runtime_state() != crate::disk::health_state::RuntimeDriveHealthState::Online {
if disk.runtime_state() != disk::health_state::RuntimeDriveHealthState::Online {
disk.reset_health_for_store_init_retry();
}
}
@@ -536,15 +536,15 @@ mod tests {
all_disks[1]
.as_ref()
.expect("disk 1 should exist")
.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Suspect);
.force_runtime_state_for_test(disk::health_state::RuntimeDriveHealthState::Suspect);
all_disks[2]
.as_ref()
.expect("disk 2 should exist")
.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Returning);
.force_runtime_state_for_test(disk::health_state::RuntimeDriveHealthState::Returning);
all_disks[3]
.as_ref()
.expect("disk 3 should exist")
.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Offline);
.force_runtime_state_for_test(disk::health_state::RuntimeDriveHealthState::Offline);
let snapshot = set_disks.drive_membership_snapshot().await;
assert_eq!(snapshot.online.len(), 1);
@@ -563,7 +563,7 @@ mod tests {
assert!(
online_disks
.iter()
.all(|disk| { disk.runtime_state() != crate::disk::health_state::RuntimeDriveHealthState::Offline }),
.all(|disk| { disk.runtime_state() != disk::health_state::RuntimeDriveHealthState::Offline }),
"offline disks should be filtered by membership snapshot"
);
@@ -601,7 +601,7 @@ mod tests {
let all_disks = set_disks.get_disks_internal().await;
for disk in all_disks.iter().flatten() {
disk.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Returning);
disk.force_runtime_state_for_test(disk::health_state::RuntimeDriveHealthState::Returning);
}
let (online_disks, infos, healing) = set_disks.get_online_disks_with_healing_and_info(false).await;
crates/ecstore/src/set_disk/mod.rs
+3 -7
View File
@@ -1885,8 +1885,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap());
let result: Result<ObjectInfo> = async {
let erasure =
coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let erasure = coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let is_inline_buffer =
runtime_sources::storage_class_should_inline(erasure.shard_file_size(data.size()), opts.versioned);
@@ -4218,8 +4217,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
let tmp_part = format!("{}x{}", Uuid::new_v4(), OffsetDateTime::now_utc().unix_timestamp());
let tmp_part_path = Arc::new(format!("{tmp_part}/{part_suffix}"));
let erasure =
coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let erasure = coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let writer_setup_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
let mut writers = Vec::with_capacity(shuffle_disks.len());
@@ -5912,9 +5910,7 @@ async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<ru
let runtime_state = disk.runtime_state();
let offline_duration_seconds = disk.offline_duration_secs();
let capacity_snapshot = disk.last_capacity_snapshot();
if runtime_state.should_probe_for_admin()
|| runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect
{
if runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect {
match disk.disk_info(&DiskInfoOptions::default()).await {
Ok(res) => {
disk.record_capacity_probe(res.total, res.used, res.free);
crates/ecstore/src/set_disk/multipart.rs
+2 -2
View File
@@ -221,7 +221,7 @@ mod tests {
#[tokio::test]
async fn collect_list_parts_results_fails_early_when_quorum_is_impossible() {
let started = std::time::Instant::now();
let started = Instant::now();
let tasks: Vec<_> = vec![
(10_u64, Err(DiskError::DiskNotFound)),
(15, Err(DiskError::DiskNotFound)),
@@ -285,7 +285,7 @@ mod tests {
#[tokio::test]
async fn collect_list_parts_results_fails_early_when_file_not_found_fallback_is_impossible() {
let started = std::time::Instant::now();
let started = Instant::now();
let tasks: Vec<_> = vec![
(5_u64, Err(DiskError::FileNotFound)),
(10, Err(DiskError::FileCorrupt)),
crates/ecstore/src/set_disk/read.rs
+37 -50
View File
@@ -1936,7 +1936,7 @@ impl SetDisks {
object, offset, length, end_offset, part_index, last_part_index, last_part_relative_offset, "Multipart read bounds"
);
let erasure = crate::erasure::coding::Erasure::new_with_options(
let erasure = coding::Erasure::new_with_options(
fi.erasure.data_blocks,
fi.erasure.parity_blocks,
fi.erasure.block_size,
@@ -1987,12 +1987,11 @@ impl SetDisks {
);
let checksum_info = fi.erasure.get_checksum_info(part_number);
let checksum_algo =
if fi.uses_legacy_checksum && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S {
rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
let read_length = till_offset.saturating_sub(read_offset);
// Read zero-copy configuration from environment variable
@@ -2278,7 +2277,7 @@ impl SetDisks {
) -> Result<GetCodecStreamingReaderBuildOutcome> {
let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi);
let erasure = crate::erasure::coding::Erasure::new_with_options(
let erasure = coding::Erasure::new_with_options(
fi.erasure.data_blocks,
fi.erasure.parity_blocks,
fi.erasure.block_size,
@@ -2362,7 +2361,7 @@ impl SetDisks {
fi: &FileInfo,
files: &[FileInfo],
disks: &[Option<DiskStore>],
erasure: &crate::erasure::coding::Erasure,
erasure: &coding::Erasure,
part_number: usize,
part_offset: usize,
part_length: usize,
@@ -2373,9 +2372,8 @@ impl SetDisks {
return Err(Error::other("codec streaming reader part length exceeds part size"));
}
let checksum_info = fi.erasure.get_checksum_info(part_number);
let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S
{
rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
HashAlgorithm::HighwayHash256SLegacy
} else {
checksum_info.algorithm
};
@@ -2427,24 +2425,19 @@ impl SetDisks {
}
let readers = reader_setup.readers;
let source =
crate::erasure::coding::decode::ParallelReader::new_with_metrics_path_read_costs_and_reconstruction_verification(
readers,
erasure.clone(),
part_offset,
part_size,
Some(metrics_path),
read_costs,
);
let source = coding::decode::ParallelReader::new_with_metrics_path_read_costs_and_reconstruction_verification(
readers,
erasure.clone(),
part_offset,
part_size,
Some(metrics_path),
read_costs,
);
let engine = build_get_codec_streaming_decode_engine(erasure.clone())?;
let reader = crate::erasure::coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(
source,
engine,
part_length,
metrics_path,
)?;
let reader =
coding::decode_reader::ErasureDecodeReader::new_with_metrics_path(source, engine, part_length, metrics_path)?;
Ok(GetCodecStreamingReaderBuildOutcome::Reader(Box::new(
crate::erasure::coding::decode_reader::SyncErasureDecodeReader::new_with_metrics_path(reader, metrics_path),
coding::decode_reader::SyncErasureDecodeReader::new_with_metrics_path(reader, metrics_path),
)))
}
}
@@ -2627,7 +2620,7 @@ mod metadata_cache_tests {
"read-repair submission should not wait for admission response"
);
tokio::time::timeout(Duration::from_secs(1), async {
timeout(Duration::from_secs(1), async {
while SLOW_READ_REPAIR_SUBMITTER_CALLS.load(Ordering::Relaxed) == 0 {
tokio::time::sleep(Duration::from_millis(5)).await;
}
@@ -2656,7 +2649,7 @@ mod metadata_cache_tests {
)
.await;
let released_key = tokio::time::timeout(Duration::from_secs(1), async {
let released_key = timeout(Duration::from_secs(1), async {
loop {
if let Some(key) = reserve_read_repair_heal(&bucket, "object", None, 0, 0).await {
break key;
@@ -3707,7 +3700,7 @@ mod tests {
);
let mut remote_fi = fi;
remote_fi.transition_status = crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string();
remote_fi.transition_status = TRANSITION_COMPLETE.to_string();
let remote = codec_streaming_test_object_info(&remote_fi);
assert_eq!(
codec_streaming_reader_gate_for_test(&None, &remote, &remote_fi, true).decision,
@@ -3831,7 +3824,7 @@ mod tests {
#[test]
fn codec_streaming_decode_engine_builder_selects_rustfs() {
temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS), || {
let erasure = crate::erasure::coding::Erasure::new(4, 2, 32);
let erasure = coding::Erasure::new(4, 2, 32);
let engine = build_get_codec_streaming_decode_engine(erasure).expect("engine should be built");
assert!(matches!(engine, CodecStreamingDecodeEngine::Rustfs(_)));
@@ -3841,17 +3834,11 @@ mod tests {
#[test]
fn codec_streaming_metrics_path_matches_selected_engine() {
temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, None::<&str>, || {
assert_eq!(
get_codec_streaming_metrics_path(),
crate::diagnostics::get::GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE
);
assert_eq!(get_codec_streaming_metrics_path(), GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE);
});
temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, Some(GET_CODEC_STREAMING_ENGINE_RUSTFS), || {
assert_eq!(
get_codec_streaming_metrics_path(),
crate::diagnostics::get::GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE
);
assert_eq!(get_codec_streaming_metrics_path(), GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE);
});
}
@@ -4067,7 +4054,7 @@ mod tests {
#[tokio::test]
async fn collect_read_multiple_results_fails_early_when_quorum_is_impossible() {
let started = std::time::Instant::now();
let started = Instant::now();
let resp = ReadMultipleResp {
bucket: "bucket".to_string(),
prefix: "prefix".to_string(),
@@ -4085,14 +4072,14 @@ mod tests {
]
.into_iter()
.map(|(delay_ms, outcome)| async move {
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
outcome
})
.collect();
let result = collect_read_multiple_results(tasks, 2).await;
assert!(result.is_err(), "quorum should become impossible before slow tail completes");
assert!(started.elapsed() < std::time::Duration::from_millis(120));
assert!(started.elapsed() < Duration::from_millis(120));
}
#[tokio::test]
@@ -4114,7 +4101,7 @@ mod tests {
]
.into_iter()
.map(|(delay_ms, outcome)| async move {
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
outcome
})
.collect();
@@ -4142,7 +4129,7 @@ mod tests {
.map(|(delay_ms, should_panic)| {
let resp = resp.clone();
async move {
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
if should_panic {
panic!("simulated task panic");
}
@@ -4160,7 +4147,7 @@ mod tests {
#[tokio::test]
async fn collect_read_parts_results_fails_early_when_quorum_is_impossible() {
let started = std::time::Instant::now();
let started = Instant::now();
let part = ObjectPartInfo {
number: 1,
etag: "etag".to_string(),
@@ -4174,14 +4161,14 @@ mod tests {
]
.into_iter()
.map(|(delay_ms, outcome)| async move {
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
outcome
})
.collect();
let result = collect_read_parts_results(tasks, 2).await;
assert!(result.is_err(), "quorum should become impossible before slow tail completes");
assert!(started.elapsed() < std::time::Duration::from_millis(120));
assert!(started.elapsed() < Duration::from_millis(120));
}
#[tokio::test]
@@ -4199,7 +4186,7 @@ mod tests {
]
.into_iter()
.map(|(delay_ms, outcome)| async move {
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
outcome
})
.collect();
@@ -4222,7 +4209,7 @@ mod tests {
.map(|(delay_ms, should_panic)| {
let part = part.clone();
async move {
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
if should_panic {
panic!("simulated task panic");
}
crates/ecstore/src/set_disk/replication.rs
-2
View File
@@ -24,9 +24,7 @@ impl SetDisks {
) -> Result<()> {
let mut oi = obj_info.clone();
oi.metadata_only = true;
Arc::make_mut(&mut oi.user_defined).remove(X_AMZ_RESTORE.as_str());
let version_id = oi.version_id.map(|v| v.to_string());
let _obj = self
.copy_object(
rustfs/src/admin/handlers/bucket_meta.rs
+44 -11
View File
@@ -380,7 +380,10 @@ impl Operation for ExportBucketMetadata {
.map_err(|e| s3_error!(InternalError, "failed to finalize export archive: {e}"))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/zip".parse().expect("valid header value"));
header.insert(CONTENT_DISPOSITION, "attachment; filename=bucket-meta.zip".parse().expect("valid header value"));
header.insert(
CONTENT_DISPOSITION,
"attachment; filename=bucket-meta.zip".parse().expect("valid header value"),
);
header.insert(CONTENT_LENGTH, zip_bytes.get_ref().len().to_string().parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(zip_bytes.into_inner())), header))
}
@@ -597,7 +600,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
let metadata = match bucket_metadatas.get_mut(bucket_name) {
Some(m) => m,
None => continue,
};
metadata.policy_config_json = content;
metadata.policy_config_updated_at = update_at;
}
@@ -617,7 +623,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
let metadata = match bucket_metadatas.get_mut(bucket_name) {
Some(m) => m,
None => continue,
};
metadata.notification_config_xml = content;
metadata.notification_config_updated_at = update_at;
}
@@ -638,7 +647,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
let metadata = match bucket_metadatas.get_mut(bucket_name) {
Some(m) => m,
None => continue,
};
metadata.lifecycle_config_xml = content;
metadata.lifecycle_config_updated_at = update_at;
}
@@ -659,7 +671,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
let metadata = match bucket_metadatas.get_mut(bucket_name) {
Some(m) => m,
None => continue,
};
metadata.encryption_config_xml = content;
metadata.encryption_config_updated_at = update_at;
}
@@ -680,7 +695,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
let metadata = match bucket_metadatas.get_mut(bucket_name) {
Some(m) => m,
None => continue,
};
metadata.tagging_config_xml = content;
metadata.tagging_config_updated_at = update_at;
}
@@ -701,7 +719,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
let metadata = match bucket_metadatas.get_mut(bucket_name) {
Some(m) => m,
None => continue,
};
metadata.quota_config_json = content;
metadata.quota_config_updated_at = update_at;
}
@@ -722,7 +743,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
let metadata = match bucket_metadatas.get_mut(bucket_name) {
Some(m) => m,
None => continue,
};
metadata.object_lock_config_xml = content;
metadata.object_lock_config_updated_at = update_at;
}
@@ -743,7 +767,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
let metadata = match bucket_metadatas.get_mut(bucket_name) {
Some(m) => m,
None => continue,
};
metadata.versioning_config_xml = content;
metadata.versioning_config_updated_at = update_at;
}
@@ -764,7 +791,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
let metadata = match bucket_metadatas.get_mut(bucket_name) {
Some(m) => m,
None => continue,
};
metadata.replication_config_xml = content;
metadata.replication_config_updated_at = update_at;
}
@@ -785,7 +815,10 @@ impl Operation for ImportBucketMetadata {
continue;
}
let metadata = match bucket_metadatas.get_mut(bucket_name) { Some(m) => m, None => continue, };
let metadata = match bucket_metadatas.get_mut(bucket_name) {
Some(m) => m,
None => continue,
};
metadata.bucket_targets_config_json = content;
metadata.bucket_targets_config_updated_at = update_at;
}
rustfs/src/admin/handlers/tier.rs
+45 -9
View File
@@ -220,31 +220,67 @@ impl Operation for AddTier {
match args.tier_type {
TierType::S3 => {
args.name = args.s3.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing S3 configuration"))?.name;
args.name = args
.s3
.clone()
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing S3 configuration"))?
.name;
}
TierType::RustFS => {
args.name = args.rustfs.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing RustFS configuration"))?.name;
args.name = args
.rustfs
.clone()
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing RustFS configuration"))?
.name;
}
TierType::MinIO => {
args.name = args.minio.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing MinIO configuration"))?.name;
args.name = args
.minio
.clone()
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing MinIO configuration"))?
.name;
}
TierType::Aliyun => {
args.name = args.aliyun.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Aliyun configuration"))?.name;
args.name = args
.aliyun
.clone()
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Aliyun configuration"))?
.name;
}
TierType::Tencent => {
args.name = args.tencent.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Tencent configuration"))?.name;
args.name = args
.tencent
.clone()
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Tencent configuration"))?
.name;
}
TierType::Huaweicloud => {
args.name = args.huaweicloud.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Huawei Cloud configuration"))?.name;
args.name = args
.huaweicloud
.clone()
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Huawei Cloud configuration"))?
.name;
}
TierType::Azure => {
args.name = args.azure.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Azure configuration"))?.name;
args.name = args
.azure
.clone()
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Azure configuration"))?
.name;
}
TierType::GCS => {
args.name = args.gcs.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing GCS configuration"))?.name;
args.name = args
.gcs
.clone()
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing GCS configuration"))?
.name;
}
TierType::R2 => {
args.name = args.r2.clone().ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing R2 configuration"))?.name;
args.name = args
.r2
.clone()
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing R2 configuration"))?
.name;
}
_ => (),
}
rustfs/src/admin/handlers/user.rs
+4 -1
View File
@@ -841,7 +841,10 @@ impl Operation for ExportIam {
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "application/zip".parse().expect("valid header value"));
header.insert(CONTENT_DISPOSITION, "attachment; filename=iam-assets.zip".parse().expect("valid header value"));
header.insert(
CONTENT_DISPOSITION,
"attachment; filename=iam-assets.zip".parse().expect("valid header value"),
);
header.insert(CONTENT_LENGTH, zip_bytes.get_ref().len().to_string().parse().expect("valid header value"));
Ok(S3Response::with_headers((StatusCode::OK, Body::from(zip_bytes.into_inner())), header))
}
rustfs/src/app/object_usecase.rs
+1
View File
@@ -3450,6 +3450,7 @@ impl DefaultObjectUsecase {
sse_customer_key_md5,
ssekms_key_id,
encryption_applied,
is_inline_fast_path: _,
} = read_setup;
let versioning_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
rustfs/src/server/layer.rs
+16 -4
View File
@@ -1435,7 +1435,10 @@ where
.unwrap());
}
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).expect("valid response body");
let mut response = Response::builder()
.status(StatusCode::OK)
.body(ResBody::default())
.expect("valid response body");
let cors_layer = ConditionalCorsLayer {
cors_origins: (*cors_origins).clone(),
};
@@ -1464,7 +1467,10 @@ where
let cors_allowed = cors_headers.contains_key(cors::response::ACCESS_CONTROL_ALLOW_ORIGIN);
let status = if cors_allowed { StatusCode::OK } else { StatusCode::FORBIDDEN };
let mut response = Response::builder().status(status).body(ResBody::default()).expect("valid response body");
let mut response = Response::builder()
.status(status)
.body(ResBody::default())
.expect("valid response body");
if cors_allowed {
for (key, value) in cors_headers.iter() {
response.headers_mut().insert(key, value.clone());
@@ -1474,7 +1480,10 @@ where
}
// No bucket-level CORS config: fall back to global/default CORS behavior.
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).expect("valid response body");
let mut response = Response::builder()
.status(StatusCode::OK)
.body(ResBody::default())
.expect("valid response body");
cors_layer.apply_cors_headers(&request_headers, response.headers_mut());
Ok(response)
});
@@ -1482,7 +1491,10 @@ where
let request_headers_clone = request_headers.clone();
return Box::pin(async move {
let mut response = Response::builder().status(StatusCode::OK).body(ResBody::default()).expect("valid response body");
let mut response = Response::builder()
.status(StatusCode::OK)
.body(ResBody::default())
.expect("valid response body");
let cors_layer = ConditionalCorsLayer {
cors_origins: (*cors_origins).clone(),
};