fix(github): throttle PR branch-detection probes to stop log flood (#1135)
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user