perf(sourcemaps): upload source maps concurrently

cmd/sourcemaps/upload.go

@@ -2,6 +2,7 @@ package sourcemaps
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -15,6 +16,7 @@ import (
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+	"golang.org/x/sync/errgroup"
 
 	cmdAnalytics "github.com/launchdarkly/ldcli/cmd/analytics"
 	"github.com/launchdarkly/ldcli/cmd/cliflags"
@@ -34,6 +36,8 @@ const (
 	defaultPath       = "."
 	defaultBackendUrl = "https://pri.observability.app.launchdarkly.com"
 
+	uploadConcurrency = 8
+
 	getSourceMapUrlsQuery = `
 	  query GetSourceMapUploadUrls($api_key: String!, $project_id: String!, $paths: [String!]!) {
 	    get_source_map_upload_urls_ld(
@@ -157,10 +161,8 @@ func runE(client resources.Client) func(cmd *cobra.Command, args []string) error
 			return fmt.Errorf("failed to get upload URLs: %w", err)
 		}
 
-		for i, file := range files {
-			if err := uploadFile(file.Path, uploadUrls[i], file.Name); err != nil {
-				return fmt.Errorf("failed to upload file %s: %w", file.Path, err)
-			}
+		if err := uploadFiles(files, uploadUrls); err != nil {
+			return err
 		}
 
 		fmt.Println("Successfully uploaded all sourcemaps")
@@ -309,6 +311,29 @@ func getSourceMapUploadUrls(apiKey, projectID string, paths []string, backendUrl
 	return urlsResp.Data.GetSourceMapUploadUrls, nil
 }
 
+// uploadFiles uploads each file to its presigned URL, running up to
+// uploadConcurrency uploads at once. After the first failure it starts no new
+// uploads and returns that failure.
+func uploadFiles(files []SourceMapFile, uploadUrls []string) error {
+	g, ctx := errgroup.WithContext(context.Background())
+	g.SetLimit(uploadConcurrency)
+
+	for i, file := range files {
+		g.Go(func() error {
+			if ctx.Err() != nil {
+				return nil
+			}
+
+			if err := uploadFile(file.Path, uploadUrls[i], file.Name); err != nil {
+				return fmt.Errorf("failed to upload file %s: %w", file.Path, err)
+			}
+			return nil
+		})
+	}
+
+	return g.Wait()
+}
+
 func uploadFile(filePath, uploadUrl, name string) error {
 	fileContent, err := os.ReadFile(filePath)
 	if err != nil {

cmd/sourcemaps/upload_test.go

@@ -9,7 +9,9 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -128,6 +130,68 @@ func TestUploadFile(t *testing.T) {
 	assert.Contains(t, err.Error(), "upload failed with status code: 500")
 }
 
+func writeUploadTestFiles(t *testing.T, n int) []SourceMapFile {
+	tempDir := t.TempDir()
+	files := make([]SourceMapFile, 0, n)
+	for i := 0; i < n; i++ {
+		name := fmt.Sprintf("chunk%d.js.map", i)
+		path := filepath.Join(tempDir, name)
+		assert.NoError(t, os.WriteFile(path, []byte("{}"), 0644))
+		files = append(files, SourceMapFile{Path: path, Name: name})
+	}
+	return files
+}
+
+func TestUploadFilesRunsConcurrently(t *testing.T) {
+	files := writeUploadTestFiles(t, uploadConcurrency*4)
+
+	var inFlight, maxInFlight, uploaded atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		n := inFlight.Add(1)
+		defer inFlight.Add(-1)
+		for {
+			m := maxInFlight.Load()
+			if n <= m || maxInFlight.CompareAndSwap(m, n) {
+				break
+			}
+		}
+		time.Sleep(20 * time.Millisecond)
+		uploaded.Add(1)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+
+	urls := make([]string, len(files))
+	for i := range urls {
+		urls[i] = fmt.Sprintf("%s/%d", server.URL, i)
+	}
+
+	assert.NoError(t, uploadFiles(files, urls))
+	assert.Equal(t, int32(len(files)), uploaded.Load())
+	assert.Greater(t, maxInFlight.Load(), int32(1))
+	assert.LessOrEqual(t, maxInFlight.Load(), int32(uploadConcurrency))
+}
+
+func TestUploadFilesStopsAfterFirstError(t *testing.T) {
+	files := writeUploadTestFiles(t, uploadConcurrency*10)
+
+	var requests atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests.Add(1)
+		w.WriteHeader(http.StatusForbidden)
+	}))
+	defer server.Close()
+
+	urls := make([]string, len(files))
+	for i := range urls {
+		urls[i] = server.URL
+	}
+
+	err := uploadFiles(files, urls)
+	assert.ErrorContains(t, err, "upload failed with status code: 403")
+	assert.Less(t, requests.Load(), int32(len(files)))
+}
+
 func TestNewUploadCmd(t *testing.T) {
 	client := resources.NewClient("")
 	cmd := NewUploadCmd(client, func(accessToken, baseURI string, analyticsOptOut bool) analytics.Tracker {

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/atotto/clipboard v0.1.4
 	github.com/aws/aws-sdk-go-v2 v1.47.3
 	github.com/aws/aws-sdk-go-v2/config v1.33.5
+	github.com/aws/aws-sdk-go-v2/credentials v1.20.5
 	github.com/aws/aws-sdk-go-v2/service/devopsagent v1.17.0
 	github.com/aws/aws-sdk-go-v2/service/iam v1.64.0
 	github.com/aws/aws-sdk-go-v2/service/sts v1.51.4
@@ -48,6 +49,7 @@ require (
 	github.com/stretchr/testify v1.11.1
 	go.uber.org/mock v0.6.0
 	golang.org/x/exp v0.0.0-20240808152545-0cdaa3abc0fa
+	golang.org/x/sync v0.19.0
 	golang.org/x/term v0.40.0
 	gopkg.in/yaml.v3 v3.0.1
 )
@@ -57,7 +59,6 @@ require (
 	github.com/alecthomas/chroma/v2 v2.14.0 // indirect
 	github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.20 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.20.5 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.0 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.6 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.6 // indirect
@@ -127,7 +128,6 @@ require (
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/mod v0.33.0 // indirect
 	golang.org/x/net v0.50.0 // indirect
-	golang.org/x/sync v0.19.0 // indirect
 	golang.org/x/sys v0.45.0 // indirect
 	golang.org/x/text v0.34.0 // indirect
 	golang.org/x/tools v0.42.0 // indirect