fix: avoid stale sibling PR sync (#1302)
This commit is contained in:
@@ -832,12 +832,11 @@ func (s *Service) detectPRForWatchOnce(ctx context.Context, watch *PRWatch, task
|
||||
}
|
||||
|
||||
func (s *Service) triggerPRStatusSync(ctx context.Context, watch *PRWatch, taskID string) (*TaskPR, error) {
|
||||
// Freshness check: skip GitHub API if recently synced. Look up by the
|
||||
// watch's own (task, repo) — the legacy GetTaskPR(ctx, taskID) returned
|
||||
// "first match" which for multi-repo tasks would mistakenly hit the
|
||||
// other repo's row and skip the sync that this watch actually needs.
|
||||
// Freshness check: skip GitHub API if this exact PR row was recently
|
||||
// synced. Multi-branch tasks can have multiple PRs in the same repo, so a
|
||||
// repo-only lookup would let a fresh sibling suppress this PR's sync.
|
||||
loadTaskPR := func(c context.Context) (*TaskPR, error) {
|
||||
tp, err := s.store.GetTaskPRByRepository(c, taskID, watch.RepositoryID)
|
||||
tp, err := s.store.GetTaskPRByRepoAndNumber(c, taskID, watch.RepositoryID, watch.PRNumber)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -846,6 +845,9 @@ func (s *Service) triggerPRStatusSync(ctx context.Context, watch *PRWatch, taskI
|
||||
}
|
||||
// Fall back to the legacy untagged row for single-repo tasks that
|
||||
// haven't been re-associated under the multi-repo schema yet.
|
||||
if watch.RepositoryID != "" {
|
||||
return nil, nil
|
||||
}
|
||||
return s.store.GetTaskPR(c, taskID)
|
||||
}
|
||||
if tp, _ := loadTaskPR(ctx); tp != nil && tp.LastSyncedAt != nil {
|
||||
@@ -880,6 +882,17 @@ func (s *Service) triggerPRStatusSync(ctx context.Context, watch *PRWatch, taskI
|
||||
if status == nil {
|
||||
return loadTaskPR(bgCtx)
|
||||
}
|
||||
if existing, loadErr := loadTaskPR(bgCtx); loadErr != nil {
|
||||
return nil, loadErr
|
||||
} else if existing == nil && status.PR != nil {
|
||||
// Gap-fill a numbered watch whose exact task_pr row was never
|
||||
// created. AssociatePRWithTask publishes the creation event; the
|
||||
// following SyncTaskPR may publish again if status fields changed.
|
||||
// That double event is harmless because clients re-fetch state.
|
||||
if _, assocErr := s.AssociatePRWithTask(bgCtx, taskID, watch.RepositoryID, status.PR); assocErr != nil {
|
||||
return nil, assocErr
|
||||
}
|
||||
}
|
||||
if syncErr := s.SyncTaskPR(bgCtx, taskID, status); syncErr != nil {
|
||||
return nil, syncErr
|
||||
}
|
||||
|
||||
@@ -241,6 +241,25 @@ func (s *Service) applyBatchedNumberedWatch(
|
||||
if err := s.store.UpdatePRWatchTimestamps(ctx, w.ID, now, nil, status.ChecksState, status.ReviewState); err != nil {
|
||||
s.logger.Error("failed to update PR watch timestamps", zap.String("id", w.ID), zap.Error(err))
|
||||
}
|
||||
// Gap-fill: a numbered watch can exist even when its exact task_pr row was
|
||||
// never created. This targeted read is unconditional because the common
|
||||
// existing-row path is cheap, and the missing-row path must repair before
|
||||
// SyncTaskPR. If AssociatePRWithTask creates the row, it publishes the
|
||||
// creation event; the following SyncTaskPR may publish a second event when
|
||||
// status fields changed. That double event is harmless because clients
|
||||
// re-fetch the task PR state.
|
||||
if existing, err := s.store.GetTaskPRByRepoAndNumber(ctx, w.TaskID, w.RepositoryID, w.PRNumber); err != nil {
|
||||
s.logger.Error("failed to load exact task PR",
|
||||
zap.String("task_id", w.TaskID), zap.String("repository_id", w.RepositoryID),
|
||||
zap.Int("pr_number", w.PRNumber), zap.Error(err))
|
||||
return PRWatchSyncResult{Watch: w, Status: status, Found: true, SyncFailed: true}
|
||||
} else if existing == nil && status.PR != nil {
|
||||
if _, assocErr := s.AssociatePRWithTask(ctx, w.TaskID, w.RepositoryID, status.PR); assocErr != nil {
|
||||
s.logger.Error("failed to associate numbered PR with task",
|
||||
zap.String("task_id", w.TaskID), zap.Int("pr_number", w.PRNumber), zap.Error(assocErr))
|
||||
return PRWatchSyncResult{Watch: w, Status: status, Found: true, SyncFailed: true}
|
||||
}
|
||||
}
|
||||
if syncErr := s.SyncTaskPR(ctx, w.TaskID, status); syncErr != nil {
|
||||
s.logger.Error("failed to sync task PR", zap.String("task_id", w.TaskID), zap.Error(syncErr))
|
||||
// SyncFailed=true so poller skips publishing PR feedback while the
|
||||
|
||||
@@ -3,6 +3,7 @@ package github
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestCreatePRWatch_AllowsMultipleBranchesPerRepo locks the multi-branch
|
||||
@@ -256,3 +257,152 @@ func TestUpdatePRWatchBranchIfSearching_MissingRow_NoOp(t *testing.T) {
|
||||
t.Fatalf("must be a no-op for missing row, got: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTriggerPRStatusSync_AssociatesExactPRWhenSiblingIsFresh(t *testing.T) {
|
||||
_, svc, mockClient, store := setupPollerTest(t)
|
||||
ctx := context.Background()
|
||||
seedTask(t, store, "task-1", false)
|
||||
|
||||
now := time.Now().UTC()
|
||||
mergedAt := now.Add(-30 * time.Minute)
|
||||
if err := store.CreateTaskPR(ctx, &TaskPR{
|
||||
TaskID: "task-1",
|
||||
RepositoryID: "repo-1",
|
||||
Owner: "owner",
|
||||
Repo: "repo",
|
||||
PRNumber: 1293,
|
||||
PRURL: "https://github.com/owner/repo/pull/1293",
|
||||
PRTitle: "First",
|
||||
HeadBranch: "feature/first",
|
||||
BaseBranch: "main",
|
||||
State: "merged",
|
||||
CreatedAt: now.Add(-2 * time.Hour),
|
||||
MergedAt: &mergedAt,
|
||||
ClosedAt: &mergedAt,
|
||||
LastSyncedAt: &now,
|
||||
}); err != nil {
|
||||
t.Fatalf("seed merged sibling: %v", err)
|
||||
}
|
||||
|
||||
watch, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 1299, "feature/second")
|
||||
if err != nil {
|
||||
t.Fatalf("CreatePRWatch: %v", err)
|
||||
}
|
||||
mockClient.AddPR(&PR{
|
||||
Number: 1299,
|
||||
Title: "Second",
|
||||
HTMLURL: "https://github.com/owner/repo/pull/1299",
|
||||
State: "open",
|
||||
HeadBranch: "feature/second",
|
||||
BaseBranch: "main",
|
||||
RepoOwner: "owner",
|
||||
RepoName: "repo",
|
||||
CreatedAt: now.Add(-1 * time.Hour),
|
||||
UpdatedAt: now,
|
||||
})
|
||||
|
||||
got, err := svc.triggerPRStatusSync(ctx, watch, "task-1")
|
||||
if err != nil {
|
||||
t.Fatalf("triggerPRStatusSync: %v", err)
|
||||
}
|
||||
if got == nil {
|
||||
t.Fatal("expected TaskPR result")
|
||||
}
|
||||
if got.PRNumber != 1299 || got.State != "open" {
|
||||
t.Fatalf("sync returned PR #%d state=%q, want PR #1299 open", got.PRNumber, got.State)
|
||||
}
|
||||
|
||||
all, err := store.ListTaskPRsByTask(ctx, "task-1")
|
||||
if err != nil {
|
||||
t.Fatalf("ListTaskPRsByTask: %v", err)
|
||||
}
|
||||
if len(all) != 2 {
|
||||
t.Fatalf("expected sync to create exact sibling row, got %d rows", len(all))
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyBatchedNumberedWatch_AssociatesExactPRWhenSiblingExists(t *testing.T) {
|
||||
_, svc, _, store := setupPollerTest(t)
|
||||
ctx := context.Background()
|
||||
seedTask(t, store, "task-1", false)
|
||||
|
||||
now := time.Now().UTC()
|
||||
if err := store.CreateTaskPR(ctx, &TaskPR{
|
||||
TaskID: "task-1",
|
||||
RepositoryID: "repo-1",
|
||||
Owner: "owner",
|
||||
Repo: "repo",
|
||||
PRNumber: 1293,
|
||||
PRURL: "https://github.com/owner/repo/pull/1293",
|
||||
PRTitle: "First",
|
||||
HeadBranch: "feature/first",
|
||||
BaseBranch: "main",
|
||||
State: "merged",
|
||||
CreatedAt: now.Add(-2 * time.Hour),
|
||||
LastSyncedAt: &now,
|
||||
}); err != nil {
|
||||
t.Fatalf("seed merged sibling: %v", err)
|
||||
}
|
||||
|
||||
watch, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 1299, "feature/second")
|
||||
if err != nil {
|
||||
t.Fatalf("CreatePRWatch: %v", err)
|
||||
}
|
||||
status := &PRStatus{
|
||||
PR: &PR{
|
||||
Number: 1299,
|
||||
Title: "Second",
|
||||
HTMLURL: "https://github.com/owner/repo/pull/1299",
|
||||
State: "open",
|
||||
HeadBranch: "feature/second",
|
||||
BaseBranch: "main",
|
||||
RepoOwner: "owner",
|
||||
RepoName: "repo",
|
||||
CreatedAt: now.Add(-1 * time.Hour),
|
||||
UpdatedAt: now,
|
||||
},
|
||||
ChecksState: "pending",
|
||||
}
|
||||
|
||||
result := svc.applyBatchedNumberedWatch(ctx, watch, map[string]*PRStatus{
|
||||
prStatusCacheKey("owner", "repo", 1299): status,
|
||||
}, now)
|
||||
if result.SyncFailed {
|
||||
t.Fatal("batched sync should not fail")
|
||||
}
|
||||
|
||||
got, err := store.GetTaskPRByRepoAndNumber(ctx, "task-1", "repo-1", 1299)
|
||||
if err != nil {
|
||||
t.Fatalf("GetTaskPRByRepoAndNumber: %v", err)
|
||||
}
|
||||
if got == nil || got.State != "open" {
|
||||
t.Fatalf("expected exact PR #1299 open row, got %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyBatchedNumberedWatch_MissingPRDataDoesNotPanic(t *testing.T) {
|
||||
_, svc, _, store := setupPollerTest(t)
|
||||
ctx := context.Background()
|
||||
seedTask(t, store, "task-1", false)
|
||||
|
||||
now := time.Now().UTC()
|
||||
watch, err := svc.CreatePRWatch(ctx, "session-1", "task-1", "repo-1", "owner", "repo", 1299, "feature/second")
|
||||
if err != nil {
|
||||
t.Fatalf("CreatePRWatch: %v", err)
|
||||
}
|
||||
|
||||
result := svc.applyBatchedNumberedWatch(ctx, watch, map[string]*PRStatus{
|
||||
prStatusCacheKey("owner", "repo", 1299): &PRStatus{ChecksState: "pending"},
|
||||
}, now)
|
||||
if !result.SyncFailed {
|
||||
t.Fatal("batched sync should fail without PR data")
|
||||
}
|
||||
|
||||
got, err := store.GetTaskPRByRepoAndNumber(ctx, "task-1", "repo-1", 1299)
|
||||
if err != nil {
|
||||
t.Fatalf("GetTaskPRByRepoAndNumber: %v", err)
|
||||
}
|
||||
if got != nil {
|
||||
t.Fatalf("expected no task PR row without PR data, got %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user