From f2014b2a900523ce65208fe4cea83c40262ff2a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Mon, 17 Aug 2026 10:14:58 +0000 Subject: [PATCH] fix(github): stop re-asking the batched PR watch query once per watch (#2742) Co-authored-by: Kandev Agent --- apps/backend/AGENTS.md | 2 +- apps/backend/internal/github/graphql.go | 54 ++- apps/backend/internal/github/graphql_test.go | 31 +- apps/backend/internal/github/mock_client.go | 18 + apps/backend/internal/github/pat_client.go | 2 +- apps/backend/internal/github/service.go | 2 + .../github/service_automation_auth.go | 75 +++- .../internal/github/service_pr_watch.go | 4 +- .../github/service_pr_watch_batched.go | 96 ++++- .../service_pr_watch_batched_budget_test.go | 329 ++++++++++++++++++ apps/backend/internal/github/ttl_cache.go | 12 + 11 files changed, 577 insertions(+), 48 deletions(-) create mode 100644 apps/backend/internal/github/service_pr_watch_batched_budget_test.go diff --git a/apps/backend/AGENTS.md b/apps/backend/AGENTS.md index b96d305e66..29b25fbe86 100644 --- a/apps/backend/AGENTS.md +++ b/apps/backend/AGENTS.md @@ -220,7 +220,7 @@ Client (WS) ← Orchestrator ← Lifecycle Manager ←──── stream update - **Event-bus wildcard parity:** New NATS wildcard subscriptions must verify equivalent `MemoryEventBus` semantics in `go test ./internal/events/bus`. - **Repository provider identity:** Provider-backed repositories are keyed by workspace, provider, normalized `provider_host` origin, full owner/namespace, and name. Persist `provider_host` when importing or resolving a remote; do not infer self-managed GitLab rows from owner/name alone. Legacy rows with an empty host have unknown identity and must fail closed for provider write/link operations. - **Repository provider branches:** Native task branch pickers derive provider identity only from the persisted workspace repository. GitHub uses the first-party service; manifest-owned providers route through the owning active plugin's workspace-scoped `repositories.branches` action. Do not add provider-specific task-service branches or accept browser-supplied repository descriptors on this path. -- **PR status sync coverage:** A `github_pr_watches` row is unique per (session, repository, branch). A watch that has **found a PR** is never re-pointed: it is that PR's only sync handle, and both the poller and the on-demand sync iterate watches, so re-pointing it froze the PR at its last-observed checks/review state — which the UI renders as an open, mergeable PR long after it merged, and which silently disabled CI auto-fix and auto-merge for every branch but the one currently checked out. Branch switches (`resetPRWatchForBranchSwitch`) and the git-status branch sync (`syncPRWatchBranch`) therefore only re-point a **searching** watch (`pr_number=0`) of the same repository, and otherwise add a watch for the new branch — one searching watch per (session, repository) plus one per PR actually opened. Watches are still not a complete index of a task's PRs (sessions get deleted, PRs get linked by URL), so the orphan sweep stays. `service_pr_unwatched.go` reconciles the `github_task_prs` rows no watch points at, and deliberately writes **lifecycle fields only** (`state` / `merged_at` / `closed_at`) — check and review aggregates belong to the watch-driven sync that actually fetches reviews and check runs, and the per-PR REST status marks its counts populated even with nothing to count, so widening this path to aggregates persists 0/0 over good data (it broke six PR-popover e2e specs exactly that way). Terminal and detached rows are excluded so the set stays bounded. **The backend is the only writer of `github_task_prs`:** the PR-feedback read makes the same upstream calls the status sync does (`GetPR` + `ListPRReviews` + `ListCheckRuns`, plus comments), so it persists through `SyncTaskPR` (`service_pr_feedback_sync.go`) and lets `github.task_pr.updated` converge every surface. The frontend must never patch PR lifecycle fields into its own store copy — that write cannot reach the DB, so it desyncs the panel from the kanban card and evaporates on reload (it also masked the sync hole above for months). Both derivations share `newPRStatus` so they cannot drift; the persist rides *inside* the TTL cache's fetch closure, so a cache hit costs no write; failures are logged and swallowed because this is a read endpoint; and the row lookup is workspace-scoped (`ListTaskPRsByPRNumber`) to match the credential scope the read was authorized under. Finally, `merged` is the one state `SyncTaskPR` refuses to leave (`resolveTerminalMergeState`) since GitHub cannot un-merge a PR — do **not** generalize that into a `merged > closed > open` rank, because `closed -> open` is a real transition (a PR can be reopened) and ranking it as a regression pins reopened PRs closed forever. +- **PR status sync coverage:** A `github_pr_watches` row is unique per (session, repository, branch). A watch that has **found a PR** is never re-pointed: it is that PR's only sync handle, and both the poller and the on-demand sync iterate watches, so re-pointing it froze the PR at its last-observed checks/review state — which the UI renders as an open, mergeable PR long after it merged, and which silently disabled CI auto-fix and auto-merge for every branch but the one currently checked out. Branch switches (`resetPRWatchForBranchSwitch`) and the git-status branch sync (`syncPRWatchBranch`) therefore only re-point a **searching** watch (`pr_number=0`) of the same repository, and otherwise add a watch for the new branch — one searching watch per (session, repository) plus one per PR actually opened. Watches are still not a complete index of a task's PRs (sessions get deleted, PRs get linked by URL), so the orphan sweep stays. `service_pr_unwatched.go` reconciles the `github_task_prs` rows no watch points at, and deliberately writes **lifecycle fields only** (`state` / `merged_at` / `closed_at`) — check and review aggregates belong to the watch-driven sync that actually fetches reviews and check runs, and the per-PR REST status marks its counts populated even with nothing to count, so widening this path to aggregates persists 0/0 over good data (it broke six PR-popover e2e specs exactly that way). Terminal and detached rows are excluded so the set stays bounded. **The backend is the only writer of `github_task_prs`:** the PR-feedback read makes the same upstream calls the status sync does (`GetPR` + `ListPRReviews` + `ListCheckRuns`, plus comments), so it persists through `SyncTaskPR` (`service_pr_feedback_sync.go`) and lets `github.task_pr.updated` converge every surface. The frontend must never patch PR lifecycle fields into its own store copy — that write cannot reach the DB, so it desyncs the panel from the kanban card and evaporates on reload (it also masked the sync hole above for months). Both derivations share `newPRStatus` so they cannot drift; the persist rides *inside* the TTL cache's fetch closure, so a cache hit costs no write; failures are logged and swallowed because this is a read endpoint; and the row lookup is workspace-scoped (`ListTaskPRsByPRNumber`) to match the credential scope the read was authorized under. Finally, `merged` is the one state `SyncTaskPR` refuses to leave (`resolveTerminalMergeState`) since GitHub cannot un-merge a PR — do **not** generalize that into a `merged > closed > open` rank, because `closed -> open` is a real transition (a PR can be reopened) and ranking it as a regression pins reopened PRs closed forever. **The batched branch query's empty answer is authoritative:** an alias resolving with **zero** nodes has definitively said "no open PR on this branch", so `branchBatchResult.ResolvedEmpty` carries that to `applyBatchedSearchingWatch`, which must not re-ask it through a per-watch `client.FindPRByBranch`. That re-ask was one `gh pr list` (GraphQL) plus one `gh api /repos/...` per searching watch per cycle *forever*, since a branch with no PR keeps that shape — measured at 5,200-9,200 GraphQL points/hour against a 5,000/hour limit with 42 searching watches, parking the poller on `github rate-limited` ~40 min into every reset window. On a definitive negative only the fork **parent** (a repository the batch never asked about) is worth a look, and `forkParentRepositoryForLookup` caches fork identity per (credential scope, owner, repo) since that answer is constant. Two-or-more open PRs is **not** definitive — `selectBatchedBranchPRNode` refuses to guess — so that case and an unresolved alias keep the full fork-network fallback, throttled on the watch's `last_checked_at` (`PRSyncFreshnessWindow`) exactly as `triggerPRDetection` is; the batched path is the production path, so a throttle guarding only the legacy per-watch path guards nothing. Changes here need a **call-count** test (`service_pr_watch_batched_budget_test.go`) — an assertion on the resulting `PRStatus` passes with the whole regression restored. - **Execution access:** Workspace-oriented handlers (files, shell, inference, ports, vscode, LSP) MUST use `GetOrEnsureExecution(ctx, sessionID)` — it recovers from backend restarts by creating executions on-demand. Only use `GetExecutionBySessionID` for operations that require a running agent process (prompt, cancel, mode). - **Generated-title ownership:** Claim generated-title ownership only after task, config, and Office eligibility is known. Config tasks, Office/External modes, and workspace-only preparation (`StartAgent: false`) must never claim it; only an eligible agent-start task may claim the generated title. - **Task lifecycle events:** Any code path that mutates a task row must publish via the event bus (`task.created` / `task.updated` / `task.deleted`) — either by going through `Service.CreateTask` / `UpdateTask` / `DeleteTask` / `ArchiveTask`, or by calling `publishTaskEvent` (or one of the `Publish*` helpers in `service_events.go`) directly. Walking `repository.TaskRepository` straight bypasses event publishing and breaks WS-driven UI like the All-Workflows kanban view. `HandoffService`'s cascade methods learned this the hard way — they now require a `TaskEventPublisher` wired via `SetTaskEventPublisher`. New cascade / bulk / cleanup paths must follow the same pattern. **Workflow steps** follow the same rule with their own publisher: step create/update/delete publish `workflow_step.created` / `.updated` / `.deleted` through `internal/workflow/stepevents.Publisher` (the WS gateway fans them out; the orchestrator re-evaluates queue admission off `.updated`). Both mutation surfaces — REST/WS in `internal/workflow/handlers`, MCP in `internal/mcp/handlers` — share that publisher and differ only in the source label they pass, so a new step-mutation path uses it rather than hand-rolling the payload; promoting a start step demotes the previous one, so also publish `.updated` for every entry in `DemotedStartSteps`. diff --git a/apps/backend/internal/github/graphql.go b/apps/backend/internal/github/graphql.go index 6935b4ff84..24d565819c 100644 --- a/apps/backend/internal/github/graphql.go +++ b/apps/backend/internal/github/graphql.go @@ -830,14 +830,35 @@ func decodeBatchedPRChunk( return continuations, nil } +// branchBatchResult is the outcome of one batched branch lookup. +// +// Statuses holds the branches that resolved to exactly one OPEN PR, keyed by +// graphqlBranchKey. ResolvedEmpty holds the branch keys whose repository alias +// resolved and reported ZERO open PRs — a definitive "no PR on this branch" +// that callers must NOT re-ask one watch at a time. A branch in neither set is +// unknown: its alias did not resolve, or it carried several open PRs and +// selectBatchedBranchPRNode deliberately refused to guess. +// +// Both maps are read-only at the call sites (they are shared through the +// service singleflight). +type branchBatchResult struct { + Statuses map[string]*PRStatus + ResolvedEmpty map[string]struct{} +} + // runBatchedBranchQuery executes the branch-lookup query in chunks and maps // each branch name to its unambiguous OPEN PR (if any). Result keys are // "owner/repo/branch" so callers can index by their input refs. -func runBatchedBranchQuery(ctx context.Context, exec GraphQLExecutor, refs []graphQLBranchRef) (map[string]*PRStatus, error) { +func runBatchedBranchQuery( + ctx context.Context, exec GraphQLExecutor, refs []graphQLBranchRef, +) (branchBatchResult, error) { if exec == nil || len(refs) == 0 { - return nil, nil + return branchBatchResult{}, nil + } + out := branchBatchResult{ + Statuses: make(map[string]*PRStatus, len(refs)), + ResolvedEmpty: make(map[string]struct{}, len(refs)), } - 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 @@ -849,22 +870,31 @@ func runBatchedBranchQuery(ctx context.Context, exec GraphQLExecutor, refs []gra Errors []graphQLError `json:"errors"` } if err := exec.ExecuteGraphQL(ctx, query, vars, &resp); err != nil { - return nil, err + return branchBatchResult{}, err } missing, residual := classifyBatchedErrors(resp.Errors, aliasMapForBranchRefs(chunk)) - if err := decodeBatchedBranchChunk(chunk, resp.Data, result); err != nil { - return nil, err + if err := decodeBatchedBranchChunk(chunk, resp.Data, &out); err != nil { + return branchBatchResult{}, err } allMissing = append(allMissing, missing...) allResidual = append(allResidual, residual...) } - return finishBatchedQuery(ctx, exec, result, allMissing, allResidual, nil) + statuses, err := finishBatchedQuery(ctx, exec, out.Statuses, allMissing, allResidual, nil) + if statuses == nil { + // finishBatchedQuery dropped the partial decode (residual errors, or a + // failed review-thread continuation). The negatives decoded alongside + // it are no longer trustworthy as a whole-response answer, so surface + // nothing rather than let a caller skip a fallback on their strength. + return branchBatchResult{}, err + } + out.Statuses = statuses + return out, err } func decodeBatchedBranchChunk( refs []graphQLBranchRef, data map[string]json.RawMessage, - result map[string]*PRStatus, + out *branchBatchResult, ) error { for i, ref := range refs { alias := fmt.Sprintf("b%d", i) @@ -882,10 +912,16 @@ func decodeBatchedBranchChunk( } node, ok := selectBatchedBranchPRNode(inner.PullRequests.Nodes) if !ok { + // Zero nodes is a definitive answer: the repository resolved and + // has no open PR on this branch. Two or more is ambiguous and + // stays unknown so the caller keeps its per-watch fallback. + if len(inner.PullRequests.Nodes) == 0 { + out.ResolvedEmpty[graphqlBranchKey(ref.Owner, ref.Repo, ref.Branch)] = struct{}{} + } continue } status := convertBatchedPRResult(&node.batchedPRResult, ref.Owner, ref.Repo, node.Number) - result[graphqlBranchKey(ref.Owner, ref.Repo, ref.Branch)] = status + out.Statuses[graphqlBranchKey(ref.Owner, ref.Repo, ref.Branch)] = status } return nil } diff --git a/apps/backend/internal/github/graphql_test.go b/apps/backend/internal/github/graphql_test.go index 0e805bedf4..e7e63585f1 100644 --- a/apps/backend/internal/github/graphql_test.go +++ b/apps/backend/internal/github/graphql_test.go @@ -610,9 +610,9 @@ func TestRunBatchedBranchQuery_DecodesPRNode(t *testing.T) { if err != nil { t.Fatalf("run: %v", err) } - status, ok := got[graphqlBranchKey("o", "r", "feat")] + status, ok := got.Statuses[graphqlBranchKey("o", "r", "feat")] if !ok { - t.Fatalf("expected branch result, got keys: %v", keysOf(got)) + t.Fatalf("expected branch result, got keys: %v", keysOf(got.Statuses)) } if status.PR == nil || status.PR.Number != 7 { t.Errorf("expected PR number 7, got %#v", status.PR) @@ -629,7 +629,7 @@ func TestRunBatchedBranchQuery_SkipsUnusedReviewThreadPagination(t *testing.T) { t.Fatalf("run: %v", err) } - status := got[graphqlBranchKey("o", "r", "feat")] + status := got.Statuses[graphqlBranchKey("o", "r", "feat")] if status == nil || status.PR == nil || status.PR.Number != 7 { t.Fatalf("branch status = %#v, want discovered PR #7", status) } @@ -656,8 +656,13 @@ func TestRunBatchedBranchQuery_EmptyNodesReturnsNoResult(t *testing.T) { if err != nil { t.Fatalf("run: %v", err) } - if len(got) != 0 { - t.Fatalf("expected no branch result, got %#v", got) + if len(got.Statuses) != 0 { + t.Fatalf("expected no branch result, got %#v", got.Statuses) + } + // Zero nodes is a definitive "no open PR on this branch"; callers rely on + // it to skip a redundant per-watch lookup. + if _, ok := got.ResolvedEmpty[graphqlBranchKey("o", "r", "feat")]; !ok { + t.Fatalf("expected branch to be recorded as definitively empty, got %#v", got.ResolvedEmpty) } } @@ -697,8 +702,13 @@ func TestRunBatchedBranchQuery_SkipsAmbiguousForkHeads(t *testing.T) { if err != nil { t.Fatalf("run: %v", err) } - if len(got) != 0 { - t.Fatalf("expected ambiguous fork heads to be skipped, got %#v", got) + if len(got.Statuses) != 0 { + t.Fatalf("expected ambiguous fork heads to be skipped, got %#v", got.Statuses) + } + // Ambiguity is not a definitive negative — the caller must keep its + // per-watch fallback rather than conclude there is no PR. + if _, ok := got.ResolvedEmpty[graphqlBranchKey("o", "r", "feat")]; ok { + t.Fatal("ambiguous branch must not be recorded as definitively empty") } } @@ -738,8 +748,11 @@ func TestRunBatchedBranchQuery_SkipsMultipleBranchMatchesEvenWithOwnerMatch(t *t if err != nil { t.Fatalf("run: %v", err) } - if len(got) != 0 { - t.Fatalf("expected duplicate branch matches to be skipped, got %#v", got) + if len(got.Statuses) != 0 { + t.Fatalf("expected duplicate branch matches to be skipped, got %#v", got.Statuses) + } + if _, ok := got.ResolvedEmpty[graphqlBranchKey("o", "r", "feat")]; ok { + t.Fatal("ambiguous branch must not be recorded as definitively empty") } } diff --git a/apps/backend/internal/github/mock_client.go b/apps/backend/internal/github/mock_client.go index e8170971c4..86c8006c4c 100644 --- a/apps/backend/internal/github/mock_client.go +++ b/apps/backend/internal/github/mock_client.go @@ -127,6 +127,10 @@ type MockClient struct { // FindPRByBranch otherwise only takes a read lock. findPRByBranchCalls atomic.Int64 + // getRepositoryCalls counts GetRepository invocations so tests can assert + // that fork-parent resolution is cached rather than re-fetched per watch. + getRepositoryCalls atomic.Int64 + // probeEntered/probeRelease let a test gate FindPRByBranch: when set, each // invocation signals on probeEntered and then blocks until probeRelease is // closed. Used to force concurrent probes to overlap and assert @@ -233,10 +237,22 @@ func (m *MockClient) FindPRByHead(ctx context.Context, owner, repo, headOwner, h // FindPRByBranchCallCount returns how many times FindPRByBranch has been // called. Used by tests asserting detection-probe throttling. +// +// This also counts FindPRByHead: the mock implements it by delegating to +// FindPRByBranch (it reuses the same branch index), so a fork-parent probe +// increments this counter too. Tests asserting on fork-parent lookups are +// reading it through that delegation — do not split the counters without +// re-reading every assertion that uses it. func (m *MockClient) FindPRByBranchCallCount() int { return int(m.findPRByBranchCalls.Load()) } +// GetRepositoryCallCount returns how many times GetRepository has been called. +// Used by tests asserting that fork-parent resolution is cached. +func (m *MockClient) GetRepositoryCallCount() int { + return int(m.getRepositoryCalls.Load()) +} + // GateFindPRByBranch installs a gate around FindPRByBranch: each invocation // signals on entered, then blocks until release is closed. Pass nil/nil to // remove the gate. Used to force concurrent probes to overlap. @@ -428,6 +444,7 @@ func (m *MockClient) HasRepositoryAccess(_ context.Context, owner, repo string) // lightweight repo-search fixture remains separate so existing autocomplete // tests do not accidentally grant write access. func (m *MockClient) GetRepository(_ context.Context, owner, repo string) (*GitHubRepository, error) { + m.getRepositoryCalls.Add(1) m.mu.RLock() defer m.mu.RUnlock() if m.reposUnavailable { @@ -1023,6 +1040,7 @@ func (m *MockClient) Reset() { m.nextGistID = 0 m.repoFiles = make(map[repoKey][]repoFileEntry) m.findPRByBranchCalls.Store(0) + m.getRepositoryCalls.Store(0) m.probeEntered = nil m.probeRelease = nil } diff --git a/apps/backend/internal/github/pat_client.go b/apps/backend/internal/github/pat_client.go index 0bf45adfff..96ce26acd9 100644 --- a/apps/backend/internal/github/pat_client.go +++ b/apps/backend/internal/github/pat_client.go @@ -221,7 +221,7 @@ func (c *PATClient) FindPRByBranch(ctx context.Context, owner, repo, branch stri if err != nil { return nil, fmt.Errorf("find PR by branch %q: %w", branch, err) } - status := statuses[graphqlBranchKey(owner, repo, branch)] + status := statuses.Statuses[graphqlBranchKey(owner, repo, branch)] if status == nil { return nil, nil } diff --git a/apps/backend/internal/github/service.go b/apps/backend/internal/github/service.go index 98cfbe7f69..eb05c9e067 100644 --- a/apps/backend/internal/github/service.go +++ b/apps/backend/internal/github/service.go @@ -128,6 +128,7 @@ type Service struct { mergeMethodsCache *ttlCache accessibleReposCache *ttlCache repoErrorCache *ttlCache + forkParentCache *ttlCache protectionCache *branchProtectionCache rateTracker *RateTracker promptResolver PromptResolver @@ -197,6 +198,7 @@ func NewService(client Client, authMethod string, secrets SecretProvider, store mergeMethodsCache: newMergeMethodsCache(), accessibleReposCache: newAccessibleReposCache(), repoErrorCache: newRepoErrorCache(), + forkParentCache: newForkParentCache(), protectionCache: newBranchProtectionCache(), rateTracker: NewRateTracker(eventBus, log), tokenClientFactory: func(token string) Client { return NewPATClient(token) }, diff --git a/apps/backend/internal/github/service_automation_auth.go b/apps/backend/internal/github/service_automation_auth.go index 45125ccd5d..7997f71cea 100644 --- a/apps/backend/internal/github/service_automation_auth.go +++ b/apps/backend/internal/github/service_automation_auth.go @@ -40,7 +40,7 @@ func (s *Service) FindPRByBranchForWorkspace( if err != nil { return nil, err } - return findPRByBranchInForkNetwork(ctx, resolved.Client, owner, repo, branch) + return s.findPRByBranchInForkNetwork(ctx, resolved.Client, resolved.CacheScope, owner, repo, branch) } // findPRByBranchInForkNetwork first searches the requested repository. When @@ -48,15 +48,25 @@ func (s *Service) FindPRByBranchForWorkspace( // open PR whose head is the exact fork owner and branch. This handles PRs that // target the canonical repository while keeping same-named branches from // another fork out of the association path. -func findPRByBranchInForkNetwork( - ctx context.Context, client Client, owner, repo, branch string, +func (s *Service) findPRByBranchInForkNetwork( + ctx context.Context, client Client, cacheScope, owner, repo, branch string, ) (*PR, error) { pr, err := client.FindPRByBranch(ctx, owner, repo, branch) if err != nil || pr != nil { return pr, err } + return s.findPRInForkParent(ctx, client, cacheScope, owner, repo, branch) +} - parentOwner, parentRepo, isFork, err := forkParentRepositoryForLookup(ctx, client, owner, repo) +// findPRInForkParent is the parent-repository half of the fork-network lookup. +// It is exported to the package for callers that already hold a definitive +// "no open PR in the requested repository" answer (see +// lookupSearchingWatchPR), so they can reach the parent without paying for a +// direct lookup whose result they already know. +func (s *Service) findPRInForkParent( + ctx context.Context, client Client, cacheScope, owner, repo, branch string, +) (*PR, error) { + parentOwner, parentRepo, isFork, err := s.forkParentRepositoryForLookup(ctx, client, cacheScope, owner, repo) if err != nil { return nil, err } @@ -68,7 +78,7 @@ func findPRByBranchInForkNetwork( if !ok { return nil, nil } - pr, err = finder.FindPRByHead(ctx, parentOwner, parentRepo, owner, repo, branch) + pr, err := finder.FindPRByHead(ctx, parentOwner, parentRepo, owner, repo, branch) if err != nil || pr == nil { return pr, err } @@ -78,29 +88,68 @@ func findPRByBranchInForkNetwork( return pr, nil } -func forkParentRepositoryForLookup( - ctx context.Context, client Client, owner, repo string, +// forkParent is the cached outcome of one fork-parent resolution. +type forkParent struct { + Owner string + Repo string + IsFork bool +} + +// forkParentRepositoryForLookup resolves whether (owner, repo) is a fork and, +// if so, which repository it forked from. +// +// The answer is cached per (credential scope, owner, repo): fork status is +// effectively immutable, but this runs once per searching watch per sync cycle, +// which meant one `gh api /repos/...` subprocess per watch per minute against a +// constant answer. Keyed by scope like the sibling caches so a repository +// resolved under one principal's credential is never served to another's. +func (s *Service) forkParentRepositoryForLookup( + ctx context.Context, client Client, cacheScope, owner, repo string, ) (parentOwner, parentRepo string, isFork bool, err error) { repositoryReader, ok := client.(repositoryDetailsReader) if !ok { return "", "", false, nil } + if s == nil || s.forkParentCache == nil { + resolved, resolveErr := resolveForkParent(ctx, repositoryReader, owner, repo) + if resolveErr != nil { + return "", "", false, resolveErr + } + return resolved.Owner, resolved.Repo, resolved.IsFork, nil + } + key := scopedCacheKey(cacheScope, repoErrorCacheKey(owner, repo)) + v, err := s.forkParentCache.doOrFetch(key, func() (any, error) { + return resolveForkParent(ctx, repositoryReader, owner, repo) + }) + if err != nil { + return "", "", false, err + } + resolved, ok := v.(forkParent) + if !ok { + return "", "", false, nil + } + return resolved.Owner, resolved.Repo, resolved.IsFork, nil +} + +func resolveForkParent( + ctx context.Context, repositoryReader repositoryDetailsReader, owner, repo string, +) (forkParent, error) { repository, err := repositoryReader.GetRepository(ctx, owner, repo) if err != nil { var apiErr *GitHubAPIError if errors.As(err, &apiErr) && apiErr.StatusCode == 404 { - return "", "", false, nil + return forkParent{}, nil } - return "", "", false, fmt.Errorf("inspect repository %s/%s for parent PR lookup: %w", owner, repo, err) + return forkParent{}, fmt.Errorf("inspect repository %s/%s for parent PR lookup: %w", owner, repo, err) } if repository == nil || !repository.Fork { - return "", "", false, nil + return forkParent{}, nil } - parentOwner, parentRepo, ok = parseForkParentFullName(repository.ParentFullName) + parentOwner, parentRepo, ok := parseForkParentFullName(repository.ParentFullName) if !ok || (strings.EqualFold(parentOwner, owner) && strings.EqualFold(parentRepo, repo)) { - return "", "", false, nil + return forkParent{}, nil } - return parentOwner, parentRepo, true, nil + return forkParent{Owner: parentOwner, Repo: parentRepo, IsFork: true}, nil } func parseForkParentFullName(fullName string) (owner, repo string, ok bool) { diff --git a/apps/backend/internal/github/service_pr_watch.go b/apps/backend/internal/github/service_pr_watch.go index fc52acff11..5a1491bd60 100644 --- a/apps/backend/internal/github/service_pr_watch.go +++ b/apps/backend/internal/github/service_pr_watch.go @@ -1108,7 +1108,9 @@ func (s *Service) detectPRForWatchOnce( // fires WHILE FindPRByBranch is running wins over our post-fetch // classification — see Service.markRepoAsMissing for the rationale. repoErrGen := s.repoErrorGenSnapshot() - pr, err := findPRByBranchInForkNetwork(ctx, resolved.Client, watch.Owner, watch.Repo, watch.Branch) + pr, err := s.findPRByBranchInForkNetwork( + ctx, resolved.Client, resolved.CacheScope, watch.Owner, watch.Repo, watch.Branch, + ) if err != nil && isRepoNotResolvableErr(err) { s.markRepoAsMissingForScope(resolved.CacheScope, watch.Owner, watch.Repo, repoErrGen) // Wrap so wsSyncTaskPR can errors.Is(err, ErrRepoNotResolvable) diff --git a/apps/backend/internal/github/service_pr_watch_batched.go b/apps/backend/internal/github/service_pr_watch_batched.go index 7b769bb205..6b88e5df2a 100644 --- a/apps/backend/internal/github/service_pr_watch_batched.go +++ b/apps/backend/internal/github/service_pr_watch_batched.go @@ -73,22 +73,36 @@ func (s *Service) syncWatchesBatchedWithClient( return nil, err } numbered, searching := splitPRWatches(watches) - statusByKey, err := s.fetchBatchedWatchStatuses(ctx, exec, cacheScope, numbered, searching) + statuses, err := s.fetchBatchedWatchStatuses(ctx, exec, cacheScope, numbered, searching) if err != nil { return nil, err } + if statuses == nil { + statuses = &batchedWatchStatuses{} + } results := make([]PRWatchSyncResult, 0, len(watches)) now := time.Now().UTC() for _, w := range numbered { - results = append(results, s.applyBatchedNumberedWatch(ctx, cacheScope, w, statusByKey, now)) + results = append(results, s.applyBatchedNumberedWatch(ctx, cacheScope, w, statuses.byKey, now)) } for _, w := range searching { - results = append(results, s.applyBatchedSearchingWatch(ctx, client, cacheScope, w, statusByKey, now)) + results = append(results, s.applyBatchedSearchingWatch(ctx, client, cacheScope, w, statuses, now)) } return results, nil } +// batchedWatchStatuses is the shared, read-only result of one batched fetch. +// +// byKey holds every PR the batch resolved, keyed by prStatusCacheKey for +// numbered watches and graphqlBranchKey for searching ones. branchResolvedEmpty +// holds the branch keys the batch answered definitively as "no open PR" — see +// branchBatchResult for why that distinction matters. +type batchedWatchStatuses struct { + byKey map[string]*PRStatus + branchResolvedEmpty map[string]struct{} +} + // 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. Watches whose @@ -112,7 +126,7 @@ func (s *Service) syncWatchesBatchedWithClient( // the fetch can't outlive the request budget. func (s *Service) fetchBatchedWatchStatuses( ctx context.Context, exec GraphQLExecutor, cacheScope string, numbered, searching []*PRWatch, -) (map[string]*PRStatus, error) { +) (*batchedWatchStatuses, error) { key := scopedCacheKey(cacheScope, batchedFetchSingleflightKey(numbered, searching)) fetchCtx, cancelFetch := derivedFetchContext(ctx) defer cancelFetch() @@ -122,8 +136,11 @@ func (s *Service) fetchBatchedWatchStatuses( // 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, cacheScope, numbered, combined, repoErrGen); err != nil { + combined := &batchedWatchStatuses{ + byKey: make(map[string]*PRStatus, len(numbered)+len(searching)), + branchResolvedEmpty: make(map[string]struct{}, len(searching)), + } + if err := s.fetchBatchedPRStatuses(fetchCtx, exec, cacheScope, numbered, combined.byKey, repoErrGen); err != nil { return nil, err } if err := s.fetchBatchedBranchStatuses(fetchCtx, exec, cacheScope, searching, combined, repoErrGen); err != nil { @@ -137,7 +154,7 @@ func (s *Service) fetchBatchedWatchStatuses( if v == nil { return nil, nil } - return v.(map[string]*PRStatus), nil + return v.(*batchedWatchStatuses), nil } // batchedFetchSingleflightKey builds a deterministic key from the sorted @@ -196,7 +213,7 @@ func (s *Service) fetchBatchedPRStatuses( // 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, cacheScope string, searching []*PRWatch, combined map[string]*PRStatus, + ctx context.Context, exec GraphQLExecutor, cacheScope string, searching []*PRWatch, combined *batchedWatchStatuses, repoErrGen uint64, ) error { refs := make([]graphQLBranchRef, 0, len(searching)) @@ -210,12 +227,20 @@ func (s *Service) fetchBatchedBranchStatuses( return nil } out, err := runBatchedBranchQuery(ctx, exec, refs) - out, err = s.absorbMissingReposErr(out, err, cacheScope, repoErrGen) + statuses, err := s.absorbMissingReposErr(out.Statuses, err, cacheScope, repoErrGen) if err != nil { return fmt.Errorf("batched branch query: %w", err) } - for k, v := range out { - combined[k] = v + for k, v := range statuses { + combined.byKey[k] = v + } + // Only carry the definitive negatives that survived error absorption. A + // purely-missing-repos error keeps its partial decode; anything else made + // absorbMissingReposErr return early above. + if statuses != nil { + for k := range out.ResolvedEmpty { + combined.branchResolvedEmpty[k] = struct{}{} + } } return nil } @@ -322,12 +347,54 @@ func (s *Service) applyBatchedNumberedWatch( return PRWatchSyncResult{Watch: w, Status: status, Found: true, Changed: changed} } +// lookupSearchingWatchPR resolves the PR for a searching watch the batched +// query returned nothing for. Returns (nil, nil) when there is no PR to +// promote — including when the probe was skipped. +// +// The batched query already asked GitHub `pullRequests(states: OPEN, +// headRefName: )` for this exact repository and branch. When it came +// back resolved-and-empty, re-asking through client.FindPRByBranch spends one +// more API call per watch per cycle to receive the same answer — and since a +// branch without a PR keeps that shape indefinitely, that call repeats forever. +// That redundancy was the dominant GitHub GraphQL consumer in production: 42 +// searching watches burned ~150 points/minute against a 5000/hour budget, which +// parked the poller on `github rate-limited` roughly 40 minutes into every +// reset window. So on a definitive negative only the fork PARENT — a different +// repository the batch never asked about — is still worth a look, and that +// lookup is itself cached per (scope, owner, repo). +// +// An alias that did not resolve, or a branch carrying several open PRs that +// selectBatchedBranchPRNode refused to guess between, is not a definitive +// negative and keeps the full fork-network fallback. +// +// Either way the probe is throttled on the watch's own last_checked_at exactly +// as triggerPRDetection's is, so a watch costs at most one upstream probe per +// freshness window no matter how many callers sync it — the frontend retries +// every 5s, and that throttle previously guarded only the legacy per-watch +// path this one replaced. The 60s poller always clears the window, so +// detection latency is unchanged. +func (s *Service) lookupSearchingWatchPR( + ctx context.Context, client Client, cacheScope string, w *PRWatch, + branchKey string, statuses *batchedWatchStatuses, lastChecked *time.Time, +) (*PR, error) { + if lastChecked != nil && time.Since(*lastChecked) < PRSyncFreshnessWindow { + return nil, nil + } + if _, definitive := statuses.branchResolvedEmpty[branchKey]; definitive { + return s.findPRInForkParent(ctx, client, cacheScope, w.Owner, w.Repo, w.Branch) + } + return s.findPRByBranchInForkNetwork(ctx, client, cacheScope, w.Owner, w.Repo, w.Branch) +} + // applyBatchedSearchingWatch mirrors Poller.applyDetectedPR on the service // side. A searching watch (pr_number=0) is promoted to a known PR when the // branch lookup returns one; otherwise we just bump last_checked_at. func (s *Service) applyBatchedSearchingWatch( - ctx context.Context, client Client, cacheScope string, w *PRWatch, statusByKey map[string]*PRStatus, now time.Time, + ctx context.Context, client Client, cacheScope string, w *PRWatch, statuses *batchedWatchStatuses, now time.Time, ) PRWatchSyncResult { + // Read the previous probe time before the stamp below overwrites it — + // lookupSearchingWatchPR throttles on it. + lastChecked := w.LastCheckedAt // Timestamps get bumped on both the "no PR found" and "PR detected" // paths, so hoist the single call above the branch to make that // invariant obvious. @@ -335,9 +402,10 @@ func (s *Service) applyBatchedSearchingWatch( if s.isRepoCachedAsMissingForScope(cacheScope, w.Owner, w.Repo) { return PRWatchSyncResult{Watch: w, SyncFailed: true} } - status, ok := statusByKey[graphqlBranchKey(w.Owner, w.Repo, w.Branch)] + branchKey := graphqlBranchKey(w.Owner, w.Repo, w.Branch) + status, ok := statuses.byKey[branchKey] if !ok || status == nil || status.PR == nil { - pr, err := findPRByBranchInForkNetwork(ctx, client, w.Owner, w.Repo, w.Branch) + pr, err := s.lookupSearchingWatchPR(ctx, client, cacheScope, w, branchKey, statuses, lastChecked) if err != nil { s.logger.Debug("failed to search parent repository for PR", zap.String("watch_id", w.ID), zap.String("branch", w.Branch), zap.Error(err)) diff --git a/apps/backend/internal/github/service_pr_watch_batched_budget_test.go b/apps/backend/internal/github/service_pr_watch_batched_budget_test.go new file mode 100644 index 0000000000..3958cbd9fc --- /dev/null +++ b/apps/backend/internal/github/service_pr_watch_batched_budget_test.go @@ -0,0 +1,329 @@ +package github + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "testing" + "time" +) + +// These tests pin the API budget of the batched watch sync, not just its +// output. The regression they guard: the batched branch query answers "no open +// PR on this branch" definitively, the apply path used to discard that answer, +// and every searching watch then re-asked it through a separate `gh pr list` +// (GraphQL) plus a `gh api /repos/...` (fork check) on EVERY sync cycle. With +// 42 searching watches that burned ~150 GraphQL points/minute against a +// 5000/hour budget and parked the poller on `github rate-limited`. +// +// Assertions here are therefore call counts. A test that only checks the +// resulting PRStatus passes with the bug fully restored. + +// branchAliasResponse builds a `query Branches` response where every supplied +// alias index resolves with zero open PRs, except those named in overrides +// (raw JSON for that alias, or omitted entirely when the value is ""). +func branchAliasResponse(t *testing.T, count int, overrides map[int]string) string { + t.Helper() + aliases := make([]string, 0, count) + for i := range count { + if raw, ok := overrides[i]; ok { + if raw == "" { + continue // alias absent from the response entirely + } + aliases = append(aliases, fmt.Sprintf("%q: %s", fmt.Sprintf("b%d", i), raw)) + continue + } + aliases = append(aliases, fmt.Sprintf("%q: {\"pullRequests\": {\"nodes\": []}}", fmt.Sprintf("b%d", i))) + } + resp := "{\"data\": {" + strings.Join(aliases, ", ") + "}}" + if !json.Valid([]byte(resp)) { + t.Fatalf("built invalid branch response: %s", resp) + } + return resp +} + +func openPRNode(number int, branch, author string) string { + return fmt.Sprintf(`{"pullRequests": {"nodes": [{ + "number": %d, + "state": "OPEN", "title": "PR %d", "url": "https://x/%d", + "isDraft": false, "mergeable": "MERGEABLE", "mergeStateStatus": "CLEAN", + "headRefName": %q, "baseRefName": "main", "headRefOid": "abc", + "author": {"login": %q}, + "createdAt": "2026-01-01T00:00:00Z", "updatedAt": "2026-01-02T00:00:00Z", + "reviews": {"nodes": []}, "reviewRequests": {"totalCount": 0}, + "commits": {"nodes": []} + }]}}`, number, number, number, branch, author) +} + +// seedSearchingWatches creates n searching (pr_number=0) watches on one repo, +// returning them in the same order the batched query aliases them. +func seedSearchingWatches(t *testing.T, store *Store, owner, repo string, n int) []*PRWatch { + t.Helper() + ctx := context.Background() + watches := make([]*PRWatch, 0, n) + for i := range n { + taskID := fmt.Sprintf("task-%d", i) + seedTask(t, store, taskID, false) + w := withTestWorkspace(&PRWatch{ + SessionID: fmt.Sprintf("session-%d", i), + TaskID: taskID, + Owner: owner, + Repo: repo, + PRNumber: 0, + Branch: fmt.Sprintf("feature/branch-%d", i), + }) + if err := store.CreatePRWatch(ctx, w); err != nil { + t.Fatalf("create PR watch %d: %v", i, err) + } + watches = append(watches, w) + } + return watches +} + +// reloadWatches re-reads the watches from the store so a second sync cycle +// sees the last_checked_at the first one stamped, exactly as the poller does. +func reloadWatches(t *testing.T, store *Store, watches []*PRWatch) []*PRWatch { + t.Helper() + ctx := context.Background() + out := make([]*PRWatch, 0, len(watches)) + for _, w := range watches { + fresh, err := store.GetPRWatchBySession(ctx, w.SessionID) + if err != nil { + t.Fatalf("reload watch %s: %v", w.SessionID, err) + } + out = append(out, fresh) + } + return out +} + +func TestSyncWatchesBatched_ResolvedEmptyBranchSkipsPerWatchLookup(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + + const watchCount = 4 + watches := seedSearchingWatches(t, store, "o", "r", watchCount) + gh.SetRepositoryDetails(GitHubRepository{FullName: "o/r", Owner: "o", Name: "r", Fork: false}) + gh.branchResponses = []string{ + branchAliasResponse(t, watchCount, nil), + branchAliasResponse(t, watchCount, nil), + } + + if _, err := svc.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, watches); err != nil { + t.Fatalf("first batched sync: %v", err) + } + if _, err := svc.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, reloadWatches(t, store, watches)); err != nil { + t.Fatalf("second batched sync: %v", err) + } + + if got := len(gh.branchQueries); got != 2 { + t.Errorf("batched branch queries = %d, want 2 (one per cycle)", got) + } + // The batched query already said "no open PR" for every branch. Re-asking + // per watch is what blew the GraphQL budget. + if got := gh.FindPRByBranchCallCount(); got != 0 { + t.Errorf("per-watch branch lookups = %d, want 0 across %d watches x 2 cycles", got, watchCount) + } + // Fork status is constant, so it is resolved once and cached — not once + // per watch per cycle. + if got := gh.GetRepositoryCallCount(); got != 1 { + t.Errorf("repository fork lookups = %d, want 1 across %d watches x 2 cycles", got, watchCount) + } + for _, w := range reloadWatches(t, store, watches) { + if w.LastCheckedAt == nil { + t.Errorf("watch %s was not stamped despite a definitive answer", w.SessionID) + } + if w.PRNumber != 0 { + t.Errorf("watch %s = PR #%d, want to stay searching", w.SessionID, w.PRNumber) + } + } +} + +func TestSyncWatchesBatched_UnresolvedAliasStillFallsBack(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + + const watchCount = 3 + watches := seedSearchingWatches(t, store, "o", "r", watchCount) + gh.SetRepositoryDetails(GitHubRepository{FullName: "o/r", Owner: "o", Name: "r", Fork: false}) + // Alias b1 is absent from the response: not a definitive negative. + gh.branchResponses = []string{branchAliasResponse(t, watchCount, map[int]string{1: ""})} + + if _, err := svc.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, watches); err != nil { + t.Fatalf("batched sync: %v", err) + } + + if got := gh.FindPRByBranchCallCount(); got != 1 { + t.Errorf("per-watch branch lookups = %d, want 1 (only the unresolved alias)", got) + } +} + +func TestSyncWatchesBatched_AmbiguousBranchStillFallsBack(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + + watches := seedSearchingWatches(t, store, "o", "r", 2) + gh.SetRepositoryDetails(GitHubRepository{FullName: "o/r", Owner: "o", Name: "r", Fork: false}) + // b0 carries two open PRs — the decoder refuses to pick one, so the + // caller must not read that as "there is no PR". + twoOpen := `{"pullRequests": {"nodes": [ + {"number": 7, "state": "OPEN", "title": "A", "url": "https://x/7", + "isDraft": false, "mergeable": "MERGEABLE", "mergeStateStatus": "CLEAN", + "headRefName": "feature/branch-0", "baseRefName": "main", "headRefOid": "a", + "author": {"login": "alice"}, + "createdAt": "2026-01-01T00:00:00Z", "updatedAt": "2026-01-01T00:00:00Z", + "reviews": {"nodes": []}, "reviewRequests": {"totalCount": 0}, "commits": {"nodes": []}}, + {"number": 8, "state": "OPEN", "title": "B", "url": "https://x/8", + "isDraft": false, "mergeable": "MERGEABLE", "mergeStateStatus": "CLEAN", + "headRefName": "feature/branch-0", "baseRefName": "main", "headRefOid": "b", + "author": {"login": "bob"}, + "createdAt": "2026-01-01T00:00:00Z", "updatedAt": "2026-01-01T00:00:00Z", + "reviews": {"nodes": []}, "reviewRequests": {"totalCount": 0}, "commits": {"nodes": []}} + ]}}` + gh.branchResponses = []string{branchAliasResponse(t, 2, map[int]string{0: twoOpen})} + + if _, err := svc.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, watches); err != nil { + t.Fatalf("batched sync: %v", err) + } + + if got := gh.FindPRByBranchCallCount(); got != 1 { + t.Errorf("per-watch branch lookups = %d, want 1 (only the ambiguous branch)", got) + } +} + +func TestSyncWatchesBatched_FallbackThrottledInsideFreshnessWindow(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + + watches := seedSearchingWatches(t, store, "o", "r", 1) + gh.SetRepositoryDetails(GitHubRepository{FullName: "o/r", Owner: "o", Name: "r", Fork: false}) + // The alias never resolves, so every cycle wants the fallback. + gh.branchResponses = []string{ + branchAliasResponse(t, 1, map[int]string{0: ""}), + branchAliasResponse(t, 1, map[int]string{0: ""}), + } + + if _, err := svc.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, watches); err != nil { + t.Fatalf("first batched sync: %v", err) + } + if _, err := svc.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, reloadWatches(t, store, watches)); err != nil { + t.Fatalf("second batched sync: %v", err) + } + if got := gh.FindPRByBranchCallCount(); got != 1 { + t.Fatalf("per-watch branch lookups = %d, want 1 (second cycle throttled)", got) + } + + // Age the watch past the freshness window and the fallback runs again. + stale := time.Now().UTC().Add(-2 * PRSyncFreshnessWindow) + if err := store.UpdatePRWatchTimestamps(ctx, watches[0].ID, stale, nil, "", ""); err != nil { + t.Fatalf("age watch: %v", err) + } + gh.branchResponses = []string{branchAliasResponse(t, 1, map[int]string{0: ""})} + if _, err := svc.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, reloadWatches(t, store, watches)); err != nil { + t.Fatalf("third batched sync: %v", err) + } + if got := gh.FindPRByBranchCallCount(); got != 2 { + t.Errorf("per-watch branch lookups = %d, want 2 after the window elapsed", got) + } +} + +func TestSyncWatchesBatched_DetectsPRWithoutPerWatchLookup(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + + watches := seedSearchingWatches(t, store, "o", "r", 2) + gh.SetRepositoryDetails(GitHubRepository{FullName: "o/r", Owner: "o", Name: "r", Fork: false}) + gh.branchResponses = []string{ + branchAliasResponse(t, 2, map[int]string{1: openPRNode(42, "feature/branch-1", "alice")}), + } + + if _, err := svc.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, watches); err != nil { + t.Fatalf("batched sync: %v", err) + } + + updated := reloadWatches(t, store, watches) + if updated[1].PRNumber != 42 { + t.Errorf("detected PR number = %d, want 42", updated[1].PRNumber) + } + if updated[0].PRNumber != 0 { + t.Errorf("watch without a PR = #%d, want to stay searching", updated[0].PRNumber) + } + if got := gh.FindPRByBranchCallCount(); got != 0 { + t.Errorf("per-watch branch lookups = %d, want 0", got) + } +} + +func TestSyncWatchesBatched_ForkParentPRStillDetected(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + + watches := seedSearchingWatches(t, store, "fork", "r", 1) + gh.SetRepositoryDetails(GitHubRepository{ + FullName: "fork/r", Owner: "fork", Name: "r", Fork: true, ParentFullName: "upstream/r", + }) + // The PR lives in the parent with the fork as its head repository. The + // batched query only asked the fork, so a definitive negative there must + // still reach the parent. + gh.AddPR(&PR{ + Number: 99, RepoOwner: "upstream", RepoName: "r", + HeadBranch: "feature/branch-0", HeadRepoOwner: "fork", HeadRepoName: "r", + State: "open", Title: "fork PR", URL: "https://x/99", BaseBranch: "main", + }) + gh.branchResponses = []string{branchAliasResponse(t, 1, nil)} + + if _, err := svc.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, watches); err != nil { + t.Fatalf("batched sync: %v", err) + } + if updated := reloadWatches(t, store, watches); updated[0].PRNumber != 99 { + t.Fatalf("fork-parent PR number = %d, want 99", updated[0].PRNumber) + } +} + +func TestSyncWatchesBatched_ForkParentPRWithForeignHeadIsRejectedAndCached(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + + watches := seedSearchingWatches(t, store, "fork", "r", 1) + gh.SetRepositoryDetails(GitHubRepository{ + FullName: "fork/r", Owner: "fork", Name: "r", Fork: true, ParentFullName: "upstream/r", + }) + // Same branch name, but opened from a different fork — must not associate, + // so the watch stays searching and both cycles take the fork-parent path. + gh.AddPR(&PR{ + Number: 99, RepoOwner: "upstream", RepoName: "r", + HeadBranch: "feature/branch-0", HeadRepoOwner: "someone-else", HeadRepoName: "r", + State: "open", Title: "other fork PR", URL: "https://x/99", BaseBranch: "main", + }) + gh.branchResponses = []string{ + branchAliasResponse(t, 1, nil), + branchAliasResponse(t, 1, nil), + } + + if _, err := svc.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, watches); err != nil { + t.Fatalf("first batched sync: %v", err) + } + if updated := reloadWatches(t, store, watches); updated[0].PRNumber != 0 { + t.Fatalf("watch = PR #%d, want to stay searching (foreign head)", updated[0].PRNumber) + } + // Age past the freshness window so the second cycle genuinely probes the + // parent again — otherwise the throttle, not the cache, would explain a + // single fork lookup. + stale := time.Now().UTC().Add(-2 * PRSyncFreshnessWindow) + if err := store.UpdatePRWatchTimestamps(ctx, watches[0].ID, stale, nil, "", ""); err != nil { + t.Fatalf("age watch: %v", err) + } + if _, err := svc.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, reloadWatches(t, store, watches)); err != nil { + t.Fatalf("second batched sync: %v", err) + } + + // The parent is a different repository the batch never asked about, so it + // is probed on both cycles — but fork identity itself is fetched once. + // The probe here is FindPRByHead, which the mock implements by delegating + // to FindPRByBranch, so this counter sees it. + if got := gh.FindPRByBranchCallCount(); got != 2 { + t.Errorf("fork-parent lookups = %d, want 2 (one per cycle)", got) + } + if got := gh.GetRepositoryCallCount(); got != 1 { + t.Errorf("repository fork lookups = %d, want 1 across 2 cycles", got) + } +} diff --git a/apps/backend/internal/github/ttl_cache.go b/apps/backend/internal/github/ttl_cache.go index 6de3fb0eb8..b5fdddf397 100644 --- a/apps/backend/internal/github/ttl_cache.go +++ b/apps/backend/internal/github/ttl_cache.go @@ -324,6 +324,18 @@ func newRepoErrorCache() *ttlCache { return c } +// newForkParentCache backs Service.forkParentRepositoryForLookup. Whether a +// repository is a fork, and of what, effectively never changes, but the lookup +// runs once per searching PR watch per sync cycle — a constant answer re-fetched +// dozens of times a minute. 10 minutes matches the negative repo cache, so a +// repository that genuinely becomes a fork is picked up within the same window +// the rest of the repo-identity caching already tolerates. +func newForkParentCache() *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