From 79d4f84bf27543c10520962c06b279909aa7ea9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Tue, 2 Jun 2026 19:37:20 +0100 Subject: [PATCH] fix(backend): cache & singleflight live PR feedback fetches (#1237) --- .../backend/internal/common/subproc/shared.go | 69 ++++ .../internal/common/subproc/shared_test.go | 93 ++++- apps/backend/internal/github/errors.go | 69 +++- apps/backend/internal/github/errors_test.go | 43 +++ apps/backend/internal/github/gh_client.go | 120 ++++--- apps/backend/internal/github/graphql.go | 170 ++++++++- apps/backend/internal/github/handlers.go | 19 +- apps/backend/internal/github/service.go | 95 ++++++ apps/backend/internal/github/service_pr.go | 48 ++- .../github/service_pr_feedback_test.go | 107 ++++++ .../internal/github/service_pr_watch.go | 95 +++++- .../github/service_pr_watch_batched.go | 186 ++++++++-- .../service_repo_negative_cache_test.go | 322 ++++++++++++++++++ .../github/service_search_cache_test.go | 2 + .../backend/internal/github/service_tokens.go | 30 +- apps/backend/internal/github/ttl_cache.go | 65 ++++ .../hooks/domains/github/use-task-pr.test.tsx | 82 +++++ apps/web/hooks/domains/github/use-task-pr.ts | 65 +++- 18 files changed, 1559 insertions(+), 121 deletions(-) create mode 100644 apps/backend/internal/github/service_pr_feedback_test.go create mode 100644 apps/backend/internal/github/service_repo_negative_cache_test.go create mode 100644 apps/web/hooks/domains/github/use-task-pr.test.tsx diff --git a/apps/backend/internal/common/subproc/shared.go b/apps/backend/internal/common/subproc/shared.go index 0ee6eee90d..8b2fe89adc 100644 --- a/apps/backend/internal/common/subproc/shared.go +++ b/apps/backend/internal/common/subproc/shared.go @@ -5,8 +5,16 @@ import ( "os" "os/exec" "strconv" + "time" ) +// defaultGHExecTimeout is the per-command exec budget the +// Run*AfterAcquire helpers apply after the throttle slot is acquired. +// 30s matches the gh client's default WithTimeout for plain `gh api` +// calls; callers needing more (paginated REST calls) or less (cheap +// `gh auth status`) pass an explicit value. +const defaultGHExecTimeout = 30 * time.Second + // Shared throttle singletons. // // Both gh and git are spawned from many packages across the backend @@ -136,3 +144,64 @@ func RunGHCombinedOutput(ctx context.Context, cmd *exec.Cmd) ([]byte, error) { defer release() return cmd.CombinedOutput() } + +// RunGHAfterAcquire / RunGHCombinedAfterAcquire mirror the git +// helper runGitCombinedAfterAcquire in apps/backend/internal/worktree: +// the exec timer starts only AFTER the throttle slot is acquired so +// throttle queue time can't burn the per-command budget. Without this +// ordering, a queued waiter inherits its parent's WS-bound deadline, +// gets a slot just before the deadline fires, and is killed with +// `signal: killed` the moment it execs — producing the killed (192×) / +// context deadline exceeded (96×) cascade in the SyncWatchesBatched +// storm logs. +// +// Returns (out, runErr, execCtxErr). The exec-ctx error lets callers +// tell a context-driven kill (timeout) from a regular gh failure when +// classifying fallbacks — see worktree.classifyGitFallbackReason for the +// pattern. The caller is expected to build `cmd` lazily (typically +// `exec.CommandContext(execCtx, "gh", args...)`) via the supplied +// builder closure so the exec context attaches to the right command. +// `execTimeout <= 0` falls back to defaultGHExecTimeout. +func RunGHCombinedAfterAcquire( + ctx context.Context, execTimeout time.Duration, build func(execCtx context.Context) *exec.Cmd, +) ([]byte, error, error) { + release, err := ghThrottle.Acquire(ctx) + if err != nil { + return nil, err, ctx.Err() + } + defer release() + execCtx, cancel := withExecTimeout(ctx, execTimeout) + defer cancel() + cmd := build(execCtx) + out, runErr := cmd.CombinedOutput() + return out, runErr, execCtx.Err() +} + +// RunGHAfterAcquire is RunGHCombinedAfterAcquire's plain `cmd.Run` sibling. +// Caller owns Stdout/Stderr wiring on the returned `cmd` (build closure +// runs synchronously inside the throttle slot so wiring happens after +// acquire, never before). +func RunGHAfterAcquire( + ctx context.Context, execTimeout time.Duration, build func(execCtx context.Context) *exec.Cmd, +) (error, error) { + release, err := ghThrottle.Acquire(ctx) + if err != nil { + return err, ctx.Err() + } + defer release() + execCtx, cancel := withExecTimeout(ctx, execTimeout) + defer cancel() + cmd := build(execCtx) + runErr := cmd.Run() + return runErr, execCtx.Err() +} + +// withExecTimeout returns a child context bounded by execTimeout +// (defaultGHExecTimeout when execTimeout<=0). Centralised so the two +// AfterAcquire helpers can't drift on default-timeout behaviour. +func withExecTimeout(ctx context.Context, execTimeout time.Duration) (context.Context, context.CancelFunc) { + if execTimeout <= 0 { + execTimeout = defaultGHExecTimeout + } + return context.WithTimeout(ctx, execTimeout) +} diff --git a/apps/backend/internal/common/subproc/shared_test.go b/apps/backend/internal/common/subproc/shared_test.go index 7ab4401e1e..57a29a0196 100644 --- a/apps/backend/internal/common/subproc/shared_test.go +++ b/apps/backend/internal/common/subproc/shared_test.go @@ -1,6 +1,13 @@ package subproc -import "testing" +import ( + "context" + "os/exec" + "runtime" + "sync" + "testing" + "time" +) func TestResolveCap(t *testing.T) { const env = "KANDEV_SUBPROC_CAP_TEST" @@ -69,3 +76,87 @@ func TestGHGitAreDistinctThrottles(t *testing.T) { t.Fatal("GH() and Git() returned the same throttle instance") } } + +// TestRunGHCombinedAfterAcquire_ExecTimeoutSurvivesAcquireWait is the +// regression test for the killed (192×) / context-deadline-exceeded +// (96×) cascade in the SyncWatchesBatched storm: when a queued waiter +// finally gets a throttle slot, its per-command exec budget MUST start +// fresh from that moment, not be eaten by the queue wait. +// +// Setup: pin the gh throttle to cap=1 and hold the slot for the +// acquireWait window. The second caller invokes +// RunGHCombinedAfterAcquire with a build closure that just records the +// execCtx deadline relative to the current wallclock. Pre-fix (timer +// started before Acquire) the recorded deadline would be in the past +// the moment Acquire returns; post-fix the deadline is acquireWait +// ahead of the AfterFunc fire, i.e. ~execTimeout ahead of now. +func TestRunGHCombinedAfterAcquire_ExecTimeoutSurvivesAcquireWait(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("uses POSIX `true` binary as a no-op subprocess") + } + restore := GH().SetCapForTest(1) + defer restore() + + ctx := context.Background() + holdRelease, err := GH().Acquire(ctx) + if err != nil { + t.Fatalf("acquire holder: %v", err) + } + + // Sized so a pre-fix regression (timer starts BEFORE acquire) yields + // a NEGATIVE remaining budget at the build callback, while a correct + // after-acquire timer yields a budget within (~50%, 100%] of + // execTimeout. The exact fraction depends on scheduler latency + // between AfterFunc firing and the second goroutine resuming, so the + // lower bound is generous. + const acquireWait = 200 * time.Millisecond + const execTimeout = 500 * time.Millisecond + + var wg sync.WaitGroup + wg.Add(1) + var ( + gotRemaining time.Duration + gotRunErr error + queued = make(chan struct{}) + ) + go func() { + defer wg.Done() + // Signal once the goroutine is about to call into the throttle — + // at that point it WILL be a waiter (cap=1, the slot is held). + // Pre-fix used time.AfterFunc(acquireWait, ...) to "give the + // waiter time to queue", which was non-deterministic across CI + // schedulers and could miss the very regression it's meant to + // catch. + close(queued) + _, gotRunErr, _ = RunGHCombinedAfterAcquire(ctx, execTimeout, func(execCtx context.Context) *exec.Cmd { + if dl, ok := execCtx.Deadline(); ok { + gotRemaining = time.Until(dl) + } + return exec.CommandContext(execCtx, "true") + }) + }() + // Wait for the goroutine to be ready to call Acquire, then yield + // until the throttle records the queue depth (waiters >= 1). This + // replaces the prior time.AfterFunc-based barrier, which fired on a + // scheduler timer rather than on the actual queue state and could + // miss the regression on slow CI hosts. After that, sleep for + // acquireWait so the pre-fix regression (timer starts BEFORE + // Acquire) would have its deadline burned by the wall-clock delay. + <-queued + for metricInt(subprocWaiters, "gh") < 1 { + runtime.Gosched() + } + time.Sleep(acquireWait) + holdRelease() + wg.Wait() + + // Pre-fix: gotRemaining would be roughly execTimeout - acquireWait, + // or negative if acquireWait > execTimeout. Post-fix: gotRemaining + // should land in (execTimeout/2, execTimeout]. + if gotRemaining <= execTimeout/2 { + t.Errorf("execCtx deadline was only %v from now (execTimeout=%v); acquire wait burned through the budget", gotRemaining, execTimeout) + } + if gotRunErr != nil { + t.Logf("`true` runErr=%v (informational; deadline budget is the assertion)", gotRunErr) + } +} diff --git a/apps/backend/internal/github/errors.go b/apps/backend/internal/github/errors.go index 4c8e85dc35..d2f4ebafa0 100644 --- a/apps/backend/internal/github/errors.go +++ b/apps/backend/internal/github/errors.go @@ -1,6 +1,9 @@ package github -import "errors" +import ( + "errors" + "strings" +) // ErrInvalidPRURL signals that a caller-supplied PR URL could not be parsed. // Used by AssociateExistingPRByURL so HTTP callers can translate the failure @@ -25,3 +28,67 @@ var ErrSelfApprove = errors.New("cannot approve your own pull request") // error so HTTP callers can distinguish a validation failure (HTTP 400) // from a secret-store write failure (HTTP 500). var ErrInvalidToken = errors.New("invalid token") + +// ErrRepoNotResolvable signals that GitHub reported the repository as +// missing, deleted, or inaccessible to the authenticated principal. The PR +// watch flood storm this fixes (SyncWatchesBatched hammering the same dead +// repo on every 5s frontend poll) classifies upstream errors via +// isRepoNotResolvableErr and feeds the result into a 10-minute negative +// cache so subsequent watch syncs short-circuit before acquiring the gh +// throttle. The cache is also flushed on token swap/clear (see +// Service.ClearRepoErrorCache wired from ConfigureToken/ClearToken) so a +// re-auth doesn't have to wait out the TTL. Use errors.Is(err, +// ErrRepoNotResolvable) at boundaries (WS handler, poller) where the +// caller needs to differentiate a "stop retrying" failure from a +// transient upstream blip. +var ErrRepoNotResolvable = errors.New("github: repository not resolvable") + +// repoNotResolvableSubstrings matches the wire-format substrings GitHub's +// GraphQL/REST APIs use for a deterministic "repository does not exist / +// not accessible" outcome: +// - "Could not resolve to a Repository" — GraphQL response for a missing +// or deleted repo (the dominant signal in the storm logs) +// - "Resource not accessible by integration" — GitHub Apps / fine-grained +// PAT scope mismatch. Treated as "dead end" here because retrying the +// same call with the same token will never succeed; a token swap +// evicts the cache via ClearRepoErrorCache. +// +// Raw "HTTP 404 / Not Found" is intentionally NOT matched here. A 404 on +// /repos/{owner}/{repo}/pulls/{N} can legitimately mean "the PR was +// deleted" while the repo itself is fine — a false-positive there would +// negative-cache the whole repo for 10 minutes off a single stale PR +// number. The GraphQL "Could not resolve to a Repository" string IS +// repo-scoped by construction and remains the canonical signal. +var repoNotResolvableSubstrings = []string{ + "could not resolve to a repository", + "resource not accessible by integration", +} + +// isRepoNotResolvableErr reports whether err carries a deterministic +// "repository missing / inaccessible" signal we should not keep retrying +// against in a tight loop. Errors flagged here are negative-cached for +// 10 minutes via Service.repoErrorCache; the cache is evicted on watch +// (re)creation and on token swap/clear so a freshly-linked repo or a +// re-auth is probed immediately. The match is case-insensitive so we +// don't depend on a stable error casing across gh CLI versions and +// GraphQL response variants. +// +// Bare HTTP 404 errors are NOT classified — too ambiguous (missing PR +// number vs missing repo) and a false positive poisons the whole repo +// for 10 minutes. Callers that know they're hitting a repo-root +// endpoint should wrap with ErrRepoNotResolvable explicitly. +func isRepoNotResolvableErr(err error) bool { + if err == nil { + return false + } + if errors.Is(err, ErrRepoNotResolvable) { + return true + } + s := strings.ToLower(err.Error()) + for _, sub := range repoNotResolvableSubstrings { + if strings.Contains(s, sub) { + return true + } + } + return false +} diff --git a/apps/backend/internal/github/errors_test.go b/apps/backend/internal/github/errors_test.go index 37fe2335f4..0b85981148 100644 --- a/apps/backend/internal/github/errors_test.go +++ b/apps/backend/internal/github/errors_test.go @@ -53,3 +53,46 @@ func TestErrorsAreClassifiable(t *testing.T) { } }) } + +// TestIsRepoNotResolvableErr pins the wire-format substrings that classify +// as "stop hammering this repo". A regression here re-opens the PR-watch +// storm that motivated the negative cache: with the classifier missing the +// canonical GraphQL "could not resolve to a repository" signal, +// SyncWatchesBatched keeps retrying the dead repo on every 5s frontend +// poll. Bare 404s are intentionally NOT classified — a 404 from +// /repos/{o}/{r}/pulls/{N} can legitimately mean "this PR is gone" while +// the repo is fine, and the negative cache must not poison the repo on +// that signal. +func TestIsRepoNotResolvableErr(t *testing.T) { + cases := []struct { + name string + err error + want bool + }{ + {"nil err", nil, false}, + {"unrelated error", errors.New("connection refused"), false}, + {"transient 500", errors.New("HTTP 500: Internal Server Error"), false}, + {"graphql repo missing", errors.New(`graphql error: Could not resolve to a Repository with the name 'NBCUDTC/bff'.`), true}, + {"graphql repo missing mixed case", errors.New("graphql error: could not resolve to a repository ..."), true}, + // Bare 404 errors are NOT repo-not-resolvable — too ambiguous. They + // may indicate a missing PR number on a live repo, and poisoning + // the repo for 10 minutes on that signal is the bug this guards + // against. See the docstring on isRepoNotResolvableErr. + {"bare gh REST 404 is ambiguous", errors.New("gh api: HTTP 404: Not Found"), false}, + {"bare 404 short form is ambiguous", errors.New("status: 404"), false}, + {"bare 404 with body suffix is ambiguous", errors.New("404 Not Found"), false}, + {"github api 404 is ambiguous", &GitHubAPIError{StatusCode: 404, Endpoint: "/repos/x/y/pulls/123", Body: "Not Found"}, false}, + {"wrapped github api 404 is ambiguous", fmt.Errorf("repo: %w", &GitHubAPIError{StatusCode: 404, Endpoint: "/repos/x/y/pulls/123"}), false}, + {"github api 403", &GitHubAPIError{StatusCode: 403, Endpoint: "/repos/x/y", Body: "Forbidden"}, false}, + {"resource not accessible", errors.New("Resource not accessible by integration"), true}, + {"resource not accessible lowercase", errors.New("resource not accessible by integration"), true}, + {"wrapped sentinel", fmt.Errorf("sync: %w", ErrRepoNotResolvable), true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := isRepoNotResolvableErr(tc.err); got != tc.want { + t.Errorf("isRepoNotResolvableErr(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} diff --git a/apps/backend/internal/github/gh_client.go b/apps/backend/internal/github/gh_client.go index dd1edfe16a..5c63e5437e 100644 --- a/apps/backend/internal/github/gh_client.go +++ b/apps/backend/internal/github/gh_client.go @@ -11,6 +11,8 @@ import ( "os/exec" "strings" "time" + + "github.com/kandev/kandev/internal/common/subproc" ) // ghSearchReposPath is the GitHub REST search-repositories path that gh CLI @@ -124,24 +126,16 @@ func (c *GHClient) IsAuthenticated(ctx context.Context) (bool, error) { // RunAuthDiagnostics executes gh auth status and captures the raw output for troubleshooting. func (c *GHClient) RunAuthDiagnostics(ctx context.Context) *AuthDiagnostics { - ctx, cancel := withDefaultGHTimeout(ctx) - defer cancel() - release, err := acquireGHSlot(ctx) - if err != nil { - return &AuthDiagnostics{ - Command: "gh auth status --hostname github.com", - Output: err.Error(), - ExitCode: -1, - } - } - defer release() - cmd := exec.CommandContext(ctx, "gh", "auth", "status", "--hostname", "github.com") var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr + runErr, execCtxErr := subproc.RunGHAfterAcquire(ctx, resolveGHExecTimeout(ctx), func(execCtx context.Context) *exec.Cmd { + cmd := exec.CommandContext(execCtx, "gh", "auth", "status", "--hostname", "github.com") + cmd.Stdout = &stdout + cmd.Stderr = &stderr + return cmd + }) exitCode := 0 - if err := cmd.Run(); err != nil { - if exitErr, ok := err.(*exec.ExitError); ok { + if runErr != nil { + if exitErr, ok := runErr.(*exec.ExitError); ok { exitCode = exitErr.ExitCode() } else { exitCode = -1 @@ -151,6 +145,17 @@ func (c *GHClient) RunAuthDiagnostics(ctx context.Context) *AuthDiagnostics { if output == "" { output = stdout.String() } + // When acquire/context fails before gh runs (or the exec is killed by + // the per-command deadline), both buffers can be empty. Surface the + // underlying error so the diagnostics panel doesn't show a blank + // "command failed with exit code -1" with no explanation. + if output == "" && runErr != nil { + if execCtxErr != nil { + output = fmt.Sprintf("gh auth status did not run: %v (context: %v)", runErr, execCtxErr) + } else { + output = fmt.Sprintf("gh auth status did not run: %v", runErr) + } + } return &AuthDiagnostics{ Command: "gh auth status --hostname github.com", Output: output, @@ -846,54 +851,67 @@ func (c *GHClient) DeleteGist(ctx context.Context, gistID string) error { const ghCLITimeout = 30 * time.Second -// withDefaultGHTimeout applies a 30s timeout if the context has no deadline. -func withDefaultGHTimeout(ctx context.Context) (context.Context, context.CancelFunc) { - if _, ok := ctx.Deadline(); ok { - return ctx, func() {} +// resolveGHExecTimeout returns the per-command exec budget used by +// run / runWithStdin. Honours the caller's ctx deadline when set (so a +// short-lived WS request stays bounded by the request deadline), but +// caps the budget at ghCLITimeout when the deadline is further out or +// absent. Used by the acquire-then-derive-timeout path so a queued +// waiter's exec timer starts AFTER it gets the throttle slot. +func resolveGHExecTimeout(ctx context.Context) time.Duration { + if dl, ok := ctx.Deadline(); ok { + remaining := time.Until(dl) + if remaining < ghCLITimeout { + return remaining + } } - return context.WithTimeout(ctx, ghCLITimeout) + return ghCLITimeout } // run executes a gh CLI command and returns its stdout output. // Stderr is captured separately to avoid contaminating JSON output. -// A default 30s timeout is applied if the context has no deadline. +// A 30s per-command exec timeout is applied AFTER the gh throttle slot +// is acquired (acquire-then-derive-timeout). Pre-fix the timer started +// before Acquire, so a queued waiter inherited a deadline that had +// already partly elapsed against throttle queue time — producing the +// `signal: killed` + `context deadline exceeded` cascade in the +// SyncWatchesBatched storm logs. func (c *GHClient) run(ctx context.Context, args ...string) (string, error) { - ctx, cancel := withDefaultGHTimeout(ctx) - defer cancel() - release, err := acquireGHSlot(ctx) - if err != nil { - return "", fmt.Errorf("gh %s: %w", firstArg(args), err) - } - defer release() - cmd := exec.CommandContext(ctx, "gh", args...) - var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr - if err := cmd.Run(); err != nil { - c.inspectRateStderr(args, stderr.String()) - return stdout.String(), fmt.Errorf("gh %s: %w: %s", args[0], err, stderr.String()) - } - return stdout.String(), nil + return c.runGH(ctx, nil, args...) } // runWithStdin is like run but pipes stdin into the gh CLI process. // Useful for `gh api --input -` calls where the body comes from JSON. func (c *GHClient) runWithStdin(ctx context.Context, stdin []byte, args ...string) (string, error) { - ctx, cancel := withDefaultGHTimeout(ctx) - defer cancel() - release, err := acquireGHSlot(ctx) - if err != nil { - return "", fmt.Errorf("gh %s: %w", firstArg(args), err) - } - defer release() - cmd := exec.CommandContext(ctx, "gh", args...) - cmd.Stdin = bytes.NewReader(stdin) + return c.runGH(ctx, stdin, args...) +} + +// runGH is the shared body of run / runWithStdin — both apply the same +// acquire-then-derive-timeout ordering, the same stderr/rate-limit +// inspection, and the same error wrap. The only delta is whether stdin +// is wired (`stdin != nil`). +func (c *GHClient) runGH(ctx context.Context, stdin []byte, args ...string) (string, error) { var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr - if err := cmd.Run(); err != nil { + execTimeout := resolveGHExecTimeout(ctx) + runErr, execCtxErr := subproc.RunGHAfterAcquire(ctx, execTimeout, func(execCtx context.Context) *exec.Cmd { + cmd := exec.CommandContext(execCtx, "gh", args...) + if stdin != nil { + cmd.Stdin = bytes.NewReader(stdin) + } + cmd.Stdout = &stdout + cmd.Stderr = &stderr + return cmd + }) + if runErr != nil { + // Surface execCtx timeouts AND cancellations as the canonical + // context error so callers (FetchRateLimit, IsAuthenticated, + // share.IsAlreadyGone) can still errors.Is(err, context.X). The + // runErr from cmd.Run on a killed child is `signal: killed`, + // which loses both classifier signals that pre-fix code relied on. + if execCtxErr != nil && (errors.Is(execCtxErr, context.DeadlineExceeded) || errors.Is(execCtxErr, context.Canceled)) { + return stdout.String(), fmt.Errorf("gh %s: %w", firstArg(args), execCtxErr) + } c.inspectRateStderr(args, stderr.String()) - return stdout.String(), fmt.Errorf("gh %s: %w: %s", args[0], err, stderr.String()) + return stdout.String(), fmt.Errorf("gh %s: %w: %s", firstArg(args), runErr, stderr.String()) } return stdout.String(), nil } diff --git a/apps/backend/internal/github/graphql.go b/apps/backend/internal/github/graphql.go index b012827234..2f05793e5e 100644 --- a/apps/backend/internal/github/graphql.go +++ b/apps/backend/internal/github/graphql.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -127,6 +128,51 @@ type graphQLError struct { Path []string `json:"path"` } +// repoRef is a minimal (owner, repo) pair used by the batched GraphQL +// helpers to surface which repositories the response flagged as +// unresolvable. Service.SyncWatchesBatched feeds these into the negative +// cache so subsequent polls short-circuit the dead-repo storm. +type repoRef struct { + Owner string + Repo string +} + +// batchedMissingReposErr is the typed error runBatchedPRQuery / +// runBatchedBranchQuery returns when the GraphQL response carried one or +// more "Could not resolve to a Repository" entries. Callers use +// errors.As to extract Repos and populate Service.repoErrorCache. When +// Inner is nil, the response contained ONLY resolution failures and +// the partial-decode result returned alongside the error is still safe +// to consume for the repos that did resolve; when Inner is non-nil, +// other (non-resolution) errors were also present and the caller should +// treat the whole batch as failed for those. +type batchedMissingReposErr struct { + Repos []repoRef + Inner error +} + +func (e *batchedMissingReposErr) Error() string { + if e.Inner != nil { + return e.Inner.Error() + } + parts := make([]string, len(e.Repos)) + for i, r := range e.Repos { + parts[i] = r.Owner + "/" + r.Repo + } + return "github: repositories not resolvable: " + strings.Join(parts, ", ") +} + +// Is lets callers errors.Is(err, ErrRepoNotResolvable) without having to +// type-assert. The sentinel is the canonical "stop hammering this repo" +// signal across the package. +func (e *batchedMissingReposErr) Is(target error) bool { + return target == ErrRepoNotResolvable +} + +// Unwrap exposes the composite non-resolution error (if any) so callers +// can errors.As / errors.Is past the batched-missing wrapper. +func (e *batchedMissingReposErr) Unwrap() error { return e.Inner } + // graphQLErrorsToErr returns a non-nil error when the GraphQL response carries // a top-level "errors" array. The first message is included so logs are // actionable; the count is appended when there are multiple. @@ -140,6 +186,82 @@ func graphQLErrorsToErr(errs []graphQLError) error { return fmt.Errorf("graphql error: %s (and %d more)", errs[0].Message, len(errs)-1) } +// classifyBatchedErrors splits the GraphQL errors[] array into "this +// repo is permanently unresolvable" and "everything else", mapping each +// resolution error's path-prefix alias back to the (owner, repo) it +// referred to via aliasToRepo. The caller (runBatchedPRQuery / +// runBatchedBranchQuery) lifts the resolution misses into the typed +// batchedMissingReposErr; the residual non-resolution errors are +// collapsed via graphQLErrorsToErr exactly as before so the storm-fix +// doesn't change behavior for unrelated GraphQL failures. +func classifyBatchedErrors(errs []graphQLError, aliasToRepo map[string]repoRef) (missing []repoRef, residual []graphQLError) { + seen := map[string]bool{} + for _, e := range errs { + if isRepoNotResolvableErr(errors.New(e.Message)) && len(e.Path) > 0 { + ref, ok := aliasToRepo[e.Path[0]] + if ok { + key := ref.Owner + "/" + ref.Repo + if !seen[key] { + seen[key] = true + missing = append(missing, ref) + } + continue + } + } + residual = append(residual, e) + } + return missing, residual +} + +// aliasMapForPRRefs returns alias -> repoRef for the +// owner/repo-sorted ordering buildBatchedPRQuery applies. A +// resolution-failure error's path[0] is always the outer repository +// alias ("repo0", "repo1", ...). +func aliasMapForPRRefs(refs []graphQLPRRef) map[string]repoRef { + byKey := map[string]repoRef{} + var keys []string + for _, r := range refs { + k := r.Owner + "/" + r.Repo + if _, ok := byKey[k]; !ok { + byKey[k] = repoRef{Owner: r.Owner, Repo: r.Repo} + keys = append(keys, k) + } + } + sort.Strings(keys) + out := make(map[string]repoRef, len(keys)) + for i, k := range keys { + out[fmt.Sprintf("repo%d", i)] = byKey[k] + } + return out +} + +// aliasMapForBranchRefs returns alias -> repoRef for the input-order +// b0, b1, ... aliases buildBatchedBranchQuery applies. +func aliasMapForBranchRefs(refs []graphQLBranchRef) map[string]repoRef { + out := make(map[string]repoRef, len(refs)) + for i, r := range refs { + out[fmt.Sprintf("b%d", i)] = repoRef{Owner: r.Owner, Repo: r.Repo} + } + return out +} + +// wrapBatchedErrors composes a classifyBatchedErrors result into the +// (result, error) shape runBatchedPRQuery / runBatchedBranchQuery +// already return. Behaviour: +// - no resolution misses, no residual errors → (nil): caller continues +// - no resolution misses, residual errors only → return composite err +// unchanged (pre-storm-fix semantics) +// - resolution misses present → return *batchedMissingReposErr; Inner +// is nil iff there were no residuals so callers can keep the +// partial decode, otherwise Inner carries the composite +func wrapBatchedErrors(missing []repoRef, residual []graphQLError) error { + residualErr := graphQLErrorsToErr(residual) + if len(missing) == 0 { + return residualErr + } + return &batchedMissingReposErr{Repos: missing, Inner: residualErr} +} + // graphQLRateLimit mirrors GitHub's top-level rateLimit field, the most // accurate source of GraphQL quota since limits are point-cost based. type graphQLRateLimit struct { @@ -411,6 +533,14 @@ func runBatchedPRQuery(ctx context.Context, exec GraphQLExecutor, refs []graphQL return nil, nil } result := make(map[string]*PRStatus, len(refs)) + // Accumulate missing / residual errors across all chunks so a chunk + // of purely-dead repos doesn't short-circuit the loop and drop the + // good data in later chunks. Pre-fix, a workspace with >50 watches + // where the first 50 happened to be dead would silently skip the + // remaining watches on every poll cycle until the first batch's + // repos landed in the negative cache. + var allMissing []repoRef + var allResidual []graphQLError for _, chunk := range chunkedRefs(refs) { query, vars := buildBatchedPRQuery(chunk) var resp struct { @@ -421,15 +551,27 @@ func runBatchedPRQuery(ctx context.Context, exec GraphQLExecutor, refs []graphQL return nil, err } // GitHub returns HTTP 200 with a top-level "errors" array for partial - // auth failures, schema mismatches, or per-alias errors. Surface them - // as an error so the caller falls back to per-watch checks instead of - // silently absorbing the failure. - if err := graphQLErrorsToErr(resp.Errors); err != nil { - return nil, err - } + // auth failures, schema mismatches, or per-alias errors. Split out + // "Could not resolve to a Repository" entries via classifyBatchedErrors + // so SyncWatchesBatched can negative-cache the dead repos and keep + // partial results for the ones that resolved; non-resolution errors + // still bubble up so the caller falls back to per-watch checks. + missing, residual := classifyBatchedErrors(resp.Errors, aliasMapForPRRefs(chunk)) if err := decodeBatchedPRChunk(chunk, resp.Data, result); err != nil { return nil, err } + allMissing = append(allMissing, missing...) + allResidual = append(allResidual, residual...) + } + if err := wrapBatchedErrors(allMissing, allResidual); err != nil { + // When the error is purely "missing repos" across all chunks, + // partial Statuses for the other refs are still in `result`; + // surface them to the caller alongside the typed error so it + // can populate the negative cache without dropping the good data. + if len(allMissing) > 0 && len(allResidual) == 0 { + return result, err + } + return nil, err } return result, nil } @@ -499,6 +641,10 @@ func runBatchedBranchQuery(ctx context.Context, exec GraphQLExecutor, refs []gra return nil, nil } result := make(map[string]*PRStatus, len(refs)) + // Same accumulation pattern as runBatchedPRQuery — see that function + // for the rationale (one dead-repo chunk must not drop later chunks). + var allMissing []repoRef + var allResidual []graphQLError for _, chunk := range chunkedRefs(refs) { query, vars := buildBatchedBranchQuery(chunk) var resp struct { @@ -508,12 +654,18 @@ func runBatchedBranchQuery(ctx context.Context, exec GraphQLExecutor, refs []gra if err := exec.ExecuteGraphQL(ctx, query, vars, &resp); err != nil { return nil, err } - if err := graphQLErrorsToErr(resp.Errors); err != nil { - return nil, err - } + missing, residual := classifyBatchedErrors(resp.Errors, aliasMapForBranchRefs(chunk)) if err := decodeBatchedBranchChunk(chunk, resp.Data, result); err != nil { return nil, err } + allMissing = append(allMissing, missing...) + allResidual = append(allResidual, residual...) + } + if err := wrapBatchedErrors(allMissing, allResidual); err != nil { + if len(allMissing) > 0 && len(allResidual) == 0 { + return result, err + } + return nil, err } return result, nil } diff --git a/apps/backend/internal/github/handlers.go b/apps/backend/internal/github/handlers.go index d233d449d5..5572b03b30 100644 --- a/apps/backend/internal/github/handlers.go +++ b/apps/backend/internal/github/handlers.go @@ -250,16 +250,27 @@ func wsGetTaskPR(svc *Service, _ *logger.Logger) func(ctx context.Context, msg * // PR yet"); multi-repo callers iterate and call setTaskPR for each so the // per-repo PR icon stays in sync. The legacy single-PR shape would have // silently dropped every repo's PR except the most-recently-updated one. +// +// permanent=true on the response signals the frontend's 5s retry loop to +// stop. It's set when every watch on the task points at a repository +// classified as unresolvable (missing, deleted, or inaccessible to the +// authenticated principal) — without this, the frontend keeps re-polling +// dead repos every 5s for the lifetime of the task, which was the +// dominant signal in the production SyncWatchesBatched storm. func wsSyncTaskPR(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { return wsWithField("task_id", func(ctx context.Context, taskID string) (interface{}, error) { - prs, err := svc.TriggerPRSyncAll(ctx, taskID) + prs, permanent, err := svc.TriggerPRSyncAllPermanent(ctx, taskID) if err != nil { return nil, err } // Return an envelope so the frontend always gets a deterministic - // shape even on empty results (`{prs: []}`); a bare `nil` would - // confuse the WS handler's success/error branching. - return map[string]interface{}{"prs": prs}, nil + // shape even on empty results (`{prs: []}`); a bare `nil` slice + // would serialize to `prs: null` and confuse the WS handler's + // success/error branching, so normalize to an empty slice here. + if prs == nil { + prs = []*TaskPR{} + } + return map[string]interface{}{"prs": prs, "permanent": permanent}, nil }) } diff --git a/apps/backend/internal/github/service.go b/apps/backend/internal/github/service.go index 8a3662a62e..e3e6c297d7 100644 --- a/apps/backend/internal/github/service.go +++ b/apps/backend/internal/github/service.go @@ -84,8 +84,10 @@ type Service struct { taskEventSubs []bus.Subscription searchCache *ttlCache prStatusCache *ttlCache + prFeedbackCache *ttlCache mergeMethodsCache *ttlCache accessibleReposCache *ttlCache + repoErrorCache *ttlCache protectionCache *branchProtectionCache rateTracker *RateTracker @@ -129,8 +131,10 @@ func NewService(client Client, authMethod string, secrets SecretProvider, store logger: log, searchCache: newTTLCache(), prStatusCache: newTTLCache(), + prFeedbackCache: newPRFeedbackCache(), mergeMethodsCache: newMergeMethodsCache(), accessibleReposCache: newAccessibleReposCache(), + repoErrorCache: newRepoErrorCache(), protectionCache: newBranchProtectionCache(), rateTracker: NewRateTracker(eventBus, log), cleanupFailureCounts: make(map[string]int), @@ -215,3 +219,94 @@ func (s *Service) IsAuthenticated() bool { func (s *Service) AuthMethod() string { return s.authMethod } + +// isRepoCachedAsMissing reports whether the (owner, repo) tuple is in the +// 10-min negative cache. Called from SyncWatchesBatched and per-watch +// probes before acquiring the gh throttle, so the storm against +// missing/unauthorized repos short-circuits without burning a slot. +func (s *Service) isRepoCachedAsMissing(owner, repo string) bool { + if s == nil || s.repoErrorCache == nil { + return false + } + v, ok := s.repoErrorCache.get(repoErrorCacheKey(owner, repo)) + if !ok { + return false + } + _, isErr := v.(cachedErr) + return isErr +} + +// repoErrorGenSnapshot returns the current cache generation. Callers +// snapshot this BEFORE issuing the fetch that may classify a repo as +// missing, and pass the snapshot to markRepoAsMissing — that way an +// evictRepoNegative / ClearRepoErrorCache that fires DURING the fetch +// (bumping gen) causes the post-fetch write to be dropped, and the +// negative cache reflects the user's latest intent rather than the +// stale classifier result. +func (s *Service) repoErrorGenSnapshot() uint64 { + if s == nil || s.repoErrorCache == nil { + return 0 + } + return s.repoErrorCache.generation() +} + +// markRepoAsMissing stores ErrRepoNotResolvable for (owner, repo) in the +// negative cache. Called after a per-watch or batched probe deterministically +// classifies the repo as missing/unauthorized (see isRepoNotResolvableErr). +// `gen` must be a snapshot taken via repoErrorGenSnapshot BEFORE the fetch +// — passing the current generation at write time would defeat the guard: +// a concurrent del() / clear() that bumped gen between the fetch start and +// the write would still be present at write time, so a write-time snapshot +// would land in the new generation and the just-evicted entry would be +// reinserted. The fetch-start snapshot turns the write into a no-op when +// any eviction has fired since the fetch began. +func (s *Service) markRepoAsMissing(owner, repo string, gen uint64) { + if s == nil || s.repoErrorCache == nil { + return + } + s.repoErrorCache.setIfCurrentGeneration( + repoErrorCacheKey(owner, repo), cachedErr{err: ErrRepoNotResolvable}, gen) +} + +// evictRepoNegative drops any negative-cache entry for (owner, repo). Called +// from the PR watch (re)create paths and AssociatePRByURL so a freshly +// linked or re-linked repository is probed immediately on the next sync, +// without waiting out the 10-min TTL. +func (s *Service) evictRepoNegative(owner, repo string) { + if s == nil || s.repoErrorCache == nil { + return + } + s.repoErrorCache.del(repoErrorCacheKey(owner, repo)) +} + +// ClearRepoErrorCache drops every entry from the negative repo-resolution +// cache. Called from ConfigureToken / ClearToken so a token swap or clear +// invalidates classifications that were made under the prior identity — +// without this, a repo that the old token couldn't see would stay +// "permanent: true" for up to 10 minutes after the user fixes auth, and +// the frontend retry loop would stop early on stale data. Nil-guards the +// cache because tests construct Service literals without going through +// NewService. +func (s *Service) ClearRepoErrorCache() { + if s == nil || s.repoErrorCache == nil { + return + } + s.repoErrorCache.clear() +} + +// ClearPRCaches drops the PR status and PR feedback caches. Called from +// the token-swap path because review visibility differs across identities +// (a PR readable to the old token may not be readable to the new one, and +// vice versa); without this, the new identity could see stale review / +// check / comment data for up to one TTL after the swap. +func (s *Service) ClearPRCaches() { + if s == nil { + return + } + if s.prStatusCache != nil { + s.prStatusCache.clear() + } + if s.prFeedbackCache != nil { + s.prFeedbackCache.clear() + } +} diff --git a/apps/backend/internal/github/service_pr.go b/apps/backend/internal/github/service_pr.go index 4c81195bdb..3ef40e02d0 100644 --- a/apps/backend/internal/github/service_pr.go +++ b/apps/backend/internal/github/service_pr.go @@ -107,12 +107,34 @@ func (s *Service) GetPR(ctx context.Context, owner, repo string, number int) (*P return s.client.GetPR(ctx, owner, repo, number) } -// GetPRFeedback fetches live PR feedback from GitHub. +// GetPRFeedback fetches live PR feedback from GitHub. Cached briefly with +// singleflight coalescing: the task page fires this for the same PR many times +// in quick succession (chat-bar CI chip, detail panel, hover popover — each on +// its own render/mount triggers), and each miss is 4 sequential GitHub REST +// calls. Without this, a burst fanned out into dozens of slow, rate-limited +// duplicate requests. The returned pointer is shared — callers must not mutate it. func (s *Service) GetPRFeedback(ctx context.Context, owner, repo string, number int) (*PRFeedback, error) { if s.client == nil { return nil, fmt.Errorf("github client not available") } - return s.client.GetPRFeedback(ctx, owner, repo, number) + // Detach the upstream call's context from the singleflight leader: a + // cancelled leader (user closes the tab) would otherwise return its + // context error to all co-waiters, cascading a single client disconnect + // into a burst of transient failures across concurrent callers. + // context.WithoutCancel strips the deadline too, so re-attach the + // parent deadline explicitly — without this, a still-active caller's + // timeout wouldn't bound the shared fetch and quota could keep being + // consumed past the deadline. + fetchCtx, cancelFetch := derivedFetchContext(ctx) + defer cancelFetch() + key := prStatusCacheKey(owner, repo, number) + v, err := s.prFeedbackCache.doOrFetch(key, func() (any, error) { + return s.client.GetPRFeedback(fetchCtx, owner, repo, number) + }) + if err != nil { + return nil, err + } + return v.(*PRFeedback), nil } // GetPRStatus fetches lightweight PR status (review + checks + mergeable). @@ -123,9 +145,13 @@ func (s *Service) GetPRStatus(ctx context.Context, owner, repo string, number in if s.client == nil { return nil, fmt.Errorf("github client not available") } + // Detach the upstream call from the singleflight leader's context; see + // GetPRFeedback for the cascading-cancel + deadline-preserve rationale. + fetchCtx, cancelFetch := derivedFetchContext(ctx) + defer cancelFetch() key := prStatusCacheKey(owner, repo, number) v, err := s.prStatusCache.doOrFetch(key, func() (any, error) { - return s.client.GetPRStatus(ctx, owner, repo, number) + return s.client.GetPRStatus(fetchCtx, owner, repo, number) }) if err != nil { return nil, err @@ -133,6 +159,22 @@ func (s *Service) GetPRStatus(ctx context.Context, owner, repo string, number in return v.(*PRStatus), nil } +// derivedFetchContext builds the upstream-fetch context used by the +// singleflight-cached GetPRFeedback / GetPRStatus paths. It strips the +// caller's cancellation via context.WithoutCancel so a single client +// disconnect doesn't cascade the same error to all co-waiters, but +// preserves the caller's deadline (when present) so the fetch can't +// outlive the request budget and keep consuming GitHub quota past it. +// Returns the derived context and a cancel func the caller must defer +// to release resources promptly. +func derivedFetchContext(ctx context.Context) (context.Context, context.CancelFunc) { + fetchCtx := context.WithoutCancel(ctx) + if deadline, ok := ctx.Deadline(); ok { + return context.WithDeadline(fetchCtx, deadline) + } + return fetchCtx, func() {} +} + // PRRef identifies a pull request by owner/repo/number. type PRRef struct { Owner string `json:"owner"` diff --git a/apps/backend/internal/github/service_pr_feedback_test.go b/apps/backend/internal/github/service_pr_feedback_test.go new file mode 100644 index 0000000000..142e199321 --- /dev/null +++ b/apps/backend/internal/github/service_pr_feedback_test.go @@ -0,0 +1,107 @@ +package github + +import ( + "context" + "sync" + "sync/atomic" + "testing" +) + +// countingFeedbackClient counts top-level GetPRFeedback calls so the cache / +// singleflight behavior of Service.GetPRFeedback can be asserted without +// reaching real GitHub. +type countingFeedbackClient struct { + *MockClient + calls atomic.Int32 +} + +func (c *countingFeedbackClient) GetPRFeedback(_ context.Context, owner, repo string, number int) (*PRFeedback, error) { + c.calls.Add(1) + return &PRFeedback{PR: &PR{RepoOwner: owner, RepoName: repo, Number: number}}, nil +} + +// Repeated feedback fetches for the same PR within the TTL window must collapse +// to a single upstream call. Before the cache was added, every call hit GitHub +// (4 sequential REST calls each), so a render/mount burst on the task page +// fanned out into dozens of slow, rate-limited duplicate requests. +func TestService_GetPRFeedback_UsesCache(t *testing.T) { + client := &countingFeedbackClient{MockClient: NewMockClient()} + svc := newTestService(client) + ctx := context.Background() + + for i := 0; i < 3; i++ { + if _, err := svc.GetPRFeedback(ctx, "o", "r", 1); err != nil { + t.Fatalf("call %d: %v", i, err) + } + } + if got := client.calls.Load(); got != 1 { + t.Fatalf("expected 1 upstream call, got %d", got) + } + + if _, err := svc.GetPRFeedback(ctx, "o", "r", 2); err != nil { + t.Fatal(err) + } + if got := client.calls.Load(); got != 2 { + t.Fatalf("expected new fetch for different number, got %d", got) + } +} + +// blockingFeedbackClient blocks inside GetPRFeedback until released, so the +// test can force many concurrent in-flight fetches for the same PR and assert +// they coalesce into one upstream call (singleflight). `started` is signaled +// the moment a goroutine enters the upstream call so the test can be certain +// at least one fetch is in flight before releasing siblings. +type blockingFeedbackClient struct { + *MockClient + calls atomic.Int32 + release chan struct{} + started chan struct{} +} + +func (c *blockingFeedbackClient) GetPRFeedback(_ context.Context, owner, repo string, number int) (*PRFeedback, error) { + c.calls.Add(1) + // Non-blocking signal so only the FIRST entrant fires `started`; later + // arrivals queue behind the singleflight without re-signaling. + select { + case c.started <- struct{}{}: + default: + } + <-c.release + return &PRFeedback{PR: &PR{RepoOwner: owner, RepoName: repo, Number: number}}, nil +} + +// Concurrent feedback fetches for the same PR must coalesce to a single +// upstream call. This reproduces the production pile-up: the task page fired ~8 +// identical GetPRFeedback requests concurrently, and without singleflight each +// independently hammered GitHub, ballooning durations to 40-73s. +func TestService_GetPRFeedback_CoalescesConcurrentCalls(t *testing.T) { + client := &blockingFeedbackClient{ + MockClient: NewMockClient(), + release: make(chan struct{}), + started: make(chan struct{}, 1), + } + svc := newTestService(client) + + const n = 16 + var done sync.WaitGroup + done.Add(n) + for i := 0; i < n; i++ { + go func() { + defer done.Done() + if _, err := svc.GetPRFeedback(context.Background(), "o", "r", 1); err != nil { + t.Errorf("concurrent fetch: %v", err) + } + }() + } + // Wait for the leader to actually enter the upstream call (not just be + // scheduled). Pre-fix this used launched.Wait, which only guaranteed the + // goroutines had STARTED — late arrivals after release could have seen + // a warm cache and skipped the singleflight, masking regressions. + <-client.started + close(client.release) + done.Wait() + + if got := client.calls.Load(); got != 1 { + t.Fatalf("expected concurrent calls to coalesce to 1 upstream call, got %d", got) + } +} diff --git a/apps/backend/internal/github/service_pr_watch.go b/apps/backend/internal/github/service_pr_watch.go index b6c4e4e6c1..d581f3f672 100644 --- a/apps/backend/internal/github/service_pr_watch.go +++ b/apps/backend/internal/github/service_pr_watch.go @@ -22,6 +22,12 @@ import ( // its own watch row. Multi-branch tasks store one watch per branch so a // secondary branch's push isn't lost behind the primary's existing watch. func (s *Service) CreatePRWatch(ctx context.Context, sessionID, taskID, repositoryID, owner, repo string, prNumber int, branch string) (*PRWatch, error) { + // Evict any negative-cache entry up front. Caller intent on either + // branch (creating a new watch or re-finding an existing one) is "I + // want this repo watched", which means a stale "missing" verdict + // from a prior auth failure should be re-probed immediately rather + // than held for the rest of the 10-min TTL. + s.evictRepoNegative(owner, repo) existing, err := s.store.GetPRWatchBySessionRepoAndBranch(ctx, sessionID, repositoryID, branch) if err != nil { return nil, err @@ -134,6 +140,10 @@ func (s *Service) CheckPRWatch(ctx context.Context, watch *PRWatch) (*PRStatus, // so each branch gets its own watch — keying on (session, repo) alone // drops secondary branches' watches behind the primary's existing row. func (s *Service) EnsurePRWatch(ctx context.Context, sessionID, taskID, repositoryID, owner, repo, branch string) (*PRWatch, error) { + // Same up-front eviction as CreatePRWatch — see that function for the + // rationale. The existing-watch early-return path must not inherit a + // stale "missing" verdict from a prior incarnation. + s.evictRepoNegative(owner, repo) existing, err := s.store.GetPRWatchBySessionRepoAndBranch(ctx, sessionID, repositoryID, branch) if err != nil { return nil, err @@ -619,20 +629,46 @@ func (s *Service) TriggerPRSync(ctx context.Context, taskID string) (*TaskPR, er // when the batched fetch itself fails; in those cases we fan out one // subprocess per watch as before. func (s *Service) TriggerPRSyncAll(ctx context.Context, taskID string) ([]*TaskPR, error) { + prs, _, err := s.triggerPRSyncAllPermanent(ctx, taskID) + return prs, err +} + +// TriggerPRSyncAllPermanent extends TriggerPRSyncAll with a "permanent" +// flag: true iff the task has at least one PR watch and every one of +// those watches is currently in the 10-min negative cache (or its repo +// is otherwise classified as unresolvable). The WS handler uses this to +// tell the frontend to stop the 5s sync-retry interval, so a task +// pointing at a deleted repo doesn't keep hammering the gh throttle for +// the lifetime of the task. +func (s *Service) TriggerPRSyncAllPermanent(ctx context.Context, taskID string) ([]*TaskPR, bool, error) { + return s.triggerPRSyncAllPermanent(ctx, taskID) +} + +func (s *Service) triggerPRSyncAllPermanent(ctx context.Context, taskID string) ([]*TaskPR, bool, error) { watches, err := s.store.ListPRWatchesByTask(ctx, taskID) if err != nil { - return nil, fmt.Errorf("list PR watches: %w", err) + return nil, false, fmt.Errorf("list PR watches: %w", err) } if len(watches) == 0 { // No watches — fall back to whatever TaskPRs already exist (e.g. // PRs imported via task-create-from-PR-URL where the watch is - // optional). Empty slice if none. + // optional). Empty slice if none. permanent=false (the task can + // still acquire a watch later via push detection). existing, listErr := s.store.ListTaskPRsByTask(ctx, taskID) if listErr != nil { - return nil, fmt.Errorf("list task PRs: %w", listErr) + return nil, false, fmt.Errorf("list task PRs: %w", listErr) } - return existing, nil + return existing, false, nil } + prs, syncErr := s.runBatchedOrPerWatchSync(ctx, taskID, watches) + return prs, s.areAllWatchesPermanentlyMissing(watches), syncErr +} + +// runBatchedOrPerWatchSync is the shared "try batched, fall back to +// per-watch" body of triggerPRSyncAllPermanent. Split out so the +// permanent-flag computation can wrap the result without duplicating +// the batched / fallback branches inline. +func (s *Service) runBatchedOrPerWatchSync(ctx context.Context, taskID string, watches []*PRWatch) ([]*TaskPR, error) { if _, batchErr := s.SyncWatchesBatched(ctx, watches); batchErr != nil { s.logger.Debug("batched PR sync failed; falling back to per-watch", zap.String("task_id", taskID), zap.Error(batchErr)) @@ -643,6 +679,24 @@ func (s *Service) TriggerPRSyncAll(ctx context.Context, taskID string) ([]*TaskP return s.store.ListTaskPRsByTask(ctx, taskID) } +// areAllWatchesPermanentlyMissing reports whether every supplied watch's +// (owner, repo) is currently in the 10-min negative cache. Used to set +// the permanent flag on wsSyncTaskPR's response so the frontend stops +// retrying when a task points only at deleted/inaccessible repositories. +// Returns false when the input is empty so an empty task doesn't get +// classified as permanently missing. +func (s *Service) areAllWatchesPermanentlyMissing(watches []*PRWatch) bool { + if len(watches) == 0 { + return false + } + for _, w := range watches { + if !s.isRepoCachedAsMissing(w.Owner, w.Repo) { + return false + } + } + return true +} + // triggerPRSyncAllPerWatch is the legacy fan-out path: one gh subprocess // per watch. Kept as a fallback for the NoopClient and for the rare case // where the batched GraphQL call fails (auth glitch, network blip) so a @@ -706,6 +760,13 @@ func (s *Service) triggerPRDetection(ctx context.Context, watch *PRWatch, taskID // searching (pr_number=0) watch. It runs inside triggerPRDetection's // singleflight so only one probe per watch is in flight at a time. func (s *Service) detectPRForWatchOnce(ctx context.Context, watch *PRWatch, taskID string) (*TaskPR, error) { + // Short-circuit when this repo is already in the 10-min negative + // cache. Without this, an unresolvable repo gets a fresh per-watch + // probe every 5s (the frontend retry cadence) until the freshness + // timestamp below stamps in — multiplied by every watch the task has. + if s.isRepoCachedAsMissing(watch.Owner, watch.Repo) { + return nil, ErrRepoNotResolvable + } // Throttle branch-detection probes. A watch still searching for its PR // (pr_number=0) has no TaskPR row yet, so triggerPRStatusSync's freshness // check can't gate it — we gate on the watch's own last_checked_at instead. @@ -716,7 +777,19 @@ func (s *Service) detectPRForWatchOnce(ctx context.Context, watch *PRWatch, task return s.store.GetTaskPRByRepository(ctx, taskID, watch.RepositoryID) } + // Snapshot the negative-cache generation BEFORE the fetch so an + // eviction (CreatePRWatch / EnsurePRWatch / explicit clear) that + // fires WHILE FindPRByBranch is running wins over our post-fetch + // classification — see Service.markRepoAsMissing for the rationale. + repoErrGen := s.repoErrorGenSnapshot() pr, err := s.client.FindPRByBranch(ctx, watch.Owner, watch.Repo, watch.Branch) + if err != nil && isRepoNotResolvableErr(err) { + s.markRepoAsMissing(watch.Owner, watch.Repo, repoErrGen) + // Wrap so wsSyncTaskPR can errors.Is(err, ErrRepoNotResolvable) + // to flag the WS response as permanent without re-running the + // classifier on the raw upstream error string. + err = fmt.Errorf("%w: %w", ErrRepoNotResolvable, err) + } // Stamp last_checked_at regardless of outcome — including on error. The // flood this fixes IS the error path (an unresolvable repo makes `gh` exit // non-zero), so stamping only on success would leave the bug unfixed. The @@ -775,13 +848,27 @@ func (s *Service) triggerPRStatusSync(ctx context.Context, watch *PRWatch, taskI } } + // Short-circuit on negative cache so a 5s frontend retry against a + // dead repo doesn't burn the gh throttle. Eviction happens on watch + // (re)create paths, so a freshly linked repo is probed immediately. + if s.isRepoCachedAsMissing(watch.Owner, watch.Repo) { + return nil, ErrRepoNotResolvable + } + // Coalesce concurrent syncs for the same PR key := fmt.Sprintf("%s/%s/%d", watch.Owner, watch.Repo, watch.PRNumber) v, err, _ := s.syncGroup.Do(key, func() (interface{}, error) { bgCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() + // Snapshot cache generation BEFORE the per-watch probe so a + // concurrent eviction wins; see Service.markRepoAsMissing. + repoErrGen := s.repoErrorGenSnapshot() status, _, checkErr := s.CheckPRWatch(bgCtx, watch) if checkErr != nil { + if isRepoNotResolvableErr(checkErr) { + s.markRepoAsMissing(watch.Owner, watch.Repo, repoErrGen) + return nil, fmt.Errorf("%w: %w", ErrRepoNotResolvable, checkErr) + } return nil, checkErr } if status == nil { diff --git a/apps/backend/internal/github/service_pr_watch_batched.go b/apps/backend/internal/github/service_pr_watch_batched.go index d3b099134f..2c3e975c27 100644 --- a/apps/backend/internal/github/service_pr_watch_batched.go +++ b/apps/backend/internal/github/service_pr_watch_batched.go @@ -2,7 +2,10 @@ package github import ( "context" + "errors" "fmt" + "sort" + "strings" "time" "go.uber.org/zap" @@ -59,38 +62,160 @@ func (s *Service) SyncWatchesBatched(ctx context.Context, watches []*PRWatch) ([ // fetchBatchedWatchStatuses runs the numbered- and branch-keyed GraphQL // queries and merges their results. Returns an error when either query -// fails so the caller can fall back to per-watch checks. +// fails so the caller can fall back to per-watch checks. Watches whose +// (owner, repo) is already cached as missing are filtered out before +// hitting GraphQL, and any newly-discovered missing repos in the +// response are fed into the negative cache so subsequent polls +// short-circuit before acquiring the gh throttle. +// +// The full fetch is gated by a service-level singleflight keyed by the +// sorted ref set. Without it, a burst of concurrent SyncWatchesBatched +// calls (poller racing the WS sync racing the workspace background +// refresh) all hit GraphQL during the window before the first caller +// seeds the negative cache — exactly the storm this change is trying to +// calm. The shared-result map is read-only at the call sites (only +// map lookups in applyBatched*), so it's safe to share. +// +// The upstream fetch detaches from the leader's cancellation via +// derivedFetchContext so one caller (WS) disconnecting mid-flight +// doesn't cascade context.Canceled to all co-waiters — same pattern as +// GetPRFeedback / GetPRStatus. The leader's deadline is preserved so +// the fetch can't outlive the request budget. func (s *Service) fetchBatchedWatchStatuses( ctx context.Context, exec GraphQLExecutor, numbered, searching []*PRWatch, ) (map[string]*PRStatus, error) { - combined := make(map[string]*PRStatus, len(numbered)+len(searching)) - if len(numbered) > 0 { - refs := make([]graphQLPRRef, 0, len(numbered)) - for _, w := range numbered { - refs = append(refs, graphQLPRRef{Owner: w.Owner, Repo: w.Repo, Number: w.PRNumber}) + key := batchedFetchSingleflightKey(numbered, searching) + fetchCtx, cancelFetch := derivedFetchContext(ctx) + defer cancelFetch() + v, err, _ := s.syncGroup.Do(key, func() (interface{}, error) { + // Snapshot the negative-cache generation BEFORE the batched + // fetch so an eviction (relink / clear) firing while the + // GraphQL call is in flight wins over the post-fetch + // markRepoAsMissing writes — see Service.markRepoAsMissing. + repoErrGen := s.repoErrorGenSnapshot() + combined := make(map[string]*PRStatus, len(numbered)+len(searching)) + if err := s.fetchBatchedPRStatuses(fetchCtx, exec, numbered, combined, repoErrGen); err != nil { + return nil, err } - out, err := runBatchedPRQuery(ctx, exec, refs) - if err != nil { - return nil, fmt.Errorf("batched PR query: %w", err) - } - for k, v := range out { - combined[k] = v + if err := s.fetchBatchedBranchStatuses(fetchCtx, exec, searching, combined, repoErrGen); err != nil { + return nil, err } + return combined, nil + }) + if err != nil { + return nil, err } - if len(searching) > 0 { - refs := make([]graphQLBranchRef, 0, len(searching)) - for _, w := range searching { - refs = append(refs, graphQLBranchRef{Owner: w.Owner, Repo: w.Repo, Branch: w.Branch}) - } - out, err := runBatchedBranchQuery(ctx, exec, refs) - if err != nil { - return nil, fmt.Errorf("batched branch query: %w", err) - } - for k, v := range out { - combined[k] = v - } + if v == nil { + return nil, nil } - return combined, nil + return v.(map[string]*PRStatus), nil +} + +// batchedFetchSingleflightKey builds a deterministic key from the sorted +// ref tuples in (numbered, searching). Sorting ensures ordering doesn't +// split equivalent calls into separate slots, and lowercasing matches +// repoErrorCacheKey so the in-flight key is stable regardless of caller +// casing (watches from the DB always have consistent casing in practice, +// but it costs nothing to make the key insensitive). Distinct "n:"/"b:" +// prefixes prevent a numbered ref colliding with a branch ref that +// happens to render the same string. +func batchedFetchSingleflightKey(numbered, searching []*PRWatch) string { + parts := make([]string, 0, len(numbered)+len(searching)) + for _, w := range numbered { + parts = append(parts, fmt.Sprintf("n:%s/%s#%d", + strings.ToLower(w.Owner), strings.ToLower(w.Repo), w.PRNumber)) + } + for _, w := range searching { + parts = append(parts, fmt.Sprintf("b:%s/%s@%s", + strings.ToLower(w.Owner), strings.ToLower(w.Repo), w.Branch)) + } + sort.Strings(parts) + return "batched-fetch:" + strings.Join(parts, "|") +} + +// fetchBatchedPRStatuses runs the numbered-watch (PR-number-keyed) batch. +// Watches whose repo is in the negative cache are dropped before building +// the GraphQL refs so the dead-repo storm doesn't burn gh throttle slots. +// Missing repos surfaced by the response are negative-cached for the next +// 10 minutes; partial results for the repos that did resolve are merged +// into `combined`. +func (s *Service) fetchBatchedPRStatuses( + ctx context.Context, exec GraphQLExecutor, numbered []*PRWatch, combined map[string]*PRStatus, + repoErrGen uint64, +) error { + refs := make([]graphQLPRRef, 0, len(numbered)) + for _, w := range numbered { + if s.isRepoCachedAsMissing(w.Owner, w.Repo) { + continue + } + refs = append(refs, graphQLPRRef{Owner: w.Owner, Repo: w.Repo, Number: w.PRNumber}) + } + if len(refs) == 0 { + return nil + } + out, err := runBatchedPRQuery(ctx, exec, refs) + out, err = s.absorbMissingReposErr(out, err, repoErrGen) + if err != nil { + return fmt.Errorf("batched PR query: %w", err) + } + for k, v := range out { + combined[k] = v + } + return nil +} + +// fetchBatchedBranchStatuses runs the branch-keyed batch with the same +// negative-cache filter and missing-repo absorption as the numbered path. +func (s *Service) fetchBatchedBranchStatuses( + ctx context.Context, exec GraphQLExecutor, searching []*PRWatch, combined map[string]*PRStatus, + repoErrGen uint64, +) error { + refs := make([]graphQLBranchRef, 0, len(searching)) + for _, w := range searching { + if s.isRepoCachedAsMissing(w.Owner, w.Repo) { + continue + } + refs = append(refs, graphQLBranchRef{Owner: w.Owner, Repo: w.Repo, Branch: w.Branch}) + } + if len(refs) == 0 { + return nil + } + out, err := runBatchedBranchQuery(ctx, exec, refs) + out, err = s.absorbMissingReposErr(out, err, repoErrGen) + if err != nil { + return fmt.Errorf("batched branch query: %w", err) + } + for k, v := range out { + combined[k] = v + } + return nil +} + +// absorbMissingReposErr extracts the negative-cacheable repos from a +// runBatched* result and seeds the cache, then returns (partial, +// remaining-err). When the error was PURELY missing-repo errors, the +// partial decode is preserved and a nil error is returned so the caller +// can use the resolved repos' data. When other errors were also +// present, the inner error is returned unchanged so the caller falls +// back to per-watch checks. `repoErrGen` is the negative-cache generation +// snapshot taken BEFORE the fetch so a concurrent eviction wins. +func (s *Service) absorbMissingReposErr(out map[string]*PRStatus, err error, repoErrGen uint64) (map[string]*PRStatus, error) { + if err == nil { + return out, nil + } + var missingErr *batchedMissingReposErr + if !errors.As(err, &missingErr) { + return nil, err + } + for _, r := range missingErr.Repos { + s.markRepoAsMissing(r.Owner, r.Repo, repoErrGen) + s.logger.Debug("repo not resolvable; negative-cached", + zap.String("owner", r.Owner), zap.String("repo", r.Repo)) + } + if missingErr.Inner != nil { + return nil, missingErr.Inner + } + return out, nil } // applyBatchedNumberedWatch mirrors Poller.applyPRStatus on the service @@ -98,6 +223,14 @@ func (s *Service) fetchBatchedWatchStatuses( func (s *Service) applyBatchedNumberedWatch( ctx context.Context, w *PRWatch, statusByKey map[string]*PRStatus, now time.Time, ) PRWatchSyncResult { + // Repo is in the 10-min negative cache — the fetch path didn't include + // it in the GraphQL batch, so the apply path can't probe upstream either. + // Bump last_checked_at to suppress retry storms and surface SyncFailed + // so the WS handler can flag the result as permanent for the frontend. + if s.isRepoCachedAsMissing(w.Owner, w.Repo) { + _ = s.store.UpdatePRWatchTimestamps(ctx, w.ID, now, nil, "", "") + return PRWatchSyncResult{Watch: w, SyncFailed: true} + } status, ok := statusByKey[prStatusCacheKey(w.Owner, w.Repo, w.PRNumber)] if !ok || status == nil { // Alias missing — best-effort liveness bump so we don't immediately re-probe. @@ -135,6 +268,9 @@ func (s *Service) applyBatchedSearchingWatch( // paths, so hoist the single call above the branch to make that // invariant obvious. _ = s.store.UpdatePRWatchTimestamps(ctx, w.ID, now, nil, "", "") + if s.isRepoCachedAsMissing(w.Owner, w.Repo) { + return PRWatchSyncResult{Watch: w, SyncFailed: true} + } status, ok := statusByKey[graphqlBranchKey(w.Owner, w.Repo, w.Branch)] if !ok || status == nil || status.PR == nil { return PRWatchSyncResult{Watch: w} diff --git a/apps/backend/internal/github/service_repo_negative_cache_test.go b/apps/backend/internal/github/service_repo_negative_cache_test.go new file mode 100644 index 0000000000..ca5146fc15 --- /dev/null +++ b/apps/backend/internal/github/service_repo_negative_cache_test.go @@ -0,0 +1,322 @@ +package github + +import ( + "context" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" +) + +// TestSyncWatchesBatched_NegativeCache_ShortCircuits is the regression +// test for the SyncWatchesBatched storm: once a repo is classified as +// unresolvable, subsequent syncs for watches on that repo must NOT issue +// a fresh GraphQL call. Before the negative cache, every 5s frontend +// retry against a dead repo (NBCUDTC/bff was the production smoking +// gun) burned a gh throttle slot and produced a fresh "Could not resolve +// to a Repository" log line. +func TestSyncWatchesBatched_NegativeCache_ShortCircuits(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + + w := &PRWatch{SessionID: "s1", TaskID: "t1", Owner: "Dead", Repo: "Repo", PRNumber: 7, Branch: "br"} + if err := store.CreatePRWatch(ctx, w); err != nil { + t.Fatalf("create watch: %v", err) + } + // CreatePRWatch evicts the negative cache, so prime it AFTER the + // watch is in place. + svc.markRepoAsMissing(w.Owner, w.Repo, svc.repoErrorGenSnapshot()) + + // No canned responses — if anything reaches gh the test fails fast + // with "no canned PR response" / "no canned branch response". + results, err := svc.SyncWatchesBatched(ctx, []*PRWatch{w}) + if err != nil { + t.Fatalf("SyncWatchesBatched: %v", err) + } + if len(gh.prQueries) != 0 { + t.Errorf("expected zero PR GraphQL calls when repo is cached as missing, got %d", len(gh.prQueries)) + } + if len(gh.branchQueries) != 0 { + t.Errorf("expected zero branch GraphQL calls when repo is cached as missing, got %d", len(gh.branchQueries)) + } + if len(results) != 1 || !results[0].SyncFailed { + t.Errorf("expected one SyncFailed result for cached-missing watch, got %+v", results) + } +} + +// TestSyncWatchesBatched_LearnsAndCachesMissingRepo verifies the +// "first call discovers the dead repo, the second call short-circuits" +// flow that collapses the storm down to one upstream call per 10 +// minutes per repo. Mirrors TestService_GetPRFeedback_UsesCache's +// repeated-call assertion but on the SyncWatchesBatched seam. +func TestSyncWatchesBatched_LearnsAndCachesMissingRepo(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + + w := &PRWatch{SessionID: "s1", TaskID: "t1", Owner: "Dead", Repo: "Repo", PRNumber: 7, Branch: "br"} + if err := store.CreatePRWatch(ctx, w); err != nil { + t.Fatalf("create watch: %v", err) + } + + // Canned response: data has "repo0": null, errors[].path = ["repo0"] + // with the deterministic "Could not resolve to a Repository" message. + // This mirrors GitHub's actual response shape for a dead repo alias. + gh.prResponses = []string{`{ + "data": {"repo0": null}, + "errors": [{"message": "Could not resolve to a Repository with the name 'Dead/Repo'.", "type": "NOT_FOUND", "path": ["repo0"]}] + }`} + + if _, err := svc.SyncWatchesBatched(ctx, []*PRWatch{w}); err != nil { + t.Fatalf("first SyncWatchesBatched: %v", err) + } + if !svc.isRepoCachedAsMissing(w.Owner, w.Repo) { + t.Fatalf("expected repo to be negative-cached after first sync; cache=%v", svc.repoErrorCache.entries) + } + + // Second call must NOT hit gh — the cache short-circuits everything. + if _, err := svc.SyncWatchesBatched(ctx, []*PRWatch{w}); err != nil { + t.Fatalf("second SyncWatchesBatched: %v", err) + } + if got := len(gh.prQueries); got != 1 { + t.Errorf("expected 1 upstream PR call across two syncs (first discovers, second short-circuits), got %d", got) + } +} + +// blockingExec wraps the existing graphQLMockClient with a controllable +// blocker. The test releases it after asserting that concurrent +// SyncWatchesBatched calls have all coalesced onto the same in-flight +// fetch — mirroring TestService_GetPRFeedback_CoalescesConcurrentCalls. +type blockingExec struct { + *graphQLMockClient + calls atomic.Int32 + release chan struct{} +} + +func (b *blockingExec) ExecuteGraphQL(ctx context.Context, query string, vars map[string]any, out any) error { + b.calls.Add(1) + <-b.release + return b.graphQLMockClient.ExecuteGraphQL(ctx, query, vars, out) +} + +// TestSyncWatchesBatched_NegativeCache_CoalescesConcurrentCalls verifies +// that N concurrent SyncWatchesBatched calls for the same dead repo +// don't fan out into N gh subprocesses while waiting for the first +// classification to land in the negative cache. The throttle protects +// us at process level, but the cache check + classifier is the seam +// that has to coalesce here — pre-fix, every concurrent caller would +// race the empty cache and queue behind the gh throttle. +func TestSyncWatchesBatched_NegativeCache_CoalescesConcurrentCalls(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + + w := &PRWatch{SessionID: "s1", TaskID: "t1", Owner: "Dead", Repo: "Repo", PRNumber: 7, Branch: "br"} + if err := store.CreatePRWatch(context.Background(), w); err != nil { + t.Fatalf("create watch: %v", err) + } + + // Prime the negative cache directly — the test then asserts that + // every concurrent caller short-circuits without queueing on + // ExecuteGraphQL. (Coalescing-via-singleflight inside the cache is + // covered by lower-level cache tests; here we just need to confirm + // SyncWatchesBatched honors the cache under concurrency.) + svc.markRepoAsMissing(w.Owner, w.Repo, svc.repoErrorGenSnapshot()) + + blocker := &blockingExec{graphQLMockClient: gh, release: make(chan struct{})} + svc.client = blocker + close(blocker.release) // never block — if any call reaches gh the test will catch it via blocker.calls + + const n = 16 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + if _, err := svc.SyncWatchesBatched(context.Background(), []*PRWatch{w}); err != nil { + t.Errorf("concurrent SyncWatchesBatched: %v", err) + } + }() + } + wg.Wait() + + if got := blocker.calls.Load(); got != 0 { + t.Fatalf("expected 0 upstream calls (every caller should short-circuit on cache), got %d", got) + } +} + +// TestSyncWatchesBatched_FirstMissCoalesces is the regression test for +// the burst-before-first-cache-hit window: when N concurrent +// SyncWatchesBatched calls race against an EMPTY negative cache for the +// same dead repo, the singleflight wrap around fetchBatchedWatchStatuses +// must collapse them to a single upstream GraphQL call. Without it, the +// pre-filter's isRepoCachedAsMissing check would all return false (cache +// is empty) and every caller would issue its own batched query, stacking +// upstream requests during the exact window the negative cache is trying +// to calm down. +func TestSyncWatchesBatched_FirstMissCoalesces(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + + w := &PRWatch{SessionID: "s1", TaskID: "t1", Owner: "Dead", Repo: "Repo", PRNumber: 7, Branch: "br"} + if err := store.CreatePRWatch(context.Background(), w); err != nil { + t.Fatalf("create watch: %v", err) + } + + // Block the first in-flight ExecuteGraphQL so concurrent callers + // have time to join the singleflight before any response lands. The + // `joined` counter is the deterministic barrier: every caller that + // joins the singleflight bumps it before the leader's fetch returns, + // so we can wait for "all n callers are inside the singleflight" + // without a timing-sensitive sleep. + release := make(chan struct{}) + var gotFirstInFlight sync.WaitGroup + gotFirstInFlight.Add(1) + var firstSignaled atomic.Bool + var joined atomic.Int32 + gh.onExecute = func() { + if firstSignaled.CompareAndSwap(false, true) { + gotFirstInFlight.Done() + } + <-release + } + gh.prResponses = []string{`{ + "data": {"repo0": null}, + "errors": [{"message": "Could not resolve to a Repository with the name 'Dead/Repo'.", "type": "NOT_FOUND", "path": ["repo0"]}] + }`} + + const n = 16 + var wg sync.WaitGroup + wg.Add(n) + errs := make(chan error, n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + joined.Add(1) + if _, err := svc.SyncWatchesBatched(context.Background(), []*PRWatch{w}); err != nil { + errs <- err + } + }() + } + // Wait until the leader is inside ExecuteGraphQL, then wait until + // every co-waiter has reached the singleflight join before releasing. + // Pre-fix this used a fixed 50ms sleep, which violated the backend's + // no-Sleep-in-pure-in-memory-tests convention and was prone to CI + // scheduler variance. + gotFirstInFlight.Wait() + for joined.Load() < int32(n) { + runtime.Gosched() + } + close(release) + wg.Wait() + close(errs) + + for err := range errs { + t.Errorf("concurrent SyncWatchesBatched: %v", err) + } + if got := len(gh.prQueries); got != 1 { + t.Errorf("expected 1 upstream PR call when concurrent first-time misses coalesce, got %d", got) + } + if !svc.isRepoCachedAsMissing(w.Owner, w.Repo) { + t.Errorf("expected repo to be negative-cached after the coalesced fetch") + } +} + +// TestService_ClearRepoErrorCache verifies that the new bulk-clear hook +// (wired from ConfigureToken / ClearToken) drops every entry. Without +// this, a repo classified as unresolvable under stale credentials would +// stay "permanent: true" for up to 10 minutes after the user fixes auth, +// and the frontend retry loop would stop early on stale data. +func TestService_ClearRepoErrorCache(t *testing.T) { + _, svc, _, _ := setupBatchedPollerTest(t) + + gen := svc.repoErrorGenSnapshot() + svc.markRepoAsMissing("Dead1", "Repo", gen) + svc.markRepoAsMissing("Dead2", "Repo", gen) + if !svc.isRepoCachedAsMissing("Dead1", "Repo") || !svc.isRepoCachedAsMissing("Dead2", "Repo") { + t.Fatalf("priming failed") + } + + svc.ClearRepoErrorCache() + + if svc.isRepoCachedAsMissing("Dead1", "Repo") { + t.Errorf("Dead1/Repo still cached after ClearRepoErrorCache") + } + if svc.isRepoCachedAsMissing("Dead2", "Repo") { + t.Errorf("Dead2/Repo still cached after ClearRepoErrorCache") + } +} + +// TestService_EvictRepoNegative_OnWatchCreate verifies that creating a +// PR watch clears any prior negative-cache entry for the repo, so a +// freshly linked repository is probed immediately instead of waiting +// out the 10-min TTL. +func TestService_EvictRepoNegative_OnWatchCreate(t *testing.T) { + _, svc, _, store := setupBatchedPollerTest(t) + ctx := context.Background() + + svc.markRepoAsMissing("Dead", "Repo", svc.repoErrorGenSnapshot()) + if !svc.isRepoCachedAsMissing("Dead", "Repo") { + t.Fatalf("priming failed; cache=%v", svc.repoErrorCache.entries) + } + + // CreatePRWatch must evict regardless of input casing, because the + // cache key is case-insensitive. + w := &PRWatch{SessionID: "s1", TaskID: "t1", Owner: "dead", Repo: "repo", PRNumber: 1, Branch: "main"} + if _, err := svc.CreatePRWatch(ctx, w.SessionID, w.TaskID, "", w.Owner, w.Repo, w.PRNumber, w.Branch); err != nil { + t.Fatalf("CreatePRWatch: %v", err) + } + _ = store + if svc.isRepoCachedAsMissing("Dead", "Repo") { + t.Errorf("expected negative cache to be evicted after CreatePRWatch; entry still present") + } +} + +// TestService_TriggerPRSyncAllPermanent_FlagsDeadRepos pins the +// permanent-flag wiring: a task whose every watch points at a cached +// missing repo gets permanent=true from TriggerPRSyncAllPermanent, so +// the WS handler can tell the frontend to stop polling. Mixed +// (one dead repo + one live one) is NOT permanent — the live one might +// still acquire a PR. +func TestService_TriggerPRSyncAllPermanent_FlagsDeadRepos(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + + w1 := &PRWatch{SessionID: "s1", TaskID: "t1", Owner: "Dead", Repo: "Repo", PRNumber: 0, Branch: "x"} + if err := store.CreatePRWatch(ctx, w1); err != nil { + t.Fatalf("create w1: %v", err) + } + svc.markRepoAsMissing(w1.Owner, w1.Repo, svc.repoErrorGenSnapshot()) + // No canned responses needed — the cache short-circuits both watches. + + _, permanent, err := svc.TriggerPRSyncAllPermanent(ctx, "t1") + if err != nil { + t.Fatalf("TriggerPRSyncAllPermanent: %v", err) + } + if !permanent { + t.Errorf("expected permanent=true when every watch's repo is cached-missing") + } + + // Add a second watch on a live repo; permanent must flip to false + // even though w1 is still cached-missing. + w2 := &PRWatch{SessionID: "s2", TaskID: "t1", Owner: "Live", Repo: "Repo", PRNumber: 0, Branch: "y"} + if err := store.CreatePRWatch(ctx, w2); err != nil { + t.Fatalf("create w2: %v", err) + } + gh.branchResponses = []string{`{"data":{"b0":{"pullRequests":{"nodes":[]}}}}`} + + _, permanent2, err := svc.TriggerPRSyncAllPermanent(ctx, "t1") + if err != nil { + t.Fatalf("TriggerPRSyncAllPermanent (mixed): %v", err) + } + if permanent2 { + t.Errorf("expected permanent=false when at least one watch's repo is live") + } + // Sanity: the live repo's branch query DID fire (single one). + if got := len(gh.branchQueries); got != 1 { + t.Errorf("expected 1 branch GraphQL call for the live repo, got %d", got) + } + // Confirm the dead repo was NOT in any query. + for _, q := range gh.branchQueries { + if strings.Contains(q, w1.Owner) { + t.Errorf("dead repo leaked into branch query: %q", q) + } + } +} diff --git a/apps/backend/internal/github/service_search_cache_test.go b/apps/backend/internal/github/service_search_cache_test.go index 25e8c6399d..81124ad89f 100644 --- a/apps/backend/internal/github/service_search_cache_test.go +++ b/apps/backend/internal/github/service_search_cache_test.go @@ -31,7 +31,9 @@ func newTestService(client Client) *Service { logger: logger.Default(), searchCache: newTTLCache(), prStatusCache: newTTLCache(), + prFeedbackCache: newPRFeedbackCache(), accessibleReposCache: newAccessibleReposCache(), + repoErrorCache: newRepoErrorCache(), } } diff --git a/apps/backend/internal/github/service_tokens.go b/apps/backend/internal/github/service_tokens.go index 8ff6c530ec..affa0829de 100644 --- a/apps/backend/internal/github/service_tokens.go +++ b/apps/backend/internal/github/service_tokens.go @@ -196,17 +196,24 @@ func (s *Service) ConfigureToken(ctx context.Context, token string) error { s.logger.Info("created GitHub token secret", zap.String("id", newID)) } - // Refresh the client to use the new token + // Swap the client AND drop identity-scoped caches atomically under + // s.mu. Clearing OUTSIDE the lock (either before or after the swap) + // opens a race window: clear-before-swap lets an old-identity fetch + // completing in the gap repopulate the just-cleared cache; clear- + // after-swap lets a new-identity caller hit the still-stale cache + // in the gap. Holding the same lock for both publishes a coherent + // (new client, empty caches) state to any goroutine that takes + // s.mu after this. The cache.clear() calls take their own locks + // (not s.mu), so this is fine to nest. The generation bump inside + // each clear() also catches any in-flight fetch that started under + // the old client and tries to write after the lock is released. s.mu.Lock() s.client = testClient s.authMethod = AuthMethodPAT - s.mu.Unlock() - - // Drop user-scoped caches — the previous token's user/orgs and merged - // repo list are no longer valid under the new identity. Without this, - // the picker would surface the prior user's repos for up to 60s after - // a token swap. s.ClearAccessibleReposCaches() + s.ClearRepoErrorCache() + s.ClearPRCaches() + s.mu.Unlock() return nil } @@ -227,14 +234,15 @@ func (s *Service) ClearToken(ctx context.Context) error { } s.logger.Info("cleared GitHub token secret", zap.String("id", existingID)) - // Reset to try gh CLI or other methods + // Reset client + drop identity-scoped caches atomically under s.mu; + // see ConfigureToken for the rationale. s.mu.Lock() s.client = nil s.authMethod = AuthMethodNone - s.mu.Unlock() - - // Drop user-scoped caches — see ConfigureToken for rationale. s.ClearAccessibleReposCaches() + s.ClearRepoErrorCache() + s.ClearPRCaches() + s.mu.Unlock() // Try to re-establish connection via other methods s.retryClientCreation(ctx) diff --git a/apps/backend/internal/github/ttl_cache.go b/apps/backend/internal/github/ttl_cache.go index ef3a9287ef..19e2672258 100644 --- a/apps/backend/internal/github/ttl_cache.go +++ b/apps/backend/internal/github/ttl_cache.go @@ -2,6 +2,7 @@ package github import ( "fmt" + "strings" "sync" "time" @@ -68,6 +69,24 @@ func newAccessibleReposCache() *ttlCache { return c } +// newPRFeedbackCache backs Service.GetPRFeedback (reviews + comments + checks, +// 4 sequential GitHub REST calls per miss). Its job is to collapse the +// render/mount bursts the task page produces — a single PR getting fetched +// dozens of times in a couple of seconds — without those duplicates each +// re-hammering GitHub and tripping its secondary rate limits. +// +// The TTL is deliberately shorter than the 30s status cache: feedback is the +// "fresh, on-demand" surface behind the PR popover/detail panel and the manual +// Refresh button, so staling it for 30s would make legitimate updates lag. +// 8s is long enough to cover a burst's tail (the singleflight in doOrFetch +// already coalesces the concurrent in-flight fetches) while keeping the worst- +// case staleness small. +func newPRFeedbackCache() *ttlCache { + c := newTTLCache() + c.ttl = 8 * time.Second + return c +} + func (c *ttlCache) get(key string) (any, bool) { c.mu.Lock() defer c.mu.Unlock() @@ -160,6 +179,28 @@ func (c *ttlCache) evictLocked() { } } +// del removes a single key from the cache. Used by Service.evictRepoNegative +// to drop a negative entry when a repository is (re)linked to a workspace +// or its watch is recreated, so a freshly-linked repo gets probed +// immediately instead of waiting out the 10-min TTL. Bumps `gen` so any +// classifier fetch that was in flight at the time of the del (e.g. the +// one whose result triggered the relink) cannot reinsert the just-evicted +// key via setIfCurrentGeneration — its post-fetch write becomes a no-op +// and the next ensure() re-probes under the new generation. +func (c *ttlCache) del(key string) { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.entries, key) + c.gen++ +} + +// cachedErr wraps an error stored in a ttlCache so that a deterministic +// upstream failure (e.g. ErrRepoNotResolvable) can be negative-cached +// alongside the cache's normal success-value entries. Currently used by +// Service.repoErrorCache via the lower-level get / setIfCurrentGeneration +// primitives (see Service.isRepoCachedAsMissing / markRepoAsMissing). +type cachedErr struct{ err error } + // doOrFetch returns a cached value when fresh; otherwise runs fetch under a // singleflight guard, caches the result, and returns it. Errors are not // cached. The returned value is shared — callers must not mutate it. @@ -205,3 +246,27 @@ func searchCacheKey(kind, filter, customQuery string, page, perPage int) string func prStatusCacheKey(owner, repo string, number int) string { return fmt.Sprintf("%s/%s#%d", owner, repo, number) } + +// newRepoErrorCache backs Service.repoErrorCache. It negative-caches +// deterministic "repository not resolvable" failures (see +// isRepoNotResolvableErr) so the SyncWatchesBatched storm against +// missing/unauthorized repos collapses to one upstream call per 10 +// minutes per repo instead of one per 5s frontend poll. The TTL is +// deliberately longer than the status / feedback caches because the +// failure mode is "this repo doesn't exist or we can't see it" — a +// state that flips back only via an explicit re-link or token swap, +// both of which evict the entry directly. +func newRepoErrorCache() *ttlCache { + c := newTTLCache() + c.ttl = 10 * time.Minute + return c +} + +// repoErrorCacheKey is the case-insensitive (owner, repo) key for +// Service.repoErrorCache. GitHub's repository names are case-insensitive +// at the API surface, so "NBCUDTC/Bff" and "nbcudtc/bff" reach the same +// upstream resource; normalising the cache key prevents the negative +// entry from being bypassed by a caller that capitalises differently. +func repoErrorCacheKey(owner, repo string) string { + return strings.ToLower(owner) + "/" + strings.ToLower(repo) +} diff --git a/apps/web/hooks/domains/github/use-task-pr.test.tsx b/apps/web/hooks/domains/github/use-task-pr.test.tsx new file mode 100644 index 0000000000..9a3d8b20f0 --- /dev/null +++ b/apps/web/hooks/domains/github/use-task-pr.test.tsx @@ -0,0 +1,82 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createElement, type ReactNode } from "react"; +import { act, cleanup, renderHook } from "@testing-library/react"; +import { StateProvider } from "@/components/state-provider"; + +const requestMock = vi.fn(); + +vi.mock("@/lib/ws/connection", () => ({ + getWebSocketClient: () => ({ request: requestMock }), +})); + +// listWorkspaceTaskPRs is only used by useWorkspacePRs (not under test +// here). Stub it so the module import doesn't fail in jsdom. +vi.mock("@/lib/api/domains/github-api", () => ({ + listWorkspaceTaskPRs: vi.fn().mockResolvedValue(null), +})); + +import { useTaskPR } from "./use-task-pr"; + +function wrapper({ children }: { children: ReactNode }) { + return createElement(StateProvider, null, children); +} + +beforeEach(() => { + vi.useFakeTimers(); + requestMock.mockReset(); +}); +afterEach(() => { + cleanup(); + vi.useRealTimers(); +}); + +describe("useTaskPR — permanent flag", () => { + // The dominant production signal in the SyncWatchesBatched storm was the + // frontend polling `github.task_pr.sync` every 5s for tasks whose repos + // were deleted/inaccessible. The backend now returns `permanent: true` + // on those responses; the hook must stop the retry interval cold. + it("stops the 5s retry interval when the backend reports permanent: true", async () => { + requestMock.mockResolvedValue({ prs: [], permanent: true }); + + renderHook(() => useTaskPR("task-1"), { wrapper }); + + // Initial freshness sync fires synchronously from the mount effect. + // Flush the resolved promise so the permanent flag is applied before + // the interval would otherwise fire. + await act(async () => { + await vi.advanceTimersByTimeAsync(0); + }); + expect(requestMock).toHaveBeenCalledTimes(1); + + // Advance well past several retry windows. If the permanent + // short-circuit regressed, this would burst 5-6 additional calls + // into requestMock. + await act(async () => { + await vi.advanceTimersByTimeAsync(5_000 * 6); + }); + expect(requestMock).toHaveBeenCalledTimes(1); + }); + + // Without permanent, the existing retry cadence must still kick in so + // tasks waiting on a freshly-pushed branch still get their PR detected. + it("retries every 5s when permanent is absent and no PR is in the store", async () => { + requestMock.mockResolvedValue({ prs: [] }); + + renderHook(() => useTaskPR("task-1"), { wrapper }); + + await act(async () => { + await vi.advanceTimersByTimeAsync(0); + }); + expect(requestMock).toHaveBeenCalledTimes(1); + + await act(async () => { + await vi.advanceTimersByTimeAsync(5_000); + }); + expect(requestMock).toHaveBeenCalledTimes(2); + + await act(async () => { + await vi.advanceTimersByTimeAsync(5_000); + }); + expect(requestMock).toHaveBeenCalledTimes(3); + }); +}); diff --git a/apps/web/hooks/domains/github/use-task-pr.ts b/apps/web/hooks/domains/github/use-task-pr.ts index 18a7ccbf5c..08ac58691e 100644 --- a/apps/web/hooks/domains/github/use-task-pr.ts +++ b/apps/web/hooks/domains/github/use-task-pr.ts @@ -52,7 +52,7 @@ export function getPrimaryTaskPR(prs: TaskPR[] | undefined): TaskPR | null { * accept the legacy bare-TaskPR shape too in case an older backend is * still running. Empty / null / unknown shapes return an empty array. */ -function normalizeSyncResponse(result: { prs?: TaskPR[] } | TaskPR | null | undefined): TaskPR[] { +function normalizeSyncResponse(result: SyncResponse): TaskPR[] { if (!result) return []; const envelope = result as { prs?: TaskPR[] }; if (Array.isArray(envelope.prs)) return envelope.prs; @@ -61,30 +61,65 @@ function normalizeSyncResponse(result: { prs?: TaskPR[] } | TaskPR | null | unde return []; } +/** + * Response shape from the `github.task_pr.sync` WS action. `permanent` + * is true when every watch on the task points at a repository the + * backend has classified as unresolvable (missing, deleted, or + * inaccessible). When set, the 5s retry interval stops — without this + * the frontend kept hammering a dead repo every 5s for the lifetime of + * the task. Older backends omit the field, so it's optional. + */ +type SyncResponse = { prs?: TaskPR[]; permanent?: boolean } | TaskPR | null | undefined; + /** Fetch a single task's PR associations, with on-demand sync via WS. */ export function useTaskPR(taskId: string | null) { const prs = useAppStore((state) => (taskId ? (state.taskPRs.byTaskId[taskId] ?? null) : null)); const pr = getPrimaryTaskPR(prs ?? undefined); const setTaskPR = useAppStore((state) => state.setTaskPR); const retryRef = useRef(0); + const permanentRef = useRef(false); + // Monotonic counter incremented before each WS request, snapshotted in + // the .then() closure. Mirrors useWorkspacePRs above. Without this, a + // stale response from a previous taskId can land after the user + // navigates to a new task and flip permanentRef.current = true for the + // active task, killing its retry loop. The reset effect below clears + // retry/permanent state on taskId change, but a still-in-flight WS + // call from the previous task can race that reset. + const requestRef = useRef(0); const refresh = useCallback(() => { if (!taskId) return; const client = getWebSocketClient(); if (!client) return; - // Backend returns `{prs: TaskPR[]}` — multi-repo tasks have one row per - // repo. We push each into the store so the per-repo PR icon stays in - // sync. Empty array means no watches yet (the freshness retry below - // handles the polling cadence). Legacy single-PR shape (`TaskPR` only) - // is detected via the absence of `.prs`. + // Backend returns `{prs: TaskPR[], permanent?: boolean}` — multi-repo + // tasks have one row per repo. We push each into the store so the + // per-repo PR icon stays in sync. Empty array means no watches yet + // (the freshness retry below handles the polling cadence). Legacy + // single-PR shape (`TaskPR` only) is detected via the absence of + // `.prs`. When `permanent` is true (every watch's repo is dead), + // exhaust the retry counter so the 5s interval below clears itself. + const requestId = ++requestRef.current; + const requestedTaskId = taskId; client - .request<{ prs?: TaskPR[] } | TaskPR | null>("github.task_pr.sync", { task_id: taskId }) + .request("github.task_pr.sync", { task_id: requestedTaskId }) .then((result) => { + // Drop responses that aren't the latest in-flight request for + // this hook instance — they'd otherwise corrupt permanentRef / + // retryRef for whatever task the user is now viewing. The + // taskId-change effect below bumps requestRef.current too, so + // requestId alone covers both stale-by-sequence and + // stale-by-task-change. + if (requestRef.current !== requestId) return; + const envelope = (result ?? {}) as { permanent?: boolean }; + if (envelope.permanent) { + permanentRef.current = true; + retryRef.current = SYNC_MAX_RETRIES; + } const list = normalizeSyncResponse(result); if (list.length === 0) return; for (const pr of list) { - if (pr.task_id) setTaskPR(taskId, pr); + if (pr.task_id) setTaskPR(requestedTaskId, pr); } retryRef.current = 0; }) @@ -93,9 +128,13 @@ export function useTaskPR(taskId: string | null) { }); }, [taskId, setTaskPR]); - // Reset retry count when taskId changes + // Reset retry/permanent state when taskId changes. Bumping requestRef + // here invalidates any still-in-flight .then() closure from the prior + // taskId so it can't write to the new task's refs. useEffect(() => { retryRef.current = 0; + permanentRef.current = false; + requestRef.current++; }, [taskId]); // Sync once when the task becomes active (freshness check). @@ -105,12 +144,14 @@ export function useTaskPR(taskId: string | null) { refresh(); }, [taskId, refresh]); - // Retry polling when no PR is in the store yet. + // Retry polling when no PR is in the store yet. permanentRef short-circuits + // the interval entirely so a task whose repos are all dead doesn't tie up + // the backend's gh throttle on every 5s tick. useEffect(() => { - if (!taskId || pr) return; + if (!taskId || pr || permanentRef.current) return; const interval = setInterval(() => { - if (retryRef.current >= SYNC_MAX_RETRIES) { + if (retryRef.current >= SYNC_MAX_RETRIES || permanentRef.current) { clearInterval(interval); return; }