sync: replace hardcoded concurrency limit with --concurrency flag

cmd/bundle/sync.go

@@ -17,11 +17,12 @@ import (
 )
 
 type syncFlags struct {
-	interval time.Duration
-	full     bool
-	watch    bool
-	output   flags.Output
-	dryRun   bool
+	interval    time.Duration
+	full        bool
+	watch       bool
+	output      flags.Output
+	dryRun      bool
+	concurrency int
 }
 
 func (f *syncFlags) syncOptionsFromBundle(cmd *cobra.Command, b *bundle.Bundle) (*sync.SyncOptions, error) {
@@ -48,6 +49,7 @@ func (f *syncFlags) syncOptionsFromBundle(cmd *cobra.Command, b *bundle.Bundle)
 	opts.Full = f.full
 	opts.PollInterval = f.interval
 	opts.DryRun = f.dryRun
+	opts.Concurrency = f.concurrency
 	return opts, nil
 }
 
@@ -74,8 +76,13 @@ Use 'databricks bundle deploy' for full resource deployment.`,
 	cmd.Flags().BoolVar(&f.watch, "watch", false, "watch local file system for changes")
 	cmd.Flags().Var(&f.output, "output", "type of the output format")
 	cmd.Flags().BoolVar(&f.dryRun, "dry-run", false, "simulate sync execution without making actual changes")
+	cmd.Flags().IntVar(&f.concurrency, "concurrency", sync.MaxRequestsInFlight, "maximum number of concurrent in-flight requests during sync")
 
 	cmd.RunE = func(cmd *cobra.Command, args []string) error {
+		if f.concurrency < 1 {
+			return fmt.Errorf("--concurrency must be a positive integer, got %d", f.concurrency)
+		}
+
 		b, err := utils.ProcessBundle(cmd, utils.ProcessOptions{})
 		if err != nil {
 			return err

cmd/sync/sync.go

@@ -35,6 +35,7 @@ type syncFlags struct {
 	dryRun      bool
 	excludeFrom string
 	includeFrom string
+	concurrency int
 }
 
 func readPatternsFile(filePath string) ([]string, error) {
@@ -89,6 +90,7 @@ func (f *syncFlags) syncOptionsFromBundle(cmd *cobra.Command, args []string, b *
 	opts.Include = append(opts.Include, f.include...)
 	opts.Include = append(opts.Include, includePatterns...)
 	opts.DryRun = f.dryRun
+	opts.Concurrency = f.concurrency
 	return opts, nil
 }
 
@@ -163,6 +165,7 @@ func (f *syncFlags) syncOptionsFromArgs(cmd *cobra.Command, args []string) (*syn
 
 		OutputHandler: outputHandler,
 		DryRun:        f.dryRun,
+		Concurrency:   f.concurrency,
 	}
 	return &opts, nil
 }
@@ -187,6 +190,7 @@ func New() *cobra.Command {
 	cmd.Flags().StringVar(&f.excludeFrom, "exclude-from", "", "file containing patterns to exclude from sync (one pattern per line)")
 	cmd.Flags().StringVar(&f.includeFrom, "include-from", "", "file containing patterns to include to sync (one pattern per line)")
 	cmd.Flags().BoolVar(&f.dryRun, "dry-run", false, "simulate sync execution without making actual changes")
+	cmd.Flags().IntVar(&f.concurrency, "concurrency", sync.MaxRequestsInFlight, "maximum number of concurrent in-flight requests during sync")
 
 	// Wrapper for [root.MustWorkspaceClient] that disables loading authentication configuration from a bundle.
 	mustWorkspaceClient := func(cmd *cobra.Command, args []string) error {
@@ -196,6 +200,10 @@ func New() *cobra.Command {
 
 	cmd.PreRunE = mustWorkspaceClient
 	cmd.RunE = func(cmd *cobra.Command, args []string) error {
+		if f.concurrency < 1 {
+			return fmt.Errorf("--concurrency must be a positive integer, got %d", f.concurrency)
+		}
+
 		var opts *sync.SyncOptions
 		var err error
 

libs/sync/sync.go

@@ -43,6 +43,12 @@ type SyncOptions struct {
 	OutputHandler OutputHandler
 
 	DryRun bool
+
+	// Concurrency is the maximum number of concurrent in-flight requests during
+	// sync (file uploads, deletes, mkdir, rmdir). When zero, MaxRequestsInFlight
+	// is used. Callers (e.g. the --concurrency flag) can set this to tune
+	// throughput against workspace API rate limits.
+	Concurrency int
 }
 
 type Sync struct {
@@ -96,6 +102,12 @@ func New(ctx context.Context, opts SyncOptions) (*Sync, error) {
 		return nil, errors.New("failed to resolve host for snapshot")
 	}
 
+	// Normalize concurrency so internal callers don't have to repeat the default.
+	// Negative values indicate a programming error and are rejected at the flag layer.
+	if opts.Concurrency <= 0 {
+		opts.Concurrency = MaxRequestsInFlight
+	}
+
 	// For full sync, we start with an empty snapshot.
 	// For incremental sync, we try to load an existing snapshot to start from.
 	var snapshot *Snapshot
@@ -119,7 +131,7 @@ func New(ctx context.Context, opts SyncOptions) (*Sync, error) {
 	var notifier EventNotifier
 	outputWaitGroup := &stdsync.WaitGroup{}
 	if opts.OutputHandler != nil {
-		ch := make(chan Event, MaxRequestsInFlight)
+		ch := make(chan Event, opts.Concurrency)
 		notifier = &ChannelNotifier{ch}
 		outputWaitGroup.Go(func() {
 			opts.OutputHandler(ctx, ch)

libs/sync/watchdog.go

@@ -10,7 +10,8 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-// Maximum number of concurrent requests during sync.
+// Default maximum number of concurrent requests during sync.
+// Callers can override this via SyncOptions.Concurrency (e.g. the --concurrency flag).
 const MaxRequestsInFlight = 20
 
 // Delete the specified path.
@@ -96,9 +97,9 @@ func groupRunSingle(ctx context.Context, group *errgroup.Group, fn func(context.
 	})
 }
 
-func groupRunParallel(ctx context.Context, paths []string, fn func(context.Context, string) error) error {
+func groupRunParallel(ctx context.Context, paths []string, limit int, fn func(context.Context, string) error) error {
 	group, ctx := errgroup.WithContext(ctx)
-	group.SetLimit(MaxRequestsInFlight)
+	group.SetLimit(limit)
 
 	for _, path := range paths {
 		groupRunSingle(ctx, group, fn, path)
@@ -110,31 +111,32 @@ func groupRunParallel(ctx context.Context, paths []string, fn func(context.Conte
 
 func (s *Sync) applyDiff(ctx context.Context, d diff) error {
 	var err error
+	limit := s.Concurrency
 
 	// Delete files in parallel.
-	err = groupRunParallel(ctx, d.delete, s.applyDelete)
+	err = groupRunParallel(ctx, d.delete, limit, s.applyDelete)
 	if err != nil {
 		return err
 	}
 
 	// Delete directories ordered by depth from leaf to root.
 	for _, group := range d.groupedRmdir() {
-		err = groupRunParallel(ctx, group, s.applyRmdir)
+		err = groupRunParallel(ctx, group, limit, s.applyRmdir)
 		if err != nil {
 			return err
 		}
 	}
 
 	// Create directories (leafs only because intermediates are created automatically).
 	for _, group := range d.groupedMkdir() {
-		err = groupRunParallel(ctx, group, s.applyMkdir)
+		err = groupRunParallel(ctx, group, limit, s.applyMkdir)
 		if err != nil {
 			return err
 		}
 	}
 
 	// Put files in parallel.
-	err = groupRunParallel(ctx, d.put, s.applyPut)
+	err = groupRunParallel(ctx, d.put, limit, s.applyPut)
 
 	return err
 }

libs/sync/watchdog_test.go

@@ -0,0 +1,44 @@
+package sync
+
+import (
+	"context"
+	"strconv"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestGroupRunParallelHonorsLimit asserts the parallelism limit passed into
+// groupRunParallel actually caps in-flight work. Without this test, swapping
+// the hardcoded limit out for SyncOptions.Concurrency could silently lose
+// the cap.
+func TestGroupRunParallelHonorsLimit(t *testing.T) {
+	const limit = 3
+
+	paths := make([]string, 20)
+	for i := range paths {
+		paths[i] = strconv.Itoa(i)
+	}
+
+	var inflight, peak atomic.Int32
+	fn := func(ctx context.Context, path string) error {
+		cur := inflight.Add(1)
+		// Track the peak observed concurrency.
+		for {
+			p := peak.Load()
+			if cur <= p || peak.CompareAndSwap(p, cur) {
+				break
+			}
+		}
+		// Hold long enough that other goroutines pile up if the limit is wrong.
+		time.Sleep(20 * time.Millisecond)
+		inflight.Add(-1)
+		return nil
+	}
+
+	err := groupRunParallel(t.Context(), paths, limit, fn)
+	require.NoError(t, err)
+	require.LessOrEqual(t, int(peak.Load()), limit, "observed concurrency exceeded limit")
+}