diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 5cadd569a1..78e49d655e 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -327,17 +327,18 @@ const ( // Event types for GitHub integration const ( - GitHubPRFeedback = "github.pr_feedback" // PR has new feedback (UI notification only) - GitHubPRStateChanged = "github.pr_state_changed" // PR state changed (merged, closed, etc.) - GitHubNewReviewPR = "github.new_pr_to_review" // New PR found needing review - GitHubNewIssue = "github.new_issue" // New issue found matching issue watch - GitHubTaskPRUpdated = "github.task_pr.updated" // TaskPR record updated (for UI refresh) - GitHubTaskPRDeleted = "github.task_pr.deleted" // TaskPR association detached (for UI refresh) - GitHubTaskCIOptionsUpdated = "github.task_ci_options.updated" // Task CI automation options updated - GitHubWatchEvent = "github.watch.event" // Watch created/deleted - GitHubRateLimitUpdated = "github.rate_limit.updated" // GitHub API rate-limit snapshot changed - GitHubPushReceived = "github.push_received" // Push webhook verified + installation resolved to workspaces - GitHubCheckRunCompleted = "github.check_run_completed" // Completed check_run webhook resolved to workspaces + GitHubPRFeedback = "github.pr_feedback" // PR has new feedback (UI notification only) + GitHubPRStateChanged = "github.pr_state_changed" // PR state changed (merged, closed, etc.) + GitHubNewReviewPR = "github.new_pr_to_review" // New PR found needing review + GitHubNewIssue = "github.new_issue" // New issue found matching issue watch + GitHubTaskPRUpdated = "github.task_pr.updated" // TaskPR record updated (for UI refresh) + GitHubTaskPRDeleted = "github.task_pr.deleted" // TaskPR association detached (for UI refresh) + GitHubTaskCIOptionsUpdated = "github.task_ci_options.updated" // Task CI automation options updated + GitHubWatchEvent = "github.watch.event" // Watch created/deleted + GitHubRateLimitUpdated = "github.rate_limit.updated" // GitHub API rate-limit snapshot changed + GitHubPRDiscoveryHealthUpdated = "github.pr_discovery_health.updated" // Workspace PR discovery health changed + GitHubPushReceived = "github.push_received" // Push webhook verified + installation resolved to workspaces + GitHubCheckRunCompleted = "github.check_run_completed" // Completed check_run webhook resolved to workspaces ) // Event types for GitLab integration diff --git a/apps/backend/internal/gateway/websocket/task_notifications.go b/apps/backend/internal/gateway/websocket/task_notifications.go index 11fe51b86a..44967b97e7 100644 --- a/apps/backend/internal/gateway/websocket/task_notifications.go +++ b/apps/backend/internal/gateway/websocket/task_notifications.go @@ -99,6 +99,7 @@ func RegisterTaskNotifications(ctx context.Context, eventBus bus.EventBus, hub * b.subscribe(eventBus, events.GitHubTaskPRDeleted, ws.ActionGitHubTaskPRDeleted) b.subscribe(eventBus, events.GitHubTaskCIOptionsUpdated, ws.ActionGitHubTaskCIOptionsUpdated) b.subscribe(eventBus, events.GitHubRateLimitUpdated, ws.ActionGitHubRateLimitUpdated) + b.subscribe(eventBus, events.GitHubPRDiscoveryHealthUpdated, ws.ActionGitHubPRDiscoveryHealthUpdated) b.subscribe(eventBus, events.GitLabTaskMRUpdated, ws.ActionGitLabTaskMRUpdated) b.subscribe(eventBus, events.GitLabTaskMROptionsUpdated, ws.ActionGitLabTaskMRAutomationUpdated) @@ -311,7 +312,8 @@ func (b *TaskEventBroadcaster) routeBroadcast( b.hub.BroadcastToWorkspace(workspaceID, msg) return nil case ws.ActionGitHubTaskPRUpdated, ws.ActionGitHubTaskPRDeleted, - ws.ActionGitHubTaskCIOptionsUpdated, ws.ActionGitLabTaskMRUpdated, ws.ActionGitLabTaskMRAutomationUpdated: + ws.ActionGitHubTaskCIOptionsUpdated, ws.ActionGitHubPRDiscoveryHealthUpdated, + ws.ActionGitLabTaskMRUpdated, ws.ActionGitLabTaskMRAutomationUpdated: // These payloads carry per-task PR/MR automation and lifecycle state. Fail closed // (drop, don't fall back to a global broadcast) when workspace // resolution came back empty and auth is enforced — an unattributed diff --git a/apps/backend/internal/gateway/websocket/task_notifications_test.go b/apps/backend/internal/gateway/websocket/task_notifications_test.go index c963293bb1..b0a5f41645 100644 --- a/apps/backend/internal/gateway/websocket/task_notifications_test.go +++ b/apps/backend/internal/gateway/websocket/task_notifications_test.go @@ -112,7 +112,7 @@ func TestTaskEventBroadcaster_NoDuplicateSubscriptions(t *testing.T) { // // Update this number when adding or removing event subscriptions in // RegisterTaskNotifications — it is intentionally exact. - const wantSubscriptions = 72 + const wantSubscriptions = 73 if got := len(b.subscriptions); got != wantSubscriptions { t.Errorf("RegisterTaskNotifications created %d subscriptions, want %d — "+ "did an event get subscribed twice?", got, wantSubscriptions) diff --git a/apps/backend/internal/github/graphql.go b/apps/backend/internal/github/graphql.go index 645c5ee8fd..fe98e64433 100644 --- a/apps/backend/internal/github/graphql.go +++ b/apps/backend/internal/github/graphql.go @@ -335,6 +335,31 @@ func graphQLErrorsToErr(errs []graphQLError) error { return fmt.Errorf("graphql error: %s (and %d more)", errs[0].Message, len(errs)-1) } +func laterRetryAt(current, candidate *time.Time) *time.Time { + if current == nil { + return copyTime(candidate) + } + if candidate != nil && candidate.After(*current) { + return copyTime(candidate) + } + return copyTime(current) +} + +func graphQLRetryAtFromData(data map[string]json.RawMessage) *time.Time { + raw, ok := data["rateLimit"] + if !ok || len(raw) == 0 || string(raw) == jsonNullLiteral { + return nil + } + var rate graphQLRateLimit + if err := json.Unmarshal(raw, &rate); err != nil || rate.ResetAt.IsZero() { + return nil + } + if rate.Remaining > 0 { + return nil + } + return copyTime(&rate.ResetAt) +} + // 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 @@ -436,7 +461,7 @@ func buildBatchedPRQuery(refs []graphQLPRRef) (string, map[string]any) { // the batched and single-PR paths returning the same data. func prFieldsBlock() string { return `state title url isDraft mergeable mergeStateStatus ` + - `headRefName baseRefName headRefOid headRepository { id name nameWithOwner cloneUrl } headRepositoryOwner { login } additions deletions changedFiles ` + + `headRefName baseRefName headRefOid headRepository { id name nameWithOwner url } headRepositoryOwner { login } additions deletions changedFiles ` + `author { login } mergedBy { login } autoMergeRequest { enabledAt } ` + `mergeQueueEntry { id state position estimatedTimeToMerge headCommit { oid } } ` + `createdAt updatedAt mergedAt closedAt ` + @@ -492,7 +517,7 @@ func convertBatchedPRResult(raw *batchedPRResult, owner, repo string, number int HeadRepoNodeID: raw.HeadRepository.ID, HeadRepoOwner: raw.HeadRepoOwner.Login, HeadRepoName: raw.HeadRepository.Name, - HeadRepoCloneURL: raw.HeadRepository.CloneURL, + HeadRepoCloneURL: httpsCloneURLFromRepositoryURL(raw.HeadRepository.URL), Draft: draft, IsDraftObserved: raw.IsDraft != nil, Mergeable: raw.Mergeable == ghMergeableState, @@ -690,7 +715,12 @@ func (c *PATClient) ExecuteGraphQL(ctx context.Context, query string, variables respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 4*1024*1024)) if resp.StatusCode >= 400 { c.maybeMarkRateExhaustedFromBody("/graphql", resp.StatusCode, respBody) - return &GitHubAPIError{StatusCode: resp.StatusCode, Endpoint: "/graphql", Body: string(respBody)} + return &GitHubAPIError{ + StatusCode: resp.StatusCode, + Endpoint: "/graphql", + Body: string(respBody), + RetryAt: retryAtFromHTTPResponse(resp, ResourceGraphQL, time.Now().UTC()), + } } if err := json.Unmarshal(respBody, out); err != nil { return fmt.Errorf("decode graphql response: %w", err) @@ -777,6 +807,7 @@ func runBatchedPRQuery(ctx context.Context, exec GraphQLExecutor, refs []graphQL // repos landed in the negative cache. var allMissing []repoRef var allResidual []graphQLError + var retryAt *time.Time var continuations []reviewThreadContinuation for _, chunk := range chunkedRefs(refs) { query, vars := buildBatchedPRQuery(chunk) @@ -787,6 +818,7 @@ func runBatchedPRQuery(ctx context.Context, exec GraphQLExecutor, refs []graphQL if err := exec.ExecuteGraphQL(ctx, query, vars, &resp); err != nil { return nil, err } + retryAt = laterRetryAt(retryAt, graphQLRetryAtFromData(resp.Data)) // GitHub returns HTTP 200 with a top-level "errors" array for partial // auth failures, schema mismatches, or per-alias errors. Split out // "Could not resolve to a Repository" entries via classifyBatchedErrors @@ -802,7 +834,7 @@ func runBatchedPRQuery(ctx context.Context, exec GraphQLExecutor, refs []graphQL allMissing = append(allMissing, missing...) allResidual = append(allResidual, residual...) } - return finishBatchedQuery(ctx, exec, result, allMissing, allResidual, continuations) + return finishBatchedQuery(ctx, exec, result, allMissing, allResidual, continuations, retryAt) } func finishBatchedQuery( @@ -812,8 +844,9 @@ func finishBatchedQuery( missing []repoRef, residual []graphQLError, continuations []reviewThreadContinuation, + retryAt *time.Time, ) (map[string]*PRStatus, error) { - batchErr := wrapBatchedErrors(missing, residual) + batchErr := withPRDiscoveryRetryAt(wrapBatchedErrors(missing, residual), retryAt) if batchErr != nil && (len(missing) == 0 || len(residual) > 0) { return nil, batchErr } @@ -863,7 +896,7 @@ func completeReviewThreadContinuations( return err } if err := graphQLErrorsToErr(resp.Errors); err != nil { - return err + return withPRDiscoveryRetryAt(err, graphQLRetryAtFromData(resp.Data)) } chunkNext, err := applyReviewThreadPageChunk(chunk, resp.Data) if err != nil { @@ -1056,6 +1089,7 @@ func runBatchedBranchQuery( // for the rationale (one dead-repo chunk must not drop later chunks). var allMissing []repoRef var allResidual []graphQLError + var retryAt *time.Time for _, chunk := range chunkedRefs(refs) { query, vars := buildBatchedBranchQuery(chunk) var resp struct { @@ -1065,6 +1099,7 @@ func runBatchedBranchQuery( if err := exec.ExecuteGraphQL(ctx, query, vars, &resp); err != nil { return branchBatchResult{}, err } + retryAt = laterRetryAt(retryAt, graphQLRetryAtFromData(resp.Data)) missing, residual := classifyBatchedErrors(resp.Errors, aliasMapForBranchRefs(chunk)) if err := decodeBatchedBranchChunk(chunk, resp.Data, &out); err != nil { return branchBatchResult{}, err @@ -1072,7 +1107,7 @@ func runBatchedBranchQuery( allMissing = append(allMissing, missing...) allResidual = append(allResidual, residual...) } - statuses, err := finishBatchedQuery(ctx, exec, out.Statuses, allMissing, allResidual, nil) + statuses, err := finishBatchedQuery(ctx, exec, out.Statuses, allMissing, allResidual, nil, retryAt) if statuses == nil { // finishBatchedQuery dropped the partial decode (residual errors, or a // failed review-thread continuation). The negatives decoded alongside diff --git a/apps/backend/internal/github/graphql_pr_discovery_contract_test.go b/apps/backend/internal/github/graphql_pr_discovery_contract_test.go new file mode 100644 index 0000000000..4dbeeff03d --- /dev/null +++ b/apps/backend/internal/github/graphql_pr_discovery_contract_test.go @@ -0,0 +1,209 @@ +package github + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" +) + +// githubRepositoryGraphQLFields is the checked subset of the Repository +// object used by PR discovery. The field names are taken from GitHub's +// Repository schema, not from gh CLI JSON aliases: +// https://docs.github.com/en/graphql/reference/objects#repository +var githubRepositoryGraphQLFields = map[string]struct{}{ + "id": {}, + "name": {}, + "nameWithOwner": {}, + "url": {}, +} + +func headRepositorySelection(query string) (map[string]struct{}, bool) { + const prefix = "headRepository {" + start := strings.Index(query, prefix) + if start < 0 { + return nil, false + } + start += len(prefix) + end := strings.Index(query[start:], "}") + if end < 0 { + return nil, false + } + fields := make(map[string]struct{}) + for _, field := range strings.Fields(query[start : start+end]) { + fields[field] = struct{}{} + } + return fields, true +} + +func TestPRDiscoveryQueryContract(t *testing.T) { + queries := map[string]string{} + if query, _ := buildBatchedPRQuery([]graphQLPRRef{{Owner: "acme", Repo: "widget", Number: 1}}); query != "" { + queries["known PR"] = query + } + if query, _ := buildBatchedBranchQuery([]graphQLBranchRef{{Owner: "acme", Repo: "widget", Branch: "feature"}}); query != "" { + queries["branch"] = query + } + + for name, query := range queries { + fields, ok := headRepositorySelection(query) + if !ok { + t.Fatalf("%s query has no headRepository selection: %s", name, query) + } + for field := range fields { + if _, supported := githubRepositoryGraphQLFields[field]; !supported { + t.Errorf("%s query selects unsupported Repository field %q", name, field) + } + } + for field := range githubRepositoryGraphQLFields { + if _, selected := fields[field]; !selected { + t.Errorf("%s query omits supported Repository field %q", name, field) + } + } + } +} + +func TestPRDiscoveryHeadIdentity(t *testing.T) { + tests := []struct { + name string + head ghRepository + headOwner string + wantTarget string + wantHead string + wantCloneURL string + }{ + { + name: "same repository", + head: ghRepository{ + ID: "R_same", + Name: "widget", + NameWithOwner: "acme/widget", + URL: "https://github.com/acme/widget", + }, + headOwner: "acme", + wantTarget: "acme/widget", + wantHead: "acme/widget", + wantCloneURL: "https://github.com/acme/widget.git", + }, + { + name: "fork keeps target distinct", + head: ghRepository{ + ID: "R_fork", + Name: "widget", + NameWithOwner: "alice/widget", + URL: "https://github.com/alice/widget.git", + }, + headOwner: "alice", + wantTarget: "acme/widget", + wantHead: "alice/widget", + wantCloneURL: "https://github.com/alice/widget.git", + }, + { + name: "missing head repository does not invent identity", + wantTarget: "acme/widget", + }, + { + name: "name with owner fills omitted owner fields", + head: ghRepository{ + NameWithOwner: "alice/widget", + }, + wantTarget: "acme/widget", + wantHead: "alice/widget", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + raw := &batchedPRResult{ + HeadRepository: tt.head, + HeadRepoOwner: ghRepositoryOwner{Login: tt.headOwner}, + } + status := convertBatchedPRResult(raw, "acme", "widget", 42) + if got := status.PR.RepoOwner + "/" + status.PR.RepoName; got != tt.wantTarget { + t.Errorf("target repository = %q, want %q", got, tt.wantTarget) + } + gotHead := status.PR.HeadRepoOwner + if status.PR.HeadRepoName != "" { + gotHead += "/" + status.PR.HeadRepoName + } + if gotHead != tt.wantHead { + t.Errorf("head repository = %q, want %q", gotHead, tt.wantHead) + } + if status.PR.HeadRepoCloneURL != tt.wantCloneURL { + t.Errorf("head clone URL = %q, want %q", status.PR.HeadRepoCloneURL, tt.wantCloneURL) + } + }) + } +} + +func TestPRDiscoveryAssociationAfterQueryRecovery(t *testing.T) { + _, service, client, store := setupBatchedPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-discovery", false) + fixedNow := time.Date(2026, 9, 11, 12, 0, 0, 0, time.UTC) + service.prDiscoveryHealth.setNow(func() time.Time { return fixedNow }) + watch := withTestWorkspace(&PRWatch{ + SessionID: "session-discovery", + TaskID: "task-discovery", + Owner: "o", + Repo: "r", + Branch: "feature/discovery", + }) + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create PR watch: %v", err) + } + + updated := make(chan *TaskPR, 1) + subscription, err := service.eventBus.Subscribe(events.GitHubTaskPRUpdated, func(_ context.Context, event *bus.Event) error { + if taskPR, ok := event.Data.(*TaskPR); ok { + updated <- taskPR + } + return nil + }) + if err != nil { + t.Fatalf("subscribe to task PR updates: %v", err) + } + t.Cleanup(func() { _ = subscription.Unsubscribe() }) + + client.branchResponses = []string{ + `{"errors":[{"type":"GRAPHQL_VALIDATION_FAILED","message":"Field 'cloneUrl' doesn't exist on type 'Repository'"}]}`, + branchAliasResponse(t, 1, map[int]string{0: openPRNode(7, "feature/discovery", "alice")}), + } + + if _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watch}); err != nil { + t.Fatalf("health-aware rejected discovery: %v", err) + } + if taskPR, err := store.GetTaskPR(ctx, "task-discovery"); err != nil { + t.Fatalf("check failed association: %v", err) + } else if taskPR != nil { + t.Fatalf("query failure created task PR association: %+v", taskPR) + } + select { + case event := <-updated: + t.Fatalf("query failure published task PR event: %+v", event) + default: + } + + fixedNow = fixedNow.Add(PRDiscoveryRetryBase) + if _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watch}); err != nil { + t.Fatalf("discovery after corrected query: %v", err) + } + associated, err := store.GetTaskPR(ctx, "task-discovery") + if err != nil { + t.Fatalf("read recovered association: %v", err) + } + if associated == nil || associated.PRNumber != 7 { + t.Fatalf("recovered association = %+v, want PR #7", associated) + } + select { + case event := <-updated: + if event.TaskID != "task-discovery" || event.PRNumber != 7 { + t.Fatalf("task PR event = %+v, want task-discovery PR #7", event) + } + case <-time.After(time.Second): + t.Fatal("recovered association did not publish task PR event") + } +} diff --git a/apps/backend/internal/github/pat_client.go b/apps/backend/internal/github/pat_client.go index ac12cb2c1b..3c0ce92519 100644 --- a/apps/backend/internal/github/pat_client.go +++ b/apps/backend/internal/github/pat_client.go @@ -10,6 +10,7 @@ import ( "io" "net/http" "net/url" + "strconv" "strings" "time" ) @@ -32,12 +33,52 @@ type GitHubAPIError struct { StatusCode int Endpoint string Body string + RetryAt *time.Time } func (e *GitHubAPIError) Error() string { return fmt.Sprintf("GitHub API %s returned %d: %s", e.Endpoint, e.StatusCode, e.Body) } +func (e *GitHubAPIError) ProviderRetryAt() *time.Time { + if e == nil || e.RetryAt == nil { + return nil + } + value := *e.RetryAt + return &value +} + +// retryAtFromHTTPResponse preserves provider-supplied retry evidence. The +// later of Retry-After and X-RateLimit-Reset is used so a secondary-limit hint +// cannot cause discovery to resume before the primary reset. +func retryAtFromHTTPResponse(resp *http.Response, defaultResource Resource, now time.Time) *time.Time { + if resp == nil { + return nil + } + var latest *time.Time + if value := strings.TrimSpace(resp.Header.Get("Retry-After")); value != "" { + var retryAt time.Time + if seconds, err := strconv.ParseInt(value, 10, 64); err == nil { + retryAt = now.Add(time.Duration(seconds) * time.Second) + } else if parsed, parseErr := http.ParseTime(value); parseErr == nil { + retryAt = parsed.UTC() + } + if retryAt.After(now) { + latest = &retryAt + } + } + if snapshot, ok := parseRateHeaders(resp, defaultResource); ok && snapshot.ResetAt.After(now) && + (latest == nil || snapshot.ResetAt.After(*latest)) { + resetAt := snapshot.ResetAt + latest = &resetAt + } + if latest == nil { + return nil + } + copy := *latest + return © +} + // WithRateTracker attaches a rate tracker so response headers are recorded. // Returns the client for chaining; safe to call before any requests are made. func (c *PATClient) WithRateTracker(t *RateTracker) *PATClient { diff --git a/apps/backend/internal/github/pat_client_test.go b/apps/backend/internal/github/pat_client_test.go index 59a3a507f6..3dd78084cb 100644 --- a/apps/backend/internal/github/pat_client_test.go +++ b/apps/backend/internal/github/pat_client_test.go @@ -83,6 +83,61 @@ func TestConvertPatPR(t *testing.T) { } } +func TestExecuteGraphQLPreservesHTTPProviderRetryDeadline(t *testing.T) { + tests := []struct { + name string + status int + body string + configure func(http.Header, time.Time) + minWait time.Duration + }{ + { + name: "429 retry-after", + status: http.StatusTooManyRequests, + body: `{"message":"rate limit exceeded"}`, + minWait: 119 * time.Second, + configure: func(headers http.Header, _ time.Time) { + headers.Set("Retry-After", "120") + }, + }, + { + name: "403 reset", + status: http.StatusForbidden, + body: `{"message":"API rate limit exceeded"}`, + minWait: 119 * time.Second, + configure: func(headers http.Header, now time.Time) { + headers.Set("X-RateLimit-Limit", "5000") + headers.Set("X-RateLimit-Remaining", "0") + headers.Set("X-RateLimit-Reset", strconv.FormatInt(now.Add(120*time.Second).Unix(), 10)) + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + now := time.Now().UTC() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + tt.configure(w.Header(), now) + w.WriteHeader(tt.status) + _, _ = w.Write([]byte(tt.body)) + })) + t.Cleanup(srv.Close) + client := newPATClientPointingAt(t, srv.URL) + var out map[string]any + err := client.ExecuteGraphQL(context.Background(), "query Test { rateLimit { resetAt } }", nil, &out) + if err == nil { + t.Fatal("ExecuteGraphQL succeeded, want provider error") + } + var apiErr *GitHubAPIError + if !errors.As(err, &apiErr) { + t.Fatalf("error = %T %v, want GitHubAPIError", err, err) + } + if apiErr.RetryAt == nil || apiErr.RetryAt.Before(now.Add(tt.minWait)) { + t.Fatalf("retry deadline = %v, want at least %v", apiErr.RetryAt, now.Add(tt.minWait)) + } + }) + } +} + // TestConvertPatPR_MissingDraftAndChangedFilesLeavesUnobserved covers // AC-12a on the REST decode path: a response that omits draft and // changed_files must decode to unobserved (Observed=false), not a diff --git a/apps/backend/internal/github/poller.go b/apps/backend/internal/github/poller.go index f1d884d9d8..394877be68 100644 --- a/apps/backend/internal/github/poller.go +++ b/apps/backend/internal/github/poller.go @@ -347,9 +347,7 @@ func (p *Poller) detectPRForWatch(ctx context.Context, watch *PRWatch) { return } - pr, err := p.service.FindPRByBranchForWorkspace( - ctx, watch.WorkspaceID, watch.Owner, watch.Repo, watch.Branch, - ) + pr, err := p.service.findPRByBranchForWatch(ctx, watch) if err != nil { p.logger.Debug("failed to search for PR by branch", zap.String("watch_id", watch.ID), diff --git a/apps/backend/internal/github/poller_test.go b/apps/backend/internal/github/poller_test.go index d5e6ef603a..49e1a30e3d 100644 --- a/apps/backend/internal/github/poller_test.go +++ b/apps/backend/internal/github/poller_test.go @@ -5,7 +5,9 @@ import ( "database/sql" "encoding/json" "errors" + "runtime" "strings" + "sync" "testing" "testing/synctest" "time" @@ -649,6 +651,326 @@ func TestTryBatchedPRWatchCheck_NumberedWatch_AppliesStatus(t *testing.T) { } } +func TestSyncWorkspaceWatchesBatched_DuplicatePRFansOutTaskAssociation(t *testing.T) { + _, service, client, store := setupBatchedPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-duplicate-a", false) + seedTask(t, store, "task-duplicate-b", false) + + watchA := withTestWorkspace(&PRWatch{ + SessionID: "session-duplicate-a", TaskID: "task-duplicate-a", Owner: "o", Repo: "r", + PRNumber: 42, Branch: "feature/shared-pr", LastCheckStatus: "pending", + }) + watchB := withTestWorkspace(&PRWatch{ + SessionID: "session-duplicate-b", TaskID: "task-duplicate-b", Owner: "o", Repo: "r", + PRNumber: 42, Branch: "feature/shared-pr", LastCheckStatus: "pending", + }) + for _, watch := range []*PRWatch{watchA, watchB} { + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create duplicate PR watch: %v", err) + } + } + client.prResponses = []string{`{"data":{"repo0":{"pr0":{ + "state":"OPEN","title":"Shared PR","url":"https://x/42", + "headRefName":"feature/shared-pr","baseRefName":"main","headRefOid":"abc", + "author":{"login":"alice"}, + "createdAt":"2026-01-01T00:00:00Z","updatedAt":"2026-01-02T00:00:00Z", + "reviews":{"nodes":[]},"reviewRequests":{"totalCount":0}, + "commits":{"nodes":[{"commit":{"statusCheckRollup":{"state":"SUCCESS"}}}]} + }}}}`} + + if _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watchA, watchB}); err != nil { + t.Fatalf("sync duplicate PR watches: %v", err) + } + if got := len(client.prQueries); got != 1 { + t.Fatalf("duplicate PR transport calls = %d, want one", got) + } + for _, taskID := range []string{"task-duplicate-a", "task-duplicate-b"} { + pr, err := store.GetTaskPR(ctx, taskID) + if err != nil || pr == nil { + t.Fatalf("task %s association = %#v, err=%v; duplicate consumer was dropped", taskID, pr, err) + } + if pr.ChecksState != "success" { + t.Errorf("task %s checks state = %q, want success", taskID, pr.ChecksState) + } + } +} + +func TestSyncWorkspaceWatchesBatched_ConcurrentEntryPointsShareDuplicateResult(t *testing.T) { + _, service, client, store := setupBatchedPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-concurrent-a", false) + seedTask(t, store, "task-concurrent-b", false) + watchA := withTestWorkspace(&PRWatch{ + SessionID: "session-concurrent-a", TaskID: "task-concurrent-a", Owner: "o", Repo: "r", + PRNumber: 43, Branch: "feature/concurrent", + }) + watchB := withTestWorkspace(&PRWatch{ + SessionID: "session-concurrent-b", TaskID: "task-concurrent-b", Owner: "o", Repo: "r", + PRNumber: 43, Branch: "feature/concurrent", + }) + for _, watch := range []*PRWatch{watchA, watchB} { + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create concurrent watch: %v", err) + } + } + healthScope := testAutomationScope(t, service, testWorkspaceID) + client.prResponses = []string{`{"data":{"repo0":{"pr0":{ + "state":"OPEN","title":"Concurrent PR","url":"https://x/43", + "headRefName":"feature/concurrent","baseRefName":"main","headRefOid":"def", + "author":{"login":"alice"},"createdAt":"2026-01-01T00:00:00Z", + "updatedAt":"2026-01-02T00:00:00Z","reviews":{"nodes":[]}, + "reviewRequests":{"totalCount":0},"commits":{"nodes":[]} + }}}}`} + started := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once + client.onExecute = func() { + once.Do(func() { close(started) }) + <-release + } + firstDone := make(chan error, 1) + go func() { + _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watchA}) + firstDone <- err + }() + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("first entry point never reached GraphQL") + } + secondDone := make(chan error, 1) + go func() { + _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watchB}) + secondDone <- err + }() + + // The second entry point must register its consumer and join the active + // target before the leader is released. This is a bounded scheduler wait, + // not a timing sleep. + deadline := time.After(2 * time.Second) + for { + service.prDiscoveryHealth.mu.Lock() + scope := service.prDiscoveryHealth.scopes[prDiscoveryScopeKey(testWorkspaceID, healthScope)] + joined := false + if scope != nil { + state := scope.targets[prDiscoveryHealthTarget{Owner: "o", Repo: "r", PRNumber: 43}.key()] + joined = state != nil && state.active && len(state.consumers) == 2 + } + service.prDiscoveryHealth.mu.Unlock() + if joined { + break + } + select { + case <-deadline: + t.Fatal("second entry point did not join active discovery target") + default: + runtime.Gosched() + } + } + close(release) + if err := <-firstDone; err != nil { + t.Fatalf("first concurrent sync: %v", err) + } + if err := <-secondDone; err != nil { + t.Fatalf("second concurrent sync: %v", err) + } + if got := len(client.prQueries); got != 1 { + t.Fatalf("concurrent duplicate transport calls = %d, want one", got) + } + for _, taskID := range []string{"task-concurrent-a", "task-concurrent-b"} { + pr, err := store.GetTaskPR(ctx, taskID) + if err != nil || pr == nil { + t.Fatalf("task %s association = %#v, err=%v; concurrent consumer was dropped", taskID, pr, err) + } + } +} + +func TestSyncWorkspaceWatchesBatched_OverlappingEntryPointsShareTargetResult(t *testing.T) { + _, service, client, store := setupBatchedPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-overlap-a", false) + seedTask(t, store, "task-overlap-b", false) + seedTask(t, store, "task-overlap-extra", false) + watchA := withTestWorkspace(&PRWatch{ + SessionID: "session-overlap-a", TaskID: "task-overlap-a", Owner: "o", Repo: "r", + PRNumber: 43, Branch: "feature/overlap", + }) + watchB := withTestWorkspace(&PRWatch{ + SessionID: "session-overlap-b", TaskID: "task-overlap-b", Owner: "o", Repo: "r", + PRNumber: 43, Branch: "feature/overlap", + }) + extra := withTestWorkspace(&PRWatch{ + SessionID: "session-overlap-extra", TaskID: "task-overlap-extra", Owner: "o", Repo: "r", + PRNumber: 44, Branch: "feature/overlap-extra", + }) + for _, watch := range []*PRWatch{watchA, watchB, extra} { + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create overlapping watch: %v", err) + } + } + healthScope := testAutomationScope(t, service, testWorkspaceID) + client.prResponses = []string{`{"data":{"repo0":{"pr0":{ + "state":"OPEN","title":"Shared PR","url":"https://x/43", + "headRefName":"feature/overlap","baseRefName":"main","headRefOid":"def", + "author":{"login":"alice"},"createdAt":"2026-01-01T00:00:00Z", + "updatedAt":"2026-01-02T00:00:00Z","reviews":{"nodes":[]}, + "reviewRequests":{"totalCount":0},"commits":{"nodes":[]} + }},"repo1":{"pr0":{ + "state":"OPEN","title":"Extra PR","url":"https://x/44", + "headRefName":"feature/overlap-extra","baseRefName":"main","headRefOid":"ghi", + "author":{"login":"alice"},"createdAt":"2026-01-01T00:00:00Z", + "updatedAt":"2026-01-02T00:00:00Z","reviews":{"nodes":[]}, + "reviewRequests":{"totalCount":0},"commits":{"nodes":[]} + }}}}`} + started := make(chan struct{}) + release := make(chan struct{}) + var once sync.Once + client.onExecute = func() { + once.Do(func() { close(started) }) + <-release + } + firstDone := make(chan error, 1) + go func() { + _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watchA, extra}) + firstDone <- err + }() + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("overlapping leader never reached GraphQL") + } + secondDone := make(chan error, 1) + go func() { + _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watchB}) + secondDone <- err + }() + deadline := time.After(2 * time.Second) + for { + service.prDiscoveryHealth.mu.Lock() + scope := service.prDiscoveryHealth.scopes[prDiscoveryScopeKey(testWorkspaceID, healthScope)] + joined := false + if scope != nil { + state := scope.targets[prDiscoveryHealthTarget{Owner: "o", Repo: "r", PRNumber: 43}.key()] + joined = state != nil && state.active && len(state.consumers) == 2 + } + service.prDiscoveryHealth.mu.Unlock() + if joined { + break + } + select { + case <-deadline: + t.Fatal("overlapping consumer did not join active discovery target") + default: + runtime.Gosched() + } + } + close(release) + if err := <-firstDone; err != nil { + t.Fatalf("overlapping leader sync: %v", err) + } + if err := <-secondDone; err != nil { + t.Fatalf("overlapping joined sync: %v", err) + } + if got := len(client.prQueries); got != 1 { + t.Fatalf("overlapping target transport calls = %d, want one", got) + } + for _, taskID := range []string{"task-overlap-a", "task-overlap-b"} { + pr, err := store.GetTaskPR(ctx, taskID) + if err != nil || pr == nil { + t.Fatalf("task %s association = %#v, err=%v; joined consumer was dropped", taskID, pr, err) + } + if pr.PRNumber != 43 { + t.Errorf("task %s PR number = %d, want 43", taskID, pr.PRNumber) + } + } +} + +func TestPollerFallbackAfterUnavailableBatchDoesNotPauseDiscovery(t *testing.T) { + poller, service, client, store := setupBatchedPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-unavailable-fallback", false) + watch := withTestWorkspace(&PRWatch{ + SessionID: "session-unavailable-fallback", TaskID: "task-unavailable-fallback", Owner: "o", Repo: "r", + Branch: "feature/unavailable-fallback", + }) + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create unavailable-fallback watch: %v", err) + } + client.AddPR(&PR{ + Number: 52, RepoOwner: "o", RepoName: "r", HeadRepoOwner: "o", HeadRepoName: "r", + HeadBranch: "feature/unavailable-fallback", State: "open", Title: "Fallback PR", + URL: "https://x/52", BaseBranch: "main", + }) + client.branchErr = errors.New("graphql transport unavailable") + if poller.tryBatchedPRWatchCheck(ctx, []*PRWatch{watch}) { + t.Fatal("unavailable batch unexpectedly succeeded") + } + // The poller's per-watch fallback must be able to acquire the released + // target immediately. A batch transport outage is not provider evidence + // that should pause the target for a minute. + poller.checkSinglePRWatch(ctx, watch) + associated, err := store.GetTaskPR(ctx, "task-unavailable-fallback") + if err != nil { + t.Fatalf("read fallback association: %v", err) + } + if associated == nil || associated.PRNumber != 52 { + t.Fatalf("fallback association = %#v, want PR #52", associated) + } + health := service.prDiscoveryHealth.snapshot(testWorkspaceID, testAutomationScope(t, service, testWorkspaceID), 0) + if health.State == PRDiscoveryHealthDegraded { + t.Fatalf("fallback transport outage left discovery degraded: %+v", health) + } +} + +func TestSyncWorkspaceWatchesBatched_DoesNotApplyAfterWatchDeletion(t *testing.T) { + _, service, client, store := setupBatchedPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-deleted-in-flight", false) + watch := withTestWorkspace(&PRWatch{ + SessionID: "session-deleted-in-flight", TaskID: "task-deleted-in-flight", Owner: "o", Repo: "r", + PRNumber: 53, Branch: "feature/deleted-in-flight", + }) + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create deleted in-flight watch: %v", err) + } + client.prResponses = []string{`{"data":{"repo0":{"pr0":{ + "state":"OPEN","title":"Deleted Watch PR","url":"https://x/53", + "headRefName":"feature/deleted-in-flight","baseRefName":"main","headRefOid":"def", + "author":{"login":"alice"},"createdAt":"2026-01-01T00:00:00Z", + "updatedAt":"2026-01-02T00:00:00Z","reviews":{"nodes":[]}, + "reviewRequests":{"totalCount":0},"commits":{"nodes":[]} + }}}}`} + started := make(chan struct{}) + release := make(chan struct{}) + client.onExecute = func() { + close(started) + <-release + } + done := make(chan error, 1) + go func() { + _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watch}) + done <- err + }() + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("deleted-watch sync never reached GraphQL") + } + if err := service.DeletePRWatch(ctx, watch.ID); err != nil { + t.Fatalf("delete in-flight watch: %v", err) + } + close(release) + if err := <-done; err != nil { + t.Fatalf("deleted-watch sync: %v", err) + } + if associated, err := store.GetTaskPR(ctx, "task-deleted-in-flight"); err != nil { + t.Fatalf("read deleted-watch association: %v", err) + } else if associated != nil { + t.Fatalf("deleted-watch association = %+v, want no persisted result", associated) + } +} + func TestTryBatchedPRWatchCheck_PersistsExactReviewThreadCount(t *testing.T) { poller, _, gh, store := setupBatchedPollerTest(t) ctx := context.Background() diff --git a/apps/backend/internal/github/repository_api.go b/apps/backend/internal/github/repository_api.go index 59bdc5e2b3..4e59ab30ef 100644 --- a/apps/backend/internal/github/repository_api.go +++ b/apps/backend/internal/github/repository_api.go @@ -1,6 +1,33 @@ package github -import "strings" +import ( + "net/url" + "strings" +) + +const httpsScheme = "https" + +// httpsCloneURLFromRepositoryURL converts GitHub GraphQL's Repository.url +// field into the HTTPS clone identity used by task repository consumers. The +// GraphQL schema exposes url, while cloneUrl is a gh CLI JSON field and is not +// valid on Repository. An absent or malformed URL remains absent so nullable +// head repository data is never replaced with an invented identity. +func httpsCloneURLFromRepositoryURL(repositoryURL string) string { + parsed, err := url.Parse(strings.TrimSpace(repositoryURL)) + if err != nil || parsed.Host == "" || strings.Trim(parsed.Path, "/") == "" || + (parsed.Scheme != "http" && parsed.Scheme != httpsScheme) { + return "" + } + parsed.Scheme = httpsScheme + parsed.RawQuery = "" + parsed.Fragment = "" + parsed.Path = strings.TrimRight(parsed.Path, "/") + if !strings.HasSuffix(parsed.Path, ".git") { + parsed.Path += ".git" + } + parsed.RawPath = "" + return parsed.String() +} // githubRepositoryResponse is the shared subset returned by GET and POST // repository endpoints. Keep the provider response private so only the diff --git a/apps/backend/internal/github/service.go b/apps/backend/internal/github/service.go index 8c8936300f..9df010f7a8 100644 --- a/apps/backend/internal/github/service.go +++ b/apps/backend/internal/github/service.go @@ -146,26 +146,29 @@ type Service struct { // a single task by ID) because reset must walk the task tree and clean // up subtasks, runs, and worktrees too. Wired post-construction via // SetCascadeTaskDeleter to avoid an import cycle with the task service. - cascadeTaskDeleter watchreset.TaskDeleter - taskSessionChecker TaskSessionChecker - syncGroup singleflight.Group - taskEventSubs []bus.Subscription - searchCache *ttlCache - prStatusCache *ttlCache - prFeedbackCache *ttlCache - mergeMethodsCache *ttlCache - accessibleReposCache *ttlCache - repoErrorCache *ttlCache - forkParentCache *ttlCache - protectionCache *branchProtectionCache - rateTracker *RateTracker - promptResolver PromptResolver - tokenClientFactory func(string) Client - ghAccountLister func(context.Context) ([]GHAccount, error) - mockAuth *MockAuthState - workspaceAuthorizer func(context.Context, string) error - freshDefaultsMu sync.Mutex - freshDefaultsDone bool + cascadeTaskDeleter watchreset.TaskDeleter + taskSessionChecker TaskSessionChecker + syncGroup singleflight.Group + taskEventSubs []bus.Subscription + searchCache *ttlCache + prStatusCache *ttlCache + prFeedbackCache *ttlCache + mergeMethodsCache *ttlCache + accessibleReposCache *ttlCache + repoErrorCache *ttlCache + forkParentCache *ttlCache + protectionCache *branchProtectionCache + rateTracker *RateTracker + prDiscoveryHealth *prDiscoveryHealthStore + prDiscoveryAttemptsMu sync.Mutex + prDiscoveryAttempts map[string]*prDiscoveryAttemptResult + promptResolver PromptResolver + tokenClientFactory func(string) Client + ghAccountLister func(context.Context) ([]GHAccount, error) + mockAuth *MockAuthState + workspaceAuthorizer func(context.Context, string) error + freshDefaultsMu sync.Mutex + freshDefaultsDone bool // cleanupFailureMu guards cleanupFailureCounts; the cleanup loop is the // only writer but the global sweep + per-watch sweep can run concurrently @@ -229,6 +232,8 @@ func NewService(client Client, authMethod string, secrets SecretProvider, store forkParentCache: newForkParentCache(), protectionCache: newBranchProtectionCache(), rateTracker: NewRateTracker(eventBus, log), + prDiscoveryHealth: newPRDiscoveryHealth(eventBus, log), + prDiscoveryAttempts: make(map[string]*prDiscoveryAttemptResult), tokenClientFactory: func(token string) Client { return NewPATClient(token) }, ghAccountLister: ListGHAccounts, cleanupFailureCounts: make(map[string]int), @@ -269,6 +274,10 @@ func (s *Service) Stop() { s.stopOnce.Do(func() { s.stopCancel() s.bgWG.Wait() + if s.prDiscoveryHealth != nil { + s.prDiscoveryHealth.clearAll() + s.invalidateAllPRDiscoveryAttempts("") + } }) } diff --git a/apps/backend/internal/github/service_app_auth.go b/apps/backend/internal/github/service_app_auth.go index 07e702f881..94f016967a 100644 --- a/apps/backend/internal/github/service_app_auth.go +++ b/apps/backend/internal/github/service_app_auth.go @@ -457,6 +457,7 @@ type WorkspaceAuthStatus struct { TokenConfigured bool `json:"token_configured"` RequiredScopes []string `json:"required_scopes"` RateLimit *GitHubRateLimitInfo `json:"rate_limit,omitempty"` + PRDiscoveryHealth *PRDiscoveryHealth `json:"pr_discovery_health,omitempty"` } func (s *Service) GetWorkspaceAuthStatus( @@ -501,6 +502,10 @@ func (s *Service) getWorkspaceAuthStatus( WorkspaceID: workspaceID, Personal: personal, RequiredScopes: RequiredGitHubScopes, AuthMethod: AuthMethodNone, } + // A status response is authoritative for the current connection. Seed an + // explicit runtime-scoped unknown projection so a disconnected or + // unauthenticated response clears health from a previous connection. + status.PRDiscoveryHealth = s.ensurePRDiscoveryHealth().snapshot(workspaceID, "", 0) if connection == nil { return status, nil } @@ -530,6 +535,11 @@ func (s *Service) getWorkspaceAuthStatus( s.seedWorkspaceRateLimit(ctx, automation, refreshRateLimit) status.Automation.RateLimit = rateLimitInfoForTracker(automation.RateTracker) status.RateLimit = status.Automation.RateLimit + status.PRDiscoveryHealth = s.ensurePRDiscoveryHealth().snapshot( + workspaceID, + credentialCacheScope(automation, CredentialPurposeAutomation), + automation.CredentialGeneration, + ) s.populateEffectivePersonalActors(ctx, workspaceID, userID, status) return status, nil diff --git a/apps/backend/internal/github/service_automation_auth.go b/apps/backend/internal/github/service_automation_auth.go index 3b6f5b311a..d01174eefc 100644 --- a/apps/backend/internal/github/service_automation_auth.go +++ b/apps/backend/internal/github/service_automation_auth.go @@ -32,6 +32,24 @@ func (s *Service) ResolveGitHubAutomationClient(ctx context.Context, workspaceID func (s *Service) FindPRByBranchForWorkspace( ctx context.Context, workspaceID, owner, repo, branch string, +) (*PR, error) { + return s.findPRByBranchForWorkspace(ctx, workspaceID, owner, repo, branch, nil) +} + +// findPRByBranchForWatch is the watch-owned variant used by the legacy poller. +// Keeping the watch identity in the health admission lets a failed direct +// probe be pruned when the watch is deleted, even when GraphQL batching is not +// available. +func (s *Service) findPRByBranchForWatch(ctx context.Context, watch *PRWatch) (*PR, error) { + if watch == nil { + return nil, ErrGitHubWorkspaceRequired + } + return s.findPRByBranchForWorkspace(ctx, watch.WorkspaceID, watch.Owner, watch.Repo, watch.Branch, watch) +} + +func (s *Service) findPRByBranchForWorkspace( + ctx context.Context, + workspaceID, owner, repo, branch string, watch *PRWatch, ) (*PR, error) { if err := s.ensureRepositoryInWorkspaceScope(ctx, workspaceID, owner, repo); err != nil { return nil, err @@ -40,7 +58,65 @@ func (s *Service) FindPRByBranchForWorkspace( if err != nil { return nil, err } - return s.findPRByBranchInForkNetwork(ctx, resolved.Client, resolved.CacheScope, owner, repo, branch) + credentialGeneration := int64(0) + if resolved.credential != nil { + credentialGeneration = resolved.credential.CredentialGeneration + } + tracked := watch != nil && watch.ID != "" + var attempt prDiscoveryWatchAttempt + if tracked { + var admitted bool + attempt, admitted = s.beginPRDiscoveryWatch(workspaceID, resolved.CacheScope, credentialGeneration, watch) + if !admitted { + return nil, nil + } + } + pr, err := s.resolvePRByBranchAttempt( + ctx, resolved.Client, resolved.CacheScope, owner, repo, branch, attempt, + ) + if err != nil { + if tracked { + category := classifyPRDiscoveryError(err) + s.completePRDiscoveryWatchAttempt( + attempt, nil, false, true, err, + category == PRDiscoveryHealthInvalidQuery || category == PRDiscoveryHealthRateLimited, + ) + s.finishPRDiscoveryWatchFailure(workspaceID, resolved.CacheScope, credentialGeneration, attempt, err) + s.forgetPRDiscoveryWatchAttempt(workspaceID, resolved.CacheScope, credentialGeneration, attempt) + } + return nil, err + } + if tracked { + if !attempt.joined { + s.completePRDiscoveryWatchAttempt(attempt, &PRStatus{PR: pr}, true, false, nil, false) + s.finishPRDiscoveryWatchSuccess(workspaceID, resolved.CacheScope, credentialGeneration, attempt) + s.forgetPRDiscoveryWatchAttempt(workspaceID, resolved.CacheScope, credentialGeneration, attempt) + } + if pr != nil { + // Finish the immutable source attempt first, then move the live + // consumer to the discovered repository/PR target. The caller still + // owns persistence, but a successful fork lookup must not leave the + // old fork key retaining the watch forever. + rebound := *watch + rebound.Owner = pr.RepoOwner + rebound.Repo = pr.RepoName + rebound.PRNumber = pr.Number + s.trackPRDiscoveryWatchConsumer( + workspaceID, resolved.CacheScope, credentialGeneration, &rebound, + ) + } + } + return pr, nil +} + +func (s *Service) resolvePRByBranchAttempt( + ctx context.Context, client Client, cacheScope, owner, repo, branch string, + attempt prDiscoveryWatchAttempt, +) (*PR, error) { + if attempt.joined { + return sharedPRDiscoveryResult(ctx, attempt) + } + return s.findPRByBranchInForkNetwork(ctx, client, cacheScope, owner, repo, branch) } // findPRByBranchInForkNetwork first searches the requested repository. When diff --git a/apps/backend/internal/github/service_connections.go b/apps/backend/internal/github/service_connections.go index 39ca93157e..52e0b3416e 100644 --- a/apps/backend/internal/github/service_connections.go +++ b/apps/backend/internal/github/service_connections.go @@ -386,6 +386,10 @@ func (s *Service) invalidateWorkspaceCredential(workspaceID string) { if s.resolver != nil { s.resolver.InvalidateWorkspace(workspaceID) } + if s.prDiscoveryHealth != nil { + s.prDiscoveryHealth.clearWorkspace(workspaceID) + s.invalidateAllPRDiscoveryAttempts(workspaceID) + } // Existing caches are still global during the compatibility phase. Clear // them on replacement until Task 05 prefixes every key by principal. s.clearAuthCaches() diff --git a/apps/backend/internal/github/service_pr_branch_fallback_test.go b/apps/backend/internal/github/service_pr_branch_fallback_test.go index 686359957a..184d0b42b0 100644 --- a/apps/backend/internal/github/service_pr_branch_fallback_test.go +++ b/apps/backend/internal/github/service_pr_branch_fallback_test.go @@ -3,6 +3,7 @@ package github import ( "context" "testing" + "time" ) const ( @@ -157,3 +158,58 @@ func TestSyncWorkspaceWatchesBatched_ForkFallsBackToParent(t *testing.T) { t.Fatalf("watch = %#v, want parent repository and PR #%d", updated, forkPRNumber) } } + +func TestSyncWorkspaceWatchesBatched_ForkRetryReleasesOriginalHealthTarget(t *testing.T) { + _, service, client, store := setupBatchedPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-fork-health", false) + seedForkNetwork(client.MockClient) + client.AddPR(forkNetworkPR()) + + fixedNow := time.Date(2026, 9, 11, 12, 0, 0, 0, time.UTC) + service.prDiscoveryHealth.setNow(func() time.Time { return fixedNow }) + scope := testAutomationScope(t, service, testWorkspaceID) + client.branchResponses = []string{ + `{"errors":[{"type":"GRAPHQL_VALIDATION_FAILED","message":"Field 'cloneUrl' doesn't exist on type 'Repository'"}]}`, + `{"data":{"b0":{"pullRequests":{"nodes":[]}}}}`, + } + watch := &PRWatch{ + WorkspaceID: testWorkspaceID, SessionID: "session-fork-health", TaskID: "task-fork-health", + RepositoryID: "repo-fork", Owner: forkOwner, Repo: forkRepo, Branch: forkPRBranch, + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create fork health watch: %v", err) + } + + if _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watch}); err != nil { + t.Fatalf("initial failed fork discovery: %v", err) + } + failed := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if failed.State != PRDiscoveryHealthDegraded || failed.FailedTargetCount != 1 { + t.Fatalf("failed fork health = %+v, want one degraded fork target", failed) + } + + fixedNow = fixedNow.Add(PRDiscoveryRetryBase) + if _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watch}); err != nil { + t.Fatalf("successful fork retry: %v", err) + } + updated, err := store.GetPRWatch(ctx, watch.ID) + if err != nil || updated == nil { + t.Fatalf("reload fork watch: %v, %#v", err, updated) + } + if updated.Owner != parentOwner || updated.Repo != parentRepo || updated.PRNumber != forkPRNumber { + t.Fatalf("fork watch = %#v, want parent repository and PR #%d", updated, forkPRNumber) + } + recovered := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if recovered.State != PRDiscoveryHealthHealthy || recovered.FailedTargetCount != 0 { + t.Fatalf("fork health after retry = %+v, want healthy with original target released", recovered) + } + + originalTarget := prDiscoveryHealthTarget{Owner: forkOwner, Repo: forkRepo, Branch: forkPRBranch} + service.prDiscoveryHealth.mu.Lock() + originalState := service.prDiscoveryHealth.scopes[prDiscoveryScopeKey(testWorkspaceID, scope)].targets[originalTarget.key()] + service.prDiscoveryHealth.mu.Unlock() + if originalState != nil { + t.Fatalf("original fork target state = %#v, want released after rebind", originalState) + } +} diff --git a/apps/backend/internal/github/service_pr_discovery_attempt.go b/apps/backend/internal/github/service_pr_discovery_attempt.go new file mode 100644 index 0000000000..ce00903ed9 --- /dev/null +++ b/apps/backend/internal/github/service_pr_discovery_attempt.go @@ -0,0 +1,440 @@ +package github + +import ( + "context" + "errors" + "fmt" + "sync" + "time" +) + +type prDiscoveryWatchAttempt struct { + target prDiscoveryHealthTarget + number uint64 + joined bool + result *prDiscoveryAttemptResult +} + +type prDiscoveryAttemptResult struct { + done chan struct{} + once sync.Once + workspaceID string + targetKey string + status *PRStatus + resolved bool + syncFailed bool + err error + suppressFallback bool +} + +func newPRDiscoveryAttemptResult(workspaceID, targetKey string) *prDiscoveryAttemptResult { + return &prDiscoveryAttemptResult{done: make(chan struct{}), workspaceID: workspaceID, targetKey: targetKey} +} + +func (r *prDiscoveryAttemptResult) complete( + status *PRStatus, resolved, syncFailed bool, err error, suppressFallback bool, +) { + if r == nil { + return + } + r.once.Do(func() { + r.status = status + r.resolved = resolved + r.syncFailed = syncFailed + r.err = err + r.suppressFallback = suppressFallback + close(r.done) + }) +} + +func (r *prDiscoveryAttemptResult) wait(ctx context.Context) (*PRStatus, bool, bool, error, bool) { + if r == nil { + return nil, false, false, nil, false + } + select { + case <-r.done: + return r.status, r.resolved, r.syncFailed, r.err, r.suppressFallback + case <-ctx.Done(): + return nil, false, false, ctx.Err(), false + } +} + +func (r *prDiscoveryAttemptResult) invalidated() bool { + if r == nil { + return false + } + select { + case <-r.done: + return r.err == nil && r.syncFailed && r.suppressFallback + default: + return false + } +} + +func sharedPRDiscoveryResult(ctx context.Context, attempt prDiscoveryWatchAttempt) (*PR, error) { + status, _, _, err, suppressFallback := attempt.result.wait(ctx) + if err != nil { + if suppressFallback { + return nil, nil + } + return nil, err + } + if status == nil { + return nil, nil + } + return status.PR, nil +} + +type prDiscoveryRetryAtProvider interface { + ProviderRetryAt() *time.Time +} + +type prDiscoveryRetryAtError struct { + err error + retryAt *time.Time +} + +func (e *prDiscoveryRetryAtError) Error() string { + if e == nil || e.err == nil { + return "" + } + return e.err.Error() +} + +func (e *prDiscoveryRetryAtError) Unwrap() error { + if e == nil { + return nil + } + return e.err +} + +func (e *prDiscoveryRetryAtError) ProviderRetryAt() *time.Time { + if e == nil { + return nil + } + return copyTime(e.retryAt) +} + +func withPRDiscoveryRetryAt(err error, retryAt *time.Time) error { + if err == nil || retryAt == nil { + return err + } + return &prDiscoveryRetryAtError{err: err, retryAt: copyTime(retryAt)} +} + +func prDiscoveryRetryAtFromError(err error) *time.Time { + if err == nil { + return nil + } + var providerErr prDiscoveryRetryAtProvider + if errors.As(err, &providerErr) { + return providerErr.ProviderRetryAt() + } + return nil +} + +func (s *Service) ensurePRDiscoveryHealth() *prDiscoveryHealthStore { + if s == nil { + return nil + } + s.mu.Lock() + defer s.mu.Unlock() + if s.prDiscoveryHealth == nil { + s.prDiscoveryHealth = newPRDiscoveryHealth(s.eventBus, s.logger) + } + return s.prDiscoveryHealth +} + +func prDiscoveryTargetForWatch(watch *PRWatch) prDiscoveryHealthTarget { + if watch == nil { + return prDiscoveryHealthTarget{} + } + return prDiscoveryHealthTarget{ + Owner: watch.Owner, Repo: watch.Repo, Branch: watch.Branch, PRNumber: watch.PRNumber, + } +} + +func (s *Service) beginPRDiscoveryWatch( + workspaceID, cacheScope string, credentialGeneration int64, watch *PRWatch, +) (prDiscoveryWatchAttempt, bool) { + target := prDiscoveryTargetForWatch(watch) + attempt := prDiscoveryWatchAttempt{target: target} + health := s.ensurePRDiscoveryHealth() + if health == nil { + return attempt, true + } + s.prDiscoveryAttemptsMu.Lock() + invalidatedTargets, number, admitted, joined, eventHealth := health.trackConsumerAndBegin( + workspaceID, cacheScope, credentialGeneration, prDiscoveryWatchID(watch), target, + ) + s.invalidatePRDiscoveryAttemptsLocked(workspaceID, invalidatedTargets) + if s.prDiscoveryAttempts == nil { + s.prDiscoveryAttempts = make(map[string]*prDiscoveryAttemptResult) + } + attempt.number = number + if admitted && workspaceID != "" && cacheScope != "" { + key := prDiscoveryAttemptKey(workspaceID, cacheScope, credentialGeneration, target) + result := s.prDiscoveryAttempts[key] + if joined && result != nil { + attempt.joined = true + } else { + result = newPRDiscoveryAttemptResult(workspaceID, target.key()) + s.prDiscoveryAttempts[key] = result + } + attempt.result = result + } + s.prDiscoveryAttemptsMu.Unlock() + if eventHealth != nil { + health.publish(workspaceID, eventHealth) + } + return attempt, admitted +} + +func (h *prDiscoveryHealthStore) beginWithJoin( + workspaceID, cacheScope string, credentialGeneration int64, target prDiscoveryHealthTarget, +) (uint64, bool, bool) { + if h == nil || workspaceID == "" || cacheScope == "" { + return 0, true, false + } + h.mu.Lock() + defer h.mu.Unlock() + scopeKey := prDiscoveryScopeKey(workspaceID, cacheScope) + delete(h.retired, scopeKey) + scope := h.scopeLocked(workspaceID, cacheScope, credentialGeneration) + return h.beginWithJoinScopeLocked(scope, target) +} + +func (h *prDiscoveryHealthStore) beginWithJoinScopeLocked( + scope *prDiscoveryHealthScope, target prDiscoveryHealthTarget, +) (uint64, bool, bool) { + if scope == nil { + return 0, false, false + } + key := target.key() + state := scope.targets[key] + now := h.nowLocked() + if state != nil { + if state.active { + return state.attempt, true, true + } + if state.retryAt != nil && now.Before(*state.retryAt) { + return 0, false, false + } + } + if state == nil { + if !h.ensureTargetCapacityLocked(scope) { + return 0, false, false + } + state = newPRDiscoveryHealthTargetState() + scope.targets[key] = state + } + scope.nextAttempt++ + state.attempt = scope.nextAttempt + state.active = true + return state.attempt, true, false +} + +func (h *prDiscoveryHealthStore) trackConsumerAndBegin( + workspaceID, cacheScope string, credentialGeneration int64, watchID string, target prDiscoveryHealthTarget, +) ([]string, uint64, bool, bool, *PRDiscoveryHealth) { + if h == nil || workspaceID == "" || cacheScope == "" || watchID == "" { + return nil, 0, true, false, nil + } + key := target.key() + invalidatedTargets := make([]string, 0, 1) + h.mu.Lock() + scopeKey := prDiscoveryScopeKey(workspaceID, cacheScope) + delete(h.retired, scopeKey) + scope := h.scopeLocked(workspaceID, cacheScope, credentialGeneration) + if scope == nil { + h.mu.Unlock() + return nil, 0, false, false, nil + } + oldKey := scope.watchTargets[watchID] + changed, removed := detachPRDiscoveryConsumerLocked(scope, watchID, oldKey, key) + if removed { + invalidatedTargets = append(invalidatedTargets, oldKey) + } + state, ok := h.targetStateLocked(scope, key) + if !ok { + var eventHealth *PRDiscoveryHealth + if changed { + snapshot := h.snapshotLocked(scope) + eventHealth = &snapshot + } + h.mu.Unlock() + return invalidatedTargets, 0, false, false, eventHealth + } + if state.consumers == nil { + state.consumers = make(map[string]struct{}) + } + if _, exists := state.consumers[watchID]; !exists { + state.consumers[watchID] = struct{}{} + scope.watchTargets[watchID] = key + changed = true + } + if changed { + scope.revision++ + } + number, admitted, joined := h.beginWithJoinScopeLocked(scope, target) + var eventHealth *PRDiscoveryHealth + if changed { + snapshot := h.snapshotLocked(scope) + eventHealth = &snapshot + } + h.mu.Unlock() + return invalidatedTargets, number, admitted, joined, eventHealth +} + +func prDiscoveryAttemptKey( + workspaceID, cacheScope string, credentialGeneration int64, target prDiscoveryHealthTarget, +) string { + return fmt.Sprintf("%d:%s|%d:%s|%d|%s", len(workspaceID), workspaceID, len(cacheScope), cacheScope, credentialGeneration, target.key()) +} + +func (s *Service) completePRDiscoveryWatchAttempt( + attempt prDiscoveryWatchAttempt, status *PRStatus, resolved, syncFailed bool, err error, suppressFallback bool, +) { + if attempt.result == nil { + return + } + attempt.result.complete(status, resolved, syncFailed, err, suppressFallback) +} + +func (s *Service) forgetPRDiscoveryWatchAttempt( + workspaceID, cacheScope string, credentialGeneration int64, attempt prDiscoveryWatchAttempt, +) { + if s == nil || attempt.result == nil || workspaceID == "" || cacheScope == "" { + return + } + key := prDiscoveryAttemptKey(workspaceID, cacheScope, credentialGeneration, attempt.target) + s.prDiscoveryAttemptsMu.Lock() + if s.prDiscoveryAttempts[key] == attempt.result { + delete(s.prDiscoveryAttempts, key) + } + s.prDiscoveryAttemptsMu.Unlock() +} + +func (s *Service) trackPRDiscoveryWatchConsumer( + workspaceID, cacheScope string, credentialGeneration int64, watch *PRWatch, +) { + s.trackPRDiscoveryWatchTarget( + workspaceID, cacheScope, credentialGeneration, watch, prDiscoveryTargetForWatch(watch), + ) +} + +func (s *Service) trackPRDiscoveryWatchTarget( + workspaceID, cacheScope string, credentialGeneration int64, watch *PRWatch, + target prDiscoveryHealthTarget, +) { + if health := s.ensurePRDiscoveryHealth(); health != nil { + invalidatedTargets := health.trackConsumer( + workspaceID, cacheScope, credentialGeneration, prDiscoveryWatchID(watch), target, + ) + s.invalidatePRDiscoveryAttempts(workspaceID, invalidatedTargets) + } +} + +func (s *Service) finishPRDiscoveryWatchSuccess( + workspaceID, cacheScope string, credentialGeneration int64, attempt prDiscoveryWatchAttempt, +) { + if health := s.ensurePRDiscoveryHealth(); health != nil { + health.finishSuccess(workspaceID, cacheScope, credentialGeneration, attempt.target, attempt.number) + } +} + +func (s *Service) finishPRDiscoveryWatchFailure( + workspaceID, cacheScope string, credentialGeneration int64, attempt prDiscoveryWatchAttempt, err error, +) { + if health := s.ensurePRDiscoveryHealth(); health != nil { + health.finishFailure( + workspaceID, cacheScope, credentialGeneration, attempt.target, attempt.number, + classifyPRDiscoveryError(err), prDiscoveryRetryAtFromError(err), + ) + } +} + +func (s *Service) finishPRDiscoveryWatchFailureCategory( + workspaceID, cacheScope string, credentialGeneration int64, attempt prDiscoveryWatchAttempt, + category PRDiscoveryHealthCategory, +) { + if health := s.ensurePRDiscoveryHealth(); health != nil { + health.finishFailure( + workspaceID, cacheScope, credentialGeneration, attempt.target, attempt.number, category, nil, + ) + } +} + +func (s *Service) releasePRDiscoveryWatch( + workspaceID, cacheScope string, credentialGeneration int64, attempt prDiscoveryWatchAttempt, +) { + if health := s.ensurePRDiscoveryHealth(); health != nil { + health.release(workspaceID, cacheScope, credentialGeneration, attempt.target, attempt.number) + } +} + +func prDiscoveryWatchID(watch *PRWatch) string { + if watch == nil { + return "" + } + return watch.ID +} + +func (s *Service) removePRDiscoveryWatchConsumer(watch *PRWatch) { + if s == nil || watch == nil { + return + } + if health := s.ensurePRDiscoveryHealth(); health != nil { + invalidatedTargets := health.removeWatch(watch.WorkspaceID, watch.ID) + s.invalidatePRDiscoveryAttempts(watch.WorkspaceID, invalidatedTargets) + } +} + +func (s *Service) invalidatePRDiscoveryAttempts(workspaceID string, targetKeys []string) { + if s == nil { + return + } + s.prDiscoveryAttemptsMu.Lock() + s.invalidatePRDiscoveryAttemptsLocked(workspaceID, targetKeys) + s.prDiscoveryAttemptsMu.Unlock() +} + +func (s *Service) invalidatePRDiscoveryAttemptsLocked(workspaceID string, targetKeys []string) { + if s == nil || workspaceID == "" || len(targetKeys) == 0 { + return + } + wanted := make(map[string]struct{}, len(targetKeys)) + for _, targetKey := range targetKeys { + if targetKey != "" { + wanted[targetKey] = struct{}{} + } + } + if len(wanted) == 0 { + return + } + for key, result := range s.prDiscoveryAttempts { + if result.workspaceID != workspaceID { + continue + } + if _, ok := wanted[result.targetKey]; !ok { + continue + } + result.complete(nil, false, true, nil, true) + delete(s.prDiscoveryAttempts, key) + } +} + +func (s *Service) invalidateAllPRDiscoveryAttempts(workspaceID string) { + if s == nil { + return + } + s.prDiscoveryAttemptsMu.Lock() + for key, result := range s.prDiscoveryAttempts { + if workspaceID != "" && result.workspaceID != workspaceID { + continue + } + result.complete(nil, false, true, nil, true) + delete(s.prDiscoveryAttempts, key) + } + s.prDiscoveryAttemptsMu.Unlock() +} diff --git a/apps/backend/internal/github/service_pr_discovery_health.go b/apps/backend/internal/github/service_pr_discovery_health.go new file mode 100644 index 0000000000..c0af325a2f --- /dev/null +++ b/apps/backend/internal/github/service_pr_discovery_health.go @@ -0,0 +1,687 @@ +package github + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "time" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" +) + +const ( + PRDiscoveryRetryBase = time.Minute + prDiscoveryRetryMax = 15 * time.Minute + prDiscoveryMaxScopes = 256 + prDiscoveryMaxTargets = 4096 +) + +// Milliseconds remain exactly representable in JavaScript while preserving +// process-start ordering for stale HTTP/WS event rejection. +var prDiscoveryRuntimeEpochSeed = time.Now().UnixMilli() + +type PRDiscoveryHealthState string + +const ( + PRDiscoveryHealthUnknown PRDiscoveryHealthState = "unknown" + PRDiscoveryHealthHealthy PRDiscoveryHealthState = "healthy" + PRDiscoveryHealthDegraded PRDiscoveryHealthState = "degraded" +) + +type PRDiscoveryHealthCategory string + +const ( + PRDiscoveryHealthInvalidQuery PRDiscoveryHealthCategory = "invalid_query" + PRDiscoveryHealthRateLimited PRDiscoveryHealthCategory = "rate_limited" + PRDiscoveryHealthUnavailable PRDiscoveryHealthCategory = "unavailable" +) + +// PRDiscoveryHealth is the bounded, credential-scoped projection of the +// provider's PR discovery path. It intentionally contains categories and +// timestamps only; provider payloads and credentials never cross this boundary. +type PRDiscoveryHealth struct { + State PRDiscoveryHealthState `json:"state"` + FailedTargetCount int `json:"failed_target_count"` + LastFailureAt *time.Time `json:"last_failure_at,omitempty"` + Category PRDiscoveryHealthCategory `json:"category,omitempty"` + RetryAt *time.Time `json:"retry_at,omitempty"` + Revision uint64 `json:"revision"` + CredentialGeneration int64 `json:"credential_generation"` + RuntimeEpoch int64 `json:"runtime_epoch"` +} + +// PRDiscoveryHealthUpdatedEvent is workspace-scoped and carries the same +// revisioned projection returned by the status endpoint. +type PRDiscoveryHealthUpdatedEvent struct { + WorkspaceID string `json:"workspace_id"` + Health *PRDiscoveryHealth `json:"health"` +} + +func (e *PRDiscoveryHealthUpdatedEvent) GetWorkspaceID() string { + if e == nil { + return "" + } + return e.WorkspaceID +} + +type prDiscoveryHealthTarget struct { + Owner string + Repo string + Branch string + PRNumber int +} + +func (t prDiscoveryHealthTarget) key() string { + if t.PRNumber > 0 { + return fmt.Sprintf("pr:%s/%s#%d", strings.ToLower(t.Owner), strings.ToLower(t.Repo), t.PRNumber) + } + return fmt.Sprintf("branch:%s/%s@%s", strings.ToLower(t.Owner), strings.ToLower(t.Repo), t.Branch) +} + +type prDiscoveryHealthTargetState struct { + attempt uint64 + active bool + failureCount int + category PRDiscoveryHealthCategory + lastFailureAt time.Time + retryAt *time.Time + consumers map[string]struct{} +} + +type prDiscoveryHealthScope struct { + workspaceID string + cacheScope string + credentialGeneration int64 + nextAttempt uint64 + revision uint64 + seen bool + targets map[string]*prDiscoveryHealthTargetState + watchTargets map[string]string + runtimeEpoch int64 +} + +type prDiscoveryHealthCapacityBlock struct { + workspaceID string + cacheScope string + blockedAt time.Time +} + +type prDiscoveryHealthStore struct { + mu sync.Mutex + now func() time.Time + bus bus.EventBus + logger *logger.Logger + scopes map[string]*prDiscoveryHealthScope + capacity map[string]prDiscoveryHealthCapacityBlock + retired map[string]bool + runtimeEpoch int64 +} + +func newPRDiscoveryHealth(eventBus bus.EventBus, log *logger.Logger) *prDiscoveryHealthStore { + return &prDiscoveryHealthStore{ + now: time.Now, + bus: eventBus, + logger: log, + scopes: make(map[string]*prDiscoveryHealthScope), + capacity: make(map[string]prDiscoveryHealthCapacityBlock), + retired: make(map[string]bool), + runtimeEpoch: atomic.AddInt64(&prDiscoveryRuntimeEpochSeed, 1), + } +} + +func (h *prDiscoveryHealthStore) setNow(now func() time.Time) { + if h == nil || now == nil { + return + } + h.mu.Lock() + h.now = now + h.mu.Unlock() +} + +func (h *prDiscoveryHealthStore) nowLocked() time.Time { + if h.now == nil { + return time.Now().UTC() + } + return h.now().UTC() +} + +func prDiscoveryScopeKey(workspaceID, cacheScope string) string { + return fmt.Sprintf("%d:%s|%d:%s", len(workspaceID), workspaceID, len(cacheScope), cacheScope) +} + +func (h *prDiscoveryHealthStore) scopeLocked(workspaceID, cacheScope string, credentialGeneration int64) *prDiscoveryHealthScope { + key := prDiscoveryScopeKey(workspaceID, cacheScope) + scope := h.scopes[key] + if scope == nil { + if !h.ensureScopeCapacityLocked() { + h.capacity[key] = prDiscoveryHealthCapacityBlock{ + workspaceID: workspaceID, + cacheScope: cacheScope, + blockedAt: h.nowLocked(), + } + if h.logger != nil { + h.logger.Warn("PR discovery health scope capacity exhausted", + zap.String("workspace_id", workspaceID), zap.String("cache_scope", cacheScope)) + } + return nil + } + delete(h.capacity, key) + scope = &prDiscoveryHealthScope{ + workspaceID: workspaceID, + cacheScope: cacheScope, + targets: make(map[string]*prDiscoveryHealthTargetState), + watchTargets: make(map[string]string), + runtimeEpoch: atomic.AddInt64(&prDiscoveryRuntimeEpochSeed, 1), + } + h.scopes[key] = scope + } + if credentialGeneration != 0 { + scope.credentialGeneration = credentialGeneration + } + return scope +} + +func (h *prDiscoveryHealthStore) ensureScopeCapacityLocked() bool { + if len(h.scopes) < prDiscoveryMaxScopes { + return true + } + for candidateKey, candidate := range h.scopes { + if !prDiscoveryScopeHasLiveTargets(candidate) { + delete(h.scopes, candidateKey) + return true + } + } + return false +} + +func prDiscoveryScopeHasLiveTargets(scope *prDiscoveryHealthScope) bool { + if scope == nil { + return false + } + for _, state := range scope.targets { + if state != nil && (state.active || len(state.consumers) > 0) { + return true + } + } + return false +} + +func newPRDiscoveryHealthTargetState() *prDiscoveryHealthTargetState { + return &prDiscoveryHealthTargetState{consumers: make(map[string]struct{})} +} + +func (h *prDiscoveryHealthStore) ensureTargetCapacityLocked(scope *prDiscoveryHealthScope) bool { + if scope == nil { + return false + } + if len(scope.targets) < prDiscoveryMaxTargets { + return true + } + for key, state := range scope.targets { + if state == nil || (!state.active && len(state.consumers) == 0) { + delete(scope.targets, key) + return true + } + } + return false +} + +func prDiscoveryTargetHasFailure(state *prDiscoveryHealthTargetState) bool { + return state != nil && (state.category != "" || !state.lastFailureAt.IsZero()) +} + +func (h *prDiscoveryHealthStore) begin( + workspaceID, cacheScope string, credentialGeneration int64, target prDiscoveryHealthTarget, +) (uint64, bool) { + attempt, admitted, _ := h.beginWithJoin(workspaceID, cacheScope, credentialGeneration, target) + return attempt, admitted +} + +// trackConsumer records the live watch identity for a target. It moves a +// watch between target keys when a searching branch changes and invalidates +// any active attempt that no longer has a live consumer. +func (h *prDiscoveryHealthStore) trackConsumer( + workspaceID, cacheScope string, credentialGeneration int64, watchID string, target prDiscoveryHealthTarget, +) []string { + if h == nil || workspaceID == "" || cacheScope == "" || watchID == "" { + return nil + } + key := target.key() + invalidatedTargets := make([]string, 0, 1) + var eventHealth *PRDiscoveryHealth + h.mu.Lock() + scopeKey := prDiscoveryScopeKey(workspaceID, cacheScope) + delete(h.retired, scopeKey) + scope := h.scopeLocked(workspaceID, cacheScope, credentialGeneration) + if scope == nil { + h.mu.Unlock() + return nil + } + oldKey := scope.watchTargets[watchID] + changed, removed := detachPRDiscoveryConsumerLocked(scope, watchID, oldKey, key) + if removed { + invalidatedTargets = append(invalidatedTargets, oldKey) + } + state, ok := h.targetStateLocked(scope, key) + if !ok { + h.mu.Unlock() + if changed { + h.publish(workspaceID, h.snapshot(workspaceID, cacheScope, credentialGeneration)) + } + return invalidatedTargets + } + if state.consumers == nil { + state.consumers = make(map[string]struct{}) + } + if _, exists := state.consumers[watchID]; !exists { + state.consumers[watchID] = struct{}{} + scope.watchTargets[watchID] = key + } + if changed { + scope.revision++ + snapshot := h.snapshotLocked(scope) + eventHealth = &snapshot + } + h.mu.Unlock() + if eventHealth != nil { + h.publish(workspaceID, eventHealth) + } + return invalidatedTargets +} + +func detachPRDiscoveryConsumerLocked(scope *prDiscoveryHealthScope, watchID, oldKey, newKey string) (bool, bool) { + if oldKey == "" || oldKey == newKey { + return false, false + } + state := scope.targets[oldKey] + if state == nil { + delete(scope.watchTargets, watchID) + return false, false + } + wasFailed := prDiscoveryTargetHasFailure(state) + delete(state.consumers, watchID) + removed := false + if len(state.consumers) == 0 { + delete(scope.targets, oldKey) + removed = true + } + delete(scope.watchTargets, watchID) + return wasFailed, removed +} + +func (h *prDiscoveryHealthStore) targetStateLocked(scope *prDiscoveryHealthScope, key string) (*prDiscoveryHealthTargetState, bool) { + state := scope.targets[key] + if state != nil { + return state, true + } + if !h.ensureTargetCapacityLocked(scope) { + return nil, false + } + state = newPRDiscoveryHealthTargetState() + scope.targets[key] = state + return state, true +} + +// removeWatch removes a watch from every credential scope for its workspace. +// A deleted or archived watch must not keep a failed target degraded forever. +func (h *prDiscoveryHealthStore) removeWatch(workspaceID, watchID string) []string { + if h == nil || workspaceID == "" || watchID == "" { + return nil + } + type publication struct { + workspaceID string + health *PRDiscoveryHealth + } + publications := make([]publication, 0) + invalidatedTargets := make([]string, 0, 1) + h.mu.Lock() + for _, scope := range h.scopes { + if scope.workspaceID != workspaceID { + continue + } + key, ok := scope.watchTargets[watchID] + if !ok { + continue + } + delete(scope.watchTargets, watchID) + state := scope.targets[key] + if state == nil { + continue + } + wasFailed := prDiscoveryTargetHasFailure(state) + delete(state.consumers, watchID) + if len(state.consumers) == 0 { + delete(scope.targets, key) + invalidatedTargets = append(invalidatedTargets, key) + } + if wasFailed { + scope.revision++ + snapshot := h.snapshotLocked(scope) + publications = append(publications, publication{workspaceID: workspaceID, health: &snapshot}) + } + } + h.mu.Unlock() + for _, item := range publications { + h.publish(item.workspaceID, item.health) + } + return invalidatedTargets +} + +func (h *prDiscoveryHealthStore) finishSuccess( + workspaceID, cacheScope string, credentialGeneration int64, target prDiscoveryHealthTarget, attempt uint64, +) { + if h == nil || workspaceID == "" || cacheScope == "" { + return + } + var eventHealth *PRDiscoveryHealth + h.mu.Lock() + if h.retired[prDiscoveryScopeKey(workspaceID, cacheScope)] { + h.mu.Unlock() + return + } + scope := h.scopes[prDiscoveryScopeKey(workspaceID, cacheScope)] + if scope == nil { + h.mu.Unlock() + return + } + state, ok := scope.targets[target.key()] + if attempt != 0 && (!ok || !state.active || state.attempt != attempt) { + h.mu.Unlock() + return + } + if ok { + wasFailed := prDiscoveryTargetHasFailure(state) + state.active = false + state.failureCount = 0 + state.category = "" + state.lastFailureAt = time.Time{} + state.retryAt = nil + if len(state.consumers) == 0 { + delete(scope.targets, target.key()) + } + ok = wasFailed + } + changed := !scope.seen || ok + scope.seen = true + if changed { + scope.revision++ + snapshot := h.snapshotLocked(scope) + eventHealth = &snapshot + } + h.mu.Unlock() + if eventHealth != nil { + h.publish(workspaceID, eventHealth) + } +} + +func (h *prDiscoveryHealthStore) release( + workspaceID, cacheScope string, credentialGeneration int64, target prDiscoveryHealthTarget, attempt uint64, +) { + if h == nil || workspaceID == "" || cacheScope == "" || attempt == 0 { + return + } + h.mu.Lock() + defer h.mu.Unlock() + if h.retired[prDiscoveryScopeKey(workspaceID, cacheScope)] { + return + } + scope := h.scopes[prDiscoveryScopeKey(workspaceID, cacheScope)] + if scope == nil { + return + } + state := scope.targets[target.key()] + if state != nil && state.active && state.attempt == attempt { + state.active = false + if len(state.consumers) == 0 { + delete(scope.targets, target.key()) + } + } +} + +func (h *prDiscoveryHealthStore) finishFailure( + workspaceID, cacheScope string, credentialGeneration int64, target prDiscoveryHealthTarget, + attempt uint64, category PRDiscoveryHealthCategory, retryAt *time.Time, +) { + if h == nil || workspaceID == "" || cacheScope == "" { + return + } + if category == "" { + category = PRDiscoveryHealthUnavailable + } + var eventHealth *PRDiscoveryHealth + h.mu.Lock() + if h.retired[prDiscoveryScopeKey(workspaceID, cacheScope)] { + h.mu.Unlock() + return + } + scope := h.scopes[prDiscoveryScopeKey(workspaceID, cacheScope)] + if scope == nil { + h.mu.Unlock() + return + } + state := scope.targets[target.key()] + if state == nil { + // A non-zero attempt with no state means the watch was removed or + // retargeted while the provider request was in flight. Do not recreate + // a degraded target for a dead watch. + if attempt != 0 { + h.mu.Unlock() + return + } + state = newPRDiscoveryHealthTargetState() + scope.targets[target.key()] = state + } + if attempt != 0 && (!state.active || state.attempt != attempt) { + h.mu.Unlock() + return + } + if stalePRDiscoveryFailure(state, attempt) { + h.mu.Unlock() + return + } + now := h.nowLocked() + state.attempt = attempt + state.active = false + state.failureCount++ + state.category = category + state.lastFailureAt = now + state.retryAt = prDiscoveryRetryDeadline(now, category, state.failureCount, retryAt) + scope.seen = true + scope.revision++ + snapshot := h.snapshotLocked(scope) + eventHealth = &snapshot + h.mu.Unlock() + h.publish(workspaceID, eventHealth) +} + +func stalePRDiscoveryFailure(state *prDiscoveryHealthTargetState, attempt uint64) bool { + if state == nil || attempt == 0 { + return false + } + return (state.active && state.attempt != attempt) || (!state.active && state.attempt > attempt) +} + +func prDiscoveryRetryDeadline( + now time.Time, category PRDiscoveryHealthCategory, failureCount int, retryAt *time.Time, +) *time.Time { + if retryAt != nil && retryAt.After(now) { + deadline := now.Add(prDiscoveryRetryMax) + if retryAt.Before(deadline) { + return copyTime(retryAt) + } + return &deadline + } + delay := PRDiscoveryRetryBase + if category == PRDiscoveryHealthRateLimited { + delay = prDiscoveryRateLimitDelay(failureCount) + } + deadline := now.Add(delay) + return &deadline +} + +func prDiscoveryRateLimitDelay(failureCount int) time.Duration { + delay := PRDiscoveryRetryBase + for i := 1; i < failureCount; i++ { + if delay >= prDiscoveryRetryMax/2 { + return prDiscoveryRetryMax + } + delay *= 2 + } + return delay +} + +func (h *prDiscoveryHealthStore) snapshot( + workspaceID, cacheScope string, credentialGeneration int64, +) *PRDiscoveryHealth { + if h == nil { + return &PRDiscoveryHealth{State: PRDiscoveryHealthUnknown, CredentialGeneration: credentialGeneration} + } + h.mu.Lock() + defer h.mu.Unlock() + if h.retired[prDiscoveryScopeKey(workspaceID, cacheScope)] { + return &PRDiscoveryHealth{ + State: PRDiscoveryHealthUnknown, CredentialGeneration: credentialGeneration, RuntimeEpoch: h.runtimeEpoch, + } + } + scope := h.scopes[prDiscoveryScopeKey(workspaceID, cacheScope)] + if scope == nil { + if blocked, ok := h.capacity[prDiscoveryScopeKey(workspaceID, cacheScope)]; ok { + retryAt := blocked.blockedAt.Add(PRDiscoveryRetryBase) + return &PRDiscoveryHealth{ + State: PRDiscoveryHealthDegraded, + FailedTargetCount: 1, + LastFailureAt: copyTime(&blocked.blockedAt), + Category: PRDiscoveryHealthUnavailable, + RetryAt: &retryAt, + CredentialGeneration: credentialGeneration, + RuntimeEpoch: h.runtimeEpoch, + } + } + return &PRDiscoveryHealth{ + State: PRDiscoveryHealthUnknown, CredentialGeneration: credentialGeneration, RuntimeEpoch: h.runtimeEpoch, + } + } + if credentialGeneration != 0 { + scope.credentialGeneration = credentialGeneration + } + snapshot := h.snapshotLocked(scope) + return &snapshot +} + +func (h *prDiscoveryHealthStore) snapshotLocked(scope *prDiscoveryHealthScope) PRDiscoveryHealth { + health := PRDiscoveryHealth{ + State: PRDiscoveryHealthUnknown, + Revision: scope.revision, + CredentialGeneration: scope.credentialGeneration, + RuntimeEpoch: scope.runtimeEpoch, + } + var latestAttempt uint64 + for _, state := range scope.targets { + if state.category == "" || state.lastFailureAt.IsZero() { + continue + } + health.FailedTargetCount++ + if health.LastFailureAt == nil || state.lastFailureAt.After(*health.LastFailureAt) || + (state.lastFailureAt.Equal(*health.LastFailureAt) && state.attempt > latestAttempt) { + lastFailure := state.lastFailureAt + health.LastFailureAt = &lastFailure + health.Category = state.category + latestAttempt = state.attempt + } + if state.retryAt != nil && (health.RetryAt == nil || state.retryAt.Before(*health.RetryAt)) { + health.RetryAt = copyTime(state.retryAt) + } + } + if health.FailedTargetCount > 0 { + health.State = PRDiscoveryHealthDegraded + } else if scope.seen { + health.State = PRDiscoveryHealthHealthy + } + return health +} + +func (h *prDiscoveryHealthStore) clearWorkspace(workspaceID string) { + if h == nil || workspaceID == "" { + return + } + h.mu.Lock() + for key, scope := range h.scopes { + if scope.workspaceID == workspaceID { + delete(h.scopes, key) + h.retired[key] = true + } + } + for key, blocked := range h.capacity { + if blocked.workspaceID == workspaceID { + delete(h.capacity, key) + } + } + h.mu.Unlock() +} + +func (h *prDiscoveryHealthStore) clearAll() { + if h == nil { + return + } + h.mu.Lock() + h.scopes = make(map[string]*prDiscoveryHealthScope) + h.capacity = make(map[string]prDiscoveryHealthCapacityBlock) + h.retired = make(map[string]bool) + h.mu.Unlock() +} + +func (h *prDiscoveryHealthStore) publish(workspaceID string, health *PRDiscoveryHealth) { + if h == nil || h.bus == nil || health == nil { + return + } + eventData := &PRDiscoveryHealthUpdatedEvent{WorkspaceID: workspaceID, Health: health} + event := bus.NewEvent(events.GitHubPRDiscoveryHealthUpdated, "github", eventData) + if err := h.bus.Publish(context.Background(), events.GitHubPRDiscoveryHealthUpdated, event); err != nil && h.logger != nil { + h.logger.Debug("publish PR discovery health event failed", zap.Error(err)) + } +} + +func copyTime(value *time.Time) *time.Time { + if value == nil { + return nil + } + copy := *value + return © +} + +func classifyPRDiscoveryError(err error) PRDiscoveryHealthCategory { + if err == nil { + return "" + } + var apiErr *GitHubAPIError + if errors.As(err, &apiErr) && (apiErr.StatusCode == 429 || containsRateLimitMarker(apiErr.Body)) { + return PRDiscoveryHealthRateLimited + } + lower := strings.ToLower(err.Error()) + if strings.Contains(lower, "graphql") && (strings.Contains(lower, "validation") || + strings.Contains(lower, "doesn't exist") || strings.Contains(lower, "does not exist") || + strings.Contains(lower, "cannot query field") || strings.Contains(lower, "unknown argument")) { + return PRDiscoveryHealthInvalidQuery + } + if containsRateLimitMarker(lower) { + return PRDiscoveryHealthRateLimited + } + return PRDiscoveryHealthUnavailable +} + +func containsRateLimitMarker(value string) bool { + lower := strings.ToLower(value) + for _, marker := range []string{"rate limit", "too many requests", "429", "abuse detection", "secondary limit"} { + if strings.Contains(lower, marker) { + return true + } + } + return false +} diff --git a/apps/backend/internal/github/service_pr_discovery_health_test.go b/apps/backend/internal/github/service_pr_discovery_health_test.go new file mode 100644 index 0000000000..6d3089a7d4 --- /dev/null +++ b/apps/backend/internal/github/service_pr_discovery_health_test.go @@ -0,0 +1,406 @@ +package github + +import ( + "context" + "fmt" + "testing" + "time" +) + +func TestPRDiscoveryHealth(t *testing.T) { + _, service, client, store := setupBatchedPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-health", false) + watch := withTestWorkspace(&PRWatch{ + SessionID: "session-health", + TaskID: "task-health", + Owner: "o", + Repo: "r", + Branch: "feature/health", + }) + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create PR watch: %v", err) + } + + fixedNow := time.Date(2026, 9, 11, 12, 0, 0, 0, time.UTC) + service.prDiscoveryHealth.setNow(func() time.Time { return fixedNow }) + scope := testAutomationScope(t, service, testWorkspaceID) + client.branchResponses = []string{ + `{"errors":[{"type":"GRAPHQL_VALIDATION_FAILED","message":"Field 'cloneUrl' doesn't exist on type 'Repository'"}]}`, + branchAliasResponse(t, 1, map[int]string{0: openPRNode(9, "feature/health", "alice")}), + } + + if _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watch}); err != nil { + t.Fatalf("health-aware discovery: %v", err) + } + failed := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if failed.State != PRDiscoveryHealthDegraded || failed.Category != PRDiscoveryHealthInvalidQuery { + t.Fatalf("health after rejected discovery = %+v, want degraded invalid_query", failed) + } + if failed.FailedTargetCount != 1 || failed.RetryAt == nil || !failed.RetryAt.Equal(fixedNow.Add(PRDiscoveryRetryBase)) { + t.Fatalf("health retry state = %+v, want one target and retry at %v", failed, fixedNow.Add(PRDiscoveryRetryBase)) + } + if got := len(client.branchQueries); got != 1 { + t.Fatalf("rejected discovery transport calls = %d, want 1", got) + } + + // A full quota refresh is independent evidence. It must not erase the + // discovery failure or make the rejected query eligible before its deadline. + recordRateLimitResources(service.rateTracker, rateLimitResources{ + Core: rateLimitBucket{Limit: 5000, Remaining: 5000, Reset: fixedNow.Add(time.Hour).Unix()}, + GraphQL: rateLimitBucket{Limit: 5000, Remaining: 5000, Reset: fixedNow.Add(time.Hour).Unix()}, + Search: rateLimitBucket{Limit: 30, Remaining: 30, Reset: fixedNow.Add(time.Hour).Unix()}, + }) + if _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watch}); err != nil { + t.Fatalf("suppressed discovery retry: %v", err) + } + stillFailed := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if stillFailed.State != PRDiscoveryHealthDegraded || stillFailed.Category != PRDiscoveryHealthInvalidQuery { + t.Fatalf("health after suppressed retry = %+v, want unchanged degradation", stillFailed) + } + if got := len(client.branchQueries); got != 1 { + t.Fatalf("suppressed discovery transport calls = %d, want one coalesced attempt", got) + } + + fixedNow = fixedNow.Add(PRDiscoveryRetryBase) + if _, err := service.SyncWorkspaceWatchesBatched(ctx, testWorkspaceID, []*PRWatch{watch}); err != nil { + t.Fatalf("discovery after retry deadline: %v", err) + } + recovered := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if recovered.State != PRDiscoveryHealthHealthy || recovered.FailedTargetCount != 0 { + t.Fatalf("health after newer successful discovery = %+v, want healthy", recovered) + } + if got := len(client.branchQueries); got != 2 { + t.Fatalf("recovered discovery transport calls = %d, want two total attempts", got) + } +} + +func TestPRDiscoveryHealthStaleCompletionAndCredentialIsolation(t *testing.T) { + fixedNow := time.Date(2026, 9, 11, 12, 0, 0, 0, time.UTC) + health := newPRDiscoveryHealth(nil, nil) + health.setNow(func() time.Time { return fixedNow }) + target := prDiscoveryHealthTarget{Owner: "o", Repo: "r", Branch: "feature/health"} + + first, admitted := health.begin("workspace-a", "credential-a", 1, target) + if !admitted { + t.Fatal("first target attempt was not admitted") + } + health.finishFailure("workspace-a", "credential-a", 1, target, first, PRDiscoveryHealthInvalidQuery, nil) + fixedNow = fixedNow.Add(PRDiscoveryRetryBase) + second, admitted := health.begin("workspace-a", "credential-a", 1, target) + if !admitted { + t.Fatal("target was not admitted after its retry deadline") + } + health.finishSuccess("workspace-a", "credential-a", 1, target, first) + stale := health.snapshot("workspace-a", "credential-a", 1) + if stale.State != PRDiscoveryHealthDegraded || stale.Category != PRDiscoveryHealthInvalidQuery { + t.Fatalf("stale completion cleared newer failure: %+v", stale) + } + health.finishSuccess("workspace-a", "credential-a", 1, target, second) + if recovered := health.snapshot("workspace-a", "credential-a", 1); recovered.State != PRDiscoveryHealthHealthy { + t.Fatalf("newer completion did not recover target: %+v", recovered) + } + + otherCredential, admitted := health.begin("workspace-a", "credential-b", 2, target) + if !admitted { + t.Fatal("replacement credential was not admitted independently") + } + health.finishFailure("workspace-a", "credential-b", 2, target, otherCredential, PRDiscoveryHealthUnavailable, nil) + if isolated := health.snapshot("workspace-a", "credential-a", 1); isolated.State != PRDiscoveryHealthHealthy { + t.Fatalf("replacement credential affected old health scope: %+v", isolated) + } + if replacement := health.snapshot("workspace-a", "credential-b", 2); replacement.State != PRDiscoveryHealthDegraded { + t.Fatalf("replacement credential lost its own failure: %+v", replacement) + } +} + +func TestPRDiscoveryHealthScopeRecreationGetsNewRuntimeEpoch(t *testing.T) { + health := newPRDiscoveryHealth(nil, nil) + target := prDiscoveryHealthTarget{Owner: "o", Repo: "r", Branch: "feature/epoch"} + first, admitted := health.begin("workspace-epoch", "credential-epoch", 1, target) + if !admitted { + t.Fatal("initial epoch target was not admitted") + } + health.finishSuccess("workspace-epoch", "credential-epoch", 1, target, first) + firstSnapshot := health.snapshot("workspace-epoch", "credential-epoch", 1) + health.clearWorkspace("workspace-epoch") + second, admitted := health.begin("workspace-epoch", "credential-epoch", 1, target) + if !admitted { + t.Fatal("recreated epoch target was not admitted") + } + health.finishSuccess("workspace-epoch", "credential-epoch", 1, target, second) + secondSnapshot := health.snapshot("workspace-epoch", "credential-epoch", 1) + if secondSnapshot.RuntimeEpoch <= firstSnapshot.RuntimeEpoch { + t.Fatalf("scope runtime epoch = %d after recreation, want greater than %d", secondSnapshot.RuntimeEpoch, firstSnapshot.RuntimeEpoch) + } +} + +func TestClassifyPRDiscoveryError(t *testing.T) { + tests := []struct { + name string + err error + want PRDiscoveryHealthCategory + }{ + { + name: "HTTP 200 GraphQL validation", + err: errString("graphql error: Field 'cloneUrl' doesn't exist on type 'Repository'"), + want: PRDiscoveryHealthInvalidQuery, + }, + { + name: "rate limit", + err: errString("HTTP 429: API rate limit exceeded"), + want: PRDiscoveryHealthRateLimited, + }, + { + name: "transport unavailable", + err: errString("dial tcp: connection refused"), + want: PRDiscoveryHealthUnavailable, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := classifyPRDiscoveryError(tt.err); got != tt.want { + t.Fatalf("classifyPRDiscoveryError() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestPRDiscoveryHealth_PreservesGraphQLRateLimitResetDeadline(t *testing.T) { + fixedNow := time.Date(2026, 9, 11, 12, 0, 0, 0, time.UTC) + resetAt := fixedNow.Add(10 * time.Minute) + executor := &stubGraphQLExecutor{response: `{"data":{"rateLimit":{"limit":5000,"remaining":0,"resetAt":"2026-09-11T12:10:00Z","cost":1}},"errors":[{"type":"RATE_LIMITED","message":"API rate limit exceeded"}]}`} + _, err := runBatchedBranchQuery(context.Background(), executor, []graphQLBranchRef{{Owner: "o", Repo: "r", Branch: "feature/rate"}}) + if err == nil { + t.Fatal("runBatchedBranchQuery succeeded, want GraphQL rate-limit error") + } + providerRetryAt := prDiscoveryRetryAtFromError(err) + if providerRetryAt == nil || !providerRetryAt.Equal(resetAt) { + t.Fatalf("GraphQL retry deadline = %v, want %v", providerRetryAt, resetAt) + } + + health := newPRDiscoveryHealth(nil, nil) + health.setNow(func() time.Time { return fixedNow }) + target := prDiscoveryHealthTarget{Owner: "o", Repo: "r", Branch: "feature/rate"} + attempt, admitted := health.begin("workspace-rate", "credential-rate", 1, target) + if !admitted { + t.Fatal("rate-limited target was not admitted") + } + health.finishFailure("workspace-rate", "credential-rate", 1, target, attempt, PRDiscoveryHealthRateLimited, providerRetryAt) + fixedNow = fixedNow.Add(2 * time.Minute) + if _, admitted = health.begin("workspace-rate", "credential-rate", 1, target); admitted { + t.Fatal("target was admitted before provider reset deadline") + } +} + +func TestPRDiscoveryHealth_DoesNotUsePositiveGraphQLRemainingAsRetryEvidence(t *testing.T) { + executor := &stubGraphQLExecutor{response: `{"data":{"rateLimit":{"limit":5000,"remaining":4999,"resetAt":"2030-09-11T12:10:00Z","cost":1}},"errors":[{"type":"GRAPHQL_VALIDATION_FAILED","message":"schema mismatch"}]}`} + _, err := runBatchedBranchQuery(context.Background(), executor, []graphQLBranchRef{{Owner: "o", Repo: "r", Branch: "feature/positive"}}) + if err == nil { + t.Fatal("runBatchedBranchQuery succeeded, want GraphQL error") + } + if retryAt := prDiscoveryRetryAtFromError(err); retryAt != nil { + t.Fatalf("GraphQL retry deadline = %v, want no quota deadline with positive remaining", retryAt) + } +} + +func TestPRDiscoveryHealth_ClampsFarFutureProviderRetryDeadline(t *testing.T) { + fixedNow := time.Date(2026, 9, 11, 12, 0, 0, 0, time.UTC) + health := newPRDiscoveryHealth(nil, nil) + health.setNow(func() time.Time { return fixedNow }) + target := prDiscoveryHealthTarget{Owner: "o", Repo: "r", Branch: "feature/far-future"} + attempt, admitted := health.begin("workspace-far-future", "credential-far-future", 1, target) + if !admitted { + t.Fatal("far-future target was not admitted") + } + providerRetryAt := fixedNow.Add(24 * time.Hour) + health.finishFailure("workspace-far-future", "credential-far-future", 1, target, attempt, PRDiscoveryHealthRateLimited, &providerRetryAt) + snapshot := health.snapshot("workspace-far-future", "credential-far-future", 1) + if snapshot.RetryAt == nil || !snapshot.RetryAt.Equal(fixedNow.Add(prDiscoveryRetryMax)) { + t.Fatalf("far-future retry deadline = %v, want %v", snapshot.RetryAt, fixedNow.Add(prDiscoveryRetryMax)) + } +} + +func TestPRDiscoveryHealth_CapacityReportsDegradedWithoutEvictingLiveScopes(t *testing.T) { + health := newPRDiscoveryHealth(nil, nil) + target := prDiscoveryHealthTarget{Owner: "o", Repo: "r", Branch: "feature/capacity"} + for index := 0; index < prDiscoveryMaxScopes; index++ { + workspaceID := fmt.Sprintf("workspace-capacity-%d", index) + if _, admitted := health.begin(workspaceID, "credential", 1, target); !admitted { + t.Fatalf("live scope %s was not admitted", workspaceID) + } + } + blockedWorkspace := "workspace-capacity-blocked" + if _, admitted := health.begin(blockedWorkspace, "credential", 1, target); admitted { + t.Fatal("capacity-blocked scope was admitted by evicting a live scope") + } + blocked := health.snapshot(blockedWorkspace, "credential", 1) + if blocked.State != PRDiscoveryHealthDegraded || blocked.FailedTargetCount != 1 || blocked.Category != PRDiscoveryHealthUnavailable { + t.Fatalf("capacity projection = %+v, want degraded unavailable state", blocked) + } + if _, admitted := health.begin("workspace-capacity-0", "credential", 1, target); !admitted { + t.Fatal("capacity admission displaced a live scope") + } +} + +func TestPRDiscoveryHealth_RemovesFailedTargetAfterLastWatchConsumer(t *testing.T) { + _, service, _, store := setupBatchedPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-health-consumers", false) + watchA := withTestWorkspace(&PRWatch{ + SessionID: "session-health-consumer-a", TaskID: "task-health-consumers", Owner: "o", Repo: "r", + Branch: "feature/health-consumers", + }) + watchB := withTestWorkspace(&PRWatch{ + SessionID: "session-health-consumer-b", TaskID: "task-health-consumers", Owner: "o", Repo: "r", + Branch: "feature/health-consumers", + }) + for _, watch := range []*PRWatch{watchA, watchB} { + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create watch: %v", err) + } + } + scope := testAutomationScope(t, service, testWorkspaceID) + failWatch := func(watch *PRWatch) { + attempt, admitted := service.beginPRDiscoveryWatch(testWorkspaceID, scope, 1, watch) + if !admitted { + t.Fatalf("watch %s was not admitted", watch.ID) + } + service.prDiscoveryHealth.finishFailure( + testWorkspaceID, scope, 1, attempt.target, attempt.number, PRDiscoveryHealthUnavailable, nil, + ) + } + failWatch(watchA) + // The second watch joins the same target, so one failed target remains. + service.trackPRDiscoveryWatchConsumer(testWorkspaceID, scope, 1, watchB) + failed := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if failed.FailedTargetCount != 1 { + t.Fatalf("shared failed target count = %d, want 1", failed.FailedTargetCount) + } + + if err := service.DeletePRWatch(ctx, watchA.ID); err != nil { + t.Fatalf("delete first consumer: %v", err) + } + stillFailed := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if stillFailed.FailedTargetCount != 1 { + t.Fatalf("health after deleting one shared consumer = %+v, want degraded", stillFailed) + } + if err := service.DeletePRWatch(ctx, watchB.ID); err != nil { + t.Fatalf("delete last consumer: %v", err) + } + recovered := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if recovered.State != PRDiscoveryHealthHealthy || recovered.FailedTargetCount != 0 || recovered.Revision <= failed.Revision { + t.Fatalf("health after deleting last consumer = %+v, want newer healthy projection", recovered) + } +} + +func TestPRDiscoveryHealth_PreservesMixedLiveTargets(t *testing.T) { + _, service, _, store := setupBatchedPollerTest(t) + ctx := context.Background() + seedTask(t, store, "task-health-mixed", false) + watchA := withTestWorkspace(&PRWatch{ + SessionID: "session-health-mixed-a", TaskID: "task-health-mixed", Owner: "o", Repo: "r", + Branch: "feature/mixed-a", + }) + watchB := withTestWorkspace(&PRWatch{ + SessionID: "session-health-mixed-b", TaskID: "task-health-mixed", Owner: "o", Repo: "r", + Branch: "feature/mixed-b", + }) + for _, watch := range []*PRWatch{watchA, watchB} { + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create mixed watch: %v", err) + } + } + scope := testAutomationScope(t, service, testWorkspaceID) + for _, watch := range []*PRWatch{watchA, watchB} { + attempt, admitted := service.beginPRDiscoveryWatch(testWorkspaceID, scope, 1, watch) + if !admitted { + t.Fatalf("watch %s was not admitted", watch.ID) + } + service.prDiscoveryHealth.finishFailure( + testWorkspaceID, scope, 1, attempt.target, attempt.number, PRDiscoveryHealthUnavailable, nil, + ) + } + failed := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if failed.FailedTargetCount != 2 { + t.Fatalf("mixed target health = %+v, want two failed targets", failed) + } + + if err := service.DeletePRWatch(ctx, watchA.ID); err != nil { + t.Fatalf("delete first mixed target: %v", err) + } + remaining := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if remaining.State != PRDiscoveryHealthDegraded || remaining.FailedTargetCount != 1 { + t.Fatalf("health after deleting one mixed target = %+v, want one remaining failure", remaining) + } + if _, admitted := service.beginPRDiscoveryWatch(testWorkspaceID, scope, 1, watchB); admitted { + t.Fatal("remaining failed target was admitted before its retry deadline") + } + + if err := service.DeletePRWatch(ctx, watchB.ID); err != nil { + t.Fatalf("delete last mixed target: %v", err) + } + recovered := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if recovered.State != PRDiscoveryHealthHealthy || recovered.FailedTargetCount != 0 || recovered.Revision <= failed.Revision { + t.Fatalf("health after deleting mixed targets = %+v, want newer healthy projection", recovered) + } +} + +func TestPRDiscoveryHealth_PrunesArchivedAndBranchSwitchedTargets(t *testing.T) { + ctx := context.Background() + t.Run("archive", func(t *testing.T) { + _, service, _, store := setupBatchedPollerTest(t) + seedTask(t, store, "task-health-archive", false) + watch := withTestWorkspace(&PRWatch{ + SessionID: "session-health-archive", TaskID: "task-health-archive", Owner: "o", Repo: "r", + Branch: "feature/archive", + }) + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create archive watch: %v", err) + } + scope := testAutomationScope(t, service, testWorkspaceID) + attempt, admitted := service.beginPRDiscoveryWatch(testWorkspaceID, scope, 1, watch) + if !admitted { + t.Fatal("archive watch was not admitted") + } + service.prDiscoveryHealth.finishFailure(testWorkspaceID, scope, 1, attempt.target, attempt.number, PRDiscoveryHealthUnavailable, nil) + service.pruneWatchesForTask(ctx, watch.TaskID, "archived") + health := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if health.State != PRDiscoveryHealthHealthy || health.FailedTargetCount != 0 { + t.Fatalf("archived target health = %+v, want healthy", health) + } + }) + + t.Run("branch switch", func(t *testing.T) { + _, service, _, store := setupBatchedPollerTest(t) + seedTask(t, store, "task-health-branch", false) + watch := withTestWorkspace(&PRWatch{ + SessionID: "session-health-branch", TaskID: "task-health-branch", Owner: "o", Repo: "r", + Branch: "feature/old-branch", + }) + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create branch watch: %v", err) + } + scope := testAutomationScope(t, service, testWorkspaceID) + attempt, admitted := service.beginPRDiscoveryWatch(testWorkspaceID, scope, 1, watch) + if !admitted { + t.Fatal("branch watch was not admitted") + } + service.prDiscoveryHealth.finishFailure(testWorkspaceID, scope, 1, attempt.target, attempt.number, PRDiscoveryHealthUnavailable, nil) + failed := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if failed.FailedTargetCount != 1 { + t.Fatalf("branch failure health = %+v, want degraded", failed) + } + if err := service.UpdatePRWatchBranchIfSearching(ctx, watch.ID, "feature/new-branch"); err != nil { + t.Fatalf("switch branch: %v", err) + } + health := service.prDiscoveryHealth.snapshot(testWorkspaceID, scope, 0) + if health.State != PRDiscoveryHealthHealthy || health.FailedTargetCount != 0 || health.Revision <= failed.Revision { + t.Fatalf("branch-switched health = %+v, want newer healthy projection", health) + } + }) +} + +type errString string + +func (e errString) Error() string { return string(e) } diff --git a/apps/backend/internal/github/service_pr_watch.go b/apps/backend/internal/github/service_pr_watch.go index 40d8843aaf..4dc5045e2c 100644 --- a/apps/backend/internal/github/service_pr_watch.go +++ b/apps/backend/internal/github/service_pr_watch.go @@ -135,12 +135,24 @@ func (s *Service) DeletePRWatch(ctx context.Context, id string) error { return err } } - return s.store.DeletePRWatch(ctx, id) + if err := s.store.DeletePRWatch(ctx, id); err != nil { + return err + } + s.removePRDiscoveryWatchConsumer(watch) + return nil } // UpdatePRWatchBranchIfSearching atomically updates branch only when pr_number = 0. func (s *Service) UpdatePRWatchBranchIfSearching(ctx context.Context, id, branch string) error { - return s.store.UpdatePRWatchBranchIfSearching(ctx, id, branch) + watch, err := s.store.GetPRWatch(ctx, id) + if err != nil { + return err + } + if err := s.store.UpdatePRWatchBranchIfSearching(ctx, id, branch); err != nil { + return err + } + s.removePRDiscoveryWatchConsumer(watch) + return nil } // UpdatePRWatchPRNumber updates a PR watch's PR number after discovery. @@ -173,7 +185,15 @@ func (s *Service) rebindPRWatchRepository(ctx context.Context, watch *PRWatch, p // ResetPRWatch atomically resets a watch's branch and clears its pr_number so // the poller re-searches for a PR on the new branch. See Store.ResetPRWatch. func (s *Service) ResetPRWatch(ctx context.Context, id, branch string) error { - return s.store.ResetPRWatch(ctx, id, branch) + watch, err := s.store.GetPRWatch(ctx, id) + if err != nil { + return err + } + if err := s.store.ResetPRWatch(ctx, id, branch); err != nil { + return err + } + s.removePRDiscoveryWatchConsumer(watch) + return nil } // CheckPRWatch fetches lightweight PR status for a watch and determines if there are changes. @@ -1594,15 +1614,52 @@ 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 := s.findPRByBranchInForkNetwork( - ctx, resolved.Client, resolved.CacheScope, watch.Owner, watch.Repo, watch.Branch, + credentialGeneration := int64(0) + if resolved.credential != nil { + credentialGeneration = resolved.credential.CredentialGeneration + } + attempt, admitted := s.beginPRDiscoveryWatch( + watch.WorkspaceID, resolved.CacheScope, credentialGeneration, watch, ) - if err != nil && isRepoNotResolvableErr(err) { - s.markRepoAsMissingForScope(resolved.CacheScope, 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) + if !admitted { + effectiveTaskID := s.reconcileTaskPROwnership(ctx, watch.SessionID, watch.TaskID, watch.RepositoryID, watch.PRNumber) + return s.store.GetTaskPRByRepository(ctx, effectiveTaskID, watch.RepositoryID) + } + var pr *PR + var err error + if attempt.joined { + pr, err = sharedPRDiscoveryResult(ctx, attempt) + } else { + 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) + // 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) + } + if err != nil { + category := classifyPRDiscoveryError(err) + s.completePRDiscoveryWatchAttempt( + attempt, nil, false, true, err, + category == PRDiscoveryHealthInvalidQuery || category == PRDiscoveryHealthRateLimited, + ) + s.finishPRDiscoveryWatchFailure( + watch.WorkspaceID, resolved.CacheScope, credentialGeneration, attempt, err, + ) + } else { + s.completePRDiscoveryWatchAttempt( + attempt, &PRStatus{PR: pr}, true, false, nil, false, + ) + s.finishPRDiscoveryWatchSuccess( + watch.WorkspaceID, resolved.CacheScope, credentialGeneration, attempt, + ) + } + s.forgetPRDiscoveryWatchAttempt( + watch.WorkspaceID, resolved.CacheScope, credentialGeneration, attempt, + ) } // Stamp last_checked_at regardless of outcome — including on error. The // flood this fixes IS the error path (an unresolvable repo makes `gh` exit @@ -1640,6 +1697,12 @@ func (s *Service) detectPRForWatchOnce( zap.String("task_id", watch.TaskID), zap.Int("pr_number", pr.Number), zap.Error(assocErr)) return nil, fmt.Errorf("associate PR: %w", assocErr) } + s.trackPRDiscoveryWatchTarget( + watch.WorkspaceID, resolved.CacheScope, credentialGeneration, watch, + prDiscoveryHealthTarget{ + Owner: watch.Owner, Repo: watch.Repo, Branch: watch.Branch, PRNumber: pr.Number, + }, + ) // Also fetch status so the first response includes review/check state watch.PRNumber = pr.Number return s.triggerPRStatusSync(ctx, watch, taskID) diff --git a/apps/backend/internal/github/service_pr_watch_batched.go b/apps/backend/internal/github/service_pr_watch_batched.go index 469973dc57..7dcbb645be 100644 --- a/apps/backend/internal/github/service_pr_watch_batched.go +++ b/apps/backend/internal/github/service_pr_watch_batched.go @@ -21,6 +21,10 @@ type PRWatchSyncResult struct { Found bool // true when status applied (PR data exists) Changed bool // numbered watches only: true if checks/review state moved SyncFailed bool // true when SyncTaskPR returned an error — callers must NOT publish events + // DiscoveryResolved indicates that the provider lookup completed, even if + // a later database write failed. It keeps health admission focused on + // provider evidence rather than task persistence. + DiscoveryResolved bool } // SyncWatchesBatched runs the batched GraphQL queries for the supplied @@ -37,7 +41,7 @@ type PRWatchSyncResult struct { // TriggerPRSyncAll / ListWorkspaceTaskPRs background refresh share, so a // 40-watch workspace fans out to ~2 gh subprocess calls instead of 40. func (s *Service) SyncWatchesBatched(ctx context.Context, watches []*PRWatch) ([]PRWatchSyncResult, error) { - return s.syncWatchesBatchedWithClient(ctx, s.client, "legacy", watches) + return s.syncWatchesBatchedWithClient(ctx, s.client, "legacy", "", 0, watches) } // SyncWorkspaceWatchesBatched resolves one automation credential and rejects @@ -60,11 +64,26 @@ func (s *Service) SyncWorkspaceWatchesBatched( if err != nil { return nil, err } - return s.syncWatchesBatchedWithClient(ctx, resolved.Client, resolved.CacheScope, watches) + credentialGeneration := int64(0) + if resolved.credential != nil { + credentialGeneration = resolved.credential.CredentialGeneration + } + return s.syncWatchesBatchedWithClient( + ctx, resolved.Client, resolved.CacheScope, workspaceID, credentialGeneration, watches, + ) +} + +type prWatchBatchGroup struct { + key string + target prDiscoveryHealthTarget + representative *PRWatch + watches []*PRWatch + numbered bool + attempt prDiscoveryWatchAttempt } func (s *Service) syncWatchesBatchedWithClient( - ctx context.Context, client Client, cacheScope string, watches []*PRWatch, + ctx context.Context, client Client, cacheScope, workspaceID string, credentialGeneration int64, watches []*PRWatch, ) ([]PRWatchSyncResult, error) { if len(watches) == 0 { return nil, nil @@ -73,26 +92,251 @@ func (s *Service) syncWatchesBatchedWithClient( if err != nil { return nil, err } - numbered, searching := splitPRWatches(watches) - statuses, err := s.fetchBatchedWatchStatuses(ctx, client, exec, cacheScope, numbered, searching) - if err != nil { - return nil, err + groupsByKey := make(map[string]*prWatchBatchGroup, len(watches)) + groups := make([]*prWatchBatchGroup, 0, len(watches)) + for _, watch := range watches { + if watch == nil { + continue + } + target := prDiscoveryTargetForWatch(watch) + key := target.key() + group := groupsByKey[key] + if group == nil { + group = &prWatchBatchGroup{ + key: key, + target: target, + representative: watch, + numbered: target.PRNumber > 0, + } + groupsByKey[key] = group + groups = append(groups, group) + } + group.watches = append(group.watches, watch) + } + admittedGroups := make([]*prWatchBatchGroup, 0, len(groups)) + leaderGroups := make([]*prWatchBatchGroup, 0, len(groups)) + joinedGroups := make([]*prWatchBatchGroup, 0, len(groups)) + for _, group := range groups { + // Register duplicate consumers before the representative performs the + // atomic admission decision. The representative is registered by + // beginPRDiscoveryWatch so another entry point cannot observe it as a + // consumer before it has joined an existing attempt. + for _, watch := range group.watches[1:] { + s.trackPRDiscoveryWatchConsumer(workspaceID, cacheScope, credentialGeneration, watch) + } + attempt, ok := s.beginPRDiscoveryWatch( + workspaceID, cacheScope, credentialGeneration, group.representative, + ) + if !ok { + continue + } + group.attempt = attempt + admittedGroups = append(admittedGroups, group) + if attempt.joined { + joinedGroups = append(joinedGroups, group) + } else { + leaderGroups = append(leaderGroups, group) + } + } + if len(admittedGroups) == 0 { + return noOpPRWatchSyncResults(watches), nil + } + admitted := make([]*PRWatch, 0, len(leaderGroups)) + for _, group := range leaderGroups { + admitted = append(admitted, group.representative) + } + var statuses *batchedWatchStatuses + if len(leaderGroups) > 0 { + numbered, searching := splitPRWatches(admitted) + statuses, err = s.fetchBatchedWatchStatuses(ctx, client, exec, cacheScope, numbered, searching) + if err != nil { + if s.handleBatchedDiscoveryFetchError( + workspaceID, cacheScope, credentialGeneration, leaderGroups, admitted, numbered, searching, err, + ) { + // Schema and throttling failures are not eligible for the legacy + // per-watch fallback: it would repeat the same rejected provider + // request once for every watch. Keep those targets paused under the + // shared retry deadline. + return noOpPRWatchSyncResults(watches), nil + } + // Other provider failures retain the existing transport fallback. Do + // not leave the batch attempt active or impose a health deadline before + // the caller can try the per-watch path. + return nil, err + } } if statuses == nil { statuses = &batchedWatchStatuses{} } - results := make([]PRWatchSyncResult, 0, len(watches)) + resultsByWatch := make(map[*PRWatch]PRWatchSyncResult, len(watches)) now := time.Now().UTC() - for _, w := range numbered { - results = append(results, s.applyBatchedNumberedWatch(ctx, cacheScope, w, statuses.byKey, now)) + for _, group := range leaderGroups { + if group.attempt.result.invalidated() { + continue + } + var representativeResult PRWatchSyncResult + for index, watch := range group.watches { + var result PRWatchSyncResult + if group.numbered { + result = s.applyBatchedNumberedWatch(ctx, cacheScope, watch, statuses.byKey, now) + } else { + result = s.applyBatchedSearchingWatch(ctx, client, cacheScope, watch, statuses, now, false, false, false) + } + resultsByWatch[watch] = result + if index == 0 { + representativeResult = result + } + } + if group.numbered { + s.completePRDiscoveryWatchAttempt( + group.attempt, representativeResult.Status, true, representativeResult.SyncFailed, nil, false, + ) + if s.isRepoCachedAsMissingForScope(cacheScope, group.target.Owner, group.target.Repo) { + s.finishPRDiscoveryWatchFailureCategory( + workspaceID, cacheScope, credentialGeneration, group.attempt, + PRDiscoveryHealthUnavailable, + ) + } else { + // A successful known-PR query can legitimately return no alias + // when the PR is no longer open. That is still a provider success. + s.finishPRDiscoveryWatchSuccess( + workspaceID, cacheScope, credentialGeneration, group.attempt, + ) + } + s.forgetPRDiscoveryWatchAttempt(workspaceID, cacheScope, credentialGeneration, group.attempt) + continue + } + + s.completePRDiscoveryWatchAttempt( + group.attempt, representativeResult.Status, representativeResult.DiscoveryResolved, + representativeResult.SyncFailed, nil, false, + ) + if representativeResult.DiscoveryResolved { + s.finishPRDiscoveryWatchSuccess( + workspaceID, cacheScope, credentialGeneration, group.attempt, + ) + for _, watch := range group.watches { + result := resultsByWatch[watch] + target := prDiscoveryTargetForWatch(watch) + if result.Status != nil && result.Status.PR != nil && !result.SyncFailed { + target = prDiscoveryHealthTarget{ + Owner: result.Status.PR.RepoOwner, + Repo: result.Status.PR.RepoName, + Branch: watch.Branch, + PRNumber: result.Status.PR.Number, + } + } + s.trackPRDiscoveryWatchTarget( + workspaceID, cacheScope, credentialGeneration, watch, target, + ) + } + s.forgetPRDiscoveryWatchAttempt(workspaceID, cacheScope, credentialGeneration, group.attempt) + continue + } + if representativeResult.SyncFailed { + // An unresolved alias whose fallback failed is an unavailable + // discovery result. If the fallback was not due yet, release the + // attempt and preserve any existing failure without inventing a new + // one. + s.finishPRDiscoveryWatchFailureCategory( + workspaceID, cacheScope, credentialGeneration, group.attempt, + PRDiscoveryHealthUnavailable, + ) + s.forgetPRDiscoveryWatchAttempt(workspaceID, cacheScope, credentialGeneration, group.attempt) + continue + } + s.releasePRDiscoveryWatch( + workspaceID, cacheScope, credentialGeneration, group.attempt, + ) + s.forgetPRDiscoveryWatchAttempt(workspaceID, cacheScope, credentialGeneration, group.attempt) } - for _, w := range searching { - results = append(results, s.applyBatchedSearchingWatch(ctx, client, cacheScope, w, statuses, now)) + for _, group := range joinedGroups { + status, resolved, syncFailed, sharedErr, suppressFallback := group.attempt.result.wait(ctx) + if sharedErr != nil { + if suppressFallback { + continue + } + return nil, sharedErr + } + if status == nil && !resolved && syncFailed { + continue + } + if group.numbered { + byKey := make(map[string]*PRStatus, 1) + if status != nil { + byKey[prStatusCacheKey(group.target.Owner, group.target.Repo, group.target.PRNumber)] = status + } + for _, watch := range group.watches { + resultsByWatch[watch] = s.applyBatchedNumberedWatch(ctx, cacheScope, watch, byKey, now) + } + } else { + byKey := make(map[string]*PRStatus, 1) + if status != nil { + byKey[graphqlBranchKey(group.target.Owner, group.target.Repo, group.target.Branch)] = status + } + sharedStatuses := &batchedWatchStatuses{byKey: byKey} + for _, watch := range group.watches { + resultsByWatch[watch] = s.applyBatchedSearchingWatch( + ctx, client, cacheScope, watch, sharedStatuses, now, true, resolved, syncFailed, + ) + } + } + } + results := make([]PRWatchSyncResult, 0, len(watches)) + for _, watch := range watches { + if result, ok := resultsByWatch[watch]; ok { + results = append(results, result) + } else { + results = append(results, PRWatchSyncResult{Watch: watch}) + } } return results, nil } +func noOpPRWatchSyncResults(watches []*PRWatch) []PRWatchSyncResult { + results := make([]PRWatchSyncResult, 0, len(watches)) + for _, watch := range watches { + results = append(results, PRWatchSyncResult{Watch: watch}) + } + return results +} + +func (s *Service) handleBatchedDiscoveryFetchError( + workspaceID, cacheScope string, + credentialGeneration int64, + leaderGroups []*prWatchBatchGroup, + admitted, numbered, searching []*PRWatch, + err error, +) bool { + failed := admitted + if strings.Contains(err.Error(), "batched branch query") { + failed = searching + } else if strings.Contains(err.Error(), "batched PR query") { + failed = numbered + } + category := classifyPRDiscoveryError(err) + pauseFailedTargets := category == PRDiscoveryHealthInvalidQuery || category == PRDiscoveryHealthRateLimited + failedSet := make(map[string]struct{}, len(failed)) + for _, watch := range failed { + failedSet[prDiscoveryTargetForWatch(watch).key()] = struct{}{} + } + for _, group := range leaderGroups { + if _, failed := failedSet[group.key]; failed && pauseFailedTargets { + s.finishPRDiscoveryWatchFailure( + workspaceID, cacheScope, credentialGeneration, group.attempt, err, + ) + } else { + s.releasePRDiscoveryWatch( + workspaceID, cacheScope, credentialGeneration, group.attempt, + ) + } + s.completePRDiscoveryWatchAttempt(group.attempt, nil, false, true, err, pauseFailedTargets) + s.forgetPRDiscoveryWatchAttempt(workspaceID, cacheScope, credentialGeneration, group.attempt) + } + return pauseFailedTargets +} + const ( workflowAttentionBatchConcurrency = 4 workflowAttentionBatchBudget = 5 * time.Second @@ -542,14 +786,16 @@ func (s *Service) applyBatchedNumberedWatch( func (s *Service) lookupSearchingWatchPR( ctx context.Context, client Client, cacheScope string, w *PRWatch, branchKey string, statuses *batchedWatchStatuses, lastChecked *time.Time, -) (*PR, error) { +) (*PR, bool, error) { if lastChecked != nil && time.Since(*lastChecked) < PRSyncFreshnessWindow { - return nil, nil + return nil, false, nil } if _, definitive := statuses.branchResolvedEmpty[branchKey]; definitive { - return s.findPRInForkParent(ctx, client, cacheScope, w.Owner, w.Repo, w.Branch) + pr, err := s.findPRInForkParent(ctx, client, cacheScope, w.Owner, w.Repo, w.Branch) + return pr, true, err } - return s.findPRByBranchInForkNetwork(ctx, client, cacheScope, w.Owner, w.Repo, w.Branch) + pr, err := s.findPRByBranchInForkNetwork(ctx, client, cacheScope, w.Owner, w.Repo, w.Branch) + return pr, true, err } // applyBatchedSearchingWatch mirrors Poller.applyDetectedPR on the service @@ -557,6 +803,7 @@ func (s *Service) lookupSearchingWatchPR( // branch lookup returns one; otherwise we just bump last_checked_at. func (s *Service) applyBatchedSearchingWatch( ctx context.Context, client Client, cacheScope string, w *PRWatch, statuses *batchedWatchStatuses, now time.Time, + skipFallback, sharedResolved, sharedSyncFailed bool, ) PRWatchSyncResult { // Read the previous probe time before the stamp below overwrites it — // lookupSearchingWatchPR throttles on it. @@ -570,27 +817,35 @@ func (s *Service) applyBatchedSearchingWatch( } branchKey := graphqlBranchKey(w.Owner, w.Repo, w.Branch) status, ok := statuses.byKey[branchKey] + discoveryResolved := ok && status != nil && status.PR != nil if !ok || status == nil || status.PR == nil { - pr, err := s.lookupSearchingWatchPR(ctx, client, cacheScope, w, branchKey, statuses, lastChecked) + if skipFallback { + return PRWatchSyncResult{Watch: w, DiscoveryResolved: sharedResolved, SyncFailed: sharedSyncFailed} + } + pr, attempted, 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)) return PRWatchSyncResult{Watch: w, SyncFailed: true} } - if pr == nil { + if !attempted { return PRWatchSyncResult{Watch: w} } + discoveryResolved = true + if pr == nil { + return PRWatchSyncResult{Watch: w, DiscoveryResolved: true} + } status = &PRStatus{PR: pr} } if err := s.rebindPRWatchRepository(ctx, w, status.PR); err != nil { s.logger.Error("failed to rebind PR watch to detected repository", zap.String("watch_id", w.ID), zap.Int("pr_number", status.PR.Number), zap.Error(err)) - return PRWatchSyncResult{Watch: w, Status: status, Found: true, SyncFailed: true} + return PRWatchSyncResult{Watch: w, Status: status, Found: true, SyncFailed: true, DiscoveryResolved: discoveryResolved} } if err := s.store.UpdatePRWatchPRNumber(ctx, w.ID, status.PR.Number); err != nil { s.logger.Error("failed to update PR watch with detected PR", zap.String("watch_id", w.ID), zap.Int("pr_number", status.PR.Number), zap.Error(err)) - return PRWatchSyncResult{Watch: w, Status: status, Found: true} + return PRWatchSyncResult{Watch: w, Status: status, Found: true, DiscoveryResolved: discoveryResolved} } if _, err := s.associatePRWithTaskForSession( ctx, w.WorkspaceID, w.SessionID, w.TaskID, w.RepositoryID, status.PR, @@ -598,9 +853,9 @@ func (s *Service) applyBatchedSearchingWatch( ); err != nil { s.logger.Error("failed to associate detected PR with task", zap.String("task_id", w.TaskID), zap.Int("pr_number", status.PR.Number), zap.Error(err)) - return PRWatchSyncResult{Watch: w, Status: status, Found: true} + return PRWatchSyncResult{Watch: w, Status: status, Found: true, DiscoveryResolved: discoveryResolved} } s.logger.Info("detected PR for session branch (batched)", zap.String("watch_id", w.ID), zap.String("branch", w.Branch), zap.Int("pr_number", status.PR.Number)) - return PRWatchSyncResult{Watch: w, Status: status, Found: true} + return PRWatchSyncResult{Watch: w, Status: status, Found: true, DiscoveryResolved: discoveryResolved} } diff --git a/apps/backend/internal/github/service_task_events.go b/apps/backend/internal/github/service_task_events.go index 8e1debbb7d..931a937d0b 100644 --- a/apps/backend/internal/github/service_task_events.go +++ b/apps/backend/internal/github/service_task_events.go @@ -166,6 +166,13 @@ func (s *Service) pruneTaskOwnedStateForTask(ctx context.Context, taskID, reason } func (s *Service) pruneWatchesForTask(ctx context.Context, taskID, reason string) { + watches, listErr := s.store.ListPRWatchesByTask(ctx, taskID) + if listErr != nil { + s.logger.Error("failed to list PR watches for task", + zap.String("task_id", taskID), + zap.String("reason", reason), + zap.Error(listErr)) + } n, err := s.store.DeletePRWatchesByTaskID(ctx, taskID) if err != nil { s.logger.Error("failed to delete PR watches for task", @@ -174,6 +181,9 @@ func (s *Service) pruneWatchesForTask(ctx context.Context, taskID, reason string zap.Error(err)) return } + for _, watch := range watches { + s.removePRDiscoveryWatchConsumer(watch) + } if n > 0 { s.logger.Info("pruned PR watches after task change", zap.String("task_id", taskID), diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index dcc4283840..a1c57ca3e5 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -537,31 +537,32 @@ const ( // GitHub integration actions const ( - ActionGitHubStatus = "github.status" - ActionGitHubTaskPRsList = "github.task_prs.list" - ActionGitHubTaskPRGet = "github.task_pr.get" - ActionGitHubPRFeedbackGet = "github.pr_feedback.get" - ActionGitHubReviewWatchesList = "github.review_watches.list" - ActionGitHubReviewWatchCreate = "github.review_watches.create" - ActionGitHubReviewWatchUpdate = "github.review_watches.update" - ActionGitHubReviewWatchDelete = "github.review_watches.delete" - ActionGitHubReviewTrigger = "github.review_watches.trigger" - ActionGitHubReviewTriggerAll = "github.review_watches.trigger_all" - ActionGitHubPRWatchesList = "github.pr_watches.list" - ActionGitHubPRWatchDelete = "github.pr_watches.delete" - ActionGitHubPRFilesGet = "github.pr_files.get" - ActionGitHubPRCommitsGet = "github.pr_commits.get" - ActionGitHubPRCommitGet = "github.pr_commit.get" - ActionGitHubTaskPRUpdated = "github.task_pr.updated" // Notification - ActionGitHubTaskPRDeleted = "github.task_pr.deleted" // Notification - ActionGitHubTaskCIOptionsUpdated = "github.task_ci_options.updated" // Notification - ActionGitHubRateLimitUpdated = "github.rate_limit.updated" // Notification - ActionGitHubPRFeedbackNotify = "github.pr_feedback.notify" // Notification - ActionGitHubNewReviewPRNotify = "github.new_review_pr.notify" // Notification - ActionGitHubTaskPRSync = "github.task_pr.sync" - ActionGitHubStats = "github.stats" - ActionGitHubCheckSessionPR = "github.check_session_pr" - ActionGitLabCheckSessionMR = "gitlab.check_session_mr" + ActionGitHubStatus = "github.status" + ActionGitHubTaskPRsList = "github.task_prs.list" + ActionGitHubTaskPRGet = "github.task_pr.get" + ActionGitHubPRFeedbackGet = "github.pr_feedback.get" + ActionGitHubReviewWatchesList = "github.review_watches.list" + ActionGitHubReviewWatchCreate = "github.review_watches.create" + ActionGitHubReviewWatchUpdate = "github.review_watches.update" + ActionGitHubReviewWatchDelete = "github.review_watches.delete" + ActionGitHubReviewTrigger = "github.review_watches.trigger" + ActionGitHubReviewTriggerAll = "github.review_watches.trigger_all" + ActionGitHubPRWatchesList = "github.pr_watches.list" + ActionGitHubPRWatchDelete = "github.pr_watches.delete" + ActionGitHubPRFilesGet = "github.pr_files.get" + ActionGitHubPRCommitsGet = "github.pr_commits.get" + ActionGitHubPRCommitGet = "github.pr_commit.get" + ActionGitHubTaskPRUpdated = "github.task_pr.updated" // Notification + ActionGitHubTaskPRDeleted = "github.task_pr.deleted" // Notification + ActionGitHubTaskCIOptionsUpdated = "github.task_ci_options.updated" // Notification + ActionGitHubRateLimitUpdated = "github.rate_limit.updated" // Notification + ActionGitHubPRDiscoveryHealthUpdated = "github.pr_discovery_health.updated" // Notification + ActionGitHubPRFeedbackNotify = "github.pr_feedback.notify" // Notification + ActionGitHubNewReviewPRNotify = "github.new_review_pr.notify" // Notification + ActionGitHubTaskPRSync = "github.task_pr.sync" + ActionGitHubStats = "github.stats" + ActionGitHubCheckSessionPR = "github.check_session_pr" + ActionGitLabCheckSessionMR = "gitlab.check_session_mr" // Issue watch actions ActionGitHubIssueWatchesList = "github.issue_watches.list" diff --git a/apps/web/components/github/github-access-help.tsx b/apps/web/components/github/github-access-help.tsx index f7ef44ab80..d4a38ef5e5 100644 --- a/apps/web/components/github/github-access-help.tsx +++ b/apps/web/components/github/github-access-help.tsx @@ -70,7 +70,11 @@ export function GitHubAccessHelp({ {description} - {content &&
{t("github:prDiscoveryFailed")}
+{t("github:prDiscoveryCategory", { category })}
++ {t("github:prDiscoveryLastFailure", { + time: relativeTime(health.last_failure_at, locale, unknownTime), + })} +
+{t("github:prDiscoveryAffectedTargets", { count: health.failed_target_count })}
+ {health.retry_at && ( ++ {t("github:prDiscoveryRetryAt", { + time: relativeTime(health.retry_at, locale, unknownTime), + })} +
+ )} +