diff --git a/backend/internal/handler/openai_gateway_handler_test.go b/backend/internal/handler/openai_gateway_handler_test.go index 0f0964413..543cda1e5 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 1efed7e9e..95ee4ce76 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 8a8349b5f..171a6c880 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 63a39826b..a5836a90d 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 2e61b6013..04990dc76 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 7c1072402..2a1993c46 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 3190aa974..616f0f2f6 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{