sdgoij/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-01-17 01:30:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: adapt to s3s typed etag support (#653)

Browse Source 
* feat:  adapt to s3s typed etag support

* refactor: streamline etag handling
This commit is contained in:
安正超
2025-10-15 21:27:20 +08:00
committed by GitHub
parent 8f310cd4a8
commit d447b3e426
6 changed files with 65 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
Cargo.lock generated
View File

@@ -7200,8 +7200,7 @@ checksum = "28d3b2b1366ec20994f1fd18c3c594f05c5dd4bc44d8bb0c1c632c8d6829481f"
[[package]]
name = "s3s"
version = "0.12.0-rc.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6fbf35b8b229398629d8df8a0cb908a70a24da69f65911fe0461fba3f4e9dda5"
source = "git+https://github.com/s3s-project/s3s.git?rev=f76013a52de2c8d44f6aa43182f1f1a938d2ad03#f76013a52de2c8d44f6aa43182f1f1a938d2ad03"
dependencies = [
"arrayvec",
"async-trait",

2
Cargo.toml
View File

@@ -214,7 +214,7 @@ rustc-hash = { version = "2.1.1" }
rustls = { version = "0.23.32", features = ["ring", "logging", "std", "tls12"], default-features = false }
rustls-pki-types = "1.12.0"
rustls-pemfile = "2.2.0"
s3s = { version = "0.12.0-rc.2", features = ["minio"] }
s3s = { git = "https://github.com/s3s-project/s3s.git", rev = "f76013a52de2c8d44f6aa43182f1f1a938d2ad03", features = ["minio"] }
schemars = "1.0.4"
serde = { version = "1.0.228", features = ["derive"] }
serde_json = { version = "1.0.145", features = ["raw_value"] }

83
crates/ecstore/src/client/object_api_utils.rs
View File

@@ -20,8 +20,8 @@
#![allow(clippy::all)]
use http::HeaderMap;
use std::io::Cursor;
use std::{collections::HashMap, sync::Arc};
use s3s::dto::ETag;
use std::{collections::HashMap, io::Cursor, sync::Arc};
use tokio::io::BufReader;
use crate::error::ErrorResponse;
@@ -148,27 +148,30 @@ pub fn new_getobjectreader(
Ok((get_fn, off as i64, length as i64))
}
/// Format an ETag value according to HTTP standards (wrap with quotes if not already wrapped)
pub fn format_etag(etag: &str) -> String {
if etag.starts_with('"') && etag.ends_with('"') {
// Already properly formatted
etag.to_string()
} else if etag.starts_with("W/\"") && etag.ends_with('"') {
// Already a weak ETag, properly formatted
etag.to_string()
} else {
// Need to wrap with quotes
format!("\"{}\"", etag)
/// Convert a raw stored ETag into the strongly-typed `s3s::dto::ETag`.
///
/// Supports already quoted (`"abc"`), weak (`W/"abc"`), or plain (`abc`) values.
pub fn to_s3s_etag(etag: &str) -> ETag {
if let Some(rest) = etag.strip_prefix("W/\"") {
if let Some(body) = rest.strip_suffix('"') {
return ETag::Weak(body.to_string());
}
return ETag::Weak(rest.to_string());
}
if let Some(body) = etag.strip_prefix('"').and_then(|rest| rest.strip_suffix('"')) {
return ETag::Strong(body.to_string());
}
ETag::Strong(etag.to_string())
}
pub fn extract_etag(metadata: &HashMap<String, String>) -> String {
let etag = if let Some(etag) = metadata.get("etag") {
etag.clone()
} else {
metadata["md5Sum"].clone()
};
format_etag(&etag)
pub fn get_raw_etag(metadata: &HashMap<String, String>) -> String {
metadata
.get("etag")
.cloned()
.or_else(|| metadata.get("md5Sum").cloned())
.unwrap_or_default()
}
#[cfg(test)]
@@ -176,30 +179,28 @@ mod tests {
use super::*;
#[test]
fn test_format_etag() {
// Test unquoted ETag - should add quotes
assert_eq!(format_etag("6af8d12c0c74b78094884349f3c8a079"), "\"6af8d12c0c74b78094884349f3c8a079\"");
// Test already quoted ETag - should not double quote
fn test_to_s3s_etag() {
// Test unquoted ETag - should become strong etag
assert_eq!(
format_etag("\"6af8d12c0c74b78094884349f3c8a079\""),
"\"6af8d12c0c74b78094884349f3c8a079\""
to_s3s_etag("6af8d12c0c74b78094884349f3c8a079"),
ETag::Strong("6af8d12c0c74b78094884349f3c8a079".to_string())
);
// Test weak ETag - should keep as is
assert_eq!(
format_etag("W/\"6af8d12c0c74b78094884349f3c8a079\""),
"W/\"6af8d12c0c74b78094884349f3c8a079\""
to_s3s_etag("\"6af8d12c0c74b78094884349f3c8a079\""),
ETag::Strong("6af8d12c0c74b78094884349f3c8a079".to_string())
);
// Test empty ETag - should add quotes
assert_eq!(format_etag(""), "\"\"");
assert_eq!(
to_s3s_etag("W/\"6af8d12c0c74b78094884349f3c8a079\""),
ETag::Weak("6af8d12c0c74b78094884349f3c8a079".to_string())
);
// Test malformed quote (only starting quote) - should wrap properly
assert_eq!(format_etag("\"incomplete"), "\"\"incomplete\"");
assert_eq!(to_s3s_etag(""), ETag::Strong(String::new()));
// Test malformed quote (only ending quote) - should wrap properly
assert_eq!(format_etag("incomplete\""), "\"incomplete\"\"");
assert_eq!(to_s3s_etag("\"incomplete"), ETag::Strong("\"incomplete".to_string()));
assert_eq!(to_s3s_etag("incomplete\""), ETag::Strong("incomplete\"".to_string()));
}
#[test]
@@ -208,15 +209,17 @@ mod tests {
// Test with etag field
metadata.insert("etag".to_string(), "abc123".to_string());
assert_eq!(extract_etag(&metadata), "\"abc123\"");
assert_eq!(get_raw_etag(&metadata), "abc123");
// Test with already quoted etag field
metadata.insert("etag".to_string(), "\"def456\"".to_string());
assert_eq!(extract_etag(&metadata), "\"def456\"");
assert_eq!(get_raw_etag(&metadata), "\"def456\"");
// Test fallback to md5Sum
metadata.remove("etag");
metadata.insert("md5Sum".to_string(), "xyz789".to_string());
assert_eq!(extract_etag(&metadata), "\"xyz789\"");
assert_eq!(get_raw_etag(&metadata), "xyz789");
metadata.clear();
assert_eq!(get_raw_etag(&metadata), "");
}
}

4
crates/ecstore/src/set_disk.rs
View File

@@ -21,7 +21,7 @@ use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
use crate::bucket::replication::check_replicate_delete;
use crate::bucket::versioning::VersioningApi;
use crate::bucket::versioning_sys::BucketVersioningSys;
use crate::client::{object_api_utils::extract_etag, transition_api::ReaderImpl};
use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl};
use crate::disk::STORAGE_FORMAT_FILE;
use crate::disk::error_reduce::{OBJECT_OP_IGNORED_ERRS, reduce_read_quorum_errs, reduce_write_quorum_errs};
use crate::disk::{
@@ -4581,7 +4581,7 @@ impl StorageAPI for SetDisks {
}*/
// Normalize ETags by removing quotes before comparison (PR #592 compatibility)
let transition_etag = rustfs_utils::path::trim_etag(&opts.transition.etag);
let stored_etag = rustfs_utils::path::trim_etag(&extract_etag(&fi.metadata));
let stored_etag = rustfs_utils::path::trim_etag(&get_raw_etag(&fi.metadata));
if !opts.mod_time.expect("err").unix_timestamp() == fi.mod_time.as_ref().expect("err").unix_timestamp()
|| transition_etag != stored_etag
{

7
crates/ecstore/src/store_api.rs
View File

@@ -35,6 +35,7 @@ use rustfs_rio::{DecompressReader, HashReader, LimitReader, WarpReader};
use rustfs_utils::CompressionAlgorithm;
use rustfs_utils::http::headers::{AMZ_OBJECT_TAGGING, RESERVED_METADATA_PREFIX_LOWER};
use rustfs_utils::path::decode_dir_object;
use s3s::dto::ETag;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fmt::Debug;
@@ -458,9 +459,13 @@ pub struct CompletePart {
impl From<s3s::dto::CompletedPart> for CompletePart {
fn from(value: s3s::dto::CompletedPart) -> Self {
let etag = value.e_tag.map(|etag| match etag {
ETag::Strong(v) => format!("\"{v}\""),
ETag::Weak(v) => format!("W/\"{v}\""),
});
Self {
part_num: value.part_number.unwrap_or_default() as usize,
etag: value.e_tag,
etag,
}
}
}

24
rustfs/src/storage/ecfs.rs
View File

@@ -49,7 +49,7 @@ use rustfs_ecstore::{
versioning::VersioningApi,
versioning_sys::BucketVersioningSys,
},
client::object_api_utils::format_etag,
client::object_api_utils::to_s3s_etag,
compress::{MIN_COMPRESSIBLE_SIZE, is_compressible},
disk::{error::DiskError, error_reduce::is_all_buckets_not_found},
error::{StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found},
@@ -437,7 +437,7 @@ impl FS {
.await
.map_err(ApiError::from)?;
let e_tag = _obj_info.etag.clone().map(|etag| format_etag(&etag));
let e_tag = _obj_info.etag.clone().map(|etag| to_s3s_etag(&etag));
// // store.put_object(bucket, object, data, opts);
@@ -725,7 +725,7 @@ impl S3 for FS {
// warn!("copy_object oi {:?}", &oi);
let object_info = oi.clone();
let copy_object_result = CopyObjectResult {
e_tag: oi.etag.map(|etag| format_etag(&etag)),
e_tag: oi.etag.map(|etag| to_s3s_etag(&etag)),
last_modified: oi.mod_time.map(Timestamp::from),
..Default::default()
};
@@ -1670,7 +1670,7 @@ impl S3 for FS {
content_type,
accept_ranges: Some("bytes".to_string()),
content_range,
e_tag: info.etag.map(|etag| format_etag(&etag)),
e_tag: info.etag.map(|etag| to_s3s_etag(&etag)),
metadata: Some(info.user_defined),
server_side_encryption,
sse_customer_algorithm,
@@ -1804,7 +1804,7 @@ impl S3 for FS {
content_length: Some(content_length),
content_type,
last_modified,
e_tag: info.etag.map(|etag| format_etag(&etag)),
e_tag: info.etag.map(|etag| to_s3s_etag(&etag)),
metadata: Some(metadata),
version_id: info.version_id.map(|v| v.to_string()),
server_side_encryption,
@@ -1955,7 +1955,7 @@ impl S3 for FS {
key: Some(v.name.to_owned()),
last_modified: v.mod_time.map(Timestamp::from),
size: Some(v.get_actual_size().unwrap_or_default()),
e_tag: v.etag.clone().map(|etag| format_etag(&etag)),
e_tag: v.etag.clone().map(|etag| to_s3s_etag(&etag)),
..Default::default()
};
@@ -2034,7 +2034,7 @@ impl S3 for FS {
size: Some(v.size),
version_id: v.version_id.map(|v| v.to_string()),
is_latest: Some(v.is_latest),
e_tag: v.etag.clone().map(|etag| format_etag(&etag)),
e_tag: v.etag.clone().map(|etag| to_s3s_etag(&etag)),
..Default::default() // TODO: another fields
}
})
@@ -2311,7 +2311,7 @@ impl S3 for FS {
.await
.map_err(ApiError::from)?;
let event_info = obj_info.clone();
let e_tag = obj_info.etag.clone().map(|etag| format_etag(&etag));
let e_tag = obj_info.etag.clone().map(|etag| to_s3s_etag(&etag));
let repoptions =
get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts);
@@ -2674,7 +2674,7 @@ impl S3 for FS {
.map_err(ApiError::from)?;
let output = UploadPartOutput {
e_tag: info.etag.map(|etag| format_etag(&etag)),
e_tag: info.etag.map(|etag| to_s3s_etag(&etag)),
..Default::default()
};
@@ -2860,7 +2860,7 @@ impl S3 for FS {
// Create response
let copy_part_result = CopyPartResult {
e_tag: part_info.etag.map(|etag| format_etag(&etag)),
e_tag: part_info.etag.map(|etag| to_s3s_etag(&etag)),
last_modified: part_info.last_mod.map(Timestamp::from),
..Default::default()
};
@@ -2913,7 +2913,7 @@ impl S3 for FS {
res.parts
.into_iter()
.map(|p| Part {
e_tag: p.etag.map(|etag| format_etag(&etag)),
e_tag: p.etag.map(|etag| to_s3s_etag(&etag)),
last_modified: p.last_mod.map(Timestamp::from),
part_number: Some(p.part_num as i32),
size: Some(p.size as i64),
@@ -3088,7 +3088,7 @@ impl S3 for FS {
let output = CompleteMultipartUploadOutput {
bucket: Some(bucket.clone()),
key: Some(key.clone()),
e_tag: obj_info.etag.clone().map(|etag| format_etag(&etag)),
e_tag: obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)),
location: Some("us-east-1".to_string()),
server_side_encryption, // TDD: Return encryption info
ssekms_key_id, // TDD: Return KMS key ID if present