Fix/stream error detection and channel fallback - #6523
Open
liyunshan0823-glitch wants to merge 2 commits into
Open
Fix/stream error detection and channel fallback#6523liyunshan0823-glitch wants to merge 2 commits into
liyunshan0823-glitch wants to merge 2 commits into
Conversation
…hunk
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 <noreply@anthropic.com>
Contributor
WalkthroughOpenAI streaming now parses SSE payloads for terminal upstream errors and stops the stream before sending a response when such an error is detected. ChangesOpenAI stream error handling
Estimated code review effort: 3 (Moderate) | ~20 minutes Possibly related PRs
Suggested reviewers: Sequence Diagram(s)sequenceDiagram
participant OpenAISSE
participant OaiStreamHandler
participant StreamResponse
OpenAISSE->>OaiStreamHandler: Send streaming SSE chunk
OaiStreamHandler->>OaiStreamHandler: Parse and classify error payload
OaiStreamHandler->>StreamResponse: Stop before response output
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
relay/channel/openai/relay-openai.go (1)
141-159: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRun the terminal-error check before
processTokenData/collectStreamFunctionCallNames.Error chunks (e.g.
rate_limit_exceeded) are currently fed intocollectStreamFunctionCallNamesandprocessTokenDatafirst. Since these aren't valid token/tool-call payloads, this wastes work and produces a spurious"error processing stream token data"log alongside the intended"upstream stream returned error before sending data"log, muddying diagnostics for the exact failure this PR is meant to surface clearly.♻️ Proposed reordering
if len(data) > 0 { + // Detect upstream business errors embedded in the SSE stream (e.g. + // rate_limit_exceeded) before treating the payload as token data. + 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 + } + } + // 对音频模型，保存倒数第二个stream data if isAudioModel && lastStreamData != "" { secondLastStreamData = lastStreamData } lastStreamData = data collectStreamFunctionCallNames(data, seenStreamToolCalls, &streamFunctionCallNames) if err := processTokenData(info.RelayMode, data, &responseTextBuilder, &toolCount); err != nil { 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 - } - } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@relay/channel/openai/relay-openai.go` around lines 141 - 159, Move the terminal-error detection block in the stream loop to run immediately after assigning `lastStreamData`, before `collectStreamFunctionCallNames` and `processTokenData`. When an upstream error chunk is detected before any response data, log it, stop the stream, and return without processing the chunk as token or tool-call data; preserve normal processing for non-error chunks.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@relay/channel/openai/helper.go`:
- Around line 258-301: The isOpenAITextStreamErrorChunk helper currently
discards partially unmarshaled payloads when optional top-level code or type
fields have incompatible JSON types. Change the payload fields used for code,
type, and message parsing to json.RawMessage, allow unmarshaling to succeed for
unrelated malformed fields, and parse each raw field only within its
corresponding detection path while preserving standard error-envelope detection
and known-code/type behavior.
---
Nitpick comments:
In `@relay/channel/openai/relay-openai.go`:
- Around line 141-159: Move the terminal-error detection block in the stream
loop to run immediately after assigning `lastStreamData`, before
`collectStreamFunctionCallNames` and `processTokenData`. When an upstream error
chunk is detected before any response data, log it, stop the stream, and return
without processing the chunk as token or tool-call data; preserve normal
processing for non-error chunks.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 43423035-a4b8-454c-a0b4-c714cd659c52
📒 Files selected for processing (2)
relay/channel/openai/helper.gorelay/channel/openai/relay-openai.go
Comment on lines
+258
to
+301
| 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, "" | ||
| } |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
fd -t f 'json.go' common
rg -nP -A 20 'func UnmarshalJsonStr' commonRepository: QuantumNous/new-api
Length of output: 1015
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '--- openai helper relevant section ---\n'
sed -n '230,330p' relay/channel/openai/helper.go
printf '\n--- common/json.go ---\n'
sed -n '1,60p' common/json.go
printf '\n--- isKnownUpstreamErrorCode usages/implementation ---\n'
rg -n -B 5 -A 25 'func isKnownUpstream errorCode|isKnownUpstreamErrorCode' relay common || true
printf '\n--- deterministic Go stdlib partial field mismatch probe ---\n'
cat > /tmp/decode_probe.go <<'GO'
package main
import (
"encoding/json"
"fmt"
)
type Payload struct {
Error json.RawMessage `json:"error"`
Code string `json:"code"`
Type string `json:"type"`
Message string `json:"message"`
}
func main() {
input := `{"error":{"message":"oops"},"code":42,"message":"msg"}`
var p Payload
err := json.Unmarshal([]byte(input), &p)
fmt.Printf("input=%q\n", input)
fmt.Printf("err=%T: %v\n", err, err)
fmt.Printf("p.Error=%q\n", p.Error)
fmt.Printf("len(p.Error)=%d\n", len(p.Error))
fmt.Printf("p.Message=%q\n", p.Message)
}
GO
if command -v go >/dev/null 2>&1; then
go run /tmp/decode_probe.go
else
echo "go not available"
fiRepository: QuantumNous/new-api
Length of output: 248
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- openai helper relevant section ---'
sed -n '230,330p' relay/channel/openai/helper.go
printf '%s\n' ''
printf '%s\n' '--- common/json.go ---'
sed -n '1,60p' common/json.go
printf '%s\n' ''
printf '%s\n' '--- isKnownUpstreamErrorCode usages/implementation ---'
rg -n -B 5 -A 25 'func isKnownUpstream errorCode|isKnownUpstreamErrorCode' relay common || true
printf '%s\n' ''
printf '%s\n' '--- deterministic Go stdlib partial field mismatch probe ---'
cat > /tmp/decode_probe.go <<'GO'
package main
import (
"encoding/json"
"fmt"
)
type Payload struct {
Error json.RawMessage `json:"error"`
Code string `json:"code"`
Type string `json:"type"`
Message string `json:"message"`
}
func main() {
input := `{"error":{"message":"oops"},"code":42,"message":"msg"}`
var p Payload
err := json.Unmarshal([]byte(input), &p)
fmt.Printf("input=%q\n", input)
fmt.Printf("err=%T: %v\n", err, err)
fmt.Printf("p.Error=%q\n", p.Error)
fmt.Printf("len(p.Error)=%d\n", len(p.Error))
fmt.Printf("p.Message=%q\n", p.Message)
}
GO
if command -v go >/dev/null 2>&1; then
go run /tmp/decode_probe.go
else
echo "go not available"
fiRepository: QuantumNous/new-api
Length of output: 7010
🏁 Script executed:
#!/bin/bash
set -euo pipefail
cat > /tmp/decode_probe2.go <<'GO'
package main
import (
"encoding/json"
"fmt"
"strings"
)
type Payload struct {
Error json.RawMessage `json:"error"`
Code string `json:"code"`
Type string `json:"type"`
Message string `json:"message"`
}
func main() {
inputs := []string{
`{"error":{"message":"oops"},"code":42,"message":"msg"}`,
`{"error":{"message":"oops"},"type":403,"message":"msg"}`,
`{"error":{"message":"oops"},"code":"server_error","message":"msg"}`,
`{"error":22,"code":"server_error","message":"msg"}`,
}
for _, s := range inputs {
var p Payload
err := json.Unmarshal([]byte(s), &p)
fmt.Printf("err=%T: %v | Error=%q | Code=%q | Type=%q | Message=%q\n",
err, err, p.Error, p.Code, p.Type, p.Message)
after := strings.ToLower(strings.TrimSpace(p.Type))
fmt.Printf(" after lower trim: %q matches error/upstream_error=%v\n", after, after == "error" || after == "upstream_error")
}
}
GO
go run /tmp/decode_probe2.goRepository: QuantumNous/new-api
Length of output: 921
Parse the top-level fields after unmarshaling instead of abandoning partial payloads.
UnmarshalJsonStr delegates to stdlib json.Unmarshal, which populates valid fields such as error.message before returning an UnmarshalTypeError for a mismatched field. When code or type is a number (e.g. 403), this helper returns false, "" and hides the OpenAI error envelope. Use json.RawMessage for optional top-level fields and parse only the fields needed by each detection path.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@relay/channel/openai/helper.go` around lines 258 - 301, The
isOpenAITextStreamErrorChunk helper currently discards partially unmarshaled
payloads when optional top-level code or type fields have incompatible JSON
types. Change the payload fields used for code, type, and message parsing to
json.RawMessage, allow unmarshaling to succeed for unrelated malformed fields,
and parse each raw field only within its corresponding detection path while
preserving standard error-envelope detection and known-code/type behavior.
11 tasks
6 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
📝 变更描述
修复流式请求中 SSE 流内嵌错误无法触发通道切换的问题。
问题： gpt-5.5 等模型在流式场景下，上游返回 HTTP 200 正常建立 SSE 连接后，在流中嵌入
rate_limit_exceeded错误，new-api无法感知该错误（只透传数据给客户端），导致
newAPIError == nil不触发通道重试。同时客户端中断连接后计费为 0。根因：
OaiStreamHandler的 dataHandler 仅检测 JSON 解析错误，不检测 SSE chunk 中的 API 业务错误。而图片流处理(
relay_image.go) 已有同等的isOpenAIImageStreamErrorEvent()机制。修复：
isOpenAITextStreamErrorChunk()函数，检测 SSE 数据中的三种错误模式：type: "error"/"upstream_error"类型OaiStreamHandler首个 chunk 未向客户端发送数据前检测到错误时，调用sr.Stop()中止流StreamScannerHandler返回后，若流非正常结束且无数据发送，向上层返回 error 触发重试/兜底🚀 变更类型
✅ 提交前检查项
go build ./relay/...），现有测试通过（go test ./relay/channel/openai/...）📸 运行证明
$ go test ./relay/channel/openai/...
ok github.com/QuantumNous/new-api/relay/channel/openai (cached)
🤖 Generated with Claude Code
Summary by CodeRabbit