fix(s3): keep checksum-type validation off the s3s error macro

The s3s footprint ratchet (scripts/check_s3s_footprint.sh) counts
s3_error! invocation lines and is lower-only: new code must route
through the gateway abstractions rather than widen the direct s3s
surface the s3gate migration is shrinking.

Raise the contradiction through ApiError::invalid_request instead. The
response is byte-for-byte identical -- From<ApiError> for S3Error carries
the InvalidRequest code and the message through unchanged -- and the
usecase already returns ApiError elsewhere, so this is the idiomatic
path rather than a way around the counter.

The explanatory comment deliberately says "the s3s error macro" instead
of naming the macro: the ratchet counts raw matches, so spelling it out
in a comment tripped the same check.

crates/e2e_test/src/checksum_upload_test.rs

@@ -22,7 +22,10 @@ mod tests {
     use aws_sdk_s3::config::{Credentials, Region, RequestChecksumCalculation};
     use aws_sdk_s3::error::ProvideErrorMetadata;
     use aws_sdk_s3::primitives::ByteStream;
-    use aws_sdk_s3::types::{ChecksumAlgorithm, ChecksumMode, CompletedMultipartUpload, CompletedPart, ServerSideEncryption};
+    use aws_sdk_s3::types::{
+        ChecksumAlgorithm, ChecksumMode, ChecksumType as SdkChecksumType, CompletedMultipartUpload, CompletedPart,
+        ServerSideEncryption,
+    };
     use aws_smithy_http_client::Builder as SmithyHttpClientBuilder;
     use md5::{Digest as Md5Digest, Md5};
     use rustfs_rio::{Checksum, ChecksumType as RioChecksumType};
@@ -596,6 +599,222 @@ mod tests {
             Some(full_checksum.as_str()),
             "Multipart object should report the same full-object CRC64NVME as direct upload"
         );
+        assert_eq!(
+            multipart_head.checksum_type(),
+            Some(&SdkChecksumType::FullObject),
+            "Multipart object with a full-object checksum must report FULL_OBJECT"
+        );
+    }
+
+    /// Create a CRC32 FULL_OBJECT multipart upload and upload every part, returning
+    /// the upload id and the `CompletedPart` list ready for CompleteMultipartUpload.
+    async fn start_full_object_crc32_upload(
+        client: &Client,
+        bucket: &str,
+        key: &str,
+        parts: &[&Vec<u8>],
+    ) -> (String, Vec<CompletedPart>) {
+        let create_result = client
+            .create_multipart_upload()
+            .bucket(bucket)
+            .key(key)
+            .checksum_algorithm(ChecksumAlgorithm::Crc32)
+            .checksum_type(SdkChecksumType::FullObject)
+            .send()
+            .await
+            .expect("Failed to create multipart upload");
+        let upload_id = create_result.upload_id().expect("No upload_id").to_string();
+
+        let mut completed_parts = Vec::new();
+        for (index, part) in parts.iter().enumerate() {
+            let part_number = index as i32 + 1;
+            let uploaded = client
+                .upload_part()
+                .bucket(bucket)
+                .key(key)
+                .upload_id(&upload_id)
+                .part_number(part_number)
+                .body(ByteStream::from((*part).clone()))
+                .checksum_algorithm(ChecksumAlgorithm::Crc32)
+                .send()
+                .await
+                .unwrap_or_else(|e| panic!("Failed to upload part {part_number}: {e:?}"));
+            completed_parts.push(
+                CompletedPart::builder()
+                    .part_number(part_number)
+                    .e_tag(uploaded.e_tag().expect("No etag for part"))
+                    .checksum_crc32(uploaded.checksum_crc32().expect("No CRC32 for part"))
+                    .build(),
+            );
+        }
+
+        (upload_id, completed_parts)
+    }
+
+    /// A multipart upload completed with a **full-object** checksum must report
+    /// `x-amz-checksum-type: FULL_OBJECT` on both GET and HEAD, the way AWS does.
+    ///
+    /// `complete_multipart_upload` used to persist the object-level checksum
+    /// record with the pre-merge checksum type, so the MULTIPART /
+    /// INCLUDES_MULTIPART flags never reached disk. `rustfs_rio::read_checksums`
+    /// only emits the FULL_OBJECT entry inside its MULTIPART branch, so these
+    /// objects came back from GET and HEAD with no checksum-type header at all.
+    /// Found while root-causing rustfs#6825.
+    #[tokio::test]
+    async fn test_full_object_multipart_reports_full_object_checksum_type() {
+        init_logging();
+        info!("TEST: full-object multipart upload round-trips x-amz-checksum-type: FULL_OBJECT");
+
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server(vec![]).await.expect("Failed to start RustFS");
+
+        let client = create_s3_client(&env);
+        let bucket = "test-full-object-checksum-type";
+        create_bucket(&client, bucket).await.expect("Failed to create bucket");
+
+        const PART_SIZE: usize = 5 * 1024 * 1024;
+        let part1: Vec<u8> = (0..PART_SIZE).map(|i| (i % 241) as u8).collect();
+        let part2: Vec<u8> = (0..PART_SIZE).map(|i| ((i + 29) % 241) as u8).collect();
+        let content: Vec<u8> = part1.iter().chain(part2.iter()).copied().collect();
+
+        // CRC32 with an explicit FULL_OBJECT type: the object checksum is the
+        // CRC32 of the whole object, not the composite hash of the part digests.
+        let full_object_crc32 = Checksum::new_from_data(RioChecksumType::CRC32, &content)
+            .expect("crc32 checksum")
+            .encoded;
+
+        let key = "full-object-multipart.bin";
+        let (upload_id, completed_parts) = start_full_object_crc32_upload(&client, bucket, key, &[&part1, &part2]).await;
+
+        client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(&upload_id)
+            .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build())
+            // Restate the full-object intent and value on CompleteMultipartUpload,
+            // exactly as an AWS SDK client does: `x-amz-checksum-type: FULL_OBJECT`
+            // plus `x-amz-checksum-crc32`, with no `x-amz-checksum-algorithm`
+            // header (CompleteMultipartUpload has no such member).
+            .checksum_type(SdkChecksumType::FullObject)
+            .checksum_crc32(full_object_crc32.clone())
+            .send()
+            .await
+            .expect("Failed to complete multipart upload");
+
+        let head = client
+            .head_object()
+            .bucket(bucket)
+            .key(key)
+            .checksum_mode(ChecksumMode::Enabled)
+            .send()
+            .await
+            .expect("Failed to head object");
+        assert_eq!(
+            head.checksum_type(),
+            Some(&SdkChecksumType::FullObject),
+            "HeadObject must report x-amz-checksum-type: FULL_OBJECT"
+        );
+        assert_eq!(
+            head.checksum_crc32(),
+            Some(full_object_crc32.as_str()),
+            "HeadObject must report the full-object CRC32, with no -<parts> suffix"
+        );
+
+        let get = client
+            .get_object()
+            .bucket(bucket)
+            .key(key)
+            .checksum_mode(ChecksumMode::Enabled)
+            .send()
+            .await
+            .expect("Failed to get object");
+        assert_eq!(
+            get.checksum_type(),
+            Some(&SdkChecksumType::FullObject),
+            "GetObject must report x-amz-checksum-type: FULL_OBJECT"
+        );
+        assert_eq!(
+            get.checksum_crc32(),
+            Some(full_object_crc32.as_str()),
+            "GetObject must report the full-object CRC32, with no -<parts> suffix"
+        );
+
+        let body = get.body.collect().await.expect("Failed to read body").into_bytes();
+        assert_eq!(body.as_ref(), content.as_slice(), "GetObject body must match the uploaded content");
+
+        info!("PASSED: full-object multipart reports FULL_OBJECT on GET and HEAD");
+    }
+
+    /// Declaring a checksum type on CompleteMultipartUpload that contradicts the
+    /// one recorded at CreateMultipartUpload must be rejected, and rejected as a
+    /// client error (4xx), not a server error.
+    #[tokio::test]
+    async fn test_complete_multipart_rejects_contradicting_checksum_type() {
+        init_logging();
+        info!("TEST: CompleteMultipartUpload rejects a checksum type that contradicts the upload");
+
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server(vec![]).await.expect("Failed to start RustFS");
+
+        let client = create_s3_client(&env);
+        let bucket = "test-checksum-type-mismatch";
+        create_bucket(&client, bucket).await.expect("Failed to create bucket");
+
+        const PART_SIZE: usize = 5 * 1024 * 1024;
+        let part1: Vec<u8> = (0..PART_SIZE).map(|i| (i % 239) as u8).collect();
+        let part2: Vec<u8> = (0..PART_SIZE).map(|i| ((i + 31) % 239) as u8).collect();
+        let content: Vec<u8> = part1.iter().chain(part2.iter()).copied().collect();
+        let full_object_crc32 = Checksum::new_from_data(RioChecksumType::CRC32, &content)
+            .expect("crc32 checksum")
+            .encoded;
+
+        let key = "checksum-type-mismatch.bin";
+        let (upload_id, completed_parts) = start_full_object_crc32_upload(&client, bucket, key, &[&part1, &part2]).await;
+
+        let err = client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(&upload_id)
+            .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build())
+            // The upload was created as FULL_OBJECT; claiming COMPOSITE here
+            // contradicts it.
+            .checksum_type(SdkChecksumType::Composite)
+            .checksum_crc32(full_object_crc32.clone())
+            .send()
+            .await
+            .expect_err("COMPOSITE on a FULL_OBJECT upload must be rejected");
+
+        let service_err = err.into_service_error();
+        let code = service_err.meta().code().unwrap_or("<no code>").to_string();
+        let message = service_err.meta().message().unwrap_or_default().to_string();
+
+        // Before the fix the storage layer refused the combination with a generic
+        // error and the caller got `500 InternalError` -- "please try again" for a
+        // request that can only ever fail.
+        assert_eq!(
+            code, "InvalidRequest",
+            "a contradicting checksum type is a client error, got {code}: {message}"
+        );
+        assert!(
+            message.contains("FULL_OBJECT") && message.contains("COMPOSITE"),
+            "the message must name the recorded and requested types, got {message}"
+        );
+
+        // The upload is untouched by the rejected completion, so a well-formed
+        // retry on the same upload id still succeeds.
+        let listed = client
+            .list_parts()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(&upload_id)
+            .send()
+            .await
+            .expect("upload must survive the rejected completion");
+        assert_eq!(listed.parts().len(), 2, "both parts must still be listed after the rejection");
+
+        info!("PASSED: contradicting checksum type rejected as InvalidRequest");
     }
 
     /// Integration test for the AWS 2026-04 additional checksum algorithms

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -2777,14 +2777,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         );
 
         if checksum_type.is_set() {
-            checksum_type
-                .merge(rustfs_rio::ChecksumType::MULTIPART)
-                .merge(rustfs_rio::ChecksumType::INCLUDES_MULTIPART);
-            if !checksum_type.full_object_requested() {
-                checksum = rustfs_rio::Checksum::new_from_data(checksum_type, &checksum_combined)
-                    .ok_or_else(|| Error::other("checksum new_from_data failed"))?;
-            }
-            fi.checksum = Some(checksum.to_bytes(&checksum_combined));
+            fi.checksum = Some(multipart_object_checksum_record(checksum, checksum_type, &checksum_combined)?);
         }
 
         fi.metadata.remove(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM);
