fix(gateway): 客户端断开统一标记 499,不再被记成 502/200 上游错误

客户端在响应提交前断开时,只有排队/换号阶段和 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 过滤规则。
This commit is contained in:
feeeei
2026-09-26 17:13:10 +08:00
parent a3eb7ef302
commit f4f8ff04db
21 changed files with 529 additions and 11 deletions
@@ -2,10 +2,13 @@ package handler
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"testing"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
)
@@ -73,3 +76,65 @@ func TestConcurrencyErrorResponse(t *testing.T) {
})
}
}
func TestGoogleConcurrencyError(t *testing.T) {
tests := []struct {
name string
err error
wantStatus int
wantGStatus string
wantMessage string
}{
{
name: "client cancellation is 499",
err: context.Canceled,
wantStatus: statusClientClosedRequest,
wantGStatus: "CANCELLED",
wantMessage: "context canceled",
},
{
name: "full wait queue is 429",
err: &WaitQueueFullError{SlotType: "user"},
wantStatus: http.StatusTooManyRequests,
wantGStatus: "RESOURCE_EXHAUSTED",
wantMessage: "Too many pending requests, please retry later",
},
{
name: "slot wait timeout is 429",
err: &ConcurrencyError{SlotType: "user", IsTimeout: true},
wantStatus: http.StatusTooManyRequests,
wantGStatus: "RESOURCE_EXHAUSTED",
wantMessage: "Concurrency limit exceeded for user, please retry later",
},
{
name: "acquire backend error is 503",
err: errors.New("redis unavailable"),
wantStatus: http.StatusServiceUnavailable,
wantGStatus: "INTERNAL",
wantMessage: "Service temporarily unavailable, please retry later",
},
}
gin.SetMode(gin.TestMode)
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
googleConcurrencyError(c, tt.err, "user")
require.Equal(t, tt.wantStatus, rec.Code)
var body struct {
Error struct {
Code int `json:"code"`
Message string `json:"message"`
Status string `json:"status"`
} `json:"error"`
}
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &body))
require.Equal(t, tt.wantStatus, body.Error.Code)
require.Equal(t, tt.wantGStatus, body.Error.Status)
require.Equal(t, tt.wantMessage, body.Error.Message)
})
}
}
@@ -330,6 +330,10 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
for {
selection, err := h.gatewayService.SelectAccountWithLoadAwareness(c.Request.Context(), apiKey.GroupID, sessionKey, reqModel, fs.FailedAccountIDs, "", int64(0)) // Gemini 不使用会话限制
if err != nil {
if failoverClientGone(c) {
reqLog.Info("gateway.account_select_aborted_client_disconnected", zap.Error(err))
return
}
if len(fs.FailedAccountIDs) == 0 {
cls := classifyNoAccountErrorFromGin(c, h.gatewayService, apiKey, reqModel, reqModel, service.PlatformGemini)
if !cls.ModelNotFound {
@@ -660,6 +664,10 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
)
selection, err := h.gatewayService.SelectAccountWithLoadAwareness(c.Request.Context(), currentAPIKey.GroupID, sessionKey, reqModel, fs.FailedAccountIDs, parsedReq.MetadataUserID, subject.UserID)
if err != nil {
if failoverClientGone(c) {
reqLog.Info("gateway.account_select_aborted_client_disconnected", zap.Error(err))
return
}
if len(fs.FailedAccountIDs) == 0 {
cls := classifyNoAccountErrorFromGin(c, h.gatewayService, currentAPIKey, reqModel, reqModel, platform)
if !cls.ModelNotFound {
@@ -1971,10 +1979,15 @@ func (h *GatewayHandler) handleStreamingAwareErrorWithCode(c *gin.Context, statu
// Writer 已被写过时(ping 已 flush)走 streamStarted 分支,
// 让 handleStreamingAwareError 通过 SSE 发协议合规的终止事件,
// 否则下游收到的就是 silent EOF。
// 客户端已断开时不补写,响应未提交则标记 499(见 failoverClientGone)。
func (h *GatewayHandler) ensureForwardErrorResponse(c *gin.Context, streamStarted bool) bool {
if c == nil || c.Writer == nil {
return false
}
if c.Request != nil && errors.Is(c.Request.Context().Err(), context.Canceled) {
failoverClientGone(c)
return false
}
if service.IsResponseCommitted(c) {
return false
}
@@ -93,6 +93,8 @@ func TestGatewayHandlerPreCancelledCompatibleRequestsDoNotSelectAccount(t *testi
require.Zero(t, schedulerCache.snapshotCalls.Load(), "a cancelled request must stop before the account selector")
_, selected := c.Get(opsAccountIDKey)
require.False(t, selected)
require.Equal(t, statusClientClosedRequest, c.Writer.Status(), "an uncommitted cancelled request is marked 499")
require.Zero(t, recorder.Body.Len())
})
}
}
@@ -168,11 +168,15 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) {
}
for {
if c.Request.Context().Err() != nil {
if failoverClientGone(c) {
return
}
selection, err := h.gatewayService.SelectAccountWithLoadAwareness(c.Request.Context(), apiKey.GroupID, selectionSessionHash, reqModel, fs.FailedAccountIDs, "", int64(0))
if err != nil {
if failoverClientGone(c) {
reqLog.Info("gateway.cc.account_select_aborted_client_disconnected", zap.Error(err))
return
}
if len(fs.FailedAccountIDs) == 0 {
cls := classifyNoAccountErrorFromGin(c, h.gatewayService, apiKey, reqModel, reqModel, groupPlatform)
cls = classifySelectionFailureError(err, cls)
@@ -1,6 +1,7 @@
package handler
import (
"context"
"encoding/json"
"errors"
"net/http"
@@ -70,6 +71,43 @@ func TestGatewayEnsureForwardErrorResponse_SkipsCommittedSSEError(t *testing.T)
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) {
@@ -170,11 +170,15 @@ func (h *GatewayHandler) Responses(c *gin.Context) {
fs := NewFailoverState(h.maxAccountSwitches, false)
for {
if requestCtx.Err() != nil {
if failoverClientGone(c) {
return
}
selection, err := h.gatewayService.SelectAccountWithLoadAwareness(requestCtx, apiKey.GroupID, sessionHash, reqModel, fs.FailedAccountIDs, "", int64(0))
if err != nil {
if failoverClientGone(c) {
reqLog.Info("gateway.responses.account_select_aborted_client_disconnected", zap.Error(err))
return
}
if len(fs.FailedAccountIDs) == 0 {
cls := classifyNoAccountErrorFromGin(c, h.gatewayService, apiKey, reqModel, reqModel, effectiveAPIKeyPlatform(c, apiKey))
cls = classifySelectionFailureError(err, cls)
@@ -0,0 +1,130 @@
//go:build unit
package handler
import (
"bytes"
"context"
"net/http"
"net/http/httptest"
"net/url"
"sync/atomic"
"testing"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey"
middleware "github.com/Wei-Shaw/sub2api/internal/server/middleware"
"github.com/Wei-Shaw/sub2api/internal/service"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
)
// clientCancelUpstream 模拟上游尚未响应时客户端断开:Do 期间取消请求 context,
// 并按 net/http 的形态返回包裹 context.Canceled 的 *url.Error。
type clientCancelUpstream struct {
service.HTTPUpstream
cancel context.CancelFunc
calls atomic.Int32
}
func (u *clientCancelUpstream) Do(req *http.Request, _ string, _ int64, _ int) (*http.Response, error) {
u.calls.Add(1)
u.cancel()
return nil, &url.Error{Op: "Post", URL: req.URL.String(), Err: context.Canceled}
}
type geminiClientCancelFixture struct {
handler *GatewayHandler
group *service.Group
apiKey *service.APIKey
upstream *clientCancelUpstream
ctx context.Context
}
func newGeminiClientCancelFixture(t *testing.T) *geminiClientCancelFixture {
t.Helper()
groupID := int64(9200)
accountID := int64(9201)
group := &service.Group{ID: groupID, Hydrated: true, Platform: service.PlatformGemini, Status: service.StatusActive}
account := &service.Account{
ID: accountID,
Platform: service.PlatformGemini,
Type: service.AccountTypeAPIKey,
Credentials: map[string]any{"api_key": "test-key"},
Status: service.StatusActive,
Schedulable: true,
Concurrency: 1,
Priority: 1,
AccountGroups: []service.AccountGroup{{AccountID: accountID, GroupID: groupID}},
}
h, cleanup := newTestGatewayHandler(t, group, []*service.Account{account})
t.Cleanup(cleanup)
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
upstream := &clientCancelUpstream{cancel: cancel}
h.geminiCompatService = service.NewGeminiMessagesCompatService(nil, nil, nil, nil, nil, nil, upstream, nil, &config.Config{})
apiKey := &service.APIKey{
ID: 9202, UserID: 9203, GroupID: &groupID, Group: group, Status: service.StatusActive,
User: &service.User{ID: 9203, Concurrency: 10, Balance: 100},
}
return &geminiClientCancelFixture{handler: h, group: group, apiKey: apiKey, upstream: upstream, ctx: ctx}
}
// serve 经 ops 错误日志中间件执行请求,返回响应与 ops 队列中的条目数。
func (f *geminiClientCancelFixture) serve(t *testing.T, route, path, body string, call func(*gin.Context)) (*httptest.ResponseRecorder, int64) {
t.Helper()
setupOpsErrorLogTestQueue(t, 4)
ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
router := gin.New()
router.Use(OpsErrorLoggerMiddleware(ops))
router.POST(route, func(c *gin.Context) {
c.Set(string(middleware.ContextKeyAPIKey), f.apiKey)
c.Set(string(middleware.ContextKeyUser), middleware.AuthSubject{UserID: f.apiKey.UserID, Concurrency: 10})
call(c)
})
req := httptest.NewRequest(http.MethodPost, path, bytes.NewBufferString(body))
req = req.WithContext(context.WithValue(f.ctx, ctxkey.Group, f.group))
req.Header.Set("Content-Type", "application/json")
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req)
return rec, OpsErrorLogQueueLength()
}
// 原生入口:上游响应前客户端断开,响应未提交,标记 499;纯客户端取消不落 ops 错误日志。
func TestGeminiV1BetaModels_ClientCancelBeforeUpstreamResponseMarks499(t *testing.T) {
gin.SetMode(gin.TestMode)
f := newGeminiClientCancelFixture(t)
rec, opsQueued := f.serve(t,
"/v1beta/models/*modelAction",
"/v1beta/models/gemini-2.5-flash:streamGenerateContent?alt=sse",
`{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}`,
f.handler.GeminiV1BetaModels,
)
require.Equal(t, int32(1), f.upstream.calls.Load())
require.Equal(t, statusClientClosedRequest, rec.Code)
require.Zero(t, rec.Body.Len())
require.Zero(t, opsQueued)
}
// Chat Completions 兼容入口(Gemini 分组):同一场景标记 499,不再补写 502。
func TestGatewayChatCompletions_GeminiClientCancelBeforeUpstreamResponseMarks499(t *testing.T) {
gin.SetMode(gin.TestMode)
f := newGeminiClientCancelFixture(t)
rec, opsQueued := f.serve(t,
"/v1/chat/completions",
"/v1/chat/completions",
`{"model":"gemini-2.5-flash","messages":[{"role":"user","content":"hi"}],"stream":true}`,
f.handler.ChatCompletions,
)
require.Equal(t, int32(1), f.upstream.calls.Load())
require.Equal(t, statusClientClosedRequest, rec.Code)
require.Zero(t, rec.Body.Len())
require.Zero(t, opsQueued)
}
@@ -370,7 +370,7 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) {
userReleaseFunc, err := geminiConcurrency.AcquireUserSlotWithWait(c, authSubject.UserID, authSubject.Concurrency, stream, &streamStarted)
if err != nil {
reqLog.Warn("gemini.user_slot_acquire_failed", zap.Error(err))
googleError(c, http.StatusTooManyRequests, err.Error())
googleConcurrencyError(c, err, "user")
return
}
// 确保请求取消时也会释放槽位,避免长连接被动中断造成泄漏
@@ -506,6 +506,10 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) {
for {
selection, err := h.gatewayService.SelectAccountWithLoadAwareness(c.Request.Context(), apiKey.GroupID, sessionKey, modelName, fs.FailedAccountIDs, "", int64(0)) // Gemini 不使用会话限制
if err != nil {
if failoverClientGone(c) {
reqLog.Info("gemini.account_select_aborted_client_disconnected", zap.Error(err))
return
}
if len(fs.FailedAccountIDs) == 0 {
cls := classifyNoAccountErrorFromGin(c, h.gatewayService, apiKey, modelName, modelName, service.PlatformGemini)
if !cls.ModelNotFound {
@@ -598,7 +602,7 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) {
)
if err != nil {
reqLog.Warn("gemini.account_slot_acquire_failed", zap.Int64("account_id", account.ID), zap.Error(err))
googleError(c, http.StatusTooManyRequests, err.Error())
googleConcurrencyError(c, err, "account")
return
}
if accountWaitCounted {
@@ -674,7 +678,8 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) {
return
}
}
// ForwardNative already wrote the response
// 转发层已写出错误响应;客户端断开时转发层不写,响应未提交则标记 499。
failoverClientGone(c)
reqLog.Error("gemini.forward_failed", zap.Int64("account_id", account.ID), zap.Error(err))
return
}
@@ -836,6 +841,13 @@ func googleError(c *gin.Context, status int, message string) {
})
}
// googleConcurrencyError 以 Google 错误格式回写并发槽获取失败,状态码与文案
// 沿用 concurrencyErrorResponse 的统一映射(客户端断开为 499)。
func googleConcurrencyError(c *gin.Context, err error, slotType string) {
status, _, _, message := concurrencyErrorResponse(err, slotType)
googleError(c, status, message)
}
func writeUpstreamResponse(c *gin.Context, res *service.UpstreamHTTPResult) {
if res == nil {
googleError(c, http.StatusBadGateway, "Empty upstream response")
@@ -197,6 +197,10 @@ func (h *OpenAIGatewayHandler) AlphaSearch(c *gin.Context) {
var failoverErr *service.UpstreamFailoverError
if !errors.As(err, &failoverErr) {
h.gatewayService.ReportOpenAIAccountScheduleResult(account, openAIAccountScheduleModel(c, account, requestedModel, false, result), false, nil, err)
if failoverClientGone(c) {
reqLog.Info("openai_alpha_search.forward_aborted_client_disconnected", zap.Int64("account_id", account.ID), zap.Error(err))
return
}
if c.Writer.Size() == writerSizeBeforeForward {
h.errorResponse(c, http.StatusBadGateway, "upstream_error", "Upstream request failed")
}
@@ -1146,6 +1146,9 @@ func OpsErrorLoggerMiddleware(ops *service.OpsService) gin.HandlerFunc {
if shouldSkipOpsErrorLog(c.Request.Context(), ops, parsed.Message, string(body), c.Request.URL.Path) {
return
}
if shouldSkipOpsClientClosed(c, ops, status) {
return
}
apiKey := getOpsAPIKey(c)
@@ -2491,6 +2494,21 @@ func shouldSkipOpsErrorLog(ctx context.Context, ops *service.OpsService, message
return false
}
// shouldSkipOpsClientClosed 按 IgnoreContextCanceled 过滤纯客户端取消的 499。
// 499 表示客户端在响应提交前断开(见 failoverClientGone),通常不带
// "context canceled" 文案,shouldSkipOpsErrorLog 的文本过滤命中不了。
// 本次请求未观察到上游错误时是纯客户端取消;已有上游错误的 499 表示上游失败后
// 客户端没等到换号结果就离开,仍按上游失败落库。
func shouldSkipOpsClientClosed(c *gin.Context, ops *service.OpsService, status int) bool {
if status != statusClientClosedRequest || ops == nil {
return false
}
if !ops.OpsAdvancedSettingsSnapshot().IgnoreContextCanceled {
return false
}
return !hasOpsUpstreamErrorContext(c)
}
// shouldSkipOpsErrorLogForCyber:cyber_policy 命中的请求由 recordCyberPolicyIfMarked
// 统一落一条 status=403 的错误请求,故中间件跳过自身落库,避免双写。
func shouldSkipOpsErrorLogForCyber(c *gin.Context) bool {
@@ -2168,3 +2168,83 @@ func TestNormalizeOpsErrorType_KeepsGeminiInBandSignalTypes(t *testing.T) {
require.Equal(t, errType, normalizeOpsErrorType(errType, "PROHIBITED_CONTENT"), errType)
}
}
type opsAdvancedSettingsRepoStub struct {
service.SettingRepository
advanced string
}
func (r *opsAdvancedSettingsRepoStub) GetValue(context.Context, string) (string, error) {
return "", service.ErrSettingNotFound
}
func (r *opsAdvancedSettingsRepoStub) GetMultiple(context.Context, []string) (map[string]string, error) {
return map[string]string{service.SettingKeyOpsAdvancedSettings: r.advanced}, nil
}
func (r *opsAdvancedSettingsRepoStub) Set(context.Context, string, string) error {
return nil
}
// serveClientClosedRequest 以已取消的请求 context 走 failoverClientGone,复现网关标记 499 的收尾路径。
func serveClientClosedRequest(t *testing.T, ops *service.OpsService, prepare func(c *gin.Context)) {
t.Helper()
router := gin.New()
router.Use(OpsErrorLoggerMiddleware(ops))
router.POST("/v1/messages", func(c *gin.Context) {
if prepare != nil {
prepare(c)
}
failoverClientGone(c)
})
ctx, cancel := context.WithCancel(context.Background())
cancel()
recorder := httptest.NewRecorder()
router.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/v1/messages", nil).WithContext(ctx))
require.Equal(t, statusClientClosedRequest, recorder.Code)
require.Zero(t, recorder.Body.Len())
}
func TestOpsErrorLoggerMiddleware_SkipsPureClientClosed(t *testing.T) {
setupOpsErrorLogTestQueue(t, 2)
gin.SetMode(gin.TestMode)
ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
require.True(t, ops.OpsAdvancedSettingsSnapshot().IgnoreContextCanceled)
serveClientClosedRequest(t, ops, nil)
require.Zero(t, OpsErrorLogQueueLength(), "纯客户端取消的 499 按 IgnoreContextCanceled 跳过")
}
func TestOpsErrorLoggerMiddleware_RecordsClientClosedAfterUpstreamFailure(t *testing.T) {
setupOpsErrorLogTestQueue(t, 2)
gin.SetMode(gin.TestMode)
ops := service.NewOpsService(nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
serveClientClosedRequest(t, ops, func(c *gin.Context) {
c.Set(service.OpsUpstreamErrorsKey, []*service.OpsUpstreamErrorEvent{{
AccountID: 7, UpstreamStatusCode: 524, Kind: "failover", Message: "upstream timeout",
}})
})
require.Equal(t, int64(1), OpsErrorLogQueueLength(), "上游失败后客户端离开仍按上游失败落库")
job := <-opsErrorLogQueue
require.Equal(t, statusClientClosedRequest, job.entry.StatusCode)
require.Equal(t, "upstream", job.entry.ErrorPhase)
require.NotNil(t, job.entry.UpstreamStatusCode)
require.Equal(t, 524, *job.entry.UpstreamStatusCode)
}
func TestOpsErrorLoggerMiddleware_RecordsClientClosedWhenIgnoreContextCanceledDisabled(t *testing.T) {
setupOpsErrorLogTestQueue(t, 2)
gin.SetMode(gin.TestMode)
settings := &opsAdvancedSettingsRepoStub{advanced: `{"ignore_context_canceled":false}`}
ops := service.NewOpsService(nil, settings, nil, nil, nil, nil, nil, nil, nil, nil, nil)
require.False(t, ops.OpsAdvancedSettingsSnapshot().IgnoreContextCanceled)
serveClientClosedRequest(t, ops, nil)
require.Equal(t, int64(1), OpsErrorLogQueueLength())
job := <-opsErrorLogQueue
require.Equal(t, statusClientClosedRequest, job.entry.StatusCode)
}
+2
View File
@@ -16,6 +16,8 @@ func HTTPStatusToGoogleStatus(status int) string {
return "NOT_FOUND"
case http.StatusTooManyRequests:
return "RESOURCE_EXHAUSTED"
case 499: // client closed request
return "CANCELLED"
default:
if status >= 500 {
return "INTERNAL"
@@ -0,0 +1,28 @@
package googleapi
import (
"net/http"
"testing"
)
func TestHTTPStatusToGoogleStatus(t *testing.T) {
cases := []struct {
status int
want string
}{
{http.StatusBadRequest, "INVALID_ARGUMENT"},
{http.StatusUnauthorized, "UNAUTHENTICATED"},
{http.StatusForbidden, "PERMISSION_DENIED"},
{http.StatusNotFound, "NOT_FOUND"},
{http.StatusTooManyRequests, "RESOURCE_EXHAUSTED"},
{499, "CANCELLED"},
{http.StatusBadGateway, "INTERNAL"},
{http.StatusServiceUnavailable, "INTERNAL"},
{http.StatusConflict, "UNKNOWN"},
}
for _, tc := range cases {
if got := HTTPStatusToGoogleStatus(tc.status); got != tc.want {
t.Errorf("HTTPStatusToGoogleStatus(%d) = %q, want %q", tc.status, got, tc.want)
}
}
}
@@ -131,7 +131,7 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
}
// 区分客户端取消和真正的上游失败,返回更准确的错误消息
if c.Request.Context().Err() != nil {
return nil, s.writeClaudeError(c, http.StatusBadGateway, "client_disconnected", "Client disconnected before upstream response")
return nil, s.writeClaudeError(c, antigravityStatusClientClosed, "client_disconnected", "Client disconnected before upstream response")
}
return nil, s.writeClaudeError(c, http.StatusBadGateway, "upstream_error", "Upstream request failed after retries")
}
@@ -374,7 +374,7 @@ func (s *AntigravityGatewayService) handleAntigravityCompatTransportError(c *gin
}
}
if c.Request.Context().Err() != nil {
return s.writeAntigravityCompatError(c, http.StatusBadGateway, "client_disconnected", "Client disconnected before upstream response")
return s.writeAntigravityCompatError(c, antigravityStatusClientClosed, "client_disconnected", "Client disconnected before upstream response")
}
return s.writeAntigravityCompatError(c, http.StatusBadGateway, "upstream_error", "Upstream request failed after retries")
}
@@ -180,7 +180,7 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
}
// 区分客户端取消和真正的上游失败,返回更准确的错误消息
if c.Request.Context().Err() != nil {
return nil, s.writeGoogleError(c, http.StatusBadGateway, "Client disconnected before upstream response")
return nil, s.writeGoogleError(c, antigravityStatusClientClosed, "Client disconnected before upstream response")
}
return nil, s.writeGoogleError(c, http.StatusBadGateway, "Upstream request failed after retries")
}
@@ -686,6 +686,9 @@ func mergeTextPartsToResponse(response map[string]any, textParts []string) map[s
return result
}
// antigravityStatusClientClosed 是客户端在上游响应前断开时回写的状态码(499 client closed request)。
const antigravityStatusClientClosed = 499
func (s *AntigravityGatewayService) writeClaudeError(c *gin.Context, status int, errType, message string) error {
MarkResponseCommitted(c)
c.JSON(status, gin.H{
@@ -790,6 +793,8 @@ func (s *AntigravityGatewayService) writeGoogleError(c *gin.Context, status int,
statusStr = "NOT_FOUND"
case 429:
statusStr = "RESOURCE_EXHAUSTED"
case antigravityStatusClientClosed:
statusStr = "CANCELLED"
case 500:
statusStr = "INTERNAL"
case 502, 503:
@@ -0,0 +1,93 @@
//go:build unit
package service
import (
"context"
"net/http"
"net/http/httptest"
"net/url"
"testing"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
"github.com/tidwall/gjson"
)
// 三条传输层失败处理的共同契约:请求 context 已取消(客户端断开)时不记 Ops 上游错误事件;
// 请求 context 仍存活时即便错误是 context.Canceled 也照常记录,不归为客户端断开。
func TestTransportErrorHandlers_ClientDisconnectRecordsNoOpsEvent(t *testing.T) {
account := &Account{ID: 9, Name: "acc", Platform: PlatformAnthropic}
clientErr := &url.Error{Op: "Post", URL: "https://upstream.example/v1/messages", Err: context.Canceled}
canceledCtx, cancel := context.WithCancel(context.Background())
cancel()
handlers := []struct {
name string
handle func(ctx context.Context, c *gin.Context) error
}{
{"anthropic", func(ctx context.Context, c *gin.Context) error {
s := &GatewayService{accountRepo: &transportTempUnschedRepoStub{}}
return s.handleUpstreamTransportError(ctx, c, account, clientErr, OpsUpstreamErrorEvent{})
}},
{"gemini", func(ctx context.Context, c *gin.Context) error {
s := &GeminiMessagesCompatService{accountRepo: &transportTempUnschedRepoStub{}}
return s.handleUpstreamTransportError(ctx, c, account, clientErr)
}},
{"openai", func(ctx context.Context, c *gin.Context) error {
s := &OpenAIGatewayService{accountRepo: &openaiTransportAccountRepoStub{}}
return s.handleOpenAIUpstreamTransportError(ctx, c, account, clientErr, false)
}},
}
for _, h := range handlers {
t.Run(h.name+"/client disconnected", func(t *testing.T) {
c := newTransportErrorTestGin(t)
err := h.handle(canceledCtx, c)
require.ErrorIs(t, err, context.Canceled)
_, recorded := c.Get(OpsUpstreamErrorsKey)
require.False(t, recorded, "客户端断开不是上游故障")
_, messageSet := c.Get(OpsUpstreamErrorMessageKey)
require.False(t, messageSet)
})
t.Run(h.name+"/request context alive", func(t *testing.T) {
c := newTransportErrorTestGin(t)
err := h.handle(context.Background(), c)
require.ErrorIs(t, err, context.Canceled)
raw, recorded := c.Get(OpsUpstreamErrorsKey)
require.True(t, recorded)
require.Len(t, raw.([]*OpsUpstreamErrorEvent), 1)
})
}
}
func TestAntigravityCompatTransportError_ClientDisconnectWrites499(t *testing.T) {
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
ctx, cancel := context.WithCancel(context.Background())
cancel()
c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil).WithContext(ctx)
err := (&AntigravityGatewayService{}).handleAntigravityCompatTransportError(c, context.Canceled)
require.Error(t, err)
require.Equal(t, antigravityStatusClientClosed, rec.Code)
require.Equal(t, "client_disconnected", gjson.Get(rec.Body.String(), "error.type").String())
}
func TestAntigravityWriteGoogleError_ClientClosedStatus(t *testing.T) {
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1beta/models/gemini-2.5-flash:generateContent", nil)
_ = (&AntigravityGatewayService{}).writeGoogleError(c, antigravityStatusClientClosed, "Client disconnected before upstream response")
require.Equal(t, antigravityStatusClientClosed, rec.Code)
require.Equal(t, "CANCELLED", gjson.Get(rec.Body.String(), "error.status").String())
}
@@ -28,7 +28,8 @@ var gatewayTransportFailoverBody = []byte(`{"type":"error","error":{"type":"upst
// 1. records the failure in Ops error logs (status 0, kind=request_error) —
// the caller passes path-specific fields (UpstreamURL, Passthrough) via
// event; identity, proxy attribution and classification fields are
// filled here from the same account snapshot that built the transport;
// filled here from the same account snapshot that built the transport.
// A client disconnect is not recorded (see isClientCanceledTransportError);
// 2. for durable faults (expired/rejected proxy creds, dead proxy,
// DNS/routing) temporarily unschedules the account and logs a stable warn
// event that alert rules can key on;
@@ -39,6 +40,9 @@ var gatewayTransportFailoverBody = []byte(`{"type":"error","error":{"type":"upst
// It deliberately does NOT write to the response: the handler owns the
// response (failover, or a protocol-correct error once failover is exhausted).
func (s *GatewayService) handleUpstreamTransportError(ctx context.Context, c *gin.Context, account *Account, err error, event OpsUpstreamErrorEvent) error {
if isClientCanceledTransportError(ctx, err) {
return err
}
safeErr := sanitizeUpstreamErrorMessage(err.Error())
setOpsUpstreamError(c, 0, safeErr, "")
event.ProxyID, event.ProxyName = opsUpstreamProxyAttribution(account)
@@ -16,7 +16,8 @@ var geminiTransportFailoverBody = []byte(`{"error":{"code":502,"message":"Upstre
// Chat Completions 兼容)上的传输层失败:Do 返回非 HTTP 错误(代理 / DNS / TCP / TLS /
// 响应头超时),没有拿到任何上游状态码。
//
// - 记录 Ops 错误事件(status 0,kind=request_error);
// - 记录 Ops 错误事件(status 0,kind=request_error);客户端断开不记录
// (见 isClientCanceledTransportError);
// - 持久性故障(代理凭据失效、端点拒绝连接、DNS/路由不可达)临时摘除账号;
// 瞬时故障(EOF / reset / 超时)账号保持可调度;
// - 客户端已断开(context.Canceled)原样返回:不换号、不摘号;
@@ -24,6 +25,9 @@ var geminiTransportFailoverBody = []byte(`{"error":{"code":502,"message":"Upstre
//
// 本函数不写响应:响应归 handler 所有(换号,或耗尽后按端点格式渲染错误)。
func (s *GeminiMessagesCompatService) handleUpstreamTransportError(ctx context.Context, c *gin.Context, account *Account, err error) error {
if isClientCanceledTransportError(ctx, err) {
return err
}
safeErr := sanitizeUpstreamErrorMessage(err.Error())
setOpsUpstreamError(c, 0, safeErr, "")
event := OpsUpstreamErrorEvent{
@@ -91,9 +91,18 @@ func classifyUpstreamTransportError(err error) upstreamTransportErrorClass {
return upstreamTransportErrorClass{}
}
// isClientCanceledTransportError reports whether a transport-level failure was
// caused by the client disconnecting: the request context itself is canceled
// and the round-trip aborted with context.Canceled. Such a failure says nothing
// about the upstream, so it is not recorded as an Ops upstream error event.
func isClientCanceledTransportError(ctx context.Context, err error) bool {
return errors.Is(err, context.Canceled) && ctx != nil && errors.Is(ctx.Err(), context.Canceled)
}
// handleOpenAIUpstreamTransportError handles a transport-level upstream failure
// (Do/DoWithTLS returned a non-HTTP error: proxy/DNS/TCP/TLS). It:
// 1. records the failure in Ops error logs (status 0, kind=request_error);
// 1. records the failure in Ops error logs (status 0, kind=request_error),
// except when the client disconnected (see isClientCanceledTransportError);
// 2. for durable faults (expired/rejected proxy creds, dead proxy, DNS/routing)
// temporarily unschedules the account (DB + in-memory) and logs a stable
// warn event that alert rules can key on;
@@ -106,6 +115,9 @@ func classifyUpstreamTransportError(err error) upstreamTransportErrorClass {
//
// passthrough tags the Ops error event for the OpenAI passthrough forward path.
func (s *OpenAIGatewayService) handleOpenAIUpstreamTransportError(ctx context.Context, c *gin.Context, account *Account, err error, passthrough bool) error {
if isClientCanceledTransportError(ctx, err) {
return err
}
safeErr := sanitizeUpstreamErrorMessage(err.Error())
setOpsUpstreamError(c, 0, safeErr, "")
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{