From bd0f2d64058dbfab55d34bb21819f13c15c66f22 Mon Sep 17 00:00:00 2001 From: haruka <1628615876@qq.com> Date: Sat, 18 Jul 2026 03:24:04 +0800 Subject: [PATCH] fix: report chat stream transport failures --- .../handler/openai_chat_completions.go | 5 +- .../handler/openai_gateway_handler.go | 50 +++++++++++++++- .../handler/openai_gateway_handler_test.go | 29 +++++++++ backend/internal/handler/ops_error_logger.go | 20 +++++-- .../internal/handler/ops_error_logger_test.go | 30 ++++++++++ .../openai_gateway_chat_completions.go | 14 +++-- .../openai_gateway_chat_completions_test.go | 57 ++++++++++++++++++ .../service/openai_stream_read_error.go | 59 +++++++++++++++++++ .../internal/service/ops_upstream_context.go | 38 ++++++++++-- 9 files changed, 284 insertions(+), 18 deletions(-) create mode 100644 backend/internal/service/openai_stream_read_error.go diff --git a/backend/internal/handler/openai_chat_completions.go b/backend/internal/handler/openai_chat_completions.go index 974d35ded6..e16962ba5d 100644 --- a/backend/internal/handler/openai_chat_completions.go +++ b/backend/internal/handler/openai_chat_completions.go @@ -305,7 +305,10 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) { upstreamErrorAlreadyCommunicated := openAIForwardErrorAlreadyCommunicated(c, writerSizeBeforeForward, err) wroteFallback := false if !upstreamErrorAlreadyCommunicated { - wroteFallback = h.ensureForwardErrorResponse(c, streamStarted) + wroteFallback = h.ensureOpenAIStreamReadErrorResponse(c, err, streamStarted) + if !wroteFallback { + wroteFallback = h.ensureForwardErrorResponse(c, streamStarted) + } } reqLog.Warn("openai_chat_completions.forward_failed", zap.Int64("account_id", account.ID), diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index ec4e06dd8b..fe71b0336d 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -2231,12 +2231,29 @@ func (h *OpenAIGatewayHandler) mapUpstreamError(statusCode int) (int, string, st // handleStreamingAwareError handles errors that may occur after streaming has started func (h *OpenAIGatewayHandler) handleStreamingAwareError(c *gin.Context, status int, errType, message string, streamStarted bool) { + h.handleStreamingAwareErrorWithCode(c, status, errType, "", message, streamStarted, false) +} + +func (h *OpenAIGatewayHandler) handleStreamingAwareErrorWithCode( + c *gin.Context, + status int, + errType string, + code string, + message string, + streamStarted bool, + countTowardsSLA bool, +) { // body-signal compact 心跳可能已把响应头提交为 200：先停心跳（建立 // happens-before，接管 ResponseWriter），并升级为流内错误处理。 if service.StopOpenAICompactSSEKeepaliveCommitted(c) { streamStarted = true } if streamStarted { + if countTowardsSLA { + service.MarkOpsStreamFailure(c, errType, code, message, status) + } else { + service.MarkOpsStreamError(c, errType, message, status) + } // /v1/responses 的严格 SDK（Codex CLI）要求终止事件必须属于 // response.completed/failed/incomplete/cancelled 集合。 // 通用 `event: error` 帧不被识别为终止事件，会导致 @@ -2249,8 +2266,15 @@ func (h *OpenAIGatewayHandler) handleStreamingAwareError(c *gin.Context, status // Stream already started, send error as SSE event then close flusher, ok := c.Writer.(http.Flusher) if ok { - // SSE 错误事件固定 schema，使用 Quote 直拼可避免额外 Marshal 分配。 - errorEvent := "event: error\ndata: " + `{"error":{"type":` + strconv.Quote(errType) + `,"message":` + strconv.Quote(message) + `}}` + "\n\n" + errorObject := gin.H{"type": errType, "message": message} + if code != "" { + errorObject["code"] = code + } + payload, err := json.Marshal(gin.H{"error": errorObject}) + if err != nil { + payload = []byte(`{"error":{"type":"upstream_error","message":"Upstream request failed"}}`) + } + errorEvent := "event: error\ndata: " + string(payload) + "\n\n" if _, err := fmt.Fprint(c.Writer, errorEvent); err != nil { _ = c.Error(err) } @@ -2260,7 +2284,27 @@ func (h *OpenAIGatewayHandler) handleStreamingAwareError(c *gin.Context, status } // Normal case: return JSON response with proper status code - h.errorResponse(c, status, errType, message) + if code == "" { + h.errorResponse(c, status, errType, message) + return + } + c.JSON(status, gin.H{"error": gin.H{ + "type": errType, "code": code, "message": message, + }}) +} + +func (h *OpenAIGatewayHandler) ensureOpenAIStreamReadErrorResponse(c *gin.Context, err error, streamStarted bool) bool { + code, message, ok := service.OpenAIUpstreamStreamReadErrorDetails(err) + if !ok || c == nil || c.Writer == nil || service.IsResponseCommitted(c) { + return false + } + if c.Writer.Written() { + streamStarted = true + } + h.handleStreamingAwareErrorWithCode( + c, http.StatusBadGateway, "upstream_error", code, message, streamStarted, true, + ) + return true } // ensureForwardErrorResponse 在 Forward 返回错误但尚未写响应时补写统一错误响应。 diff --git a/backend/internal/handler/openai_gateway_handler_test.go b/backend/internal/handler/openai_gateway_handler_test.go index 931f3b3f65..05760c7edc 100644 --- a/backend/internal/handler/openai_gateway_handler_test.go +++ b/backend/internal/handler/openai_gateway_handler_test.go @@ -96,6 +96,35 @@ func TestOpenAIHandleStreamingAwareError_JSONEscaping(t *testing.T) { } } +func TestOpenAIHandleStreamingAwareErrorWithCode_EmitsStableClassification(t *testing.T) { + gin.SetMode(gin.TestMode) + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil) + + h := &OpenAIGatewayHandler{} + h.handleStreamingAwareErrorWithCode( + c, + http.StatusBadGateway, + "upstream_error", + service.OpenAIUpstreamHTTP2StreamErrorCode, + "Upstream HTTP/2 stream failed", + true, + true, + ) + + body := w.Body.String() + require.Contains(t, body, "event: error\n") + require.Equal(t, "upstream_error", gjson.Get(body[strings.Index(body, "{"):], "error.type").String()) + require.Equal(t, service.OpenAIUpstreamHTTP2StreamErrorCode, gjson.Get(body[strings.Index(body, "{"):], "error.code").String()) + require.NotContains(t, body, "stream ID") + + streamErr, ok := service.GetOpsStreamError(c) + require.True(t, ok) + require.True(t, streamErr.CountTowardsSLA) + require.Equal(t, http.StatusBadGateway, streamErr.IntendedStatus) +} + func TestOpenAIForwardSucceededForScheduling(t *testing.T) { require.True(t, openAIForwardSucceededForScheduling(nil)) require.True(t, openAIForwardSucceededForScheduling(&service.OpenAIForwardResult{})) diff --git a/backend/internal/handler/ops_error_logger.go b/backend/internal/handler/ops_error_logger.go index f32ca9c054..99cbf42a9e 100644 --- a/backend/internal/handler/ops_error_logger.go +++ b/backend/internal/handler/ops_error_logger.go @@ -1072,8 +1072,20 @@ func logOpsStreamError(c *gin.Context, ops *service.OpsService, wireStatus int) if classifyStatus <= 0 { classifyStatus = wireStatus } - normalizedType := normalizeOpsErrorType(streamErr.ErrType, "") - phase, isBusinessLimited, errorOwner, errorSource := classifyOpsErrorLog(c, normalizedType, streamErr.Message, "", classifyStatus) + normalizedType := normalizeOpsErrorType(streamErr.ErrType, streamErr.Code) + phase, isBusinessLimited, errorOwner, errorSource := classifyOpsErrorLog(c, normalizedType, streamErr.Message, streamErr.Code, classifyStatus) + recordedStatus := wireStatus + if streamErr.CountTowardsSLA && streamErr.IntendedStatus >= 400 { + recordedStatus = streamErr.IntendedStatus + } + errorBody := "" + if streamErr.Code != "" { + if payload, err := json.Marshal(gin.H{"error": gin.H{ + "type": normalizedType, "code": streamErr.Code, "message": streamErr.Message, + }}); err == nil { + errorBody = string(payload) + } + } apiKey := getOpsAPIKey(c) clientRequestID, _ := c.Request.Context().Value(ctxkey.ClientRequestID).(string) @@ -1140,12 +1152,12 @@ func logOpsStreamError(c *gin.Context, ops *service.OpsService, wireStatus int) ErrorPhase: phase, ErrorType: normalizedType, Severity: classifyOpsSeverity(normalizedType, classifyStatus), - StatusCode: wireStatus, + StatusCode: recordedStatus, IsBusinessLimited: isBusinessLimited, IsCountTokens: isCountTokensRequest(c), ErrorMessage: streamErr.Message, - ErrorBody: "", + ErrorBody: errorBody, ErrorSource: errorSource, ErrorOwner: errorOwner, diff --git a/backend/internal/handler/ops_error_logger_test.go b/backend/internal/handler/ops_error_logger_test.go index 89bdd938c1..893c503d5d 100644 --- a/backend/internal/handler/ops_error_logger_test.go +++ b/backend/internal/handler/ops_error_logger_test.go @@ -182,6 +182,36 @@ func TestLogOpsStreamError_RecordsInBandConcurrencyLimit(t *testing.T) { require.Equal(t, "Concurrency limit exceeded for account, please retry later", job.entry.ErrorMessage) } +func TestLogOpsStreamError_UpstreamFailureCountsTowardsSLA(t *testing.T) { + setupOpsErrorLogTestQueue(t, 4) + + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil) + c.Set(opsModelKey, "gpt-5.6-sol") + + service.MarkOpsStreamFailure( + c, + "upstream_error", + service.OpenAIUpstreamHTTP2StreamErrorCode, + "Upstream HTTP/2 stream failed", + http.StatusBadGateway, + ) + + ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil) + logOpsStreamError(c, ops, http.StatusOK) + + job := <-opsErrorLogQueue + require.NotNil(t, job.entry) + require.Equal(t, http.StatusBadGateway, job.entry.StatusCode) + require.Equal(t, "upstream_error", job.entry.ErrorType) + require.Equal(t, "upstream", job.entry.ErrorPhase) + require.Equal(t, "provider", job.entry.ErrorOwner) + require.False(t, job.entry.IsBusinessLimited) + require.Contains(t, job.entry.ErrorBody, service.OpenAIUpstreamHTTP2StreamErrorCode) +} + // 未标记流内错误时 logOpsStreamError 必须是 no-op（不误记正常的 200 流）。 func TestLogOpsStreamError_NoopWhenNotMarked(t *testing.T) { setupOpsErrorLogTestQueue(t, 4) diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index fe24b3db36..cccec46056 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -779,9 +779,9 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse( handleScanErr := func(err error) { if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { - logger.L().Warn("openai chat_completions stream: read error", + logger.FromContext(c.Request.Context()).Warn("openai chat_completions stream: read error", zap.Error(err), - zap.String("request_id", requestID), + zap.String("upstream_request_id", requestID), ) } } @@ -820,7 +820,10 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse( } if err := scanner.Err(); err != nil { handleScanErr(err) - return resultWithUsage(), fmt.Errorf("stream usage incomplete: %w", err) + if clientDisconnected || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return resultWithUsage(), fmt.Errorf("stream usage incomplete: %w", err) + } + return resultWithUsage(), newOpenAIUpstreamStreamReadError(err) } if frame, ok := parser.Finish(); ok { if strings.TrimSpace(frame.Data) == "[DONE]" { @@ -892,7 +895,10 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse( } if ev.err != nil { handleScanErr(ev.err) - return resultWithUsage(), fmt.Errorf("stream usage incomplete: %w", ev.err) + if clientDisconnected || errors.Is(ev.err, context.Canceled) || errors.Is(ev.err, context.DeadlineExceeded) { + return resultWithUsage(), fmt.Errorf("stream usage incomplete: %w", ev.err) + } + return resultWithUsage(), newOpenAIUpstreamStreamReadError(ev.err) } lastDataAt = time.Now() line := ev.line diff --git a/backend/internal/service/openai_gateway_chat_completions_test.go b/backend/internal/service/openai_gateway_chat_completions_test.go index 5186598a70..b307aa4ede 100644 --- a/backend/internal/service/openai_gateway_chat_completions_test.go +++ b/backend/internal/service/openai_gateway_chat_completions_test.go @@ -32,6 +32,63 @@ func (w *openAIChatFailingWriter) Write(p []byte) (int, error) { return w.ResponseWriter.Write(p) } +type openAIChatStreamReadErrorCloser struct { + payload []byte + err error + sent bool +} + +func (r *openAIChatStreamReadErrorCloser) Read(p []byte) (int, error) { + if !r.sent { + r.sent = true + return copy(p, r.payload), nil + } + return 0, r.err +} + +func (r *openAIChatStreamReadErrorCloser) Close() error { return nil } + +func TestHandleChatStreamingResponse_ClassifiesHTTP2ReadError(t *testing.T) { + gin.SetMode(gin.TestMode) + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil) + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{ + "Content-Type": []string{"text/event-stream"}, + "x-request-id": []string{"upstream-rid"}, + }, + Body: &openAIChatStreamReadErrorCloser{ + payload: []byte("data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n"), + err: errors.New("stream error: stream ID 5; INTERNAL_ERROR; received from peer"), + }, + } + svc := &OpenAIGatewayService{cfg: &config.Config{}} + + result, err := svc.handleChatStreamingResponse( + resp, + c, + &Account{ID: 1, Name: "openai-oauth", Platform: PlatformOpenAI}, + "gpt-5.6-sol", + "gpt-5.6-sol", + "gpt-5.6-sol", + time.Now(), + 0, + ) + + require.Error(t, err) + require.NotNil(t, result) + require.True(t, c.Writer.Written(), "partial output must make replay unsafe") + code, message, ok := OpenAIUpstreamStreamReadErrorDetails(err) + require.True(t, ok) + require.Equal(t, OpenAIUpstreamHTTP2StreamErrorCode, code) + require.Equal(t, "Upstream HTTP/2 stream failed", message) + require.NotContains(t, message, "stream ID") + require.NotContains(t, message, "INTERNAL_ERROR") +} + func TestNormalizeResponsesRequestServiceTier(t *testing.T) { t.Parallel() diff --git a/backend/internal/service/openai_stream_read_error.go b/backend/internal/service/openai_stream_read_error.go new file mode 100644 index 0000000000..b318446e02 --- /dev/null +++ b/backend/internal/service/openai_stream_read_error.go @@ -0,0 +1,59 @@ +package service + +import ( + "errors" + "fmt" + "strings" +) + +const ( + // OpenAIUpstreamHTTP2StreamErrorCode is returned to OpenAI-compatible clients + // when an upstream HTTP/2 response stream is reset after the request started. + OpenAIUpstreamHTTP2StreamErrorCode = "upstream_http2_stream_error" + OpenAIUpstreamStreamReadErrorCode = "upstream_stream_read_error" +) + +type openAIUpstreamStreamReadError struct { + cause error + clientCode string + clientMessage string +} + +func (e *openAIUpstreamStreamReadError) Error() string { + return fmt.Sprintf("stream usage incomplete: %v", e.cause) +} + +func (e *openAIUpstreamStreamReadError) Unwrap() error { return e.cause } + +func newOpenAIUpstreamStreamReadError(err error) error { + code, message := classifyOpenAIUpstreamStreamReadError(err) + return &openAIUpstreamStreamReadError{ + cause: err, + clientCode: code, + clientMessage: message, + } +} + +// OpenAIUpstreamStreamReadErrorDetails returns the stable, sanitized client +// classification attached to an upstream stream read failure. +func OpenAIUpstreamStreamReadErrorDetails(err error) (code, message string, ok bool) { + var streamErr *openAIUpstreamStreamReadError + if !errors.As(err, &streamErr) || streamErr == nil { + return "", "", false + } + return streamErr.clientCode, streamErr.clientMessage, true +} + +func classifyOpenAIUpstreamStreamReadError(err error) (code, message string) { + if err != nil { + lower := strings.ToLower(err.Error()) + // net/http's HTTP/2 stream error is unexported. Its stable text contains + // "stream error: stream ID ..."; match only the transport signature and + // never pass the original text to the client. + if strings.Contains(lower, "stream error: stream id ") || + (strings.Contains(lower, "http2:") && strings.Contains(lower, "stream")) { + return OpenAIUpstreamHTTP2StreamErrorCode, "Upstream HTTP/2 stream failed" + } + } + return OpenAIUpstreamStreamReadErrorCode, "Upstream response stream was interrupted" +} diff --git a/backend/internal/service/ops_upstream_context.go b/backend/internal/service/ops_upstream_context.go index 2d28f1a168..35a97cc2f5 100644 --- a/backend/internal/service/ops_upstream_context.go +++ b/backend/internal/service/ops_upstream_context.go @@ -102,28 +102,54 @@ func HasOpsClientBusinessLimited(c *gin.Context) bool { type OpsStreamError struct { // ErrType 是写入 SSE 帧的对客错误类型（如 rate_limit_error / upstream_error / api_error）。 ErrType string + // Code 是可选的稳定错误分类；用于既保留通用 OpenAI error.type，又向客户端和 Ops + // 暴露可编程判断的细分类（如 upstream_http2_stream_error）。 + Code string // Message 是写入 SSE 帧的对客错误消息。 Message string // IntendedStatus 是流若未固化本应返回的 HTTP 状态码（如并发限流的 429）。 - // 仅用于错误分级(severity/classification)；实际 wire 状态码仍为 200。 + // 默认仅用于错误分级；CountTowardsSLA=true 时也作为 Ops 的逻辑状态码。 IntendedStatus int + // CountTowardsSLA 表示虽然 wire 状态已固化为 200，请求在应用语义上仍然失败， + // Ops 应使用 IntendedStatus 计入错误率/SLA。 + CountTowardsSLA bool } // MarkOpsStreamError 记录一次就地 SSE 错误，供 ops 日志采集。 // 采用「首个标记生效」策略：同一请求若先后补发多帧（如上游透传错误后又追加通用兜底帧）， // 保留最先记录的根因错误，而不是被后续的 "Upstream request failed" 覆盖。 func MarkOpsStreamError(c *gin.Context, errType, message string, intendedStatus int) { + markOpsStreamError(c, OpsStreamError{ + ErrType: errType, + Message: message, + IntendedStatus: intendedStatus, + }) +} + +// MarkOpsStreamFailure records an in-band stream error that represents a failed +// request and therefore must count towards Ops error rate/SLA despite HTTP 200 +// already being committed on the wire. +func MarkOpsStreamFailure(c *gin.Context, errType, code, message string, intendedStatus int) { + markOpsStreamError(c, OpsStreamError{ + ErrType: errType, + Code: code, + Message: message, + IntendedStatus: intendedStatus, + CountTowardsSLA: true, + }) +} + +func markOpsStreamError(c *gin.Context, streamErr OpsStreamError) { if c == nil { return } if _, exists := c.Get(OpsStreamErrorKey); exists { return } - c.Set(OpsStreamErrorKey, OpsStreamError{ - ErrType: strings.TrimSpace(errType), - Message: strings.TrimSpace(message), - IntendedStatus: intendedStatus, - }) + streamErr.ErrType = strings.TrimSpace(streamErr.ErrType) + streamErr.Code = strings.TrimSpace(streamErr.Code) + streamErr.Message = strings.TrimSpace(streamErr.Message) + c.Set(OpsStreamErrorKey, streamErr) } // GetOpsStreamError 返回本请求记录的就地 SSE 错误（若有）。