mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-14 20:12:39 +08:00
feeeei
f4f8ff04db
客户端在响应提交前断开时,只有排队/换号阶段和 OpenAI 的兜底写入会标 499。 Claude 非流式与 Gemini 的 Chat Completions / Messages / Responses 兼容入口 在等待上游时断开,由 GatewayHandler.ensureForwardErrorResponse 补写 502; Gemini 原生入口什么都不写,落成 200 并被 ops 记成一条 provider 的 "Recovered upstream error";Antigravity 回写 502。这些行都按上游失败计入 SLA 与上游错误数。 - GatewayHandler.ensureForwardErrorResponse 补客户端断开检查,与 OpenAI 版一致 - GatewayHandler 的 Messages / ChatCompletions / Responses 与 Gemini 原生入口: 首次选号失败时先判断客户端断开;ChatCompletions / Responses 循环开头改用 failoverClientGone,不再直接 return 落成 200 - Gemini 原生入口:forward 失败时补 499 标记;并发槽错误改走 concurrencyErrorResponse 统一映射(断开 499、排队满/超时 429、其余 503) - Antigravity 上游响应前断开回写 499;googleapi 与 Antigravity 的 Google 错误 格式把 499 映射为 CANCELLED - OpenAI alpha search 非 failover 错误分支补断开检查 - Claude / Gemini / OpenAI 传输层错误:请求 context 已取消时不记 ops 上游事件 - ops:IgnoreContextCanceled 开启时跳过没有上游错误上下文的 499;带上游错误的 499(上游失败后客户端没等到换号结果就离开)仍按上游失败落库 计费不变:Gemini 流式中途断开的用量处理不在本次范围。新增 handler/service 单测覆盖各入口的断开场景与 ops 过滤规则。
223 lines
8.3 KiB
Go
223 lines
8.3 KiB
Go
package handler
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"strings"
|
||
"testing"
|
||
|
||
"github.com/Wei-Shaw/sub2api/internal/service"
|
||
"github.com/gin-gonic/gin"
|
||
"github.com/stretchr/testify/assert"
|
||
"github.com/stretchr/testify/require"
|
||
)
|
||
|
||
func TestGatewayEnsureForwardErrorResponse_WritesFallbackWhenNotWritten(t *testing.T) {
|
||
gin.SetMode(gin.TestMode)
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
c.Request = httptest.NewRequest(http.MethodGet, "/", nil)
|
||
|
||
h := &GatewayHandler{}
|
||
wrote := h.ensureForwardErrorResponse(c, false)
|
||
|
||
require.True(t, wrote)
|
||
require.Equal(t, http.StatusBadGateway, w.Code)
|
||
|
||
var parsed map[string]any
|
||
err := json.Unmarshal(w.Body.Bytes(), &parsed)
|
||
require.NoError(t, err)
|
||
assert.Equal(t, "error", parsed["type"])
|
||
errorObj, ok := parsed["error"].(map[string]any)
|
||
require.True(t, ok)
|
||
assert.Equal(t, "upstream_error", errorObj["type"])
|
||
assert.Equal(t, "Upstream request failed", errorObj["message"])
|
||
}
|
||
|
||
// Writer 已写后 ensureForwardErrorResponse 必须把错误以 SSE 形式追加，
|
||
// 而不是 silent EOF。非 /responses 路径走 legacy data:{"type":"error"} 分支。
|
||
func TestGatewayEnsureForwardErrorResponse_AppendsSSEAfterWritten(t *testing.T) {
|
||
gin.SetMode(gin.TestMode)
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
c.Request = httptest.NewRequest(http.MethodGet, "/", nil)
|
||
c.String(http.StatusTeapot, "already written")
|
||
|
||
h := &GatewayHandler{}
|
||
wrote := h.ensureForwardErrorResponse(c, false)
|
||
|
||
require.True(t, wrote)
|
||
require.Equal(t, http.StatusTeapot, w.Code)
|
||
assert.Contains(t, w.Body.String(), "already written")
|
||
assert.Contains(t, w.Body.String(), `data: {"type":"error"`)
|
||
}
|
||
|
||
func TestGatewayEnsureForwardErrorResponse_SkipsCommittedSSEError(t *testing.T) {
|
||
gin.SetMode(gin.TestMode)
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
c.Request = httptest.NewRequest(http.MethodPost, EndpointResponses, nil)
|
||
c.Header("Content-Type", "text/event-stream")
|
||
_, _ = c.Writer.WriteString("event: error\ndata: {\"type\":\"error\"}\n\n")
|
||
service.MarkResponseCommitted(c)
|
||
|
||
h := &GatewayHandler{}
|
||
wrote := h.ensureForwardErrorResponse(c, true)
|
||
|
||
require.False(t, wrote)
|
||
require.Equal(t, 1, strings.Count(w.Body.String(), "event: error"))
|
||
}
|
||
|
||
// 客户端已断开且响应未提交：不补写 502，标记 499 供访问日志与 ops 归类。
|
||
func TestGatewayEnsureForwardErrorResponse_SkipsCanceledClient(t *testing.T) {
|
||
gin.SetMode(gin.TestMode)
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil).WithContext(ctx)
|
||
cancel()
|
||
|
||
h := &GatewayHandler{}
|
||
wrote := h.ensureForwardErrorResponse(c, false)
|
||
|
||
require.False(t, wrote)
|
||
require.Equal(t, statusClientClosedRequest, c.Writer.Status())
|
||
require.Empty(t, w.Body.String())
|
||
}
|
||
|
||
// 客户端已断开但流已开始：状态码已固化为 200，不再向断开的连接追加错误帧。
|
||
func TestGatewayEnsureForwardErrorResponse_CanceledClientAfterStreamStartedAppendsNothing(t *testing.T) {
|
||
gin.SetMode(gin.TestMode)
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil).WithContext(ctx)
|
||
c.Header("Content-Type", "text/event-stream")
|
||
_, _ = c.Writer.WriteString(":\n\n")
|
||
cancel()
|
||
|
||
h := &GatewayHandler{}
|
||
wrote := h.ensureForwardErrorResponse(c, true)
|
||
|
||
require.False(t, wrote)
|
||
require.Equal(t, http.StatusOK, c.Writer.Status())
|
||
require.Equal(t, ":\n\n", w.Body.String())
|
||
require.Empty(t, service.GetOpsStreamErrors(c))
|
||
}
|
||
|
||
// case B 回归：Anthropic-backed /responses，Writer 已被写过时
|
||
// ensureForwardErrorResponse 仍要发 response.failed。
|
||
func TestGatewayEnsureForwardErrorResponse_ResponsesRouteAfterWrittenEmitsResponseFailed(t *testing.T) {
|
||
gin.SetMode(gin.TestMode)
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
c.Request = httptest.NewRequest(http.MethodPost, EndpointResponses, nil)
|
||
_, _ = c.Writer.WriteString(":\n\n")
|
||
|
||
h := &GatewayHandler{}
|
||
wrote := h.ensureForwardErrorResponse(c, false)
|
||
|
||
require.True(t, wrote)
|
||
body := w.Body.String()
|
||
assert.Contains(t, body, ":\n\n")
|
||
assert.Contains(t, body, "event: response.failed\n")
|
||
assert.Contains(t, body, `"type":"response.failed"`)
|
||
}
|
||
|
||
func TestGatewayForwardErrorAlreadyCommunicated(t *testing.T) {
|
||
gin.SetMode(gin.TestMode)
|
||
|
||
t.Run("json error already written", func(t *testing.T) {
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil)
|
||
before := c.Writer.Size()
|
||
c.JSON(http.StatusBadGateway, gin.H{
|
||
"type": "error",
|
||
"error": gin.H{
|
||
"type": "upstream_error",
|
||
"message": "Your Claude Code version (2.1.39) is below the minimum required version (2.1.81). Please update: npm update -g @anthropic-ai/claude-code",
|
||
},
|
||
})
|
||
|
||
reported := gatewayForwardErrorAlreadyCommunicated(c, before, errors.New("upstream error: 400 message=version too low"))
|
||
|
||
require.True(t, reported)
|
||
body := w.Body.String()
|
||
assert.NotContains(t, body, `data: {"type":"error"`)
|
||
})
|
||
|
||
t.Run("sse ping still needs fallback", func(t *testing.T) {
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil)
|
||
c.Header("Content-Type", "text/event-stream")
|
||
before := c.Writer.Size()
|
||
_, _ = c.Writer.WriteString(":\n\n")
|
||
|
||
reported := gatewayForwardErrorAlreadyCommunicated(c, before, errors.New("stream read error: unexpected EOF"))
|
||
|
||
require.False(t, reported)
|
||
})
|
||
|
||
t.Run("no write still needs fallback", func(t *testing.T) {
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil)
|
||
|
||
reported := gatewayForwardErrorAlreadyCommunicated(c, c.Writer.Size(), errors.New("upstream request failed"))
|
||
|
||
require.False(t, reported)
|
||
})
|
||
|
||
// apikey 场景核心回归：复刻 GatewayService.handleErrorResponse 的 case 400 ——
|
||
// 原样透传上游 JSON body 后返回 err。此时错误已经完整告知客户端，
|
||
// handler 不得再追加 data:{"type":"error"} 帧，否则响应被污染成「JSON + 一行 data:」。
|
||
t.Run("upstream 400 json passthrough via c.Data", func(t *testing.T) {
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil)
|
||
before := c.Writer.Size()
|
||
upstreamBody := []byte(`{"type":"error","error":{"type":"upstream_error","message":"Your Claude Code version (2.1.39) is below the minimum required version (2.1.81). Please update: npm update -g @anthropic-ai/claude-code"}}`)
|
||
c.Data(http.StatusBadRequest, "application/json", upstreamBody)
|
||
|
||
reported := gatewayForwardErrorAlreadyCommunicated(c, before, errors.New("upstream error: 400 message=version too low"))
|
||
|
||
require.True(t, reported)
|
||
body := w.Body.String()
|
||
assert.NotContains(t, body, `data: {"type":"error"`)
|
||
// 客户端只应收到上游那一份错误，没有被追加第二份。
|
||
assert.Equal(t, 1, strings.Count(body, `"type":"error"`))
|
||
})
|
||
|
||
// 流式已开始（已 flush 真实 SSE 事件，不只是 ping）+ 上游中途 400：
|
||
// HTTP 200 已固化，仍需 handler 补协议级终止帧，故不算「已完整告知」。
|
||
t.Run("streaming 400 mid-stream still needs fallback", func(t *testing.T) {
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil)
|
||
c.Header("Content-Type", "text/event-stream")
|
||
before := c.Writer.Size()
|
||
_, _ = c.Writer.WriteString("event: message_start\ndata: {\"type\":\"message_start\"}\n\n")
|
||
|
||
reported := gatewayForwardErrorAlreadyCommunicated(c, before, errors.New("upstream error: 400 message=version too low"))
|
||
|
||
require.False(t, reported)
|
||
})
|
||
|
||
// 防御边界：err 为 nil 时永远不算「已告知」，避免在成功路径误吞兜底逻辑。
|
||
t.Run("nil error never reports communicated", func(t *testing.T) {
|
||
w := httptest.NewRecorder()
|
||
c, _ := gin.CreateTestContext(w)
|
||
c.Request = httptest.NewRequest(http.MethodPost, EndpointMessages, nil)
|
||
c.JSON(http.StatusOK, gin.H{"ok": true})
|
||
|
||
reported := gatewayForwardErrorAlreadyCommunicated(c, 0, nil)
|
||
|
||
require.False(t, reported)
|
||
})
|
||
}
|