fix(private-org-sync): process repos in parallel with configurable workers

Extract per-repo sync logic into a syncRepo function and process repos
concurrently using a worker pool. The --parallelism flag (default 4)
controls the number of concurrent workers. Each worker gets its own copy
of the gitSyncer struct, avoiding races on the mutable logger field.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: petr-muller

cmd/private-org-sync/main.go

@@ -9,6 +9,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/sirupsen/logrus"
@@ -65,6 +66,7 @@ type options struct {
 	confirm              bool
 	failOnNonexistentDst bool
 	debug                bool
+	parallelism          int
 }
 
 const defaultPrefix = "https://github.com"
@@ -144,6 +146,7 @@ func gatherOptions() options {
 	fs.BoolVar(&o.failOnNonexistentDst, "fail-on-missing-destination", false, "Set true to make the tool to consider missing sync destination as an error")
 
 	fs.BoolVar(&o.debug, "debug", false, "Set true to enable debug logging level")
+	fs.IntVar(&o.parallelism, "parallelism", 4, "Number of repos to sync in parallel")
 
 	o.Options.Bind(fs)
 	o.WhitelistOptions.Bind(fs)
@@ -362,6 +365,8 @@ type location struct {
 	org, repo, branch string
 }
 
+type repoKey struct{ org, repo string }
+
 func (l location) String() string {
 	return fmt.Sprintf("%s/%s@%s", l.org, l.repo, l.branch)
 }
@@ -675,7 +680,6 @@ func main() {
 	}
 
 	// Group locations by (org, repo) so we can initialize each repo once
-	type repoKey struct{ org, repo string }
 	grouped := make(map[repoKey][]location)
 	for source := range locations {
 		key := repoKey{org: source.org, repo: source.repo}
@@ -688,108 +692,141 @@ func main() {
 		flattenedOrgs.Insert(o.org)
 	}
 
+	type repoWork struct {
+		key      repoKey
+		branches []location
+	}
+	work := make(chan repoWork)
+	var errsMu sync.Mutex
+	var wg sync.WaitGroup
+
+	for i := 0; i < o.parallelism; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for item := range work {
+				repoErrs := syncRepo(syncer, o.targetOrg, flattenedOrgs, item.key, item.branches)
+				if len(repoErrs) > 0 {
+					errsMu.Lock()
+					errs = append(errs, repoErrs...)
+					errsMu.Unlock()
+				}
+			}
+		}()
+	}
+
 	for key, branches := range grouped {
-		repoLogger := logrus.WithFields(logrus.Fields{"org": key.org, "repo": key.repo})
-		syncer.logger = repoLogger
+		work <- repoWork{key: key, branches: branches}
+	}
+	close(work)
+	wg.Wait()
 
-		dstRepo := key.repo
-		if !flattenedOrgs.Has(key.org) {
-			dstRepo = fmt.Sprintf("%s-%s", key.org, key.repo)
-		}
+	if len(errs) > 0 {
+		logrus.WithError(utilerrors.NewAggregate(errs)).Fatal("There were failures")
+	}
+}
 
