diff --git a/internal/classification/client.go b/internal/classification/client.go index 984d003..cef5caa 100644 --- a/internal/classification/client.go +++ b/internal/classification/client.go @@ -365,6 +365,12 @@ func (c *Client) complete(ctx context.Context, gate *ratelimit.Controller, r com Code int `json:"code"` } _ = json.Unmarshal(envelope.Error, &detail) + if detail.Code == http.StatusTooManyRequests { + // An upstream rate limit tunneled through HTTP 200 must arm the + // same cooldown as a transport 429: later Acquire calls fail fast + // instead of pacing more requests into a throttled endpoint. + return "", gate.ReportLimit() + } if detail.Code != 0 { return "", fmt.Errorf("AI provider reported an error (code %d)", detail.Code) } diff --git a/internal/classification/client_test.go b/internal/classification/client_test.go index 42f6f44..de8f422 100644 --- a/internal/classification/client_test.go +++ b/internal/classification/client_test.go @@ -11,6 +11,7 @@ import ( "reflect" "strings" "testing" + "time" "finance-duck/internal/domain" "finance-duck/internal/ratelimit" @@ -358,6 +359,30 @@ func TestMalformedEnvelopesRejected(t *testing.T) { } } +// An upstream rate limit tunneled inside an HTTP 200 envelope must arm the +// shared cooldown like a transport 429: the next classification fails fast +// instead of pacing another request into a throttled endpoint. +func TestEnvelope429ArmsSharedCooldown(t *testing.T) { + f, d := fixture() + calls := 0 + c := mockClient(t, func(w http.ResponseWriter, r *http.Request) { + calls++ + _, _ = io.WriteString(w, `{"error":{"code":429,"message":"private"},"choices":[]}`) + }) + c.rate.Store(&ratelimit.Controller{InitialBackoff: time.Minute}) + _, err := c.Classify(context.Background(), f, d, true) + var limit *ratelimit.RateLimitError + if err == nil || !errors.As(err, &limit) || strings.Contains(err.Error(), "private") { + t.Fatalf("envelope 429 not reported as a rate limit: %v", err) + } + if _, err = c.Classify(context.Background(), f, d, true); err == nil || !errors.As(err, &limit) { + t.Fatalf("cooldown not armed: %v", err) + } + if calls != 1 { + t.Fatalf("throttled endpoint was contacted again: %d calls", calls) + } +} + type failingTransport struct{} func (failingTransport) RoundTrip(*http.Request) (*http.Response, error) { diff --git a/internal/ratelimit/controller.go b/internal/ratelimit/controller.go index 00b5fed..cb0a2c5 100644 --- a/internal/ratelimit/controller.go +++ b/internal/ratelimit/controller.go @@ -109,6 +109,43 @@ func (g *Controller) Release() { <-g.active } +// recordLimit escalates the consecutive-failure backoff, retains the cooldown +// and learns spacing. Callers hold the Acquire gate, like Do's 429 branch. +func (g *Controller) recordLimit(header string) *RateLimitError { + if g.backoff <= 0 { + g.backoff = g.InitialBackoff + if g.backoff <= 0 { + g.backoff = time.Second + } + } else if g.backoff >= maxBackoff/2 { + g.backoff = max(g.backoff, maxBackoff) + } else { + g.backoff *= 2 + } + fallback := max(g.backoff, g.MinimumInterval, g.learnedInterval) + limit := retryLimit(header, time.Now(), fallback) + g.mu.Lock() + g.limit = limit + g.mu.Unlock() + // Keep the most conservative learned cadence for this controller's + // lifetime, capped at 30 seconds. The actual provider deadline is never + // capped; persistent failures separately escalate up to 15 minutes. + learned := maxLearnedInterval + if !limit.unbounded { + learned = min(learned, time.Until(limit.next)) + } + g.learnedInterval = max(g.learnedInterval, learned) + return limit +} + +// ReportLimit records a rate limit the provider communicated outside the HTTP +// status — typically inside an HTTP 200 error envelope — so later Acquire +// calls fail fast during the cooldown exactly as after a transport HTTP 429. +// It must be called while holding an Acquire, like Do. +func (g *Controller) ReportLimit() *RateLimitError { + return g.recordLimit("") +} + // retryLimit never converts a positive overflowing delay into a short wait. // Delays beyond time.Duration's range disable retries rather than truncate the // provider's instruction. HTTP dates retain their absolute timestamp unchanged. @@ -202,29 +239,7 @@ func (g *Controller) Do(ctx context.Context, attempt func(context.Context) (*htt } return resp, nil } - if g.backoff <= 0 { - g.backoff = g.InitialBackoff - if g.backoff <= 0 { - g.backoff = time.Second - } - } else if g.backoff >= maxBackoff/2 { - g.backoff = max(g.backoff, maxBackoff) - } else { - g.backoff *= 2 - } - fallback := max(g.backoff, g.MinimumInterval, g.learnedInterval) - limit := retryLimit(resp.Header.Get("Retry-After"), time.Now(), fallback) - g.mu.Lock() - g.limit = limit - g.mu.Unlock() - // Keep the most conservative learned cadence for this controller's - // lifetime, capped at 30 seconds. The actual provider deadline is never - // capped; persistent failures separately escalate up to 15 minutes. - learned := maxLearnedInterval - if !limit.unbounded { - learned = min(learned, time.Until(limit.next)) - } - g.learnedInterval = max(g.learnedInterval, learned) + limit := g.recordLimit(resp.Header.Get("Retry-After")) // Never read or expose provider errors, and release each response before // any sleep or retry. Other responses are processed by the caller. resp.Body.Close()