Merge pull request #1472 from Altinity/hotfix_putMultipartCRC32

refactoring for CRC32 to properly size calculation,

Author: Slach

ChangeLog.md

@@ -5,6 +5,9 @@ NEW FEATURES
 - add `gcs.allow_multipart_download` (env `GCS_ALLOW_MULTIPART_DOWNLOAD`, default `false`) with `gcs.download_concurrency` — download each file as parallel range reads (part size is `chunk_size`) into a temporary file, mirrors `s3.allow_multipart_download`, fix [#1028](https://github.com/Altinity/clickhouse-backup/issues/1028)
 - add `general.disable_environment_override` (settable ONLY in the config file, no environment variable name on purpose, default `false`) — when `true`, config values come only from the config file: all environment variables and the `--env` CLI flag are ignored during config loading; protects against accidental overrides such as Kubernetes service-discovery variables, fix [#1079](https://github.com/Altinity/clickhouse-backup/issues/1079)
 
+BUG FIXES
+- fix silent truncation of compressed backup uploads to S3 when `s3.check_sum_algorithm: CRC32` (Object Lock buckets): the multipart upload stopped at the declared size (sum of raw file sizes), dropping the tar headers/padding/trailer tail of the archive; the stream is now consumed until EOF, fix [#1471](https://github.com/Altinity/clickhouse-backup/issues/1471)
+
 # v2.7.4
 
 NEW FEATURES

pkg/storage/s3.go

@@ -448,7 +448,7 @@ func (s *S3) PutFileAbsolute(ctx context.Context, key string, r io.ReadCloser, l
 
 	// transfermanager.UploadObject sends the part checksum as a trailer, which S3 Object Lock rejects on UploadPart. Fall back to manual multipart with per-part x-amz-checksum-crc32 header, fix https://github.com/Altinity/clickhouse-backup/issues/829
 	if s.Config.CheckSumAlgorithm == string(s3types.ChecksumAlgorithmCrc32) && localSize > partSize {
-		return s.putFileMultipartCRC32(ctx, &params, r, localSize, partSize)
+		return s.putFileMultipartCRC32(ctx, s.client, &params, r, localSize, partSize)
 	}
 
 	uploadInput := &transfermanager.UploadObjectInput{
@@ -480,15 +480,23 @@ func (s *S3) PutFileAbsolute(ctx context.Context, key string, r io.ReadCloser, l
 	return nil
 }
 
-func (s *S3) putFileMultipartCRC32(ctx context.Context, putParams *s3.PutObjectInput, r io.Reader, localSize, partSize int64) error {
+// s3MultipartAPI is the subset of *s3.Client used by putFileMultipartCRC32, extracted for unit testing
+type s3MultipartAPI interface {
+	CreateMultipartUpload(ctx context.Context, params *s3.CreateMultipartUploadInput, optFns ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error)
+	UploadPart(ctx context.Context, params *s3.UploadPartInput, optFns ...func(*s3.Options)) (*s3.UploadPartOutput, error)
+	CompleteMultipartUpload(ctx context.Context, params *s3.CompleteMultipartUploadInput, optFns ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error)
+	AbortMultipartUpload(ctx context.Context, params *s3.AbortMultipartUploadInput, optFns ...func(*s3.Options)) (*s3.AbortMultipartUploadOutput, error)
+}
+
+func (s *S3) putFileMultipartCRC32(ctx context.Context, s3MultipartAPIClient s3MultipartAPI, putParams *s3.PutObjectInput, r io.Reader, localSize, partSize int64) error {
 	createParams := &s3.CreateMultipartUploadInput{
 		Bucket:       putParams.Bucket,
 		Key:          putParams.Key,
 		StorageClass: putParams.StorageClass,
 	}
 	s.enrichCreateMultipartUploadParams(createParams)
 
-	initResp, err := s.client.CreateMultipartUpload(ctx, createParams)
+	initResp, err := s3MultipartAPIClient.CreateMultipartUpload(ctx, createParams)
 	if err != nil {
 		return errors.Wrap(err, "S3 putFileMultipartCRC32 CreateMultipartUpload")
 	}
@@ -503,54 +511,57 @@ func (s *S3) putFileMultipartCRC32(ctx context.Context, putParams *s3.PutObjectI
 		if s.Config.RequestPayer != "" {
 			abortParams.RequestPayer = s3types.RequestPayer(s.Config.RequestPayer)
 		}
-		if _, abortErr := s.client.AbortMultipartUpload(context.Background(), abortParams); abortErr != nil {
+		if _, abortErr := s3MultipartAPIClient.AbortMultipartUpload(context.Background(), abortParams); abortErr != nil {
 			return errors.Wrapf(cause, "aborting putFileMultipartCRC32 multipart upload: %v, original error was", abortErr)
 		}
 		return cause
 	}
 
+	// localSize is only a capacity hint: for compressed streams UploadCompressedStream declares the
+	// sum of raw file sizes, while the actual tar stream is larger (headers/padding/trailer), so read
+	// until EOF instead of stopping at localSize, fix https://github.com/Altinity/clickhouse-backup/issues/1471
 	buf := make([]byte, partSize)
 	parts := make([]s3types.CompletedPart, 0, (localSize+partSize-1)/partSize)
 	var partNumber int32 = 1
-	remaining := localSize
-	for remaining > 0 {
-		toRead := partSize
-		if remaining < toRead {
-			toRead = remaining
-		}
-		if _, readErr := io.ReadFull(r, buf[:toRead]); readErr != nil {
+	for {
+		n, readErr := io.ReadFull(r, buf)
+		if readErr != nil && readErr != io.EOF && readErr != io.ErrUnexpectedEOF {
 			return abort(errors.Wrapf(readErr, "S3 putFileMultipartCRC32 read part=%d", partNumber))
 		}
-		h := crc32.NewIEEE()
-		if _, writeErr := h.Write(buf[:toRead]); writeErr != nil {
-			return errors.Wrapf(writeErr, "S3 putFileMultipartCRC32 write part=%d", partNumber)
-		}
-		uploadParams := &s3.UploadPartInput{
-			Bucket:            putParams.Bucket,
-			Key:               putParams.Key,
-			UploadId:          uploadID,
-			PartNumber:        aws.Int32(partNumber),
-			Body:              bytes.NewReader(buf[:toRead]),
-			ChecksumAlgorithm: s3types.ChecksumAlgorithmCrc32,
-			ChecksumCRC32:     aws.String(base64.StdEncoding.EncodeToString(h.Sum(nil))),
-		}
-		if s.Config.RequestPayer != "" {
-			uploadParams.RequestPayer = s3types.RequestPayer(s.Config.RequestPayer)
+		if n > 0 {
+			h := crc32.NewIEEE()
+			if _, writeErr := h.Write(buf[:n]); writeErr != nil {
+				return errors.Wrapf(writeErr, "S3 putFileMultipartCRC32 write part=%d", partNumber)
+			}
+			uploadParams := &s3.UploadPartInput{
+				Bucket:            putParams.Bucket,
+				Key:               putParams.Key,
+				UploadId:          uploadID,
+				PartNumber:        aws.Int32(partNumber),
+				Body:              bytes.NewReader(buf[:n]),
+				ChecksumAlgorithm: s3types.ChecksumAlgorithmCrc32,
+				ChecksumCRC32:     aws.String(base64.StdEncoding.EncodeToString(h.Sum(nil))),
+			}
+			if s.Config.RequestPayer != "" {
+				uploadParams.RequestPayer = s3types.RequestPayer(s.Config.RequestPayer)
+			}
+			partResp, uploadErr := s3MultipartAPIClient.UploadPart(ctx, uploadParams)
+			if uploadErr != nil {
+				return abort(errors.Wrapf(uploadErr, "S3 putFileMultipartCRC32 UploadPart part=%d", partNumber))
+			}
+			parts = append(parts, s3types.CompletedPart{
+				ETag:          partResp.ETag,
+				PartNumber:    aws.Int32(partNumber),
+				ChecksumCRC32: partResp.ChecksumCRC32,
+			})
+			partNumber++
 		}
-		partResp, uploadErr := s.client.UploadPart(ctx, uploadParams)
-		if uploadErr != nil {
-			return abort(errors.Wrapf(uploadErr, "S3 putFileMultipartCRC32 UploadPart part=%d", partNumber))
+		if readErr != nil {
+			break
 		}
-		parts = append(parts, s3types.CompletedPart{
-			ETag:          partResp.ETag,
-			PartNumber:    aws.Int32(partNumber),
-			ChecksumCRC32: partResp.ChecksumCRC32,
-		})
-		partNumber++
-		remaining -= toRead
 	}
 
-	if _, completeErr := s.client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
+	if _, completeErr := s3MultipartAPIClient.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
 		Bucket:          putParams.Bucket,
 		Key:             putParams.Key,
 		UploadId:        uploadID,

pkg/storage/s3_test.go

@@ -1,13 +1,80 @@
 package storage
 
 import (
+	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"testing"
 
+	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/smithy-go"
 )
 
+type fakeS3MultipartAPI struct {
+	uploaded  bytes.Buffer
+	completed bool
+	aborted   bool
+}
+
+func (f *fakeS3MultipartAPI) CreateMultipartUpload(_ context.Context, _ *s3.CreateMultipartUploadInput, _ ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error) {
+	return &s3.CreateMultipartUploadOutput{UploadId: aws.String("test-upload-id")}, nil
+}
+
+func (f *fakeS3MultipartAPI) UploadPart(_ context.Context, params *s3.UploadPartInput, _ ...func(*s3.Options)) (*s3.UploadPartOutput, error) {
+	if _, err := f.uploaded.ReadFrom(params.Body); err != nil {
+		return nil, err
+	}
+	return &s3.UploadPartOutput{
+		ETag:          aws.String(fmt.Sprintf("etag-%d", *params.PartNumber)),
+		ChecksumCRC32: params.ChecksumCRC32,
+	}, nil
+}
+
+func (f *fakeS3MultipartAPI) CompleteMultipartUpload(_ context.Context, _ *s3.CompleteMultipartUploadInput, _ ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error) {
+	f.completed = true
+	return &s3.CompleteMultipartUploadOutput{}, nil
+}
+
+func (f *fakeS3MultipartAPI) AbortMultipartUpload(_ context.Context, _ *s3.AbortMultipartUploadInput, _ ...func(*s3.Options)) (*s3.AbortMultipartUploadOutput, error) {
+	f.aborted = true
+	return &s3.AbortMultipartUploadOutput{}, nil
+}
+
+// TestPutFileMultipartCRC32ReadsUntilEOF - https://github.com/Altinity/clickhouse-backup/issues/1471
+// UploadCompressedStream declares localSize as the sum of raw on-disk file sizes, but the actual
+// tar stream is larger (512-byte header per file, 512-byte content padding, 1024-byte trailer).
+// The upload must consume the stream until EOF instead of stopping at the declared size,
+// otherwise the archive tail is silently truncated.
+func TestPutFileMultipartCRC32ReadsUntilEOF(t *testing.T) {
+	const partSize = 8 * 1024
+	const declaredSize = 3*partSize + 100 // sum of raw file sizes, what UploadCompressedStream computes
+	const trueSize = declaredSize + 2560  // + tar headers/padding/trailer
+
+	data := make([]byte, trueSize)
+	for i := range data {
+		data[i] = byte(i % 251)
+	}
+
+	fake := &fakeS3MultipartAPI{}
+	s := &S3{Config: &config.S3Config{Bucket: "bucket"}}
+	params := &s3.PutObjectInput{Bucket: aws.String("bucket"), Key: aws.String("backup/shadow/default/table/default_all_1_1_0.tar")}
+	if err := s.putFileMultipartCRC32(context.Background(), fake, params, bytes.NewReader(data), declaredSize, partSize); err != nil {
+		t.Fatalf("putFileMultipartCRC32 error: %v", err)
+	}
+	if !fake.completed {
+		t.Fatal("CompleteMultipartUpload was not called")
+	}
+	if fake.aborted {
+		t.Fatal("AbortMultipartUpload was called unexpectedly")
+	}
+	if !bytes.Equal(fake.uploaded.Bytes(), data) {
+		t.Fatalf("uploaded %d bytes, want %d - the tail of the stream was truncated", fake.uploaded.Len(), trueSize)
+	}
+}
+
 // TestS3CopySource - `x-amz-copy-source` must be URL-encoded, keys with literal `%`/`#`/non-ASCII
 // (TablePathEncode names in shadow paths) were decoded server side into a different key, so
 // CopyObject failed with NoSuchKey while StatFile on the same key succeeded