perf(agentctl): parallelise the multi-repo git fan-outs (#2138)

This commit is contained in:
João Manso
2026-08-02 15:37:31 +01:00
committed by GitHub
parent e077658a61
commit 1a59ae23b5
2 changed files with 432 additions and 72 deletions
+161 -72
View File
@@ -14,6 +14,7 @@ import (
"github.com/gin-gonic/gin"
"github.com/kandev/kandev/internal/agentctl/server/process"
"go.uber.org/zap"
"golang.org/x/sync/errgroup"
)
// safeBranchRefPattern mirrors the one in workspace_git_status.go so the
@@ -673,7 +674,11 @@ func (s *Server) handleGitLog(c *gin.Context) {
}
}
result, err := s.runGitLogForRepo(c, req, limit, req.Repo)
// c.Request.Context(), not c: *gin.Context satisfies context.Context, so
// passing c compiles but hands over a context whose Done() is nil unless
// ContextWithFallback is set — which it is not here. The git subprocesses
// would then outlive a client disconnect.
result, err := s.runGitLogForRepo(c.Request.Context(), req, limit, req.Repo)
if err != nil {
s.handleGitError(c, "log", err)
return
@@ -700,7 +705,7 @@ func (s *Server) computeMergeBase(
// runGitLogForRepo runs git log against a single repo subpath. Returns a
// result-with-error or a non-nil error for transport failures.
func (s *Server) runGitLogForRepo(
c *gin.Context,
ctx context.Context,
req GitLogRequest,
limit int,
repo string,
@@ -726,7 +731,7 @@ func (s *Server) runGitLogForRepo(
req.TargetBranch = ""
}
if req.TargetBranch != "" {
mergeBase, err := s.computeMergeBase(c.Request.Context(), gitOp, req.TargetBranch)
mergeBase, err := s.computeMergeBase(ctx, gitOp, req.TargetBranch)
if err == nil && mergeBase != "" {
baseCommit = mergeBase
} else {
@@ -738,7 +743,7 @@ func (s *Server) runGitLogForRepo(
// into the workspace's full HEAD history, mismatching the
// numstat-driven stats which fall through cleanly to per-file
// sums in the same scenario.
if tip, tipErr := gitOp.GetRevParse(c.Request.Context(), req.TargetBranch); tipErr == nil && tip != "" {
if tip, tipErr := gitOp.GetRevParse(ctx, req.TargetBranch); tipErr == nil && tip != "" {
baseCommit = tip
} else if err != nil {
s.logger.Warn("failed to compute merge-base and branch tip, falling back to since",
@@ -753,13 +758,13 @@ func (s *Server) runGitLogForRepo(
// ref, so a live non-default target (e.g. origin/develop) keeps its own
// authoritative merge-base. No-op when the base is already current.
baseCommit = gitOp.CorrectStaleComparisonBase(
c.Request.Context(),
ctx,
baseCommit,
req.TargetBranch,
)
}
return gitOp.GetLog(c.Request.Context(), baseCommit, limit)
return gitOp.GetLog(ctx, baseCommit, limit)
}
// perRepoLogOutcome captures the outcome of running git log against a single
@@ -851,39 +856,53 @@ func (s *Server) handleGitLogMultiRepo(
subpaths []string,
limit int,
) {
outcomes := make([]perRepoLogOutcome, 0, len(subpaths))
for _, sub := range subpaths {
// Multi-repo: the caller-supplied `since` is a SHA that only exists in
// the primary repo, and `target_branch` is the primary repo's base
// branch (e.g. "main"). Both can be wrong for sibling repos — running
// `git log <foreign-sha>..HEAD` in lvc fails outright and the repo's
// commits silently disappear from the merged response. Compute a
// per-repo base through the workspace tracker so the task's configured
// base branch and normal integration-branch fallbacks both work.
perRepoReq := req
perRepoReq.Since = ""
perRepoReq.TargetBranch = ""
if base := s.resolvePerRepoBase(c, sub); base != "" {
perRepoReq.Since = base
}
result, err := s.runGitLogForRepo(c, perRepoReq, limit, sub)
if err != nil {
s.logger.Warn("git log for repo failed",
zap.String("repo", sub), zap.Error(err))
} else if !result.Success {
s.logger.Warn("git log for repo returned failure",
zap.String("repo", sub), zap.String("error", result.Error))
}
outcomes = append(outcomes, perRepoLogOutcome{
subpath: sub,
result: result,
err: err,
})
}
// Parallel fan-out, mirroring handleGitStatusMulti. Each repo is an
// independent git invocation against its own worktree, so running them
// serially made latency the sum over repositories: on Windows, where a git
// spawn alone costs ~150ms, an 11-repo task could exceed the gateway's
// 60s singleflight budget and the caller would see a timeout rather than a
// slow answer. Writing by index keeps the merge order deterministic.
ctx := c.Request.Context()
outcomes := fanOutRepos(ctx, subpaths, func(ctx context.Context, sub string) perRepoLogOutcome {
return s.collectLogForRepo(ctx, req, limit, sub)
})
c.JSON(http.StatusOK, mergeGitLogResults(outcomes, limit))
}
// collectLogForRepo runs git log for one repository of a multi-repo fan-out.
// It takes a plain context and never touches the gin context, which is what
// makes it safe to call from several goroutines at once.
func (s *Server) collectLogForRepo(
ctx context.Context,
req GitLogRequest,
limit int,
sub string,
) perRepoLogOutcome {
// Multi-repo: the caller-supplied `since` is a SHA that only exists in
// the primary repo, and `target_branch` is the primary repo's base
// branch (e.g. "main"). Both can be wrong for sibling repos — running
// `git log <foreign-sha>..HEAD` in lvc fails outright and the repo's
// commits silently disappear from the merged response. Compute a
// per-repo base through the workspace tracker so the task's configured
// base branch and normal integration-branch fallbacks both work.
perRepoReq := req
perRepoReq.Since = ""
perRepoReq.TargetBranch = ""
if base := s.resolvePerRepoBase(ctx, sub); base != "" {
perRepoReq.Since = base
}
result, err := s.runGitLogForRepo(ctx, perRepoReq, limit, sub)
if err != nil {
s.logger.Warn("git log for repo failed",
zap.String("repo", sub), zap.Error(err))
} else if !result.Success {
s.logger.Warn("git log for repo returned failure",
zap.String("repo", sub), zap.String("error", result.Error))
}
return perRepoLogOutcome{subpath: sub, result: result, err: err}
}
// handleGitCumulativeDiff handles GET /api/v1/git/cumulative-diff.
//
// Multi-repo: when the caller doesn't pin a repo (?repo= empty) and the
@@ -923,9 +942,13 @@ func (s *Server) handleGitCumulativeDiff(c *gin.Context) {
}
}
result := s.runGitCumulativeDiffForRepo(c, req.Base, req.TargetBranch, req.Repo)
if result == nil {
return // gitOp lookup error already wrote the response
result, status, err := s.runGitCumulativeDiffForRepo(c.Request.Context(), req.Base, req.TargetBranch, req.Repo)
if err != nil {
c.JSON(status, process.CumulativeDiffResult{
Success: false,
Error: err.Error(),
})
return
}
c.JSON(http.StatusOK, result)
}
@@ -940,19 +963,15 @@ func (s *Server) handleGitCumulativeDiff(c *gin.Context) {
// the diff updates as main moves forward and excludes file changes brought
// in via merges from main.
func (s *Server) runGitCumulativeDiffForRepo(
c *gin.Context,
ctx context.Context,
base, targetBranch, repo string,
) *process.CumulativeDiffResult {
) (*process.CumulativeDiffResult, int, error) {
gitOp, gitOpErr := s.procMgr.GitOperatorFor(repo)
if gitOpErr != nil {
c.JSON(http.StatusBadRequest, process.CumulativeDiffResult{
Success: false,
Error: gitOpErr.Error(),
})
return nil
return nil, http.StatusBadRequest, gitOpErr
}
if targetBranch != "" {
switch mb, err := s.computeMergeBase(c.Request.Context(), gitOp, targetBranch); {
switch mb, err := s.computeMergeBase(ctx, gitOp, targetBranch); {
case err != nil:
s.logger.Warn("cumulative diff: merge-base failed, using stored base",
zap.String("target_branch", targetBranch),
@@ -973,21 +992,17 @@ func (s *Server) runGitCumulativeDiffForRepo(
// the diff excludes changes that already landed on the integration
// branch. Only fires when the target has no live upstream ref. No-op
// when the base is already current.
base = gitOp.CorrectStaleComparisonBase(c.Request.Context(), base, targetBranch)
base = gitOp.CorrectStaleComparisonBase(ctx, base, targetBranch)
}
result, err := gitOp.GetCumulativeDiff(c.Request.Context(), base)
result, err := gitOp.GetCumulativeDiff(ctx, base)
if err != nil {
s.logger.Error("git cumulative diff failed",
zap.String("base", base),
zap.String("repo", repo),
zap.Error(err))
c.JSON(http.StatusInternalServerError, process.CumulativeDiffResult{
Success: false,
Error: err.Error(),
})
return nil
return nil, http.StatusInternalServerError, err
}
return result
return result, http.StatusOK, nil
}
// handleGitCumulativeDiffMultiRepo fans cumulative diff out across every
@@ -1002,33 +1017,45 @@ func (s *Server) handleGitCumulativeDiffMultiRepo(
req GitCumulativeDiffRequest,
subpaths []string,
) {
// Parallel fan-out for the same reason as handleGitLogMultiRepo: each repo
// is an independent git invocation, and serially the request took the sum
// over repositories, which on Windows is enough to blow the gateway's 60s
// budget on a task with many repos.
ctx := c.Request.Context()
outcomes := fanOutRepos(ctx, subpaths, func(ctx context.Context, sub string) perRepoDiffOutcome {
return s.collectCumulativeDiffForRepo(ctx, sub)
})
// Merging stays sequential: it writes into one shared map, and doing it in
// subpath order keeps the merged output independent of goroutine timing.
merged := process.CumulativeDiffResult{
Files: make(map[string]interface{}),
Success: true,
}
anyOK := false
for _, sub := range subpaths {
base := s.resolvePerRepoBase(c, sub)
if base == "" {
s.logger.Warn("cumulative diff: no per-repo base, skipping",
zap.String("repo", sub))
continue
for _, outcome := range outcomes {
// A hard failure still aborts the whole request, as it did serially.
// Reporting the first one in subpath order rather than the first to
// occur keeps the response deterministic now that the repos race.
if outcome.err != nil {
c.JSON(outcome.status, process.CumulativeDiffResult{
Success: false,
Error: outcome.err.Error(),
})
return
}
// Multi-repo: base is already resolved per-repo via resolvePerRepoBase,
// so we pass empty target_branch to skip the second merge-base attempt.
result := s.runGitCumulativeDiffForRepo(c, base, "", sub)
if result == nil {
return // response already written
if outcome.result == nil {
continue // no per-repo base; already logged
}
if !result.Success {
if !outcome.result.Success {
s.logger.Warn("cumulative diff for repo returned failure",
zap.String("repo", sub),
zap.String("error", result.Error))
zap.String("repo", outcome.subpath),
zap.String("error", outcome.result.Error))
continue
}
anyOK = true
merged.TotalCommits += result.TotalCommits
mergeCumulativeFiles(merged.Files, result.Files, sub, result.BaseCommit)
merged.TotalCommits += outcome.result.TotalCommits
mergeCumulativeFiles(merged.Files, outcome.result.Files, outcome.subpath, outcome.result.BaseCommit)
}
if !anyOK {
merged.Success = false
@@ -1037,6 +1064,69 @@ func (s *Server) handleGitCumulativeDiffMultiRepo(
c.JSON(http.StatusOK, merged)
}
// perRepoDiffOutcome carries one repository's cumulative-diff result out of the
// fan-out. A nil result with a nil error means the repo had no resolvable base
// and was skipped; a non-nil error is a hard failure that aborts the request,
// with status holding the code the serial version would have written.
type perRepoDiffOutcome struct {
subpath string
result *process.CumulativeDiffResult
status int
err error
}
// collectCumulativeDiffForRepo runs the cumulative diff for one repository of a
// multi-repo fan-out. Like collectLogForRepo it takes a plain context and never
// writes to the gin context, so it is safe to run concurrently.
func (s *Server) collectCumulativeDiffForRepo(ctx context.Context, sub string) perRepoDiffOutcome {
base := s.resolvePerRepoBase(ctx, sub)
if base == "" {
s.logger.Warn("cumulative diff: no per-repo base, skipping",
zap.String("repo", sub))
return perRepoDiffOutcome{subpath: sub}
}
// Multi-repo: base is already resolved per-repo via resolvePerRepoBase,
// so we pass empty target_branch to skip the second merge-base attempt.
result, status, err := s.runGitCumulativeDiffForRepo(ctx, base, "", sub)
return perRepoDiffOutcome{subpath: sub, result: result, status: status, err: err}
}
// fanOutRepos runs collect once per repository, concurrently, and returns the
// results in subpath order regardless of which finished first. Every repository
// is an independent git invocation against its own worktree, so serial
// execution made request latency the sum over repositories rather than the
// slowest one — on Windows, where a git spawn costs ~150ms before doing any
// work, a task with a dozen repositories could exceed the gateway's 60s budget
// and the caller would see a timeout instead of a slow answer.
//
// Concurrency is bounded so a single request cannot occupy the whole shared git
// budget. The process-wide throttle in common/subproc caps subprocesses at 12 by
// default, and it is shared with background workspace polling — an unbounded
// fan-out over a dozen repositories takes all of it, which is what pushed the
// pollers into their own command timeouts.
//
// maxGitFanout is a judgement call rather than a derived number: subproc exposes
// no public accessor for its cap, and the admission design being discussed in
// #2150 is where the real answer belongs. Until then this leaves headroom for
// polling while still parallelising the common case.
const maxGitFanout = 8
func fanOutRepos[T any](ctx context.Context, subpaths []string, collect func(context.Context, string) T) []T {
results := make([]T, len(subpaths))
group, groupCtx := errgroup.WithContext(ctx)
group.SetLimit(maxGitFanout)
for i, sub := range subpaths {
group.Go(func() error {
results[i] = collect(groupCtx, sub)
return nil
})
}
// collect never returns an error — per-repository failures travel inside T
// so the merge can report them — so Wait only surfaces context cancellation.
_ = group.Wait()
return results
}
// resolvePerRepoBase returns the comparison anchor owned by the repository's
// workspace tracker. Multi-repo tasks share no single base commit across
// repos, and each tracker carries that repository's configured task base.
@@ -1046,12 +1136,11 @@ func (s *Server) handleGitCumulativeDiffMultiRepo(
// comparison policy here using the tracker's resolved base branch.
// Without this, a repo whose base branch is a merged/deleted stacked parent
// lingering only as a local ref keeps inflating the commit count and diff.
func (s *Server) resolvePerRepoBase(c *gin.Context, repo string) string {
func (s *Server) resolvePerRepoBase(ctx context.Context, repo string) string {
tracker, err := s.procMgr.GetWorkspaceTrackerFor(repo)
if err != nil {
return ""
}
ctx := c.Request.Context()
base, baseBranch := tracker.ResolveBaseAnchor(ctx)
if base == "" || baseBranch == "" {
return base
@@ -0,0 +1,271 @@
package api
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/kandev/kandev/internal/agentctl/server/config"
"github.com/kandev/kandev/internal/agentctl/server/process"
"github.com/kandev/kandev/internal/common/logger"
)
// TestFanOutReposRunsConcurrently proves parallelism without measuring how long
// anything takes. Every collect call blocks on a gate that opens only once
// maxGitFanout of them have arrived, so serial execution cannot pass: the first
// would wait on the gate forever and the rest would never start. The timeout is
// a failure detector rather than a threshold, so machine speed does not enter
// into it.
//
// The gate is plain channels rather than the git shim in
// git_status_fresh_concurrency_test.go, which is //go:build !windows because it
// needs mkfifo. This one runs everywhere and covers the shared helper both
// multi-repo handlers use.
func TestFanOutReposRunsConcurrently(t *testing.T) {
subpaths := []string{"frontend", "backend", "shared"}
gate := make(chan struct{})
var arrived sync.WaitGroup
arrived.Add(len(subpaths))
done := make(chan []string, 1)
go func() {
done <- fanOutRepos(context.Background(), subpaths, func(_ context.Context, sub string) string {
arrived.Done()
<-gate
return sub
})
}()
waitOrFail(t, &arrived, "not every repository entered the fan-out; the work is running serially")
close(gate)
select {
case got := <-done:
if len(got) != len(subpaths) {
t.Fatalf("got %d results, want %d", len(got), len(subpaths))
}
// Order follows subpaths, not completion order, or the merged output
// would depend on goroutine scheduling.
for i, want := range subpaths {
if got[i] != want {
t.Errorf("result[%d] = %q, want %q — results must be ordered by subpath", i, got[i], want)
}
}
case <-time.After(10 * time.Second):
t.Fatal("fan-out did not return after the gate opened")
}
}
// TestFanOutReposRespectsLimit pins the bound. With more repositories than
// maxGitFanout, the group must admit exactly maxGitFanout at once — enough to
// parallelise, never enough for one request to take the whole shared git
// budget from the background pollers.
func TestFanOutReposRespectsLimit(t *testing.T) {
subpaths := make([]string, maxGitFanout+4)
for i := range subpaths {
subpaths[i] = fmt.Sprintf("repo%d", i)
}
var inFlight, peak int64
admitted := make(chan struct{}, len(subpaths))
release := make(chan struct{})
done := make(chan []string, 1)
go func() {
done <- fanOutRepos(context.Background(), subpaths, func(_ context.Context, sub string) string {
cur := atomic.AddInt64(&inFlight, 1)
for {
old := atomic.LoadInt64(&peak)
if cur <= old || atomic.CompareAndSwapInt64(&peak, old, cur) {
break
}
}
admitted <- struct{}{}
<-release
atomic.AddInt64(&inFlight, -1)
return sub
})
}()
// Reading exactly maxGitFanout admissions blocks until the group is full,
// which is the parallelism half. It cannot over-read: the rest are held
// behind SetLimit until we release these.
for i := 0; i < maxGitFanout; i++ {
select {
case <-admitted:
case <-time.After(10 * time.Second):
t.Fatalf("only %d of %d workers were admitted; the group is not parallel", i, maxGitFanout)
}
}
if got := atomic.LoadInt64(&peak); got > maxGitFanout {
t.Fatalf("peak concurrency = %d, want at most %d", got, maxGitFanout)
}
close(release)
select {
case got := <-done:
if len(got) != len(subpaths) {
t.Fatalf("got %d results, want %d", len(got), len(subpaths))
}
case <-time.After(20 * time.Second):
t.Fatal("fan-out did not drain after the gate opened")
}
if got := atomic.LoadInt64(&peak); got != maxGitFanout {
t.Errorf("peak concurrency = %d, want exactly %d", got, maxGitFanout)
}
}
// TestFanOutReposPropagatesCancellation proves the context handed to each
// collect call is derived from the caller's, so a client that disconnects stops
// the git work in every repository rather than only the ones not yet started.
func TestFanOutReposPropagatesCancellation(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
subpaths := []string{"frontend", "backend"}
var arrived sync.WaitGroup
arrived.Add(len(subpaths))
observed := make(chan error, len(subpaths))
done := make(chan []string, 1)
go func() {
done <- fanOutRepos(ctx, subpaths, func(callCtx context.Context, sub string) string {
arrived.Done()
<-callCtx.Done()
observed <- callCtx.Err()
return sub
})
}()
waitOrFail(t, &arrived, "workers did not start")
cancel()
select {
case <-done:
case <-time.After(10 * time.Second):
t.Fatal("fan-out did not return after cancellation; the context is not reaching the workers")
}
for i := 0; i < len(subpaths); i++ {
if err := <-observed; !errors.Is(err, context.Canceled) {
t.Errorf("worker %d saw %v, want context.Canceled", i, err)
}
}
}
// waitOrFail blocks on wg and fails with msg if it does not complete, so the
// barrier reads the same way in every test above.
func waitOrFail(t *testing.T, wg *sync.WaitGroup, msg string) {
t.Helper()
ready := make(chan struct{})
go func() {
wg.Wait()
close(ready)
}()
select {
case <-ready:
case <-time.After(10 * time.Second):
t.Fatal(msg)
}
}
// TestGitLogCancellationReachesGitOperation guards the single-repo path.
// runGitLogForRepo takes a context.Context and *gin.Context satisfies that
// interface, so handing it the gin context compiles — but this server is built
// with gin.New() and does not enable ContextWithFallback, so that context's
// Done() is nil and cancellation never reaches the subprocesses. With the bug
// the handler runs to completion and returns commits for a request the caller
// had already abandoned.
func TestGitLogCancellationReachesGitOperation(t *testing.T) {
repoDir := t.TempDir()
seedCancellationRepo(t, repoDir)
base := strings.TrimSpace(runGitAPI(t, repoDir, "rev-parse", "main"))
log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error"})
cfg := &config.InstanceConfig{WorkDir: repoDir}
srv := NewServer(cfg, process.NewManager(cfg, log), nil, nil, log)
path := "/api/v1/git/log?limit=100&since=" + base
assertCancellationStopsGitLog(t, srv, path)
}
// TestGitLogMultiRepoCancellationReachesGitOperation covers the same guarantee
// on the fan-out path, where every repository runs on its own goroutine and has
// to inherit the request context rather than the gin one.
func TestGitLogMultiRepoCancellationReachesGitOperation(t *testing.T) {
taskRoot := t.TempDir()
bases := map[string]string{}
for _, repo := range []string{"frontend", "backend"} {
repoDir := filepath.Join(taskRoot, repo)
if err := os.Mkdir(repoDir, 0o755); err != nil {
t.Fatalf("mkdir %s: %v", repo, err)
}
seedCancellationRepo(t, repoDir)
bases[repo] = "main"
}
log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error"})
cfg := &config.InstanceConfig{WorkDir: taskRoot, BaseBranches: bases}
srv := NewServer(cfg, process.NewManager(cfg, log), nil, nil, log)
assertCancellationStopsGitLog(t, srv, "/api/v1/git/log?limit=100")
}
// seedCancellationRepo builds a repository with one commit on main and one on a
// feature branch, so a git log over the divergence has something to return.
func seedCancellationRepo(t *testing.T, repoDir string) {
t.Helper()
runGitAPI(t, repoDir, "init", "--initial-branch=main")
runGitAPI(t, repoDir, "config", "user.email", "test@test.com")
runGitAPI(t, repoDir, "config", "user.name", "Test User")
writeFileAPI(t, repoDir, "README.md", "base\n")
runGitAPI(t, repoDir, "add", ".")
runGitAPI(t, repoDir, "commit", "-m", "initial")
runGitAPI(t, repoDir, "checkout", "-b", "feature/cancel")
writeFileAPI(t, repoDir, "changed.txt", "change\n")
runGitAPI(t, repoDir, "add", ".")
runGitAPI(t, repoDir, "commit", "-m", "feature change")
}
// assertCancellationStopsGitLog checks the same request twice: once live, to
// prove the fixture actually produces commits, and once with an already
// cancelled context, which must not. Without the live half a broken fixture
// would look like a passing cancellation test.
func assertCancellationStopsGitLog(t *testing.T, srv *Server, path string) {
t.Helper()
live := httptest.NewRecorder()
srv.Router().ServeHTTP(live, httptest.NewRequest(http.MethodGet, path, nil))
var liveResult process.GitLogResult
if err := json.Unmarshal(live.Body.Bytes(), &liveResult); err != nil {
t.Fatalf("decode live git log: %v", err)
}
if len(liveResult.Commits) == 0 {
t.Fatalf("fixture returned no commits with a live context: %s", live.Body.String())
}
ctx, cancel := context.WithCancel(context.Background())
cancel()
cancelled := httptest.NewRecorder()
srv.Router().ServeHTTP(cancelled, httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx))
var cancelledResult process.GitLogResult
if err := json.Unmarshal(cancelled.Body.Bytes(), &cancelledResult); err != nil {
// A non-JSON body means the handler errored out, which is also a
// legitimate way for cancellation to surface.
return
}
if len(cancelledResult.Commits) > 0 {
t.Errorf("returned %d commits for a cancelled request; cancellation is not reaching the git operation: %s",
len(cancelledResult.Commits), cancelled.Body.String())
}
}