From 8f97953e519847225c0a0b3e331fcdf58ef05406 Mon Sep 17 00:00:00 2001 From: li Date: Thu, 9 Jul 2026 17:57:52 +0800 Subject: [PATCH] =?UTF-8?q?fix(gateway):=20SSE=20response.failed=20?= =?UTF-8?q?=E5=BA=94=E7=94=A8=E9=94=99=E8=AF=AF=E9=80=8F=E4=BC=A0=E8=A7=84?= =?UTF-8?q?=E5=88=99=EF=BC=8C=E4=B8=8D=E5=86=8D=E7=A1=AC=E7=BC=96=E7=A0=81?= =?UTF-8?q?=20502?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes #3857 --- .../openai_gateway_chat_completions.go | 35 ++++- .../service/openai_gateway_messages.go | 34 ++++- ...ateway_response_failed_passthrough_test.go | 143 ++++++++++++++++++ 3 files changed, 206 insertions(+), 6 deletions(-) create mode 100644 backend/internal/service/openai_gateway_response_failed_passthrough_test.go diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index 3c086d5ee4..297b7428a8 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -425,6 +425,21 @@ func (s *OpenAIGatewayService) handleChatBufferedStreamingResponse( return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message) } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payload, message) + // response.failed 到达在 HTTP 200 SSE 流上，无真实 HTTP 错误码，传 0。 + if status, errType, errMsg, matched := applyErrorPassthroughRule( + c, account.Platform, 0, payload, + http.StatusBadGateway, "upstream_error", message, + ); matched { + if status == 0 { + status = http.StatusBadGateway + } + if errMsg == "" { + errMsg = message + } + MarkResponseCommitted(c) + writeChatCompletionsError(c, status, errType, errMsg) + return nil, fmt.Errorf("upstream response failed (passthrough): %s", errMsg) + } writeChatCompletionsError(c, http.StatusBadGateway, "upstream_error", message) return nil, fmt.Errorf("upstream response failed: %s", message) } @@ -581,14 +596,28 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse( return true } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payloadBytes, message) + defaultStatus, defaultErrType, defaultMsg := http.StatusBadGateway, "upstream_error", message + if status, errType, errMsg, matched := applyErrorPassthroughRule( + c, account.Platform, 0, payloadBytes, + defaultStatus, defaultErrType, defaultMsg, + ); matched { + if status == 0 { + status = defaultStatus + } + if errMsg == "" { + errMsg = defaultMsg + } + defaultStatus, defaultErrType, defaultMsg = status, errType, errMsg + MarkResponseCommitted(c) + } errorPayload, _ := json.Marshal(gin.H{ "error": gin.H{ - "type": "upstream_error", - "message": message, + "type": defaultErrType, + "message": defaultMsg, }, }) if c != nil && c.Writer != nil && !c.Writer.Written() { - writeChatCompletionsError(c, http.StatusBadGateway, "upstream_error", message) + writeChatCompletionsError(c, defaultStatus, defaultErrType, defaultMsg) clientOutputStarted = true } else if c != nil && c.Writer != nil && !clientDisconnected { if _, err := fmt.Fprintf(c.Writer, "data: %s\n\n", errorPayload); err != nil { diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index c2f51795c8..18442939d0 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -465,6 +465,20 @@ func (s *OpenAIGatewayService) handleAnthropicBufferedStreamingResponse( return nil, s.newOpenAIStreamFailoverError(c, account, false, requestID, payload, message) } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payload, message) + if status, errType, errMsg, matched := applyErrorPassthroughRule( + c, account.Platform, 0, payload, + http.StatusBadGateway, "api_error", message, + ); matched { + if status == 0 { + status = http.StatusBadGateway + } + if errMsg == "" { + errMsg = message + } + MarkResponseCommitted(c) + writeAnthropicError(c, status, errType, errMsg) + return nil, fmt.Errorf("upstream response failed (passthrough): %s", errMsg) + } writeAnthropicError(c, http.StatusBadGateway, "api_error", message) return nil, fmt.Errorf("upstream response failed: %s", message) } @@ -804,18 +818,32 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse( return true } message = s.recordOpenAIStreamUpstreamError(c, account, false, requestID, "http_error", payloadBytes, message) + errStatus, errType, errMsg := http.StatusBadGateway, "api_error", message + if status, et, em, matched := applyErrorPassthroughRule( + c, account.Platform, 0, payloadBytes, + errStatus, errType, errMsg, + ); matched { + if status == 0 { + status = errStatus + } + if em == "" { + em = errMsg + } + errStatus, errType, errMsg = status, et, em + MarkResponseCommitted(c) + } if !clientDisconnected { if !clientOutputStarted { - writeAnthropicError(c, http.StatusBadGateway, "api_error", message) + writeAnthropicError(c, errStatus, errType, errMsg) clientOutputStarted = true } else { writeStreamHeaders() - if _, err := fmt.Fprint(c.Writer, buildAnthropicStreamErrorSSE("api_error", message)); err == nil { + if _, err := fmt.Fprint(c.Writer, buildAnthropicStreamErrorSSE(errType, errMsg)); err == nil { c.Writer.Flush() } } } - streamNonFailoverErr = fmt.Errorf("upstream response failed: %s", message) + streamNonFailoverErr = fmt.Errorf("upstream response failed: %s", errMsg) return true } } diff --git a/backend/internal/service/openai_gateway_response_failed_passthrough_test.go b/backend/internal/service/openai_gateway_response_failed_passthrough_test.go new file mode 100644 index 0000000000..56fd66c94c --- /dev/null +++ b/backend/internal/service/openai_gateway_response_failed_passthrough_test.go @@ -0,0 +1,143 @@ +//go:build unit + +package service + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/model" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +func buildContextLengthFailedSSE() string { + failed := `{"type":"response.failed","response":{"id":"resp_err","object":"response","status":"failed","error":{"code":"context_length_exceeded","type":"invalid_request_error","message":"Your input exceeds the context window of this model. Please adjust your input and try again."},"output":[],"usage":{"input_tokens":100000,"output_tokens":0,"total_tokens":100000}}}` + return fmt.Sprintf("data: %s\n\n", failed) +} + +func bindPassthroughRule(c *gin.Context, platform string, keywords []string, responseCode int) { + svc := &ErrorPassthroughService{} + rules := make([]*cachedPassthroughRule, 0, len(keywords)) + for i, kw := range keywords { + code := responseCode + rules = append(rules, &cachedPassthroughRule{ + ErrorPassthroughRule: &model.ErrorPassthroughRule{ + ID: int64(i + 1), + Enabled: true, + Platforms: []string{platform}, + MatchMode: model.MatchModeAny, + Keywords: []string{kw}, + ResponseCode: &code, + PassthroughBody: true, + }, + lowerKeywords: []string{strings.ToLower(kw)}, + lowerPlatforms: []string{strings.ToLower(platform)}, + }) + } + svc.localCacheMu.Lock() + svc.localCache = rules + svc.localCacheMu.Unlock() + BindErrorPassthroughService(c, svc) +} + +func TestForwardAsChatCompletions_ResponseFailed_PassthroughRule(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":false}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + bindPassthroughRule(c, "openai", []string{"context_length_exceeded"}, 400) + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(buildContextLengthFailedSSE())), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + account := rawChatCompletionsTestAccount() + _, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "") + + require.Error(t, err) + require.Contains(t, err.Error(), "passthrough") + require.Equal(t, 400, rec.Code, "passthrough rule should override 502 to 400") + + respBody := rec.Body.String() + errType := gjson.Get(respBody, "error.type").String() + require.Equal(t, "upstream_error", errType) + errMsg := gjson.Get(respBody, "error.message").String() + require.NotEmpty(t, errMsg, "passthrough should preserve error message") + require.Contains(t, errMsg, "context window") +} + +func TestForwardAsAnthropic_ResponseFailed_PassthroughRule(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":32,"messages":[{"role":"user","content":"hello"}],"stream":false}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + bindPassthroughRule(c, "openai", []string{"context_length_exceeded"}, 400) + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(buildContextLengthFailedSSE())), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + account := rawChatCompletionsTestAccount() + _, err := svc.ForwardAsAnthropic(context.Background(), c, account, body, "", "") + + require.Error(t, err) + require.Contains(t, err.Error(), "passthrough") + require.Equal(t, 400, rec.Code, "passthrough rule should override 502 to 400") + respBody := rec.Body.String() + errMsg := gjson.Get(respBody, "error.message").String() + require.NotEmpty(t, errMsg, "passthrough should preserve error message") +} + +func TestForwardAsChatCompletions_ResponseFailed_NoRule_Still502(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":false}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(buildContextLengthFailedSSE())), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + account := rawChatCompletionsTestAccount() + _, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "") + + require.Error(t, err) + require.Equal(t, http.StatusBadGateway, rec.Code, "without passthrough rule should still be 502") +}