From 2bebffb05e1f9da8b235dea6f811a36173806e0a Mon Sep 17 00:00:00 2001 From: Eric Dahl Date: Mon, 28 Sep 2026 22:55:13 -0400 Subject: [PATCH] refactor: let Pacing absorb the token bookkeeping ratelimit held only the Interval formula while ~120 lines of per-token state (tokenPace, fetchCost, due, repriceTokens, Throttles) lived in the poller, beside a second 403 backoff keyed by Org. ratelimit.Pacer now owns all of it, keyed by token only: Due, LimitedUntil, Observe, Failed, EndCycle, Throttles, with an injected clock. A 403 now holds the token rather than the Org, so Orgs sharing a token back off together instead of each collecting its own 403. Closes #51 --- CONTEXT.md | 3 + internal/poller/poller.go | 252 +++++-------------------------- internal/ratelimit/pacer.go | 196 ++++++++++++++++++++++++ internal/ratelimit/pacer_test.go | 117 ++++++++++++++ internal/state/state.go | 9 +- 5 files changed, 357 insertions(+), 220 deletions(-) create mode 100644 internal/ratelimit/pacer.go create mode 100644 internal/ratelimit/pacer_test.go diff --git a/CONTEXT.md b/CONTEXT.md index e8a68dd..7594108 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -165,6 +165,9 @@ interval = max(configured, time to reset / affordable) so they are paced together; pacing them separately would spend it twice. - A healthy token is never held back: while there is budget to spare the formula lands below the configured interval, and the configured one wins. +- **Rate limited**: when GitHub refuses a token anyway (a 403 with a reset, + or a secondary limit with retry-after), that token is held until the reset. + Its Repos go stale with the reason, unlike paced Repos, which just wait. When a token is paced below its configured interval, the title bar says so — which Org, how much budget is left, the pace, and when it recovers. Healthy diff --git a/internal/poller/poller.go b/internal/poller/poller.go index 5a8664d..1cfd23e 100644 --- a/internal/poller/poller.go +++ b/internal/poller/poller.go @@ -2,15 +2,10 @@ package poller import ( "context" - "errors" "fmt" - "sort" - "strings" "sync" "time" - "github.com/google/go-github/v72/github" - "github.com/ericdahl-dev/git-green/internal/aggregator" "github.com/ericdahl-dev/git-green/internal/config" githubclient "github.com/ericdahl-dev/git-green/internal/github" @@ -49,14 +44,9 @@ type Poller struct { // expanded holds the "owner/name" of repos whose rows are open in the UI. // Only those fetch per-run jobs and per-PR runs; see RepoQuery.Detail. expanded map[string]bool - // paced tracks each token's REST budget and when it may next be polled. - // Keyed by token rather than by org because orgs sharing a token share a - // budget, and pacing them independently would spend it twice over. - paced map[string]*tokenPace - // rateLimitedUntil records, per org, when GitHub said the quota resets. - // Polling that org is skipped until then rather than spending every cycle - // collecting 403s. - rateLimitedUntil map[string]time.Time + // pacer paces each token against its REST budget and holds tokens + // GitHub has rate-limited until their reset. + pacer *ratelimit.Pacer // now is swappable in tests so threshold crossings can be exercised // without waiting on the wall clock. now func() time.Time @@ -74,17 +64,18 @@ func New(cfg *config.Config, factory ClientFactory) *Poller { Stoplight: aggregator.StoplightGrey, } } - return &Poller{ - cfg: cfg, - factory: factory, - dispatcher: webhooks.New(cfg.Webhooks), - current: repos, - stuck: make(map[string]*stuckEntry), - paced: make(map[string]*tokenPace), - expanded: make(map[string]bool), - rateLimitedUntil: make(map[string]time.Time), - now: time.Now, + p := &Poller{ + cfg: cfg, + factory: factory, + dispatcher: webhooks.New(cfg.Webhooks), + current: repos, + stuck: make(map[string]*stuckEntry), + expanded: make(map[string]bool), + now: time.Now, } + // Read the clock through p.now so tests that swap it move pacing too. + p.pacer = ratelimit.NewPacer(func() time.Time { return p.now() }) + return p } // SetExpandedRepos records which repo rows are open in the UI, as @@ -105,166 +96,9 @@ func (p *Poller) isExpanded(owner, name string) bool { return p.expanded[owner+"/"+name] } -// rateLimitPause reports how long an org is still rate-limited for, and clears -// the entry once it has elapsed. -func (p *Poller) rateLimitPause(org string) (time.Duration, bool) { - p.mu.Lock() - defer p.mu.Unlock() - until, ok := p.rateLimitedUntil[org] - if !ok { - return 0, false - } - if remaining := until.Sub(p.now()); remaining > 0 { - return remaining, true - } - delete(p.rateLimitedUntil, org) - return 0, false -} - -// noteRateLimit records a reset time reported by GitHub so subsequent polls -// for that org back off instead of burning cycles on 403s. -func (p *Poller) noteRateLimit(org string, err error) { - var rlErr *github.RateLimitError - var abuseErr *github.AbuseRateLimitError - var until time.Time - switch { - case errors.As(err, &rlErr): - until = rlErr.Rate.Reset.Time - case errors.As(err, &abuseErr): - if abuseErr.RetryAfter != nil { - until = p.now().Add(*abuseErr.RetryAfter) - } - default: - return - } - if until.IsZero() { - return - } - p.mu.Lock() - p.rateLimitedUntil[org] = until - p.mu.Unlock() - logx.Debug("rate limited", "org", org, "until", until) -} - -// tokenPace is one token's budget and the pace it buys. -type tokenPace struct { - budget ratelimit.Budget - cost int // REST calls one full cycle costs this token - interval time.Duration - nextDue time.Time - orgs map[string]bool -} - -// fetchCost is what one repo's fetch spent, reported back so the cycle can be -// priced per token. -type fetchCost struct { - token string - calls int - budget ratelimit.Budget - spent bool // false when the repo was skipped or never reached the API - skipped bool -} - -// due reports whether a token may be polled this cycle. An unpaced token — one -// that has never reported a budget, or is comfortably inside it — is always due. -func (p *Poller) due(token string) bool { - p.mu.Lock() - defer p.mu.Unlock() - pace, ok := p.paced[token] - if !ok || pace.nextDue.IsZero() { - return true - } - return !p.now().Before(pace.nextDue) -} - -// repriceTokens folds a cycle's spending into each token's pace. A token that -// was skipped this cycle keeps the pace it already had. -func (p *Poller) repriceTokens(costs []fetchCost) { - configured := time.Duration(p.cfg.Settings.PollInterval) * time.Second - now := p.now() - - p.mu.Lock() - defer p.mu.Unlock() - for _, c := range costs { - if !c.spent || c.token == "" { - continue - } - pace, ok := p.paced[c.token] - if !ok { - pace = &tokenPace{orgs: make(map[string]bool)} - p.paced[c.token] = pace - } - pace.cost += c.calls - // Repos race, so take the leanest reading of the cycle: it is the one - // closest to what is actually left. - if c.budget.Known() && (!pace.budget.Known() || c.budget.Remaining < pace.budget.Remaining) { - pace.budget = c.budget - } - } - - for token, pace := range p.paced { - if !anySpent(costs, token) { - continue - } - interval, throttled := ratelimit.Interval(configured, pace.budget, pace.cost, now) - pace.interval = interval - if throttled { - pace.nextDue = now.Add(interval) - logx.Debug("throttling token", "orgs", orgList(pace.orgs), - "remaining", pace.budget.Remaining, "cost", pace.cost, "interval", interval) - } else { - // Healthy tokens are never held back: the ticker alone decides - // when they poll, exactly as before pacing existed. - pace.nextDue = time.Time{} - } - // The cost is re-measured every cycle: expanding a repo changes it. - pace.cost = 0 - } -} - -func anySpent(costs []fetchCost, token string) bool { - for _, c := range costs { - if c.spent && c.token == token { - return true - } - } - return false -} - -func orgList(orgs map[string]bool) []string { - out := make([]string, 0, len(orgs)) - for o := range orgs { - out = append(out, o) - } - sort.Strings(out) - return out -} - // Throttles reports every token currently polling slower than configured, for -// the title bar. Healthy tokens are omitted — there is nothing to say. -func (p *Poller) Throttles() []state.Throttle { - configured := time.Duration(p.cfg.Settings.PollInterval) * time.Second - p.mu.Lock() - defer p.mu.Unlock() - - var out []state.Throttle - for _, pace := range p.paced { - if pace.interval <= configured || !pace.budget.Known() { - continue - } - out = append(out, state.Throttle{ - Orgs: orgList(pace.orgs), - Remaining: pace.budget.Remaining, - Limit: pace.budget.Limit, - Reset: pace.budget.Reset, - Interval: pace.interval, - }) - } - sort.Slice(out, func(a, b int) bool { - return strings.Join(out[a].Orgs, ",") < strings.Join(out[b].Orgs, ",") - }) - return out -} +// the title bar. +func (p *Poller) Throttles() []state.Throttle { return p.pacer.Throttles() } // Snapshot returns an immutable view of the current (possibly initial) state. func (p *Poller) Snapshot() state.Snapshot { @@ -322,8 +156,6 @@ func (p *Poller) fetch(ctx context.Context, ch chan<- state.Snapshot) { copy(previous, p.current) p.mu.Unlock() - costs := make([]fetchCost, len(enabled)) - for i, repo := range enabled { wg.Add(1) go func(i int, repo config.Repo) { @@ -339,17 +171,16 @@ func (p *Poller) fetch(ctx context.Context, ch chan<- state.Snapshot) { // A token polling below the configured rate skips its repos this // cycle. Their last-known state stands rather than going stale: // nothing failed, the dashboard is just refreshing less often. - if token, err := p.cfg.TokenForOrg(repo.Owner); err == nil && !p.due(token) { + if p.pacedOut(repo.Owner) { results[i] = prev - costs[i] = fetchCost{token: token, skipped: true} return } - results[i], costs[i] = p.fetchRepo(ctx, repo, prev) + results[i] = p.fetchRepo(ctx, repo, prev) }(i, repo) } wg.Wait() - p.repriceTokens(costs) + p.pacer.EndCycle(time.Duration(p.cfg.Settings.PollInterval) * time.Second) p.mu.Lock() p.current = results @@ -371,7 +202,19 @@ func (p *Poller) fetch(ctx context.Context, ch chan<- state.Snapshot) { } } -func (p *Poller) fetchRepo(ctx context.Context, repo config.Repo, prev state.RepoState) (state.RepoState, fetchCost) { +// pacedOut reports whether an Org's token is being paced this cycle. A token +// GitHub rate-limited is not paced out: its Repos still go through fetchRepo, +// which marks them stale with the reason. +func (p *Poller) pacedOut(org string) bool { + token, err := p.cfg.TokenForOrg(org) + if err != nil || p.pacer.Due(token) { + return false + } + _, limited := p.pacer.LimitedUntil(token) + return !limited +} + +func (p *Poller) fetchRepo(ctx context.Context, repo config.Repo, prev state.RepoState) state.RepoState { logx.Debug("fetch repo", "owner", repo.Owner, "name", repo.Name) // Keep whatever branch was already resolved. Falling back to the config // value (often empty) would throw the resolution away on every error and @@ -382,8 +225,7 @@ func (p *Poller) fetchRepo(ctx context.Context, repo config.Repo, prev state.Rep resolved = prev.Branch } - var cost fetchCost - stale := func(err error) (state.RepoState, fetchCost) { + stale := func(err error) state.RepoState { now := time.Now() return state.RepoState{ Owner: repo.Owner, @@ -394,19 +236,16 @@ func (p *Poller) fetchRepo(ctx context.Context, repo config.Repo, prev state.Rep PRs: prev.PRs, StaleAt: &now, Err: err, - }, cost - } - - if pause, limited := p.rateLimitPause(repo.Owner); limited { - return stale(fmt.Errorf("rate limited for %s; retrying in %s", repo.Owner, pause.Round(time.Second))) + } } token, err := p.cfg.TokenForOrg(repo.Owner) if err != nil { return stale(err) } - cost.token = token - p.noteTokenOrg(token, repo.Owner) + if until, limited := p.pacer.LimitedUntil(token); limited { + return stale(fmt.Errorf("rate limited for %s; retrying in %s", repo.Owner, until.Sub(p.now()).Round(time.Second))) + } client := p.factory(token) // Use the previously-resolved branch so we avoid a Repositories.Get call every poll. @@ -419,9 +258,9 @@ func (p *Poller) fetchRepo(ctx context.Context, repo config.Repo, prev state.Rep } data, err := client.FetchAll(ctx, q) - cost.calls, cost.budget, cost.spent = data.Calls, data.Budget, true + p.pacer.Observe(token, repo.Owner, data.Calls, data.Budget) if err != nil { - p.noteRateLimit(repo.Owner, err) + p.pacer.Failed(token, err) return stale(err) } @@ -452,20 +291,7 @@ func (p *Poller) fetchRepo(ctx context.Context, repo config.Repo, prev state.Rep PRs: prStates, StaleAt: nil, Err: nil, - }, cost -} - -// noteTokenOrg remembers which orgs ride on a token, so a throttle notice can -// name them. -func (p *Poller) noteTokenOrg(token, org string) { - p.mu.Lock() - defer p.mu.Unlock() - pace, ok := p.paced[token] - if !ok { - pace = &tokenPace{orgs: make(map[string]bool)} - p.paced[token] = pace } - pace.orgs[org] = true } // stuckKey identifies one stuck condition: a repo's default branch, or one of diff --git a/internal/ratelimit/pacer.go b/internal/ratelimit/pacer.go new file mode 100644 index 0000000..15bb022 --- /dev/null +++ b/internal/ratelimit/pacer.go @@ -0,0 +1,196 @@ +package ratelimit + +import ( + "errors" + "sort" + "strings" + "sync" + "time" + + "github.com/google/go-github/v72/github" + + "github.com/ericdahl-dev/git-green/internal/logx" +) + +// Throttle reports one token polling slower than configured because its REST +// budget is running down. +type Throttle struct { + Orgs []string + Remaining int + Limit int + Reset time.Time + Interval time.Duration +} + +// Pacer paces every token against its own REST budget. It is keyed by token, +// never by Org: Orgs sharing a token share one budget, and pacing them +// separately would spend it twice. +// +// Each poll cycle, the poller asks Due before fetching a token's Repos, +// reports each fetch with Observe (or Failed, for a 403), and closes the +// cycle with EndCycle, which re-prices every token that spent. It is safe for +// concurrent use by the fetches of one cycle. +type Pacer struct { + mu sync.Mutex + now func() time.Time + configured time.Duration // as of the last EndCycle, for Throttles + tokens map[string]*pace +} + +type pace struct { + orgs map[string]bool + budget Budget + interval time.Duration + nextDue time.Time // zero while healthy: the ticker alone decides + limited time.Time // GitHub's 403 reset; zero when not rate limited + + // This cycle's spending, folded in by EndCycle. + spent bool + cost int + low Budget // leanest budget reported this cycle +} + +// NewPacer returns a Pacer reading time from now. +func NewPacer(now func() time.Time) *Pacer { + return &Pacer{now: now, tokens: make(map[string]*pace)} +} + +func (p *Pacer) token(t string) *pace { + pc, ok := p.tokens[t] + if !ok { + pc = &pace{orgs: make(map[string]bool)} + p.tokens[t] = pc + } + return pc +} + +// Due reports whether a token may be polled now. A token that has never +// reported a budget, or is comfortably inside it, is always due. +func (p *Pacer) Due(token string) bool { + p.mu.Lock() + defer p.mu.Unlock() + pc, ok := p.tokens[token] + if !ok { + return true + } + now := p.now() + if now.Before(pc.limited) { + return false + } + return pc.nextDue.IsZero() || !now.Before(pc.nextDue) +} + +// LimitedUntil reports when a rate-limited token may be polled again. Unlike a +// paced token, whose Repos simply wait, a rate-limited one was refused by +// GitHub, and its Repos show that. +func (p *Pacer) LimitedUntil(token string) (time.Time, bool) { + p.mu.Lock() + defer p.mu.Unlock() + pc, ok := p.tokens[token] + if !ok || !p.now().Before(pc.limited) { + return time.Time{}, false + } + return pc.limited, true +} + +// Observe records one fetch for org on token: the REST calls it made and the +// budget GitHub reported afterwards. +func (p *Pacer) Observe(token, org string, calls int, b Budget) { + p.mu.Lock() + defer p.mu.Unlock() + pc := p.token(token) + pc.orgs[org] = true + pc.spent = true + pc.cost += calls + // Repos race, so keep the leanest reading of the cycle: it is the one + // closest to what is actually left. + if b.Known() && (!pc.low.Known() || b.Remaining < pc.low.Remaining) { + pc.low = b + } +} + +// Failed records a fetch error. When GitHub refused the token for its rate +// limit, the token is held until the reset GitHub gave, and Failed reports +// true. +func (p *Pacer) Failed(token string, err error) bool { + var rlErr *github.RateLimitError + var abuseErr *github.AbuseRateLimitError + var until time.Time + switch { + case errors.As(err, &rlErr): + until = rlErr.Rate.Reset.Time + case errors.As(err, &abuseErr) && abuseErr.RetryAfter != nil: + until = p.now().Add(*abuseErr.RetryAfter) + } + if until.IsZero() { + return false + } + p.mu.Lock() + p.token(token).limited = until + p.mu.Unlock() + logx.Debug("rate limited", "until", until) + return true +} + +// EndCycle folds the cycle's spending into each token's pace. A token that +// was skipped this cycle keeps the pace it already had. +// +// The fastest pace that lasts the window is what is left after the Reserve, +// divided by the measured cost per cycle, spread over the time until reset. +// The cost is re-measured every cycle, since expanding a Repo changes it. +func (p *Pacer) EndCycle(configured time.Duration) { + p.mu.Lock() + defer p.mu.Unlock() + p.configured = configured + now := p.now() + for _, pc := range p.tokens { + if !pc.spent { + continue + } + if pc.low.Known() { + pc.budget = pc.low + } + interval, throttled := Interval(configured, pc.budget, pc.cost, now) + pc.interval = interval + pc.nextDue = time.Time{} + if throttled { + pc.nextDue = now.Add(interval) + logx.Debug("throttling token", "orgs", sortedOrgs(pc.orgs), + "remaining", pc.budget.Remaining, "cost", pc.cost, "interval", interval) + } + pc.spent, pc.cost, pc.low = false, 0, Budget{} + } +} + +// Throttles reports every token polling slower than configured, for the title +// bar. Healthy tokens are omitted. +func (p *Pacer) Throttles() []Throttle { + p.mu.Lock() + defer p.mu.Unlock() + var out []Throttle + for _, pc := range p.tokens { + if pc.interval <= p.configured || !pc.budget.Known() { + continue + } + out = append(out, Throttle{ + Orgs: sortedOrgs(pc.orgs), + Remaining: pc.budget.Remaining, + Limit: pc.budget.Limit, + Reset: pc.budget.Reset, + Interval: pc.interval, + }) + } + sort.Slice(out, func(a, b int) bool { + return strings.Join(out[a].Orgs, ",") < strings.Join(out[b].Orgs, ",") + }) + return out +} + +func sortedOrgs(orgs map[string]bool) []string { + out := make([]string, 0, len(orgs)) + for o := range orgs { + out = append(out, o) + } + sort.Strings(out) + return out +} diff --git a/internal/ratelimit/pacer_test.go b/internal/ratelimit/pacer_test.go new file mode 100644 index 0000000..656d0ac --- /dev/null +++ b/internal/ratelimit/pacer_test.go @@ -0,0 +1,117 @@ +package ratelimit + +import ( + "errors" + "testing" + "time" + + "github.com/google/go-github/v72/github" +) + +const configured = time.Minute + +type clock struct{ t time.Time } + +func (c *clock) now() time.Time { return c.t } + +func newTestPacer() (*Pacer, *clock) { + c := &clock{t: time.Date(2026, 9, 28, 12, 0, 0, 0, time.UTC)} + return NewPacer(c.now), c +} + +// cycle runs one poll cycle: each org on the token spends calls against b. +func cycle(p *Pacer, token string, calls int, b Budget, orgs ...string) { + for _, org := range orgs { + p.Observe(token, org, calls, b) + } + p.EndCycle(configured) +} + +func TestPacerNeverHoldsBackAHealthyToken(t *testing.T) { + p, c := newTestPacer() + cycle(p, "tok", 20, Budget{Remaining: 4800, Limit: 5000, Reset: c.t.Add(50 * time.Minute)}, "acme") + + if !p.Due("tok") { + t.Error("a healthy token must stay due") + } + if got := p.Throttles(); len(got) != 0 { + t.Errorf("healthy token reported as throttled: %+v", got) + } +} + +func TestPacerSlowsAThinTokenForEveryOrgOnIt(t *testing.T) { + p, c := newTestPacer() + cycle(p, "tok", 150, Budget{Remaining: 400, Limit: 5000, Reset: c.t.Add(30 * time.Minute)}, "acme", "globex") + + if p.Due("tok") { + t.Fatal("a thin token must sit out the next cycle") + } + th := p.Throttles() + if len(th) != 1 { + t.Fatalf("got %d throttles, want 1", len(th)) + } + if got := th[0].Orgs; len(got) != 2 || got[0] != "acme" || got[1] != "globex" { + t.Errorf("throttle names %v, want both orgs sharing the token", got) + } + if th[0].Interval <= configured { + t.Errorf("interval %s is not slower than configured", th[0].Interval) + } + + c.t = c.t.Add(th[0].Interval + time.Second) + if !p.Due("tok") { + t.Error("the token must be due once its interval has passed") + } +} + +// A cycle that skipped the token keeps the pace it already had rather than +// reading zero spend as a healthy budget. +func TestPacerKeepsPaceAcrossSkippedCycles(t *testing.T) { + p, c := newTestPacer() + cycle(p, "tok", 150, Budget{Remaining: 400, Limit: 5000, Reset: c.t.Add(30 * time.Minute)}, "acme") + p.EndCycle(configured) // nothing spent + if p.Due("tok") { + t.Error("an idle cycle must not reset the pace") + } +} + +func TestPacerHoldsARateLimitedTokenUntilReset(t *testing.T) { + p, c := newTestPacer() + reset := c.t.Add(30 * time.Minute) + if !p.Failed("tok", &github.RateLimitError{Rate: github.Rate{Reset: github.Timestamp{Time: reset}}}) { + t.Fatal("a RateLimitError must be recognised") + } + + if p.Due("tok") { + t.Error("a rate-limited token must not be due") + } + if until, ok := p.LimitedUntil("tok"); !ok || !until.Equal(reset) { + t.Errorf("LimitedUntil = %v, %v; want %v", until, ok, reset) + } + + c.t = reset.Add(time.Second) + if !p.Due("tok") { + t.Error("the token must be due after the reset") + } + if _, ok := p.LimitedUntil("tok"); ok { + t.Error("the limit must clear after the reset") + } +} + +func TestPacerHonoursRetryAfter(t *testing.T) { + p, c := newTestPacer() + wait := 90 * time.Second + p.Failed("tok", &github.AbuseRateLimitError{RetryAfter: &wait}) + if until, ok := p.LimitedUntil("tok"); !ok || !until.Equal(c.t.Add(wait)) { + t.Errorf("LimitedUntil = %v, %v; want %v", until, ok, c.t.Add(wait)) + } +} + +func TestPacerIgnoresOtherErrors(t *testing.T) { + p, _ := newTestPacer() + if p.Failed("tok", errors.New("boom")) { + t.Error("an ordinary error is not a rate limit") + } + if !p.Due("tok") { + t.Error("an ordinary error must not hold the token back") + } +} diff --git a/internal/state/state.go b/internal/state/state.go index 703eb5d..c61f639 100644 --- a/internal/state/state.go +++ b/internal/state/state.go @@ -5,6 +5,7 @@ import ( "github.com/ericdahl-dev/git-green/internal/aggregator" githubclient "github.com/ericdahl-dev/git-green/internal/github" + "github.com/ericdahl-dev/git-green/internal/ratelimit" ) // PRState holds CI state for a single open pull request. @@ -48,13 +49,7 @@ func (r RepoState) IsStale() bool { // Throttle reports one token polling slower than configured because its REST // budget is running down. -type Throttle struct { - Orgs []string - Remaining int - Limit int - Reset time.Time - Interval time.Duration -} +type Throttle = ratelimit.Throttle // Snapshot is an immutable view of all repo states at a point in time. type Snapshot struct {