From 6c7203d83b5c2efb7be89f03d6ceebac9f7c986d Mon Sep 17 00:00:00 2001 From: shaw Date: Tue, 16 Jun 2026 20:25:09 +0800 Subject: [PATCH] fix(gateway): preserve SSE event:error body so ops logs reflect real upstream errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When an Anthropic upstream returned HTTP 200 but then emitted an SSE `event: error` frame (overloaded_error / rate_limit_error / api_error / etc.), Forward's stream branch matched on `err.Error() == "have error in stream"` and returned `UpstreamFailoverError{StatusCode: 403}` with no ResponseBody. That dropped three pieces of evidence: - handleFailoverExhausted → ExtractUpstreamErrorMessage(nil) = "" → ops_error_logs.upstream_error_message was empty. - errorPassthroughService.MatchRule(_, 403, nil) could only match rules without keywords, so keyword-based passthrough rules silently never fired. - upstream_errors carried no stream_error record, leaving ops looking at a generic 403 with no clue whether the upstream was throttled, overloaded, or rejecting the request. ping-during-slot-wait amplified this by skipping failover (writerSizeBeforeForward guard), so 403s ballooned in the ops view well past the upstream's actual rate. Fix: - Introduce *sseStreamErrorEventError that carries the SSE data line. Error() still returns "have error in stream" so existing log searches keep working. - Forward extracts via errors.As, appends an OpsUpstreamErrorEvent (kind="stream_error", with the sanitized message and a truncated raw body honoring LogUpstreamErrorBody*), and returns UpstreamFailoverError{StatusCode: 403, ResponseBody: rawJSON}. StatusCode 403 is preserved verbatim: mapUpstreamError, failover decisions (shouldFailoverUpstreamError(403)=true), client-visible message, RetryableOnSameAccount, and rateLimitService side-effects (this path already didn't invoke them) all match prior behavior. OAuth and API Key accounts share this path; the API-Key passthrough branch is independent and already forwards SSE error frames untouched, so it's unaffected. Adds four unit tests: typed-error contract + RawData, empty data line, event:error after partial stream output (streamStarted=true), and non-JSON data line. --- backend/internal/service/gateway_service.go | 54 ++++++- .../service/gateway_streaming_test.go | 140 ++++++++++++++++++ 2 files changed, 191 insertions(+), 3 deletions(-) diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 5e333ae7e4..f7f29fef3b 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -586,6 +586,17 @@ func (e *UpstreamFailoverError) Error() string { return fmt.Sprintf("upstream error: %d (failover)", e.StatusCode) } +// sseStreamErrorEventError 表示上游 SSE 流体内出现 event:error 帧。 +// RawData 是该事件 data: 行的原始 JSON 字符串 +// （Anthropic 标准结构 {"type":"error","error":{"type":"...","message":"..."}}）。 +// Error() 保持原字符串以兼容现有日志/检索；调用方应通过 errors.As +// 提取 RawData 并构造 UpstreamFailoverError.ResponseBody。 +type sseStreamErrorEventError struct { + RawData string +} + +func (e *sseStreamErrorEventError) Error() string { return "have error in stream" } + // TempUnscheduleRetryableError 对 RetryableOnSameAccount 类型的 failover 错误触发临时封禁。 // 由 handler 层在同账号重试全部用尽、切换账号时调用。 func (s *GatewayService) TempUnscheduleRetryableError(ctx context.Context, accountID int64, failoverErr *UpstreamFailoverError) { @@ -5396,9 +5407,46 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A if reqStream { streamResult, err := s.handleStreamingResponse(ctx, resp, c, account, startTime, originalModel, reqModel, shouldMimicClaudeCode) if err != nil { - if err.Error() == "have error in stream" { + var sseErr *sseStreamErrorEventError + if errors.As(err, &sseErr) { + // 上游 HTTP 200 + SSE 流体内出现 event:error 帧。 + // 保留 StatusCode=403 以兼容既有 failover/客户端响应语义， + // 但补全 ResponseBody 与 ops 上下文，让运维日志能反映上游真实错误。 + body := []byte(sseErr.RawData) + + upstreamMsg := sanitizeUpstreamErrorMessage( + strings.TrimSpace(extractUpstreamErrorMessage(body)), + ) + + upstreamDetail := "" + if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody { + maxBytes := s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes + if maxBytes <= 0 { + maxBytes = 2048 + } + upstreamDetail = truncateString(sseErr.RawData, maxBytes) + } + + appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ + Platform: account.Platform, + AccountID: account.ID, + AccountName: account.Name, + UpstreamStatusCode: 403, + UpstreamRequestID: resp.Header.Get("x-request-id"), + Kind: "stream_error", + Message: upstreamMsg, + Detail: upstreamDetail, + }) + + logger.LegacyPrintf("service.gateway", + "[Forward] SSE error event in stream: Account=%d(%s) RequestID=%s Body=%s", + account.ID, account.Name, resp.Header.Get("x-request-id"), + truncateString(sseErr.RawData, 1000), + ) + return nil, &UpstreamFailoverError{ - StatusCode: 403, + StatusCode: 403, + ResponseBody: body, } } return nil, err @@ -8061,7 +8109,7 @@ func (s *GatewayService) handleStreamingResponse(ctx context.Context, resp *http } if eventName == "error" { - return nil, dataLine, nil, errors.New("have error in stream") + return nil, dataLine, nil, &sseStreamErrorEventError{RawData: dataLine} } if dataLine == "" { diff --git a/backend/internal/service/gateway_streaming_test.go b/backend/internal/service/gateway_streaming_test.go index ef09a882e8..df8e8e4f80 100644 --- a/backend/internal/service/gateway_streaming_test.go +++ b/backend/internal/service/gateway_streaming_test.go @@ -393,3 +393,143 @@ func TestHandleStreamingResponse_FailoverBodyDoesNotLeakAddresses(t *testing.T) require.Contains(t, body, "connection reset by peer") require.Contains(t, body, "upstream stream disconnected") } + +// 上游 HTTP 200 + SSE 流体内 event:error 帧应被识别为 *sseStreamErrorEventError， +// 且 RawData 等于上游 data: 行的原始 JSON。这是 Forward 主流程后续把 dataLine +// 透传到 UpstreamFailoverError.ResponseBody 与 ops_error_logs 的前提。 +func TestHandleStreamingResponse_SSEErrorEvent_ReturnsTypedErrorWithRawData(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newMinimalGatewayService() + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + + const errorJSON = `{"type":"error","error":{"type":"overloaded_error","message":"Anthropic upstream is overloaded"}}` + + pr, pw := io.Pipe() + resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: pr} + + go func() { + defer func() { _ = pw.Close() }() + _, _ = pw.Write([]byte("event: error\ndata: " + errorJSON + "\n\n")) + }() + + result, err := svc.handleStreamingResponse(context.Background(), resp, c, &Account{ID: 1}, time.Now(), "model", "model", false) + _ = pr.Close() + + require.Error(t, err) + require.Nil(t, result) + + // typed error 必须可被 errors.As 匹配，RawData 必须保留上游 dataLine 原文 + var sseErr *sseStreamErrorEventError + require.True(t, errors.As(err, &sseErr), "SSE event:error 必须包成 *sseStreamErrorEventError，期望: %v", err) + require.Equal(t, errorJSON, sseErr.RawData) + + // 字符串兼容：保留与旧实现一致的 "have error in stream"，避免破坏依赖该字符串的日志检索 + require.Equal(t, "have error in stream", err.Error()) + + // 在 Forward 主流程中调用方依赖 ExtractUpstreamErrorMessage 从 RawData 解析出 message + extracted := ExtractUpstreamErrorMessage([]byte(sseErr.RawData)) + require.Equal(t, "Anthropic upstream is overloaded", extracted) +} + +// 边界用例：上游只发了 event: error 而没有 data 行。RawData 为空， +// 调用方不得 panic，UpstreamFailoverError.ResponseBody 应回退为空切片。 +func TestHandleStreamingResponse_SSEErrorEvent_EmptyDataLine(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newMinimalGatewayService() + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + + pr, pw := io.Pipe() + resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: pr} + + go func() { + defer func() { _ = pw.Close() }() + _, _ = pw.Write([]byte("event: error\n\n")) + }() + + _, err := svc.handleStreamingResponse(context.Background(), resp, c, &Account{ID: 1}, time.Now(), "model", "model", false) + _ = pr.Close() + + require.Error(t, err) + var sseErr *sseStreamErrorEventError + require.True(t, errors.As(err, &sseErr), "即使 data 行为空，也必须返回 typed error 让上层走 stream_error 分支") + require.Equal(t, "", sseErr.RawData) +} + +// 对抗用例：上游先发 message_start 再发 event:error，模拟"流已开始写客户端"+SSE error 帧。 +// 这是 ping 放大场景的服务侧近似（c.Writer 已被写后才出现 error）。 +// 必须仍然返回 *sseStreamErrorEventError 且 RawData 包含真实错误体， +// 让 Forward 调用方能正确补全 ResponseBody 与 ops 事件。 +func TestHandleStreamingResponse_SSEErrorEvent_AfterPartialStreamOutput(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newMinimalGatewayService() + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + + const errorJSON = `{"type":"error","error":{"type":"rate_limit_error","message":"Rate limited"}}` + + pr, pw := io.Pipe() + resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: pr} + + go func() { + defer func() { _ = pw.Close() }() + // 先发 message_start，让 handleStreamingResponse 把它转发到客户端 → c.Writer 已被写 + _, _ = pw.Write([]byte(`data: {"type":"message_start","message":{"usage":{"input_tokens":5}}}` + "\n\n")) + // 紧接着发 event:error + _, _ = pw.Write([]byte("event: error\ndata: " + errorJSON + "\n\n")) + }() + + _, err := svc.handleStreamingResponse(context.Background(), resp, c, &Account{ID: 1}, time.Now(), "model", "model", false) + _ = pr.Close() + + require.Error(t, err) + var sseErr *sseStreamErrorEventError + require.True(t, errors.As(err, &sseErr), "已发数据后再来的 SSE event:error 必须仍包成 typed error，期望: %v", err) + require.Equal(t, errorJSON, sseErr.RawData) + + // c.Writer 必定已被写过（message_start 已转发）— 这是 handler 838 行 streamStarted 守卫触发的条件， + // 修复前/后均会让 handler 直接走 handleFailoverExhausted 而非切账号；不变。 + require.Greater(t, rec.Body.Len(), 0, "message_start 应被转发到客户端") + require.Contains(t, rec.Body.String(), "message_start") +} + +// 对抗用例：上游发 event:error 但 data 行不是合法 JSON。 +// RawData 必须保留原始字节，ExtractUpstreamErrorMessage 不得 panic， +// upstreamMsg 回退为空字符串（不再丢失原始诊断线索 — Detail 字段仍保留原始 body）。 +func TestHandleStreamingResponse_SSEErrorEvent_NonJSONDataLine(t *testing.T) { + gin.SetMode(gin.TestMode) + svc := newMinimalGatewayService() + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + + pr, pw := io.Pipe() + resp := &http.Response{StatusCode: http.StatusOK, Header: http.Header{}, Body: pr} + + go func() { + defer func() { _ = pw.Close() }() + _, _ = pw.Write([]byte("event: error\ndata: not-a-json-payload\n\n")) + }() + + _, err := svc.handleStreamingResponse(context.Background(), resp, c, &Account{ID: 1}, time.Now(), "model", "model", false) + _ = pr.Close() + + require.Error(t, err) + var sseErr *sseStreamErrorEventError + require.True(t, errors.As(err, &sseErr)) + require.Equal(t, "not-a-json-payload", sseErr.RawData) + + // gjson 对非 JSON 输入返回空字符串，不 panic — Forward 主流程靠这个 invariant 安全地走下去 + require.NotPanics(t, func() { + _ = ExtractUpstreamErrorMessage([]byte(sseErr.RawData)) + }) + require.Equal(t, "", ExtractUpstreamErrorMessage([]byte(sseErr.RawData))) +}