@@ -3528,6 +3521,43 @@ fn resolve_complete_etag(opts: &ObjectOptions, uploaded_parts: &[CompletePart])
     get_complete_multipart_md5(uploaded_parts)
 }
 
+/// Serialize the object-level checksum record that `complete_multipart_upload`
+/// persists into `FileInfo::checksum`.
+///
+/// `full_object` is the running checksum accumulated with
+/// [`rustfs_rio::Checksum::add_part`] over the parts (only meaningful when
+/// `checksum_type` asks for a full-object checksum); `combined` is the
+/// concatenation of the raw per-part digests.
+///
+/// Both shapes must be written with MULTIPART | INCLUDES_MULTIPART set. Those
+/// flags are what make [`rustfs_rio::read_checksums`] treat the object as
+/// multipart and — for the full-object shape — emit
+/// `x-amz-checksum-type: FULL_OBJECT`, which GET/HEAD echo back. `merge` takes
+/// `&mut self`, so the caller's `ChecksumType` and the copy already inside
+/// `full_object` drift apart at the merge; the full-object branch therefore has
+/// to be handed the merged type explicitly. It must *not* be rebuilt from
+/// `combined`: hashing the concatenated part digests yields the COMPOSITE value,
+/// a different number than the merged full-object one the client sent.
+fn multipart_object_checksum_record(
+    mut full_object: rustfs_rio::Checksum,
+    mut checksum_type: rustfs_rio::ChecksumType,
+    combined: &[u8],
+) -> Result<Bytes> {
+    checksum_type
+        .merge(rustfs_rio::ChecksumType::MULTIPART)
+        .merge(rustfs_rio::ChecksumType::INCLUDES_MULTIPART);
+
+    let checksum = if checksum_type.full_object_requested() {
+        full_object.checksum_type = checksum_type;
+        full_object
+    } else {
+        rustfs_rio::Checksum::new_from_data(checksum_type, combined)
+            .ok_or_else(|| Error::other("checksum new_from_data failed"))?
+    };
+
+    Ok(checksum.to_bytes(combined))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -8611,4 +8641,113 @@ mod tests {
         let computed = resolve_complete_etag(&ObjectOptions::default(), &[]);
         assert_eq!(computed, get_complete_multipart_md5(&[]));
     }
+
+    /// Accumulate the per-part digests and the running full-object checksum the
+    /// way the `complete_multipart_upload` part loop does, so the record tests
+    /// below exercise the same inputs `multipart_object_checksum_record` gets in
+    /// production.
+    fn accumulate_parts(
+        checksum_type: rustfs_rio::ChecksumType,
+        parts: &[&[u8]],
+        merge_full_object: bool,
+    ) -> (rustfs_rio::Checksum, Vec<u8>) {
+        let mut running = rustfs_rio::Checksum {
+            checksum_type,
+            ..Default::default()
+        };
+        let mut combined = Vec::new();
+        for part in parts {
+            let part_checksum = rustfs_rio::Checksum::new_from_data(checksum_type, part).expect("part checksum");
+            if merge_full_object {
+                running.add_part(&part_checksum, part.len() as i64).expect("add_part");
+            }
+            combined.extend_from_slice(part_checksum.raw.as_slice());
+        }
+        (running, combined)
+    }
+
+    /// A full-object multipart checksum must be persisted with the MULTIPART /
+    /// INCLUDES_MULTIPART flags, so the reader reports the object as multipart
+    /// and emits `x-amz-checksum-type: FULL_OBJECT`. Before the fix the record
+    /// carried the pre-merge type (the copy `Checksum` took at construction),
+    /// the reader never entered its MULTIPART branch, and GET/HEAD answered with
+    /// no checksum-type header at all.
+    #[test]
+    fn full_object_multipart_record_persists_merged_type_and_full_object_value() {
+        let part1 = b"full-object multipart part one payload".as_slice();
+        let part2 = b"full-object multipart part two payload".as_slice();
+        let checksum_type = rustfs_rio::ChecksumType::from_string_with_obj_type("crc32", "FULL_OBJECT");
+        assert!(checksum_type.full_object_requested());
+
+        let (running, combined) = accumulate_parts(checksum_type, &[part1, part2], true);
+        let record = multipart_object_checksum_record(running, checksum_type, &combined).expect("record");
+
+        let (map, is_multipart) = rustfs_rio::read_checksums(record.as_ref(), 0);
+        assert!(is_multipart, "full-object multipart record must read back as multipart");
+        assert_eq!(
+            map.get("x-amz-checksum-type").map(String::as_str),
+            Some("FULL_OBJECT"),
+            "full-object record must drive the FULL_OBJECT response header, got {map:?}"
+        );
+
+        // The persisted value stays the full-object checksum (CRC32 of the whole
+        // object), never the COMPOSITE hash of the concatenated part digests, and
+        // it carries no `-<parts>` suffix.
+        let whole: Vec<u8> = part1.iter().chain(part2.iter()).copied().collect();
+        let full_object = rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::CRC32, &whole).expect("full object");
+        let composite = rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::CRC32, &combined).expect("composite");
+        assert_ne!(full_object.encoded, composite.encoded, "test would not discriminate the two shapes");
+        assert_eq!(map.get("CRC32").map(String::as_str), Some(full_object.encoded.as_str()));
+
+        // Per-part digests are still recoverable from the record (INCLUDES_MULTIPART).
+        let (part_map, _) = rustfs_rio::read_checksums(record.as_ref(), 2);
+        let part2_checksum = rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::CRC32, part2).expect("part 2");
+        assert_eq!(part_map.get("CRC32").map(String::as_str), Some(part2_checksum.encoded.as_str()));
+    }
+
+    /// The composite shape is unchanged by the fix: value hashed from the
+    /// concatenated part digests, `-<parts>` suffix, no FULL_OBJECT entry.
+    #[test]
+    fn composite_multipart_record_keeps_part_count_suffix() {
+        let part1 = b"composite multipart part one payload".as_slice();
+        let part2 = b"composite multipart part two payload".as_slice();
+        let checksum_type = rustfs_rio::ChecksumType::from_string("sha256");
+        assert!(!checksum_type.full_object_requested());
+
+        let (running, combined) = accumulate_parts(checksum_type, &[part1, part2], false);
+        let record = multipart_object_checksum_record(running, checksum_type, &combined).expect("record");
+
+        let (map, is_multipart) = rustfs_rio::read_checksums(record.as_ref(), 0);
+        assert!(is_multipart);
+        assert_eq!(map.get("x-amz-checksum-type"), None, "composite must not claim FULL_OBJECT");
+        let composite = rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::SHA256, &combined).expect("composite");
+        assert_eq!(map.get("SHA256").map(String::as_str), Some(format!("{}-2", composite.encoded).as_str()));
+    }
+
+    /// Records written by builds from before the fix carry the bare algorithm
+    /// type with no MULTIPART flags and no trailing part block. Those bytes must
+    /// keep reading back to the same checksum value they always did — the fix
+    /// only changes what newly completed uploads write.
+    #[test]
+    fn legacy_full_object_record_without_flags_still_reads_back() {
+        let part1 = b"full-object multipart part one payload".as_slice();
+        let part2 = b"full-object multipart part two payload".as_slice();
+        let checksum_type = rustfs_rio::ChecksumType::from_string_with_obj_type("crc32", "FULL_OBJECT");
+        let (running, combined) = accumulate_parts(checksum_type, &[part1, part2], true);
+
+        // Exactly what the pre-fix code emitted: the unmerged type, serialized
+        // with the part digests offered but never appended.
+        let legacy = running.to_bytes(&combined);
+
+        let (map, is_multipart) = rustfs_rio::read_checksums(legacy.as_ref(), 0);
+        assert!(!is_multipart, "legacy record has no MULTIPART flag");
+        assert_eq!(map.get("x-amz-checksum-type"), None, "legacy record carries no type entry");
+        let whole: Vec<u8> = part1.iter().chain(part2.iter()).copied().collect();
+        let full_object = rustfs_rio::Checksum::new_from_data(rustfs_rio::ChecksumType::CRC32, &whole).expect("full object");
+        assert_eq!(
+            map.get("CRC32").map(String::as_str),
+            Some(full_object.encoded.as_str()),
+            "legacy records must keep returning their stored checksum value"
+        );
+    }
 }

