Merge pull request #904 from james-6-23/fix-pool-mode-retry

fix: OpenAI临时性400错误支持池模式同账号重试 & HelpTooltip层级修复
This commit is contained in:
Wesley Liddick
2026-03-10 09:08:12 +08:00
committed by GitHub
4 changed files with 160 additions and 25 deletions

View File

@@ -2,6 +2,7 @@ package service
import ( import (
"bufio" "bufio"
"bytes"
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
@@ -140,12 +141,13 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic(
// 8. Handle error response with failover // 8. Handle error response with failover
if resp.StatusCode >= 400 { if resp.StatusCode >= 400 {
if s.shouldFailoverUpstreamError(resp.StatusCode) { respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20)) _ = resp.Body.Close()
_ = resp.Body.Close() resp.Body = io.NopCloser(bytes.NewReader(respBody))
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody)) upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
upstreamDetail := "" upstreamDetail := ""
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
@@ -167,7 +169,11 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic(
if s.rateLimitService != nil { if s.rateLimitService != nil {
s.rateLimitService.HandleUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody) s.rateLimitService.HandleUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody)
} }
return nil, &UpstreamFailoverError{StatusCode: resp.StatusCode, ResponseBody: respBody} return nil, &UpstreamFailoverError{
StatusCode: resp.StatusCode,
ResponseBody: respBody,
RetryableOnSameAccount: account.IsPoolMode() && isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody),
}
} }
// Non-failover error: return Anthropic-formatted error to client // Non-failover error: return Anthropic-formatted error to client
return s.handleAnthropicErrorResponse(resp, c, account) return s.handleAnthropicErrorResponse(resp, c, account)

View File

@@ -911,6 +911,36 @@ func isOpenAIInstructionsRequiredError(upstreamStatusCode int, upstreamMsg strin
return false return false
} }
func isOpenAITransientProcessingError(upstreamStatusCode int, upstreamMsg string, upstreamBody []byte) bool {
if upstreamStatusCode != http.StatusBadRequest {
return false
}
match := func(text string) bool {
lower := strings.ToLower(strings.TrimSpace(text))
if lower == "" {
return false
}
if strings.Contains(lower, "an error occurred while processing your request") {
return true
}
return strings.Contains(lower, "you can retry your request") &&
strings.Contains(lower, "help.openai.com") &&
strings.Contains(lower, "request id")
}
if match(upstreamMsg) {
return true
}
if len(upstreamBody) == 0 {
return false
}
if match(gjson.GetBytes(upstreamBody, "error.message").String()) {
return true
}
return match(string(upstreamBody))
}
// ExtractSessionID extracts the raw session ID from headers or body without hashing. // ExtractSessionID extracts the raw session ID from headers or body without hashing.
// Used by ForwardAsAnthropic to pass as prompt_cache_key for upstream cache. // Used by ForwardAsAnthropic to pass as prompt_cache_key for upstream cache.
func (s *OpenAIGatewayService) ExtractSessionID(c *gin.Context, body []byte) string { func (s *OpenAIGatewayService) ExtractSessionID(c *gin.Context, body []byte) string {
@@ -1518,6 +1548,13 @@ func (s *OpenAIGatewayService) shouldFailoverUpstreamError(statusCode int) bool
} }
} }
func (s *OpenAIGatewayService) shouldFailoverOpenAIUpstreamResponse(statusCode int, upstreamMsg string, upstreamBody []byte) bool {
if s.shouldFailoverUpstreamError(statusCode) {
return true
}
return isOpenAITransientProcessingError(statusCode, upstreamMsg, upstreamBody)
}
func (s *OpenAIGatewayService) handleFailoverSideEffects(ctx context.Context, resp *http.Response, account *Account) { func (s *OpenAIGatewayService) handleFailoverSideEffects(ctx context.Context, resp *http.Response, account *Account) {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20)) body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
s.rateLimitService.HandleUpstreamError(ctx, account, resp.StatusCode, resp.Header, body) s.rateLimitService.HandleUpstreamError(ctx, account, resp.StatusCode, resp.Header, body)
@@ -2016,13 +2053,13 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
// Handle error response // Handle error response
if resp.StatusCode >= 400 { if resp.StatusCode >= 400 {
if s.shouldFailoverUpstreamError(resp.StatusCode) { respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20)) _ = resp.Body.Close()
_ = resp.Body.Close() resp.Body = io.NopCloser(bytes.NewReader(respBody))
resp.Body = io.NopCloser(bytes.NewReader(respBody))
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody)) upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
upstreamDetail := "" upstreamDetail := ""
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes
@@ -2046,7 +2083,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
return nil, &UpstreamFailoverError{ return nil, &UpstreamFailoverError{
StatusCode: resp.StatusCode, StatusCode: resp.StatusCode,
ResponseBody: respBody, ResponseBody: respBody,
RetryableOnSameAccount: account.IsPoolMode() && isPoolModeRetryableStatus(resp.StatusCode), RetryableOnSameAccount: account.IsPoolMode() && (isPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)),
} }
} }
return s.handleErrorResponse(ctx, resp, c, account, body) return s.handleErrorResponse(ctx, resp, c, account, body)

