feat(cli): parallel --rain fetch with bounded worker pool (PR D)

README.md

@@ -429,7 +429,7 @@ rain_panel_size = "comfortable"
 
 ## Interactive TUI
 
-`git-rain --rain` opens an interactive picker. Repositories stream in as the filesystem scan finds them — no waiting for the full scan to complete before you can start picking. After you confirm, the tool runs the **default full fetch** (`git fetch --all`, prune opt-in) unless you passed **`--fetch-mainline`**, or **full branch hydration** is implied by **`--sync`**, **`--risky`**, **`risky_mode`** in config, a **non-mainline `branch_mode`**, or **any `--branch-mode`** on the CLI. Quitting (**`q`** or **`ctrl+c`**) cancels the in-progress scan (in-flight `git` subprocesses are aborted via the scan context); **`ctrl+c`** outside raw TTY mode is treated like cancel.
+`git-rain --rain` opens an interactive picker. Repositories stream in as the filesystem scan finds them — no waiting for the full scan to complete before you can start picking. After you confirm, selected repos run through the same **bounded `fetch_workers` pool** as the default CLI (so one stalled remote does not block the rest). The tool runs the **default full fetch** (`git fetch --all`, prune opt-in) unless you passed **`--fetch-mainline`**, or **full branch hydration** is implied by **`--sync`**, **`--risky`**, **`risky_mode`** in config, a **non-mainline `branch_mode`**, or **any `--branch-mode`** on the CLI. Quitting (**`q`** or **`ctrl+c`**) cancels the in-progress scan (in-flight `git` subprocesses are aborted via the scan context); **`ctrl+c`** outside raw TTY mode is treated like cancel.
 
 Status strip and optional log panel are driven from structured session events. Confirming selection still exits TUI and runs fetch/sync work in normal CLI output.
 

cmd/default_stream_stall_test.go

@@ -294,7 +294,7 @@ func TestRunRainOnRepos_CancelSkipsRemaining(t *testing.T) {
 	done := make(chan error, 1)
 	start := time.Now()
 	go func() {
-		done <- runRainOnRepos(&registry.Registry{}, repos, rainOpts, false)
+		done <- runRainOnRepos(&registry.Registry{}, repos, rainOpts, false, 1)
 	}()
 
 	time.Sleep(300 * time.Millisecond)

cmd/rain_parallel_test.go

@@ -0,0 +1,137 @@
+package cmd
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/git-rain/git-rain/internal/git"
+	"github.com/git-rain/git-rain/internal/registry"
+	testutil "github.com/git-fire/git-testkit"
+)
+
+func TestRunRainOnRepos_ParallelStallDoesNotBlockHealthy(t *testing.T) {
+	baseURL, cleanup := startStallTCPListener(t)
+	defer cleanup()
+
+	scenario := testutil.NewScenario(t)
+	remote := scenario.CreateBareRepo("upstream")
+	healthy := scenario.CreateRepo("healthy").
+		AddFile("README.md", "ok\n").
+		Commit("init").
+		WithRemote("origin", remote)
+	healthy.Push("origin", healthy.GetDefaultBranch())
+
+	stall := testutil.CreateTestRepo(t, testutil.RepoOptions{
+		Name: "stall",
+		Remotes: map[string]string{
+			"origin": baseURL + "/stall.git",
+		},
+	})
+
+	repos := []git.Repository{
+		{Name: "healthy", Path: healthy.Path()},
+		{Name: "stall", Path: stall},
+	}
+
+	timeoutSec := 2
+	rainOpts := git.RainOptions{FetchTimeoutSec: timeoutSec}
+
+	oldStdout := os.Stdout
+	r, w, _ := os.Pipe()
+	os.Stdout = w
+	defer func() { os.Stdout = oldStdout }()
+
+	start := time.Now()
+	err := runRainOnRepos(&registry.Registry{}, repos, rainOpts, false, 2)
+	elapsed := time.Since(start)
+
+	_ = w.Close()
+	var out bytes.Buffer
+	_, _ = io.Copy(&out, r)
+
+	if err != nil {
+		t.Fatalf("runRainOnRepos: %v", err)
+	}
+	body := out.String()
+	if !strings.Contains(body, "rain delivered") {
+		t.Fatalf("expected summary, got:\n%s", body)
+	}
+	if !strings.Contains(body, "timed out") {
+		t.Fatalf("expected stall timeout in output, got:\n%s", body)
+	}
+	if elapsed > time.Duration(timeoutSec+8)*time.Second {
+		t.Fatalf("parallel rain took %v; expected ~%ds timeout bound", elapsed, timeoutSec)
+	}
+}
+
+func TestRunRainReposParallel_PreservesSelectionIndex(t *testing.T) {
+	baseURL, cleanup := startStallTCPListener(t)
+	defer cleanup()
+
+	scenario := testutil.NewScenario(t)
+	remote := scenario.CreateBareRepo("upstream")
+	fast := scenario.CreateRepo("repo-alpha").
+		AddFile("README.md", "ok\n").
+		Commit("init").
+		WithRemote("origin", remote)
+	fast.Push("origin", fast.GetDefaultBranch())
+
+	stall := testutil.CreateTestRepo(t, testutil.RepoOptions{
+		Name: "repo-beta",
+		Remotes: map[string]string{
+			"origin": baseURL + "/stall.git",
+		},
+	})
+
+	repos := []git.Repository{
+		{Name: "repo-alpha", Path: fast.Path()},
+		{Name: "repo-beta", Path: stall},
+	}
+
+	oldJSON := rainJSON
+	rainJSON = true
+	defer func() { rainJSON = oldJSON }()
+
+	oldStdout := os.Stdout
+	r, w, _ := os.Pipe()
+	os.Stdout = w
+	defer func() { os.Stdout = oldStdout }()
+
+	totals, reports := runRainReposParallel(&registry.Registry{}, repos, git.RainOptions{FetchTimeoutSec: 2}, false, 2)
+	_ = w.Close()
+	var out bytes.Buffer
+	_, _ = io.Copy(&out, r)
+
+	if totals.updated < 1 {
+		t.Fatalf("expected at least one updated repo, totals=%+v", totals)
+	}
+	if len(reports) != 2 {
+		t.Fatalf("expected 2 json reports, got %d", len(reports))
+	}
+
+	byName := map[string]jsonRepoResult{}
+	for _, rep := range reports {
+		byName[rep.Name] = rep
+	}
+	if byName["repo-alpha"].Index != 1 || byName["repo-alpha"].Total != 2 {
+		t.Fatalf("repo-alpha index/total = %d/%d, want 1/2", byName["repo-alpha"].Index, byName["repo-alpha"].Total)
+	}
+	if byName["repo-beta"].Index != 2 || byName["repo-beta"].Total != 2 {
+		t.Fatalf("repo-beta index/total = %d/%d, want 2/2", byName["repo-beta"].Index, byName["repo-beta"].Total)
+	}
+
+	var doc jsonRunDocument
+	if err := json.Unmarshal(out.Bytes(), &doc); err == nil && len(doc.Repositories) > 0 {
+		for _, rep := range doc.Repositories {
+			want := byName[rep.Name].Index
+			if rep.Index != want {
+				t.Fatalf("stdout json repo %s index %d, want %d", rep.Name, rep.Index, want)
+			}
+		}
+	}
+}

