fix(s3): preserve typed upload errors across body readers (#8816)

* fix(s3): report payload SHA256 mismatches accurately

Distinguish signed payload SHA256 failures from MD5 and additional
checksum failures at the shared API error boundary. Preserve the
payload error through archive extraction and use typed s3s errors.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* refactor(s3): classify body failures in one bounded pass

Share typed body-error classification across direct and storage I/O.
Retain marker precedence while replacing repeated source-chain scans,
and map storage-wrapped short HashReader bodies to IncompleteBody.

Cover nested priorities, real short reads, and one traversal budget.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(s3): route upload stream errors through the app facade

Keep the real stream-error fixtures behind the existing test-only S3
boundary without increasing the direct s3s import footprint.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(s3): preserve typed body failures across archive reads

Route TAR entry failures through the shared archive error adapter and
preserve every classified body error. Keep generic archive parsing as
InvalidArgument without formatting discarded transport diagnostics.

Use the existing timeout marker for PUT and Snowball body guards. Add
regressions for real stream verification after sniffing and timeout
responses, and tighten the reduced S3 macro footprint baseline.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/e2e_test/src/negative_sigv4_test.rs

@@ -566,47 +566,54 @@ async fn tampered_payload_is_rejected() -> Result<(), Box<dyn std::error::Error
     let mut env = RustFSTestEnvironment::new().await?;
     setup(&mut env).await?;
 
-    let key = "tampered-payload.txt";
-    let path = format!("/{BUCKET}/{key}");
-    let claimed_body = b"the-body-i-claim-to-send";
-    let actual_body = b"the-body-i-really-send!!";
-    assert_eq!(claimed_body.len(), actual_body.len(), "keep content-length stable for the mismatch");
+    use md5::{Digest, Md5};
 
     let signer = SigV4::new(&env);
-    // Sign over the hash of the CLAIMED body (single-chunk payload hash), then
-    // send a different body of equal length.
-    let headers = signer.sign("PUT", &path, "", &sha256_hex(claimed_body));
-
-    let result = send_signed(&env, reqwest::Method::PUT, &path, &headers, Some(actual_body.to_vec())).await;
-    match result {
-        Ok(resp) => {
-            let status = resp.status();
-            let body = resp.text().await.unwrap_or_default();
+    let client = env.create_s3_client();
+    // Empty, small-eager, zero-copy, and streaming PUTs must share the error contract.
+    for size in [0, 24, 2 * 1024 * 1024, 17 * 1024 * 1024] {
+        let actual_body = vec![b'a'; size];
+        let wrong_hash = sha256_hex(b"a different payload");
+        let content_md5 = base64_simd::STANDARD.encode_to_string(Md5::digest(&actual_body));
+        for with_md5 in [false, true] {
+            let key = format!("tampered-payload-{size}-{with_md5}.txt");
+            let path = format!("/{BUCKET}/{key}");
+            let extra = if with_md5 {
+                vec![("content-md5", content_md5.as_str())]
+            } else {
+                vec![]
+            };
+            let headers = signer.sign_with_extra_headers("PUT", &path, "", &wrong_hash, &extra);
+            let mut request = local_http_client()
+                .put(format!("{}{}", env.url, path))
+                .header("x-amz-date", &headers.amz_date)
+                .header("x-amz-content-sha256", &headers.content_sha256)
+                .header("authorization", &headers.authorization);
+            for (name, value) in extra {
+                request = request.header(name, value);
+            }
+            let response = request.body(actual_body.clone()).send().await?;
+            let status = response.status();
+            let body = response.text().await?;
+            assert_eq!(status, reqwest::StatusCode::BAD_REQUEST, "payload mismatch: {body}");
+            assert_error_code(&body, "XAmzContentSHA256Mismatch");
             assert!(
-                status.is_client_error(),
-                "payload mismatch must be rejected with a client error, got {status}, body:\n{body}"
+                body.replace("&apos;", "'")
+                    .contains("<Message>The provided 'x-amz-content-sha256' header does not match what was computed.</Message>"),
+                "wrong message: {body}"
             );
-            info!(%status, "tampered payload rejected with error status");
-        }
-        // A mid-stream hash-mismatch abort surfacing as a transport error is
-        // also a valid rejection (definitely not a 200 success).
-        Err(err) => {
-            assert!(!err.is_connect(), "connection failure is not proof of payload rejection: {err}");
-            assert!(!err.is_timeout(), "request timeout is not proof of payload rejection: {err}");
-            info!(%err, "tampered payload rejected via mid-stream transport error");
+
+            let absent = client
+                .get_object()
+                .bucket(BUCKET)
+                .key(&key)
+                .send()
+                .await
+                .expect_err("a tampered payload must not publish an object");
+            assert_eq!(absent.raw_response().map(|response| response.status().as_u16()), Some(404));
+            assert_eq!(absent.as_service_error().and_then(ProvideErrorMetadata::code), Some("NoSuchKey"));
         }
     }
-
-    let absent = env
-        .create_s3_client()
-        .get_object()
-        .bucket(BUCKET)
-        .key(key)
-        .send()
-        .await
-        .expect_err("a tampered payload must not publish an object");
-    assert_eq!(absent.raw_response().map(|response| response.status().as_u16()), Some(404));
-    assert_eq!(absent.as_service_error().and_then(ProvideErrorMetadata::code), Some("NoSuchKey"));
     Ok(())
 }
 
@@ -638,9 +645,9 @@ async fn tampered_upload_part_payload_is_rejected() -> Result<(), Box<dyn std::e
     assert_eq!(
         status,
         reqwest::StatusCode::BAD_REQUEST,
-        "multipart payload mismatch must be rejected as BadDigest, body:\n{body}"
+        "multipart payload mismatch must be rejected as XAmzContentSHA256Mismatch, body:\n{body}"
     );
-    assert_error_code(&body, "BadDigest");
+    assert_error_code(&body, "XAmzContentSHA256Mismatch");
 
     let parts = client
         .list_parts()

rustfs/src/app/object/extract.rs

@@ -1208,13 +1208,12 @@ pub fn normalize_extract_entry_key(path: &str, prefix: Option<&str>, is_dir: boo
 }
 
 fn map_extract_archive_error(err: std::io::Error) -> S3Error {
-    let message = err.to_string();
     let api_error = ApiError::from(err);
-    if matches!(api_error.code, S3ErrorCode::BadDigest | S3ErrorCode::IncompleteBody) {
+    if api_error.code != S3ErrorCode::InternalError {
         return api_error.into();
     }
 
-    let mut archive_error = s3_error!(InvalidArgument, "Failed to process archive entry: {}", message);
+    let mut archive_error = s3_error!(InvalidArgument, "Failed to process archive entry: {}", api_error.message);
     archive_error.set_source(Box::new(api_error));
     archive_error
 }
@@ -2239,14 +2238,7 @@ impl DefaultObjectUsecase {
         }
 
         while let Some(entry) = entries.next().await {
-            let mut f = match entry {
-                Ok(f) => f,
-                Err(error) => {
-                    error!(error = %error, "Archive entry read failed");
-                    flush_extract_batch(&mut batch, &mut batch_state, &commit_context, extract_options.ignore_errors).await?;
-                    return Err(s3_error!(InvalidArgument, "Failed to read archive entry: {:?}", error));
-                }
-            };
+            let mut f = extract_try!(entry.map_err(map_extract_archive_error));
             extracted_entry_count = extracted_entry_count.saturating_add(1);
             extract_try!(validate_put_object_extract_entry_count(extracted_entry_count, extract_limits,));
             let entry_size = f.effective_size();
@@ -3475,6 +3467,123 @@ mod tests {
         }
     }
 
+    #[test]
+    fn archive_error_mapping_preserves_classified_body_errors() {
+        use crate::error::{ClientBodyReadTimeout, ServerSideSourceReadError, UploadLimitExceeded};
+        for (error, code) in [
+            (
+                io::Error::other(ClientBodyReadTimeout {
+                    timeout: Duration::from_secs(1),
+                    raw_bytes_received: 16,
+                }),
+                S3ErrorCode::RequestTimeout,
+            ),
+            (
+                io::Error::other(s3s::BodySizeLimitExceeded { size: 6, limit: 5 }),
+                S3ErrorCode::EntityTooLarge,
+            ),
+            (io::Error::other(UploadLimitExceeded { limit: 5 }), S3ErrorCode::EntityTooLarge),
+            (
+                io::Error::other(ServerSideSourceReadError::new("CopyObject", io::Error::other("source unavailable"))),
+                S3ErrorCode::ServiceUnavailable,
+            ),
+        ] {
+            let error = map_extract_archive_error(error);
+            assert_eq!(error.code(), &code);
+            assert_eq!(error.message(), Some(ApiError::error_code_to_message(&code).as_str()));
+            assert!(error.source().is_some());
+        }
+        let error = map_extract_archive_error(io::Error::new(io::ErrorKind::InvalidData, "invalid TAR header"));
+        assert_eq!(error.code(), &S3ErrorCode::InvalidArgument);
+        assert_eq!(error.message(), Some("Failed to process archive entry: invalid TAR header"));
+        assert!(error.source().is_some());
+    }
+
+    #[test]
+    fn archive_body_errors_do_not_format_discarded_diagnostics() {
+        #[derive(Debug)]
+        struct DisplayProbe {
+            source: io::Error,
+            displays: Arc<AtomicUsize>,
+        }
+        impl std::fmt::Display for DisplayProbe {
+            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                self.displays.fetch_add(1, Ordering::Relaxed);
+                f.write_str("archive transport diagnostics")
+            }
+        }
+        impl std::error::Error for DisplayProbe {
+            fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+                Some(&self.source)
+            }
+        }
+        let displays = Arc::new(AtomicUsize::new(0));
+        let error = map_extract_archive_error(io::Error::other(DisplayProbe {
+            source: io::Error::other(UploadStreamError::Sha256Mismatch),
+            displays: Arc::clone(&displays),
+        }));
+        assert_eq!(error.code().as_str(), "XAmzContentSHA256Mismatch");
+        assert_eq!(
+            displays.load(Ordering::Relaxed),
+            0,
+            "classified body failures need only their public message"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn snowball_extract_preserves_sha256_failure_after_sniffing() {
+        use crate::app::storage_api::s3::UploadStream;
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+        use http::Method;
+
+        let (store, context) = real_store_test_context().await;
+        let bucket = format!("extract-body-error-{}", Uuid::new_v4());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create extract test bucket");
+        let mut header = Header::new_gnu();
+        header.set_path("member.txt").expect("valid member path");
+        header.set_size(0);
+        header.set_mode(0o644);
+        header.set_cksum();
+        let mut archive = header.as_bytes().to_vec();
+        archive.extend_from_slice(&[0; 1024]);
+        let size = archive.len();
+        // Sniffing consumes only the prefix; the final frame fails while TAR
+        // reads its first header, before a member can be prepared or committed.
+        let frames = futures::stream::iter([
+            Ok::<_, StdError>(Bytes::copy_from_slice(&archive[..16])),
+            Ok(Bytes::copy_from_slice(&archive[16..])),
+        ]);
+        let body = StreamingBlob::wrap(UploadStream::new(frames, size, s3s_sigv4::Sha256Sum::from_bytes([0; 32])));
+        let input = PutObjectInput::builder()
+            .bucket(bucket.clone())
+            .key("archive.tar".to_string())
+            .content_length(Some(i64::try_from(size).expect("fixture size")))
+            .body(Some(body))
+            .build()
+            .expect("extract input");
+        let mut request = build_request(input, Method::PUT);
+        request
+            .headers
+            .insert("x-amz-meta-snowball-auto-extract", HeaderValue::from_static("true"));
+        let error = DefaultObjectUsecase::with_context(Some(context))
+            .execute_put_object_extract(request)
+            .await
+            .expect_err("a bad payload digest must reject the archive request");
+        assert_eq!(error.code().as_str(), "XAmzContentSHA256Mismatch");
+        assert_eq!(error.status_code(), Some(http::StatusCode::BAD_REQUEST));
+        for key in ["archive.tar", "member.txt"] {
+            let missing = store
+                .get_object_info(&bucket, key, &ObjectOptions::default())
+                .await
+                .expect_err("rejected archive cannot publish a member");
+            assert!(is_err_object_not_found(&missing), "{missing}");
+        }
+    }
+
     #[tokio::test]
     async fn archive_etag_reader_validates_sha256_before_completion() {
         let payload = b"archive-with-wrong-sha256".to_vec();
@@ -3502,6 +3611,9 @@ mod tests {
 
         assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
         assert!(!state.lock().expect("archive state lock must remain healthy").body_complete);
+        let error = map_extract_archive_error(err);
+        assert_eq!(error.code().as_str(), "XAmzContentSHA256Mismatch");
+        assert_eq!(error.status_code(), Some(http::StatusCode::BAD_REQUEST));
     }
 
     fn pax_record(key: &str, value: &[u8]) -> Vec<u8> {
@@ -3648,6 +3760,9 @@ mod tests {
 
         assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
         assert!(!state.lock().expect("archive state lock must remain healthy").body_complete);
+        let error = map_extract_archive_error(err);
+        assert_eq!(error.code(), &S3ErrorCode::BadDigest);
+        assert_eq!(error.status_code(), Some(http::StatusCode::BAD_REQUEST));
     }
 
     #[tokio::test]
@@ -4777,7 +4892,7 @@ mod tests {
         );
     }
 
-    #[tokio::test]
+    #[tokio::test(start_paused = true)]
     async fn snowball_extract_body_guard_aborts_stalled_upload() {
         let body = StreamingBlob::wrap(futures::stream::pending::<Result<Bytes, std::io::Error>>());
         let mut guarded = guard_put_object_body_read_timeout(
@@ -4798,6 +4913,9 @@ mod tests {
             .downcast_ref::<std::io::Error>()
             .expect("stall error should retain its I/O kind");
         assert_eq!(io_err.kind(), std::io::ErrorKind::TimedOut);
+        let error = map_extract_archive_error(s3s_body_error_to_io(err));
+        assert_eq!(error.code(), &S3ErrorCode::RequestTimeout);
+        assert_eq!(error.status_code(), Some(http::StatusCode::BAD_REQUEST));
     }
 
     #[test]

rustfs/src/app/object/put.rs

@@ -18,7 +18,7 @@ use super::*;
 
 use crate::app::trailer_adapter::trailer_source;
 use crate::auth::{RUSTFS_MAX_CONTENT_LENGTH_QUERY, VerifiedPresignedRequest, parse_presigned_put_max_content_length};
-use crate::error::{UploadLimitExceeded, normalize_put_read_quorum_error};
+use crate::error::{ClientBodyReadTimeout, UploadLimitExceeded, normalize_put_read_quorum_error};
 static PUT_FAILURE_LOGS: rustfs_utils::LogThrottle = rustfs_utils::LogThrottle::new(5_000);
 
 #[cfg(test)]
@@ -293,12 +293,10 @@ impl Stream for RequestBodyReadTimeout {
                     );
                     return Poll::Ready(Some(Err(Box::new(std::io::Error::new(
                         std::io::ErrorKind::TimedOut,
-                        format!(
-                            "request body read stalled: received {} of {} bytes, no data for {}s",
-                            this.received,
-                            expected_display,
-                            this.timeout.as_secs()
-                        ),
+                        ClientBodyReadTimeout {
+                            timeout: this.timeout,
+                            raw_bytes_received: this.received,
+                        },
                     )) as StdError)));
                 }
 
@@ -2970,7 +2968,7 @@ mod tests {
         assert!(!object_traffic_health.snapshot().write_stalled);
     }
 
-    #[tokio::test]
+    #[tokio::test(start_paused = true)]
     async fn put_object_body_read_timeout_guard_aborts_on_stall() {
         // Inner stream never yields and never reports EOF (a proxy that forwarded
         // a partial body then went silent while holding the connection open).
@@ -2993,6 +2991,9 @@ mod tests {
             .downcast_ref::<std::io::Error>()
             .expect("stall error should wrap an io::Error");
         assert_eq!(io_err.kind(), std::io::ErrorKind::TimedOut);
+        let error = S3Error::from(ApiError::from(s3s_body_error_to_io(err)));
+        assert_eq!(error.code(), &S3ErrorCode::RequestTimeout);
+        assert_eq!(error.status_code(), Some(http::StatusCode::BAD_REQUEST));
 
         // After a stall the guard terminates the stream instead of re-polling the
         // abandoned inner stream.
@@ -3260,28 +3261,28 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn read_small_put_body_maps_upload_stream_sha256_mismatch_to_bad_digest() {
+    async fn read_small_put_body_maps_upload_stream_sha256_mismatch_to_content_sha256_mismatch() {
         let body = StreamReader::new(futures::stream::iter(vec![Err::<Bytes, std::io::Error>(s3s_body_error_to_io(Box::new(
-            MockUploadStreamSha256Mismatch,
+            UploadStreamError::Sha256Mismatch,
         )))]));
 
         let error = read_small_put_body_exact_direct(body, 1)
             .await
             .expect_err("SHA256 mismatch should reject the small PUT body");
 
-        assert_eq!(error.code(), &S3ErrorCode::BadDigest);
+        assert_eq!(error.code().as_str(), "XAmzContentSHA256Mismatch");
     }
 
     #[tokio::test]
-    async fn read_zero_copy_put_body_maps_upload_stream_sha256_mismatch_to_bad_digest() {
-        let body = futures::stream::iter(vec![Err::<Bytes, MockUploadStreamSha256Mismatch>(MockUploadStreamSha256Mismatch)]);
+    async fn read_zero_copy_put_body_maps_upload_stream_sha256_mismatch_to_content_sha256_mismatch() {
+        let body = futures::stream::iter(vec![Err::<Bytes, UploadStreamError>(UploadStreamError::Sha256Mismatch)]);
 
         let error = match read_zero_copy_put_body_exact(body, 1).await {
             Ok(_) => panic!("SHA256 mismatch should reject the zero-copy PUT body"),
             Err(error) => error,
         };
 
-        assert_eq!(error.code(), &S3ErrorCode::BadDigest);
+        assert_eq!(error.code().as_str(), "XAmzContentSHA256Mismatch");
     }
 
     struct FragmentedBody {

rustfs/src/app/object/shared.rs

@@ -1783,13 +1783,11 @@ mod tests {
 
     #[test]
     fn s3s_body_error_to_io_preserves_upload_stream_error_source() {
-        let error = s3s_body_error_to_io(Box::new(MockUploadStreamSha256Mismatch));
+        let error = s3s_body_error_to_io(Box::new(UploadStreamError::Sha256Mismatch));
 
         assert!(matches!(
-            error
-                .get_ref()
-                .and_then(|source| source.downcast_ref::<MockUploadStreamSha256Mismatch>()),
-            Some(MockUploadStreamSha256Mismatch)
+            error.get_ref().and_then(|source| source.downcast_ref::<UploadStreamError>()),
+            Some(UploadStreamError::Sha256Mismatch)
         ));
     }
 

rustfs/src/app/object/test_support.rs

@@ -18,16 +18,7 @@ use super::*;
 use http::{Extensions, HeaderMap, Method, Uri};
 use std::sync::Arc;
 
-#[derive(Debug)]
-pub(super) struct MockUploadStreamSha256Mismatch;
-
-impl std::fmt::Display for MockUploadStreamSha256Mismatch {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str("UploadStreamError: Sha256Mismatch")
-    }
-}
-
-impl std::error::Error for MockUploadStreamSha256Mismatch {}
+pub(super) use crate::app::storage_api::s3::UploadStreamError;
 
 pub(super) fn build_request<T>(input: T, method: Method) -> S3Request<T> {
     S3Request {

rustfs/src/app/storage_api.rs

@@ -54,6 +54,8 @@ pub(crate) mod s3 {
     #[cfg(test)]
     pub(crate) use s3s::service::{S3Service, S3ServiceBuilder};
     #[cfg(test)]
+    pub(crate) use s3s::stream::upload_stream::{UploadStream, UploadStreamError};
+    #[cfg(test)]
     pub(crate) use s3s::xml::{Serialize as XmlSerialize, Serializer as XmlSerializer};
     #[cfg(test)]
     pub(crate) use s3s::{Body, S3, S3Response};

rustfs/src/error.rs

@@ -17,8 +17,12 @@ use crate::storage_api::error::{PoolMetadataError, QuotaError, StorageError, unr
 use http::StatusCode;
 use rustfs_kms::KmsUnavailableError;
 use rustfs_lock::LockError;
+use s3s::stream::{aws_chunked_stream::AwsChunkedStreamError, upload_stream::UploadStreamError};
 use s3s::{S3Error, S3ErrorCode};
 
+const CONTENT_SHA256_MISMATCH_CODE: &str = "XAmzContentSHA256Mismatch";
+const CONTENT_SHA256_MISMATCH_MESSAGE: &str = "The provided 'x-amz-content-sha256' header does not match what was computed.";
+
 const MAX_VERSIONS_EXCEEDED_CODE: &str = "MaxVersionsExceeded";
 const MAX_VERSIONS_EXCEEDED_MESSAGE: &str = "You've exceeded the limit on the number of versions you can create on this object";
 const SLOW_DOWN_READ_CODE: &str = "SlowDownRead";
@@ -164,7 +168,12 @@ fn data_plane_kms_error(error: &rustfs_kms::KmsError) -> Option<(S3ErrorCode, St
 /// exist" has to say so itself.
 fn custom_error_status(code: &S3ErrorCode) -> Option<StatusCode> {
     match code {
-        S3ErrorCode::Custom(custom) if &**custom == KMS_KEY_NOT_FOUND_ERROR_CODE || &**custom == MAX_VERSIONS_EXCEEDED_CODE => {
+        S3ErrorCode::Custom(custom)
+            if matches!(
+                &**custom,
+                KMS_KEY_NOT_FOUND_ERROR_CODE | MAX_VERSIONS_EXCEEDED_CODE | CONTENT_SHA256_MISMATCH_CODE
+            ) =>
+        {
             Some(StatusCode::BAD_REQUEST)
         }
         S3ErrorCode::Custom(custom) if &**custom == SLOW_DOWN_READ_CODE => Some(StatusCode::SERVICE_UNAVAILABLE),
@@ -398,6 +407,9 @@ impl ApiError {
             S3ErrorCode::InvalidArgument => "Invalid argument".to_string(),
             S3ErrorCode::InvalidStorageClass => "Invalid storage class.".to_string(),
             S3ErrorCode::AccessDenied => "Access Denied.".to_string(),
+            S3ErrorCode::Custom(code) if &**code == CONTENT_SHA256_MISMATCH_CODE => {
+                CONTENT_SHA256_MISMATCH_MESSAGE.to_string()
+            }
             S3ErrorCode::BadDigest => "The Content-Md5 you specified did not match what we received.".to_string(),
             S3ErrorCode::EntityTooSmall => "Your proposed upload is smaller than the minimum allowed object size.".to_string(),
             S3ErrorCode::EntityTooLarge => "Your proposed upload exceeds the maximum allowed object size.".to_string(),
@@ -534,32 +546,55 @@ impl ApiError {
     }
 }
 
-fn error_chain_has_type<T>(err: &(dyn std::error::Error + 'static)) -> bool
-where
-    T: std::error::Error + 'static,
-{
-    error_chain_find(err, |error| error.is::<T>().then_some(())).is_some()
-}
+/// Classify body failures in one bounded pass, retaining precedence across wrappers.
+/// A digest failure wins immediately; other markers must wait for deeper causes.
+fn body_read_error_code(err: &(dyn std::error::Error + 'static)) -> Option<S3ErrorCode> {
+    let mut upload_limit_exceeded = false;
+    let mut server_source_failed = false;
+    let mut client_body_timed_out = false;
+    let mut body_size_exceeded = false;
+    let mut incomplete_body = false;
+    let digest = error_chain_find(err, |error| {
+        if error.is::<rustfs_rio::Sha256Mismatch>()
+            || matches!(error.downcast_ref::<UploadStreamError>(), Some(UploadStreamError::Sha256Mismatch))
+        {
+            return Some(S3ErrorCode::Custom(CONTENT_SHA256_MISMATCH_CODE.into()));
+        }
+        if error.is::<rustfs_rio::ChecksumMismatch>() || error.is::<rustfs_rio::BadDigest>() {
+            return Some(S3ErrorCode::BadDigest);
+        }
 
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
-enum S3sBodyStreamError {
-    Sha256Mismatch,
-    IncompleteBody,
-}
-
-fn classify_s3s_body_stream_error_display(err: &(dyn std::error::Error + 'static)) -> Option<S3sBodyStreamError> {
-    match err.to_string().as_str() {
-        "UploadStreamError: Sha256Mismatch" => Some(S3sBodyStreamError::Sha256Mismatch),
-        "UploadStreamError: LengthMismatch"
-        | "UploadStreamError: Incomplete"
-        | "AwsChunkedStreamError: LengthMismatch"
-        | "AwsChunkedStreamError: Incomplete" => Some(S3sBodyStreamError::IncompleteBody),
-        _ => None,
-    }
-}
-
-fn error_chain_s3s_body_stream_error(err: &(dyn std::error::Error + 'static)) -> Option<S3sBodyStreamError> {
-    error_chain_find(err, classify_s3s_body_stream_error_display)
+        upload_limit_exceeded |= error.is::<UploadLimitExceeded>();
+        server_source_failed |= error.is::<ServerSideSourceReadError>();
+        client_body_timed_out |= error.is::<ClientBodyReadTimeout>();
+        body_size_exceeded |= error.is::<s3s::BodySizeLimitExceeded>();
+        incomplete_body |= error.is::<rustfs_rio::IncompleteBody>()
+            || matches!(
+                error.downcast_ref::<UploadStreamError>(),
+                Some(UploadStreamError::LengthMismatch | UploadStreamError::Incomplete)
+            )
+            || matches!(
+                error.downcast_ref::<AwsChunkedStreamError>(),
+                Some(AwsChunkedStreamError::LengthMismatch | AwsChunkedStreamError::Incomplete)
+            )
+            || is_hyper_body_eof(error);
+        None
+    });
+    digest.or_else(|| {
+        if upload_limit_exceeded {
+            Some(S3ErrorCode::EntityTooLarge)
+        } else if server_source_failed {
+            Some(S3ErrorCode::ServiceUnavailable)
+        } else if client_body_timed_out {
+            Some(S3ErrorCode::RequestTimeout)
+        } else if body_size_exceeded {
+            Some(S3ErrorCode::EntityTooLarge)
+        } else if incomplete_body {
+            Some(S3ErrorCode::IncompleteBody)
+        } else {
+            None
+        }
+    })
 }
 
 /// `io::Error::source` skips its custom payload itself. Visit that payload
@@ -584,10 +619,6 @@ fn error_chain_find<T>(
     None
 }
 
-fn error_chain_has_body_size_limit_exceeded(err: &(dyn std::error::Error + 'static)) -> bool {
-    error_chain_has_type::<s3s::BodySizeLimitExceeded>(err)
-}
-
 /// hyper reports a request body whose connection hit EOF before
 /// `Content-Length` bytes arrived as a `Kind::Body` error carrying an
 /// `UnexpectedEof` `io::Error` (its `IncompleteBody` marker is private).
@@ -599,10 +630,6 @@ fn is_hyper_body_eof(err: &(dyn std::error::Error + 'static)) -> bool {
         .is_some_and(|io_err| io_err.kind() == std::io::ErrorKind::UnexpectedEof)
 }
 
-fn error_chain_has_hyper_body_eof(err: &(dyn std::error::Error + 'static)) -> bool {
-    error_chain_find(err, |error| is_hyper_body_eof(error).then_some(())).is_some()
-}
-
 impl From<ApiError> for S3Error {
     fn from(err: ApiError) -> Self {
         let status = custom_error_status(&err.code);
@@ -674,65 +701,10 @@ impl From<StorageError> for ApiError {
         if let StorageError::Io(ref io_err) = err
             && let Some(inner) = io_err.get_ref()
         {
-            let s3s_body_stream_error = error_chain_s3s_body_stream_error(inner);
-
-            // Preserve client-provided body and digest failures across I/O boundaries.
-            if inner.downcast_ref::<rustfs_rio::ChecksumMismatch>().is_some()
-                || inner.downcast_ref::<rustfs_rio::BadDigest>().is_some()
-                || inner.downcast_ref::<rustfs_rio::Sha256Mismatch>().is_some()
-                || matches!(s3s_body_stream_error, Some(S3sBodyStreamError::Sha256Mismatch))
-            {
+            if let Some(code) = body_read_error_code(inner) {
                 return ApiError {
-                    code: S3ErrorCode::BadDigest,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::BadDigest),
-                    source: Some(Box::new(err)),
-                };
-            }
-
-            if error_chain_has_type::<UploadLimitExceeded>(inner) {
-                return ApiError {
-                    code: S3ErrorCode::EntityTooLarge,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
-                    source: Some(Box::new(err)),
-                };
-            }
-
-            if error_chain_has_type::<ServerSideSourceReadError>(inner) {
-                return ApiError {
-                    code: S3ErrorCode::ServiceUnavailable,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable),
-                    source: Some(Box::new(err)),
-                };
-            }
-
-            if error_chain_has_type::<ClientBodyReadTimeout>(inner) {
-                return ApiError {
-                    code: S3ErrorCode::RequestTimeout,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::RequestTimeout),
-                    source: Some(Box::new(err)),
-                };
-            }
-
-            if error_chain_has_body_size_limit_exceeded(inner) {
-                return ApiError {
-                    code: S3ErrorCode::EntityTooLarge,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
-                    source: Some(Box::new(err)),
-                };
-            }
-
-            if error_chain_has_hyper_body_eof(inner) {
-                return ApiError {
-                    code: S3ErrorCode::IncompleteBody,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody),
-                    source: Some(Box::new(err)),
-                };
-            }
-
-            if matches!(s3s_body_stream_error, Some(S3sBodyStreamError::IncompleteBody)) {
-                return ApiError {
-                    code: S3ErrorCode::IncompleteBody,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody),
+                    message: ApiError::error_code_to_message(&code),
+                    code,
                     source: Some(Box::new(err)),
                 };
             }
@@ -852,72 +824,14 @@ impl From<HTTPRangeError> for ApiError {
 
 impl From<std::io::Error> for ApiError {
     fn from(err: std::io::Error) -> Self {
-        // Map client-provided digest mismatches to BadDigest.
-        if let Some(inner) = err.get_ref() {
-            let s3s_body_stream_error = error_chain_s3s_body_stream_error(inner);
-            if error_chain_has_type::<rustfs_rio::ChecksumMismatch>(inner)
-                || error_chain_has_type::<rustfs_rio::BadDigest>(inner)
-                || error_chain_has_type::<rustfs_rio::Sha256Mismatch>(inner)
-                || matches!(s3s_body_stream_error, Some(S3sBodyStreamError::Sha256Mismatch))
-            {
-                return ApiError {
-                    code: S3ErrorCode::BadDigest,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::BadDigest),
-                    source: Some(Box::new(err)),
-                };
-            }
-            if error_chain_has_type::<UploadLimitExceeded>(inner) {
-                return ApiError {
-                    code: S3ErrorCode::EntityTooLarge,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
-                    source: Some(Box::new(err)),
-                };
-            }
-            if error_chain_has_type::<ServerSideSourceReadError>(inner) {
-                return ApiError {
-                    code: S3ErrorCode::ServiceUnavailable,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable),
-                    source: Some(Box::new(err)),
-                };
-            }
-            if error_chain_has_type::<ClientBodyReadTimeout>(inner) {
-                return ApiError {
-                    code: S3ErrorCode::RequestTimeout,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::RequestTimeout),
-                    source: Some(Box::new(err)),
-                };
-            }
-
-            if error_chain_has_body_size_limit_exceeded(inner) {
-                return ApiError {
-                    code: S3ErrorCode::EntityTooLarge,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
-                    source: Some(Box::new(err)),
-                };
-            }
-
-            if error_chain_has_hyper_body_eof(inner) {
-                return ApiError {
-                    code: S3ErrorCode::IncompleteBody,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody),
-                    source: Some(Box::new(err)),
-                };
-            }
-
-            if matches!(s3s_body_stream_error, Some(S3sBodyStreamError::IncompleteBody)) {
-                return ApiError {
-                    code: S3ErrorCode::IncompleteBody,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody),
-                    source: Some(Box::new(err)),
-                };
-            }
-            if error_chain_has_type::<rustfs_rio::IncompleteBody>(inner) {
-                return ApiError {
-                    code: S3ErrorCode::IncompleteBody,
-                    message: ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody),
-                    source: Some(Box::new(err)),
-                };
-            }
+        if let Some(inner) = err.get_ref()
+            && let Some(code) = body_read_error_code(inner)
+        {
+            return ApiError {
+                message: ApiError::error_code_to_message(&code),
+                code,
+                source: Some(Box::new(err)),
+            };
         }
         ApiError {
             code: S3ErrorCode::InternalError,
@@ -965,9 +879,17 @@ mod tests {
     use metrics_util::debugging::{DebugValue, DebuggingRecorder};
     use rustfs_io_metrics::s3_http_metrics::S3HttpRequestGuard;
     use rustfs_s3_ops::S3Operation;
+    use s3s::stream::{aws_chunked_stream::AwsChunkedStreamError, upload_stream::UploadStreamError};
     use s3s::{S3Error, S3ErrorCode};
     use std::io::{Error as IoError, ErrorKind};
 
+    fn error_chain_has_type<T>(err: &(dyn std::error::Error + 'static)) -> bool
+    where
+        T: std::error::Error + 'static,
+    {
+        error_chain_find(err, |error| error.is::<T>().then_some(())).is_some()
+    }
+
     #[tokio::test]
     async fn lock_timeout_response_advises_retry_without_exposing_resource() {
         use http_body_util::BodyExt as _;
@@ -1238,34 +1160,6 @@ mod tests {
         assert!(!format!("{:?}", error.diagnostic()).contains("secret"));
     }
 
-    #[derive(Debug)]
-    enum MockUploadStreamError {
-        Underlying(IoError),
-        Sha256Mismatch,
-        LengthMismatch,
-        Incomplete,
-    }
-
-    impl std::fmt::Display for MockUploadStreamError {
-        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-            match self {
-                Self::Underlying(err) => write!(f, "UploadStreamError: Underlying: {err}"),
-                Self::Sha256Mismatch => f.write_str("UploadStreamError: Sha256Mismatch"),
-                Self::LengthMismatch => f.write_str("UploadStreamError: LengthMismatch"),
-                Self::Incomplete => f.write_str("UploadStreamError: Incomplete"),
-            }
-        }
-    }
-
-    impl std::error::Error for MockUploadStreamError {
-        fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
-            match self {
-                Self::Underlying(err) => Some(err),
-                Self::Sha256Mismatch | Self::LengthMismatch | Self::Incomplete => None,
-            }
-        }
-    }
-
     #[derive(Debug)]
     struct MockS3sBodyStreamError(&'static str);
 
@@ -1314,94 +1208,124 @@ mod tests {
     }
 
     #[test]
-    fn sha256_mismatch_io_errors_map_to_bad_digest() {
-        let io_error = IoError::new(
-            ErrorKind::InvalidData,
-            rustfs_rio::Sha256Mismatch {
-                expected_sha256: "expected".to_string(),
-                calculated_sha256: "calculated".to_string(),
-            },
-        );
-        let api_error = ApiError::from(io_error);
-        assert_eq!(api_error.code, S3ErrorCode::BadDigest);
-        assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest));
-
-        let storage_error = StorageError::Io(IoError::new(
-            ErrorKind::InvalidData,
-            rustfs_rio::Sha256Mismatch {
-                expected_sha256: "expected".to_string(),
-                calculated_sha256: "calculated".to_string(),
-            },
-        ));
-        let api_error = ApiError::from(storage_error);
-        assert_eq!(api_error.code, S3ErrorCode::BadDigest);
-        assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest));
+    fn payload_sha256_mismatch_io_errors_map_to_content_sha256_mismatch() {
+        for storage in [false, true] {
+            for nested in [false, true] {
+                for error in [
+                    IoError::other(rustfs_rio::Sha256Mismatch {
+                        expected_sha256: "expected".to_string(),
+                        calculated_sha256: "calculated".to_string(),
+                    }),
+                    IoError::other(UploadStreamError::Sha256Mismatch),
+                ] {
+                    let error = if nested { IoError::other(error) } else { error };
+                    let api_error = if storage {
+                        ApiError::from(StorageError::Io(error))
+                    } else {
+                        ApiError::from(error)
+                    };
+                    assert_eq!(api_error.code.as_str(), "XAmzContentSHA256Mismatch");
+                    assert_eq!(
+                        api_error.message,
+                        "The provided 'x-amz-content-sha256' header does not match what was computed."
+                    );
+                    assert!(
+                        error_chain_has_type::<rustfs_rio::Sha256Mismatch>(&api_error)
+                            || error_chain_has_type::<UploadStreamError>(&api_error)
+                    );
+                    let s3_error = S3Error::from(api_error);
+                    assert_eq!(s3_error.status_code(), Some(StatusCode::BAD_REQUEST));
+                    assert!(s3_error.source().is_some());
+                }
+            }
+        }
     }
 
     #[test]
-    fn content_md5_mismatch_io_errors_map_to_bad_digest() {
-        let bad_digest = || rustfs_rio::BadDigest {
-            expected_md5: "expected".to_string(),
-            calculated_md5: "calculated".to_string(),
-        };
-        let api_error = ApiError::from(IoError::new(ErrorKind::InvalidData, bad_digest()));
-        assert_eq!(api_error.code, S3ErrorCode::BadDigest);
-        assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest));
-
-        let api_error = ApiError::from(StorageError::Io(IoError::new(ErrorKind::InvalidData, bad_digest())));
-        assert_eq!(api_error.code, S3ErrorCode::BadDigest);
-        assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest));
-    }
-
-    #[test]
-    fn upload_stream_sha256_mismatch_maps_to_bad_digest() {
-        let api_error = ApiError::from(IoError::other(MockUploadStreamError::Sha256Mismatch));
-        assert_eq!(api_error.code, S3ErrorCode::BadDigest);
-        assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest));
-
-        let api_error = ApiError::from(StorageError::Io(IoError::other(MockUploadStreamError::Sha256Mismatch)));
-        assert_eq!(api_error.code, S3ErrorCode::BadDigest);
-        assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest));
+    fn content_md5_and_additional_checksum_mismatches_map_to_bad_digest() {
+        for storage in [false, true] {
+            for nested in [false, true] {
+                for error in [
+                    IoError::other(rustfs_rio::BadDigest {
+                        expected_md5: "expected".to_string(),
+                        calculated_md5: "calculated".to_string(),
+                    }),
+                    IoError::other(rustfs_rio::ChecksumMismatch {
+                        want: "expected".into(),
+                        got: "calculated".into(),
+                    }),
+                ] {
+                    let error = if nested { IoError::other(error) } else { error };
+                    let api_error = if storage {
+                        ApiError::from(StorageError::Io(error))
+                    } else {
+                        ApiError::from(error)
+                    };
+                    assert_eq!(api_error.code, S3ErrorCode::BadDigest);
+                    assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest));
+                    assert_eq!(S3Error::from(api_error).status_code(), Some(StatusCode::BAD_REQUEST));
+                }
+            }
+        }
     }
 
     #[test]
     fn underlying_upload_stream_errors_remain_internal() {
-        let api_error =
-            ApiError::from(IoError::other(MockUploadStreamError::Underlying(IoError::other("underlying body error"))));
+        let api_error = ApiError::from(IoError::other(UploadStreamError::Underlying(Box::new(IoError::other(
+            "underlying body error",
+        )))));
         assert_eq!(api_error.code, S3ErrorCode::InternalError);
 
-        let api_error = ApiError::from(StorageError::Io(IoError::other(MockUploadStreamError::Underlying(IoError::other(
-            "underlying body error",
+        let api_error = ApiError::from(StorageError::Io(IoError::other(UploadStreamError::Underlying(Box::new(
+            IoError::other("underlying body error"),
         )))));
         assert_eq!(api_error.code, S3ErrorCode::InternalError);
     }
 
     #[test]
     fn s3s_incomplete_body_stream_errors_map_to_incomplete_body() {
-        for error in [MockUploadStreamError::LengthMismatch, MockUploadStreamError::Incomplete] {
+        for error in [UploadStreamError::LengthMismatch, UploadStreamError::Incomplete] {
             let api_error = ApiError::from(IoError::other(error));
             assert_eq!(api_error.code, S3ErrorCode::IncompleteBody);
             assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody));
         }
 
-        for error in [MockUploadStreamError::LengthMismatch, MockUploadStreamError::Incomplete] {
+        for error in [UploadStreamError::LengthMismatch, UploadStreamError::Incomplete] {
             let api_error = ApiError::from(StorageError::Io(IoError::other(error)));
             assert_eq!(api_error.code, S3ErrorCode::IncompleteBody);
             assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody));
         }
 
-        for message in ["AwsChunkedStreamError: LengthMismatch", "AwsChunkedStreamError: Incomplete"] {
-            let api_error = ApiError::from(IoError::other(MockS3sBodyStreamError(message)));
-            assert_eq!(api_error.code, S3ErrorCode::IncompleteBody, "{message}");
-            assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody));
+        for storage in [false, true] {
+            for nested in [false, true] {
+                for error in [AwsChunkedStreamError::LengthMismatch, AwsChunkedStreamError::Incomplete] {
+                    let error = IoError::other(error);
+                    let error = if nested { IoError::other(error) } else { error };
+                    let api_error = if storage {
+                        ApiError::from(StorageError::Io(error))
+                    } else {
+                        ApiError::from(error)
+                    };
+                    assert_eq!(api_error.code, S3ErrorCode::IncompleteBody);
+                    assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody));
+                }
+            }
+        }
+    }
 
-            let api_error = ApiError::from(IoError::other(IoError::other(MockS3sBodyStreamError(message))));
-            assert_eq!(api_error.code, S3ErrorCode::IncompleteBody, "{message}");
-            assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody));
-
-            let api_error = ApiError::from(StorageError::Io(IoError::other(MockS3sBodyStreamError(message))));
-            assert_eq!(api_error.code, S3ErrorCode::IncompleteBody, "{message}");
-            assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::IncompleteBody));
+    #[test]
+    fn body_error_display_text_does_not_impersonate_typed_failures() {
+        for message in [
+            "UploadStreamError: Sha256Mismatch",
+            "UploadStreamError: Incomplete",
+            "AwsChunkedStreamError: LengthMismatch",
+        ] {
+            for error in [
+                ApiError::from(IoError::other(MockS3sBodyStreamError(message))),
+                ApiError::from(StorageError::Io(IoError::other(MockS3sBodyStreamError(message)))),
+            ] {
+                assert_eq!(error.code, S3ErrorCode::InternalError);
+            }
         }
     }
 
@@ -1477,8 +1401,12 @@ mod tests {
 
     #[test]
     fn body_error_classification_bounds_cyclic_source_chains() {
+        use std::sync::{
+            Arc,
+            atomic::{AtomicUsize, Ordering},
+        };
         #[derive(Debug)]
-        struct Cycle;
+        struct Cycle(Arc<AtomicUsize>);
         impl std::fmt::Display for Cycle {
             fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
                 f.write_str("cycle")
@@ -1486,11 +1414,70 @@ mod tests {
         }
         impl std::error::Error for Cycle {
             fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+                self.0.fetch_add(1, Ordering::Relaxed);
                 Some(self)
             }
         }
-        assert!(!error_chain_has_type::<ClientBodyReadTimeout>(&Cycle));
-        assert_eq!(ApiError::from(IoError::other(Cycle)).code, S3ErrorCode::InternalError);
+        let visits = Arc::new(AtomicUsize::new(0));
+        let error = ApiError::from(IoError::other(Cycle(Arc::clone(&visits))));
+        assert_eq!(error.code, S3ErrorCode::InternalError);
+        assert!(visits.load(Ordering::Relaxed) <= 64, "body classification must share one chain budget");
+    }
+
+    #[test]
+    fn body_error_classification_preserves_nested_marker_precedence() {
+        for storage in [false, true] {
+            for (inner, expected) in [
+                (IoError::other(UploadStreamError::Sha256Mismatch), CONTENT_SHA256_MISMATCH_CODE),
+                (
+                    IoError::other(rustfs_rio::BadDigest {
+                        expected_md5: "expected".into(),
+                        calculated_md5: "calculated".into(),
+                    }),
+                    "BadDigest",
+                ),
+                (IoError::other(UploadLimitExceeded { limit: 5 }), "EntityTooLarge"),
+                (
+                    IoError::other(ClientBodyReadTimeout {
+                        timeout: std::time::Duration::from_secs(300),
+                        raw_bytes_received: 0,
+                    }),
+                    "ServiceUnavailable",
+                ),
+                (IoError::other(s3s::BodySizeLimitExceeded { size: 6, limit: 5 }), "ServiceUnavailable"),
+            ] {
+                let error = IoError::other(ServerSideSourceReadError::new("CopyObject", inner));
+                let error = if storage {
+                    ApiError::from(StorageError::Io(error))
+                } else {
+                    ApiError::from(error)
+                };
+                assert_eq!(error.code.as_str(), expected);
+                assert!(error.source.is_some());
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn short_hash_reader_body_maps_to_incomplete_body_across_io_boundaries() {
+        use tokio::io::AsyncReadExt;
+        for storage in [false, true] {
+            let mut reader =
+                rustfs_rio::HashReader::from_stream(std::io::Cursor::new(b"short".to_vec()), 10, 10, None, None, false)
+                    .expect("valid declared body length");
+            let error = reader
+                .read_to_end(&mut Vec::new())
+                .await
+                .expect_err("five bytes cannot satisfy declared ten");
+            let error = if storage {
+                ApiError::from(StorageError::Io(error))
+            } else {
+                ApiError::from(error)
+            };
+            assert_eq!(error.code, S3ErrorCode::IncompleteBody);
+            assert!(error_chain_has_type::<rustfs_rio::IncompleteBody>(&error));
+            assert_eq!(S3Error::from(error).status_code(), Some(StatusCode::BAD_REQUEST));
+        }
     }
 
     /// Exercise the public error type through the actual body budget.

rustfs/src/storage/options.rs

@@ -1065,7 +1065,7 @@ pub fn parse_copy_source_range(range_str: &str) -> S3Result<HTTPRangeSpec> {
 /// digest, for both header-signed and presigned SigV4 requests, and rejects a
 /// malformed header outright. Every request that reaches a handler with that
 /// header has therefore already had its body verified (or is about to fail in
-/// the body stream with the same `BadDigest` mapping), so hashing it again in
+/// the body stream with `XAmzContentSHA256Mismatch`), so hashing it again in
 /// `HashReader` would cost a second full SHA-256 pass per PUT for no extra
 /// protection. Streaming payloads are verified chunk by chunk by s3s and never
 /// produced a digest here.

scripts/check_s3s_footprint.sh

@@ -65,7 +65,7 @@ cd "$(dirname "$0")/.."
 # other reductions. The merged tree measures 209 import files and 1586
 # invocation lines, tightening main's reviewed 1592-line baseline by six.
 S3S_IMPORT_FILES_BASELINE=209
-S3_ERROR_LINES_BASELINE=1586
+S3_ERROR_LINES_BASELINE=1572
 # ecstore-scoped ratchet (rustfs/backlog#1842): the storage engine must not
 # know S3 wire/DTO types (ARCHITECTURE.md invariant 4). The S3-*consuming*
 # client was extracted to crates/s3-client, where s3s usage is legitimate;