sync: rename flag to --concurrency and extend to bundle sync

Match the name of `fs cp --concurrency` (#4132) for consistency, and
expose the same flag on `databricks bundle sync` so both sync entry
points are tunable.

Co-authored-by: Isaac

Author: pietern

acceptance/bundle/help/bundle-sync/output.txt

@@ -15,6 +15,7 @@ Usage:
   databricks bundle sync [flags]
 
 Flags:
+      --concurrency int     number of parallel requests to the workspace (default 20)
       --dry-run             simulate sync execution without making actual changes
       --full                perform full synchronization (default is incremental)
   -h, --help                help for sync

acceptance/cmd/sync-without-args/output.txt

@@ -6,17 +6,17 @@ Usage:
   databricks sync [flags] SRC DST
 
 Flags:
-      --dry-run                       simulate sync execution without making actual changes
-      --exclude strings               patterns to exclude from sync (can be specified multiple times)
-      --exclude-from string           file containing patterns to exclude from sync (one pattern per line)
-      --full                          perform full synchronization (default is incremental)
-  -h, --help                          help for sync
-      --include strings               patterns to include in sync (can be specified multiple times)
-      --include-from string           file containing patterns to include to sync (one pattern per line)
-      --interval duration             file system polling interval (for --watch) (default 1s)
-      --max-concurrent-requests int   maximum number of concurrent requests to the workspace (default 20)
-      --output type                   type of output format (default text)
-      --watch                         watch local file system for changes
+      --concurrency int       number of parallel requests to the workspace (default 20)
+      --dry-run               simulate sync execution without making actual changes
+      --exclude strings       patterns to exclude from sync (can be specified multiple times)
+      --exclude-from string   file containing patterns to exclude from sync (one pattern per line)
+      --full                  perform full synchronization (default is incremental)
+  -h, --help                  help for sync
+      --include strings       patterns to include in sync (can be specified multiple times)
+      --include-from string   file containing patterns to include to sync (one pattern per line)
+      --interval duration     file system polling interval (for --watch) (default 1s)
+      --output type           type of output format (default text)
+      --watch                 watch local file system for changes
 
 Global Flags:
       --debug            enable debug logging

cmd/bundle/sync.go

@@ -2,6 +2,7 @@ package bundle
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"time"
@@ -16,12 +17,22 @@ import (
 	"github.com/spf13/cobra"
 )
 
+var errInvalidConcurrency = errors.New("--concurrency must be at least 1")
+
 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) validate() error {
+	if f.concurrency < 1 {
+		return errInvalidConcurrency
+	}
+	return nil
 }
 
 func (f *syncFlags) syncOptionsFromBundle(cmd *cobra.Command, b *bundle.Bundle) (*sync.SyncOptions, error) {
@@ -48,6 +59,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 +86,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, "number of parallel requests to the workspace")
 
 	cmd.RunE = func(cmd *cobra.Command, args []string) error {
+		if err := f.validate(); err != nil {
+			return err
+		}
+
 		b, err := utils.ProcessBundle(cmd, utils.ProcessOptions{})
 		if err != nil {
 			return err

cmd/sync/sync.go

@@ -24,23 +24,25 @@ import (
 	"github.com/spf13/cobra"
 )
 
+var errInvalidConcurrency = errors.New("--concurrency must be at least 1")
+
 type syncFlags struct {
 	// project files polling interval
-	interval              time.Duration
-	full                  bool
-	watch                 bool
-	output                flags.Output
-	exclude               []string
-	include               []string
-	dryRun                bool
-	excludeFrom           string
-	includeFrom           string
-	maxConcurrentRequests int
+	interval    time.Duration
+	full        bool
+	watch       bool
+	output      flags.Output
+	exclude     []string
+	include     []string
+	dryRun      bool
+	excludeFrom string
+	includeFrom string
+	concurrency int
 }
 
 func (f *syncFlags) validate() error {
-	if f.maxConcurrentRequests < 1 {
-		return fmt.Errorf("--max-concurrent-requests must be >= 1, got %d", f.maxConcurrentRequests)
+	if f.concurrency < 1 {
+		return errInvalidConcurrency
 	}
 	return nil
 }
@@ -97,7 +99,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.MaxConcurrentRequests = f.maxConcurrentRequests
+	opts.Concurrency = f.concurrency
 	return opts, nil
 }
 
@@ -170,9 +172,9 @@ func (f *syncFlags) syncOptionsFromArgs(cmd *cobra.Command, args []string) (*syn
 		SnapshotBasePath: filepath.Join(args[0], ".databricks"),
 		WorkspaceClient:  client,
 
-		OutputHandler:         outputHandler,
-		DryRun:                f.dryRun,
-		MaxConcurrentRequests: f.maxConcurrentRequests,
+		OutputHandler: outputHandler,
+		DryRun:        f.dryRun,
+		Concurrency:   f.concurrency,
 	}
 	return &opts, nil
 }
@@ -197,7 +199,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.maxConcurrentRequests, "max-concurrent-requests", sync.MaxRequestsInFlight, "maximum number of concurrent requests to the workspace")
+	cmd.Flags().IntVar(&f.concurrency, "concurrency", sync.MaxRequestsInFlight, "number of parallel requests to the workspace")
 
 	// Wrapper for [root.MustWorkspaceClient] that disables loading authentication configuration from a bundle.
 	mustWorkspaceClient := func(cmd *cobra.Command, args []string) error {

cmd/sync/sync_test.go

@@ -32,13 +32,13 @@ func TestSyncOptionsFromBundle(t *testing.T) {
 		},
 	}
 
