seewe/sub2api
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-14 03:52:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(openai): preserve model scope on stream failover

Browse Source
This commit is contained in:
shaw
2026-08-29 15:25:10 +08:00
parent 571d1e1d9b
commit 804679d996
6 changed files with 51 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/service/openai_account_runtime_block_fastpath_test.go
+26
View File
@@ -176,6 +176,32 @@ func TestOpenAIStream429_SparkQuotaUsesQuotaHeaders(t *testing.T) {
require.False(t, svc.isOpenAIAccountRuntimeBlocked(account))
}
func TestOpenAIStreamFailover_Spark429KeepsModelScope(t *testing.T) {
repo := &oauth429RateLimitRepo{}
rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil)
svc := &OpenAIGatewayService{rateLimitService: rateLimits}
rateLimits.SetAccountRuntimeBlocker(svc)
account := &Account{ID: 432, Platform: PlatformOpenAI, Type: AccountTypeOAuth}
headers := http.Header{}
headers.Set("x-codex-primary-used-percent", "100")
headers.Set("x-codex-primary-reset-after-seconds", "604800")
headers.Set("x-codex-primary-window-minutes", "10080")
headers.Set("x-codex-secondary-used-percent", "20")
headers.Set("x-codex-secondary-reset-after-seconds", "3600")
headers.Set("x-codex-secondary-window-minutes", "300")
payload := []byte(`{"type":"error","error":{"type":"rate_limit_error","code":"rate_limit_exceeded"}}`)
failoverErr := svc.newOpenAIStreamFailoverErrorWithModel(
nil, account, false, "", payload, "quota exhausted", "gpt-5.3-codex-spark", headers,
)
require.Equal(t, http.StatusTooManyRequests, failoverErr.StatusCode)
require.Equal(t, 0, repo.setRateLimitedCalls)
require.Equal(t, 1, repo.setModelRateLimitCalls)
require.Equal(t, "gpt-5.3-codex-spark", repo.lastModelRateLimitKey)
require.False(t, svc.isOpenAIAccountRuntimeBlocked(account))
}
func TestOpenAIWSErrorEvent_OrdinaryModelIgnoresHandshakeQuotaHeaders(t *testing.T) {
repo := &oauth429RateLimitRepo{}
rateLimits := NewRateLimitService(repo, nil, &config.Config{}, nil, nil)
backend/internal/service/openai_gateway_chat_completions.go
+2 -2
View File
@@ -511,7 +511,7 @@ func (s *OpenAIGatewayService) handleChatBufferedStreamingResponse(
}
message := openAICompatFailedResponseMessage(finalResponse)
if openAIStreamFailedEventShouldFailover(payload, message) {
return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message, resp.Header)
return nil, s.newOpenAIStreamFailoverErrorWithModel(c, account, false, requestID, payload, message, upstreamModel, resp.Header)
}
message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payload, message)
// response.failed 到达在 HTTP 200 SSE 流上，无真实 HTTP 错误码；统一走语义
@@ -766,7 +766,7 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse(
shouldFailover = openAIStreamErrorEventShouldFailover(payloadBytes, message)
}
if !clientOutputStarted && shouldFailover {
streamFailoverErr = s.newOpenAIStreamFailoverError(c, account, false, requestID, payloadBytes, message, resp.Header)
streamFailoverErr = s.newOpenAIStreamFailoverErrorWithModel(c, account, false, requestID, payloadBytes, message, upstreamModel, resp.Header)
return true
}
message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payloadBytes, message)
backend/internal/service/openai_gateway_messages.go
+2 -2
View File
@@ -605,7 +605,7 @@ func (s *OpenAIGatewayService) handleAnthropicBufferedStreamingResponse(
}
message := openAICompatFailedResponseMessage(finalResponse)
if openAIStreamFailedEventShouldFailover(payload, message) {
return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message, resp.Header)
return nil, s.newOpenAIStreamFailoverErrorWithModel(c, account, false, requestID, payload, message, upstreamModel, resp.Header)
}
message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payload, message)
// 统一走语义状态推断 + body 归一化（与 /v1/responses 路径一致），
@@ -1038,7 +1038,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse(
shouldFailover = openAIStreamErrorEventShouldFailover(payloadBytes, message)
}
if !clientOutputStarted && shouldFailover {
streamFailoverErr = s.newOpenAIStreamFailoverError(c, account, false, requestID, payloadBytes, message, resp.Header)
streamFailoverErr = s.newOpenAIStreamFailoverErrorWithModel(c, account, false, requestID, payloadBytes, message, upstreamModel, resp.Header)
return true
}
message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payloadBytes, message)
backend/internal/service/openai_gateway_passthrough.go
+18 -4
View File
@@ -1658,6 +1658,19 @@ func (s *OpenAIGatewayService) newOpenAIStreamFailoverError(
payload []byte,
message string,
responseHeaders ...http.Header,
) *UpstreamFailoverError {
return s.newOpenAIStreamFailoverErrorWithModel(c, account, passthrough, upstreamRequestID, payload, message, "", responseHeaders...)
}
func (s *OpenAIGatewayService) newOpenAIStreamFailoverErrorWithModel(
c *gin.Context,
account *Account,
passthrough bool,
upstreamRequestID string,
payload []byte,
message string,
canonicalModel string,
responseHeaders ...http.Header,
) *UpstreamFailoverError {
message = sanitizeUpstreamErrorMessage(strings.TrimSpace(message))
if message == "" {
@@ -1667,7 +1680,7 @@ func (s *OpenAIGatewayService) newOpenAIStreamFailoverError(
if len(responseHeaders) > 0 && responseHeaders[0] != nil {
headers = responseHeaders[0].Clone()
}
statusCode, shouldDisable := s.handleOpenAIStreamTerminalAccountSideEffects(c, account, payload, message, headers)
statusCode, shouldDisable := s.handleOpenAIStreamTerminalAccountSideEffects(c, account, payload, message, headers, canonicalModel)
// 流内 failed 事件承载于 HTTP 200；使用事件的语义状态更新账号健康，
// 再由 failover 引擎按 StatusCode/RetryableOnSameAccount 决定恢复策略。
message = s.recordOpenAIStreamUpstreamError(c, account, passthrough, upstreamRequestID, "failover", payload, message)
@@ -1738,6 +1751,7 @@ func (s *OpenAIGatewayService) nonStreamingTerminalFailureFailover(
terminalType string,
payload []byte,
message string,
canonicalModel ...string,
) *UpstreamFailoverError {
if account == nil || IsResponseCommitted(c) {
return nil
@@ -1755,7 +1769,7 @@ func (s *OpenAIGatewayService) nonStreamingTerminalFailureFailover(
headers = resp.Header
upstreamRequestID = strings.TrimSpace(resp.Header.Get("x-request-id"))
}
return s.newOpenAIStreamFailoverError(c, account, passthrough, upstreamRequestID, payload, message, headers)
return s.newOpenAIStreamFailoverErrorWithModel(c, account, passthrough, upstreamRequestID, payload, message, firstNonEmpty(canonicalModel...), headers)
}
func (s *OpenAIGatewayService) handleStreamingResponsePassthrough(
@@ -2024,7 +2038,7 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough(
}
if shouldFailover {
return resultWithUsage(),
s.newOpenAIStreamFailoverError(c, account, true, upstreamRequestID, dataBytes, failedMessage, resp.Header)
s.newOpenAIStreamFailoverErrorWithModel(c, account, true, upstreamRequestID, dataBytes, failedMessage, mappedModel, resp.Header)
}
if !cyberHit && !sawBareError {
if status, errType, errMsg, matched := applyOpenAIStreamFailedErrorPassthroughRule(c, account.Platform, dataBytes, failedMessage); matched {
@@ -2277,7 +2291,7 @@ func (s *OpenAIGatewayService) handlePassthroughSSEToJSON(resp *http.Response, c
if compactErr := newOpenAICompactFallbackSignal(c, terminalPayload, msg); compactErr != nil {
return nil, compactErr
}
if failoverErr := s.nonStreamingTerminalFailureFailover(c, resp, account, true, terminalType, terminalPayload, msg); failoverErr != nil {
if failoverErr := s.nonStreamingTerminalFailureFailover(c, resp, account, true, terminalType, terminalPayload, msg, mappedModel); failoverErr != nil {
return nil, failoverErr
}
return nil, s.writeOpenAINonStreamingProtocolError(resp, c, msg)
backend/internal/service/openai_gateway_response_handling.go
+2 -2
View File
@@ -572,7 +572,7 @@ func (s *OpenAIGatewayService) handleStreamingResponseWithReasoning(ctx context.
}
if shouldFailover {
sawFailedEvent = true
streamEarlyErr = s.newOpenAIStreamFailoverError(c, account, false, upstreamRequestID, dataBytes, failedMessage, resp.Header)
streamEarlyErr = s.newOpenAIStreamFailoverErrorWithModel(c, account, false, upstreamRequestID, dataBytes, failedMessage, mappedModel, resp.Header)
return
}
if !cyberHit && !sawBareError {
@@ -1684,7 +1684,7 @@ func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Conte
if compactErr := newOpenAICompactFallbackSignal(c, terminalPayload, msg); compactErr != nil {
return nil, compactErr
}
if failoverErr := s.nonStreamingTerminalFailureFailover(c, resp, account, false, terminalType, terminalPayload, msg); failoverErr != nil {
if failoverErr := s.nonStreamingTerminalFailureFailover(c, resp, account, false, terminalType, terminalPayload, msg, mappedModel); failoverErr != nil {
return nil, failoverErr
}
return nil, s.writeOpenAINonStreamingProtocolError(resp, c, msg)
backend/internal/service/openai_ws_http_bridge.go
+1 -1
View File
@@ -695,7 +695,7 @@ func (s *OpenAIGatewayService) proxyOpenAIWSHTTPBridgeTurn(
if account.Platform == PlatformGrok {
return nil, newOpenAIUpstreamFailoverError(statusCode, resp.Header, upstreamMessage, errMessage, false)
}
return nil, s.newOpenAIStreamFailoverError(c, account, true, resp.Header.Get("x-request-id"), upstreamMessage, errMessage, resp.Header)
return nil, s.newOpenAIStreamFailoverErrorWithModel(c, account, true, resp.Header.Get("x-request-id"), upstreamMessage, errMessage, mappedModel, resp.Header)
}
if account.Platform != PlatformGrok && !failureAccountSideEffectsApplied {
if eventType == "response.failed" || (!officialOpenAIResponses && shouldFailover && !requestScopedCapacity) {