cmd/rain_workers.go

@@ -0,0 +1,85 @@
+package cmd
+
+import (
+	"io"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+
+	"github.com/git-rain/git-rain/internal/git"
+	"github.com/git-rain/git-rain/internal/registry"
+)
+
+type parallelRepoJob struct {
+	repo  git.Repository
+	index int // 1-based position in the selected repos slice
+}
+
+// runRainReposParallel runs fetch/rain for repos using the same bounded worker
+// pool as the default streaming CLI path.
+func runRainReposParallel(reg *registry.Registry, repos []git.Repository, rainOpts git.RainOptions, fullSync bool, fetchWorkers int) (rainTotals, []jsonRepoResult) {
+	if len(repos) == 0 {
+		return rainTotals{}, nil
+	}
+
+	total := len(repos)
+	workers := fetchWorkerCount(fetchWorkers)
+	repoChan := make(chan parallelRepoJob, len(repos))
+	for i, repo := range repos {
+		repoChan <- parallelRepoJob{repo: repo, index: i + 1}
+	}
+	close(repoChan)
+
+	var printMu sync.Mutex
+	var (
+		updated int64
+		skipped int64
+		frozen  int64
+		failed  int64
+	)
+	var jsonReports []jsonRepoResult
+	var jsonReportsMu sync.Mutex
+
+	runCtx := rainOpts.RunCtx
+	var workersWG sync.WaitGroup
+	for w := 0; w < workers; w++ {
+		workersWG.Add(1)
+		go func() {
+			defer workersWG.Done()
+			for job := range repoChan {
+				repo := job.repo
+				if runCtx != nil && runCtx.Err() != nil {
+					continue
+				}
+				var buf strings.Builder
+				delta, report := runRainOnRepo(reg, repo, rainOpts, fullSync, job.index, total, &buf, !rainJSON)
+
+				printMu.Lock()
+				if !rainJSON {
+					_, _ = io.WriteString(os.Stdout, buf.String())
+				}
+				printMu.Unlock()
+
+				if rainJSON {
+					jsonReportsMu.Lock()
+					jsonReports = append(jsonReports, report)
+					jsonReportsMu.Unlock()
+				}
+
+				atomic.AddInt64(&updated, int64(delta.updated))
+				atomic.AddInt64(&skipped, int64(delta.skipped))
+				atomic.AddInt64(&frozen, int64(delta.frozen))
+				atomic.AddInt64(&failed, int64(delta.failed))
+			}
+		}()
+	}
+	workersWG.Wait()
+
+	return rainTotals{
+		updated: int(updated),
+		skipped: int(skipped),
+		frozen:  int(frozen),
+		failed:  int(failed),
+	}, jsonReports
+}

