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 <qiuzgang@gmail.com>

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;
 

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<dyn Error + Send + Sync>> {
+    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(())
+}

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() {

rustfs/src/error.rs

@@ -112,6 +112,14 @@ fn custom_error_status(code: &S3ErrorCode) -> Option<StatusCode> {
     }
 }
 
+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 {

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<String, String>,
 }
 
 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"));