fix(backend): cache & singleflight live PR feedback fetches (#1237)

This commit is contained in:
Carlos Florêncio
2026-06-02 19:37:20 +01:00
committed by GitHub
parent 570c747959
commit 79d4f84bf2
18 changed files with 1559 additions and 121 deletions
@@ -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)
}
@@ -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)
}
}
+68 -1
View File
@@ -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
}
@@ -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)
}
})
}
}
+69 -51
View File
@@ -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
}
+161 -9
View File
@@ -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
}
+15 -4
View File
@@ -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
})
}
+95
View File
@@ -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()
}
}
+45 -3
View File
@@ -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"`
@@ -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)
}
}
@@ -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 {
@@ -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}
@@ -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)
}
}
}
@@ -31,7 +31,9 @@ func newTestService(client Client) *Service {
logger: logger.Default(),
searchCache: newTTLCache(),
prStatusCache: newTTLCache(),
prFeedbackCache: newPRFeedbackCache(),
accessibleReposCache: newAccessibleReposCache(),
repoErrorCache: newRepoErrorCache(),
}
}
+19 -11
View File
@@ -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)
+65
View File
@@ -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)
}
@@ -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);
});
});
+53 -12
View File
@@ -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<SyncResponse>("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;
}