View File

@@ -211,6 +211,26 @@ func TestLogOpenAIInstructionsRequiredDebug_NonTargetErrorSkipped(t *testing.T)
require.False(t, logSink.ContainsMessage("OpenAI 上游返回 Instructions are required已记录请求详情用于排查")) require.False(t, logSink.ContainsMessage("OpenAI 上游返回 Instructions are required已记录请求详情用于排查"))
} }
func TestIsOpenAITransientProcessingError(t *testing.T) {
require.True(t, isOpenAITransientProcessingError(
http.StatusBadRequest,
"An error occurred while processing your request.",
nil,
))
require.True(t, isOpenAITransientProcessingError(
http.StatusBadRequest,
"",
[]byte(`{"error":{"message":"An error occurred while processing your request. You can retry your request, or contact us through our help center at help.openai.com if the error persists. Please include the request ID req_123 in your message."}}`),
))
require.False(t, isOpenAITransientProcessingError(
http.StatusBadRequest,
"Missing required parameter: 'instructions'",
[]byte(`{"error":{"message":"Missing required parameter: 'instructions'"}}`),
))
}
func TestOpenAIGatewayService_Forward_LogsInstructionsRequiredDetails(t *testing.T) { func TestOpenAIGatewayService_Forward_LogsInstructionsRequiredDetails(t *testing.T) {
gin.SetMode(gin.TestMode) gin.SetMode(gin.TestMode)
logSink, restore := captureStructuredLog(t) logSink, restore := captureStructuredLog(t)
@@ -264,3 +284,51 @@ func TestOpenAIGatewayService_Forward_LogsInstructionsRequiredDetails(t *testing
require.True(t, logSink.ContainsField("request_body_size")) require.True(t, logSink.ContainsField("request_body_size"))
require.False(t, logSink.ContainsField("request_body_preview")) require.False(t, logSink.ContainsField("request_body_preview"))
} }
func TestOpenAIGatewayService_Forward_TransientProcessingErrorTriggersFailover(t *testing.T) {
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(nil))
c.Request.Header.Set("User-Agent", "codex_cli_rs/0.1.0")
c.Request.Header.Set("Content-Type", "application/json")
upstream := &httpUpstreamRecorder{
resp: &http.Response{
StatusCode: http.StatusBadRequest,
Header: http.Header{
"Content-Type": []string{"application/json"},
"x-request-id": []string{"rid-processing-400"},
},
Body: io.NopCloser(strings.NewReader(`{"error":{"message":"An error occurred while processing your request. You can retry your request, or contact us through our help center at help.openai.com if the error persists. Please include the request ID req_123 in your message.","type":"invalid_request_error"}}`)),
},
}
svc := &OpenAIGatewayService{
cfg: &config.Config{
Gateway: config.GatewayConfig{ForceCodexCLI: false},
},
httpUpstream: upstream,
}
account := &Account{
ID: 1001,
Name: "codex max套餐",
Platform: PlatformOpenAI,
Type: AccountTypeAPIKey,
Concurrency: 1,
Credentials: map[string]any{"api_key": "sk-test"},
Status: StatusActive,
Schedulable: true,
RateMultiplier: f64p(1),
}
body := []byte(`{"model":"gpt-5.1-codex","stream":false,"input":[{"type":"text","text":"hello"}]}`)
_, err := svc.Forward(context.Background(), c, account, body)
require.Error(t, err)
var failoverErr *UpstreamFailoverError
require.ErrorAs(t, err, &failoverErr)
require.Equal(t, http.StatusBadRequest, failoverErr.StatusCode)
require.Contains(t, string(failoverErr.ResponseBody), "An error occurred while processing your request")
require.False(t, c.Writer.Written(), "service 层应返回 failover 错误给上层换号,而不是直接向客户端写响应")
}

