Repository: minio/minio (language: go)

Issue:
Support AWS S3 new checksums
## NOTE
If this case is urgent, please subscribe to [Subnet](https://min.io/pricing) so that our 24/7 support team may help you faster.

<!--- Provide a general summary of the issue in the Title above -->

## Expected Behavior


## Current Behavior

New AWS SDK v2 (2.30.0) allows configuring request/response checksum validation.

when:

```
  s3client.requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED);
  s3client.responseChecksumValidation(ResponseChecksumValidation.WHEN_REQUIRED);
```

```
DeleteObjectsRequest request = DeleteObjectsRequest.builder()
                        .overrideConfiguration(context::applyCredentialProviderOverride)
                        .requestPayer(requestPayer)
                        .bucket(bucket)
                        .delete(builder -> builder.objects(objects).quiet(true))
                        .build();

                try {
                    DeleteObjectsResponse response = client.deleteObjects(request);
                    for (S3Error error : response.errors()) {
                        failures.put("s3://%s/%s".formatted(bucket, error.key()), error.code());
                    }
                }
                catch (SdkException e) {
                    throw new TrinoFileSystemException("Error while batch deleting files", e);
                }
```

This now throws an error:

```
Caused by: software.amazon.awssdk.services.s3.model.S3Exception: Missing required header for this request: Content-Md5. (Service: S3, Status Code: 400, Request ID: 181B26ADB054EC02, Extended Request ID: 39697cb5cac07bf2a227e0cd44c7b2e9184e7598e2abe41c55eb5028b44676ca)
	at software.amazon.awssdk.protocols.xml.internal.unmarshall.AwsXmlPredicatedResponseHandler.handleErrorResponse(AwsXmlPredicatedResponseHandler.java:155)
	at software.amazon.awssdk.protocols.xml.internal.unmarshall.AwsXmlPredicatedResponseHandler.handleResponse(AwsXmlPredicatedResponseHandler.java:107)
	at software.amazon.awssdk.protocols.xml.internal.unmarshall.AwsXmlPredicatedResponseHandler.handle(AwsXmlPredicatedResponseHandler.java:84)
	at software.amazon.awssdk.protocols.xml.internal.unmarshall.AwsXmlPredicatedResponseHandler.handle(AwsXmlPredicatedResponseHandler.java:42)
	at software.amazon.awssdk.awscore.client.handler.AwsSyncClientHandler$Crc32ValidationResponseHandler.handle(AwsSyncClientHandler.java:93)
	at software.amazon.awssdk.core.internal.handler.BaseClientHandler.lambda$successTransformationResponseHandler$7(BaseClientHandler.java:279)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.HandleResponseStage.execute(HandleResponseStage.java:50)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.HandleResponseStage.execute(HandleResponseStage.java:38)
	at software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallAttemptTimeoutTrackingStage.execute(ApiCallAttemptTimeoutTrackingStage.java:74)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallAttemptTimeoutTrackingStage.execute(ApiCallAttemptTimeoutTrackingStage.java:43)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.TimeoutExceptionHandlingStage.execute(TimeoutExceptionHandlingStage.java:79)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.TimeoutExceptionHandlingStage.execute(TimeoutExceptionHandlingStage.java:41)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallAttemptMetricCollectionStage.execute(ApiCallAttemptMetricCollectionStage.java:55)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallAttemptMetricCollectionStage.execute(ApiCallAttemptMetricCollectionStage.java:39)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage.executeRequest(RetryableStage.java:93)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage.execute(RetryableStage.java:56)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.RetryableStage.execute(RetryableStage.java:36)
	at software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206)
	at software.amazon.awssdk.core.internal.http.StreamManagingStage.execute(StreamManagingStage.java:53)
	at software.amazon.awssdk.core.internal.http.StreamManagingStage.execute(StreamManagingStage.java:35)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallTimeoutTrackingStage.executeWithTimer(ApiCallTimeoutTrackingStage.java:82)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallTimeoutTrackingStage.execute(ApiCallTimeoutTrackingStage.java:62)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallTimeoutTrackingStage.execute(ApiCallTimeoutTrackingStage.java:43)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallMetricCollectionStage.execute(ApiCallMetricCollectionStage.java:50)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ApiCallMetricCollectionStage.execute(ApiCallMetricCollectionStage.java:32)
	at software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206)
	at software.amazon.awssdk.core.internal.http.pipeline.RequestPipelineBuilder$ComposingRequestPipelineStage.execute(RequestPipelineBuilder.java:206)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ExecutionFailureExceptionReportingStage.execute(ExecutionFailureExceptionReportingStage.java:37)
	at software.amazon.awssdk.core.internal.http.pipeline.stages.ExecutionFailureExceptionReportingStage.execute(ExecutionFailureExceptionReportingStage.java:26)
	at software.amazon.awssdk.core.internal.http.AmazonSyncHttpClient$RequestExecutionBuilderImpl.execute(AmazonSyncHttpClient.java:210)
	at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.invoke(BaseSyncClientHandler.java:103)
	at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.doExecute(BaseSyncClientHandler.java:173)
	at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.lambda$execute$1(BaseSyncClientHandler.java:80)
	at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.measureApiCallSuccess(BaseSyncClientHandler.java:182)
	at software.amazon.awssdk.core.internal.handler.BaseSyncClientHandler.execute(BaseSyncClientHandler.java:74)
	at software.amazon.awssdk.core.client.handler.SdkSyncClientHandler.execute(SdkSyncClientHandler.java:45)
	at software.amazon.awssdk.awscore.client.handler.AwsSyncClientHandler.execute(AwsSyncClientHandler.java:53)
	at software.amazon.awssdk.services.s3.DefaultS3Client.deleteObjects(DefaultS3Client.java:3628)
	at io.trino.filesystem.s3.S3FileSystem.deleteObjects(S3FileSystem.java:205)
	... 8 more
```

This works for S3, LocalStack but not for Minio.


## Possible Solution
<!--- Not obligatory, but suggest a fix/reason for the bug, -->
<!--- or ideas how to implement the addition or change -->

None.

## Steps to Reproduce (for bugs)
<!--- Provide a link to a live example, or an unambiguous set of steps to -->
<!--- reproduce this bug. Include code to reproduce, if relevant -->
<!--- and make sure you have followed https://github.com/minio/minio/tree/release/docs/debugging to capture relevant logs -->

1.
2.
3.
4.

## Context
<!--- How has this issue affected you? What are you trying to accomplish? -->
<!--- Providing context helps us come up with a solution that is most useful in the real world -->

## Regression
Derivation from the S3 protocol behaviour

<!-- Is this issue a regression? (Yes / No) -->
<!-- If Yes, optionally please include minio version or commit id or PR# that caused this regression, if you have these details. -->

## Your Environment
<!--- Include as many relevant details about the environment you experienced the bug in -->
* Version used (`minio --version`): RELEASE.2024-12-18T13-15-44Z
* Server setup and configuration: local docker 
* Operating System and version (`uname -a`):



Discussion / hints:
This is in AIStor only for now, alongside full object checksums.
This rules out Minio for local testing with S3 client libraries so not helpful at all that some other product implements this.
Closing this as fixed as part of the AiStor releases. 

Which makes minio unusable with AWS SDK. great job!
@harshavardhana sure, this means that we will drop Minio support from Trino as it's unusable with new S3 SDK and this will hold us back from upgrading which is unacceptable 
These things may come back into community release perhaps later this year; however, this is immediately available for customers.

> [@harshavardhana](https://github.com/harshavardhana) Sure, this means that we will drop Minio support from Trino as it's unusable with the new S3 SDK and this will hold us back from upgrading, which is unacceptable

If you are from Trino, don't hesitate to contact partners@min.io, where AiStor can be made available for Trino (starburstdata). 

We have customers already using Trino with AiStor releases in production.
@harshavardhana I'm Trino maintainer and Starburst employee, we are using Minio for integration tests in the CI/CD in parallel to S3, LocalStack, Adobe S3 mock and at the moment Minio is the only S3-complaint solution not working with the latest AWS SDK v2.

Using licensed version isn't an option as we want these CI/CD pipelines to work for non-maintainers without access to secrets and in the local development.
Understood I will get back on this @wendigo 
Can you share the test code here? (preferably self contained that I can run)
```
git clone https://github.com/trinodb/trino
git checkout serafin/minio-s3-aws-sdk
./mvnw -T2C clean install -nsu -DskipTests -Dmaven.javadoc.skip=true -Dmaven.source.skip=true -Dair.check.skip-all=true -pl '!:trino-server-rpm,!:trino-docs,!:trino-proxy,!:trino-verifier,!:trino-benchto-benchmarks'
./mvnw test -pl ':trino-filesystem-s3' -Dtest=TestS3FileSystemMinIo
```

This will start Minio docker container (`RELEASE.2024-12-18T13-15-44Z`) and run `io.trino.filesystem.s3.TestS3FileSystemMinIo` test class.

This will fail only for `deleteObjects` calls:

```
[ERROR] Errors:
[ERROR]   TestS3FileSystemMinIo.testDeleteDirectory:115->AbstractTestTrinoFileSystem.testDeleteDirectory:876 » TrinoFileSystem Error while batch deleting files
[ERROR]   TestS3FileSystemMinIo>AbstractTestS3FileSystem.testDeleteEmptyDirectoryWithDeepHierarchy:245 » TrinoFileSystem Error while batch deleting files
[ERROR]   TestS3FileSystemMinIo>AbstractTestTrinoFileSystem.testDeleteFiles:836 » TrinoFileSystem Error while batch deleting files
[ERROR]   TestS3FileSystemMinIo>AbstractTestS3FileSystem.testExistingDirectoryWithTrailingSlash:219 » TrinoFileSystem Error while batch deleting files
[INFO]
[ERROR] Tests run: 27, Failures: 0, Errors: 4, Skipped: 1
```

Single `deleteObject` works fine, `deleteObjects` (batch call) doesn't.
Another report for visibility: https://github.com/minio/minio/issues/20849

We will backport the changes for AWS S3 checksums to the community release.
@harshavardhana Great to hear! Looking forward to it

Fix patch:
diff --git a/.github/workflows/run-mint.sh b/.github/workflows/run-mint.sh
index df1bec9b7a1e0..0bbc1cbaf6150 100755
--- a/.github/workflows/run-mint.sh
+++ b/.github/workflows/run-mint.sh
@@ -7,7 +7,6 @@ export ACCESS_KEY="$2"
 export SECRET_KEY="$3"
 export JOB_NAME="$4"
 export MINT_MODE="full"
-export MINT_NO_FULL_OBJECT="true"
 
 docker system prune -f || true
 docker volume prune -f || true
@@ -39,7 +38,6 @@ docker run --rm --net=mint_default \
 	-e ACCESS_KEY="${ACCESS_KEY}" \
 	-e SECRET_KEY="${SECRET_KEY}" \
 	-e ENABLE_HTTPS=0 \
-	-e MINT_NO_FULL_OBJECT="${MINT_NO_FULL_OBJECT}" \
 	-e MINT_MODE="${MINT_MODE}" \
 	docker.io/minio/mint:edge
 
diff --git a/cmd/api-errors.go b/cmd/api-errors.go
index aa58332013800..6ea04d674e0cf 100644
--- a/cmd/api-errors.go
+++ b/cmd/api-errors.go
@@ -629,7 +629,7 @@ var errorCodes = errorCodeMap{
 	},
 	ErrMissingContentMD5: {
 		Code:           "MissingContentMD5",
-		Description:    "Missing required header for this request: Content-Md5.",
+		Description:    "Missing or invalid required header for this request: Content-Md5 or Amz-Content-Checksum",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
 	ErrMissingSecurityHeader: {
diff --git a/cmd/api-response.go b/cmd/api-response.go
index 391b6d7cc456b..db93d30d631c0 100644
--- a/cmd/api-response.go
+++ b/cmd/api-response.go
@@ -166,10 +166,11 @@ type Part struct {
 	Size         int64
 
 	// Checksum values
-	ChecksumCRC32  string `xml:"ChecksumCRC32,omitempty"`
-	ChecksumCRC32C string `xml:"ChecksumCRC32C,omitempty"`
-	ChecksumSHA1   string `xml:"ChecksumSHA1,omitempty"`
-	ChecksumSHA256 string `xml:"ChecksumSHA256,omitempty"`
+	ChecksumCRC32     string `xml:"ChecksumCRC32,omitempty"`
+	ChecksumCRC32C    string `xml:"ChecksumCRC32C,omitempty"`
+	ChecksumSHA1      string `xml:"ChecksumSHA1,omitempty"`
+	ChecksumSHA256    string `xml:"ChecksumSHA256,omitempty"`
+	ChecksumCRC64NVME string `xml:",omitempty"`
 }
 
 // ListPartsResponse - format for list parts response.
@@ -192,6 +193,8 @@ type ListPartsResponse struct {
 	IsTruncated          bool
 
 	ChecksumAlgorithm string
+	ChecksumType      string
+
 	// List of parts.
 	Parts []Part `xml:"Part"`
 }
@@ -413,10 +416,11 @@ type CompleteMultipartUploadResponse struct {
 	Key      string
 	ETag     string
 
-	ChecksumCRC32  string `xml:"ChecksumCRC32,omitempty"`
-	ChecksumCRC32C string `xml:"ChecksumCRC32C,omitempty"`
-	ChecksumSHA1   string `xml:"ChecksumSHA1,omitempty"`
-	ChecksumSHA256 string `xml:"ChecksumSHA256,omitempty"`
+	ChecksumCRC32     string `xml:"ChecksumCRC32,omitempty"`
+	ChecksumCRC32C    string `xml:"ChecksumCRC32C,omitempty"`
+	ChecksumSHA1      string `xml:"ChecksumSHA1,omitempty"`
+	ChecksumSHA256    string `xml:"ChecksumSHA256,omitempty"`
+	ChecksumCRC64NVME string `xml:",omitempty"`
 }
 
 // DeleteError structure.
@@ -793,11 +797,12 @@ func generateCompleteMultipartUploadResponse(bucket, key, location string, oi Ob
 		Bucket:   bucket,
 		Key:      key,
 		// AWS S3 quotes the ETag in XML, make sure we are compatible here.
-		ETag:           "\"" + oi.ETag + "\"",
-		ChecksumSHA1:   cs[hash.ChecksumSHA1.String()],
-		ChecksumSHA256: cs[hash.ChecksumSHA256.String()],
-		ChecksumCRC32:  cs[hash.ChecksumCRC32.String()],
-		ChecksumCRC32C: cs[hash.ChecksumCRC32C.String()],
+		ETag:              "\"" + oi.ETag + "\"",
+		ChecksumSHA1:      cs[hash.ChecksumSHA1.String()],
+		ChecksumSHA256:    cs[hash.ChecksumSHA256.String()],
+		ChecksumCRC32:     cs[hash.ChecksumCRC32.String()],
+		ChecksumCRC32C:    cs[hash.ChecksumCRC32C.String()],
+		ChecksumCRC64NVME: cs[hash.ChecksumCRC64NVME.String()],
 	}
 	return c
 }
@@ -825,6 +830,7 @@ func generateListPartsResponse(partsInfo ListPartsInfo, encodingType string) Lis
 	listPartsResponse.IsTruncated = partsInfo.IsTruncated
 	listPartsResponse.NextPartNumberMarker = partsInfo.NextPartNumberMarker
 	listPartsResponse.ChecksumAlgorithm = partsInfo.ChecksumAlgorithm
+	listPartsResponse.ChecksumType = partsInfo.ChecksumType
 
 	listPartsResponse.Parts = make([]Part, len(partsInfo.Parts))
 	for index, part := range partsInfo.Parts {
@@ -837,6 +843,7 @@ func generateListPartsResponse(partsInfo ListPartsInfo, encodingType string) Lis
 		newPart.ChecksumCRC32C = part.ChecksumCRC32C
 		newPart.ChecksumSHA1 = part.ChecksumSHA1
 		newPart.ChecksumSHA256 = part.ChecksumSHA256
+		newPart.ChecksumCRC64NVME = part.ChecksumCRC64NVME
 		listPartsResponse.Parts[index] = newPart
 	}
 	return listPartsResponse
diff --git a/cmd/bucket-handlers.go b/cmd/bucket-handlers.go
index b61c4d900f422..de442e3a89ebf 100644
--- a/cmd/bucket-handlers.go
+++ b/cmd/bucket-handlers.go
@@ -429,7 +429,7 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 
 	// Content-Md5 is required should be set
 	// http://docs.aws.amazon.com/AmazonS3/latest/API/multiobjectdeleteapi.html
-	if _, ok := r.Header[xhttp.ContentMD5]; !ok {
+	if !validateLengthAndChecksum(r) {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingContentMD5), r.URL)
 		return
 	}
diff --git a/cmd/bucket-lifecycle-handlers.go b/cmd/bucket-lifecycle-handlers.go
index 442086b9cc640..e917c9adf53dc 100644
--- a/cmd/bucket-lifecycle-handlers.go
+++ b/cmd/bucket-lifecycle-handlers.go
@@ -19,7 +19,6 @@ package cmd
 
 import (
 	"encoding/xml"
-	"io"
 	"net/http"
 	"strconv"
 	"time"
@@ -53,7 +52,7 @@ func (api objectAPIHandlers) PutBucketLifecycleHandler(w http.ResponseWriter, r
 	bucket := vars["bucket"]
 
 	// PutBucketLifecycle always needs a Content-Md5
-	if _, ok := r.Header[xhttp.ContentMD5]; !ok {
+	if !validateLengthAndChecksum(r) {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingContentMD5), r.URL)
 		return
 	}
@@ -70,7 +69,7 @@ func (api objectAPIHandlers) PutBucketLifecycleHandler(w http.ResponseWriter, r
 		return
 	}
 
-	bucketLifecycle, err := lifecycle.ParseLifecycleConfigWithID(io.LimitReader(r.Body, r.ContentLength))
+	bucketLifecycle, err := lifecycle.ParseLifecycleConfigWithID(r.Body)
 	if err != nil {
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
 		return
diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go
index cc90fc8a213d8..f278c91ce2d34 100644
--- a/cmd/bucket-replication.go
+++ b/cmd/bucket-replication.go
@@ -794,18 +794,23 @@ func putReplicationOpts(ctx context.Context, sc string, objInfo ObjectInfo, part
 			meta[k] = v
 		}
 	}
-
 	if len(objInfo.Checksum) > 0 {
 		// Add encrypted CRC to metadata for SSE-C objects.
 		if isSSEC {
 			meta[ReplicationSsecChecksumHeader] = base64.StdEncoding.EncodeToString(objInfo.Checksum)
 		} else {
-			for _, pi := range objInfo.Parts {
-				if pi.Number == partNum {
-					for k, v := range pi.Checksums {
-						meta[k] = v
+			if objInfo.isMultipart() && partNum > 0 {
+				for _, pi := range objInfo.Parts {
+					if pi.Number == partNum {
+						for k, v := range pi.Checksums { // for PutObjectPart
+							meta[k] = v
+						}
 					}
 				}
+			} else {
+				for k, v := range getCRCMeta(objInfo, 0, nil) { // for PutObject/NewMultipartUpload
+					meta[k] = v
+				}
 			}
 		}
 	}
@@ -1666,7 +1671,7 @@ func replicateObjectWithMultipart(ctx context.Context, c *minio.Core, bucket, ob
 		cHeader := http.Header{}
 		cHeader.Add(xhttp.MinIOSourceReplicationRequest, "true")
 		if !isSSEC {
-			for k, v := range partInfo.Checksums {
+			for k, v := range getCRCMeta(objInfo, partInfo.Number, nil) {
 				cHeader.Add(k, v)
 			}
 		}
@@ -1690,12 +1695,13 @@ func replicateObjectWithMultipart(ctx context.Context, c *minio.Core, bucket, ob
 			return fmt.Errorf("ssec(%t): Part size mismatch: got %d, want %d", isSSEC, pInfo.Size, size)
 		}
 		uploadedParts = append(uploadedParts, minio.CompletePart{
-			PartNumber:     pInfo.PartNumber,
-			ETag:           pInfo.ETag,
-			ChecksumCRC32:  pInfo.ChecksumCRC32,
-			ChecksumCRC32C: pInfo.ChecksumCRC32C,
-			ChecksumSHA1:   pInfo.ChecksumSHA1,
-			ChecksumSHA256: pInfo.ChecksumSHA256,
+			PartNumber:        pInfo.PartNumber,
+			ETag:              pInfo.ETag,
+			ChecksumCRC32:     pInfo.ChecksumCRC32,
+			ChecksumCRC32C:    pInfo.ChecksumCRC32C,
+			ChecksumSHA1:      pInfo.ChecksumSHA1,
+			ChecksumSHA256:    pInfo.ChecksumSHA256,
+			ChecksumCRC64NVME: pInfo.ChecksumCRC64NVME,
 		})
 	}
 	userMeta := map[string]string{
@@ -1708,6 +1714,12 @@ func replicateObjectWithMultipart(ctx context.Context, c *minio.Core, bucket, ob
 	// really big value but its okay on heavily loaded systems. This is just tail end timeout.
 	cctx, ccancel := context.WithTimeout(ctx, 10*time.Minute)
 	defer ccancel()
+
+	if len(objInfo.Checksum) > 0 {
+		for k, v := range getCRCMeta(objInfo, 0, nil) {
+			userMeta[k] = v
+		}
+	}
 	_, err = c.CompleteMultipartUpload(cctx, bucket, object, uploadID, uploadedParts, minio.PutObjectOptions{
 		UserMetadata: userMeta,
 		Internal: minio.AdvancedPutOptions{
@@ -3753,3 +3765,19 @@ type validateReplicationDestinationOptions struct {
 
 	checkReadyErr sync.Map
 }
+
+func getCRCMeta(oi ObjectInfo, partNum int, h http.Header) map[string]string {
+	meta := make(map[string]string)
+	cs := oi.decryptChecksums(partNum, h)
+	for k, v := range cs {
+		cksum := hash.NewChecksumString(k, v)
+		if cksum == nil {
+			continue
+		}
+		if cksum.Valid() {
+			meta[cksum.Type.Key()] = v
+		}
+		meta[xhttp.AmzChecksumType] = cksum.Type.ObjType()
+	}
+	return meta
+}
diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go
index a0d47849f5656..c440928c187f9 100644
--- a/cmd/erasure-multipart.go
+++ b/cmd/erasure-multipart.go
@@ -480,6 +480,7 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 
 	if opts.WantChecksum != nil && opts.WantChecksum.Type.IsSet() {
 		userDefined[hash.MinIOMultipartChecksum] = opts.WantChecksum.Type.String()
+		userDefined[hash.MinIOMultipartChecksumType] = opts.WantChecksum.Type.ObjType()
 	}
 
 	modTime := opts.MTime
@@ -508,6 +509,7 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 	return &NewMultipartUploadResult{
 		UploadID:     uploadID,
 		ChecksumAlgo: userDefined[hash.MinIOMultipartChecksum],
+		ChecksumType: userDefined[hash.MinIOMultipartChecksumType],
 	}, nil
 }
 
@@ -765,15 +767,16 @@ func (er erasureObjects) PutObjectPart(ctx context.Context, bucket, object, uplo
 
 	// Return success.
 	return PartInfo{
-		PartNumber:     partInfo.Number,
-		ETag:           partInfo.ETag,
-		LastModified:   partInfo.ModTime,
-		Size:           partInfo.Size,
-		ActualSize:     partInfo.ActualSize,
-		ChecksumCRC32:  partInfo.Checksums["CRC32"],
-		ChecksumCRC32C: partInfo.Checksums["CRC32C"],
-		ChecksumSHA1:   partInfo.Checksums["SHA1"],
-		ChecksumSHA256: partInfo.Checksums["SHA256"],
+		PartNumber:        partInfo.Number,
+		ETag:              partInfo.ETag,
+		LastModified:      partInfo.ModTime,
+		Size:              partInfo.Size,
+		ActualSize:        partInfo.ActualSize,
+		ChecksumCRC32:     partInfo.Checksums["CRC32"],
+		ChecksumCRC32C:    partInfo.Checksums["CRC32C"],
+		ChecksumSHA1:      partInfo.Checksums["SHA1"],
+		ChecksumSHA256:    partInfo.Checksums["SHA256"],
+		ChecksumCRC64NVME: partInfo.Checksums["CRC64NVME"],
 	}, nil
 }
 
@@ -895,6 +898,7 @@ func (er erasureObjects) ListObjectParts(ctx context.Context, bucket, object, up
 	result.PartNumberMarker = partNumberMarker
 	result.UserDefined = cloneMSS(fi.Metadata)
 	result.ChecksumAlgorithm = fi.Metadata[hash.MinIOMultipartChecksum]
+	result.ChecksumType = fi.Metadata[hash.MinIOMultipartChecksumType]
 
 	if maxParts == 0 {
 		return result, nil
@@ -941,15 +945,16 @@ func (er erasureObjects) ListObjectParts(ctx context.Context, bucket, object, up
 	count := maxParts
 	for _, objPart := range objParts {
 		result.Parts = append(result.Parts, PartInfo{
-			PartNumber:     objPart.Number,
-			LastModified:   objPart.ModTime,
-			ETag:           objPart.ETag,
-			Size:           objPart.Size,
-			ActualSize:     objPart.ActualSize,
-			ChecksumCRC32:  objPart.Checksums["CRC32"],
-			ChecksumCRC32C: objPart.Checksums["CRC32C"],
-			ChecksumSHA1:   objPart.Checksums["SHA1"],
-			ChecksumSHA256: objPart.Checksums["SHA256"],
+			PartNumber:        objPart.Number,
+			LastModified:      objPart.ModTime,
+			ETag:              objPart.ETag,
+			Size:              objPart.Size,
+			ActualSize:        objPart.ActualSize,
+			ChecksumCRC32:     objPart.Checksums["CRC32"],
+			ChecksumCRC32C:    objPart.Checksums["CRC32C"],
+			ChecksumSHA1:      objPart.Checksums["SHA1"],
+			ChecksumSHA256:    objPart.Checksums["SHA256"],
+			ChecksumCRC64NVME: objPart.Checksums["CRC64NVME"],
 		})
 		count--
 		if count == 0 {
@@ -1131,12 +1136,12 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 	// Checksum type set when upload started.
 	var checksumType hash.ChecksumType
 	if cs := fi.Metadata[hash.MinIOMultipartChecksum]; cs != "" {
-		checksumType = hash.NewChecksumType(cs)
+		checksumType = hash.NewChecksumType(cs, fi.Metadata[hash.MinIOMultipartChecksumType])
 		if opts.WantChecksum != nil && !opts.WantChecksum.Type.Is(checksumType) {
 			return oi, InvalidArgument{
 				Bucket: bucket,
 				Object: fi.Name,
-				Err:    fmt.Errorf("checksum type mismatch"),
+				Err:    fmt.Errorf("checksum type mismatch. got %q (%s) expected %q (%s)", checksumType.String(), checksumType.ObjType(), opts.WantChecksum.Type.String(), opts.WantChecksum.Type.ObjType()),
 			}
 		}
 	}
@@ -1216,6 +1221,9 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 	// Allocate parts similar to incoming slice.
 	fi.Parts = make([]ObjectPartInfo, len(parts))
 
+	var checksum hash.Checksum
+	checksum.Type = checksumType
+
 	// Validate each part and then commit to disk.
 	for i, part := range parts {
 		partIdx := objectPartIndex(currentFI.Parts, part.PartNumber)
@@ -1249,10 +1257,11 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 				}
 			}
 			wantCS := map[string]string{
-				hash.ChecksumCRC32.String():  part.ChecksumCRC32,
-				hash.ChecksumCRC32C.String(): part.ChecksumCRC32C,
-				hash.ChecksumSHA1.String():   part.ChecksumSHA1,
-				hash.ChecksumSHA256.String(): part.ChecksumSHA256,
+				hash.ChecksumCRC32.String():     part.ChecksumCRC32,
+				hash.ChecksumCRC32C.String():    part.ChecksumCRC32C,
+				hash.ChecksumSHA1.String():      part.ChecksumSHA1,
+				hash.ChecksumSHA256.String():    part.ChecksumSHA256,
+				hash.ChecksumCRC64NVME.String(): part.ChecksumCRC64NVME,
 			}
 			if wantCS[checksumType.String()] != crc {
 				return oi, InvalidPart{
@@ -1267,6 +1276,15 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 					PartNumber: part.PartNumber,
 				}
 			}
+			if checksumType.FullObjectRequested() {
+				if err := checksum.AddPart(*cs, expPart.ActualSize); err != nil {
+					return oi, InvalidPart{
+						PartNumber: part.PartNumber,
+						ExpETag:    "<nil>",
+						GotETag:    err.Error(),
+					}
+				}
+			}
 			checksumCombined = append(checksumCombined, cs.Raw...)
 		}
 
@@ -1297,9 +1315,19 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 	}
 
 	if opts.WantChecksum != nil {
-		err := opts.WantChecksum.Matches(checksumCombined, len(parts))
-		if err != nil {
-			return oi, err
+		if checksumType.FullObjectRequested() {
+			if opts.WantChecksum.Encoded != checksum.Encoded {
+				err := hash.ChecksumMismatch{
+					Want: opts.WantChecksum.Encoded,
+					Got:  checksum.Encoded,
+				}
+				return oi, err
+			}
+		} else {
+			err := opts.WantChecksum.Matches(checksumCombined, len(parts))
+			if err != nil {
+				return oi, err
+			}
 		}
 	}
 
@@ -1313,14 +1341,18 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 
 	if checksumType.IsSet() {
 		checksumType |= hash.ChecksumMultipart | hash.ChecksumIncludesMultipart
-		var cs *hash.Checksum
-		cs = hash.NewChecksumFromData(checksumType, checksumCombined)
-		fi.Checksum = cs.AppendTo(nil, checksumCombined)
+		checksum.Type = checksumType
+		if !checksumType.FullObjectRequested() {
+			checksum = *hash.NewChecksumFromData(checksumType, checksumCombined)
+		}
+		fi.Checksum = checksum.AppendTo(nil, checksumCombined)
 		if opts.EncryptFn != nil {
 			fi.Checksum = opts.EncryptFn("object-checksum", fi.Checksum)
 		}
 	}
-	delete(fi.Metadata, hash.MinIOMultipartChecksum) // Not needed in final object.
+	// Remove superfluous internal headers.
+	delete(fi.Metadata, hash.MinIOMultipartChecksum)
+	delete(fi.Metadata, hash.MinIOMultipartChecksumType)
 
 	// Save the final object size and modtime.
 	fi.Size = objectSize
diff --git a/cmd/object-api-datatypes.go b/cmd/object-api-datatypes.go
index cad12432bbf92..de9893c68e5f3 100644
--- a/cmd/object-api-datatypes.go
+++ b/cmd/object-api-datatypes.go
@@ -419,6 +419,9 @@ type ListPartsInfo struct {
 
 	// ChecksumAlgorithm if set
 	ChecksumAlgorithm string
+
+	// ChecksumType if set
+	ChecksumType string
 }
 
 // Lookup - returns if uploadID is valid
@@ -597,10 +600,11 @@ type PartInfo struct {
 	ActualSize int64
 
 	// Checksum values
-	ChecksumCRC32  string
-	ChecksumCRC32C string
-	ChecksumSHA1   string
-	ChecksumSHA256 string
+	ChecksumCRC32     string
+	ChecksumCRC32C    string
+	ChecksumSHA1      string
+	ChecksumSHA256    string
+	ChecksumCRC64NVME string
 }
 
 // CompletePart - represents the part that was completed, this is sent by the client
@@ -613,11 +617,14 @@ type CompletePart struct {
 	// Entity tag returned when the part was uploaded.
 	ETag string
 
+	Size int64
+
 	// Checksum values. Optional.
-	ChecksumCRC32  string
-	ChecksumCRC32C string
-	ChecksumSHA1   string
-	ChecksumSHA256 string
+	ChecksumCRC32     string
+	ChecksumCRC32C    string
+	ChecksumSHA1      string
+	ChecksumSHA256    string
+	ChecksumCRC64NVME string
 }
 
 // CompleteMultipartUpload - represents list of parts which are completed, this is sent by the
@@ -630,6 +637,7 @@ type CompleteMultipartUpload struct {
 type NewMultipartUploadResult struct {
 	UploadID     string
 	ChecksumAlgo string
+	ChecksumType string
 }
 
 type getObjectAttributesResponse struct {
@@ -641,10 +649,11 @@ type getObjectAttributesResponse struct {
 }
 
 type objectAttributesChecksum struct {
-	ChecksumCRC32  string `xml:",omitempty"`
-	ChecksumCRC32C string `xml:",omitempty"`
-	ChecksumSHA1   string `xml:",omitempty"`
-	ChecksumSHA256 string `xml:",omitempty"`
+	ChecksumCRC32     string `xml:",omitempty"`
+	ChecksumCRC32C    string `xml:",omitempty"`
+	ChecksumSHA1      string `xml:",omitempty"`
+	ChecksumSHA256    string `xml:",omitempty"`
+	ChecksumCRC64NVME string `xml:",omitempty"`
 }
 
 type objectAttributesParts struct {
@@ -657,12 +666,13 @@ type objectAttributesParts struct {
 }
 
 type objectAttributesPart struct {
-	PartNumber     int
-	Size           int64
-	ChecksumCRC32  string `xml:",omitempty"`
-	ChecksumCRC32C string `xml:",omitempty"`
-	ChecksumSHA1   string `xml:",omitempty"`
-	ChecksumSHA256 string `xml:",omitempty"`
+	PartNumber        int
+	Size              int64
+	ChecksumCRC32     string `xml:",omitempty"`
+	ChecksumCRC32C    string `xml:",omitempty"`
+	ChecksumSHA1      string `xml:",omitempty"`
+	ChecksumSHA256    string `xml:",omitempty"`
+	ChecksumCRC64NVME string `xml:",omitempty"`
 }
 
 type objectAttributesErrorResponse struct {
diff --git a/cmd/object-api-datatypes_gen.go b/cmd/object-api-datatypes_gen.go
index 5bef95e80ec0b..5b2c60a0239b5 100644
--- a/cmd/object-api-datatypes_gen.go
+++ b/cmd/object-api-datatypes_gen.go
@@ -201,13 +201,16 @@ func (z *CompleteMultipartUpload) Msgsize() (s int) {
 // MarshalMsg implements msgp.Marshaler
 func (z *CompletePart) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.Require(b, z.Msgsize())
-	// map header, size 6
+	// map header, size 8
 	// string "PartNumber"
-	o = append(o, 0x86, 0xaa, 0x50, 0x61, 0x72, 0x74, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72)
+	o = append(o, 0x88, 0xaa, 0x50, 0x61, 0x72, 0x74, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72)
 	o = msgp.AppendInt(o, z.PartNumber)
 	// string "ETag"
 	o = append(o, 0xa4, 0x45, 0x54, 0x61, 0x67)
 	o = msgp.AppendString(o, z.ETag)
+	// string "Size"
+	o = append(o, 0xa4, 0x53, 0x69, 0x7a, 0x65)
+	o = msgp.AppendInt64(o, z.Size)
 	// string "ChecksumCRC32"
 	o = append(o, 0xad, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x43, 0x52, 0x43, 0x33, 0x32)
 	o = msgp.AppendString(o, z.ChecksumCRC32)
@@ -220,6 +223,9 @@ func (z *CompletePart) MarshalMsg(b []byte) (o []byte, err error) {
 	// string "ChecksumSHA256"
 	o = append(o, 0xae, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x53, 0x48, 0x41, 0x32, 0x35, 0x36)
 	o = msgp.AppendString(o, z.ChecksumSHA256)
+	// string "ChecksumCRC64NVME"
+	o = append(o, 0xb1, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x43, 0x52, 0x43, 0x36, 0x34, 0x4e, 0x56, 0x4d, 0x45)
+	o = msgp.AppendString(o, z.ChecksumCRC64NVME)
 	return
 }
 
@@ -253,6 +259,12 @@ func (z *CompletePart) UnmarshalMsg(bts []byte) (o []byte, err error) {
 				err = msgp.WrapError(err, "ETag")
 				return
 			}
+		case "Size":
+			z.Size, bts, err = msgp.ReadInt64Bytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "Size")
+				return
+			}
 		case "ChecksumCRC32":
 			z.ChecksumCRC32, bts, err = msgp.ReadStringBytes(bts)
 			if err != nil {
@@ -277,6 +289,12 @@ func (z *CompletePart) UnmarshalMsg(bts []byte) (o []byte, err error) {
 				err = msgp.WrapError(err, "ChecksumSHA256")
 				return
 			}
+		case "ChecksumCRC64NVME":
+			z.ChecksumCRC64NVME, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "ChecksumCRC64NVME")
+				return
+			}
 		default:
 			bts, err = msgp.Skip(bts)
 			if err != nil {
@@ -291,7 +309,7 @@ func (z *CompletePart) UnmarshalMsg(bts []byte) (o []byte, err error) {
 
 // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
 func (z *CompletePart) Msgsize() (s int) {
-	s = 1 + 11 + msgp.IntSize + 5 + msgp.StringPrefixSize + len(z.ETag) + 14 + msgp.StringPrefixSize + len(z.ChecksumCRC32) + 15 + msgp.StringPrefixSize + len(z.ChecksumCRC32C) + 13 + msgp.StringPrefixSize + len(z.ChecksumSHA1) + 15 + msgp.StringPrefixSize + len(z.ChecksumSHA256)
+	s = 1 + 11 + msgp.IntSize + 5 + msgp.StringPrefixSize + len(z.ETag) + 5 + msgp.Int64Size + 14 + msgp.StringPrefixSize + len(z.ChecksumCRC32) + 15 + msgp.StringPrefixSize + len(z.ChecksumCRC32C) + 13 + msgp.StringPrefixSize + len(z.ChecksumSHA1) + 15 + msgp.StringPrefixSize + len(z.ChecksumSHA256) + 18 + msgp.StringPrefixSize + len(z.ChecksumCRC64NVME)
 	return
 }
 
@@ -956,9 +974,9 @@ func (z *ListObjectsV2Info) Msgsize() (s int) {
 // MarshalMsg implements msgp.Marshaler
 func (z *ListPartsInfo) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.Require(b, z.Msgsize())
-	// map header, size 11
+	// map header, size 12
 	// string "Bucket"
-	o = append(o, 0x8b, 0xa6, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74)
+	o = append(o, 0x8c, 0xa6, 0x42, 0x75, 0x63, 0x6b, 0x65, 0x74)
 	o = msgp.AppendString(o, z.Bucket)
 	// string "Object"
 	o = append(o, 0xa6, 0x4f, 0x62, 0x6a, 0x65, 0x63, 0x74)
@@ -1001,6 +1019,9 @@ func (z *ListPartsInfo) MarshalMsg(b []byte) (o []byte, err error) {
 	// string "ChecksumAlgorithm"
 	o = append(o, 0xb1, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x41, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d)
 	o = msgp.AppendString(o, z.ChecksumAlgorithm)
+	// string "ChecksumType"
+	o = append(o, 0xac, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x54, 0x79, 0x70, 0x65)
+	o = msgp.AppendString(o, z.ChecksumType)
 	return
 }
 
@@ -1125,6 +1146,12 @@ func (z *ListPartsInfo) UnmarshalMsg(bts []byte) (o []byte, err error) {
 				err = msgp.WrapError(err, "ChecksumAlgorithm")
 				return
 			}
+		case "ChecksumType":
+			z.ChecksumType, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "ChecksumType")
+				return
+			}
 		default:
 			bts, err = msgp.Skip(bts)
 			if err != nil {
@@ -1150,7 +1177,7 @@ func (z *ListPartsInfo) Msgsize() (s int) {
 			s += msgp.StringPrefixSize + len(za0002) + msgp.StringPrefixSize + len(za0003)
 		}
 	}
-	s += 18 + msgp.StringPrefixSize + len(z.ChecksumAlgorithm)
+	s += 18 + msgp.StringPrefixSize + len(z.ChecksumAlgorithm) + 13 + msgp.StringPrefixSize + len(z.ChecksumType)
 	return
 }
 
@@ -1279,13 +1306,16 @@ func (z *MultipartInfo) Msgsize() (s int) {
 // MarshalMsg implements msgp.Marshaler
 func (z NewMultipartUploadResult) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.Require(b, z.Msgsize())
-	// map header, size 2
+	// map header, size 3
 	// string "UploadID"
-	o = append(o, 0x82, 0xa8, 0x55, 0x70, 0x6c, 0x6f, 0x61, 0x64, 0x49, 0x44)
+	o = append(o, 0x83, 0xa8, 0x55, 0x70, 0x6c, 0x6f, 0x61, 0x64, 0x49, 0x44)
 	o = msgp.AppendString(o, z.UploadID)
 	// string "ChecksumAlgo"
 	o = append(o, 0xac, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x41, 0x6c, 0x67, 0x6f)
 	o = msgp.AppendString(o, z.ChecksumAlgo)
+	// string "ChecksumType"
+	o = append(o, 0xac, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x54, 0x79, 0x70, 0x65)
+	o = msgp.AppendString(o, z.ChecksumType)
 	return
 }
 
@@ -1319,6 +1349,12 @@ func (z *NewMultipartUploadResult) UnmarshalMsg(bts []byte) (o []byte, err error
 				err = msgp.WrapError(err, "ChecksumAlgo")
 				return
 			}
+		case "ChecksumType":
+			z.ChecksumType, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "ChecksumType")
+				return
+			}
 		default:
 			bts, err = msgp.Skip(bts)
 			if err != nil {
@@ -1333,7 +1369,7 @@ func (z *NewMultipartUploadResult) UnmarshalMsg(bts []byte) (o []byte, err error
 
 // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
 func (z NewMultipartUploadResult) Msgsize() (s int) {
-	s = 1 + 9 + msgp.StringPrefixSize + len(z.UploadID) + 13 + msgp.StringPrefixSize + len(z.ChecksumAlgo)
+	s = 1 + 9 + msgp.StringPrefixSize + len(z.UploadID) + 13 + msgp.StringPrefixSize + len(z.ChecksumAlgo) + 13 + msgp.StringPrefixSize + len(z.ChecksumType)
 	return
 }
 
@@ -1772,9 +1808,9 @@ func (z *ObjectInfo) Msgsize() (s int) {
 // MarshalMsg implements msgp.Marshaler
 func (z *PartInfo) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.Require(b, z.Msgsize())
-	// map header, size 9
+	// map header, size 10
 	// string "PartNumber"
-	o = append(o, 0x89, 0xaa, 0x50, 0x61, 0x72, 0x74, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72)
+	o = append(o, 0x8a, 0xaa, 0x50, 0x61, 0x72, 0x74, 0x4e, 0x75, 0x6d, 0x62, 0x65, 0x72)
 	o = msgp.AppendInt(o, z.PartNumber)
 	// string "LastModified"
 	o = append(o, 0xac, 0x4c, 0x61, 0x73, 0x74, 0x4d, 0x6f, 0x64, 0x69, 0x66, 0x69, 0x65, 0x64)
@@ -1800,6 +1836,9 @@ func (z *PartInfo) MarshalMsg(b []byte) (o []byte, err error) {
 	// string "ChecksumSHA256"
 	o = append(o, 0xae, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x53, 0x48, 0x41, 0x32, 0x35, 0x36)
 	o = msgp.AppendString(o, z.ChecksumSHA256)
+	// string "ChecksumCRC64NVME"
+	o = append(o, 0xb1, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x73, 0x75, 0x6d, 0x43, 0x52, 0x43, 0x36, 0x34, 0x4e, 0x56, 0x4d, 0x45)
+	o = msgp.AppendString(o, z.ChecksumCRC64NVME)
 	return
 }
 
@@ -1875,6 +1914,12 @@ func (z *PartInfo) UnmarshalMsg(bts []byte) (o []byte, err error) {
 				err = msgp.WrapError(err, "ChecksumSHA256")
 				return
 			}
+		case "ChecksumCRC64NVME":
+			z.ChecksumCRC64NVME, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "ChecksumCRC64NVME")
+				return
+			}
 		default:
 			bts, err = msgp.Skip(bts)
 			if err != nil {
@@ -1889,7 +1934,7 @@ func (z *PartInfo) UnmarshalMsg(bts []byte) (o []byte, err error) {
 
 // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
 func (z *PartInfo) Msgsize() (s int) {
-	s = 1 + 11 + msgp.IntSize + 13 + msgp.TimeSize + 5 + msgp.StringPrefixSize + len(z.ETag) + 5 + msgp.Int64Size + 11 + msgp.Int64Size + 14 + msgp.StringPrefixSize + len(z.ChecksumCRC32) + 15 + msgp.StringPrefixSize + len(z.ChecksumCRC32C) + 13 + msgp.StringPrefixSize + len(z.ChecksumSHA1) + 15 + msgp.StringPrefixSize + len(z.ChecksumSHA256)
+	s = 1 + 11 + msgp.IntSize + 13 + msgp.TimeSize + 5 + msgp.StringPrefixSize + len(z.ETag) + 5 + msgp.Int64Size + 11 + msgp.Int64Size + 14 + msgp.StringPrefixSize + len(z.ChecksumCRC32) + 15 + msgp.StringPrefixSize + len(z.ChecksumCRC32C) + 13 + msgp.StringPrefixSize + len(z.ChecksumSHA1) + 15 + msgp.StringPrefixSize + len(z.ChecksumSHA256) + 18 + msgp.StringPrefixSize + len(z.ChecksumCRC64NVME)
 	return
 }
 
diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go
index b918f24857132..31a140dc087fe 100644
--- a/cmd/object-handlers.go
+++ b/cmd/object-handlers.go
@@ -636,10 +636,11 @@ func (api objectAPIHandlers) getObjectAttributesHandler(ctx context.Context, obj
 		// AWS does not appear to append part number on this API call.
 		if len(chkSums) > 0 {
 			OA.Checksum = &objectAttributesChecksum{
-				ChecksumCRC32:  strings.Split(chkSums["CRC32"], "-")[0],
-				ChecksumCRC32C: strings.Split(chkSums["CRC32C"], "-")[0],
-				ChecksumSHA1:   strings.Split(chkSums["SHA1"], "-")[0],
-				ChecksumSHA256: strings.Split(chkSums["SHA256"], "-")[0],
+				ChecksumCRC32:     strings.Split(chkSums["CRC32"], "-")[0],
+				ChecksumCRC32C:    strings.Split(chkSums["CRC32C"], "-")[0],
+				ChecksumSHA1:      strings.Split(chkSums["SHA1"], "-")[0],
+				ChecksumSHA256:    strings.Split(chkSums["SHA256"], "-")[0],
+				ChecksumCRC64NVME: strings.Split(chkSums["CRC64NVME"], "-")[0],
 			}
 		}
 	}
@@ -678,12 +679,13 @@ func (api objectAPIHandlers) getObjectAttributesHandler(ctx context.Context, obj
 
 				OA.ObjectParts.NextPartNumberMarker = v.Number
 				OA.ObjectParts.Parts = append(OA.ObjectParts.Parts, &objectAttributesPart{
-					ChecksumSHA1:   objInfo.Parts[i].Checksums["SHA1"],
-					ChecksumSHA256: objInfo.Parts[i].Checksums["SHA256"],
-					ChecksumCRC32:  objInfo.Parts[i].Checksums["CRC32"],
-					ChecksumCRC32C: objInfo.Parts[i].Checksums["CRC32C"],
-					PartNumber:     objInfo.Parts[i].Number,
-					Size:           objInfo.Parts[i].Size,
+					ChecksumSHA1:      objInfo.Parts[i].Checksums["SHA1"],
+					ChecksumSHA256:    objInfo.Parts[i].Checksums["SHA256"],
+					ChecksumCRC32:     objInfo.Parts[i].Checksums["CRC32"],
+					ChecksumCRC32C:    objInfo.Parts[i].Checksums["CRC32C"],
+					ChecksumCRC64NVME: objInfo.Parts[i].Checksums["CRC64NVME"],
+					PartNumber:        objInfo.Parts[i].Number,
+					Size:              objInfo.Parts[i].Size,
 				})
 			}
 		}
@@ -2731,7 +2733,7 @@ func (api objectAPIHandlers) PutObjectLegalHoldHandler(w http.ResponseWriter, r
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
 		return
 	}
-	if !hasContentMD5(r.Header) {
+	if !validateLengthAndChecksum(r) {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingContentMD5), r.URL)
 		return
 	}
@@ -2741,7 +2743,7 @@ func (api objectAPIHandlers) PutObjectLegalHoldHandler(w http.ResponseWriter, r
 		return
 	}
 
-	legalHold, err := objectlock.ParseObjectLegalHold(io.LimitReader(r.Body, r.ContentLength))
+	legalHold, err := objectlock.ParseObjectLegalHold(r.Body)
 	if err != nil {
 		apiErr := errorCodes.ToAPIErr(ErrMalformedXML)
 		apiErr.Description = err.Error()
@@ -2889,7 +2891,7 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
 		return
 	}
 
-	if !hasContentMD5(r.Header) {
+	if !validateLengthAndChecksum(r) {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingContentMD5), r.URL)
 		return
 	}
diff --git a/cmd/object-multipart-handlers.go b/cmd/object-multipart-handlers.go
index bd0c01684c279..037b1ae2159bc 100644
--- a/cmd/object-multipart-handlers.go
+++ b/cmd/object-multipart-handlers.go
@@ -214,7 +214,7 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r
 		}
 	}
 
-	checksumType := hash.NewChecksumType(r.Header.Get(xhttp.AmzChecksumAlgo))
+	checksumType := hash.NewChecksumHeader(r.Header)
 	if checksumType.Is(hash.ChecksumInvalid) {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequestParameter), r.URL)
 		return
@@ -233,6 +233,9 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r
 	response := generateInitiateMultipartUploadResponse(bucket, object, res.UploadID)
 	if res.ChecksumAlgo != "" {
 		w.Header().Set(xhttp.AmzChecksumAlgo, res.ChecksumAlgo)
+		if res.ChecksumType != "" {
+			w.Header().Set(xhttp.AmzChecksumType, res.ChecksumType)
+		}
 	}
 	encodedSuccessResponse := encodeResponse(response)
 
diff --git a/cmd/utils.go b/cmd/utils.go
index be3a9c55ebf56..00724474cc531 100644
--- a/cmd/utils.go
+++ b/cmd/utils.go
@@ -20,7 +20,9 @@ package cmd
 import (
 	"bytes"
 	"context"
+	"crypto/md5"
 	"crypto/tls"
+	"encoding/base64"
 	"encoding/json"
 	"encoding/xml"
 	"errors"
@@ -254,10 +256,26 @@ func xmlDecoder(body io.Reader, v interface{}, size int64) error {
 	return err
 }
 
-// hasContentMD5 returns true if Content-MD5 header is set.
-func hasContentMD5(h http.Header) bool {
-	_, ok := h[xhttp.ContentMD5]
-	return ok
+// validateLengthAndChecksum returns if a content checksum is set,
+// and will replace r.Body with a reader that checks the provided checksum
+func validateLengthAndChecksum(r *http.Request) bool {
+	if mdFive := r.Header.Get(xhttp.ContentMD5); mdFive != "" {
+		want, err := base64.StdEncoding.DecodeString(mdFive)
+		if err != nil {
+			return false
+		}
+		r.Body = hash.NewChecker(r.Body, md5.New(), want, r.ContentLength)
+		return true
+	}
+	cs, err := hash.GetContentChecksum(r.Header)
+	if err != nil {
+		return false
+	}
+	if !cs.Type.IsSet() {
+		return false
+	}
+	r.Body = hash.NewChecker(r.Body, cs.Type.Hasher(), cs.Raw, r.ContentLength)
+	return true
 }
 
 // http://docs.aws.amazon.com/AmazonS3/latest/dev/UploadingObjects.html
diff --git a/docs/site-replication/run-replication-with-checksum-header.sh b/docs/site-replication/run-replication-with-checksum-header.sh
index 9e8040843d017..f7bf81a2222d7 100755
--- a/docs/site-replication/run-replication-with-checksum-header.sh
+++ b/docs/site-replication/run-replication-with-checksum-header.sh
@@ -159,11 +159,11 @@ DEST_OBJ_1_ETAG=$(echo "${DEST_OUT_1}" | jq '.ETag')
 DEST_OBJ_2_ETAG=$(echo "${DEST_OUT_2}" | jq '.ETag')
 
 # Check the replication of checksums and etags
-if [ "${SRC_OBJ_1_CHKSUM}" != "${DEST_OBJ_1_CHKSUM}" ]; then
+if [[ ${SRC_OBJ_1_CHKSUM} != "${DEST_OBJ_1_CHKSUM}" && ${DEST_OBJ_1_CHKSUM} != "" ]]; then
 	echo "BUG: Checksums dont match for 'obj'. Source: ${SRC_OBJ_1_CHKSUM}, Destination: ${DEST_OBJ_1_CHKSUM}"
 	exit_1
 fi
-if [ "${SRC_OBJ_2_CHKSUM}" != "${DEST_OBJ_2_CHKSUM}" ]; then
+if [[ ${SRC_OBJ_2_CHKSUM} != "${DEST_OBJ_2_CHKSUM}" && ${DEST_OBJ_2_CHKSUM} != "" ]]; then
 	echo "BUG: Checksums dont match for 'mpartobj'. Source: ${SRC_OBJ_2_CHKSUM}, Destination: ${DEST_OBJ_2_CHKSUM}"
 	exit_1
 fi
@@ -242,11 +242,11 @@ DEST_OBJ_1_ETAG=$(echo "${DEST_OUT_1}" | jq '.ETag')
 DEST_OBJ_2_ETAG=$(echo "${DEST_OUT_2}" | jq '.ETag')
 
 # Check the replication of checksums and etags
-if [ "${SRC_OBJ_1_CHKSUM}" != "${DEST_OBJ_1_CHKSUM}" ]; then
+if [[ ${SRC_OBJ_1_CHKSUM} != "${DEST_OBJ_1_CHKSUM}" && ${DEST_OBJ_1_CHKSUM} != "" ]]; then
 	echo "BUG: Checksums dont match for 'obj2'. Source: ${SRC_OBJ_1_CHKSUM}, Destination: ${DEST_OBJ_1_CHKSUM}"
 	exit_1
 fi
-if [ "${SRC_OBJ_2_CHKSUM}" != "${DEST_OBJ_2_CHKSUM}" ]; then
+if [[ ${SRC_OBJ_2_CHKSUM} != "${DEST_OBJ_2_CHKSUM}" && ${DEST_OBJ_2_CHKSUM} != "" ]]; then
 	echo "BUG: Checksums dont match for 'mpartobj2'. Source: ${SRC_OBJ_2_CHKSUM}, Destination: ${DEST_OBJ_2_CHKSUM}"
 	exit_1
 fi
diff --git a/internal/hash/checker.go b/internal/hash/checker.go
new file mode 100644
index 0000000000000..df7a8918978f0
--- /dev/null
+++ b/internal/hash/checker.go
@@ -0,0 +1,70 @@
+// Copyright (c) 2015-2024 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package hash
+
+import (
+	"bytes"
+	"errors"
+	"hash"
+	"io"
+
+	"github.com/minio/minio/internal/ioutil"
+)
+
+// Checker allows to verify the checksum of a reader.
+type Checker struct {
+	c io.Closer
+	r io.Reader
+	h hash.Hash
+
+	want []byte
+}
+
+// NewChecker ensures that content with the specified length is read from rc.
+// Calling Close on this will close upstream.
+func NewChecker(rc io.ReadCloser, h hash.Hash, wantSum []byte, length int64) *Checker {
+	return &Checker{c: rc, r: ioutil.HardLimitReader(rc, length), h: h, want: wantSum}
+}
+
+// Read satisfies io.Reader
+func (c Checker) Read(p []byte) (n int, err error) {
+	n, err = c.r.Read(p)
+	if n > 0 {
+		c.h.Write(p[:n])
+	}
+	if errors.Is(err, io.EOF) {
+		got := c.h.Sum(nil)
+		if !bytes.Equal(got, c.want) {
+			return n, ErrInvalidChecksum
+		}
+		return n, err
+	}
+	return n, err
+}
+
+// Close satisfies io.Closer
+func (c Checker) Close() error {
+	err := c.c.Close()
+	if err == nil {
+		got := c.h.Sum(nil)
+		if !bytes.Equal(got, c.want) {
+			return ErrInvalidChecksum
+		}
+	}
+	return err
+}
diff --git a/internal/hash/checksum.go b/internal/hash/checksum.go
index a169e5acf36c5..32772fc07e3a0 100644
--- a/internal/hash/checksum.go
+++ b/internal/hash/checksum.go
@@ -26,6 +26,7 @@ import (
 	"fmt"
 	"hash"
 	"hash/crc32"
+	"hash/crc64"
 	"net/http"
 	"strconv"
 	"strings"
@@ -42,6 +43,9 @@ func hashLogIf(ctx context.Context, err error) {
 // MinIOMultipartChecksum is as metadata on multipart uploads to indicate checksum type.
 const MinIOMultipartChecksum = "x-minio-multipart-checksum"
 
+// MinIOMultipartChecksumType is as metadata on multipart uploads to indicate checksum type.
+const MinIOMultipartChecksumType = "x-minio-multipart-checksum-type"
+
 // ChecksumType contains information about the checksum type.
 type ChecksumType uint32
 
@@ -65,11 +69,21 @@ const (
 	ChecksumMultipart
 	// ChecksumIncludesMultipart indicates the checksum also contains part checksums.
 	ChecksumIncludesMultipart
+	// ChecksumCRC64NVME indicates CRC64 with 0xad93d23594c93659 polynomial.
+	ChecksumCRC64NVME
+	// ChecksumFullObject indicates the checksum is of the full object,
+	// not checksum of checksums. Should only be set on ChecksumMultipart
+	ChecksumFullObject
 
 	// ChecksumNone indicates no checksum.
 	ChecksumNone ChecksumType = 0
+
+	baseTypeMask = ChecksumSHA256 | ChecksumSHA1 | ChecksumCRC32 | ChecksumCRC32C | ChecksumCRC64NVME
 )
 
+// BaseChecksumTypes is a list of all the base checksum types.
+var BaseChecksumTypes = []ChecksumType{ChecksumSHA256, ChecksumSHA1, ChecksumCRC32, ChecksumCRC64NVME, ChecksumCRC32C}
+
 // Checksum is a type and base 64 encoded value.
 type Checksum struct {
 	Type      ChecksumType
@@ -86,6 +100,11 @@ func (c ChecksumType) Is(t ChecksumType) bool {
 	return c&t == t
 }
 
+// Base returns the base checksum (if any)
+func (c ChecksumType) Base() ChecksumType {
+	return c & baseTypeMask
+}
+
 // Key returns the header key.
 // returns empty string if invalid or none.
 func (c ChecksumType) Key() string {
@@ -98,6 +117,8 @@ func (c ChecksumType) Key() string {
 		return xhttp.AmzChecksumSHA1
 	case c.Is(ChecksumSHA256):
 		return xhttp.AmzChecksumSHA256
+	case c.Is(ChecksumCRC64NVME):
+		return xhttp.AmzChecksumCRC64NVME
 	}
 	return ""
 }
@@ -113,32 +134,56 @@ func (c ChecksumType) RawByteLen() int {
 		return sha1.Size
 	case c.Is(ChecksumSHA256):
 		return sha256.Size
+	case c.Is(ChecksumCRC64NVME):
+		return crc64.Size
 	}
 	return 0
 }
 
 // IsSet returns whether the type is valid and known.
 func (c ChecksumType) IsSet() bool {
-	return !c.Is(ChecksumInvalid) && !c.Is(ChecksumNone)
+	return !c.Is(ChecksumInvalid) && !c.Base().Is(ChecksumNone)
 }
 
-// NewChecksumType returns a checksum type based on the algorithm string.
-func NewChecksumType(alg string) ChecksumType {
+// NewChecksumType returns a checksum type based on the algorithm string and obj type.
+func NewChecksumType(alg, objType string) ChecksumType {
+	full := ChecksumFullObject
+	if objType != xhttp.AmzChecksumTypeFullObject {
+		full = 0
+	}
+
 	switch strings.ToUpper(alg) {
 	case "CRC32":
-		return ChecksumCRC32
+		return ChecksumCRC32 | full
 	case "CRC32C":
-		return ChecksumCRC32C
+		return ChecksumCRC32C | full
 	case "SHA1":
+		if full != 0 {
+			return ChecksumInvalid
+		}
 		return ChecksumSHA1
 	case "SHA256":
+		if full != 0 {
+			return ChecksumInvalid
+		}
 		return ChecksumSHA256
+	case "CRC64NVME":
+		// AWS seems to ignore full value, and just assume it.
+		return ChecksumCRC64NVME
 	case "":
+		if full != 0 {
+			return ChecksumInvalid
+		}
 		return ChecksumNone
 	}
 	return ChecksumInvalid
 }
 
+// NewChecksumHeader returns a checksum type based on the algorithm string.
+func NewChecksumHeader(h http.Header) ChecksumType {
+	return NewChecksumType(h.Get(xhttp.AmzChecksumAlgo), h.Get(xhttp.AmzChecksumType))
+}
+
 // String returns the type as a string.
 func (c ChecksumType) String() string {
 	switch {
@@ -150,12 +195,35 @@ func (c ChecksumType) String() string {
 		return "SHA1"
 	case c.Is(ChecksumSHA256):
 		return "SHA256"
+	case c.Is(ChecksumCRC64NVME):
+		return "CRC64NVME"
 	case c.Is(ChecksumNone):
 		return ""
 	}
 	return "invalid"
 }
 
+// FullObjectRequested will return if the checksum type indicates full object checksum was requested.
+func (c ChecksumType) FullObjectRequested() bool {
+	return c&(ChecksumFullObject) == ChecksumFullObject || c.Is(ChecksumCRC64NVME)
+}
+
+// ObjType returns a string to return as x-amz-checksum-type.
+func (c ChecksumType) ObjType() string {
+	if c.FullObjectRequested() {
+		return xhttp.AmzChecksumTypeFullObject
+	}
+	if c.IsSet() {
+		return xhttp.AmzChecksumTypeComposite
+	}
+	return ""
+}
+
+// CanMerge will return if the checksum type indicates that checksums can be merged.
+func (c ChecksumType) CanMerge() bool {
+	return c.Is(ChecksumCRC64NVME) || c.Is(ChecksumCRC32C) || c.Is(ChecksumCRC32)
+}
+
 // Hasher returns a hasher corresponding to the checksum type.
 // Returns nil if no checksum.
 func (c ChecksumType) Hasher() hash.Hash {
@@ -168,6 +236,8 @@ func (c ChecksumType) Hasher() hash.Hash {
 		return sha1.New()
 	case c.Is(ChecksumSHA256):
 		return sha256.New()
+	case c.Is(ChecksumCRC64NVME):
+		return crc64.New(crc64Table)
 	}
 	return nil
 }
@@ -214,7 +284,11 @@ func ReadCheckSums(b []byte, part int) map[string]string {
 			if n < 0 {
 				break
 			}
-			cs = fmt.Sprintf("%s-%d", cs, t)
+			if !typ.FullObjectRequested() {
+				cs = fmt.Sprintf("%s-%d", cs, t)
+			} else if part <= 0 {
+				res[xhttp.AmzChecksumType] = xhttp.AmzChecksumTypeFullObject
+			}
 			b = b[n:]
 			if part > 0 {
 				cs = ""
@@ -322,7 +396,7 @@ func NewChecksumWithType(alg ChecksumType, value string) *Checksum {
 
 // NewChecksumString returns a new checksum from specified algorithm and base64 encoded value.
 func NewChecksumString(alg, value string) *Checksum {
-	return NewChecksumWithType(NewChecksumType(alg), value)
+	return NewChecksumWithType(NewChecksumType(alg, ""), value)
 }
 
 // AppendTo will append the checksum to b.
@@ -377,8 +451,7 @@ func (c Checksum) Valid() bool {
 	if len(c.Encoded) == 0 || c.Type.Trailing() {
 		return c.Type.Is(ChecksumNone) || c.Type.Trailing()
 	}
-	raw := c.Raw
-	return c.Type.RawByteLen() == len(raw)
+	return c.Type.RawByteLen() == len(c.Raw)
 }
 
 // Matches returns whether given content matches c.
@@ -440,6 +513,10 @@ func TransferChecksumHeader(w http.ResponseWriter, r *http.Request) {
 // AddChecksumHeader will transfer any checksum value that has been checked.
 func AddChecksumHeader(w http.ResponseWriter, c map[string]string) {
 	for k, v := range c {
+		if k == xhttp.AmzChecksumType {
+			w.Header().Set(xhttp.AmzChecksumType, v)
+			continue
+		}
 		cksum := NewChecksumString(k, v)
 		if cksum == nil {
 			continue
@@ -458,19 +535,11 @@ func GetContentChecksum(h http.Header) (*Checksum, error) {
 		var res *Checksum
 		for _, header := range trailing {
 			var duplicates bool
-			switch {
-			case strings.EqualFold(header, ChecksumCRC32C.Key()):
-				duplicates = res != nil
-				res = NewChecksumWithType(ChecksumCRC32C|ChecksumTrailing, "")
-			case strings.EqualFold(header, ChecksumCRC32.Key()):
-				duplicates = res != nil
-				res = NewChecksumWithType(ChecksumCRC32|ChecksumTrailing, "")
-			case strings.EqualFold(header, ChecksumSHA256.Key()):
-				duplicates = res != nil
-				res = NewChecksumWithType(ChecksumSHA256|ChecksumTrailing, "")
-			case strings.EqualFold(header, ChecksumSHA1.Key()):
-				duplicates = res != nil
-				res = NewChecksumWithType(ChecksumSHA1|ChecksumTrailing, "")
+			for _, t := range BaseChecksumTypes {
+				if strings.EqualFold(t.Key(), header) {
+					duplicates = res != nil
+					res = NewChecksumWithType(t|ChecksumTrailing, "")
+				}
 			}
 			if duplicates {
 				return nil, ErrInvalidChecksum
@@ -500,7 +569,13 @@ func getContentChecksum(h http.Header) (t ChecksumType, s string) {
 	t = ChecksumNone
 	alg := h.Get(xhttp.AmzChecksumAlgo)
 	if alg != "" {
-		t |= NewChecksumType(alg)
+		t |= NewChecksumHeader(h)
+		if h.Get(xhttp.AmzChecksumType) == xhttp.AmzChecksumTypeFullObject {
+			if !t.CanMerge() {
+				return ChecksumInvalid, ""
+			}
+			t |= ChecksumFullObject
+		}
 		if t.IsSet() {
 			hdr := t.Key()
 			if s = h.Get(hdr); s == "" {
@@ -519,12 +594,19 @@ func getContentChecksum(h http.Header) (t ChecksumType, s string) {
 				t = c
 				s = got
 			}
+			if h.Get(xhttp.AmzChecksumType) == xhttp.AmzChecksumTypeFullObject {
+				if !t.CanMerge() {
+					t = ChecksumInvalid
+					s = ""
+					return
+				}
+				t |= ChecksumFullObject
+			}
 			return
 		}
 	}
-	checkType(ChecksumCRC32)
-	checkType(ChecksumCRC32C)
-	checkType(ChecksumSHA1)
-	checkType(ChecksumSHA256)
+	for _, t := range BaseChecksumTypes {
+		checkType(t)
+	}
 	return t, s
 }
diff --git a/internal/hash/crc.go b/internal/hash/crc.go
new file mode 100644
index 0000000000000..6fc1d3ba81b70
--- /dev/null
+++ b/internal/hash/crc.go
@@ -0,0 +1,219 @@
+// Copyright (c) 2015-2024 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package hash
+
+import (
+	"encoding/base64"
+	"encoding/binary"
+	"fmt"
+	"hash/crc32"
+	"hash/crc64"
+	"math/bits"
+)
+
+// AddPart will merge a part checksum into the current,
+// as if the content of each was appended.
+// The size of the content that produced the second checksum must be provided.
+// Not all checksum types can be merged, use the CanMerge method to check.
+// Checksum types must match.
+func (c *Checksum) AddPart(other Checksum, size int64) error {
+	if !other.Type.CanMerge() {
+		return fmt.Errorf("checksum type cannot be merged")
+	}
+	if size == 0 {
+		return nil
+	}
+	if !c.Type.Is(other.Type.Base()) {
+		return fmt.Errorf("checksum type does not match got %s and %s", c.Type.String(), other.Type.String())
+	}
+	// If never set, just add first checksum.
+	if len(c.Raw) == 0 {
+		c.Raw = other.Raw
+		c.Encoded = other.Encoded
+		return nil
+	}
+	if !c.Valid() {
+		return fmt.Errorf("invalid base checksum")
+	}
+	if !other.Valid() {
+		return fmt.Errorf("invalid part checksum")
+	}
+
+	switch c.Type.Base() {
+	case ChecksumCRC32:
+		v := crc32Combine(crc32.IEEE, binary.BigEndian.Uint32(c.Raw), binary.BigEndian.Uint32(other.Raw), size)
+		binary.BigEndian.PutUint32(c.Raw, v)
+	case ChecksumCRC32C:
+		v := crc32Combine(crc32.Castagnoli, binary.BigEndian.Uint32(c.Raw), binary.BigEndian.Uint32(other.Raw), size)
+		binary.BigEndian.PutUint32(c.Raw, v)
+	case ChecksumCRC64NVME:
+		v := crc64Combine(bits.Reverse64(crc64NVMEPolynomial), binary.BigEndian.Uint64(c.Raw), binary.BigEndian.Uint64(other.Raw), size)
+		binary.BigEndian.PutUint64(c.Raw, v)
+	default:
+		return fmt.Errorf("unknown checksum type: %s", c.Type.String())
+	}
+	c.Encoded = base64.StdEncoding.EncodeToString(c.Raw)
+	return nil
+}
+
+const crc64NVMEPolynomial = 0xad93d23594c93659
+
+var crc64Table = crc64.MakeTable(bits.Reverse64(crc64NVMEPolynomial))
+
+// Following is ported from C to Go in 2016 by Justin Ruggles, with minimal alteration.
+// Used uint for unsigned long. Used uint32 for input arguments in order to match
+// the Go hash/crc32 package. zlib CRC32 combine (https://github.com/madler/zlib)
+// Modified for hash/crc64 by Klaus Post, 2024.
+func gf2MatrixTimes(mat []uint64, vec uint64) uint64 {
+	var sum uint64
+
+	for vec != 0 {
+		if vec&1 != 0 {
+			sum ^= mat[0]
+		}
+		vec >>= 1
+		mat = mat[1:]
+	}
+	return sum
+}
+
+func gf2MatrixSquare(square, mat []uint64) {
+	if len(square) != len(mat) {
+		panic("square matrix size mismatch")
+	}
+	for n := range mat {
+		square[n] = gf2MatrixTimes(mat, mat[n])
+	}
+}
+
+// crc32Combine returns the combined CRC-32 hash value of the two passed CRC-32
+// hash values crc1 and crc2. poly represents the generator polynomial
+// and len2 specifies the byte length that the crc2 hash covers.
+func crc32Combine(poly uint32, crc1, crc2 uint32, len2 int64) uint32 {
+	// degenerate case (also disallow negative lengths)
+	if len2 <= 0 {
+		return crc1
+	}
+
+	even := make([]uint64, 32) // even-power-of-two zeros operator
+	odd := make([]uint64, 32)  // odd-power-of-two zeros operator
+
+	// put operator for one zero bit in odd
+	odd[0] = uint64(poly) // CRC-32 polynomial
+	row := uint64(1)
+	for n := 1; n < 32; n++ {
+		odd[n] = row
+		row <<= 1
+	}
+
+	// put operator for two zero bits in even
+	gf2MatrixSquare(even, odd)
+
+	// put operator for four zero bits in odd
+	gf2MatrixSquare(odd, even)
+
+	// apply len2 zeros to crc1 (first square will put the operator for one
+	// zero byte, eight zero bits, in even)
+	crc1n := uint64(crc1)
+	for {
+		// apply zeros operator for this bit of len2
+		gf2MatrixSquare(even, odd)
+		if len2&1 != 0 {
+			crc1n = gf2MatrixTimes(even, crc1n)
+		}
+		len2 >>= 1
+
+		// if no more bits set, then done
+		if len2 == 0 {
+			break
+		}
+
+		// another iteration of the loop with odd and even swapped
+		gf2MatrixSquare(odd, even)
+		if len2&1 != 0 {
+			crc1n = gf2MatrixTimes(odd, crc1n)
+		}
+		len2 >>= 1
+
+		// if no more bits set, then done
+		if len2 == 0 {
+			break
+		}
+	}
+
+	// return combined crc
+	crc1n ^= uint64(crc2)
+	return uint32(crc1n)
+}
+
+func crc64Combine(poly uint64, crc1, crc2 uint64, len2 int64) uint64 {
+	// degenerate case (also disallow negative lengths)
+	if len2 <= 0 {
+		return crc1
+	}
+
+	even := make([]uint64, 64) // even-power-of-two zeros operator
+	odd := make([]uint64, 64)  // odd-power-of-two zeros operator
+
+	// put operator for one zero bit in odd
+	odd[0] = poly // CRC-64 polynomial
+	row := uint64(1)
+	for n := 1; n < 64; n++ {
+		odd[n] = row
+		row <<= 1
+	}
+
+	// put operator for two zero bits in even
+	gf2MatrixSquare(even, odd)
+
+	// put operator for four zero bits in odd
+	gf2MatrixSquare(odd, even)
+
+	// apply len2 zeros to crc1 (first square will put the operator for one
+	// zero byte, eight zero bits, in even)
+	crc1n := crc1
+	for {
+		// apply zeros operator for this bit of len2
+		gf2MatrixSquare(even, odd)
+		if len2&1 != 0 {
+			crc1n = gf2MatrixTimes(even, crc1n)
+		}
+		len2 >>= 1
+
+		// if no more bits set, then done
+		if len2 == 0 {
+			break
+		}
+
+		// another iteration of the loop with odd and even swapped
+		gf2MatrixSquare(odd, even)
+		if len2&1 != 0 {
+			crc1n = gf2MatrixTimes(odd, crc1n)
+		}
+		len2 >>= 1
+
+		// if no more bits set, then done
+		if len2 == 0 {
+			break
+		}
+	}
+
+	// return combined crc
+	crc1n ^= crc2
+	return crc1n
+}
diff --git a/internal/hash/reader.go b/internal/hash/reader.go
index caca03186d531..272452f06a23b 100644
--- a/internal/hash/reader.go
+++ b/internal/hash/reader.go
@@ -257,26 +257,6 @@ func (r *Reader) Read(p []byte) (int, error) {
 		r.contentHasher.Write(p[:n])
 	}
 
-	// If we have reached our expected size,
-	// do one more read to ensure we are at EOF
-	// and that any trailers have been read.
-	attempts := 0
-	for err == nil && r.size >= 0 && r.bytesRead >= r.size {
-		attempts++
-		if r.bytesRead > r.size {
-			return 0, SizeTooLarge{Want: r.size, Got: r.bytesRead}
-		}
-		var tmp [1]byte
-		var n2 int
-		n2, err = r.src.Read(tmp[:])
-		if n2 > 0 {
-			return 0, SizeTooLarge{Want: r.size, Got: r.bytesRead}
-		}
-		if attempts == 100 {
-			return 0, io.ErrNoProgress
-		}
-	}
-
 	if err == io.EOF { // Verify content SHA256, if set.
 		if r.expectedMin > 0 {
 			if r.bytesRead < r.expectedMin {
diff --git a/internal/http/headers.go b/internal/http/headers.go
index edfca9d9bf4ef..7940b34b97b6f 100644
--- a/internal/http/headers.go
+++ b/internal/http/headers.go
@@ -170,12 +170,16 @@ const (
 	MinIOServerStatus = "x-minio-server-status"
 
 	// Content Checksums
-	AmzChecksumAlgo   = "x-amz-checksum-algorithm"
-	AmzChecksumCRC32  = "x-amz-checksum-crc32"
-	AmzChecksumCRC32C = "x-amz-checksum-crc32c"
-	AmzChecksumSHA1   = "x-amz-checksum-sha1"
-	AmzChecksumSHA256 = "x-amz-checksum-sha256"
-	AmzChecksumMode   = "x-amz-checksum-mode"
+	AmzChecksumAlgo           = "x-amz-checksum-algorithm"
+	AmzChecksumCRC32          = "x-amz-checksum-crc32"
+	AmzChecksumCRC32C         = "x-amz-checksum-crc32c"
+	AmzChecksumSHA1           = "x-amz-checksum-sha1"
+	AmzChecksumSHA256         = "x-amz-checksum-sha256"
+	AmzChecksumCRC64NVME      = "x-amz-checksum-crc64nvme"
+	AmzChecksumMode           = "x-amz-checksum-mode"
+	AmzChecksumType           = "x-amz-checksum-type"
+	AmzChecksumTypeFullObject = "FULL_OBJECT"
+	AmzChecksumTypeComposite  = "COMPOSITE"
 
 	// Post Policy related
 	AmzMetaUUID = "X-Amz-Meta-Uuid"


Test patch: