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

refactor(app): reuse bucket listing helpers (#2449)

Browse Source
This commit is contained in:
安正超
2026-04-10 07:06:11 +08:00
committed by GitHub
parent 722cab500c
commit a9507f8ec7
1 changed files with 99 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/app/bucket_usecase.rs
+99 -119
View File
@@ -23,7 +23,10 @@ use crate::error::ApiError;
use crate::server::RemoteAddr;
use crate::storage::access::{ReqInfo, authorize_request, req_info_ref};
use crate::storage::helper::{OperationHelper, spawn_background_with_context};
use crate::storage::s3_api::bucket::{build_list_buckets_output, build_list_objects_v2_output};
use crate::storage::s3_api::bucket::{
ListObjectVersionsParams, ListObjectsV2Params, build_list_buckets_output, build_list_object_versions_output,
build_list_objects_v2_output, parse_list_object_versions_params, parse_list_objects_v2_params,
};
use crate::storage::s3_api::common::rustfs_owner;
use crate::storage::s3_api::{acl, encryption, replication, tagging};
use crate::storage::*;
@@ -1841,43 +1844,18 @@ impl DefaultBucketUsecase {
..
} = req.input;
let prefix = prefix.unwrap_or_default();
// Log debug info for prefixes with special characters to help diagnose encoding issues
if prefix.contains([' ', '+', '%', '\n', '\r', '\0']) {
debug!("LIST objects with special characters in prefix: {:?}", prefix);
}
let max_keys = max_keys.unwrap_or(1000);
if max_keys < 0 {
return Err(S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid max keys".to_string()));
}
let delimiter = delimiter.filter(|v| !v.is_empty());
let ListObjectsV2Params {
prefix,
max_keys,
delimiter,
response_start_after,
start_after_for_query,
response_continuation_token,
decoded_continuation_token,
} = parse_list_objects_v2_params(prefix, delimiter, max_keys, continuation_token, start_after)?;
validate_list_object_unordered_with_delimiter(delimiter.as_ref(), req.uri.query())?;
// Save original start_after for response (per S3 API spec, must echo back if provided)
let response_start_after = start_after.clone();
let start_after_for_query = start_after.filter(|v| !v.is_empty());
// Save original continuation_token for response (per S3 API spec, must echo back if provided)
// Note: empty string should still be echoed back in the response
let response_continuation_token = continuation_token.clone();
let continuation_token_for_query = continuation_token.filter(|v| !v.is_empty());
// Decode continuation_token from base64 for internal use
let decoded_continuation_token = continuation_token_for_query
.map(|token| {
base64_simd::STANDARD
.decode_to_vec(token.as_bytes())
.map_err(|_| s3_error!(InvalidArgument, "Invalid continuation token"))
.and_then(|bytes| {
String::from_utf8(bytes).map_err(|_| s3_error!(InvalidArgument, "Invalid continuation token"))
})
})
.transpose()?;
let store = get_validated_store(&bucket).await?;
let incl_deleted = get_header(&req.headers, rustfs_utils::http::SUFFIX_INCLUDE_DELETED)
@@ -1934,31 +1912,18 @@ impl DefaultBucketUsecase {
..
} = input;
let prefix = prefix.unwrap_or_default();
let max_keys = max_keys.unwrap_or(1000);
if max_keys < 0 {
return Err(S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid max keys".to_string()));
}
let ListObjectsV2Params {
prefix,
max_keys,
delimiter,
response_start_after,
start_after_for_query,
response_continuation_token,
decoded_continuation_token,
} = parse_list_objects_v2_params(prefix, delimiter, max_keys, continuation_token, start_after)?;
let delimiter = delimiter.filter(|value| !value.is_empty());
validate_list_object_unordered_with_delimiter(delimiter.as_ref(), req.uri.query())?;
let response_start_after = start_after.clone();
let start_after_for_query = start_after.filter(|value| !value.is_empty());
let response_continuation_token = continuation_token.clone();
let continuation_token_for_query = continuation_token.filter(|value| !value.is_empty());
let decoded_continuation_token = continuation_token_for_query
.map(|token| {
base64_simd::STANDARD
.decode_to_vec(token.as_bytes())
.map_err(|_| s3_error!(InvalidArgument, "Invalid continuation token"))
.and_then(|bytes| {
String::from_utf8(bytes).map_err(|_| s3_error!(InvalidArgument, "Invalid continuation token"))
})
})
.transpose()?;
let store = get_validated_store(&bucket).await?;
let incl_deleted = get_header(&req.headers, rustfs_utils::http::SUFFIX_INCLUDE_DELETED)
.map(|value| value.as_ref() == "true")
@@ -2012,12 +1977,13 @@ impl DefaultBucketUsecase {
..
} = req.input;
let prefix = prefix.unwrap_or_default();
let max_keys = max_keys.unwrap_or(1000);
let key_marker = key_marker.filter(|v| !v.is_empty());
let version_id_marker = version_id_marker.filter(|v| !v.is_empty());
let delimiter = delimiter.filter(|v| !v.is_empty());
let ListObjectVersionsParams {
prefix,
delimiter,
key_marker,
version_id_marker,
max_keys,
} = parse_list_object_versions_params(prefix, delimiter, key_marker, version_id_marker, max_keys)?;
let store = get_validated_store(&bucket).await?;
@@ -2026,57 +1992,7 @@ impl DefaultBucketUsecase {
.await
.map_err(ApiError::from)?;
let objects: Vec<ObjectVersion> = object_infos
.objects
.iter()
.filter(|v| !v.name.is_empty() && !v.delete_marker)
.map(|v| ObjectVersion {
key: Some(v.name.to_owned()),
last_modified: v.mod_time.map(Timestamp::from),
size: Some(v.size),
version_id: Some(v.version_id.map(|v| v.to_string()).unwrap_or_else(|| "null".to_string())),
is_latest: Some(v.is_latest),
e_tag: v.etag.clone().map(|etag| to_s3s_etag(&etag)),
storage_class: v.storage_class.clone().map(ObjectVersionStorageClass::from),
..Default::default()
})
.collect();
let common_prefixes = object_infos
.prefixes
.into_iter()
.map(|v| CommonPrefix { prefix: Some(v) })
.collect();
let delete_markers = object_infos
.objects
.iter()
.filter(|o| o.delete_marker)
.map(|o| DeleteMarkerEntry {
key: Some(o.name.clone()),
version_id: Some(o.version_id.map(|v| v.to_string()).unwrap_or_else(|| "null".to_string())),
is_latest: Some(o.is_latest),
last_modified: o.mod_time.map(Timestamp::from),
..Default::default()
})
.collect::<Vec<_>>();
let next_key_marker = object_infos.next_marker.filter(|v| !v.is_empty());
let next_version_id_marker = object_infos.next_version_idmarker.filter(|v| !v.is_empty());
let output = ListObjectVersionsOutput {
is_truncated: Some(object_infos.is_truncated),
max_keys: Some(max_keys),
delimiter,
name: Some(bucket),
prefix: Some(prefix),
common_prefixes: Some(common_prefixes),
versions: Some(objects),
delete_markers: Some(delete_markers),
next_key_marker,
next_version_id_marker,
..Default::default()
};
let output = build_list_object_versions_output(object_infos, bucket, prefix, delimiter, max_keys);
Ok(S3Response::new(output))
}
@@ -2101,11 +2017,13 @@ impl DefaultBucketUsecase {
..
} = input;
let prefix = prefix.unwrap_or_default();
let max_keys = max_keys.unwrap_or(1000);
let key_marker = key_marker.filter(|value| !value.is_empty());
let version_id_marker = version_id_marker.filter(|value| !value.is_empty());
let delimiter = delimiter.filter(|value| !value.is_empty());
let ListObjectVersionsParams {
prefix,
delimiter,
key_marker,
version_id_marker,
max_keys,
} = parse_list_object_versions_params(prefix, delimiter, key_marker, version_id_marker, max_keys)?;
let store = get_validated_store(&bucket).await?;
let object_infos = store
@@ -2953,6 +2871,22 @@ mod tests {
assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
}
#[tokio::test]
async fn execute_list_objects_v2_rejects_invalid_continuation_token_before_store_lookup() {
let input = ListObjectsV2Input::builder()
.bucket("test-bucket".to_string())
.continuation_token(Some("%%%".to_string()))
.build()
.unwrap();
let req = build_request(input, Method::GET);
let usecase = DefaultBucketUsecase::without_context();
let err = usecase.execute_list_objects_v2(req).await.unwrap_err();
assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
assert_eq!(err.message(), Some("Invalid continuation token"));
}
#[tokio::test]
async fn execute_list_objects_v2m_rejects_negative_max_keys() {
let input = ListObjectsV2Input::builder()
@@ -2967,4 +2901,50 @@ mod tests {
let err = usecase.execute_list_objects_v2m(req).await.unwrap_err();
assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
}
#[tokio::test]
async fn execute_list_objects_v2m_rejects_invalid_continuation_token_before_store_lookup() {
let input = ListObjectsV2Input::builder()
.bucket("test-bucket".to_string())
.continuation_token(Some("%%%".to_string()))
.build()
.unwrap();
let req = build_request(input, Method::GET);
let usecase = DefaultBucketUsecase::without_context();
let err = usecase.execute_list_objects_v2m(req).await.unwrap_err();
assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
assert_eq!(err.message(), Some("Invalid continuation token"));
}
#[tokio::test]
async fn execute_list_object_versions_rejects_negative_max_keys_before_store_lookup() {
let input = ListObjectVersionsInput::builder()
.bucket("test-bucket".to_string())
.max_keys(Some(-1))
.build()
.unwrap();
let req = build_request(input, Method::GET);
let usecase = DefaultBucketUsecase::without_context();
let err = usecase.execute_list_object_versions(req).await.unwrap_err();
assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
}
#[tokio::test]
async fn execute_list_object_versions_m_rejects_negative_max_keys_before_store_lookup() {
let input = ListObjectVersionsInput::builder()
.bucket("test-bucket".to_string())
.max_keys(Some(-1))
.build()
.unwrap();
let req = build_request(input, Method::GET);
let usecase = DefaultBucketUsecase::without_context();
let err = usecase.execute_list_object_versions_m(req).await.unwrap_err();
assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
}
}