fix: 识别上游 HTTP 200 错误体,正确记账并返回 502

问题:OpenRouter 等上游超时时返回 HTTP 200 但响应体/SSE 内含
error 字段(如 {"error":{"code":504}}),网关只检查 StatusCode>=400,
导致:1) 客户端收到假 200 + 空内容;2) 用量被错误记为 success。

修复:
- bufferResponse 检测 JSON 错误体(openai 风格 error / anthropic 风格 type=error),
  识别时以 502 返回给客户端,并向 Dispatch 返回错误码
- streamResponse 扫描 SSE data 载荷中的 error 字段,识别时返回错误码
- Dispatch 对带错误码的响应按失败记账(status=error, error_code=upstream_error),
  不产生费用
This commit is contained in:
Sakurasan
2026-09-01 03:31:55 +08:00
parent 104ecc4691
commit 20f365d11a
+68 -10
View File
@@ -302,13 +302,25 @@ func (g *Gateway) Dispatch(c *gin.Context, req *Request) {
g.channelSvc.RecordSuccess(ch.ID)
// Stream or buffer response;tok 从上游响应(SSE usage 块或非流式 JSON)提取。
// 上游可能返回 HTTP 200 但 body/SSE 内带 error(OpenRouter 超时等),
// 此时按失败记账(errCode 非空),非流式错误体以 502 返回给客户端。
var tok convert.TokenUsage
var errCode string
if req.Stream {
tok = g.streamResponse(c, resp, req.Protocol, targetFormat, req.rawBuf)
tok, errCode = g.streamResponse(c, resp, req.Protocol, targetFormat, req.rawBuf)
} else {
tok = g.bufferResponse(c, resp, req.Protocol, targetFormat, req.rawBuf)
tok, errCode, _ = g.bufferResponse(c, resp, req.Protocol, targetFormat, req.rawBuf)
}
resp.Body.Close()
if errCode != "" {
// 记账为失败(错误码),不产生费用;响应内容已由 buffer/stream 写出
g.recordUsage(req, cand, ch, usage.Event{
IsError: true,
ErrorCode: errCode,
LatencyMS: int(time.Since(start).Milliseconds()),
}, tok)
return
}
// 成功记录:用量 + 定价计费。
g.recordUsage(req, cand, ch, usage.Event{
LatencyMS: int(time.Since(start).Milliseconds()),
@@ -433,7 +445,9 @@ func (g *Gateway) setHeaders(req *http.Request, ch *store.Channel, apiKey string
// streamResponse 流式响应:按 \n\n 分块零缓冲转发;跨协议时逐行转换。
// 返回从上游 SSE usage 块累计的 token 用量(按上游协议解析)。
// capture 非 nil 时把上游原始行累积进去(原始响应记录)。
func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string, capture *strings.Builder) convert.TokenUsage {
// 上游部分实现(如 OpenRouter)在超时时返回 HTTP 200 但 SSE data 内带
// error 字段;检测到则返回错误码,供 Dispatch 按失败记账。
func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string, capture *strings.Builder) (convert.TokenUsage, string) {
w := c.Writer
c.Header("Content-Type", "text/event-stream")
c.Header("Cache-Control", "no-cache")
@@ -452,6 +466,7 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt
// 同时喂入用量累计器(usage 块可能出现在任一事件)。
r := bufio.NewReaderSize(resp.Body, 32*1024)
accum := convert.NewStreamUsageAccum()
errCode := ""
for {
buf := []byte{}
for {
@@ -463,14 +478,14 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt
buf = append(buf, line...)
if err == io.EOF {
if len(buf) == 0 {
return accum.Usage()
return accum.Usage(), errCode
}
if !bytes.HasSuffix(buf, []byte("\n")) {
buf = append(buf, '\n')
}
} else if err != nil {
log.Printf("stream read error: %v", err)
return accum.Usage()
return accum.Usage(), errCode
}
if len(buf) >= 2 && bytes.HasSuffix(buf, []byte("\n\n")) {
break
@@ -485,6 +500,9 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt
// 先解析用量(data: {...} 行),再决定转发内容。
for _, data := range sseDataPayloads(buf) {
accum.Feed(data, upstreamProto)
if errCode == "" && streamChunkHasError(data) {
errCode = "upstream_stream_error"
}
}
out := buf
@@ -495,7 +513,7 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt
continue
}
if _, err := w.Write(out); err != nil {
return accum.Usage() // 客户端已断开
return accum.Usage(), errCode // 客户端已断开
}
if flusher != nil {
flusher.Flush()
@@ -505,11 +523,24 @@ func (g *Gateway) streamResponse(c *gin.Context, resp *http.Response, clientProt
// 发完 [DONE] 后不关连接,继续读会阻塞到超时;据此主动收尾。
// responses 协议没有 [DONE],以 response.completed 事件收尾。
if streamTerminated(buf, upstreamProto) {
return accum.Usage()
return accum.Usage(), errCode
}
}
}
// streamChunkHasError 判断一块 SSE data 载荷是否带 error 字段(OpenRouter 超时等)。
func streamChunkHasError(data []byte) bool {
var m map[string]any
if json.Unmarshal(data, &m) != nil {
return false
}
if _, ok := m["error"]; ok {
return true
}
// responses 协议错误事件可能形如 {"type":"error",...}
return m["type"] == "error"
}
// streamTerminated 判断一块 SSE 是否为上游流的结束事件。
func streamTerminated(chunk []byte, proto string) bool {
switch proto {
@@ -546,11 +577,14 @@ func sseDataPayloads(chunk []byte) [][]byte {
return out
}
func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string, capture *strings.Builder) convert.TokenUsage {
// bufferResponse 非流式响应:整体读取、可选转换后写回。
// 返回 (用量, 错误码, 是否错误)。部分上游(如 OpenRouter)在超时时返回
// HTTP 200 但 JSON 内含 error 字段,需要识别并让调用方按失败处理。
func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProto, upstreamProto string, capture *strings.Builder) (convert.TokenUsage, string, bool) {
body, err := io.ReadAll(resp.Body)
if err != nil {
g.writeError(c, http.StatusBadGateway, "failed to read response")
return convert.TokenUsage{}
return convert.TokenUsage{}, "", false
}
// 原始响应捕获(仅管理员+开关开启时启用)。
@@ -561,6 +595,9 @@ func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProt
// 用量从上游原始响应体提取(先于转换,转换会改字段名)。
tok, _ := convert.ExtractUsageJSON(body, upstreamProto)
// HTTP 200 但带 error 字段(OpenRouter 超时 504 等):识别并转失败。
errCode, isErr := bodyHasError(body)
out := body
if upstreamProto != clientProto {
if converted, cerr := convert.ConvertResponse(body, upstreamProto, clientProto); cerr == nil {
@@ -573,8 +610,29 @@ func (g *Gateway) bufferResponse(c *gin.Context, resp *http.Response, clientProt
// 直通:部分上游(如 OpenRouter)的 non-stream 响应在 JSON 前夹带空白/注释
out = convert.CleanJSON(body)
}
// 上游错误体:用 502 返回,让客户端感知失败(不伪装成 200)。
if isErr {
c.Data(http.StatusBadGateway, "application/json", out)
return tok, errCode, true
}
c.Data(resp.StatusCode, "application/json", out)
return tok
return tok, errCode, false
}
// bodyHasError 判断 JSON 响应体是否带 error 字段(openai 风格 {"error":{...}} 或
// anthropic 风格 {"type":"error",...})。返回 (错误码, 是否错误)。找不到 JSON 返回 ("", false)。
func bodyHasError(body []byte) (string, bool) {
var m map[string]any
if json.Unmarshal(bytes.TrimSpace(body), &m) != nil {
return "", false
}
if _, ok := m["error"]; ok {
return "upstream_error", true
}
if m["type"] == "error" {
return "upstream_error", true
}
return "", false
}
func (g *Gateway) writeError(c *gin.Context, status int, message string) {