diff --git a/apps/backend/internal/github/mock_client.go b/apps/backend/internal/github/mock_client.go index 15920a067b..6f0e8a7fa6 100644 --- a/apps/backend/internal/github/mock_client.go +++ b/apps/backend/internal/github/mock_client.go @@ -5,6 +5,7 @@ import ( "fmt" "strings" "sync" + "sync/atomic" "time" ) @@ -77,6 +78,18 @@ type MockClient struct { gists map[string]mockGist deletedGists []string nextGistID int + + // findPRByBranchCalls counts FindPRByBranch invocations so tests can + // assert that branch-detection probes are throttled. Atomic because + // FindPRByBranch otherwise only takes a read lock. + findPRByBranchCalls atomic.Int64 + + // probeEntered/probeRelease let a test gate FindPRByBranch: when set, each + // invocation signals on probeEntered and then blocks until probeRelease is + // closed. Used to force concurrent probes to overlap and assert + // singleflight coalescing. + probeEntered chan string + probeRelease chan struct{} } // mockGist captures a gist that was created via the mock client so tests @@ -133,12 +146,38 @@ func (m *MockClient) GetPR(_ context.Context, owner, repo string, number int) (* } func (m *MockClient) FindPRByBranch(_ context.Context, owner, repo, branch string) (*PR, error) { + m.findPRByBranchCalls.Add(1) m.mu.RLock() - defer m.mu.RUnlock() pr := m.prsByBranch[branchKey{owner, repo, branch}] + entered, release := m.probeEntered, m.probeRelease + m.mu.RUnlock() + // Gate (if a test installed one) outside the lock so a blocked probe + // doesn't wedge other mock operations. + if entered != nil { + entered <- branch + } + if release != nil { + <-release + } return pr, nil } +// FindPRByBranchCallCount returns how many times FindPRByBranch has been +// called. Used by tests asserting detection-probe throttling. +func (m *MockClient) FindPRByBranchCallCount() int { + return int(m.findPRByBranchCalls.Load()) +} + +// GateFindPRByBranch installs a gate around FindPRByBranch: each invocation +// signals on entered, then blocks until release is closed. Pass nil/nil to +// remove the gate. Used to force concurrent probes to overlap. +func (m *MockClient) GateFindPRByBranch(entered chan string, release chan struct{}) { + m.mu.Lock() + defer m.mu.Unlock() + m.probeEntered = entered + m.probeRelease = release +} + func (m *MockClient) ListAuthoredPRs(_ context.Context, owner, repo string) ([]*PR, error) { m.mu.RLock() defer m.mu.RUnlock() @@ -527,6 +566,9 @@ func (m *MockClient) Reset() { m.gists = make(map[string]mockGist) m.deletedGists = nil m.nextGistID = 0 + m.findPRByBranchCalls.Store(0) + m.probeEntered = nil + m.probeRelease = nil } // SubmittedReviews returns all recorded SubmitReview calls. diff --git a/apps/backend/internal/github/service_pr_sync_test.go b/apps/backend/internal/github/service_pr_sync_test.go index d1be09f36e..31420f5841 100644 --- a/apps/backend/internal/github/service_pr_sync_test.go +++ b/apps/backend/internal/github/service_pr_sync_test.go @@ -2,7 +2,9 @@ package github import ( "context" + "sync" "testing" + "testing/synctest" ) func TestTriggerPRSync_SyncsExistingWatch(t *testing.T) { @@ -108,3 +110,102 @@ func TestTriggerPRSync_NoWatch(t *testing.T) { t.Errorf("expected nil TaskPR for task without watch, got %+v", result) } } + +// TestTriggerPRSyncAll_ThrottlesDetectionProbe reproduces the log-flood bug: +// a pr_number=0 watch on a branch that has no PR (e.g. an unresolvable repo) +// was re-probing GitHub on every on-demand sync, because the detection path — +// unlike the status-sync path — had no freshness guard and never stamped +// last_checked_at. The frontend re-syncs every 5s while no PR is found, so +// each repeated sync hit `gh` again and logged a warning. After the fix the +// first sync probes once and stamps the watch; the immediate second sync is +// throttled within prSyncFreshnessWindow and must NOT probe again. +func TestTriggerPRSyncAll_ThrottlesDetectionProbe(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + + // Watch with pr_number=0 on a branch that has no PR in the mock — every + // FindPRByBranch returns (nil, nil), mirroring "PR not found yet". + watch := &PRWatch{ + SessionID: "s1", + TaskID: "t1", + Owner: "org", + Repo: "missing", + PRNumber: 0, + Branch: "feat/never-merged", + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatal(err) + } + + if _, err := svc.TriggerPRSyncAll(ctx, "t1"); err != nil { + t.Fatalf("first TriggerPRSyncAll: %v", err) + } + if got := mockClient.FindPRByBranchCallCount(); got != 1 { + t.Fatalf("expected exactly 1 detection probe on first sync, got %d", got) + } + + // Second sync immediately after — well within prSyncFreshnessWindow. + if _, err := svc.TriggerPRSyncAll(ctx, "t1"); err != nil { + t.Fatalf("second TriggerPRSyncAll: %v", err) + } + if got := mockClient.FindPRByBranchCallCount(); got != 1 { + t.Errorf("expected detection probe to be throttled on second sync (still 1 call), got %d", got) + } +} + +// TestTriggerPRDetection_CoalescesConcurrentProbes verifies that parallel +// syncs for the same watch collapse to a single GitHub probe. The freshness +// guard alone is racy — concurrent callers can all read a stale +// last_checked_at and probe simultaneously — so detection runs inside a +// per-watch singleflight. We force overlap by blocking the probe: the leader +// enters FindPRByBranch and parks; the followers must coalesce into it rather +// than issue their own probes. +// +// synctest.Wait() supplies the otherwise-unobservable "all followers have +// joined the singleflight" signal: it returns only once every goroutine is +// durably blocked (the leader on the release channel, the followers inside +// singleflight). That makes the coalescing assertion deterministic without a +// fixed sleep. +func TestTriggerPRDetection_CoalescesConcurrentProbes(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + + watch := &PRWatch{ + SessionID: "s1", + TaskID: "t1", + Owner: "org", + Repo: "missing", + PRNumber: 0, + Branch: "feat/never-merged", + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatal(err) + } + + const concurrency = 6 + release := make(chan struct{}) + mockClient.GateFindPRByBranch(nil, release) // block the leader inside FindPRByBranch + + var wg sync.WaitGroup + wg.Add(concurrency) + for range concurrency { + go func() { + defer wg.Done() + if _, err := svc.TriggerPRSyncAll(ctx, "t1"); err != nil { + t.Errorf("TriggerPRSyncAll: %v", err) + } + }() + } + + // Block until the leader is parked in FindPRByBranch and every other + // caller has coalesced into its singleflight call. + synctest.Wait() + close(release) + wg.Wait() + + if got := mockClient.FindPRByBranchCallCount(); got != 1 { + t.Errorf("expected concurrent detection probes to coalesce to 1 GitHub call, got %d", got) + } + }) +} diff --git a/apps/backend/internal/github/service_pr_watch.go b/apps/backend/internal/github/service_pr_watch.go index f109cfd384..7bc646f332 100644 --- a/apps/backend/internal/github/service_pr_watch.go +++ b/apps/backend/internal/github/service_pr_watch.go @@ -562,7 +562,13 @@ func (s *Service) TriggerPRSyncAll(ctx context.Context, taskID string) ([]*TaskP tp, syncErr = s.triggerPRStatusSync(ctx, w, taskID) } if syncErr != nil { - s.logger.Warn("per-repo PR sync failed", + // Debug, not Warn: this is best-effort background reconciliation and + // the most common failure — a branch with no PR yet, or an + // unresolvable repo — is an expected steady state, not an + // operational warning. The background poller logs the same failure + // at Debug (see poller.detectPRForWatch / checkSinglePRWatch); the + // WS handler still returns the error to the caller. + s.logger.Debug("per-repo PR sync failed", zap.String("task_id", taskID), zap.String("repository_id", w.RepositoryID), zap.Int("pr_number", w.PRNumber), @@ -580,7 +586,58 @@ func (s *Service) triggerPRDetection(ctx context.Context, watch *PRWatch, taskID if s.client == nil { return nil, nil } + // Coalesce concurrent detection probes for the same watch. Without this the + // freshness check below is racy: parallel callers (e.g. the 5s frontend + // retry racing the workspace background refresh) can all read a stale + // last_checked_at and probe GitHub simultaneously, defeating the throttle. + // Keyed distinctly from triggerPRStatusSync's owner/repo/number key so the + // two never collide. singleflight only blocks same-key calls, so the nested + // triggerPRStatusSync (different key) below cannot deadlock. + key := "pr-detect:" + watch.ID + v, err, _ := s.syncGroup.Do(key, func() (interface{}, error) { + return s.detectPRForWatchOnce(ctx, watch, taskID) + }) + if err != nil { + return nil, err + } + if v == nil { + return nil, nil + } + return v.(*TaskPR), nil +} + +// detectPRForWatchOnce performs a single branch-detection probe for a +// searching (pr_number=0) watch. It runs inside triggerPRDetection's +// singleflight so only one probe per watch is in flight at a time. +func (s *Service) detectPRForWatchOnce(ctx context.Context, watch *PRWatch, taskID string) (*TaskPR, error) { + // Throttle branch-detection probes. A watch still searching for its PR + // (pr_number=0) has no TaskPR row yet, so triggerPRStatusSync's freshness + // check can't gate it — we gate on the watch's own last_checked_at instead. + // Without this, a branch whose PR never appears (e.g. an unresolvable repo) + // re-hits `gh` on every on-demand sync, and the frontend re-syncs every 5s + // while no PR is found — flooding the logs with identical failures. + if watch.LastCheckedAt != nil && time.Since(*watch.LastCheckedAt) < prSyncFreshnessWindow { + return s.store.GetTaskPRByRepository(ctx, taskID, watch.RepositoryID) + } + pr, err := s.client.FindPRByBranch(ctx, watch.Owner, watch.Repo, watch.Branch) + // Stamp last_checked_at regardless of outcome — including on error. The + // flood this fixes IS the error path (an unresolvable repo makes `gh` exit + // non-zero), so stamping only on success would leave the bug unfixed. The + // tradeoff: a transient GitHub error throttles the next on-demand probe for + // prSyncFreshnessWindow, which is fine — it stops hammering a failing + // endpoint, and the 60s background poller still retries. This diverges + // intentionally from poller.detectPRForWatch, which stamps only on success. + // The empty-string check/review args clear last_check_status / + // last_review_state. That's harmless here — this path only runs for + // pr_number=0 (searching) watches, which never carry those fields — and + // matches the poller's detection stamp. Only call this on searching + // watches; for a resolved PR use the status-sync path instead. + now := time.Now().UTC() + if tsErr := s.store.UpdatePRWatchTimestamps(ctx, watch.ID, now, nil, "", ""); tsErr != nil { + s.logger.Debug("failed to stamp PR watch after detection probe", + zap.String("watch_id", watch.ID), zap.Error(tsErr)) + } if err != nil || pr == nil { return nil, err }