diff --git a/README.md b/README.md index ff21e05..1b9a3a3 100644 --- a/README.md +++ b/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. 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) diff --git a/cmd/rain_parallel_test.go b/cmd/rain_parallel_test.go new file mode 100644 index 0000000..5e514a0 --- /dev/null +++ b/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(®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) + } +} + +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(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) + } + } + } +} diff --git a/cmd/rain_workers.go b/cmd/rain_workers.go new file mode 100644 index 0000000..45befb0 --- /dev/null +++ b/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 +} diff --git a/cmd/root.go b/cmd/root.go index 427188c..bc68d75 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. @@ -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) }