perf(upload): skip suite objects the bucket already holds

pkg/upload/s3.go

@@ -113,6 +113,7 @@ func (u *s3Uploader) Preflight(ctx context.Context) error {
 type uploadJob struct {
 	localPath string
 	key       string
+	size      int64
 }
 
 // Upload walks localDir and uploads all files to S3 under the configured prefix.
@@ -149,6 +150,7 @@ func (u *s3Uploader) collectJobs(localDir, keyPrefix string) ([]uploadJob, error
 		jobs = append(jobs, uploadJob{
 			localPath: path,
 			key:       keyPrefix + "/" + filepath.ToSlash(relPath),
+			size:      info.Size(),
 		})
 
 		return nil
@@ -302,7 +304,9 @@ func (u *s3Uploader) resolvePrefix(baseName string) string {
 }
 
 // UploadSuiteDir uploads all files in a suite directory to S3 under
-// prefix + "/suites/" + dirname.
+// prefix + "/suites/" + dirname. Objects already there at the same size are
+// skipped: a suite hash is a digest of its step file contents, so a given key
+// under it always holds the same bytes and re-sending them is pure waste.
 func (u *s3Uploader) UploadSuiteDir(ctx context.Context, localSuiteDir string) error {
 	prefix := u.cfg.Prefix
 	if prefix == "" {
@@ -316,7 +320,77 @@ func (u *s3Uploader) UploadSuiteDir(ctx context.Context, localSuiteDir string) e
 		return fmt.Errorf("walking suite directory %s: %w", localSuiteDir, err)
 	}
 
-	return u.uploadJobs(ctx, jobs, keyPrefix)
+	remote, err := u.listSizes(ctx, keyPrefix)
+	if err != nil {
+		// A listing failure only costs us the optimisation, so fall back to
+		// uploading everything rather than failing the suite.
+		u.log.WithError(err).WithField("prefix", keyPrefix).
+			Warn("Failed to list existing suite objects; uploading all files")
+
+		return u.uploadJobs(ctx, jobs, keyPrefix)
+	}
+
+	pending := make([]uploadJob, 0, len(jobs))
+
+	var skipped, skippedBytes int64
+
+	for _, job := range jobs {
+		// summary.json is rewritten every run — metadata labels change without
+		// affecting the suite hash — so it is the one file never skipped.
+		if size, ok := remote[job.key]; ok && size == job.size &&
+			filepath.Base(job.key) != suiteSummaryFile {
+			skipped++
+			skippedBytes += job.size
+
+			continue
+		}
+
+		pending = append(pending, job)
+	}
+
+	if skipped > 0 {
+		u.log.WithFields(logrus.Fields{
+			"skipped":       skipped,
+			"skipped_bytes": skippedBytes,
+			"pending":       len(pending),
+			"prefix":        keyPrefix,
+		}).Info("Skipping suite objects already present at the same size")
+	}
+
+	return u.uploadJobs(ctx, pending, keyPrefix)
+}
+
+// suiteSummaryFile is the one file in a suite directory that is rewritten
+// rather than content-addressed.
+const suiteSummaryFile = "summary.json"
+
+// listSizes returns the size of every object under prefix, keyed by full key.
+func (u *s3Uploader) listSizes(
+	ctx context.Context, prefix string,
+) (map[string]int64, error) {
+	sizes := make(map[string]int64)
+
+	paginator := s3.NewListObjectsV2Paginator(u.client, &s3.ListObjectsV2Input{
+		Bucket: aws.String(u.cfg.Bucket),
+		Prefix: aws.String(prefix + "/"),
+	})
+
+	for paginator.HasMorePages() {
+		page, err := paginator.NextPage(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("listing objects under %q: %w", prefix, err)
+		}
+
+		for _, obj := range page.Contents {
+			if obj.Key == nil || obj.Size == nil {
+				continue
+			}
+
+			sizes[*obj.Key] = *obj.Size
+		}
+	}
+
+	return sizes, nil
 }
 
 // detectContentType returns a MIME type based on file extension.

pkg/upload/s3_test.go

@@ -6,6 +6,8 @@ import (
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"strconv"
+	"strings"
 	"sync"
 	"testing"
 
@@ -96,6 +98,21 @@ type fakeS3 struct {
 	singlePuts []string
 	multiparts []string
 	parts      int
+	// stored mimics bucket contents so ListObjectsV2 can report them back.
+	stored    map[string]int64
+	partBytes map[string]int64
+}
+
+func newFakeS3() *fakeS3 {
+	return &fakeS3{
+		stored:    make(map[string]int64),
+		partBytes: make(map[string]int64),
+	}
+}
+
+// store records an object under its bucket-relative key. Callers hold f.mu.
+func (f *fakeS3) store(path string, size int64) {
+	f.stored[strings.TrimPrefix(path, "/b/")] = size
 }
 
 func (f *fakeS3) handler() http.Handler {
@@ -107,24 +124,42 @@ func (f *fakeS3) handler() http.Handler {
 		defer f.mu.Unlock()
 
 		switch {
+		case r.Method == http.MethodGet && q.Get("list-type") == "2":
+			w.Header().Set("Content-Type", "application/xml")
+
+			body := `<ListBucketResult><IsTruncated>false</IsTruncated>`
+
+			for k, size := range f.stored {
+				if strings.HasPrefix(k, q.Get("prefix")) {
+					body += `<Contents><Key>` + k +
+						`</Key><Size>` + strconv.FormatInt(size, 10) + `</Size></Contents>`
+				}
+			}
+
+			_, _ = w.Write([]byte(body + `</ListBucketResult>`))
 		case r.Method == http.MethodPost && q.Has("uploads"):
 			w.Header().Set("Content-Type", "application/xml")
 			f.multiparts = append(f.multiparts, key)
 			_, _ = w.Write([]byte(`<InitiateMultipartUploadResult>` +
 				`<Bucket>b</Bucket><Key>` + key + `</Key><UploadId>up-1</UploadId>` +
 				`</InitiateMultipartUploadResult>`))
 		case r.Method == http.MethodPut && q.Has("partNumber"):
-			_, _ = io.Copy(io.Discard, r.Body)
+			n, _ := io.Copy(io.Discard, r.Body)
 			f.parts++
+			f.partBytes[key] += n
 			w.Header().Set("ETag", `"etag"`)
 		case r.Method == http.MethodPost && q.Has("uploadId"):
 			w.Header().Set("Content-Type", "application/xml")
+
+			f.store(key, f.partBytes[key])
+
 			_, _ = w.Write([]byte(`<CompleteMultipartUploadResult>` +
 				`<Bucket>b</Bucket><Key>` + key + `</Key><ETag>"etag"</ETag>` +
 				`</CompleteMultipartUploadResult>`))
 		case r.Method == http.MethodPut:
-			_, _ = io.Copy(io.Discard, r.Body)
+			n, _ := io.Copy(io.Discard, r.Body)
 			f.singlePuts = append(f.singlePuts, key)
+			f.store(key, n)
 			w.Header().Set("ETag", `"etag"`)
 		default:
 			w.WriteHeader(http.StatusNotImplemented)
@@ -136,7 +171,7 @@ func (f *fakeS3) handler() http.Handler {
 // PutObject caps at 5 GiB and fails with EntityTooLarge on the multi-GB
 // pre-run bundles a stateful suite carries.
 func TestUploadFileSplitsLargeFiles(t *testing.T) {
-	fake := &fakeS3{}
+	fake := newFakeS3()
 	srv := httptest.NewServer(fake.handler())
 
 	defer srv.Close()
@@ -173,3 +208,54 @@ func TestUploadFileSplitsLargeFiles(t *testing.T) {
 	assert.Equal(t, 2, fake.parts)
 	assert.Equal(t, []string{"/b/suites/h/small.bin"}, fake.singlePuts)
 }
+
+// A suite directory is content-addressed by its hash, so re-uploading it on
+// every run re-sends bytes the bucket already holds — for a stateful suite
+// that is ~12 GB a run, most of it one pre-run bundle.
+func TestUploadSuiteDirSkipsUnchangedObjects(t *testing.T) {
+	fake := newFakeS3()
+	srv := httptest.NewServer(fake.handler())
+
+	defer srv.Close()
+
+	uploader, err := NewS3Uploader(logrus.New(), &config.S3UploadConfig{
+		Bucket:          "b",
+		EndpointURL:     srv.URL,
+		Region:          "us-east-1",
+		AccessKeyID:     "id",
+		SecretAccessKey: "secret",
+		ForcePathStyle:  true,
+		ParallelUploads: 4,
+		Prefix:          "results",
+	})
+	require.NoError(t, err)
+
+	dir := filepath.Join(t.TempDir(), "0d93b5bf3b970403")
+	require.NoError(t, os.MkdirAll(filepath.Join(dir, "benchmark", "t1"), 0o750))
+	require.NoError(t, os.MkdirAll(filepath.Join(dir, ".eest-meta"), 0o750))
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "benchmark", "t1", "test.request"), []byte("payload"), 0o600))
+	require.NoError(t, os.WriteFile(filepath.Join(dir, ".eest-meta", "fixtures.ini"), []byte("meta"), 0o600))
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "summary.json"), []byte(`{"hash":"x"}`), 0o600))
+
+	require.NoError(t, uploader.UploadSuiteDir(t.Context(), dir))
+	assert.Len(t, fake.singlePuts, 3, "first upload sends every file")
+
+	fake.singlePuts = nil
+
+	require.NoError(t, uploader.UploadSuiteDir(t.Context(), dir))
+
+	// summary.json is rewritten every run — labels change without changing the
+	// suite hash — so it alone is re-sent.
+	assert.Equal(t, []string{"/b/results/suites/0d93b5bf3b970403/summary.json"}, fake.singlePuts)
+
+	// A file whose size no longer matches is re-sent. Same size with different
+	// bytes is deliberately not detected: a suite key is content-addressed, so
+	// that cannot happen without the hash changing too.
+	fake.singlePuts = nil
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "benchmark", "t1", "test.request"), []byte("longer payload"), 0o600))
+	require.NoError(t, uploader.UploadSuiteDir(t.Context(), dir))
+	assert.ElementsMatch(t, []string{
+		"/b/results/suites/0d93b5bf3b970403/summary.json",
+		"/b/results/suites/0d93b5bf3b970403/benchmark/t1/test.request",
+	}, fake.singlePuts)
+}