fix(api): preserve typed upload digest errors (#6564)

Cargo.lock

@@ -10802,8 +10802,7 @@ checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
 [[package]]
 name = "s3s"
 version = "0.15.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "652451a66fefda01a13dc7df24aa2af9e70c7058f98bc08fe2f7c552eaa9f1e3"
+source = "git+https://github.com/rustfs/s3s.git?rev=5f22e8d0a37e83f531f653024aac11c72586479a#5f22e8d0a37e83f531f653024aac11c72586479a"
 dependencies = [
  "arc-swap",
  "arrayvec",

Cargo.toml

@@ -293,7 +293,7 @@ rustify = { version = "0.7", default-features = false }
 rustix = { version = "1.1.4" }
 rust-embed = { version = "8.12.0" }
 rustc-hash = { version = "2.1.3" }
-s3s = { version = "0.15.0", features = ["minio"] }
+s3s = { git = "https://github.com/rustfs/s3s.git", rev = "5f22e8d0a37e83f531f653024aac11c72586479a", version = "0.15.0", features = ["minio"] }
 serial_test = "4.0.1"
 shadow-rs = { default-features = false, version = "2.0.0" }
 siphasher = "1.0.3"

crates/rio/src/hash_reader.rs

@@ -560,7 +560,13 @@ impl AsyncRead for HashReader {
                         let sha256 = hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower);
                         if sha256 != *expected_sha256 {
                             error!("SHA256 mismatch, expected={:?}, actual={:?}", expected_sha256, sha256);
-                            return Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "SHA256 mismatch")));
+                            return Poll::Ready(Err(std::io::Error::new(
+                                std::io::ErrorKind::InvalidData,
+                                crate::errors::Sha256Mismatch {
+                                    expected_sha256: expected_sha256.clone(),
+                                    calculated_sha256: sha256,
+                                },
+                            )));
                         }
                     }
 
@@ -774,6 +780,30 @@ mod tests {
         assert_eq!(buf, data);
     }
 
+    #[tokio::test]
+    async fn sha256_mismatch_retains_typed_io_error_source() {
+        let data = b"tampered payload";
+        let expected_sha256 = "0".repeat(64);
+        let reader = BufReader::new(Cursor::new(&data[..]));
+        let mut hash_reader =
+            HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, Some(expected_sha256.clone()), false)
+                .expect("operation should succeed");
+
+        let error = hash_reader
+            .read_to_end(&mut Vec::new())
+            .await
+            .expect_err("SHA256 mismatch should fail");
+        let mismatch = error
+            .get_ref()
+            .and_then(|source| source.downcast_ref::<crate::Sha256Mismatch>())
+            .expect("SHA256 mismatch should remain typed");
+
+        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
+        assert_eq!(mismatch.expected_sha256, expected_sha256);
+        assert_eq!(mismatch.calculated_sha256.len(), 64);
+        assert_ne!(mismatch.calculated_sha256, mismatch.expected_sha256);
+    }
+
     #[tokio::test]
     async fn test_add_calculated_checksum_records_checksum() {
         let data = b"server-side copy checksum";

rustfs/src/app/object_usecase.rs

@@ -235,22 +235,6 @@ use crate::app::object_traffic_health::ObjectTrafficHealth;
 type S3StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
 
 fn s3s_body_error_to_io(err: StdError) -> io::Error {
-    match err.to_string().as_str() {
-        "UploadStreamError: Sha256Mismatch" => {
-            return io::Error::new(
-                io::ErrorKind::InvalidData,
-                rustfs_rio::ChecksumMismatch {
-                    want: AMZ_CONTENT_SHA256.to_string(),
-                    got: "payload sha256".to_string(),
-                },
-            );
-        }
-        "UploadStreamError: Incomplete" | "UploadStreamError: LengthMismatch" => {
-            return io::Error::new(io::ErrorKind::UnexpectedEof, rustfs_rio::IncompleteBody { remaining: 0 });
-        }
-        _ => {}
-    }
-
     io::Error::other(err)
 }
 
@@ -16250,6 +16234,43 @@ mod tests {
         );
     }
 
