From 76a13a5a8d56dca2befb8a3d70daf356c540f336 Mon Sep 17 00:00:00 2001 From: wucm667 Date: Thu, 13 Aug 2026 02:43:03 +0800 Subject: [PATCH] fix(gateway): handle Anthropic SSE overload errors --- backend/internal/service/gateway_forward.go | 18 ++- .../gateway_forward_partial_usage_test.go | 114 ++++++++++++++++++ 2 files changed, 128 insertions(+), 4 deletions(-) diff --git a/backend/internal/service/gateway_forward.go b/backend/internal/service/gateway_forward.go index d10cb1ea3b..863e4d2e83 100644 --- a/backend/internal/service/gateway_forward.go +++ b/backend/internal/service/gateway_forward.go @@ -14,6 +14,7 @@ import ( "github.com/Wei-Shaw/sub2api/internal/pkg/claude" "github.com/Wei-Shaw/sub2api/internal/pkg/logger" + "github.com/tidwall/gjson" "github.com/gin-gonic/gin" ) @@ -800,14 +801,23 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A var firstTokenMs *int var clientDisconnect bool if reqStream { + writerSizeBeforeStream := c.Writer.Size() streamResult, err := s.handleStreamingResponse(ctx, resp, c, account, startTime, originalModel, reqModel, shouldMimicClaudeCode) if err != nil { var sseErr *sseStreamErrorEventError if errors.As(err, &sseErr) { // 上游 HTTP 200 + SSE 流体内出现 event:error 帧。 - // 保留 StatusCode=403 以兼容既有 failover/客户端响应语义， - // 但补全 ResponseBody 与 ops 上下文，让运维日志能反映上游真实错误。 body := []byte(sseErr.RawData) + semanticStatus := http.StatusForbidden + if c.Writer.Size() == writerSizeBeforeStream && gjson.GetBytes(body, "error.type").String() == "overloaded_error" { + semanticStatus = 529 + syntheticResp := &http.Response{ + StatusCode: semanticStatus, + Header: resp.Header.Clone(), + Body: io.NopCloser(bytes.NewReader(body)), + } + s.handleFailoverSideEffects(ctx, syntheticResp, account, reqModel) + } upstreamMsg := sanitizeUpstreamErrorMessage( strings.TrimSpace(extractUpstreamErrorMessage(body)), @@ -826,7 +836,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A Platform: account.Platform, AccountID: account.ID, AccountName: account.Name, - UpstreamStatusCode: 403, + UpstreamStatusCode: semanticStatus, UpstreamRequestID: resp.Header.Get("x-request-id"), Kind: "stream_error", Message: upstreamMsg, @@ -840,7 +850,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A ) return nil, &UpstreamFailoverError{ - StatusCode: 403, + StatusCode: semanticStatus, ResponseBody: body, } } diff --git a/backend/internal/service/gateway_forward_partial_usage_test.go b/backend/internal/service/gateway_forward_partial_usage_test.go index f6701ca8b0..2b7666378e 100644 --- a/backend/internal/service/gateway_forward_partial_usage_test.go +++ b/backend/internal/service/gateway_forward_partial_usage_test.go @@ -8,12 +8,37 @@ import ( "net/http/httptest" "strings" "testing" + "time" "github.com/Wei-Shaw/sub2api/internal/config" "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" ) +type gatewayForwardErrorPolicyRepoStub struct { + AccountRepository + tempCalls int + modelRateLimitCalls []gatewayForwardModelRateLimitCall +} + +type gatewayForwardModelRateLimitCall struct { + accountID int64 + scope string +} + +func (r *gatewayForwardErrorPolicyRepoStub) SetTempUnschedulable(context.Context, int64, time.Time, string) error { + r.tempCalls++ + return nil +} + +func (r *gatewayForwardErrorPolicyRepoStub) SetModelRateLimit(_ context.Context, id int64, scope string, _ time.Time, _ ...string) error { + r.modelRateLimitCalls = append(r.modelRateLimitCalls, gatewayForwardModelRateLimitCall{ + accountID: id, + scope: scope, + }) + return nil +} + // 本文件覆盖 issue #5148：流式转发中途出错（缺失 terminal 事件、读错误等）时， // 已观测到的上游 usage 不得随错误一起被丢弃，Forward 必须把部分结果与错误一同 // 返回，供 handler 照常提交 usage 记录。 @@ -183,6 +208,95 @@ func TestGatewayService_Forward_FailoverErrorKeepsNilResult(t *testing.T) { require.Nil(t, result, "failover 错误必须保持 result=nil，防止重试成功后双重计费") } +func TestGatewayService_Forward_PreOutputSSEOverloadedErrorUsesSemantic529(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + + body := []byte(`{"model":"claude-3-5-sonnet-latest","stream":true,"messages":[{"role":"user","content":"hello"}]}`) + parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), PlatformAnthropic) + require.NoError(t, err) + + const errorJSON = `{"type":"error","error":{"details":null,"type":"overloaded_error","message":"Overloaded"},"request_id":"req_01"}` + upstream := &anthropicHTTPUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader("event: error\ndata: " + errorJSON + "\n\n")), + }} + repo := &gatewayForwardErrorPolicyRepoStub{} + cfg := &config.Config{Gateway: config.GatewayConfig{MaxLineSize: defaultMaxLineSize}} + svc := &GatewayService{ + cfg: cfg, + responseHeaderFilter: compileResponseHeaderFilter(cfg), + httpUpstream: upstream, + rateLimitService: NewRateLimitService(repo, nil, cfg, nil, nil), + deferredService: &DeferredService{}, + } + account := newAnthropicOAuthAccountForPartialUsageTest() + account.Credentials["temp_unschedulable_enabled"] = true + account.Credentials["temp_unschedulable_rules"] = []any{map[string]any{ + "error_code": float64(529), + "keywords": []any{"Overloaded"}, + "duration_minutes": float64(10), + }} + + result, err := svc.Forward(context.Background(), c, account, parsed) + require.Error(t, err) + require.Nil(t, result) + + var failoverErr *UpstreamFailoverError + require.ErrorAs(t, err, &failoverErr) + require.Equal(t, 529, failoverErr.StatusCode) + require.JSONEq(t, errorJSON, string(failoverErr.ResponseBody)) + require.Len(t, repo.modelRateLimitCalls, 1, "synthetic 529 must participate in temp-unschedulable rules") + require.Equal(t, account.ID, repo.modelRateLimitCalls[0].accountID) + require.Equal(t, parsed.Model, repo.modelRateLimitCalls[0].scope) + require.Empty(t, rec.Body.String(), "pre-output overload must remain eligible for account failover") +} + +func TestGatewayService_Forward_PostOutputSSEOverloadedErrorKeepsExistingStatus(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + + body := []byte(`{"model":"claude-3-5-sonnet-latest","stream":true,"messages":[{"role":"user","content":"hello"}]}`) + parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), PlatformAnthropic) + require.NoError(t, err) + + const errorJSON = `{"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}` + fixture := "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":1}}}\n\n" + + "event: error\ndata: " + errorJSON + "\n\n" + upstream := &anthropicHTTPUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(fixture)), + }} + repo := &gatewayForwardErrorPolicyRepoStub{} + cfg := &config.Config{Gateway: config.GatewayConfig{MaxLineSize: defaultMaxLineSize}} + svc := &GatewayService{ + cfg: cfg, + responseHeaderFilter: compileResponseHeaderFilter(cfg), + httpUpstream: upstream, + rateLimitService: NewRateLimitService(repo, nil, cfg, nil, nil), + deferredService: &DeferredService{}, + } + + result, err := svc.Forward(context.Background(), c, newAnthropicOAuthAccountForPartialUsageTest(), parsed) + require.Error(t, err) + require.Nil(t, result) + + var failoverErr *UpstreamFailoverError + require.ErrorAs(t, err, &failoverErr) + require.Equal(t, http.StatusForbidden, failoverErr.StatusCode) + require.JSONEq(t, errorJSON, string(failoverErr.ResponseBody)) + require.Zero(t, repo.tempCalls) + require.Contains(t, rec.Body.String(), "message_start") +} + func TestGatewayService_AnthropicAPIKeyPassthrough_ForwardStreamMissingTerminalPreservesPartialUsage(t *testing.T) { gin.SetMode(gin.TestMode)