s3: return the multipart object checksum in the CompleteMultipartUpload response (#11101)

* s3: return the multipart object checksum in the CompleteMultipartUpload body

S3 carries the flexible-checksum members of CompleteMultipartUploadResult in
the XML body, not in response headers, so every SDK read back an empty
checksum from an upload that asked for one.

Claude-Session: https://claude.ai/code/session_01Huux1uh7JxAbf8yypMYrMk

* s3: echo the checksum algorithm and type from CreateMultipartUpload

The upload directory already records both, but the response dropped them, so a
client could not confirm which checksum its parts had to carry.

Claude-Session: https://claude.ai/code/session_01Huux1uh7JxAbf8yypMYrMk

* test: multipart upload reports the object checksum it was asked for

Covers every algorithm end to end: the create response echoes the algorithm and
type, the complete response carries the checksum, and it matches what a later
HEAD reports.

Claude-Session: https://claude.ai/code/session_01Huux1uh7JxAbf8yypMYrMk

test/s3/checksum/s3_checksum_multipart_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http/httptest"
 	"net/http/httputil"
 	"net/url"
+	"strings"
 	"testing"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -17,15 +18,18 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestMultipartCopyPreservesChecksum(t *testing.T) {
-	// aws-sdk-go-v2 sends flexible checksums as unsigned streaming trailers, which
-	// it refuses over plain HTTP, so front the HTTP endpoint with a TLS proxy.
+// newTrailerChecksumClient returns a client that can send flexible checksums.
+// aws-sdk-go-v2 sends them as unsigned streaming trailers, which it refuses over
+// plain HTTP, so front the HTTP endpoint with a TLS proxy.
+func newTrailerChecksumClient(t *testing.T) *s3.Client {
+	t.Helper()
+
 	target, err := url.Parse(defaultConfig.Endpoint)
 	require.NoError(t, err)
 
 	proxy := httputil.NewSingleHostReverseProxy(target)
 	server := httptest.NewTLSServer(proxy)
-	defer server.Close()
+	t.Cleanup(server.Close)
 
 	cfg, err := config.LoadDefaultConfig(context.Background(),
 		config.WithRegion(defaultConfig.Region),
@@ -34,10 +38,14 @@ func TestMultipartCopyPreservesChecksum(t *testing.T) {
 		config.WithHTTPClient(server.Client()),
 	)
 	require.NoError(t, err)
-	client := s3.NewFromConfig(cfg, func(o *s3.Options) {
+	return s3.NewFromConfig(cfg, func(o *s3.Options) {
 		o.BaseEndpoint = aws.String(server.URL)
 		o.UsePathStyle = true
 	})
+}
+
+func TestMultipartCopyPreservesChecksum(t *testing.T) {
+	client := newTrailerChecksumClient(t)
 
 	bucket := uniqueBucket()
 	createBucket(t, client, bucket)
@@ -122,3 +130,120 @@ func TestMultipartCopyPreservesChecksum(t *testing.T) {
 		})
 	}
 }
+
+// A multipart upload that asked for a checksum must report it in the
+// CompleteMultipartUpload response, alongside the algorithm and type that
+// CreateMultipartUpload echoed back.
+func TestMultipartUploadReturnsObjectChecksum(t *testing.T) {
+	client := newTrailerChecksumClient(t)
+
+	bucket := uniqueBucket()
+	createBucket(t, client, bucket)
+	defer cleanupBucket(t, client, bucket)
+
+	cases := []struct {
+		algorithm    types.ChecksumAlgorithm
+		expectedType types.ChecksumType
+		completeSum  func(*s3.CompleteMultipartUploadOutput) *string
+		headSum      func(*s3.HeadObjectOutput) *string
+		partSum      func(*s3.UploadPartOutput) *string
+		setPart      func(*types.CompletedPart, *string)
+	}{
+		{
+			algorithm:    types.ChecksumAlgorithmCrc32,
+			expectedType: types.ChecksumTypeComposite,
+			completeSum:  func(o *s3.CompleteMultipartUploadOutput) *string { return o.ChecksumCRC32 },
+			headSum:      func(o *s3.HeadObjectOutput) *string { return o.ChecksumCRC32 },
+			partSum:      func(o *s3.UploadPartOutput) *string { return o.ChecksumCRC32 },
+			setPart:      func(p *types.CompletedPart, v *string) { p.ChecksumCRC32 = v },
+		},
+		{
+			algorithm:    types.ChecksumAlgorithmCrc32c,
+			expectedType: types.ChecksumTypeComposite,
+			completeSum:  func(o *s3.CompleteMultipartUploadOutput) *string { return o.ChecksumCRC32C },
+			headSum:      func(o *s3.HeadObjectOutput) *string { return o.ChecksumCRC32C },
+			partSum:      func(o *s3.UploadPartOutput) *string { return o.ChecksumCRC32C },
+			setPart:      func(p *types.CompletedPart, v *string) { p.ChecksumCRC32C = v },
+		},
+		{
+			algorithm:    types.ChecksumAlgorithmCrc64nvme,
+			expectedType: types.ChecksumTypeFullObject,
+			completeSum:  func(o *s3.CompleteMultipartUploadOutput) *string { return o.ChecksumCRC64NVME },
+			headSum:      func(o *s3.HeadObjectOutput) *string { return o.ChecksumCRC64NVME },
+			partSum:      func(o *s3.UploadPartOutput) *string { return o.ChecksumCRC64NVME },
+			setPart:      func(p *types.CompletedPart, v *string) { p.ChecksumCRC64NVME = v },
+		},
+		{
+			algorithm:    types.ChecksumAlgorithmSha1,
+			expectedType: types.ChecksumTypeComposite,
+			completeSum:  func(o *s3.CompleteMultipartUploadOutput) *string { return o.ChecksumSHA1 },
+			headSum:      func(o *s3.HeadObjectOutput) *string { return o.ChecksumSHA1 },
+			partSum:      func(o *s3.UploadPartOutput) *string { return o.ChecksumSHA1 },
+			setPart:      func(p *types.CompletedPart, v *string) { p.ChecksumSHA1 = v },
+		},
+		{
+			algorithm:    types.ChecksumAlgorithmSha256,
+			expectedType: types.ChecksumTypeComposite,
+			completeSum:  func(o *s3.CompleteMultipartUploadOutput) *string { return o.ChecksumSHA256 },
+			headSum:      func(o *s3.HeadObjectOutput) *string { return o.ChecksumSHA256 },
+			partSum:      func(o *s3.UploadPartOutput) *string { return o.ChecksumSHA256 },
+			setPart:      func(p *types.CompletedPart, v *string) { p.ChecksumSHA256 = v },
+		},
+	}
+
+	// Every part but the last has to reach the 5MB multipart minimum.
+	parts := [][]byte{bytes.Repeat([]byte("a"), 5*1024*1024), []byte("tail")}
+
+	for _, tc := range cases {
+		t.Run(string(tc.algorithm), func(t *testing.T) {
+			key := "multipart-" + string(tc.algorithm)
+			create, err := client.CreateMultipartUpload(context.Background(), &s3.CreateMultipartUploadInput{
+				Bucket:            aws.String(bucket),
+				Key:               aws.String(key),
+				ChecksumAlgorithm: tc.algorithm,
+			})
+			require.NoError(t, err)
+			require.Equal(t, tc.algorithm, create.ChecksumAlgorithm)
+			require.Equal(t, tc.expectedType, create.ChecksumType)
+
+			var completed []types.CompletedPart
+			for i, data := range parts {
+				part, err := client.UploadPart(context.Background(), &s3.UploadPartInput{
+					Bucket:            aws.String(bucket),
+					Key:               aws.String(key),
+					UploadId:          create.UploadId,
+					PartNumber:        aws.Int32(int32(i + 1)),
+					Body:              bytes.NewReader(data),
+					ChecksumAlgorithm: tc.algorithm,
+				})
+				require.NoError(t, err)
+				require.NotEmpty(t, aws.ToString(tc.partSum(part)))
+
+				entry := types.CompletedPart{ETag: part.ETag, PartNumber: aws.Int32(int32(i + 1))}
+				tc.setPart(&entry, tc.partSum(part))
+				completed = append(completed, entry)
+			}
+
+			done, err := client.CompleteMultipartUpload(context.Background(), &s3.CompleteMultipartUploadInput{
+				Bucket:          aws.String(bucket),
+				Key:             aws.String(key),
+				UploadId:        create.UploadId,
+				MultipartUpload: &types.CompletedMultipartUpload{Parts: completed},
+			})
+			require.NoError(t, err)
+			require.NotEmpty(t, aws.ToString(tc.completeSum(done)))
+			require.Equal(t, tc.expectedType, done.ChecksumType)
+			if tc.expectedType == types.ChecksumTypeComposite {
+				require.True(t, strings.HasSuffix(aws.ToString(tc.completeSum(done)), fmt.Sprintf("-%d", len(parts))))
+			}
+
+			head, err := client.HeadObject(context.Background(), &s3.HeadObjectInput{
+				Bucket:       aws.String(bucket),
+				Key:          aws.String(key),
+				ChecksumMode: types.ChecksumModeEnabled,
+			})
+			require.NoError(t, err)
+			require.Equal(t, aws.ToString(tc.completeSum(done)), aws.ToString(tc.headSum(head)))
+		})
+	}
+}