From 6c8863169087ff047fc2d8e13eb71e49e1a6ee1a Mon Sep 17 00:00:00 2001 From: erio Date: Tue, 9 Jun 2026 14:52:32 +0800 Subject: [PATCH] fix(gateway): prevent double-write on error passthrough responses Service layer writes a complete JSON error response then returns error. Handler's ensureForwardErrorResponse couldn't distinguish this from "no response written" and appended an SSE event, corrupting the body. Use gin.Context flag: service marks MarkResponseCommitted(c) after writing, ensureForwardErrorResponse checks IsResponseCommitted(c) and skips. Zero function signature changes, zero error wrapping. --- backend/internal/handler/gateway_handler.go | 3 + .../handler/openai_gateway_handler.go | 8 +- .../service/antigravity_gateway_service.go | 1 + .../service/error_passthrough_runtime_test.go | 83 +++++++++++++++++++ backend/internal/service/gateway_service.go | 3 + .../service/gemini_messages_compat_service.go | 1 + .../service/openai_gateway_service.go | 5 ++ .../internal/service/ops_upstream_context.go | 15 ++++ 8 files changed, 114 insertions(+), 5 deletions(-) diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index 8853bffbe6..edae3526ee 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -1605,6 +1605,9 @@ func (h *GatewayHandler) ensureForwardErrorResponse(c *gin.Context, streamStarte if c == nil || c.Writer == nil { return false } + if service.IsResponseCommitted(c) { + return false + } if c.Writer.Written() { streamStarted = true } diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 6730529743..80e00b7fc4 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -1847,11 +1847,9 @@ func (h *OpenAIGatewayHandler) ensureForwardErrorResponse(c *gin.Context, stream if c == nil || c.Writer == nil { return false } - // 旧实现在 Writer.Written 时直接 return false，导致 ping 已 flush 之后的 - // 上游错误（http2 timeout、连接中断等）完全无法把错误传给客户端—— - // HTTP 200 已锁死，TCP 直接 EOF，Codex CLI 报 "stream closed before response.completed"。 - // 这里改成：Writer 已写过时强制走 streamStarted 分支，让 - // handleStreamingAwareError 通过 SSE 发协议合规的 response.failed。 + if service.IsResponseCommitted(c) { + return false + } if c.Writer.Written() { streamStarted = true } diff --git a/backend/internal/service/antigravity_gateway_service.go b/backend/internal/service/antigravity_gateway_service.go index f79d20a2ca..b93eb9685a 100644 --- a/backend/internal/service/antigravity_gateway_service.go +++ b/backend/internal/service/antigravity_gateway_service.go @@ -3657,6 +3657,7 @@ func (s *AntigravityGatewayService) WriteMappedClaudeError(c *gin.Context, accou } func (s *AntigravityGatewayService) writeMappedClaudeError(c *gin.Context, account *Account, upstreamStatus int, upstreamRequestID string, body []byte) error { + MarkResponseCommitted(c) upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body)) upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) logBody, maxBytes := s.getLogConfig() diff --git a/backend/internal/service/error_passthrough_runtime_test.go b/backend/internal/service/error_passthrough_runtime_test.go index 7032d15b95..b9e882cc90 100644 --- a/backend/internal/service/error_passthrough_runtime_test.go +++ b/backend/internal/service/error_passthrough_runtime_test.go @@ -251,6 +251,89 @@ func TestApplyErrorPassthroughRule_NoSkipMonitoringDoesNotSetContextKey(t *testi assert.False(t, exists, "OpsSkipPassthroughKey should NOT be set when skip_monitoring=false") } +// ---- ResponseCommittedKey: service 层写完错误响应后标记，handler 层检查跳过兜底写入 ---- + +func TestHandleErrorResponse_SetsResponseCommitted(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + + svc := &GatewayService{} + resp := &http.Response{ + StatusCode: http.StatusBadRequest, + Body: io.NopCloser(bytes.NewReader([]byte(`{"error":{"message":"temperature: range: 0..1"}}`))), + Header: http.Header{}, + } + account := &Account{ID: 100, Platform: PlatformAnthropic, Type: AccountTypeAPIKey} + + _, err := svc.handleErrorResponse(context.Background(), resp, c, account) + require.Error(t, err) + assert.True(t, IsResponseCommitted(c), "non-failover error path must mark response committed") + var payload map[string]any + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &payload)) +} + +func TestHandleErrorResponse_PassthroughRuleSetsCommitted(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + + ruleSvc := &ErrorPassthroughService{} + ruleSvc.setLocalCache([]*model.ErrorPassthroughRule{ + newNonFailoverPassthroughRule(http.StatusBadRequest, "temperature", http.StatusBadRequest, "参数错误"), + }) + BindErrorPassthroughService(c, ruleSvc) + + svc := &GatewayService{} + resp := &http.Response{ + StatusCode: http.StatusBadRequest, + Body: io.NopCloser(bytes.NewReader([]byte(`{"error":{"message":"temperature: range: 0..1"}}`))), + Header: http.Header{}, + } + account := &Account{ID: 200, Platform: PlatformAnthropic, Type: AccountTypeAPIKey} + + _, err := svc.handleErrorResponse(context.Background(), resp, c, account) + require.Error(t, err) + assert.True(t, IsResponseCommitted(c), "passthrough rule path must mark response committed") + assert.Equal(t, http.StatusBadRequest, rec.Code) + var payload map[string]any + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &payload)) + errField := payload["error"].(map[string]any) + assert.Equal(t, "参数错误", errField["message"]) +} + +func TestOpenAIHandleErrorResponse_SetsResponseCommitted(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + + svc := &OpenAIGatewayService{} + resp := &http.Response{ + StatusCode: http.StatusTooManyRequests, + Body: io.NopCloser(bytes.NewReader([]byte(`{"error":{"message":"rate limit exceeded"}}`))), + Header: http.Header{}, + } + account := &Account{ID: 101, Platform: PlatformOpenAI, Type: AccountTypeAPIKey} + + _, err := svc.handleErrorResponse(context.Background(), resp, c, account, nil) + require.Error(t, err) + assert.True(t, IsResponseCommitted(c), "OpenAI non-failover path must mark response committed") +} + +func TestGeminiWriteGeminiMappedError_SetsResponseCommitted(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + + svc := &GeminiMessagesCompatService{} + body := []byte(`{"error":{"message":"invalid field"}}`) + account := &Account{ID: 102, Platform: PlatformGemini, Type: AccountTypeAPIKey} + + err := svc.writeGeminiMappedError(c, account, http.StatusBadRequest, "req-99", body) + require.Error(t, err) + assert.True(t, IsResponseCommitted(c), "Gemini path must mark response committed") +} + func newNonFailoverPassthroughRule(statusCode int, keyword string, respCode int, customMessage string) *model.ErrorPassthroughRule { return &model.ErrorPassthroughRule{ ID: 1, diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 531c93e4e7..20a14b53f4 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -7353,6 +7353,8 @@ func (s *GatewayService) handleErrorResponse(ctx context.Context, resp *http.Res return nil, &UpstreamFailoverError{StatusCode: resp.StatusCode, ResponseBody: body} } + MarkResponseCommitted(c) + // 记录上游错误响应体摘要便于排障（可选：由配置控制；不回显到客户端） if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { logger.LegacyPrintf("service.gateway", @@ -7476,6 +7478,7 @@ func (s *GatewayService) handleFailoverSideEffects(ctx context.Context, resp *ht // OAuth 403：标记账号异常 // API Key 未配置错误码：仅返回错误，不标记账号 func (s *GatewayService) handleRetryExhaustedError(ctx context.Context, resp *http.Response, c *gin.Context, account *Account) (*ForwardResult, error) { + MarkResponseCommitted(c) // Capture upstream error body before side-effects consume the stream. respBody, _ := s.readUpstreamErrorBody(resp) _ = resp.Body.Close() diff --git a/backend/internal/service/gemini_messages_compat_service.go b/backend/internal/service/gemini_messages_compat_service.go index 86073d9ca5..0d88140e24 100644 --- a/backend/internal/service/gemini_messages_compat_service.go +++ b/backend/internal/service/gemini_messages_compat_service.go @@ -1700,6 +1700,7 @@ func sanitizeUpstreamErrorMessage(msg string) string { } func (s *GeminiMessagesCompatService) writeGeminiMappedError(c *gin.Context, account *Account, upstreamStatus int, upstreamRequestID string, body []byte) error { + MarkResponseCommitted(c) upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body)) upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg) upstreamDetail := "" diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index be4dc539ff..0f19d043fa 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -3552,6 +3552,7 @@ func (s *OpenAIGatewayService) handleErrorResponsePassthrough( account *Account, requestBody []byte, ) error { + MarkResponseCommitted(c) body := s.readUpstreamErrorBody(resp) upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body)) @@ -4359,6 +4360,8 @@ func (s *OpenAIGatewayService) handleErrorResponse( } } + MarkResponseCommitted(c) + // Return appropriate error response var errType, errMsg string var statusCode int @@ -4498,6 +4501,8 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse( } } + MarkResponseCommitted(c) + // Map status code to error type and write response errType := "api_error" switch { diff --git a/backend/internal/service/ops_upstream_context.go b/backend/internal/service/ops_upstream_context.go index 2405f306fe..4a1e36aa6f 100644 --- a/backend/internal/service/ops_upstream_context.go +++ b/backend/internal/service/ops_upstream_context.go @@ -34,6 +34,10 @@ const ( // Client-side configuration denials should remain visible in ops_error_logs, // but should be excluded from SLA/error-rate calculations. + // ResponseCommittedKey 由 handleErrorResponse 系列函数在写完 HTTP 错误响应后设置。 + // ensureForwardErrorResponse 检查此 key，为 true 时跳过兜底写入，避免在已完成的 JSON 后追加 SSE。 + ResponseCommittedKey = "response_committed" + OpsClientBusinessLimitedKey = "ops_client_business_limited" OpsClientBusinessLimitedReasonKey = "ops_client_business_limited_reason" OpsClientBusinessLimitedReasonIPRestriction = "api_key_ip_restriction" @@ -43,6 +47,17 @@ const ( OpsClientBusinessLimitedReasonLocalPolicyDenied = "local_policy_denied" ) +func MarkResponseCommitted(c *gin.Context) { c.Set(ResponseCommittedKey, true) } + +func IsResponseCommitted(c *gin.Context) bool { + v, ok := c.Get(ResponseCommittedKey) + if !ok { + return false + } + b, _ := v.(bool) + return b +} + func SetOpsLatencyMs(c *gin.Context, key string, value int64) { if c == nil || strings.TrimSpace(key) == "" || value < 0 { return