View File

@@ -1,18 +1,40 @@
<script setup lang="ts"> <script setup lang="ts">
import { ref } from 'vue' import { ref, useTemplateRef, nextTick } from 'vue'
defineProps<{ defineProps<{
content?: string content?: string
}>() }>()
const show = ref(false) const show = ref(false)
const triggerRef = useTemplateRef<HTMLElement>('trigger')
const tooltipStyle = ref({ top: '0px', left: '0px' })
function onEnter() {
show.value = true
nextTick(updatePosition)
}
function onLeave() {
show.value = false
}
function updatePosition() {
const el = triggerRef.value
if (!el) return
const rect = el.getBoundingClientRect()
tooltipStyle.value = {
top: `${rect.top + window.scrollY}px`,
left: `${rect.left + rect.width / 2 + window.scrollX}px`,
}
}
</script> </script>
<template> <template>
<div <div
ref="trigger"
class="group relative ml-1 inline-flex items-center align-middle" class="group relative ml-1 inline-flex items-center align-middle"
@mouseenter="show = true" @mouseenter="onEnter"
@mouseleave="show = false" @mouseleave="onLeave"
> >
<!-- Trigger Icon --> <!-- Trigger Icon -->
<slot name="trigger"> <slot name="trigger">
@@ -31,14 +53,16 @@ const show = ref(false)
</svg> </svg>
</slot> </slot>
<!-- Popover Content --> <!-- Teleport to body to escape modal overflow clipping -->
<div <Teleport to="body">
v-show="show" <div
class="absolute bottom-full left-1/2 z-50 mb-2 w-64 -translate-x-1/2 rounded-lg bg-gray-900 p-3 text-xs leading-relaxed text-white shadow-xl ring-1 ring-white/10 opacity-0 transition-opacity duration-200 group-hover:opacity-100 dark:bg-gray-800" v-show="show"
> class="fixed z-[99999] w-64 -translate-x-1/2 -translate-y-full rounded-lg bg-gray-900 p-3 text-xs leading-relaxed text-white shadow-xl ring-1 ring-white/10 dark:bg-gray-800"
<slot>{{ content }}</slot> :style="{ top: `calc(${tooltipStyle.top} - 8px)`, left: tooltipStyle.left }"
<div class="absolute -bottom-1 left-1/2 h-2 w-2 -translate-x-1/2 rotate-45 bg-gray-900 dark:bg-gray-800"></div> >
</div> <slot>{{ content }}</slot>
<div class="absolute -bottom-1 left-1/2 h-2 w-2 -translate-x-1/2 rotate-45 bg-gray-900 dark:bg-gray-800"></div>
</div>
</Teleport>
</div> </div>
</template> </template>