From 89c0433878f44acc6bc42df45f0262a9bd8e9ede Mon Sep 17 00:00:00 2001 From: openshift-trt Date: Fri, 24 Jul 2026 12:48:27 +0000 Subject: [PATCH] TRT-2799: Fix SHA mismatch in syncPRStatus, convert to bulk operations syncPRStatus compared MergeCommitSHA (the merge commit GitHub creates) against prow_pull_requests.SHA (the PR head commit from prow job specs). These are different commits so the comparison never matched, leaving merged_at NULL on 82% of rows. Replace the per-PR loop with a batch pipeline that uses Head.SHA from the GitHub PR objects (matching the original correct GetPRSHAMerged behavior). The new implementation queries distinct repos, fetches recently merged PRs via ListRecentlyMergedPRs, then uses CopyToTempTable for a bulk UPDATE and bulk DELETE instead of per-row DB round-trips. Co-Authored-By: Claude Opus 4.6 --- pkg/dataloader/prowloader/github/github.go | 38 ++++++ .../prowloader/github/github_test.go | 64 +++++++++ pkg/dataloader/prowloader/prow.go | 126 ++++++++++++------ 3 files changed, 186 insertions(+), 42 deletions(-) diff --git a/pkg/dataloader/prowloader/github/github.go b/pkg/dataloader/prowloader/github/github.go index dd62a6d8af..2cbacfe6f8 100644 --- a/pkg/dataloader/prowloader/github/github.go +++ b/pkg/dataloader/prowloader/github/github.go @@ -45,6 +45,12 @@ type PREntry struct { State *string } +type MergedPR struct { + Number int + HeadSHA string + MergedAt time.Time +} + type Client struct { ctx context.Context cache map[prlocator]*PREntry @@ -220,6 +226,38 @@ func (c *Client) IsPrRecentlyMerged(org, repo string, number int) (*time.Time, * return nil, nil, err } +func (c *Client) ListRecentlyMergedPRs(org, repo string) ([]MergedPR, error) { + c.closedCacheLock.Lock() + defer c.closedCacheLock.Unlock() + if c.closedCache[org] == nil { + c.closedCache[org] = make(map[string]map[int]*gh.PullRequest) + } + + var fetchErr error + if c.closedCache[org][repo] == nil { + c.closedCache[org][repo], fetchErr = c.gitHubListClosedPRs(org, repo) + if fetchErr != nil { + log.WithError(fetchErr).Errorf("Error fetching closed PRs for %s/%s", org, repo) + } + } + + var merged []MergedPR + for _, pr := range c.closedCache[org][repo] { + if pr == nil || pr.MergedAt == nil || pr.Number == nil { + continue + } + if pr.Head == nil || pr.Head.SHA == nil { + continue + } + merged = append(merged, MergedPR{ + Number: *pr.Number, + HeadSHA: *pr.Head.SHA, + MergedAt: *pr.MergedAt, + }) + } + return merged, fetchErr +} + func (c *Client) IsWithinRateLimitThreshold() bool { rate, err := c.gitHubCoreRateFetch() diff --git a/pkg/dataloader/prowloader/github/github_test.go b/pkg/dataloader/prowloader/github/github_test.go index 04d5f0b4e2..31c89d1a63 100644 --- a/pkg/dataloader/prowloader/github/github_test.go +++ b/pkg/dataloader/prowloader/github/github_test.go @@ -175,6 +175,70 @@ func TestClient_GetPRSHAMerged(t *testing.T) { } +func TestClient_ListRecentlyMergedPRs(t *testing.T) { + now := time.Now() + headSHA1 := "aaa111" + headSHA2 := "bbb222" + mergeSHA1 := "ccc333" + num1 := 10 + num2 := 20 + num3 := 30 + + client := &Client{ + ctx: context.TODO(), + closedCache: make(map[string]map[string]map[int]*gh.PullRequest), + gitHubListClosedPRs: func(org, repo string) (map[int]*gh.PullRequest, error) { + return map[int]*gh.PullRequest{ + num1: { + Number: &num1, + MergedAt: &now, + MergeCommitSHA: &mergeSHA1, + Head: &gh.PullRequestBranch{SHA: &headSHA1}, + }, + num2: { + Number: &num2, + MergedAt: nil, // closed but not merged + Head: &gh.PullRequestBranch{SHA: &headSHA2}, + }, + num3: { + Number: &num3, + MergedAt: &now, + Head: nil, // merged but missing Head (shouldn't happen, but be safe) + }, + }, nil + }, + } + + merged, err := client.ListRecentlyMergedPRs(openshift, kubernetes) + if err != nil { + t.Fatalf("ListRecentlyMergedPRs() unexpected error: %v", err) + } + + if len(merged) != 1 { + t.Fatalf("ListRecentlyMergedPRs() want 1 merged PR, got %d", len(merged)) + } + + m := merged[0] + if m.Number != num1 { + t.Errorf("ListRecentlyMergedPRs() number = %d, want %d", m.Number, num1) + } + if m.HeadSHA != headSHA1 { + t.Errorf("ListRecentlyMergedPRs() HeadSHA = %s, want %s (not MergeCommitSHA %s)", m.HeadSHA, headSHA1, mergeSHA1) + } + if !m.MergedAt.Equal(now) { + t.Errorf("ListRecentlyMergedPRs() MergedAt = %v, want %v", m.MergedAt, now) + } + + // Second call should use cache, not refetch + merged2, err := client.ListRecentlyMergedPRs(openshift, kubernetes) + if err != nil { + t.Fatalf("ListRecentlyMergedPRs() second call unexpected error: %v", err) + } + if len(merged2) != 1 { + t.Errorf("ListRecentlyMergedPRs() second call want 1, got %d", len(merged2)) + } +} + func TestClient_IsCommentIdMatch(t *testing.T) { client := &Client{commentMetaRegEx: regexp.MustCompile(commentIDRegex)} diff --git a/pkg/dataloader/prowloader/prow.go b/pkg/dataloader/prowloader/prow.go index 463e76578a..ff44fa2058 100644 --- a/pkg/dataloader/prowloader/prow.go +++ b/pkg/dataloader/prowloader/prow.go @@ -30,7 +30,6 @@ import ( "github.com/prometheus/client_golang/prometheus/push" log "github.com/sirupsen/logrus" "google.golang.org/api/iterator" - "gorm.io/gorm" "gorm.io/gorm/clause" "k8s.io/apimachinery/pkg/util/sets" @@ -814,59 +813,102 @@ func (pl *ProwLoader) syncPRStatus() error { return nil } - pulls := make([]models.ProwPullRequest, 0) + type orgRepo struct { + Org string + Repo string + } + var repos []orgRepo if res := pl.dbc.DB. Table("prow_pull_requests"). - Where("merged_at IS NULL").Scan(&pulls); res.Error != nil && !errors.Is(res.Error, gorm.ErrRecordNotFound) { - return errors.Wrap(res.Error, "could not fetch prow_pull_requests") - } - - for _, pr := range pulls { - logger := log.WithField("org", pr.Org). - WithField("repo", pr.Repo). - WithField("number", pr.Number). - WithField("sha", pr.SHA) - - // first check to see if this pr has recently closed (indicating it may have merged) - recentMergedAt, mergeCommitSha, err := pl.githubClient.IsPrRecentlyMerged(pr.Org, pr.Repo, pr.Number) - - // the client should have logged the error, we want - // to see if we are rate limited or not, if so return - // otherwise keep processing + Select("DISTINCT org, repo"). + Where("merged_at IS NULL"). + Scan(&repos); res.Error != nil { + return errors.Wrap(res.Error, "could not fetch distinct repos from prow_pull_requests") + } + + type mergedRow struct { + Org string + Repo string + Number int + SHA string + MergedAt time.Time + } + var allMerged []mergedRow + for _, r := range repos { + merged, err := pl.githubClient.ListRecentlyMergedPRs(r.Org, r.Repo) if err != nil { if pl.githubClient.IsWithinRateLimitThreshold() { return err } + continue } + for _, m := range merged { + allMerged = append(allMerged, mergedRow{ + Org: r.Org, + Repo: r.Repo, + Number: m.Number, + SHA: m.HeadSHA, + MergedAt: m.MergedAt, + }) + } + } - if recentMergedAt != nil { - // we have the recentMergedAt but, we don't know if it is associated with this SHA so do - // the SHA specific verification - if mergeCommitSha != nil && *mergeCommitSha == pr.SHA { - if pr.MergedAt != recentMergedAt { - pr.MergedAt = recentMergedAt - if res := pl.dbc.DB.Save(pr); res.Error != nil { - logger.WithError(res.Error).Errorf("unexpected error updating pull request %s (%s)", pr.Link, pr.SHA) - continue - } - } - } - - // if we see that any sha has merged for this pr then we should clear out any risk analysis pending comment records - // if we don't get them here we will catch them before writing the risk analysis comment - // but, we should clean up here if possible - pendingComments, err := pl.ghCommenter.QueryPRPendingComments(pr.Org, pr.Repo, pr.Number, models.CommentTypeRiskAnalysis) + if len(allMerged) == 0 { + return nil + } - if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { - logger.WithError(err).Error("Unable to fetch pending comments ") - } + log.WithField("count", len(allMerged)).Info("syncing merged PR status in bulk") - for _, pc := range pendingComments { - pcp := pc - pl.ghCommenter.ClearPendingRecord(pcp.Org, pcp.Repo, pcp.PullNumber, pcp.SHA, models.CommentTypeRiskAnalysis, &pcp) - } + ctx := pl.ctx + sqlDB, err := pl.dbc.DB.DB() + if err != nil { + return fmt.Errorf("getting sql.DB: %w", err) + } + conn, err := stdlib.AcquireConn(sqlDB) + if err != nil { + return fmt.Errorf("acquiring pgx conn: %w", err) + } + defer func() { + if err := stdlib.ReleaseConn(sqlDB, conn); err != nil { + log.WithError(err).Error("failed to release pgx conn") } + }() + + cleanup, err := db.CopyToTempTable(ctx, conn, "tmp_merged_prs", allMerged, + []db.TempColumn[mergedRow]{ + {Name: "org", Type: "text NOT NULL", Value: func(r *mergedRow) any { return r.Org }}, + {Name: "repo", Type: "text NOT NULL", Value: func(r *mergedRow) any { return r.Repo }}, + {Name: "number", Type: "int NOT NULL", Value: func(r *mergedRow) any { return r.Number }}, + {Name: "sha", Type: "text NOT NULL", Value: func(r *mergedRow) any { return r.SHA }}, + {Name: "merged_at", Type: "timestamptz NOT NULL", Value: func(r *mergedRow) any { return r.MergedAt }}, + }, + ) + if err != nil { + return fmt.Errorf("copying merged PRs to temp table: %w", err) + } + defer cleanup() + + res, err := conn.Exec(ctx, ` + UPDATE prow_pull_requests p + SET merged_at = t.merged_at, updated_at = NOW() + FROM tmp_merged_prs t + WHERE p.sha = t.sha AND p.merged_at IS NULL + `) + if err != nil { + return fmt.Errorf("bulk updating merged_at: %w", err) + } + log.WithField("rows", res.RowsAffected()).Info("updated prow_pull_requests merged_at") + + res, err = conn.Exec(ctx, ` + DELETE FROM pull_request_comments c + USING (SELECT DISTINCT org, repo, number FROM tmp_merged_prs) t + WHERE c.org = t.org AND c.repo = t.repo AND c.pull_number = t.number + AND c.comment_type = $1 + `, int(models.CommentTypeRiskAnalysis)) + if err != nil { + return fmt.Errorf("bulk deleting pending comments for merged PRs: %w", err) } + log.WithField("rows", res.RowsAffected()).Info("deleted pending risk analysis comments for merged PRs") return nil }