From 99d087c1efb64aa7f6127b88d5131a8c57363948 Mon Sep 17 00:00:00 2001 From: Wes McKinney Date: Wed, 7 Oct 2026 09:02:55 -0500 Subject: [PATCH] fix: keep PR conflict status current after relay updates A relay branch hint could consume the list change that scheduled sync uses to refresh mergeability. PR details could also remain recently fetched while GitHub had not yet resolved their mergeability, delaying a conflict warning until a later refresh or merge attempt. Keep those unresolved PRs eligible within the existing provider budget and preserve normal sync's opportunity to observe the complete state. Generated with Codex Co-authored-by: Codex --- context/github-sync-invariants.md | 6 ++ internal/github/queue.go | 22 ++--- internal/github/relay.go | 4 +- internal/github/relay_test.go | 101 +++++++++++++++++++++ internal/github/sync.go | 29 +++--- internal/github/sync_test.go | 22 ++++- internal/server/pullservertest/api_test.go | 7 +- 7 files changed, 162 insertions(+), 29 deletions(-) diff --git a/context/github-sync-invariants.md b/context/github-sync-invariants.md index f6b91f74e..f31d0b49d 100644 --- a/context/github-sync-invariants.md +++ b/context/github-sync-invariants.md @@ -82,6 +82,9 @@ what "current" means. - Daily coverage is a target, not permission to exceed quota. Show remaining overdue open items, including never-fetched items, rather than implying complete freshness (`internal/github/sync.go::countOverdueDetails`). +- Unknown or missing GitHub mergeability remains eligible for budgeted detail refresh; + it can resolve after a base push without changing the PR's `updated_at`. + (`internal/github/sync.go::buildDetailQueueItems`) - Comment-only polling must respect dormant-item cadence. Admitted open-item detail checks still check comments on parent 304s before advancing freshness; edits and deletions may leave the parent unchanged (`internal/github/sync.go::markUnchangedIssueDetailFetched`). @@ -839,6 +842,9 @@ error or cancellation unchanged and never adopts. accepting an HTTP stream is not evidence of compatibility. (`internal/github/relay.go::RunRelay`) - Relay hints accelerate normal polling; each consumer still uses its own credentials and rate gates. (`internal/github/relay.go::refreshRelayHint`) +- Relay ref reads must not advance the normal-sync list ETag: REST indexing omits + mergeability, while that validator also gates the richer GraphQL refresh. + (`internal/github/relay.go::refreshRelayRefs`) - One `RunRelay` loop owns the subscription and reconnects with jittered exponential backoff from the backoff library, 1s rising to a 30s base ceiling plus jitter, reset only after a stream stayed open. The `[relay]` config has no poll interval; `relay.poll_interval` is rejected at load. (`internal/github/relay.go::RunRelay`) diff --git a/internal/github/queue.go b/internal/github/queue.go index 2c99ea376..167754081 100644 --- a/internal/github/queue.go +++ b/internal/github/queue.go @@ -29,14 +29,15 @@ type QueueItem struct { Score float64 // Scoring inputs - UpdatedAt time.Time - DetailFetchedAt *time.Time - CIHadPending bool - Starred bool - Watched bool - IsOpen bool - LargeRepo bool - dailyDue bool + UpdatedAt time.Time + DetailFetchedAt *time.Time + CIHadPending bool + MergeabilityPending bool + Starred bool + Watched bool + IsOpen bool + LargeRepo bool + dailyDue bool } // WorstCaseCost returns the maximum wire attempts this item's @@ -125,9 +126,8 @@ func isEligible(qi *QueueItem, now time.Time) bool { return true } - // CI had pending checks — always eligible regardless of - // updated_at. - if qi.CIHadPending { + // CI and mergeability can resolve without changing updated_at. + if qi.CIHadPending || qi.MergeabilityPending { return true } diff --git a/internal/github/relay.go b/internal/github/relay.go index 8e24ed5e7..9b4b80b23 100644 --- a/internal/github/relay.go +++ b/internal/github/relay.go @@ -348,7 +348,9 @@ func (s *Syncer) refreshRelayRefs(ctx context.Context, repo RepoRef) error { return err } requestedAt := s.nowUTC() - prs, err := client.ListOpenPullRequests(ctx, repo.Owner, repo.Name) + // This read only indexes REST fields. Leave the list ETag to normal sync, + // where a changed list also triggers the GraphQL mergeability refresh. + prs, err := client.ListOpenPullRequests(platformgithub.WithUnconditionalRead(ctx), repo.Owner, repo.Name) if platformgithub.IsNotModified(err) { return nil } diff --git a/internal/github/relay_test.go b/internal/github/relay_test.go index b907fc123..b7d696094 100644 --- a/internal/github/relay_test.go +++ b/internal/github/relay_test.go @@ -20,6 +20,7 @@ import ( "go.kenn.io/forge/internal/db" "go.kenn.io/forge/internal/testutil/reposeed" "go.kenn.io/forge/platform" + platformgithub "go.kenn.io/forge/platform/github" ) // relayStatuses forwards every published relay status so tests wait on the @@ -450,3 +451,103 @@ func TestRelayDisabledIssueRespectsCooldown(t *testing.T) { require.NoError(syncer.refreshRelayHint(WithSyncBudget(ctx), hint)) assert.Equal(2, calls) } + +func TestRelayRefsDoesNotConsumeNormalSyncListChange(t *testing.T) { + t.Parallel() + require := require.New(t) + assert := assert.New(t) + ctx := t.Context() + database := openTestDB(t) + repo := RepoRef{Platform: platform.KindGitHub, PlatformHost: "github.com", Key: platform.RepositoryIDKey(1002), Owner: "team", Name: "project"} + repoID, err := reposeed.Seed(ctx, database, db.RepoIdentity{ + Platform: "github", PlatformHost: "github.com", Key: repo.Key, Owner: repo.Owner, Name: repo.Name, + }) + require.NoError(err) + provider := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/api/v3/repositories/1002": + _, _ = fmt.Fprint(w, `{"id":1002,"name":"project","owner":{"login":"team"},"default_branch":"main","has_issues":true}`) + case "/api/v3/repos/team/project/pulls": + if r.Header.Get("If-None-Match") == `"new-base"` { + w.WriteHeader(http.StatusNotModified) + return + } + w.Header().Set("ETag", `"new-base"`) + _, _ = fmt.Fprint(w, `[{"id":700,"number":7,"title":"Test PR","state":"open","head":{"sha":"head","ref":"feature"},"base":{"sha":"new-base","ref":"main","repo":{"id":1002,"name":"project","owner":{"login":"team"}}},"created_at":"2026-09-01T00:00:00Z","updated_at":"2026-09-01T00:00:00Z"}]`) + default: + http.Error(w, "unexpected provider request", http.StatusNotFound) + } + })) + t.Cleanup(provider.Close) + client, err := NewClient(testTokenSource("token"), "github.com", nil, nil, WithBaseURLForTesting(provider.URL)) + require.NoError(err) + syncer := NewSyncer(map[string]Client{"github.com": client}, database, nil, []RepoRef{repo}, time.Minute, nil, testBudget(1000)) + require.NoError(syncer.refreshRelayHint(WithSyncBudget(ctx), activityrelay.Hint{ + Provider: "github", Host: "github.com", RepositoryID: 1002, Target: activityrelay.RepositoryRefs, + })) + stored, err := database.GetMergeRequestByRepoIDAndNumber(ctx, repoID, 7) + require.NoError(err) + require.NotNil(stored) + assert.Equal("new-base", stored.PlatformBaseSHA) + assert.Empty(stored.MergeableState, "the relay list read has not observed mergeability") + + // Normal sync must still see a changed list so it can run its GraphQL + // mergeability refresh. The relay only indexed the REST list fields. + prs, err := client.ListOpenPullRequests(ctx, repo.Owner, repo.Name) + require.NoError(err) + require.Len(prs, 1) + assert.Equal(7, prs[0].GetNumber()) + _, err = client.ListOpenPullRequests(ctx, repo.Owner, repo.Name) + assert.True(platformgithub.IsNotModified(err), "normal sync still benefits from its own list ETag") +} + +func TestSyncRefreshesMergeabilityAfterBaseChange(t *testing.T) { + t.Parallel() + for _, relay := range []bool{false, true} { + t.Run(fmt.Sprintf("relay=%v", relay), func(t *testing.T) { + t.Parallel() + require := require.New(t) + assert := assert.New(t) + ctx := t.Context() + database := openTestDB(t) + repo := RepoRef{Owner: "owner", Name: "repo", PlatformHost: "github.com", Key: platform.RepositoryIDKey(testRepoID("owner", "repo"))} + updated := time.Now().UTC().Add(-time.Hour) + pr := buildOpenPR(1, updated) + pr.Base.SHA = new("old-base") + pr.MergeableState = new("clean") + client := &mockClient{openPRs: []*gh.PullRequest{pr}} + syncer := NewSyncer(map[string]Client{"github.com": client}, database, nil, []RepoRef{repo}, time.Minute, nil, testBudget(1000)) + syncer.RunOnce(ctx) + stored, err := database.GetMergeRequest(ctx, "github", "github.com", "owner", "repo", 1) + require.NoError(err) + require.NotNil(stored) + assert.Equal("clean", stored.MergeableState) + require.NotNil(stored.DetailFetchedAt) + + // A push to main changes the base, not the PR head or updated_at. + listed := buildOpenPR(1, updated) + listed.Base.SHA = new("new-base") + client.openPRs = []*gh.PullRequest{listed} + client.getPullRequestFn = func(context.Context, string, string, int) (*gh.PullRequest, error) { + full := buildOpenPR(1, updated) + full.Base.SHA = new("new-base") + full.MergeableState = new("dirty") + return full, nil + } + if relay { + require.NoError(syncer.refreshRelayHint(WithSyncBudget(ctx), activityrelay.Hint{ + Provider: "github", Host: "github.com", RepositoryID: testRepoID("owner", "repo"), Target: activityrelay.RepositoryRefs, + })) + // The scheduled pass may get a 304 after the hint indexed the list. + client.listOpenPRsErr = notModifiedErr() + } + syncer.RunOnce(ctx) + stored, err = database.GetMergeRequest(ctx, "github", "github.com", "owner", "repo", 1) + require.NoError(err) + require.NotNil(stored) + assert.Equal("new-base", stored.PlatformBaseSHA) + assert.Equal("dirty", stored.MergeableState) + }) + } +} diff --git a/internal/github/sync.go b/internal/github/sync.go index 58ccae724..f40c5c59f 100644 --- a/internal/github/sync.go +++ b/internal/github/sync.go @@ -10785,20 +10785,23 @@ func (s *Syncer) buildDetailQueueItems( repo.Owner, repo.Name, ) + fmt.Sprintf("#%d", pr.Number) ciHadPending := pr.CIHadPending || ciHasPending(pr.CIChecksJSON) + mergeabilityPending := repo.Platform == "github" && + (pr.MergeableState == "" || pr.MergeableState == "unknown") items = append(items, QueueItem{ - Type: QueueItemPR, - Platform: platform.Kind(repo.Platform), - RepoOwner: repo.Owner, - RepoName: repo.Name, - Number: pr.Number, - PlatformHost: repo.PlatformHost, - UpdatedAt: pr.UpdatedAt, - DetailFetchedAt: pr.DetailFetchedAt, - CIHadPending: ciHadPending, - Starred: pr.Starred, - Watched: watched[watchKey], - IsOpen: true, - LargeRepo: prCountsByRepoID[pr.RepoID] >= largeRepoBulkGraphQLThreshold, + Type: QueueItemPR, + Platform: platform.Kind(repo.Platform), + RepoOwner: repo.Owner, + RepoName: repo.Name, + Number: pr.Number, + PlatformHost: repo.PlatformHost, + UpdatedAt: pr.UpdatedAt, + DetailFetchedAt: pr.DetailFetchedAt, + CIHadPending: ciHadPending, + MergeabilityPending: mergeabilityPending, + Starred: pr.Starred, + Watched: watched[watchKey], + IsOpen: true, + LargeRepo: prCountsByRepoID[pr.RepoID] >= largeRepoBulkGraphQLThreshold, }) } diff --git a/internal/github/sync_test.go b/internal/github/sync_test.go index 6c4055412..c22aafa5e 100644 --- a/internal/github/sync_test.go +++ b/internal/github/sync_test.go @@ -6480,6 +6480,7 @@ func TestSyncTriggersFullFetchForUnknownMergeableState(t *testing.T) { // Build a list PR with diff stats set so the zero-stats condition // doesn't trigger the full fetch independently. listPR := buildOpenPR(1, now) + listPR.Base.SHA = new("base123") additions := 10 deletions := 5 listPR.Additions = &additions @@ -6496,12 +6497,13 @@ func TestSyncTriggersFullFetchForUnknownMergeableState(t *testing.T) { mc.getPullRequestFn = func(_ context.Context, _, _ string, _ int) (*gh.PullRequest, error) { fetchCount++ p := buildOpenPR(1, now) + p.Base.SHA = new("base123") a, d2 := 10, 5 p.Additions = &a p.Deletions = &d2 state := "unknown" if fetchCount >= 2 { - state = "clean" + state = "dirty" } p.MergeableState = &state return p, nil @@ -6518,6 +6520,18 @@ func TestSyncTriggersFullFetchForUnknownMergeableState(t *testing.T) { require.NotNil(stored) assert.Equal("unknown", stored.MergeableState) assert.Equal(1, fetchCount, "first sync should trigger one full fetch via detail drain") + + // GitHub finishes computing mergeability without changing updated_at. + // A recent detail fetch must not hide the now-known conflict. + syncer.RunOnce(ctx) + stored, err = d.GetMergeRequest(ctx, "github", "github.com", "owner", "repo", 1) + require.NoError(err) + require.NotNil(stored) + assert.Equal("dirty", stored.MergeableState) + assert.Equal(2, fetchCount, "unknown mergeability must be checked on the next sync") + + syncer.RunOnce(ctx) + assert.Equal(2, fetchCount, "resolved mergeability resumes the normal detail cadence") } func TestSyncPreservesFieldsOnFullFetchFailure(t *testing.T) { @@ -13362,7 +13376,9 @@ func TestRunOnceLargeExistingRepoSkipsBulkGraphQLAndFetchesChangedPRDetail(t *te if number == 1 { updatedAt = changedAt } - openPRs = append(openPRs, buildOpenPR(number, updatedAt)) + listed := buildOpenPR(number, updatedAt) + listed.Base.SHA = new("base123") + openPRs = append(openPRs, listed) _, err := d.UpsertMergeRequest(ctx, &db.MergeRequest{ RepoID: repoID, PlatformID: int64(number * 1000), @@ -13374,6 +13390,8 @@ func TestRunOnceLargeExistingRepoSkipsBulkGraphQLAndFetchesChangedPRDetail(t *te HeadBranch: "feature-branch", BaseBranch: "main", PlatformHeadSHA: "abc123def456", + PlatformBaseSHA: "base123", + MergeableState: "clean", CreatedAt: unchangedAt, UpdatedAt: unchangedAt, LastActivityAt: unchangedAt, diff --git a/internal/server/pullservertest/api_test.go b/internal/server/pullservertest/api_test.go index 3d351dfa8..3d3c2ad9d 100644 --- a/internal/server/pullservertest/api_test.go +++ b/internal/server/pullservertest/api_test.go @@ -190,7 +190,7 @@ func TestE2ELargeRepoSkipsGraphQLAndUsesConditionalPRDetail(t *testing.T) { UpdatedAt: &updated, Comments: &comments, Head: &gh.PullRequestBranch{SHA: &headSHA, Ref: &headRef}, - Base: &gh.PullRequestBranch{Ref: &baseRef}, + Base: &gh.PullRequestBranch{Ref: &baseRef, SHA: new("base123")}, } } @@ -266,6 +266,8 @@ func TestE2ELargeRepoSkipsGraphQLAndUsesConditionalPRDetail(t *testing.T) { HeadBranch: fmt.Sprintf("feature-%d", number), BaseBranch: "main", PlatformHeadSHA: fmt.Sprintf("head-%d", number), + PlatformBaseSHA: "base123", + MergeableState: "clean", CreatedAt: unchangedAt, UpdatedAt: unchangedAt, LastActivityAt: unchangedAt, @@ -349,7 +351,7 @@ func TestE2EConditionalPRDetailRefreshesInlineModerationThroughAPI(t *testing.T) ID: &id, Number: &number, State: &state, Title: &title, HTMLURL: &url, User: &gh.User{Login: &author}, CreatedAt: ×tamp, UpdatedAt: ×tamp, Head: &gh.PullRequestBranch{SHA: &headSHA, Ref: &headRef}, - Base: &gh.PullRequestBranch{Ref: &baseRef}, + Base: &gh.PullRequestBranch{Ref: &baseRef, SHA: new("base123")}, } } @@ -418,6 +420,7 @@ func TestE2EConditionalPRDetailRefreshesInlineModerationThroughAPI(t *testing.T) Title: fmt.Sprintf("existing PR %d", number), Author: "alice", State: "open", HeadBranch: fmt.Sprintf("feature-%d", number), BaseBranch: "main", PlatformHeadSHA: fmt.Sprintf("head-%d", number), CreatedAt: now, UpdatedAt: now, + PlatformBaseSHA: "base123", MergeableState: "clean", LastActivityAt: now, DetailFetchedAt: detailFetchedAt, }) require.NoError(err)