Use a part reader + max 10k parts (#52)

* Use a part reader to avoid reading the files into memory

* Fix log

* Fix more vars

* Max parts is 10000

Author: cmmarslender

pkg/amazon/s3.go

@@ -1,10 +1,10 @@
 package amazon
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"log/slog"
 	"os"
 	"sync"
@@ -100,8 +100,9 @@ func MultiPartUpload(input MultiPartUploadInput) error {
 
 	// Get the total number of parts we will upload
 	numParts := getTotalNumberParts(fileSize, input.PartSize)
+	partSize := getPartSize(fileSize, numParts, input.PartSize)
 	if input.Logger != nil {
-		input.Logger.Debug("will upload file in parts", "file", input.Filepath, "parts", numParts)
+		input.Logger.Debug("will upload file in parts", "file", input.Filepath, "parts", numParts, "partSize", partSize, "fileSize", fileSize)
 	}
 
 	var (
@@ -114,8 +115,8 @@ func MultiPartUpload(input MultiPartUploadInput) error {
 	orderedParts := make([]*s3.CompletedPart, numParts)
 	for i := int64(0); i < numParts; i++ {
 		partNumber := i + 1
-		offset := i * input.PartSize
-		bytesToRead := min(input.PartSize, fileSize-offset)
+		offset := i * partSize
+		bytesToRead := min(partSize, fileSize-offset)
 
 		wg.Add(1)
 		go func(partNumber int64, bytesToRead int64, offset int64) {
@@ -125,23 +126,18 @@ func MultiPartUpload(input MultiPartUploadInput) error {
 			}()
 			defer wg.Done()
 
-			partBuffer := make([]byte, bytesToRead)
-			_, err := file.ReadAt(partBuffer, offset)
-			if err != nil {
-				ch <- err
-				return
-			}
+			partReader := io.NewSectionReader(file, offset, bytesToRead)
 
 			if input.Logger != nil {
-				input.Logger.Debug("uploading file part", "file", input.Filepath, "part", partNumber, "size", len(partBuffer))
+				input.Logger.Debug("uploading file part", "file", input.Filepath, "part", partNumber, "size", bytesToRead)
 			}
 
 			resp, err := input.Svc.UploadPart(&s3.UploadPartInput{
 				Bucket:     aws.String(input.DestinationBucket),
 				Key:        aws.String(input.DestinationKey),
 				UploadId:   &uploadID,
 				PartNumber: aws.Int64(partNumber),
-				Body:       bytes.NewReader(partBuffer),
+				Body:       partReader,
 			})
 			if err != nil {
 				ch <- fmt.Errorf("error uploading part %d : %w", partNumber, err)
@@ -155,7 +151,7 @@ func MultiPartUpload(input MultiPartUploadInput) error {
 			}
 
 			if input.Logger != nil {
-				input.Logger.Debug("finished uploading file part", "file", input.Filepath, "part", partNumber, "size", len(partBuffer))
+				input.Logger.Debug("finished uploading file part", "file", input.Filepath, "part", partNumber, "size", bytesToRead)
 			}
 		}(partNumber, bytesToRead, offset)
 	}
@@ -197,7 +193,18 @@ func min(a, b int64) int64 {
 
 func getTotalNumberParts(filesize int64, partsize int64) int64 {
 	if filesize%partsize == 0 {
-		return filesize / partsize
+		return min(10000, filesize/partsize)
+	}
+	return min(10000, filesize/partsize+1)
+}
+
+func getPartSize(filesize int64, numParts int64, defaultPartSize int64) int64 {
+	if numParts < 10000 {
+		return defaultPartSize
+	}
+	if filesize%numParts == 0 {
+		return filesize / numParts
 	}
-	return filesize/partsize + 1
+	// numParts-1 to account for any rounding (makes the parts slightly larger)
+	return filesize / (numParts - 1)
 }

pkg/amazon/s3_test.go

@@ -12,7 +12,8 @@ func TestGetTotalNumberParts(t *testing.T) {
 		{2, 1, 2}, // Test quotient where division produced a whole number
 		// Test with a couple much larger numbers
 		{104857600, 5242880, 20},       // 100MB file, 5MB chunks, 20 total chunks
-		{132070244351, 8388200, 15745}, // ~123GB file, ~8MB chunks, 15745 total chunks
+		{132070244351, 8388200, 10000}, // ~123GB file, ~8MB chunks, 15745 total chunks - limit is 10,000 so expect 10k
+		{100658400000, 8388200, 10000}, // Testing a perfect division into parts
 	}
 
 	for _, test := range tests {
@@ -22,3 +23,28 @@ func TestGetTotalNumberParts(t *testing.T) {
 		}
 	}
 }
+
+func TestGetPartSize(t *testing.T) {
+	tests := []struct {
+		FileSize int64
+		PartSize int64
+		Expected int64
+	}{
+		{5, 2, 2}, // Test quotient where division produced a non-whole number
+		{2, 1, 1}, // Test quotient where division produced a whole number
+		// Test with a couple much larger numbers
+		{104857600, 5242880, 5242880},     // 100MB file, 5MB chunks, 20 total chunks
+		{132070244351, 8388200, 13208345}, // ~123GB file, ~8MB chunks, 15745 total chunks - limit is 10,000 so expect 10k
+		{132083450000, 8388200, 13208345}, // Should be an exact filesize to fit perfectly into the parts
+		{132083450001, 8388200, 13209665}, // One too big to fit perfectly into the parts, so expect larger part sizes to accommodate
+		{132096658344, 8388200, 13210986}, // -1 too big to fit perfectly into the parts, so expect larger part sizes to accommodate
+	}
+
+	for _, test := range tests {
+		numParts := getTotalNumberParts(test.FileSize, test.PartSize)
+		result := getPartSize(test.FileSize, numParts, test.PartSize)
+		if result != test.Expected {
+			t.Errorf("operation failed for %d / %d. Expected %d, got %d", test.FileSize, test.PartSize, test.Expected, result)
+		}
+	}
+}