From 8f6d06db21f456a10945daf89f14c25d11998669 Mon Sep 17 00:00:00 2001 From: Hauser Date: Sun, 13 Sep 2026 22:57:25 +0800 Subject: [PATCH] fix(s3): classify copy source part read failures (#7746) Map PartMissingOrCorrupt to SlowDownRead only at the GetObject/CopyObject source-reader boundary so quota metadata corruption keeps its internal fail-closed response. Add store and e2e coverage for Harbor-style multipart staging CopyObject boundaries. Co-authored-by: zhi22915 --- crates/e2e_test/src/lib.rs | 3 + .../src/multipart_copy_readiness_test.rs | 164 ++++++++++++++++++ crates/ecstore/src/store/init.rs | 45 +++++ rustfs/src/error.rs | 16 ++ rustfs/src/storage/sse.rs | 14 +- 5 files changed, 241 insertions(+), 1 deletion(-) create mode 100644 crates/e2e_test/src/multipart_copy_readiness_test.rs diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index a9d35da00..b741be195 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -254,6 +254,9 @@ mod copy_object_version_restore_test; #[cfg(test)] mod copy_object_checksum_test; +#[cfg(test)] +mod multipart_copy_readiness_test; + #[cfg(test)] mod ssec_copy_test; diff --git a/crates/e2e_test/src/multipart_copy_readiness_test.rs b/crates/e2e_test/src/multipart_copy_readiness_test.rs new file mode 100644 index 000000000..a668539b7 --- /dev/null +++ b/crates/e2e_test/src/multipart_copy_readiness_test.rs @@ -0,0 +1,164 @@ +// Copyright 2026 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. + +//! Harbor / Docker Distribution multipart staging to CopyObject regressions. + +use crate::common::{RustFSTestEnvironment, init_logging}; +use aws_sdk_s3::error::ProvideErrorMetadata; +use aws_sdk_s3::primitives::ByteStream; +use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart, StorageClass}; +use std::error::Error; +use tracing::info; + +const BUCKET: &str = "multipart-copy-readiness"; +const SOURCE_KEY: &str = "docker/registry/v2/repositories/example/_uploads/upload-id/data"; +const TARGET_KEY: &str = "docker/registry/v2/blobs/sha256/c0/digest/data"; +const UNFINISHED_SOURCE_KEY: &str = "docker/registry/v2/repositories/example/_uploads/upload-id-unfinished/data"; +const UNFINISHED_TARGET_KEY: &str = "docker/registry/v2/blobs/sha256/c1/digest/data"; + +fn list_contains_key(output: &aws_sdk_s3::operation::list_objects_v2::ListObjectsV2Output, key: &str) -> bool { + output + .contents() + .iter() + .any(|object| object.key().is_some_and(|candidate| candidate == key)) +} + +#[tokio::test] +async fn harbor_style_multipart_staging_copy_object_boundaries() -> Result<(), Box> { + init_logging(); + info!("backlog#2185: Harbor-style one-part MPU staging followed by CopyObject"); + + let mut env = RustFSTestEnvironment::new().await?; + env.start_rustfs_server(vec![]).await?; + + let client = env.create_s3_client(); + env.create_test_bucket(BUCKET).await?; + + let payload = vec![0xAB; 273]; + let create = client.create_multipart_upload().bucket(BUCKET).key(SOURCE_KEY).send().await?; + let upload_id = create.upload_id().ok_or("missing upload id")?.to_string(); + let part = client + .upload_part() + .bucket(BUCKET) + .key(SOURCE_KEY) + .upload_id(&upload_id) + .part_number(1) + .body(ByteStream::from(payload.clone())) + .send() + .await?; + let completed = CompletedMultipartUpload::builder() + .parts( + CompletedPart::builder() + .part_number(1) + .set_e_tag(part.e_tag().map(str::to_string)) + .build(), + ) + .build(); + client + .complete_multipart_upload() + .bucket(BUCKET) + .key(SOURCE_KEY) + .upload_id(&upload_id) + .multipart_upload(completed) + .send() + .await?; + + let completed_list = client + .list_objects_v2() + .bucket(BUCKET) + .prefix(SOURCE_KEY) + .max_keys(1) + .send() + .await?; + assert!( + list_contains_key(&completed_list, SOURCE_KEY), + "completed multipart staging object must be immediately list-visible" + ); + + client + .copy_object() + .bucket(BUCKET) + .key(TARGET_KEY) + .copy_source(format!("/{BUCKET}/{SOURCE_KEY}")) + .storage_class(StorageClass::Standard) + .send() + .await?; + + let copied = client.get_object().bucket(BUCKET).key(TARGET_KEY).send().await?; + let copied_body = copied.body.collect().await?.into_bytes(); + assert_eq!(copied_body.as_ref(), payload.as_slice()); + + let unfinished = client + .create_multipart_upload() + .bucket(BUCKET) + .key(UNFINISHED_SOURCE_KEY) + .send() + .await?; + let unfinished_upload_id = unfinished.upload_id().ok_or("missing unfinished upload id")?.to_string(); + client + .upload_part() + .bucket(BUCKET) + .key(UNFINISHED_SOURCE_KEY) + .upload_id(&unfinished_upload_id) + .part_number(1) + .body(ByteStream::from_static(b"not-yet-committed")) + .send() + .await?; + + let unfinished_list = client + .list_objects_v2() + .bucket(BUCKET) + .prefix(UNFINISHED_SOURCE_KEY) + .max_keys(1) + .send() + .await?; + assert!( + !list_contains_key(&unfinished_list, UNFINISHED_SOURCE_KEY), + "UploadPart alone must not publish the staging object namespace entry" + ); + + let copy_err = client + .copy_object() + .bucket(BUCKET) + .key(UNFINISHED_TARGET_KEY) + .copy_source(format!("/{BUCKET}/{UNFINISHED_SOURCE_KEY}")) + .storage_class(StorageClass::Standard) + .send() + .await + .expect_err("copying an uncompleted multipart upload source must be rejected"); + assert_eq!( + copy_err.raw_response().map(|response| response.status().as_u16()), + Some(404), + "unfinished MPU source must not leak a 5xx response: {copy_err:?}" + ); + assert_eq!( + copy_err.as_service_error().and_then(ProvideErrorMetadata::code), + Some("NoSuchKey"), + "unfinished MPU source must be reported as a missing ordinary object: {copy_err:?}" + ); + + client + .abort_multipart_upload() + .bucket(BUCKET) + .key(UNFINISHED_SOURCE_KEY) + .upload_id(unfinished_upload_id) + .send() + .await?; + client.delete_object().bucket(BUCKET).key(TARGET_KEY).send().await?; + client.delete_object().bucket(BUCKET).key(SOURCE_KEY).send().await?; + env.delete_test_bucket(BUCKET).await?; + env.stop_server(); + + Ok(()) +} diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index a95871805..51d34b3ea 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -3135,6 +3135,51 @@ mod tests { shutdown.cancel(); } + #[cfg(feature = "test-util")] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[serial_test::serial(storage_class_env)] + async fn unfinished_multipart_upload_is_not_copy_source_readable() { + let temp_dir = tempfile::tempdir().expect("create unfinished multipart copy store dir"); + let (_ctx, store, shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "unfinished-multipart-copy", &[1])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await; + + let bucket = format!("unfinished-multipart-copy-{}", Uuid::new_v4()); + let source_object = "docker/registry/v2/repositories/example/_uploads/upload-id/data"; + let payload = vec![0xCD; 273]; + + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create bucket for unfinished multipart copy source"); + let upload = store + .new_multipart_upload(&bucket, source_object, &ObjectOptions::default()) + .await + .expect("create source multipart upload"); + let mut part_reader = PutObjReader::from_vec(payload); + store + .put_object_part(&bucket, source_object, &upload.upload_id, 1, &mut part_reader, &ObjectOptions::default()) + .await + .expect("stage unfinished multipart source part"); + + let source_err = match store + .get_object_reader(&bucket, source_object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + { + Ok(_) => panic!("an uncompleted multipart upload must not be readable as a CopyObject source"), + Err(err) => err, + }; + assert!( + matches!( + source_err, + StorageError::ObjectNotFound(_, _) | StorageError::FileNotFound | StorageError::VersionNotFound(_, _, _) + ), + "unexpected unfinished multipart source error: {source_err:?}" + ); + + shutdown.cancel(); + } + #[tokio::test] #[serial_test::serial(storage_class_env)] async fn pool_metadata_preflight_recovery_preserves_single_and_multi_pool_public_mutations() { diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index 9531f795e..cd08f0f58 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -112,6 +112,14 @@ fn custom_error_status(code: &S3ErrorCode) -> Option { } } +pub(crate) fn slow_down_read_api_error(err: StorageError) -> ApiError { + ApiError { + code: S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into()), + message: SLOW_DOWN_READ_MESSAGE.to_string(), + source: Some(Box::new(err)), + } +} + /// Marks a request body that exceeded a presigned upload size capability. /// /// This marker must survive the body-reader and storage layers so the client @@ -1219,6 +1227,14 @@ mod tests { assert_eq!(s3_error.status_code(), Some(StatusCode::SERVICE_UNAVAILABLE)); } + #[test] + fn test_slow_down_read_api_error_maps_to_retryable_status() { + let api_error = slow_down_read_api_error(StorageError::PartMissingOrCorrupt); + + assert_eq!(api_error.code, S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into())); + assert_eq!(S3Error::from(api_error).status_code(), Some(StatusCode::SERVICE_UNAVAILABLE)); + } + #[test] fn test_unknown_authoritative_quota_usage_maps_to_retryable_error() { let api_error = ApiError::from(QuotaError::UsageUnavailable { diff --git a/rustfs/src/storage/sse.rs b/rustfs/src/storage/sse.rs index c47e267f1..0b9404cde 100644 --- a/rustfs/src/storage/sse.rs +++ b/rustfs/src/storage/sse.rs @@ -159,7 +159,7 @@ fn md5_base64(input: impl AsRef<[u8]>) -> String { use super::Error; use super::get_bucket_sse_config; -use crate::error::ApiError; +use crate::error::{ApiError, slow_down_read_api_error}; use rustfs_utils::http::headers::{ AMZ_ENCRYPTION_AES, AMZ_ENCRYPTION_KMS, AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY, AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, AMZ_SERVER_SIDE_ENCRYPTION_KMS_CONTEXT, @@ -711,6 +711,10 @@ pub(crate) fn validate_sse_headers_for_read(metadata: &HashMap, } pub(crate) fn map_get_object_reader_error(err: StorageError) -> ApiError { + if matches!(err, StorageError::PartMissingOrCorrupt) { + return slow_down_read_api_error(err); + } + if let StorageError::Io(io_error) = &err && let Some(resolution_error) = io_error .get_ref() @@ -7524,6 +7528,14 @@ mod tests { assert_eq!(err.message, "KMS unavailable"); } + #[test] + fn test_map_get_object_reader_error_maps_part_missing_to_slow_down_read() { + let err = map_get_object_reader_error(StorageError::PartMissingOrCorrupt); + + assert_eq!(err.code, S3ErrorCode::Custom("SlowDownRead".into())); + assert_eq!(err.message, "Resource requested is unreadable, please reduce your request rate"); + } + #[test] fn test_map_get_object_reader_error_redacts_non_ssec_internal_errors() { let err = map_get_object_reader_error(StorageError::other("plain io failure"));