From 3a7e271c5f033530a424670215077bde808a6c1a Mon Sep 17 00:00:00 2001 From: kristofferR Date: Sun, 26 Jul 2026 17:58:33 +0200 Subject: [PATCH] Let the fire slot serialize the metered review and nothing else MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pump returned as soon as it had progressed the round holding the fire slot. The rescue scan further down said it covered "blocked or slot-busy" starvation, and its condition tested both — but control never reached it with the slot held, so the slot-busy half had never run. A summary-only round, whose CodeRabbit review is never coming at all, therefore sat queued for as long as an unrelated PR's review took. The slot exists to serialize one account-metered review; the account window bounds that same allowance. Neither has anything to say about a round that spends no quota. So progressing the slot round no longer ends the pump: it reloads, then sweeps the quota-free rounds behind it. A pump still never both progresses and fires, because a quota-free verdict is not a fire. The scan is now one function used by both paths that can starve, rather than a loop inlined into the tail of the account-blocked case, which is how one of its two callers came to be unreachable. --- internal/crq/service.go | 136 ++++++++++++++++++++++------------- internal/crq/service_test.go | 83 ++++++++++++++++++++- 2 files changed, 167 insertions(+), 52 deletions(-) diff --git a/internal/crq/service.go b/internal/crq/service.go index 2f385945..2c19b709 100644 --- a/internal/crq/service.go +++ b/internal/crq/service.go @@ -221,10 +221,28 @@ func (s *Service) Pump(ctx context.Context) (PumpResult, error) { return PumpResult{}, err } - // 1. The round holding the fire slot: progress it and return, mirroring v2's - // "handle in-flight first" so a single pump never both progresses and fires. + // 1. The round holding the fire slot: progress it first, mirroring v2's + // "handle in-flight first" so a single pump never both progresses and + // fires. It does not end the pump, though. The slot serializes the METERED + // fire and nothing else, so rounds whose next step spends no CodeRabbit + // quota still get their turn below. if slot := st.SlotRound(); slot != nil { - return s.progressSlotRound(ctx, *slot) + res, err := s.progressSlotRound(ctx, *slot) + if err != nil { + return res, err + } + // Reload: progressing the slot round may have released it, and the sweep + // must not decide against a snapshot that says otherwise. + st, _, err = s.store.Load(ctx) + if err != nil { + return PumpResult{}, err + } + if free, handled, err := s.sweepQuotaFree(ctx, st, s.clock(), slot.Repo, slot.PR); err != nil { + return PumpResult{}, err + } else if handled { + return free, nil + } + return res, nil } // 2. Reviewing rounds no longer hold the slot; sweep the oldest one toward @@ -286,58 +304,78 @@ func (s *Service) Pump(ctx context.Context) (PumpResult, error) { if err != nil { return result, err } - // A blocked or slot-busy front of the queue must not starve later PRs of - // resolutions that spend NO CodeRabbit quota — a co-reviewer defer, or a - // summary-only round whose review is never coming from CodeRabbit at all. - // Scan a bounded number of following queued rounds and apply any quota-free - // verdict (each costs one observation; ETag caching keeps it cheap). - // - // The scan is deliberately verdict-driven rather than flag-driven: - // `decideCoDeferred` already honours RateLimitCoDegrade internally, so a - // disabled degrade simply never yields FireCoDeferred, while a summary-only - // round must be rescued regardless of that flag. + // A blocked front of the queue must not starve later PRs of resolutions that + // spend NO CodeRabbit quota — a co-reviewer defer, or a summary-only round + // whose review is never coming from CodeRabbit at all. accountBlocked := global.BlockedUntil != nil && global.BlockedUntil.After(now) - if decision.Verdict == engine.FireNo && (accountBlocked || !global.SlotFree) { - // Rotate the window start across pumps. A fixed start meant that when the - // three rounds behind the FIFO head all needed CodeRabbit, every pump - // observed those same three and stopped — a fourth, quota-free round was - // never even looked at until the unrelated block cleared. - queued := st.QueuedRounds(now) - scanned := 0 - policy := s.policy() - for i := range queued { - r := queued[(i+s.scanOffset)%len(queued)] - if r.Repo == next.Repo && r.PR == next.PR { - continue - } - if scanned >= 3 { - break - } - // No cheap pre-filter here. "Every trigger already posted" is not - // proof that nothing is left to do: a primary-unavailable round in - // that state still needs its quota-free FireDedupe/FireCoReviewWait, - // and a co-review answer to an earlier deferred command needs - // collecting. Skipping those left them behind the account block for - // hours. The scan budget below bounds the cost instead. - scanned++ - round := r - robs, oerr := s.observe(ctx, round.Repo, round.PR, &round, now) - if oerr != nil { - continue - } - d := engine.DecideFire(s.global(st, now), round, robs.eng, now, policy) - if !quotaFreeVerdict(d.Verdict) { - continue - } - return s.applyFire(ctx, round, robs.eng, d, now) - } - if len(queued) > 0 { - s.scanOffset = (s.scanOffset + scanned + 1) % len(queued) + if decision.Verdict == engine.FireNo && accountBlocked { + if free, handled, err := s.sweepQuotaFree(ctx, st, now, next.Repo, next.PR); err != nil { + return PumpResult{}, err + } else if handled { + return free, nil } } return result, nil } +// sweepQuotaFree gives the queued rounds that spend no CodeRabbit quota their +// turn when the metered front cannot move — because another PR holds the fire +// slot, or the account window is shut. +// +// Neither of those has any authority here. The slot exists to serialize one +// account-metered review, and the window bounds that same allowance; a +// co-reviewer defer or a summary-only round asks for neither. Leaving them in +// the FIFO is what stranded PRs behind an unrelated block for hours. +// +// skipRepo/skipPR is the round the caller already decided about, so one pump +// never applies two verdicts to it. +// +// It observes at most scanBudget rounds per pump and rotates where it starts. +// A fixed start meant that when the rounds behind the head all needed +// CodeRabbit, every pump observed those same few and stopped — a quota-free +// round further back was never even looked at until the unrelated block cleared. +func (s *Service) sweepQuotaFree(ctx context.Context, st State, now time.Time, skipRepo string, skipPR int) (PumpResult, bool, error) { + const scanBudget = 3 + queued := st.QueuedRounds(now) + if len(queued) == 0 { + return PumpResult{}, false, nil + } + policy := s.policy() + global := s.global(st, now) + scanned := 0 + defer func() { s.scanOffset = (s.scanOffset + scanned + 1) % len(queued) }() + for i := range queued { + round := queued[(i+s.scanOffset)%len(queued)] + if round.Repo == skipRepo && round.PR == skipPR { + continue + } + if scanned >= scanBudget { + break + } + // No cheap pre-filter here. "Every trigger already posted" is not proof + // that nothing is left to do: a primary-unavailable round in that state + // still needs its quota-free FireDedupe/FireCoReviewWait, and a co-review + // answer to an earlier deferred command needs collecting. Skipping those + // left them behind the account block for hours. The budget above bounds + // the cost instead. + scanned++ + obs, err := s.observe(ctx, round.Repo, round.PR, &round, now) + if err != nil { + continue + } + d := engine.DecideFire(global, round, obs.eng, now, policy) + if !quotaFreeVerdict(d.Verdict) { + continue + } + res, err := s.applyFire(ctx, round, obs.eng, d, now) + if err != nil { + return PumpResult{}, false, err + } + return res, true, nil + } + return PumpResult{}, false, nil +} + // advanceQuotaFree resolves ONE PR's round directly, bypassing the account-wide // FIFO, when its verdict spends no CodeRabbit quota. It is the loop's escape // hatch from a queue it does not belong in: a summary-only round has no review diff --git a/internal/crq/service_test.go b/internal/crq/service_test.go index 19c42419..1463d3ea 100644 --- a/internal/crq/service_test.go +++ b/internal/crq/service_test.go @@ -2456,6 +2456,82 @@ func TestPumpRescuesSummaryOnlyRoundBehindBlockedQueue(t *testing.T) { } } +// TestPumpSweepsQuotaFreeRoundWhileTheSlotIsHeld pins the rule the fire slot +// actually stands for: it serializes ONE account-metered review, and has no +// authority over work that spends none. +// +// Pump used to return the moment it had progressed the slot holder. Its own +// rescue scan claimed to cover "blocked or slot-busy", but the slot-busy half +// was unreachable — control never got there. So a summary-only round, whose +// CodeRabbit review is never coming at all, sat queued for as long as an +// unrelated PR's review ran. +func TestPumpSweepsQuotaFreeRoundWhileTheSlotIsHeld(t *testing.T) { + ctx := context.Background() + cfg := firingConfig() + cfg.RequiredBots = []string{"coderabbitai[bot]", dialect.CodexBotLogin} + cfg.CoBots = codexCoBots(cfg.RequiredBots) + cfg.FeedbackBots = cfg.RequiredBots + gh := newFakeGitHub() + gh.graphQL = noForcePush + now := time.Now().UTC() + + // The slot holder: an unrelated PR whose CodeRabbit review is in flight. + holderSHA := "1111111111111111" + holder := ghapi.Pull{State: "open"} + holder.Head.SHA = holderSHA + gh.pulls[fakeKey("o/holder", 10)] = holder + + // Behind it: a summary-only PR. CodeRabbit posted only its Free-plan + // walkthrough, which is the whole review it will ever produce. + backSHA := "2222222222222222" + backPull := ghapi.Pull{State: "open"} + backPull.Head.SHA = backSHA + gh.pulls[fakeKey("o/back", 20)] = backPull + walkthrough := ghapi.IssueComment{ + ID: 900, + Body: corpusMessage(t, "coderabbit/summary-only-free-plan.md"), + CreatedAt: now.Add(-time.Minute), + UpdatedAt: now.Add(-time.Minute), + } + walkthrough.User.Login = "coderabbitai[bot]" + gh.comments[fakeKey("o/back", 20)] = []ghapi.IssueComment{walkthrough} + + store := NewMemoryStore(cfg) + svc := NewService(cfg, gh, store, nil) + svc.now = func() time.Time { return now } + + // The holder is fired and holds the slot; the summary-only round is queued + // behind it. + seedRound(t, store, cfg, "o/holder", 10, holderSHA, PhaseFired, now.Add(-time.Minute), 701) + if _, err := svc.Enqueue(ctx, "o/back", 20); err != nil { + t.Fatal(err) + } + if st, _, err := store.Load(ctx); err != nil { + t.Fatal(err) + } else if st.SlotRound() == nil { + t.Fatal("the test needs the slot held to mean anything") + } + + if _, err := svc.Pump(ctx); err != nil { + t.Fatal(err) + } + + st, _, _ := store.Load(ctx) + back := st.Round("o/back", 20) + if back == nil || back.Phase == PhaseQueued { + t.Fatalf("the summary-only round must move while the slot is held, got %#v", back) + } + // The slot is not what let it move, and it did not take it. + if slot := st.SlotRound(); slot == nil || slot.Repo != "o/holder" { + t.Fatalf("the slot must still belong to the holder, got %#v", slot) + } + for _, p := range gh.posted { + if strings.Contains(p, cfg.ReviewCommand) { + t.Fatalf("a quota-free sweep must never post the CodeRabbit command, posted=%v", gh.posted) + } + } +} + // TestWaitResolvesSummaryOnlyWithoutTheQueue pins the architectural rule the // dogfood exposed: a summary-only round is NOT a queue citizen. The Seq FIFO and // the FireSlot exist solely to serialize CodeRabbit's account-wide review limit, @@ -2463,9 +2539,10 @@ func TestPumpRescuesSummaryOnlyRoundBehindBlockedQueue(t *testing.T) { // an account block, nor another PR holding the slot, may delay it. Its // co-reviewers are ready now; the loop must run them now. // -// The starvation is staged at its worst: another PR HOLDS the fire slot, so Pump -// short-circuits on the slot holder and never reaches the FIFO or its bounded -// rescue scan at all. Only the direct quota-free path can resolve this round. +// The starvation is staged at its worst: another PR HOLDS the fire slot. Pump +// now sweeps quota-free rounds even then (see +// TestPumpSweepsQuotaFreeRoundWhileTheSlotIsHeld); this pins that Wait resolves +// the round directly, without depending on a pump happening to run. func TestWaitResolvesSummaryOnlyWithoutTheQueue(t *testing.T) { cfg := firingConfig() cfg.RequiredBots = []string{"coderabbitai[bot]", dialect.CodexBotLogin}