feat(CLI): add --parallel flag to phrase pull for concurrent downloads

Download locale files using up to 4 concurrent requests (matching the
Phrase API concurrency limit) via errgroup.

Results are collected and printed in order after all downloads complete
for clean output. A shared mutex coordinates rate-limit pauses across
all workers.

Only supported in sync mode; --parallel with --async warns and ignores.

Author: tetienne

clients/cli/cmd/internal/pull.go

@@ -10,6 +10,7 @@ import (
 	"path/filepath"
 	"reflect"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/phrase/phrase-cli/cmd/internal/paths"
@@ -18,6 +19,7 @@ import (
 
 	"github.com/antihax/optional"
 	"github.com/phrase/phrase-go/v4"
+	"golang.org/x/sync/errgroup"
 )
 
 const (
@@ -26,13 +28,16 @@ const (
 	asyncRetryCount  = 360 // 30 minutes
 )
 
+const maxParallelDownloads = 4 // Phrase API allows max 4 concurrent requests
+
 var Config *phrase.Config
 
 type PullCommand struct {
 	phrase.Config
 	Branch             string
 	UseLocalBranchName bool
 	Async              bool
+	Parallel           bool
 }
 
 var Auth context.Context
@@ -82,7 +87,15 @@ func (cmd *PullCommand) Run(config *phrase.Config) error {
 	}
 
 	for _, target := range targets {
-		err := target.Pull(client, cmd.Async)
+		var err error
+		if cmd.Parallel && !cmd.Async {
+			err = target.PullParallel(client)
+		} else {
+			if cmd.Parallel && cmd.Async {
+				print.Warn("--parallel is not supported with --async, ignoring parallel")
+			}
+			err = target.Pull(client, cmd.Async)
+		}
 		if err != nil {
 			return err
 		}
@@ -146,25 +159,144 @@ func (target *Target) Pull(client *phrase.APIClient, async bool) error {
 	return nil
 }
 
-func (target *Target) DownloadAndWriteToFile(client *phrase.APIClient, localeFile *LocaleFile, async bool) error {
-	localVarOptionals := phrase.LocaleDownloadOpts{}
+type downloadResult struct {
+	message string
+	path    string
+	errMsg  string
+}
+
+func (target *Target) PullParallel(client *phrase.APIClient) error {
+	if err := target.CheckPreconditions(); err != nil {
+		return err
+	}
+
+	localeFiles, err := target.LocaleFiles()
+	if err != nil {
+		return err
+	}
+
+	// Ensure all destination files/dirs exist before parallel downloads
+	for _, lf := range localeFiles {
+		if err := createFile(lf.Path); err != nil {
+			return err
+		}
+	}
+
+	results := make([]downloadResult, len(localeFiles))
+	var rateMu sync.Mutex
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeoutInMinutes)
+	defer cancel()
+	g, ctx := errgroup.WithContext(ctx)
+	g.SetLimit(maxParallelDownloads)
+
+	for i, lf := range localeFiles {
+		g.Go(func() error {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			}
+
+			opts, err := target.buildDownloadOpts(lf)
+			if err != nil {
+				err = fmt.Errorf("%s for %s", err, lf.Path)
+				results[i] = downloadResult{errMsg: err.Error()}
+				return err
+			}
+
+			err = target.downloadWithRateLimitRetry(client, lf, opts, &rateMu)
+			if err != nil {
+				if openapiError, ok := err.(phrase.GenericOpenAPIError); ok {
+					print.Warn("API response: %s", openapiError.Body())
+				}
+				err = fmt.Errorf("%s for %s", err, lf.Path)
+				results[i] = downloadResult{errMsg: err.Error()}
+				return err
+			}
+
+			results[i] = downloadResult{
+				message: lf.Message(),
+				path:    lf.RelPath(),
+			}
+			return nil
+		})
+	}
+
+	waitErr := g.Wait()
+
+	// Print results in original order: successes and failures
+	var skipCount int
+	for _, r := range results {
+		if r.path != "" {
+			print.Success("Downloaded %s to %s", r.message, r.path)
+		} else if r.errMsg != "" {
+			print.Failure("Failed %s", r.errMsg)
+		} else {
+			skipCount++
+		}
+	}
+	if skipCount > 0 {
+		print.Warn("%d download(s) skipped due to earlier failure", skipCount)
+	}
+
+	return waitErr
+}
+
+// downloadWithRateLimitRetry downloads a locale file with rate-limit coordination.
+// The shared mutex pauses all workers when any worker hits the API rate limit.
+func (target *Target) downloadWithRateLimitRetry(client *phrase.APIClient, localeFile *LocaleFile, opts phrase.LocaleDownloadOpts, rateMu *sync.Mutex) error {
+	// Gate: block if another worker is waiting out a rate limit
+	rateMu.Lock()
+	rateMu.Unlock()
+
+	file, response, err := client.LocalesApi.LocaleDownload(Auth, target.ProjectID, localeFile.ID, &opts)
+	if err != nil {
+		if response != nil && response.Rate.Remaining == 0 {
+			// Hold the mutex while waiting for rate limit reset;
+			// this blocks other workers from making requests.
+			rateMu.Lock()
+			waitForRateLimit(response.Rate)
+			rateMu.Unlock()
+
+			file, _, err = client.LocalesApi.LocaleDownload(Auth, target.ProjectID, localeFile.ID, &opts)
+			if err != nil {
+				return err
+			}
+		} else {
+			return err
+		}
+	}
+	return copyToDestination(file, localeFile.Path)
+}
+
+// buildDownloadOpts prepares the LocaleDownloadOpts for a locale file download.
+func (target *Target) buildDownloadOpts(localeFile *LocaleFile) (phrase.LocaleDownloadOpts, error) {
+	opts := phrase.LocaleDownloadOpts{}
 
 	if target.Params != nil {
-		localVarOptionals = target.Params.LocaleDownloadOpts
+		opts = target.Params.LocaleDownloadOpts
 		translationKeyPrefix, err := placeholders.ResolveTranslationKeyPrefix(target.Params.TranslationKeyPrefix, localeFile.Path)
 		if err != nil {
-			return err
+			return opts, err
 		}
-		localVarOptionals.TranslationKeyPrefix = translationKeyPrefix
+		opts.TranslationKeyPrefix = translationKeyPrefix
 	}
 
-	if localVarOptionals.FileFormat.Value() == "" {
-		localVarOptionals.FileFormat = optional.NewString(localeFile.FileFormat)
+	if opts.FileFormat.Value() == "" {
+		opts.FileFormat = optional.NewString(localeFile.FileFormat)
 	}
 
 	if localeFile.Tag != "" {
-		localVarOptionals.Tags = optional.NewString(localeFile.Tag)
-		localVarOptionals.Tag = optional.EmptyString()
+		opts.Tags = optional.NewString(localeFile.Tag)
+		opts.Tag = optional.EmptyString()
+	}
+
+	return opts, nil
+}
+
+func (target *Target) DownloadAndWriteToFile(client *phrase.APIClient, localeFile *LocaleFile, async bool) error {
+	localVarOptionals, err := target.buildDownloadOpts(localeFile)
+	if err != nil {
+		return err
 	}
 
 	debugFprintln("Target file pattern:", target.File)
@@ -182,9 +314,8 @@ func (target *Target) DownloadAndWriteToFile(client *phrase.APIClient, localeFil
 
 	if async {
 		return target.downloadAsynchronously(client, localeFile, localVarOptionals)
-	} else {
-		return target.downloadSynchronously(client, localeFile, localVarOptionals)
 	}
+	return target.downloadSynchronously(client, localeFile, localVarOptionals)
 }
 
 func (target *Target) downloadAsynchronously(client *phrase.APIClient, localeFile *LocaleFile, downloadOpts phrase.LocaleDownloadOpts) error {

clients/cli/cmd/internal/pull_parallel_test.go

@@ -0,0 +1,113 @@
+package internal
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestRateLimitMutex_AllowsConcurrentWhenUnlocked(t *testing.T) {
+	var mu sync.Mutex
+	var concurrent int64
+	var maxConcurrent int64
+
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			// Simulate the gate pattern from downloadWithRateLimitRetry
+			mu.Lock()
+			mu.Unlock()
+
+			c := atomic.AddInt64(&concurrent, 1)
+			for {
+				old := atomic.LoadInt64(&maxConcurrent)
+				if c <= old || atomic.CompareAndSwapInt64(&maxConcurrent, old, c) {
+					break
+				}
+			}
+
+			time.Sleep(time.Millisecond)
+			atomic.AddInt64(&concurrent, -1)
+		}()
+	}
+	wg.Wait()
+
+	if maxConcurrent < 2 {
+		t.Errorf("expected concurrent execution when mutex is free, max concurrent was %d", maxConcurrent)
+	}
+}
+
+func TestRateLimitMutex_BlocksAllWorkersWhenHeld(t *testing.T) {
+	var mu sync.Mutex
+	ready := make(chan struct{}, 4)
+
+	// Simulate a rate-limited worker holding the mutex
+	mu.Lock()
+
+	var wg sync.WaitGroup
+	for i := 0; i < 4; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ready <- struct{}{}
+			mu.Lock()
+			mu.Unlock()
+		}()
+	}
+
+	// Wait for all goroutines to start
+	for i := 0; i < 4; i++ {
+		<-ready
+	}
+	// Give goroutines time to reach the mu.Lock() call
+	time.Sleep(10 * time.Millisecond)
+
+	// Release the lock (simulating rate limit wait done)
+	mu.Unlock()
+	wg.Wait()
+}
+
+func TestBuildDownloadOpts_DefaultFileFormat(t *testing.T) {
+	target := &Target{
+		File:      "locales/<locale_name>.json",
+		ProjectID: "proj1",
+	}
+	localeFile := &LocaleFile{
+		FileFormat: "json",
+		Tag:        "",
+	}
+
+	opts, err := target.buildDownloadOpts(localeFile)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if opts.FileFormat.Value() != "json" {
+		t.Errorf("expected file format 'json', got %q", opts.FileFormat.Value())
+	}
+}
+
+func TestBuildDownloadOpts_TagHandling(t *testing.T) {
+	target := &Target{
+		File:      "locales/<locale_name>/<tag>.json",
+		ProjectID: "proj1",
+	}
+	localeFile := &LocaleFile{
+		FileFormat: "json",
+		Tag:        "web",
+	}
+
+	opts, err := target.buildDownloadOpts(localeFile)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if opts.Tags.Value() != "web" {
+		t.Errorf("expected tags 'web', got %q", opts.Tags.Value())
+	}
+	if opts.Tag.Value() != "" {
+		t.Errorf("expected tag to be empty string, got %q", opts.Tag.Value())
+	}
+}

clients/cli/cmd/pull.go

@@ -21,6 +21,7 @@ func initPull() {
 				Branch:             params.GetString("branch"),
 				UseLocalBranchName: params.GetBool("use-local-branch-name"),
 				Async:              params.GetBool("async"),
+				Parallel:           params.GetBool("parallel"),
 			}
 			err := cmdPull.Run(Config)
 			if err != nil {
@@ -33,5 +34,6 @@ func initPull() {
 	AddFlag(pullCmd, "string", "branch", "b", "branch", false)
 	AddFlag(pullCmd, "bool", "use-local-branch-name", "", "use local branch name", false)
 	AddFlag(pullCmd, "bool", "async", "a", "use asynchronous locale downloads (recommended for large number of keys)", false)
+	AddFlag(pullCmd, "bool", "parallel", "p", "download locale files in parallel (max 4 concurrent requests)", false)
 	params.BindPFlags(pullCmd.Flags())
 }

clients/cli/go.mod

@@ -37,5 +37,6 @@ require (
 	github.com/pelletier/go-toml v1.2.0 // indirect
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/stretchr/testify v1.9.0 // indirect
+	golang.org/x/sync v0.12.0
 	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
 )

clients/cli/go.sum

@@ -215,10 +215,8 @@ github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn
 github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
 github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
-github.com/phrase/phrase-go/v4 v4.18.1 h1:y1sv4z8ufEQB+kJA8ymSiH8nRAvH8gGoVSB5/7jvYEQ=
-github.com/phrase/phrase-go/v4 v4.18.1/go.mod h1:4XplKvrbHS2LDaXfFp9xrVDtO5xk2WHFm0htutwwd8c=
-github.com/phrase/phrase-go/v4 v4.19.0 h1:tNliCxO/0SMu2viLE9idzADBUoY9C6CqrDmp3ntgpQI=
-github.com/phrase/phrase-go/v4 v4.19.0/go.mod h1:4XplKvrbHS2LDaXfFp9xrVDtO5xk2WHFm0htutwwd8c=
+github.com/phrase/phrase-go/v4 v4.20.0 h1:UTgos4elHzv83XbC6hK2ulDVEvvi1215r8NzE8jC3bc=
+github.com/phrase/phrase-go/v4 v4.20.0/go.mod h1:4XplKvrbHS2LDaXfFp9xrVDtO5xk2WHFm0htutwwd8c=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -370,6 +368,8 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw=
+golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=