TRT-2799: syncPRStatus: Fix SHA mismatch bug (MergeCommitSHA vs Head.SHA)

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()
 

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)}
 

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
 }