rustfs/src/app/multipart_usecase.rs

@@ -104,7 +104,7 @@ use rustfs_utils::http::{
     SUFFIX_MAX_TOTAL_OBJECT_SIZE, SUFFIX_PLAINTEXT_CHECKSUM, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS,
     SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST, contains_key_str, get_consistent_str, get_header,
     get_source_scheme,
-    headers::{AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING, AMZ_STORAGE_CLASS},
+    headers::{AMZ_CHECKSUM_TYPE, AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING, AMZ_STORAGE_CLASS},
     insert_str,
 };
 use s3s::dto::{
@@ -288,6 +288,50 @@ fn has_complete_multipart_object_lock_headers(headers: &HeaderMap) -> bool {
         || has_bypass_governance_header(headers)
 }
 
+/// Reject a CompleteMultipartUpload whose `x-amz-checksum-type` contradicts the
+/// type the upload was created with, the way AWS does (`InvalidRequest`).
+///
+/// The storage layer already refuses the combination -- `complete_multipart_upload`
+/// compares `opts.want_checksum` against the algorithm/type pair recorded under
+/// `x-rustfs-multipart-checksum*` at CreateMultipartUpload -- but it refuses with a
+/// generic error that reaches the caller as `500 InternalError`, telling them to
+/// retry a request that can only ever fail. The recorded type is already in hand
+/// here, so the contradiction is answered as the client error it is.
+///
+/// Uploads created without a checksum algorithm record no type. There is nothing
+/// to contradict in that case, so the header is left alone rather than newly
+/// rejected.
+fn validate_complete_multipart_checksum_type(headers: &HeaderMap, upload_metadata: &HashMap<String, String>) -> S3Result<()> {
+    let Some(requested) = headers
+        .get(AMZ_CHECKSUM_TYPE)
+        .and_then(|value| value.to_str().ok())
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+    else {
+        return Ok(());
+    };
+
+    let Some(recorded) = upload_metadata
+        .get(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM_TYPE)
+        .map(|value| value.trim())
+        .filter(|value| !value.is_empty())
+    else {
+        return Ok(());
+    };
+
+    if requested != recorded {
+        // Routed through `ApiError` rather than the s3s error macro so this
+        // validation does not widen the direct s3s surface the s3gate migration
+        // is shrinking (scripts/check_s3s_footprint.sh); the response is identical.
+        return Err(ApiError::invalid_request(format!(
+            "The upload was created with checksum type {recorded}. The complete request must use the same checksum type, got {requested}."
+        ))
+        .into());
+    }
+
+    Ok(())
+}
+
 fn internal_object_info_lookup_opts(mut opts: ObjectOptions) -> ObjectOptions {
     opts.http_preconditions = None;
     opts
@@ -655,6 +699,7 @@ impl DefaultMultipartUsecase {
             }
             .validate_complete_multipart_ssec(&multipart_info.user_defined)?;
         }
+        validate_complete_multipart_checksum_type(&req.headers, &multipart_info.user_defined)?;
         let cache_adapter = self.object_data_cache();
         let _ = invalidate_object_data_cache_before_mutation(&cache_adapter, &bucket, &key).await;
 
@@ -1844,6 +1889,64 @@ mod tests {
     use temp_env::async_with_vars;
     use tokio::io::AsyncReadExt;
 
+    fn upload_metadata_with_checksum_type(recorded: &str) -> HashMap<String, String> {
+        HashMap::from([(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM_TYPE.to_string(), recorded.to_string())])
+    }
+
+    fn headers_with_checksum_type(requested: &str) -> HeaderMap {
+        let mut headers = HeaderMap::new();
+        headers.insert(AMZ_CHECKSUM_TYPE, HeaderValue::from_str(requested).expect("header value"));
+        headers
+    }
+
+    /// A CompleteMultipartUpload that restates the type the upload was created
+    /// with is the normal AWS SDK request shape and must pass through.
+    #[test]
+    fn complete_multipart_checksum_type_matching_the_upload_is_accepted() {
+        for kind in ["FULL_OBJECT", "COMPOSITE"] {
+            validate_complete_multipart_checksum_type(
+                &headers_with_checksum_type(kind),
+                &upload_metadata_with_checksum_type(kind),
+            )
+            .unwrap_or_else(|err| panic!("{kind} must be accepted, got {err:?}"));
+        }
+    }
+
+    /// Contradicting the recorded type is a client error, not an internal one:
+    /// before this check the storage layer refused it as a generic error and the
+    /// caller saw `500 InternalError`.
+    #[test]
+    fn complete_multipart_checksum_type_contradicting_the_upload_is_rejected() {
+        for (requested, recorded) in [("COMPOSITE", "FULL_OBJECT"), ("FULL_OBJECT", "COMPOSITE")] {
+            let err = validate_complete_multipart_checksum_type(
+                &headers_with_checksum_type(requested),
+                &upload_metadata_with_checksum_type(recorded),
+            )
+            .expect_err("contradicting checksum type must be rejected");
+            assert_eq!(*err.code(), S3ErrorCode::InvalidRequest, "must be a client error");
+            let message = err.message().unwrap_or_default().to_string();
+            assert!(
+                message.contains(requested) && message.contains(recorded),
+                "message must name both types, got {message}"
+            );
+        }
+    }
+
+    /// No header, an empty header, and an upload that recorded no checksum type
+    /// all leave the request untouched -- the check only resolves contradictions.
+    #[test]
+    fn complete_multipart_checksum_type_without_both_sides_is_accepted() {
+        validate_complete_multipart_checksum_type(&HeaderMap::new(), &upload_metadata_with_checksum_type("FULL_OBJECT"))
+            .expect("absent header is not a contradiction");
+        validate_complete_multipart_checksum_type(
+            &headers_with_checksum_type(""),
+            &upload_metadata_with_checksum_type("FULL_OBJECT"),
+        )
+        .expect("empty header is not a contradiction");
+        validate_complete_multipart_checksum_type(&headers_with_checksum_type("FULL_OBJECT"), &HashMap::new())
+            .expect("upload without a recorded checksum type is not a contradiction");
+    }
+
     fn s3_op_total(op: S3Operation) -> u64 {
         rustfs_io_metrics::s3_op_metrics_snapshot()
             .into_iter()