-		gitDir, err := syncer.makeGitDir(key.org, key.repo)
-		if err != nil {
-			for _, source := range branches {
-				errs = append(errs, fmt.Errorf("%s: %w", source.String(), err))
-			}
-			continue
-		}
+func syncRepo(syncer gitSyncer, targetOrg string, flattenedOrgs sets.Set[string], key repoKey, branches []location) []error {
+	repoLogger := logrus.WithFields(logrus.Fields{"org": key.org, "repo": key.repo})
+	syncer.logger = repoLogger
 
-		if err := syncer.initRepo(gitDir, key.org, key.repo); err != nil {
-			for _, source := range branches {
-				errs = append(errs, fmt.Errorf("%s: %w", source.String(), err))
-			}
-			continue
-		}
+	var errs []error
 
-		// ls-remote source and destination in parallel
-		destUrlRaw := fmt.Sprintf("%s/%s/%s", syncer.prefix, o.targetOrg, dstRepo)
-		destUrl, err := url.Parse(destUrlRaw)
-		if err != nil {
-			repoLogger.WithField("remote-url", destUrlRaw).WithError(err).Error("Failed to construct URL for the destination remote")
-			for _, source := range branches {
-				errs = append(errs, fmt.Errorf("%s: failed to construct URL for the destination remote", source.String()))
-			}
-			continue
-		}
-		if syncer.token != "" {
-			destUrl.User = url.User(syncer.token)
+	dstRepo := key.repo
+	if !flattenedOrgs.Has(key.org) {
+		dstRepo = fmt.Sprintf("%s-%s", key.org, key.repo)
+	}
+
+	gitDir, err := syncer.makeGitDir(key.org, key.repo)
+	if err != nil {
+		for _, source := range branches {
+			errs = append(errs, fmt.Errorf("%s: %w", source.String(), err))
 		}
+		return errs
+	}
 
-		srcRemote := fmt.Sprintf("%s-%s", key.org, key.repo)
+	if err := syncer.initRepo(gitDir, key.org, key.repo); err != nil {
+		for _, source := range branches {
+			errs = append(errs, fmt.Errorf("%s: %w", source.String(), err))
+		}
+		return errs
+	}
 
-		type lsRemoteResult struct {
-			heads RemoteBranchHeads
-			err   error
+	// ls-remote source and destination in parallel
+	destUrlRaw := fmt.Sprintf("%s/%s/%s", syncer.prefix, targetOrg, dstRepo)
+	destUrl, err := url.Parse(destUrlRaw)
+	if err != nil {
+		repoLogger.WithField("remote-url", destUrlRaw).WithError(err).Error("Failed to construct URL for the destination remote")
+		for _, source := range branches {
+			errs = append(errs, fmt.Errorf("%s: failed to construct URL for the destination remote", source.String()))
 		}
-		dstResult := make(chan lsRemoteResult, 1)
-		srcResult := make(chan lsRemoteResult, 1)
-		go func() {
-			heads, err := getRemoteBranchHeads(repoLogger, syncer.git, gitDir, destUrl.String())
-			dstResult <- lsRemoteResult{heads, err}
-		}()
-		go func() {
-			heads, err := getRemoteBranchHeads(repoLogger, withRetryOnNonzero(syncer.git, 5), gitDir, srcRemote)
-			srcResult <- lsRemoteResult{heads, err}
-		}()
+		return errs
+	}
+	if syncer.token != "" {
+		destUrl.User = url.User(syncer.token)
+	}
 
-		dst := <-dstResult
-		src := <-srcResult
+	srcRemote := fmt.Sprintf("%s-%s", key.org, key.repo)
 
-		if dst.err != nil {
-			message := "destination repository does not exist or we cannot access it"
-			if syncer.failOnNonexistentDst {
-				repoLogger.Errorf("%s", message)
-				for _, source := range branches {
-					errs = append(errs, fmt.Errorf("%s: %s", source.String(), message))
-				}
-			} else {
-				repoLogger.Warn(message)
-			}
-			continue
-		}
+	type lsRemoteResult struct {
+		heads RemoteBranchHeads
+		err   error
+	}
+	dstResult := make(chan lsRemoteResult, 1)
+	srcResult := make(chan lsRemoteResult, 1)
+	go func() {
+		heads, err := getRemoteBranchHeads(repoLogger, syncer.git, gitDir, destUrl.String())
+		dstResult <- lsRemoteResult{heads, err}
+	}()
+	go func() {
+		heads, err := getRemoteBranchHeads(repoLogger, withRetryOnNonzero(syncer.git, 5), gitDir, srcRemote)
+		srcResult <- lsRemoteResult{heads, err}
+	}()
 
-		if src.err != nil {
-			repoLogger.WithError(src.err).Error("Failed to determine branch HEADs in source")
+	dst := <-dstResult
+	src := <-srcResult
+
+	if dst.err != nil {
+		message := "destination repository does not exist or we cannot access it"
+		if syncer.failOnNonexistentDst {
+			repoLogger.Errorf("%s", message)
 			for _, source := range branches {
-				errs = append(errs, fmt.Errorf("%s: failed to determine branch HEADs in source", source.String()))
+				errs = append(errs, fmt.Errorf("%s: %s", source.String(), message))
 			}
-			continue
+		} else {
+			repoLogger.Warn(message)
 		}
+		return errs
+	}
 
-		dstHeads := dst.heads
-		srcHeads := src.heads
-
+	if src.err != nil {
+		repoLogger.WithError(src.err).Error("Failed to determine branch HEADs in source")
 		for _, source := range branches {
-			syncer.logger = config.LoggerForInfo(config.Info{
-				Metadata: api.Metadata{
-					Org:    source.org,
-					Repo:   source.repo,
-					Branch: source.branch,
-				},
-			})
-
-			destination := location{org: o.targetOrg, repo: dstRepo, branch: source.branch}
-
-			if err := syncer.mirror(gitDir, source, destination, srcHeads, dstHeads, destUrl); err != nil {
-				errs = append(errs, fmt.Errorf("%s->%s: %w", source.String(), destination.String(), err))
-			}
+			errs = append(errs, fmt.Errorf("%s: failed to determine branch HEADs in source", source.String()))
 		}
+		return errs
 	}
 
-	if len(errs) > 0 {
-		logrus.WithError(utilerrors.NewAggregate(errs)).Fatal("There were failures")
+	dstHeads := dst.heads
+	srcHeads := src.heads
+
+	for _, source := range branches {
+		syncer.logger = config.LoggerForInfo(config.Info{
+			Metadata: api.Metadata{
+				Org:    source.org,
+				Repo:   source.repo,
+				Branch: source.branch,
+			},
+		})
+
+		destination := location{org: targetOrg, repo: dstRepo, branch: source.branch}
+
+		if err := syncer.mirror(gitDir, source, destination, srcHeads, dstHeads, destUrl); err != nil {
+			errs = append(errs, fmt.Errorf("%s->%s: %w", source.String(), destination.String(), err))
+		}
 	}
+
+	return errs
 }
 
 func getWhitelistedLocations(whitelist map[string][]string, git gitFunc, prefix, token string) (map[location]struct{}, []error) {