fix(gateway): preserve SSE event:error body so ops logs reflect real upstream errors

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.
2026-09-24 16:05:44 +08:00 · 2026-06-16 20:25:09 +08:00
parent b8a482e127
commit 6c7203d83b
2 changed files with 191 additions and 3 deletions
@@ -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 == "" {
@@ -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)))
+}