From 85a27fae39f9ba0a2b35c791e998cf873c901eb1 Mon Sep 17 00:00:00 2001 From: feeeei Date: Thu, 30 Jul 2026 15:22:16 +0800 Subject: [PATCH] fix(openai): retry SSE rate limits as HTTP 429 OpenAI Responses may return code=rate_limit_exceeded in a response.failed SSE event while the HTTP status remains 200. Classify these failures as 429 so configured pool-mode retries and account failover are applied. No account-level rate-limit state is written on this path: the 200-stream response headers carry normal quota snapshots, and retry semantics stay owned by the failover engine. Ignore compact keepalive bytes when determining whether semantic output has started, preserving safe retries before real output. --- .../handler/openai_gateway_handler_test.go | 116 ++++++++++++++++++ .../openai_compact_sse_keepalive_test.go | 14 +++ .../openai_gateway_chat_completions.go | 4 +- .../service/openai_gateway_messages.go | 4 +- .../service/openai_gateway_passthrough.go | 56 +++++++-- .../openai_gateway_response_handling.go | 2 +- .../service/openai_gateway_service_test.go | 111 +++++++++++++++++ 7 files changed, 294 insertions(+), 13 deletions(-) diff --git a/backend/internal/handler/openai_gateway_handler_test.go b/backend/internal/handler/openai_gateway_handler_test.go index 0f09644138..543cda1e55 100644 --- a/backend/internal/handler/openai_gateway_handler_test.go +++ b/backend/internal/handler/openai_gateway_handler_test.go @@ -1812,6 +1812,40 @@ func (u *openAIHTTPPassthroughFailoverUpstream) calls() []int64 { return append([]int64(nil), u.accountIDs...) } +type openAIHTTPPassthroughSSERateLimitUpstream struct { + service.HTTPUpstream + mu sync.Mutex + accountIDs []int64 +} + +func (u *openAIHTTPPassthroughSSERateLimitUpstream) Do(_ *http.Request, _ string, accountID int64, _ int) (*http.Response, error) { + u.mu.Lock() + u.accountIDs = append(u.accountIDs, accountID) + u.mu.Unlock() + body := strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_rate_limited"}}`, + "", + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_rate_limited","status":"failed","error":{"code":"rate_limit_exceeded","message":"Concurrency limit exceeded for account, please retry later"}}}`, + "", + }, "\n") + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{ + "Content-Type": []string{"text/event-stream"}, + "Retry-After": []string{"1"}, + }, + Body: io.NopCloser(strings.NewReader(body)), + }, nil +} + +func (u *openAIHTTPPassthroughSSERateLimitUpstream) calls() []int64 { + u.mu.Lock() + defer u.mu.Unlock() + return append([]int64(nil), u.accountIDs...) +} + func (s *openAIWSFailoverHandlerAccountRepoStub) ListSchedulableByPlatform(ctx context.Context, platform string) ([]service.Account, error) { out := make([]service.Account, 0, len(s.accounts)) for _, account := range s.accounts { @@ -2012,6 +2046,88 @@ func TestOpenAIResponses_APIKeyPassthroughPool5xxRetriesThenExhaustsMaxSwitches( require.Equal(t, "Upstream service temporarily unavailable", gjson.GetBytes(rec.Body.Bytes(), "error.message").String()) } +func TestOpenAIResponses_APIKeyPassthroughSSERateLimitUsesConfiguredPoolRetry(t *testing.T) { + gin.SetMode(gin.TestMode) + groupID := int64(4204) + accounts := []service.Account{ + { + ID: 9912, Name: "pool-sse-rate-limit", Platform: service.PlatformOpenAI, + Type: service.AccountTypeAPIKey, Status: service.StatusActive, Schedulable: true, Priority: 1, + Credentials: map[string]any{ + "api_key": "sk-pool", + "base_url": "https://api.example.test", + "pool_mode": true, + "pool_mode_retry_count": float64(1), + "pool_mode_retry_status_codes": []any{float64(http.StatusTooManyRequests)}, + }, + Extra: map[string]any{"openai_passthrough": true}, + }, + } + cfg := &config.Config{RunMode: config.RunModeSimple} + cfg.Default.RateMultiplier = 1 + cfg.Security.URLAllowlist.Enabled = false + cfg.Gateway.MaxAccountSwitches = 1 + + accountRepo := &openAIWSFailoverHandlerAccountRepoStub{accounts: accounts} + upstream := &openAIHTTPPassthroughSSERateLimitUpstream{} + billingCacheSvc := service.NewBillingCacheService(nil, nil, nil, nil, nil, nil, cfg, nil) + t.Cleanup(billingCacheSvc.Stop) + gatewaySvc := service.NewOpenAIGatewayService( + accountRepo, + nil, + nil, + nil, + nil, + nil, + nil, + cfg, + nil, + nil, + service.NewBillingService(cfg, nil), + nil, + billingCacheSvc, + upstream, + &service.DeferredService{}, + nil, + nil, + nil, + nil, + nil, + nil, + nil, + ) + h := NewOpenAIGatewayHandler( + gatewaySvc, + service.NewConcurrencyService(nil), + billingCacheSvc, + service.NewAPIKeyService(nil, nil, nil, nil, nil, nil, cfg), + nil, + nil, + nil, + nil, + cfg, + ) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/openai/v1/responses", strings.NewReader(`{"model":"gpt-5.6-sol","input":"hello","stream":true}`)) + c.Request.Header.Set("Content-Type", "application/json") + c.Set(string(middleware.ContextKeyAPIKey), &service.APIKey{ + ID: 1804, GroupID: &groupID, + User: &service.User{ID: 1704, Status: service.StatusActive}, + Group: &service.Group{ID: groupID, Platform: service.PlatformOpenAI, Status: service.StatusActive}, + }) + c.Set(string(middleware.ContextKeyUser), middleware.AuthSubject{UserID: 1704, Concurrency: 0}) + + h.Responses(c) + + require.Equal(t, []int64{9912, 9912}, upstream.calls()) + require.Equal(t, http.StatusTooManyRequests, rec.Code) + require.Equal(t, "1", rec.Header().Get("Retry-After")) + require.Equal(t, "rate_limit_error", gjson.GetBytes(rec.Body.Bytes(), "error.type").String()) + require.Equal(t, "Upstream rate limit exceeded, please retry later", gjson.GetBytes(rec.Body.Bytes(), "error.message").String()) +} + func TestOpenAIResponsesWebSocket_FailoverOnUpstreamUsageLimitEvent(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/backend/internal/service/openai_compact_sse_keepalive_test.go b/backend/internal/service/openai_compact_sse_keepalive_test.go index 1efed7e9e4..95ee4ce76c 100644 --- a/backend/internal/service/openai_compact_sse_keepalive_test.go +++ b/backend/internal/service/openai_compact_sse_keepalive_test.go @@ -284,6 +284,20 @@ func TestOpenAICompactKeepaliveAdjustedWrittenSize_ExcludesHeartbeatBytes(t *tes require.Contains(t, rec.Body.String(), ": keepalive\n\n") } +func TestOpenAIStreamClientOutputStarted_IgnoresCompactKeepaliveBytes(t *testing.T) { + c, _ := newCompactBridgeTestContext(t, true) + stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval) + defer stop() + waitForKeepaliveBeats() + + require.True(t, c.Writer.Written()) + require.False(t, openAIStreamClientOutputStarted(c, false), "keepalive comments are not semantic output") + + _, err := c.Writer.Write([]byte("real-output")) + require.NoError(t, err) + require.True(t, openAIStreamClientOutputStarted(c, false)) +} + // fast policy block 在心跳未提交时保持 403 JSON 原语义。 func TestWriteOpenAIFastPolicyBlockedResponse_BeforeKeepaliveCommit(t *testing.T) { c, rec := newCompactBridgeTestContext(t, true) diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index 8a8349b5f8..171a6c8805 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -441,7 +441,7 @@ func (s *OpenAIGatewayService) handleChatBufferedStreamingResponse( } message := openAICompatFailedResponseMessage(finalResponse) if openAIStreamFailedEventShouldFailover(payload, message) { - return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message) + return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message, resp.Header) } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payload, message) // response.failed 到达在 HTTP 200 SSE 流上，无真实 HTTP 错误码；统一走语义 @@ -608,7 +608,7 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse( return true } if openAIStreamFailedEventShouldFailover(payloadBytes, message) { - streamFailoverErr = s.newOpenAIStreamFailoverError(c, account, false, requestID, payloadBytes, message) + streamFailoverErr = s.newOpenAIStreamFailoverError(c, account, false, requestID, payloadBytes, message, resp.Header) return true } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payloadBytes, message) diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index 63a39826b0..a5836a90d3 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -561,7 +561,7 @@ func (s *OpenAIGatewayService) handleAnthropicBufferedStreamingResponse( } message := openAICompatFailedResponseMessage(finalResponse) if openAIStreamFailedEventShouldFailover(payload, message) { - return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message) + return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message, resp.Header) } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payload, message) // 统一走语义状态推断 + body 归一化（与 /v1/responses 路径一致）， @@ -917,7 +917,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( // two model streams together. Surface a proper Anthropic error event // instead of returning a failover error that the handler cannot retry. if !clientOutputStarted && openAIStreamFailedEventShouldFailover(payloadBytes, message) { - streamFailoverErr = s.newOpenAIStreamFailoverError(c, account, false, requestID, payloadBytes, message) + streamFailoverErr = s.newOpenAIStreamFailoverError(c, account, false, requestID, payloadBytes, message, resp.Header) return true } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payloadBytes, message) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 2e61b60131..04990dc761 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -738,7 +738,13 @@ func openAIStreamClientOutputStarted(c *gin.Context, localStarted bool) bool { if localStarted { return true } - return c != nil && c.Writer != nil && c.Writer.Written() + if c == nil || c.Writer == nil { + return false + } + // compact keepalive comments commit the HTTP response as 200, but they are + // not semantic model output and therefore must not block a safe retry. + // Without a compact keepalive this is equivalent to checking Writer.Size(). + return OpenAICompactKeepaliveAdjustedWrittenSize(c) >= 0 } func openAIStreamEventIsPreamble(eventType string) bool { @@ -776,10 +782,10 @@ func openAIStreamFailedEventSemanticStatus(payload []byte, message string) int { } combined := strings.TrimSpace(errType + " " + code + " " + strings.ToLower(strings.TrimSpace(message))) switch { - case strings.Contains(errType, "invalid_request"): - return http.StatusBadRequest case strings.Contains(combined, "rate_limit"): return http.StatusTooManyRequests + case strings.Contains(errType, "invalid_request"): + return http.StatusBadRequest case strings.Contains(combined, "authentication") || strings.Contains(combined, "unauthorized") || strings.Contains(combined, "invalid_api_key"): return http.StatusUnauthorized case strings.Contains(combined, "permission") || strings.Contains(combined, "forbidden") || strings.Contains(combined, "access denied"): @@ -791,6 +797,19 @@ func openAIStreamFailedEventSemanticStatus(payload []byte, message string) int { } } +func openAIStreamFailureStatus(payload []byte, message string) int { + if len(bytes.TrimSpace(payload)) == 0 || !gjson.ValidBytes(payload) { + return http.StatusBadGateway + } + // Keep the existing 502 failover behavior for other response.failed events. + // Only rate limits need promotion because they participate in the account's + // configurable 429 same-account retry policy. + if openAIStreamFailedEventSemanticStatus(payload, message) == http.StatusTooManyRequests { + return http.StatusTooManyRequests + } + return http.StatusBadGateway +} + func openAIStreamFailedEventPassthroughBody(payload []byte, failedMessage string) []byte { if len(payload) == 0 || !gjson.ValidBytes(payload) { return payload @@ -867,6 +886,12 @@ func openAIStreamFailedEventShouldFailover(payload []byte, message string) bool if isOpenAIContextWindowError(message, payload) { return false } + // A response.failed event is transported over HTTP 200. Prefer its semantic + // rate-limit status over a generic/invalid_request error type so it can enter + // the same 429 retry policy as a regular upstream HTTP response. + if openAIStreamFailureStatus(payload, message) == http.StatusTooManyRequests { + return true + } if isOpenAITransientProcessingError(http.StatusBadRequest, message, payload) { return true } @@ -921,6 +946,7 @@ func (s *OpenAIGatewayService) recordOpenAIStreamUpstreamError( if message == "" { message = "OpenAI upstream response failed" } + statusCode := openAIStreamFailureStatus(payload, message) detail := "" if len(payload) > 0 && s != nil && s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes @@ -930,10 +956,10 @@ func (s *OpenAIGatewayService) recordOpenAIStreamUpstreamError( detail = truncateString(string(payload), maxBytes) } if c != nil { - setOpsUpstreamError(c, http.StatusBadGateway, message, detail) + setOpsUpstreamError(c, statusCode, message, detail) event := OpsUpstreamErrorEvent{ Platform: PlatformOpenAI, - UpstreamStatusCode: http.StatusBadGateway, + UpstreamStatusCode: statusCode, UpstreamRequestID: strings.TrimSpace(upstreamRequestID), Passthrough: passthrough, Kind: kind, @@ -957,21 +983,35 @@ func (s *OpenAIGatewayService) newOpenAIStreamFailoverError( upstreamRequestID string, payload []byte, message string, + responseHeaders ...http.Header, ) *UpstreamFailoverError { message = sanitizeUpstreamErrorMessage(strings.TrimSpace(message)) if message == "" { message = "OpenAI stream disconnected before completion" } + statusCode := openAIStreamFailureStatus(payload, message) + var headers http.Header + if len(responseHeaders) > 0 && responseHeaders[0] != nil { + headers = responseHeaders[0].Clone() + } + // 流内 failed 事件承载于 HTTP 200，响应头是正常配额快照而非限流信号， + // 不写账号级限流/封禁状态；重试与切号由 failover 引擎按 + // StatusCode/RetryableOnSameAccount 决定。 message = s.recordOpenAIStreamUpstreamError(c, account, passthrough, upstreamRequestID, "failover", payload, message) + errType := "upstream_error" + if statusCode == http.StatusTooManyRequests { + errType = "rate_limit_error" + } body, _ := json.Marshal(gin.H{ "error": gin.H{ - "type": "upstream_error", + "type": errType, "message": message, }, }) return &UpstreamFailoverError{ - StatusCode: http.StatusBadGateway, + StatusCode: statusCode, ResponseBody: body, + ResponseHeaders: headers, RetryableOnSameAccount: openAIStreamFailedEventRetryableOnSameAccount(account, payload, message), } } @@ -1127,7 +1167,7 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( } if openAIStreamFailedEventShouldFailover(dataBytes, failedMessage) { return resultWithUsage(), - s.newOpenAIStreamFailoverError(c, account, true, upstreamRequestID, dataBytes, failedMessage) + s.newOpenAIStreamFailoverError(c, account, true, upstreamRequestID, dataBytes, failedMessage, resp.Header) } } forceFlushFailedEvent = true diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 7c10724028..2a1993c463 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -449,7 +449,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context. } if openAIStreamFailedEventShouldFailover(dataBytes, failedMessage) { sawFailedEvent = true - streamEarlyErr = s.newOpenAIStreamFailoverError(c, account, false, upstreamRequestID, dataBytes, failedMessage) + streamEarlyErr = s.newOpenAIStreamFailoverError(c, account, false, upstreamRequestID, dataBytes, failedMessage, resp.Header) return } } diff --git a/backend/internal/service/openai_gateway_service_test.go b/backend/internal/service/openai_gateway_service_test.go index 3190aa974b..616f0f2f6a 100644 --- a/backend/internal/service/openai_gateway_service_test.go +++ b/backend/internal/service/openai_gateway_service_test.go @@ -1577,6 +1577,117 @@ func TestOpenAIStreamingResponseFailedBeforeOutputServerOverloadedCodeReturnsFai require.Empty(t, rec.Body.String()) } +func TestOpenAIStreamingResponseFailedBeforeOutputRateLimitUsesPoolRetryPolicy(t *testing.T) { + gin.SetMode(gin.TestMode) + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + StreamDataIntervalTimeout: 0, + StreamKeepaliveInterval: 0, + MaxLineSize: defaultMaxLineSize, + }, + } + svc := &OpenAIGatewayService{cfg: cfg} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/", nil) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_1"}}`, + "", + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_1","status":"failed","error":{"code":"rate_limit_exceeded","message":"Concurrency limit exceeded for account, please retry later"}}}`, + "", + }, "\n"))), + Header: http.Header{ + "X-Request-Id": []string{"rid-rate-limit-failed"}, + "Retry-After": []string{"1"}, + }, + } + account := &Account{ + ID: 1, + Platform: PlatformOpenAI, + Type: AccountTypeAPIKey, + Name: "pool-account", + Credentials: map[string]any{ + "pool_mode": true, + "pool_mode_retry_count": float64(1), + "pool_mode_retry_status_codes": []any{float64(http.StatusTooManyRequests)}, + }, + } + + _, err := svc.handleStreamingResponse(c.Request.Context(), resp, c, account, time.Now(), "model", "model") + require.Error(t, err) + var failoverErr *UpstreamFailoverError + require.ErrorAs(t, err, &failoverErr) + require.Equal(t, http.StatusTooManyRequests, failoverErr.StatusCode) + require.True(t, failoverErr.RetryableOnSameAccount) + require.Equal(t, "1", failoverErr.ResponseHeaders.Get("Retry-After")) + require.Equal(t, "rate_limit_error", gjson.GetBytes(failoverErr.ResponseBody, "error.type").String()) + require.Contains(t, string(failoverErr.ResponseBody), "Concurrency limit exceeded") + require.False(t, c.Writer.Written()) + require.Empty(t, rec.Body.String()) + + opsVal, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok) + opsEvents, ok := opsVal.([]*OpsUpstreamErrorEvent) + require.True(t, ok) + require.NotEmpty(t, opsEvents) + require.Equal(t, http.StatusTooManyRequests, opsEvents[len(opsEvents)-1].UpstreamStatusCode) +} + +// 流内 rate limit 只产生 failover 错误，不写账号级限流/封禁状态： +// HTTP 200 流的 x-codex-* 头是正常配额快照，不能按 429 头驱动账号冷却。 +func TestOpenAIStreamingResponseFailedRateLimitDoesNotBlockAccountScheduling(t *testing.T) { + gin.SetMode(gin.TestMode) + cfg := &config.Config{ + Gateway: config.GatewayConfig{ + StreamDataIntervalTimeout: 0, + StreamKeepaliveInterval: 0, + MaxLineSize: defaultMaxLineSize, + }, + } + svc := &OpenAIGatewayService{cfg: cfg} + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/", nil) + + resp := &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_1"}}`, + "", + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_1","status":"failed","error":{"code":"rate_limit_exceeded","message":"Concurrency limit exceeded for account, please retry later"}}}`, + "", + }, "\n"))), + Header: http.Header{ + "X-Codex-Primary-Used-Percent": []string{"12"}, + "X-Codex-Primary-Reset-After-Seconds": []string{"604800"}, + "Retry-After": []string{"1"}, + }, + } + account := &Account{ + ID: 11, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + Name: "oauth-account", + } + + _, err := svc.handleStreamingResponse(c.Request.Context(), resp, c, account, time.Now(), "model", "model") + require.Error(t, err) + var failoverErr *UpstreamFailoverError + require.ErrorAs(t, err, &failoverErr) + require.Equal(t, http.StatusTooManyRequests, failoverErr.StatusCode) + require.False(t, failoverErr.RetryableOnSameAccount) + require.False(t, svc.isOpenAIAccountRuntimeBlocked(account)) +} + func TestOpenAIStreamingResponseFailedAfterOutputSanitizesVerboseResponseForClient(t *testing.T) { gin.SetMode(gin.TestMode) cfg := &config.Config{