From f4f8ff04db3ff7d27993869796aa14ae039b16af Mon Sep 17 00:00:00 2001 From: feeeei Date: Sat, 26 Sep 2026 16:59:08 +0800 Subject: [PATCH] =?UTF-8?q?fix(gateway):=20=E5=AE=A2=E6=88=B7=E7=AB=AF?= =?UTF-8?q?=E6=96=AD=E5=BC=80=E7=BB=9F=E4=B8=80=E6=A0=87=E8=AE=B0=20499,?= =?UTF-8?q?=E4=B8=8D=E5=86=8D=E8=A2=AB=E8=AE=B0=E6=88=90=20502/200=20?= =?UTF-8?q?=E4=B8=8A=E6=B8=B8=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 客户端在响应提交前断开时,只有排队/换号阶段和 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 过滤规则。 --- .../concurrency_error_response_test.go | 65 +++++++++ backend/internal/handler/gateway_handler.go | 13 ++ .../gateway_handler_cancellation_test.go | 2 + .../gateway_handler_chat_completions.go | 6 +- .../gateway_handler_error_fallback_test.go | 38 +++++ .../handler/gateway_handler_responses.go | 6 +- .../handler/gemini_client_cancel_test.go | 130 ++++++++++++++++++ .../internal/handler/gemini_v1beta_handler.go | 18 ++- .../internal/handler/openai_alpha_search.go | 4 + backend/internal/handler/ops_error_logger.go | 18 +++ .../internal/handler/ops_error_logger_test.go | 80 +++++++++++ backend/internal/pkg/googleapi/status.go | 2 + backend/internal/pkg/googleapi/status_test.go | 28 ++++ .../service/antigravity_gateway_claude.go | 2 +- .../service/antigravity_gateway_compat.go | 2 +- .../service/antigravity_gateway_gemini.go | 2 +- .../service/antigravity_gateway_streaming.go | 5 + .../service/client_disconnect_status_test.go | 93 +++++++++++++ .../gateway_upstream_transport_error.go | 6 +- .../gemini_upstream_transport_error.go | 6 +- .../openai_upstream_transport_error.go | 14 +- 21 files changed, 529 insertions(+), 11 deletions(-) create mode 100644 backend/internal/handler/gemini_client_cancel_test.go create mode 100644 backend/internal/pkg/googleapi/status_test.go create mode 100644 backend/internal/service/client_disconnect_status_test.go diff --git a/backend/internal/handler/concurrency_error_response_test.go b/backend/internal/handler/concurrency_error_response_test.go index 2d1b3d6ad2..4a6b8a79ae 100644 --- a/backend/internal/handler/concurrency_error_response_test.go +++ b/backend/internal/handler/concurrency_error_response_test.go @@ -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) + }) + } +} diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index 3bd69cead7..2f52c0bfe4 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -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 } diff --git a/backend/internal/handler/gateway_handler_cancellation_test.go b/backend/internal/handler/gateway_handler_cancellation_test.go index 07bde1eb13..9573aa8391 100644 --- a/backend/internal/handler/gateway_handler_cancellation_test.go +++ b/backend/internal/handler/gateway_handler_cancellation_test.go @@ -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()) }) } } diff --git a/backend/internal/handler/gateway_handler_chat_completions.go b/backend/internal/handler/gateway_handler_chat_completions.go index 7a98504596..9eec2bb3d1 100644 --- a/backend/internal/handler/gateway_handler_chat_completions.go +++ b/backend/internal/handler/gateway_handler_chat_completions.go @@ -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) diff --git a/backend/internal/handler/gateway_handler_error_fallback_test.go b/backend/internal/handler/gateway_handler_error_fallback_test.go index a38580dfc6..917a85e009 100644 --- a/backend/internal/handler/gateway_handler_error_fallback_test.go +++ b/backend/internal/handler/gateway_handler_error_fallback_test.go @@ -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) { diff --git a/backend/internal/handler/gateway_handler_responses.go b/backend/internal/handler/gateway_handler_responses.go index a170877315..8cf07f1e9d 100644 --- a/backend/internal/handler/gateway_handler_responses.go +++ b/backend/internal/handler/gateway_handler_responses.go @@ -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) diff --git a/backend/internal/handler/gemini_client_cancel_test.go b/backend/internal/handler/gemini_client_cancel_test.go new file mode 100644 index 0000000000..aad811f025 --- /dev/null +++ b/backend/internal/handler/gemini_client_cancel_test.go @@ -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) +} diff --git a/backend/internal/handler/gemini_v1beta_handler.go b/backend/internal/handler/gemini_v1beta_handler.go index a7ae9170c1..140e294794 100644 --- a/backend/internal/handler/gemini_v1beta_handler.go +++ b/backend/internal/handler/gemini_v1beta_handler.go @@ -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") diff --git a/backend/internal/handler/openai_alpha_search.go b/backend/internal/handler/openai_alpha_search.go index 3846c93a7f..66c50be59a 100644 --- a/backend/internal/handler/openai_alpha_search.go +++ b/backend/internal/handler/openai_alpha_search.go @@ -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") } diff --git a/backend/internal/handler/ops_error_logger.go b/backend/internal/handler/ops_error_logger.go index ea58e3af25..a4e30814ca 100644 --- a/backend/internal/handler/ops_error_logger.go +++ b/backend/internal/handler/ops_error_logger.go @@ -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 { diff --git a/backend/internal/handler/ops_error_logger_test.go b/backend/internal/handler/ops_error_logger_test.go index 8a9e8f8859..7714f4e349 100644 --- a/backend/internal/handler/ops_error_logger_test.go +++ b/backend/internal/handler/ops_error_logger_test.go @@ -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) +} diff --git a/backend/internal/pkg/googleapi/status.go b/backend/internal/pkg/googleapi/status.go index 5eb0c54add..18a0eacb41 100644 --- a/backend/internal/pkg/googleapi/status.go +++ b/backend/internal/pkg/googleapi/status.go @@ -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" diff --git a/backend/internal/pkg/googleapi/status_test.go b/backend/internal/pkg/googleapi/status_test.go new file mode 100644 index 0000000000..123515be0f --- /dev/null +++ b/backend/internal/pkg/googleapi/status_test.go @@ -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) + } + } +} diff --git a/backend/internal/service/antigravity_gateway_claude.go b/backend/internal/service/antigravity_gateway_claude.go index 7f03536b7f..264cbe9980 100644 --- a/backend/internal/service/antigravity_gateway_claude.go +++ b/backend/internal/service/antigravity_gateway_claude.go @@ -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") } diff --git a/backend/internal/service/antigravity_gateway_compat.go b/backend/internal/service/antigravity_gateway_compat.go index 8763da57b0..b62e32c842 100644 --- a/backend/internal/service/antigravity_gateway_compat.go +++ b/backend/internal/service/antigravity_gateway_compat.go @@ -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") } diff --git a/backend/internal/service/antigravity_gateway_gemini.go b/backend/internal/service/antigravity_gateway_gemini.go index ce7fc69edf..3ce13e6de4 100644 --- a/backend/internal/service/antigravity_gateway_gemini.go +++ b/backend/internal/service/antigravity_gateway_gemini.go @@ -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") } diff --git a/backend/internal/service/antigravity_gateway_streaming.go b/backend/internal/service/antigravity_gateway_streaming.go index 59d4e459b7..675f0d7f87 100644 --- a/backend/internal/service/antigravity_gateway_streaming.go +++ b/backend/internal/service/antigravity_gateway_streaming.go @@ -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: diff --git a/backend/internal/service/client_disconnect_status_test.go b/backend/internal/service/client_disconnect_status_test.go new file mode 100644 index 0000000000..a2e4080af4 --- /dev/null +++ b/backend/internal/service/client_disconnect_status_test.go @@ -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()) +} diff --git a/backend/internal/service/gateway_upstream_transport_error.go b/backend/internal/service/gateway_upstream_transport_error.go index 7781d5cedd..a9f459fed9 100644 --- a/backend/internal/service/gateway_upstream_transport_error.go +++ b/backend/internal/service/gateway_upstream_transport_error.go @@ -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) diff --git a/backend/internal/service/gemini_upstream_transport_error.go b/backend/internal/service/gemini_upstream_transport_error.go index ea3f7ebc66..27eaec57fb 100644 --- a/backend/internal/service/gemini_upstream_transport_error.go +++ b/backend/internal/service/gemini_upstream_transport_error.go @@ -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{ diff --git a/backend/internal/service/openai_upstream_transport_error.go b/backend/internal/service/openai_upstream_transport_error.go index d72685b3a9..8c56f000da 100644 --- a/backend/internal/service/openai_upstream_transport_error.go +++ b/backend/internal/service/openai_upstream_transport_error.go @@ -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{