cmd/root.go

@@ -359,31 +359,6 @@ func printRainSummary(totals rainTotals, runCtx context.Context) error {
 	return nil
 }
 
-// processRainRepositories runs fetch or full rain for each repository sequentially
-// and accumulates outcome counts. Each repo's output is written as a single block
-// directly to stdout. Used for the --rain TUI post-processing path; the default
-// CLI run uses runRainDefaultStream for parallel execution.
-// Stops starting new repos when RunCtx is cancelled (SIGINT/SIGTERM).
-func processRainRepositories(reg *registry.Registry, repos []git.Repository, rainOpts git.RainOptions, fullSync bool, totals *rainTotals, jsonReports *[]jsonRepoResult) {
-	for i, repo := range repos {
-		if rainOpts.RunCtx != nil && rainOpts.RunCtx.Err() != nil {
-			return
-		}
-		var buf strings.Builder
-		delta, report := runRainOnRepo(reg, repo, rainOpts, fullSync, i+1, len(repos), &buf, !rainJSON)
-		if !rainJSON {
-			_, _ = io.WriteString(os.Stdout, buf.String())
-		}
-		if jsonReports != nil {
-			*jsonReports = append(*jsonReports, report)
-		}
-		totals.updated += delta.updated
-		totals.skipped += delta.skipped
-		totals.frozen += delta.frozen
-		totals.failed += delta.failed
-	}
-}
-
 // reportRunInterrupted handles SIGINT/SIGTERM after work stops. Human mode prints
 // Aborted. and returns nil; --json emits schema_version 1 then a non-zero error so
 // agents do not treat cancel as a successful empty run.
@@ -746,7 +721,7 @@ func runRainTUIStream(cfg *config.Config, reg *registry.Registry, regPath string
 	}
 	fmt.Println()
 
-	return runRainOnRepos(reg, selected, rainOpts, fullSync)
+	return runRainOnRepos(reg, selected, rainOpts, fullSync, cfg.Global.FetchWorkers)
 }
 
 // fetchWorkerCount returns the per-run parallel fetch worker count, clamped
@@ -1043,7 +1018,7 @@ func runRainDefaultStream(cfg *config.Config, reg *registry.Registry, regPath st
 }
 
 // runRainOnRepos runs fetch or full rain on a pre-selected list of repos.
-func runRainOnRepos(reg *registry.Registry, repos []git.Repository, opts git.RainOptions, fullSync bool) error {
+func runRainOnRepos(reg *registry.Registry, repos []git.Repository, opts git.RainOptions, fullSync bool, fetchWorkers int) error {
 	if !rainJSON {
 		fmt.Println("🌧️  Git Rain")
 	}
@@ -1062,9 +1037,7 @@ func runRainOnRepos(reg *registry.Registry, repos []git.Repository, opts git.Rai
 		fmt.Println()
 	}
 
-	totals := rainTotals{}
-	var jsonReports []jsonRepoResult
-	processRainRepositories(reg, repos, opts, fullSync, &totals, &jsonReports)
+	totals, jsonReports := runRainReposParallel(reg, repos, opts, fullSync, fetchWorkers)
 
 	if opts.RunCtx != nil && opts.RunCtx.Err() != nil {
 		return reportRunInterrupted(opts, fullSync, totals, jsonReports)
@@ -1074,6 +1047,9 @@ func runRainOnRepos(reg *registry.Registry, repos []git.Repository, opts git.Rai
 		if err := emitJSONRun(buildJSONRunMeta(opts, fullSync), totals, jsonReports); err != nil {
 			return err
 		}
+		if opts.RunCtx != nil && opts.RunCtx.Err() != nil {
+			return fmt.Errorf("interrupted")
+		}
 		if totals.failed > 0 {
 			return fmt.Errorf("%d branch(es) failed — see --json output", totals.failed)
 		}