diff --git a/backend/internal/handler/concurrency_error_response.go b/backend/internal/handler/concurrency_error_response.go index 911f5a54d7..cd631a47b4 100644 --- a/backend/internal/handler/concurrency_error_response.go +++ b/backend/internal/handler/concurrency_error_response.go @@ -9,10 +9,15 @@ import ( const statusClientClosedRequest = 499 -func concurrencyErrorResponse(err error, slotType string) (int, string, string) { +const ( + gatewayQueueFullCode = "gateway_queue_full" + gatewayConcurrencyLimitCode = "gateway_concurrency_limit" +) + +func concurrencyErrorResponse(err error, slotType string) (int, string, string, string) { var waitQueueFullErr *WaitQueueFullError if errors.As(err, &waitQueueFullErr) { - return http.StatusTooManyRequests, "rate_limit_error", + return http.StatusTooManyRequests, "rate_limit_error", gatewayQueueFullCode, "Too many pending requests, please retry later" } @@ -21,13 +26,13 @@ func concurrencyErrorResponse(err error, slotType string) (int, string, string) if concurrencyErr.SlotType != "" { slotType = concurrencyErr.SlotType } - return http.StatusTooManyRequests, "rate_limit_error", + return http.StatusTooManyRequests, "rate_limit_error", gatewayConcurrencyLimitCode, fmt.Sprintf("Concurrency limit exceeded for %s, please retry later", slotType) } if errors.Is(err, context.Canceled) { - return statusClientClosedRequest, "api_error", "context canceled" + return statusClientClosedRequest, "api_error", "", "context canceled" } - return http.StatusServiceUnavailable, "api_error", "Service temporarily unavailable, please retry later" + return http.StatusServiceUnavailable, "api_error", "", "Service temporarily unavailable, please retry later" } diff --git a/backend/internal/handler/concurrency_error_response_test.go b/backend/internal/handler/concurrency_error_response_test.go index a2e6b9ab93..2d1b3d6ad2 100644 --- a/backend/internal/handler/concurrency_error_response_test.go +++ b/backend/internal/handler/concurrency_error_response_test.go @@ -16,6 +16,7 @@ func TestConcurrencyErrorResponse(t *testing.T) { slotType string wantStatus int wantType string + wantCode string wantMessage string }{ { @@ -24,8 +25,18 @@ func TestConcurrencyErrorResponse(t *testing.T) { slotType: "user", wantStatus: http.StatusTooManyRequests, wantType: "rate_limit_error", + wantCode: gatewayConcurrencyLimitCode, wantMessage: "Concurrency limit exceeded for account, please retry later", }, + { + name: "full local wait queue has gateway code", + err: &WaitQueueFullError{SlotType: "account"}, + slotType: "account", + wantStatus: http.StatusTooManyRequests, + wantType: "rate_limit_error", + wantCode: gatewayQueueFullCode, + wantMessage: "Too many pending requests, please retry later", + }, { name: "client cancellation is not classified as concurrency limit", err: context.Canceled, @@ -54,9 +65,10 @@ func TestConcurrencyErrorResponse(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - status, errType, message := concurrencyErrorResponse(tt.err, tt.slotType) + status, errType, code, message := concurrencyErrorResponse(tt.err, tt.slotType) require.Equal(t, tt.wantStatus, status) require.Equal(t, tt.wantType, errType) + require.Equal(t, tt.wantCode, code) require.Equal(t, tt.wantMessage, message) }) } diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index a4bcda50fa..9384d46643 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -395,7 +395,7 @@ func (h *GatewayHandler) Messages(c *gin.Context) { zap.Int64("account_id", account.ID), zap.Int("max_waiting", selection.WaitPlan.MaxWaiting), ) - h.handleStreamingAwareError(c, http.StatusTooManyRequests, "rate_limit_error", "Too many pending requests, please retry later", streamStarted) + h.handleStreamingAwareErrorWithCode(c, http.StatusTooManyRequests, "rate_limit_error", gatewayQueueFullCode, "Too many pending requests, please retry later", streamStarted) return } if err == nil && canWait { @@ -719,7 +719,7 @@ func (h *GatewayHandler) Messages(c *gin.Context) { zap.Int64("account_id", account.ID), zap.Int("max_waiting", selection.WaitPlan.MaxWaiting), ) - h.handleStreamingAwareError(c, http.StatusTooManyRequests, "rate_limit_error", "Too many pending requests, please retry later", streamStarted) + h.handleStreamingAwareErrorWithCode(c, http.StatusTooManyRequests, "rate_limit_error", gatewayQueueFullCode, "Too many pending requests, please retry later", streamStarted) return } if err == nil && canWait { @@ -1853,8 +1853,8 @@ func (h *GatewayHandler) calculateSubscriptionRemaining(group *service.Group, su // handleConcurrencyError handles concurrency-related acquire errors. func (h *GatewayHandler) handleConcurrencyError(c *gin.Context, err error, slotType string, streamStarted bool) { - status, errType, message := concurrencyErrorResponse(err, slotType) - h.handleStreamingAwareError(c, status, errType, message, streamStarted) + status, errType, code, message := concurrencyErrorResponse(err, slotType) + h.handleStreamingAwareErrorWithCode(c, status, errType, code, message, streamStarted) } func (h *GatewayHandler) handleFailoverExhausted(c *gin.Context, failoverErr *service.UpstreamFailoverError, platform string, streamStarted bool) { @@ -1925,6 +1925,10 @@ func (h *GatewayHandler) mapUpstreamError(statusCode int) (int, string, string) // handleStreamingAwareError handles errors that may occur after streaming has started func (h *GatewayHandler) handleStreamingAwareError(c *gin.Context, status int, errType, message string, streamStarted bool) { + h.handleStreamingAwareErrorWithCode(c, status, errType, "", message, streamStarted) +} + +func (h *GatewayHandler) handleStreamingAwareErrorWithCode(c *gin.Context, status int, errType, code, message string, streamStarted bool) { if streamStarted { // 响应状态码已固化为 200（ping/部分数据已 flush），错误只能就地以 SSE 帧回传。 // 标记本次流内错误，供 ops_error_logger 补记——否则该中间件按 status>=400 采集， @@ -1935,7 +1939,7 @@ func (h *GatewayHandler) handleStreamingAwareError(c *gin.Context, status int, e // response.completed/failed/incomplete/cancelled 集合。 // Anthropic-backed Responses 路径同样会因为通用 error 帧被拒。 if inboundIsResponses(c) { - if writeResponsesFailedSSE(c, errType, message) { + if writeResponsesFailedSSE(c, errType, code, message) { return } } @@ -1943,7 +1947,11 @@ func (h *GatewayHandler) handleStreamingAwareError(c *gin.Context, status int, e flusher, ok := c.Writer.(http.Flusher) if ok { // SSE 错误事件固定 schema，使用 Quote 直拼可避免额外 Marshal 分配。 - errorEvent := `data: {"type":"error","error":{"type":` + strconv.Quote(errType) + `,"message":` + strconv.Quote(message) + `}}` + "\n\n" + errorCode := "" + if code != "" { + errorCode = `,"code":` + strconv.Quote(code) + } + errorEvent := `data: {"type":"error","error":{"type":` + strconv.Quote(errType) + errorCode + `,"message":` + strconv.Quote(message) + `}}` + "\n\n" if _, err := fmt.Fprint(c.Writer, errorEvent); err != nil { _ = c.Error(err) } @@ -1953,7 +1961,7 @@ func (h *GatewayHandler) handleStreamingAwareError(c *gin.Context, status int, e } // Normal case: return JSON response with proper status code - h.errorResponse(c, status, errType, message) + h.errorResponseWithCode(c, status, errType, code, message) } // ensureForwardErrorResponse 在 Forward 返回错误但尚未写响应时补写统一错误响应。 @@ -2045,12 +2053,17 @@ func (h *GatewayHandler) checkClaudeCodeVersion(c *gin.Context) bool { // errorResponse 返回Claude API格式的错误响应 func (h *GatewayHandler) errorResponse(c *gin.Context, status int, errType, message string) { + h.errorResponseWithCode(c, status, errType, "", message) +} + +func (h *GatewayHandler) errorResponseWithCode(c *gin.Context, status int, errType, code, message string) { + errorObject := gin.H{"type": errType, "message": message} + if code != "" { + errorObject["code"] = code + } c.JSON(status, gin.H{ - "type": "error", - "error": gin.H{ - "type": errType, - "message": message, - }, + "type": "error", + "error": errorObject, }) } diff --git a/backend/internal/handler/gateway_handler_responses.go b/backend/internal/handler/gateway_handler_responses.go index 04a61cb16e..a170877315 100644 --- a/backend/internal/handler/gateway_handler_responses.go +++ b/backend/internal/handler/gateway_handler_responses.go @@ -394,7 +394,7 @@ func (h *GatewayHandler) handleResponsesFailoverExhausted(c *gin.Context, lastEr // generic response.failed. service.MarkOpsStreamError(c, code, message, status) if c != nil && c.Writer != nil && (c.Writer.Size() <= 0 || gatewayStreamHasOnlyHeartbeats(c)) { - writeResponsesFailedSSE(c, code, message) + writeResponsesFailedSSE(c, code, "", message) } return } diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index ce67e07e85..771169016b 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -2004,7 +2004,7 @@ func (h *OpenAIGatewayHandler) acquireResponsesAccountSlot( return h.acquireOpenAIAccountSlot(c, groupID, sessionHash, selection, reqStream, streamStarted, reqLog, nil) } -type openAISlotErrorWriter func(status int, errType, message string) +type openAISlotErrorWriter func(status int, errType, code, message string) // acquireOpenAIAccountSlot centralizes scheduler selection admission. The // optional error writer lets non-Responses endpoints retain their wire format @@ -2020,13 +2020,13 @@ func (h *OpenAIGatewayHandler) acquireOpenAIAccountSlot( writeError openAISlotErrorWriter, ) (func(), openAISlotAcquireResult) { if writeError == nil { - writeError = func(status int, errType, message string) { - h.handleStreamingAwareError(c, status, errType, message, *streamStarted) + writeError = func(status int, errType, code, message string) { + h.handleStreamingAwareErrorWithCode(c, status, errType, code, message, *streamStarted, false) } } if selection == nil || selection.Account == nil { markOpsRoutingCapacityLimited(c) - writeError(http.StatusServiceUnavailable, "api_error", "No available accounts") + writeError(http.StatusServiceUnavailable, "api_error", "", "No available accounts") return nil, openAISlotAcquireFailed } @@ -2056,7 +2056,7 @@ func (h *OpenAIGatewayHandler) acquireOpenAIAccountSlot( } if selection.WaitPlan == nil { markOpsRoutingCapacityLimited(c) - writeError(http.StatusServiceUnavailable, "api_error", "No available accounts") + writeError(http.StatusServiceUnavailable, "api_error", "", "No available accounts") return nil, openAISlotAcquireFailed } @@ -2067,8 +2067,8 @@ func (h *OpenAIGatewayHandler) acquireOpenAIAccountSlot( ) if err != nil { reqLog.Warn("openai.account_slot_quick_acquire_failed", zap.Int64("account_id", account.ID), zap.Error(err)) - status, errType, message := concurrencyErrorResponse(err, "account") - writeError(status, errType, message) + status, errType, code, message := concurrencyErrorResponse(err, "account") + writeError(status, errType, code, message) return nil, openAISlotAcquireFailed } if fastAcquired { @@ -2098,7 +2098,7 @@ func (h *OpenAIGatewayHandler) acquireOpenAIAccountSlot( zap.Int64("account_id", account.ID), zap.Int("max_waiting", selection.WaitPlan.MaxWaiting), ) - writeError(http.StatusTooManyRequests, "rate_limit_error", "Too many pending requests, please retry later") + writeError(http.StatusTooManyRequests, "rate_limit_error", gatewayQueueFullCode, "Too many pending requests, please retry later") return nil, openAISlotAcquireFailed } @@ -2121,8 +2121,8 @@ func (h *OpenAIGatewayHandler) acquireOpenAIAccountSlot( ) if err != nil { reqLog.Warn("openai.account_slot_acquire_failed", zap.Int64("account_id", account.ID), zap.Error(err)) - status, errType, message := concurrencyErrorResponse(err, "account") - writeError(status, errType, message) + status, errType, code, message := concurrencyErrorResponse(err, "account") + writeError(status, errType, code, message) return nil, openAISlotAcquireFailed } @@ -3172,14 +3172,14 @@ func (h *OpenAIGatewayHandler) acquireImageGenerationSlot(c *gin.Context, stream if acquired { return release, true } - h.handleStreamingAwareError(c, http.StatusTooManyRequests, "rate_limit_error", "Image generation concurrency limit exceeded, please retry later", streamStarted) + h.handleStreamingAwareErrorWithCode(c, http.StatusTooManyRequests, "rate_limit_error", gatewayConcurrencyLimitCode, "Image generation concurrency limit exceeded, please retry later", streamStarted, false) return nil, false } // handleConcurrencyError handles concurrency-related acquire errors. func (h *OpenAIGatewayHandler) handleConcurrencyError(c *gin.Context, err error, slotType string, streamStarted bool) { - status, errType, message := concurrencyErrorResponse(err, slotType) - h.handleStreamingAwareError(c, status, errType, message, streamStarted) + status, errType, code, message := concurrencyErrorResponse(err, slotType) + h.handleStreamingAwareErrorWithCode(c, status, errType, code, message, streamStarted, false) } func (h *OpenAIGatewayHandler) handleFailoverExhausted(c *gin.Context, failoverErr *service.UpstreamFailoverError, streamStarted bool) { @@ -3359,7 +3359,7 @@ func (h *OpenAIGatewayHandler) handleStreamingAwareErrorWithCode( // 通用 `event: error` 帧不被识别为终止事件，会导致 // "stream closed before response.completed"。 if inboundIsResponses(c) { - if writeResponsesFailedSSE(c, errType, message) { + if writeResponsesFailedSSE(c, errType, code, message) { return } } @@ -3526,7 +3526,7 @@ func (h *OpenAIGatewayHandler) errorResponse(c *gin.Context, status int, errType // 提交的 SSE 流交错，必须降级为 response.failed 终止事件（#3887）。 if service.StopOpenAICompactSSEKeepaliveCommitted(c) { service.MarkOpsStreamError(c, errType, message, status) - if writeResponsesFailedSSE(c, errType, message) { + if writeResponsesFailedSSE(c, errType, "", message) { return } } @@ -3844,7 +3844,7 @@ func (h *OpenAIGatewayHandler) rejectIfCyberSessionBlocked(c *gin.Context, apiKe // 写 JSON（#3887）。 if service.StopOpenAICompactSSEKeepaliveCommitted(c) { service.MarkOpsStreamError(c, "permission_error", cyberSessionBlockedClientMsg, http.StatusForbidden) - if writeResponsesFailedSSE(c, "permission_error", cyberSessionBlockedClientMsg) { + if writeResponsesFailedSSE(c, "permission_error", "", cyberSessionBlockedClientMsg) { h.enqueueCyberSessionBlockedOpsEntry(c, apiKey, model, key) return true } diff --git a/backend/internal/handler/stream_error_event.go b/backend/internal/handler/stream_error_event.go index 8ba6095cfb..f60e6c7834 100644 --- a/backend/internal/handler/stream_error_event.go +++ b/backend/internal/handler/stream_error_event.go @@ -56,7 +56,7 @@ type responsesFailedEvent struct { // 返回 false 表示 writer 不支持 Flusher，无法以 SSE 形式回报错误； // 此时 caller 也无法回退到 JSON（HTTP 200 已固化），通常意味着连接已经损坏， // 应当让请求处理函数 return，由上层关闭连接。 -func writeResponsesFailedSSE(c *gin.Context, errType, message string) bool { +func writeResponsesFailedSSE(c *gin.Context, errType, code, message string) bool { flusher, ok := c.Writer.(http.Flusher) if !ok { return false @@ -72,7 +72,7 @@ func writeResponsesFailedSSE(c *gin.Context, errType, message string) bool { Status: "failed", Output: []any{}, Error: responsesFailedError{ - Code: mapResponsesErrorCode(errType), + Code: mapResponsesErrorCode(errType, code), Message: message, }, }, @@ -155,7 +155,10 @@ func requestModel(c *gin.Context) string { // mapResponsesErrorCode 把内部 errType 映射为 Responses 协议常见的 error.code。 // 无明确映射时原样返回，保证至少可读。 -func mapResponsesErrorCode(errType string) string { +func mapResponsesErrorCode(errType, code string) string { + if code != "" { + return code + } switch errType { case "rate_limit_error": return "rate_limit_exceeded" diff --git a/backend/internal/handler/stream_error_event_test.go b/backend/internal/handler/stream_error_event_test.go index 187deda19d..94089e9fae 100644 --- a/backend/internal/handler/stream_error_event_test.go +++ b/backend/internal/handler/stream_error_event_test.go @@ -12,6 +12,7 @@ import ( "github.com/gin-gonic/gin" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" ) // Regression for the production incident on 2026-05-24 around 9:13 CST: @@ -77,6 +78,60 @@ func TestOpenAIHandleStreamingAwareError_ResponsesStreamingEmitsResponseFailed(t assert.Equal(t, "Concurrency limit exceeded for user, please retry later", errObj["message"]) } +func TestOpenAIAdmissionError_SynchronousNonResponsesIncludesGatewayCode(t *testing.T) { + c, w := newGinContextForEndpoint(t, EndpointChatCompletions) + h := &OpenAIGatewayHandler{} + h.handleStreamingAwareErrorWithCode(c, http.StatusTooManyRequests, "rate_limit_error", + gatewayQueueFullCode, "Too many pending requests, please retry later", false, false) + + assert.Equal(t, http.StatusTooManyRequests, w.Code) + assert.Equal(t, "rate_limit_error", gjson.GetBytes(w.Body.Bytes(), "error.type").String()) + assert.Equal(t, gatewayQueueFullCode, gjson.GetBytes(w.Body.Bytes(), "error.code").String()) +} + +func TestOpenAIAdmissionError_SynchronousResponsesIncludesGatewayCode(t *testing.T) { + c, w := newGinContextForEndpoint(t, EndpointResponses) + h := &OpenAIGatewayHandler{} + h.handleStreamingAwareErrorWithCode(c, http.StatusTooManyRequests, "rate_limit_error", + gatewayConcurrencyLimitCode, "Concurrency limit exceeded for account, please retry later", false, false) + + assert.Equal(t, http.StatusTooManyRequests, w.Code) + assert.Equal(t, "rate_limit_error", gjson.GetBytes(w.Body.Bytes(), "error.type").String()) + assert.Equal(t, gatewayConcurrencyLimitCode, gjson.GetBytes(w.Body.Bytes(), "error.code").String()) +} + +func TestOpenAIAdmissionError_ResponsesFailedIncludesGatewayCode(t *testing.T) { + c, w := newGinContextForEndpoint(t, EndpointResponses) + h := &OpenAIGatewayHandler{} + h.handleStreamingAwareErrorWithCode(c, http.StatusTooManyRequests, "rate_limit_error", + gatewayConcurrencyLimitCode, "Concurrency limit exceeded for account, please retry later", true, false) + + _, errObj := parseResponsesFailedSSE(t, w.Body.String()) + assert.Equal(t, gatewayConcurrencyLimitCode, errObj["code"]) +} + +func TestOpenAIAdmissionError_StreamingNonResponsesIncludesGatewayCode(t *testing.T) { + c, w := newGinContextForEndpoint(t, EndpointChatCompletions) + h := &OpenAIGatewayHandler{} + h.handleStreamingAwareErrorWithCode(c, http.StatusTooManyRequests, "rate_limit_error", + gatewayQueueFullCode, "Too many pending requests, please retry later", true, false) + + body := w.Body.String() + assert.True(t, strings.HasPrefix(body, "event: error\n")) + payload := body[strings.Index(body, "{"):] + assert.Equal(t, "rate_limit_error", gjson.Get(payload, "error.type").String()) + assert.Equal(t, gatewayQueueFullCode, gjson.Get(payload, "error.code").String()) +} + +func TestOpenAIUpstreamRateLimitWithoutLocalCodeRemainsUnchanged(t *testing.T) { + c, w := newGinContextForEndpoint(t, EndpointChatCompletions) + h := &OpenAIGatewayHandler{} + h.handleStreamingAwareError(c, http.StatusTooManyRequests, "rate_limit_error", "upstream limited", false) + + assert.Equal(t, "rate_limit_error", gjson.GetBytes(w.Body.Bytes(), "error.type").String()) + assert.False(t, gjson.GetBytes(w.Body.Bytes(), "error.code").Exists()) +} + // 当 setOpsRequestContext 写过 model，合成事件应回填该字段（与 codebase 已有 makeResponsesCompletedEvent 对齐）。 func TestOpenAIHandleStreamingAwareError_ResponsesStreamingIncludesModel(t *testing.T) { c, w := newGinContextForEndpoint(t, EndpointResponses) @@ -164,6 +219,41 @@ func TestGatewayHandleStreamingAwareError_ResponsesStreamingEmitsResponseFailed( assert.Equal(t, "upstream gone", errObj["message"]) } +func TestGatewayAdmissionError_SynchronousIncludesGatewayCode(t *testing.T) { + c, w := newGinContextForEndpoint(t, EndpointMessages) + h := &GatewayHandler{} + h.handleStreamingAwareErrorWithCode(c, http.StatusTooManyRequests, "rate_limit_error", + gatewayQueueFullCode, "Too many pending requests, please retry later", false) + + assert.Equal(t, http.StatusTooManyRequests, w.Code) + assert.Equal(t, "error", gjson.GetBytes(w.Body.Bytes(), "type").String()) + assert.Equal(t, "rate_limit_error", gjson.GetBytes(w.Body.Bytes(), "error.type").String()) + assert.Equal(t, gatewayQueueFullCode, gjson.GetBytes(w.Body.Bytes(), "error.code").String()) +} + +func TestGatewayAdmissionError_MessagesStreamingIncludesGatewayCode(t *testing.T) { + c, w := newGinContextForEndpoint(t, EndpointMessages) + h := &GatewayHandler{} + h.handleStreamingAwareErrorWithCode(c, http.StatusTooManyRequests, "rate_limit_error", + gatewayConcurrencyLimitCode, "Concurrency limit exceeded for account, please retry later", true) + + body := w.Body.String() + assert.True(t, strings.HasPrefix(body, `data: {"type":"error"`)) + payload := body[strings.Index(body, "{"):] + assert.Equal(t, "rate_limit_error", gjson.Get(payload, "error.type").String()) + assert.Equal(t, gatewayConcurrencyLimitCode, gjson.Get(payload, "error.code").String()) +} + +func TestGatewayAdmissionError_BareResponsesFailedIncludesGatewayCode(t *testing.T) { + c, w := newGinContextForEndpoint(t, "/responses") + h := &GatewayHandler{} + h.handleStreamingAwareErrorWithCode(c, http.StatusTooManyRequests, "rate_limit_error", + gatewayQueueFullCode, "Too many pending requests, please retry later", true) + + _, errObj := parseResponsesFailedSSE(t, w.Body.String()) + assert.Equal(t, gatewayQueueFullCode, errObj["code"]) +} + // Gateway handler: /v1/messages preserves the legacy data:{type:error,...} format // (Anthropic spec accepts a type:"error" stream event). func TestGatewayHandleStreamingAwareError_MessagesStreamingKeepsLegacy(t *testing.T) { @@ -264,6 +354,7 @@ func TestMapResponsesErrorCode(t *testing.T) { {"custom_thing", "custom_thing"}, } for _, tc := range cases { - assert.Equal(t, tc.out, mapResponsesErrorCode(tc.in), "in=%q", tc.in) + assert.Equal(t, tc.out, mapResponsesErrorCode(tc.in, ""), "in=%q", tc.in) } + assert.Equal(t, gatewayQueueFullCode, mapResponsesErrorCode("rate_limit_error", gatewayQueueFullCode)) }