From eeff9cd5eea578136a81758bf8f37ccd32251342 Mon Sep 17 00:00:00 2001 From: liyunshan Date: Wed, 29 Jul 2026 11:37:20 +0800 Subject: [PATCH] fix(stream): detect upstream errors in SSE stream before first data chunk When a streaming (SSE) request gets an HTTP 200 but the upstream embeds a terminal error (e.g. rate_limit_exceeded) inside the SSE stream before any content, the relay would silently forward the error to the client, return nil error to the retry loop, and never fall back to another channel. This caused the scenarios where streaming requests hit a rate-limited channel but never retried to a higher-priority channel. Changes: - Add isOpenAITextStreamErrorChunk() to detect inline SSE errors by: 1. Standard OpenAI error envelope ({"error": {"message": ...}}) 2. Known failure codes (rate_limit_exceeded, server_error, etc.) 3. Explicit error/upstream_error type field - In OaiStreamHandler, when such an error is detected before any data has been sent to the client (SendResponseCount == 0), stop the stream immediately via sr.Stop(). - After StreamScannerHandler returns, if the stream ended abnormally and nothing was sent to the client, return a non-nil error so the retry loop in controller/relay.go can try another channel. The image stream handler (relay_image.go) already had equivalent error detection via isOpenAIImageStreamErrorEvent(); this brings text stream handling to parity. Co-Authored-By: Claude Opus 4.8 --- relay/channel/openai/helper.go | 74 ++++++++++++++++++++++++++++ relay/channel/openai/relay-openai.go | 22 +++++++++ 2 files changed, 96 insertions(+) diff --git a/relay/channel/openai/helper.go b/relay/channel/openai/helper.go index 666235ff5633..dc4675cdef64 100644 --- a/relay/channel/openai/helper.go +++ b/relay/channel/openai/helper.go @@ -1,6 +1,7 @@ package openai import ( + "encoding/json" "fmt" "strings" @@ -242,3 +243,76 @@ func sendResponsesStreamData(c *gin.Context, streamResponse dto.ResponsesStreamR } _ = helper.ResponseChunkData(c, streamResponse, data) } + +// isOpenAITextStreamErrorChunk detects whether an SSE data payload represents a +// terminal error from the upstream (e.g. rate_limit_exceeded embedded in a +// streaming response with HTTP 200). Such chunks are not valid message content +// and, when detected before any data has been sent to the client, should abort +// the stream so the retry loop can try another channel. +// +// Detection logic (in order): +// 1. A top-level "error" field containing a JSON object with a non-empty message. +// 2. A "code" field whose value matches a known upstream failure code +// (rate_limit_exceeded, server_error, etc.). +// 3. A "type" field of "error" or "upstream_error". +func isOpenAITextStreamErrorChunk(data string) (bool, string) { + if data == "" { + return false, "" + } + + var payload struct { + Error json.RawMessage `json:"error"` + Code string `json:"code"` + Type string `json:"type"` + Message string `json:"message"` + } + if err := common.UnmarshalJsonStr(data, &payload); err != nil { + return false, "" + } + + // 1. Standard OpenAI error envelope: {"error": {"message": "...", ...}} + if len(payload.Error) > 0 { + var oaiErr types.OpenAIError + if err := common.Unmarshal(payload.Error, &oaiErr); err == nil && oaiErr.Message != "" { + return true, oaiErr.Message + } + } + + // 2. Known failure codes embedded in SSE data chunks + if isKnownUpstreamErrorCode(payload.Code) { + msg := payload.Message + if msg == "" { + msg = fmt.Sprintf("upstream error code: %s", payload.Code) + } + return true, msg + } + + // 3. Explicit error/upstream_error type + payloadType := strings.ToLower(strings.TrimSpace(payload.Type)) + if payloadType == "error" || payloadType == "upstream_error" { + msg := payload.Message + if msg == "" { + msg = "upstream stream returned error event" + } + return true, msg + } + + return false, "" +} + +// knownUpstreamErrorCodes lists SSE payload "code" values that represent a +// terminal upstream failure (not deliverable content). +var knownUpstreamErrorCodes = map[string]bool{ + "rate_limit_exceeded": true, + "rate_limit_reached": true, + "server_error": true, + "internal_error": true, + "upstream_error": true, +} + +func isKnownUpstreamErrorCode(code string) bool { + if code == "" { + return false + } + return knownUpstreamErrorCodes[code] +} diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index 9a0619eb27f5..8ceb3c96497b 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -144,9 +144,31 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re logger.LogError(c, "error processing stream token data: "+err.Error()) sr.Error(err) } + + // Detect upstream business errors embedded in the SSE stream (e.g. + // rate_limit_exceeded). When detected before any data has been + // sent to the client (SendResponseCount == 0), abort the stream so + // the caller can retry with a different channel. + if info.SendResponseCount == 0 { + if isErr, errMsg := isOpenAITextStreamErrorChunk(data); isErr { + logger.LogError(c, "upstream stream returned error before sending data: "+errMsg) + sr.Stop(fmt.Errorf("upstream stream error: %s", errMsg)) + return + } + } } }) + // Return an error when the stream failed before any data was sent to the + // client so the retry loop can fall back to another channel. + if info.StreamStatus != nil && !info.StreamStatus.IsNormalEnd() && info.SendResponseCount == 0 { + return usage, types.NewOpenAIError( + fmt.Errorf("upstream stream failed before sending data: %s", info.StreamStatus.Summary()), + types.ErrorCodeBadResponseStatusCode, + http.StatusInternalServerError, + ) + } + // 对音频模型，从倒数第二个stream data中提取usage信息 if isAudioModel && secondLastStreamData != "" { var streamResp struct {