From 0cb07224f4c50e7fb710337c2ba1be8dff220e74 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 11 Oct 2026 11:02:06 +0000 Subject: [PATCH 1/7] feat(cli): parallel --rain fetch with bounded worker pool Run post-picker repos through the same fetch_workers pool as the default streaming path so one stalled remote cannot block later repos. Co-authored-by: Ben Schellenberger --- README.md | 2 +- cmd/rain_parallel_test.go | 69 +++++++++++++++++++++++++++++++++++++++ cmd/rain_workers.go | 69 +++++++++++++++++++++++++++++++++++++++ cmd/root.go | 26 ++------------- 4 files changed, 142 insertions(+), 24 deletions(-) create mode 100644 cmd/rain_parallel_test.go create mode 100644 cmd/rain_workers.go diff --git a/README.md b/README.md index 052b70a..396cb93 100644 --- a/README.md +++ b/README.md @@ -391,7 +391,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. diff --git a/cmd/rain_parallel_test.go b/cmd/rain_parallel_test.go new file mode 100644 index 0000000..f9c6dd3 --- /dev/null +++ b/cmd/rain_parallel_test.go @@ -0,0 +1,69 @@ +package cmd + +import ( + "bytes" + "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(®istry.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) + } +} diff --git a/cmd/rain_workers.go b/cmd/rain_workers.go new file mode 100644 index 0000000..8a61282 --- /dev/null +++ b/cmd/rain_workers.go @@ -0,0 +1,69 @@ +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" +) + +// 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 { + if len(repos) == 0 { + return rainTotals{} + } + + workers := fetchWorkerCount(fetchWorkers) + repoChan := make(chan git.Repository, len(repos)) + for _, repo := range repos { + repoChan <- repo + } + close(repoChan) + + var printMu sync.Mutex + var ( + updated int64 + skipped int64 + frozen int64 + failed int64 + seq int64 + ) + total := int64(len(repos)) + + runCtx := rainOpts.RunCtx + var workersWG sync.WaitGroup + for w := 0; w < workers; w++ { + workersWG.Add(1) + go func() { + defer workersWG.Done() + for repo := range repoChan { + if runCtx != nil && runCtx.Err() != nil { + break + } + current := int(atomic.AddInt64(&seq, 1)) + var buf strings.Builder + delta := runRainOnRepo(reg, repo, rainOpts, fullSync, current, int(total), &buf) + printMu.Lock() + _, _ = io.WriteString(os.Stdout, buf.String()) + printMu.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), + } +} diff --git a/cmd/root.go b/cmd/root.go index e82e256..cf241f4 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -346,25 +346,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. -func processRainRepositories(reg *registry.Registry, repos []git.Repository, rainOpts git.RainOptions, fullSync bool, totals *rainTotals) { - for i, repo := range repos { - if rainOpts.RunCtx != nil && rainOpts.RunCtx.Err() != nil { - break - } - var buf strings.Builder - delta := runRainOnRepo(reg, repo, rainOpts, fullSync, i+1, len(repos), &buf) - _, _ = io.WriteString(os.Stdout, buf.String()) - totals.updated += delta.updated - totals.skipped += delta.skipped - totals.frozen += delta.frozen - totals.failed += delta.failed - } -} - // runRainOnRepo runs the rain operation for a single repository, writing its // entire output block (header + branch lines + trailing blank line) into out. // totalStr is rendered as "[N/M] reponame"; pass "?" for M when the total is @@ -612,7 +593,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 @@ -866,7 +847,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 { fmt.Println("🌧️ Git Rain") if fullSync { if opts.RiskyMode { @@ -881,8 +862,7 @@ func runRainOnRepos(reg *registry.Registry, repos []git.Repository, opts git.Rai } fmt.Println() - totals := rainTotals{} - processRainRepositories(reg, repos, opts, fullSync, &totals) + totals := runRainReposParallel(reg, repos, opts, fullSync, fetchWorkers) return printRainSummary(totals, opts.RunCtx) } From 0e1c82fef4c72922546ee3960b89907e8fc4acc5 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 11 Oct 2026 11:10:01 +0000 Subject: [PATCH 2/7] fix(git): classify timeout and cancel errors after process kill Co-authored-by: Ben Schellenberger --- internal/git/network_git.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/internal/git/network_git.go b/internal/git/network_git.go index 00e0d92..bfcb597 100644 --- a/internal/git/network_git.go +++ b/internal/git/network_git.go @@ -67,6 +67,12 @@ func RunNetworkGit(parent context.Context, cmd *exec.Cmd, timeoutSec int) ([]byt select { case err := <-waitDone: if err != nil { + if errors.Is(runCtx.Err(), context.DeadlineExceeded) && timeoutSec > 0 { + return combined.Bytes(), &FetchTimedOutError{Seconds: timeoutSec} + } + if runCtx.Err() != nil { + return combined.Bytes(), runCtx.Err() + } return combined.Bytes(), err } return combined.Bytes(), nil From 3e35f61f7a952958306e8b7294e867425d6f4e7d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 11 Oct 2026 12:29:03 +0000 Subject: [PATCH 3/7] fix(cli): remove redundant newline in skipped fetch message Co-authored-by: Ben Schellenberger --- cmd/root.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/root.go b/cmd/root.go index 7c5f216..430fa94 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -373,7 +373,7 @@ func runRainOnRepo(reg *registry.Registry, repo git.Repository, rainOpts git.Rai if !fullSync && !rainFetchMainline { if fetchErr := fetchOnly(repoOpts.RunCtx, repo.Path, repoOpts); fetchErr != nil { if errors.Is(fetchErr, context.Canceled) { - _, _ = fmt.Fprintln(out, " ~ skipped\n") + _, _ = fmt.Fprintln(out, " ~ skipped") _, _ = fmt.Fprintln(out) return delta } From dd4d06478718e705980dcf3584c1571c44d23e2d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 12 Oct 2026 18:03:22 +0000 Subject: [PATCH 4/7] fix(cmd): pass fetch_workers to runRainOnRepos in stall test Co-authored-by: Ben Schellenberger --- cmd/default_stream_stall_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/default_stream_stall_test.go b/cmd/default_stream_stall_test.go index dff0391..ad9924b 100644 --- a/cmd/default_stream_stall_test.go +++ b/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(®istry.Registry{}, repos, rainOpts, false) + done <- runRainOnRepos(®istry.Registry{}, repos, rainOpts, false, 1) }() time.Sleep(300 * time.Millisecond) From 9e3eeeb51f7522961804ad2db126ba5a1b504719 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 12 Oct 2026 18:06:29 +0000 Subject: [PATCH 5/7] fix(cmd): drop unused processRainRepositories after parallel runRainOnRepos Co-authored-by: Ben Schellenberger --- cmd/root.go | 25 ------------------------- 1 file changed, 25 deletions(-) diff --git a/cmd/root.go b/cmd/root.go index 3ee8a4b..52fa3d0 100644 --- a/cmd/root.go +++ b/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. From cd617d414c313a08ad177d50b74da75762d01daf Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 13 Oct 2026 19:00:27 +0000 Subject: [PATCH 6/7] fix(cmd): use selection order for parallel rain repo indexes Co-authored-by: Ben Schellenberger --- cmd/rain_parallel_test.go | 68 +++++++++++++++++++++++++++++++++++++++ cmd/rain_workers.go | 20 +++++++----- 2 files changed, 80 insertions(+), 8 deletions(-) diff --git a/cmd/rain_parallel_test.go b/cmd/rain_parallel_test.go index f9c6dd3..bf70991 100644 --- a/cmd/rain_parallel_test.go +++ b/cmd/rain_parallel_test.go @@ -2,6 +2,7 @@ package cmd import ( "bytes" + "encoding/json" "io" "os" "strings" @@ -67,3 +68,70 @@ func TestRunRainOnRepos_ParallelStallDoesNotBlockHealthy(t *testing.T) { 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(®istry.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([]byte(out.String()), &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) + } + } + } +} diff --git a/cmd/rain_workers.go b/cmd/rain_workers.go index 99022f0..45befb0 100644 --- a/cmd/rain_workers.go +++ b/cmd/rain_workers.go @@ -11,6 +11,11 @@ import ( "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) { @@ -18,10 +23,11 @@ func runRainReposParallel(reg *registry.Registry, repos []git.Repository, rainOp return rainTotals{}, nil } + total := len(repos) workers := fetchWorkerCount(fetchWorkers) - repoChan := make(chan git.Repository, len(repos)) - for _, repo := range repos { - repoChan <- repo + repoChan := make(chan parallelRepoJob, len(repos)) + for i, repo := range repos { + repoChan <- parallelRepoJob{repo: repo, index: i + 1} } close(repoChan) @@ -31,9 +37,7 @@ func runRainReposParallel(reg *registry.Registry, repos []git.Repository, rainOp skipped int64 frozen int64 failed int64 - seq int64 ) - total := int64(len(repos)) var jsonReports []jsonRepoResult var jsonReportsMu sync.Mutex @@ -43,13 +47,13 @@ func runRainReposParallel(reg *registry.Registry, repos []git.Repository, rainOp workersWG.Add(1) go func() { defer workersWG.Done() - for repo := range repoChan { + for job := range repoChan { + repo := job.repo if runCtx != nil && runCtx.Err() != nil { continue } - current := int(atomic.AddInt64(&seq, 1)) var buf strings.Builder - delta, report := runRainOnRepo(reg, repo, rainOpts, fullSync, current, int(total), &buf, !rainJSON) + delta, report := runRainOnRepo(reg, repo, rainOpts, fullSync, job.index, total, &buf, !rainJSON) printMu.Lock() if !rainJSON { From 08b65847cdc59df6230021a7211afae4f9d84e7b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 13 Oct 2026 22:44:07 +0000 Subject: [PATCH 7/7] fix(cmd): use bytes.Buffer.Bytes in parallel index JSON test (staticcheck S1030) Co-authored-by: Ben Schellenberger --- cmd/rain_parallel_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmd/rain_parallel_test.go b/cmd/rain_parallel_test.go index bf70991..5e514a0 100644 --- a/cmd/rain_parallel_test.go +++ b/cmd/rain_parallel_test.go @@ -126,7 +126,7 @@ func TestRunRainReposParallel_PreservesSelectionIndex(t *testing.T) { } var doc jsonRunDocument - if err := json.Unmarshal([]byte(out.String()), &doc); err == nil && len(doc.Repositories) > 0 { + 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 {