-	f := syncFlags{maxConcurrentRequests: 5}
+	f := syncFlags{concurrency: 5}
 	opts, err := f.syncOptionsFromBundle(New(), []string{}, b)
 	require.NoError(t, err)
 	assert.Equal(t, tempDir, opts.LocalRoot.Native())
 	assert.Equal(t, "/Users/jane@doe.com/path", opts.RemotePath)
 	assert.Equal(t, filepath.Join(tempDir, ".databricks", "bundle", "default"), opts.SnapshotBasePath)
-	assert.Equal(t, 5, opts.MaxConcurrentRequests)
+	assert.Equal(t, 5, opts.Concurrency)
 	assert.NotNil(t, opts.WorkspaceClient)
 }
 
@@ -57,25 +57,25 @@ func TestSyncOptionsFromArgs(t *testing.T) {
 	local := t.TempDir()
 	remote := "/remote"
 
-	f := syncFlags{maxConcurrentRequests: 7}
+	f := syncFlags{concurrency: 7}
 	cmd := New()
 	cmd.SetContext(cmdctx.SetWorkspaceClient(t.Context(), nil))
 	opts, err := f.syncOptionsFromArgs(cmd, []string{local, remote})
 	require.NoError(t, err)
 	assert.Equal(t, local, opts.LocalRoot.Native())
 	assert.Equal(t, remote, opts.RemotePath)
-	assert.Equal(t, 7, opts.MaxConcurrentRequests)
+	assert.Equal(t, 7, opts.Concurrency)
 }
 
 func TestSyncFlagsValidate(t *testing.T) {
-	f := syncFlags{maxConcurrentRequests: 1}
+	f := syncFlags{concurrency: 1}
 	require.NoError(t, f.validate())
 
-	f.maxConcurrentRequests = 0
-	require.ErrorContains(t, f.validate(), "--max-concurrent-requests must be >= 1")
+	f.concurrency = 0
+	require.ErrorIs(t, f.validate(), errInvalidConcurrency)
 
-	f.maxConcurrentRequests = -3
-	require.ErrorContains(t, f.validate(), "--max-concurrent-requests must be >= 1")
+	f.concurrency = -3
+	require.ErrorIs(t, f.validate(), errInvalidConcurrency)
 }
 
 func TestExcludeFromFlag(t *testing.T) {

libs/sync/sync.go

@@ -44,7 +44,7 @@ type SyncOptions struct {
 
 	DryRun bool
 
-	MaxConcurrentRequests int
+	Concurrency int
 }
 
 type Sync struct {
@@ -98,8 +98,8 @@ func New(ctx context.Context, opts SyncOptions) (*Sync, error) {
 		return nil, errors.New("failed to resolve host for snapshot")
 	}
 
-	if opts.MaxConcurrentRequests <= 0 {
-		opts.MaxConcurrentRequests = MaxRequestsInFlight
+	if opts.Concurrency <= 0 {
+		opts.Concurrency = MaxRequestsInFlight
 	}
 
 	// For full sync, we start with an empty snapshot.

libs/sync/watchdog.go

@@ -110,7 +110,7 @@ func groupRunParallel(ctx context.Context, paths []string, limit int, fn func(co
 
 func (s *Sync) applyDiff(ctx context.Context, d diff) error {
 	var err error
-	limit := s.MaxConcurrentRequests
+	limit := s.Concurrency
 
 	// Delete files in parallel.
 	err = groupRunParallel(ctx, d.delete, limit, s.applyDelete)