+    #[test]
+    fn s3s_body_error_to_io_preserves_upload_stream_error_source() {
+        let error = s3s_body_error_to_io(Box::new(s3s::UploadStreamError::Sha256Mismatch));
+
+        assert!(matches!(
+            error
+                .get_ref()
+                .and_then(|source| source.downcast_ref::<s3s::UploadStreamError>()),
+            Some(s3s::UploadStreamError::Sha256Mismatch)
+        ));
+    }
+
+    #[tokio::test]
+    async fn read_small_put_body_maps_upload_stream_sha256_mismatch_to_bad_digest() {
+        let body = StreamReader::new(futures::stream::iter(vec![Err::<Bytes, std::io::Error>(s3s_body_error_to_io(Box::new(
+            s3s::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);
+    }
+
+    #[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, s3s::UploadStreamError>(s3s::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);
+    }
+
     struct FragmentedBody {
         data: std::io::Cursor<Vec<u8>>,
     }

rustfs/src/error.rs

@@ -225,6 +225,28 @@ where
     false
 }
 
+fn error_chain_has_upload_stream_sha256_mismatch(err: &(dyn std::error::Error + 'static)) -> bool {
+    if matches!(err.downcast_ref::<s3s::UploadStreamError>(), Some(s3s::UploadStreamError::Sha256Mismatch)) {
+        return true;
+    }
+
+    if let Some(io_err) = err.downcast_ref::<std::io::Error>()
+        && let Some(inner) = io_err.get_ref()
+        && error_chain_has_upload_stream_sha256_mismatch(inner)
+    {
+        return true;
+    }
+
+    let mut current = err.source();
+    while let Some(err) = current {
+        if matches!(err.downcast_ref::<s3s::UploadStreamError>(), Some(s3s::UploadStreamError::Sha256Mismatch)) {
+            return true;
+        }
+        current = err.source();
+    }
+    false
+}
+
 impl From<ApiError> for S3Error {
     fn from(err: ApiError) -> Self {
         let mut s3e = S3Error::with_message(err.code, err.message);
@@ -237,11 +259,13 @@ impl From<ApiError> for S3Error {
 
 impl From<StorageError> for ApiError {
     fn from(err: StorageError) -> Self {
-        // Special handling for Io errors that may contain ChecksumMismatch
+        // Preserve typed client-provided digest failures across I/O boundaries.
         if let StorageError::Io(ref io_err) = err
             && let Some(inner) = io_err.get_ref()
             && (inner.downcast_ref::<rustfs_rio::ChecksumMismatch>().is_some()
-                || inner.downcast_ref::<rustfs_rio::BadDigest>().is_some())
+                || inner.downcast_ref::<rustfs_rio::BadDigest>().is_some()
+                || inner.downcast_ref::<rustfs_rio::Sha256Mismatch>().is_some()
+                || error_chain_has_upload_stream_sha256_mismatch(inner))
         {
             return ApiError {
                 code: S3ErrorCode::BadDigest,
@@ -362,9 +386,12 @@ impl From<HTTPRangeError> for ApiError {
 
 impl From<std::io::Error> for ApiError {
     fn from(err: std::io::Error) -> Self {
-        // Check if the error is a ChecksumMismatch (BadDigest)
+        // Map client-provided digest mismatches to BadDigest.
         if let Some(inner) = err.get_ref() {
-            if error_chain_has_type::<rustfs_rio::ChecksumMismatch>(inner) || error_chain_has_type::<rustfs_rio::BadDigest>(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)
+                || error_chain_has_upload_stream_sha256_mismatch(inner)
             {
                 return ApiError {
                     code: S3ErrorCode::BadDigest,
@@ -461,6 +488,67 @@ 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));
+    }
+
+    #[test]
+    fn upload_stream_sha256_mismatch_maps_to_bad_digest() {
+        let api_error = ApiError::from(IoError::other(s3s::UploadStreamError::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(s3s::UploadStreamError::Sha256Mismatch)));
+        assert_eq!(api_error.code, S3ErrorCode::BadDigest);
+        assert_eq!(api_error.message, ApiError::error_code_to_message(&S3ErrorCode::BadDigest));
+    }
+
+    #[test]
+    fn other_upload_stream_errors_do_not_map_to_bad_digest() {
+        let errors = [
+            s3s::UploadStreamError::Underlying(Box::new(IoError::other("underlying body error"))),
+            s3s::UploadStreamError::LengthMismatch,
+            s3s::UploadStreamError::Incomplete,
+        ];
+
+        for error in errors {
+            let api_error = ApiError::from(IoError::other(error));
+            assert_eq!(api_error.code, S3ErrorCode::InternalError);
+        }
+
+        let errors = [
+            s3s::UploadStreamError::Underlying(Box::new(IoError::other("underlying body error"))),
+            s3s::UploadStreamError::LengthMismatch,
+            s3s::UploadStreamError::Incomplete,
+        ];
+
+        for error in errors {
+            let api_error = ApiError::from(StorageError::Io(IoError::other(error)));
+            assert_eq!(api_error.code, S3ErrorCode::InternalError);
+        }
+    }
+
     #[test]
     fn test_api_error_surfaces_invalid_argument_reason() {
         let err = StorageError::InvalidArgument(