Designing a Resilient S3 Multipart Upload Strategy for Large Files

// snippet-2
package main

import (
	"context"
	"fmt"
	"io"
	"os"
	"sync"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
)

const (
	BucketName = "my-production-storage"
	ObjectKey  = "large-backups/db-backup-2026.tar.gz"
	ChunkSize  = 10 * 1024 * 1024 // 10MB Chunks
	NumWorkers = 5                // Parallel upload threads
)

type PartUploadJob struct {
	PartNumber int32
	Offset     int64
	Size       int64
}

type PartUploadResult struct {
	CompletedPart types.CompletedPart
	Err           error
}

func main() {
	ctx := context.TODO()
	cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion("us-east-1"))
	if err != nil {
		panic(err)
	}
	s3Client := s3.NewFromConfig(cfg)

	file, err := os.Open("db-backup-2026.tar.gz")
	if err != nil {
		panic(err)
	}
	defer file.Close()

	fileInfo, _ := file.Stat()
	fileSize := fileInfo.Size()

	// 1. Initiate Multipart Upload
	initOutput, err := s3Client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{
		Bucket: aws.String(BucketName),
		Key:    aws.String(ObjectKey),
	})
	if err != nil {
		panic(err)
	}
	uploadID := initOutput.UploadId
	fmt.Printf("Multipart Upload initiated. Upload ID: %s\n", *uploadID)

	// Calculate parts
	var parts []PartUploadJob
	var offset int64
	var partNum int32 = 1
	for offset < fileSize {
		size := int64(ChunkSize)
		if offset+size > fileSize {
			size = fileSize - offset
		}
		parts = append(parts, PartUploadJob{
			PartNumber: partNum,
			Offset:     offset,
			Size:       size,
		})
		offset += size
		partNum++
	}

	jobs := make(chan PartUploadJob, len(parts))
	results := make(chan PartUploadResult, len(parts))

	// Start Worker Pool
	var wg sync.WaitGroup
	for w := 1; w <= NumWorkers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for job := range jobs {
				// Read chunk from file
				buffer := make([]byte, job.Size)
				_, readErr := file.ReadAt(buffer, job.Offset)
				if readErr != nil && readErr != io.EOF {
					results <- PartUploadResult{Err: readErr}
					continue
				}

				// Upload chunk
				partOutput, uploadErr := s3Client.UploadPart(ctx, &s3.UploadPartInput{
					Bucket:     aws.String(BucketName),
					Key:        aws.String(ObjectKey),
					UploadId:   uploadID,
					PartNumber: aws.Int32(job.PartNumber),
					Body:       io.NopCloser(io.NewSectionReader(file, job.Offset, job.Size)),
				})
				if uploadErr != nil {
					results <- PartUploadResult{Err: uploadErr}
					continue
				}

				results <- PartUploadResult{
					CompletedPart: types.CompletedPart{
						PartNumber: aws.Int32(job.PartNumber),
						ETag:       partOutput.ETag,
					},
				}
			}
		}()
	}

	// Enqueue jobs
	for _, job := range parts {
		jobs <- job
	}
	close(jobs)

	// Wait for workers in background
	go func() {
		wg.Wait()
		close(results)
	}()

	// Collect results
	var completedParts []types.CompletedPart
	for res := range results {
		if res.Err != nil {
			// Abort on any failure to clean S3
			_, _ = s3Client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{
				Bucket:   aws.String(BucketName),
				Key:      aws.String(ObjectKey),
				UploadId: uploadID,
			})
			panic(fmt.Sprintf("Upload failed, transaction aborted: %v", res.Err))
		}
		completedParts = append(completedParts, res.CompletedPart)
	}

	// 3. Complete Multipart Upload
	_, err = s3Client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
		Bucket:   aws.String(BucketName),
		Key:      aws.String(ObjectKey),
		UploadId: uploadID,
		MultipartUpload: &types.CompletedMultipartUpload{
			Parts: completedParts,
		},
	})
	if err != nil {
		panic(err)
	}
	fmt.Println("Multipart Upload completed successfully!")
}