From 1a59ae23b55d02311afc926221e07aba9df0e687 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Manso?= Date: Sun, 2 Aug 2026 15:37:31 +0100 Subject: [PATCH] perf(agentctl): parallelise the multi-repo git fan-outs (#2138) --- .../internal/agentctl/server/api/git.go | 233 ++++++++++----- .../agentctl/server/api/git_fanout_test.go | 271 ++++++++++++++++++ 2 files changed, 432 insertions(+), 72 deletions(-) create mode 100644 apps/backend/internal/agentctl/server/api/git_fanout_test.go diff --git a/apps/backend/internal/agentctl/server/api/git.go b/apps/backend/internal/agentctl/server/api/git.go index 5f9219b293..aa54dc7a93 100644 --- a/apps/backend/internal/agentctl/server/api/git.go +++ b/apps/backend/internal/agentctl/server/api/git.go @@ -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 ..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 ..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 diff --git a/apps/backend/internal/agentctl/server/api/git_fanout_test.go b/apps/backend/internal/agentctl/server/api/git_fanout_test.go new file mode 100644 index 0000000000..e1c5ed8a4a --- /dev/null +++ b/apps/backend/internal/agentctl/server/api/git_fanout_test.go @@ -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()) + } +}