From 143491e0af6c98ea5e2aa9c03f7789bc0e3c7168 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 28 Jun 2026 11:41:11 +0800 Subject: [PATCH] fmt and improve import --- .../replication/replication_resyncer.rs | 27 ++++-- .../src/client/api_get_object_attributes.rs | 27 +++--- crates/ecstore/src/layout/mod.rs | 14 +++ crates/ecstore/src/object_api/readers.rs | 48 +++++----- crates/ecstore/src/object_api/types.rs | 32 +++++-- crates/ecstore/src/runtime/global.rs | 6 +- crates/ecstore/src/runtime/sources.rs | 4 +- crates/ecstore/src/set_disk/heal.rs | 10 +-- crates/ecstore/src/set_disk/lock.rs | 12 +-- crates/ecstore/src/set_disk/mod.rs | 10 +-- crates/ecstore/src/set_disk/multipart.rs | 4 +- crates/ecstore/src/set_disk/read.rs | 87 ++++++++----------- crates/ecstore/src/set_disk/replication.rs | 2 - rustfs/src/admin/handlers/bucket_meta.rs | 55 +++++++++--- rustfs/src/admin/handlers/tier.rs | 54 ++++++++++-- rustfs/src/admin/handlers/user.rs | 5 +- rustfs/src/app/object_usecase.rs | 1 + rustfs/src/server/layer.rs | 20 ++++- 18 files changed, 269 insertions(+), 149 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 7221444b9..2e470685b 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -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; } diff --git a/crates/ecstore/src/client/api_get_object_attributes.rs b/crates/ecstore/src/client/api_get_object_attributes.rs index 576c36201..eb6c55be6 100644 --- a/crates/ecstore/src/client/api_get_object_attributes.rs +++ b/crates/ecstore/src/client/api_get_object_attributes.rs @@ -133,7 +133,8 @@ struct ObjectAttributePart { impl ObjectAttributes { pub async fn parse_response(&mut self, h: &HeaderMap, body_vec: Vec) -> 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::(&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::(&err_body) { Ok(result) => result, Err(err) => { diff --git a/crates/ecstore/src/layout/mod.rs b/crates/ecstore/src/layout/mod.rs index 16ee4be1d..d6ad89a43 100644 --- a/crates/ecstore/src/layout/mod.rs +++ b/crates/ecstore/src/layout/mod.rs @@ -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 diff --git a/crates/ecstore/src/object_api/readers.rs b/crates/ecstore/src/object_api/readers.rs index b5812d451..2d258a4af 100644 --- a/crates/ecstore/src/object_api/readers.rs +++ b/crates/ecstore/src/object_api/readers.rs @@ -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 { +fn decode_compression_index(index: Option<&Bytes>) -> Option { crate::io_support::rio::decode_compression_index_bytes(index?) } @@ -833,11 +847,7 @@ impl RangedDecompressReader { } impl AsyncRead for RangedDecompressReader { - fn poll_read( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - buf: &mut tokio::io::ReadBuf<'_>, - ) -> std::task::Poll> { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { use std::pin::Pin; use std::task::Poll; use tokio::io::ReadBuf; @@ -985,11 +995,7 @@ impl StreamConsumer { } impl AsyncRead for StreamConsumer { - fn poll_read( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - buf: &mut tokio::io::ReadBuf<'_>, - ) -> std::task::Poll> { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { 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 { +fn multipart_part_numbers(parts: &[ObjectPartInfo]) -> Vec { parts.iter().map(|part| part.number).collect() } @@ -1605,13 +1611,13 @@ mod tests { fn ssec_headers_from_key(key_bytes: [u8; 32]) -> HeaderMap { 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); diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index ba8f12265..c38892863 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -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, diff --git a/crates/ecstore/src/runtime/global.rs b/crates/ecstore/src/runtime/global.rs index e79cbd096..407c64152 100644 --- a/crates/ecstore/src/runtime/global.rs +++ b/crates/ecstore/src/runtime/global.rs @@ -338,9 +338,7 @@ pub fn shutdown_background_services() { /// * `Ok(())` if successful /// * `Err(Arc)` if setting fails (client already set) /// -pub fn set_global_lock_client( - client: Arc, -) -> Result<(), Arc> { +pub fn set_global_lock_client(client: Arc) -> Result<(), Arc> { GLOBAL_LOCAL_LOCK_CLIENT.set(client) } @@ -349,7 +347,7 @@ pub fn set_global_lock_client( /// # Returns /// * `Option>` - The global lock client, if set /// -pub fn get_global_lock_client() -> Option> { +pub fn get_global_lock_client() -> Option> { GLOBAL_LOCAL_LOCK_CLIENT.get().cloned() } diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index 6bb4e45e9..9d895e9bb 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -237,9 +237,9 @@ pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option pub(crate) fn backend_storage_class_parities(default_standard_parity: usize) -> (Option, Option) { 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) diff --git a/crates/ecstore/src/set_disk/heal.rs b/crates/ecstore/src/set_disk/heal.rs index f193b1c0a..21478aa7f 100644 --- a/crates/ecstore/src/set_disk/heal.rs +++ b/crates/ecstore/src/set_disk/heal.rs @@ -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, ©_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 { diff --git a/crates/ecstore/src/set_disk/lock.rs b/crates/ecstore/src/set_disk/lock.rs index cde9c9af7..d7f748822 100644 --- a/crates/ecstore/src/set_disk/lock.rs +++ b/crates/ecstore/src/set_disk/lock.rs @@ -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; diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 96ba51a1a..409b156f0 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -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 = 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], eps: &[Endpoint]) -> Vec { disk.record_capacity_probe(res.total, res.used, res.free); diff --git a/crates/ecstore/src/set_disk/multipart.rs b/crates/ecstore/src/set_disk/multipart.rs index 14bdbb559..038909fec 100644 --- a/crates/ecstore/src/set_disk/multipart.rs +++ b/crates/ecstore/src/set_disk/multipart.rs @@ -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)), diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index ae583921d..e3708aca1 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -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 { 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], - 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"); } diff --git a/crates/ecstore/src/set_disk/replication.rs b/crates/ecstore/src/set_disk/replication.rs index a35e281da..954f0b582 100644 --- a/crates/ecstore/src/set_disk/replication.rs +++ b/crates/ecstore/src/set_disk/replication.rs @@ -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( diff --git a/rustfs/src/admin/handlers/bucket_meta.rs b/rustfs/src/admin/handlers/bucket_meta.rs index f3d865043..fac7179b2 100644 --- a/rustfs/src/admin/handlers/bucket_meta.rs +++ b/rustfs/src/admin/handlers/bucket_meta.rs @@ -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; } diff --git a/rustfs/src/admin/handlers/tier.rs b/rustfs/src/admin/handlers/tier.rs index f5d7fb4b8..7703e6b8c 100644 --- a/rustfs/src/admin/handlers/tier.rs +++ b/rustfs/src/admin/handlers/tier.rs @@ -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; } _ => (), } diff --git a/rustfs/src/admin/handlers/user.rs b/rustfs/src/admin/handlers/user.rs index 500c9de35..1cdb6eeea 100644 --- a/rustfs/src/admin/handlers/user.rs +++ b/rustfs/src/admin/handlers/user.rs @@ -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)) } diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 7ae07d2e5..7578dc940 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -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); diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index b1b1dc67f..1c1c3a5d0 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -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(), };