diff --git a/backend/internal/handler/failover_loop.go b/backend/internal/handler/failover_loop.go index 15d2966299..7ce1d03a8c 100644 --- a/backend/internal/handler/failover_loop.go +++ b/backend/internal/handler/failover_loop.go @@ -5,6 +5,8 @@ import ( "net/http" "time" + "github.com/gin-gonic/gin" + "github.com/Wei-Shaw/sub2api/internal/pkg/logger" "github.com/Wei-Shaw/sub2api/internal/service" "go.uber.org/zap" @@ -71,6 +73,8 @@ func (s *FailoverState) HandleFailoverError( retryLimit int, failoverErr *service.UpstreamFailoverError, ) FailoverAction { + // 客户端已断开：failover 只会用已取消的 context 重新选号并必然失败， + // 不应再被当成账号耗尽处理（误报 502）。 if ctx != nil && ctx.Err() != nil { return FailoverCanceled } @@ -141,6 +145,12 @@ func (s *FailoverState) HandleFailoverError( // 返回 FailoverExhausted 时，调用方应返回错误响应。 // 返回 FailoverCanceled 时，调用方应直接 return。 func (s *FailoverState) HandleSelectionExhausted(ctx context.Context) FailoverAction { + // 客户端已断开时选号失败是 context canceled 的必然结果， + // 不代表账号耗尽，直接按取消终止。 + if ctx.Err() != nil { + return FailoverCanceled + } + if s.LastFailoverErr != nil && s.LastFailoverErr.StatusCode == http.StatusServiceUnavailable && s.SwitchCount <= s.MaxSwitches { @@ -169,6 +179,28 @@ func needForceCacheBilling(hasBoundSession bool, failoverErr *service.UpstreamFa return hasBoundSession || (failoverErr != nil && failoverErr.ForceCacheBilling) } +// failoverClientGone 判断下游客户端是否已断开（请求 context 已取消）。 +// 客户端断开后 failover 必须静默终止：用已取消的 context 重新选号只会得到 +// context.Canceled，并被误报成账号耗尽（通用 502）；上游 detach 的在途请求 +// 照常完成计费，但不再为无人接收的响应启动新的上游尝试。 +// 响应尚未提交时把状态码标记为 499（client closed request），供访问日志归类。 +func failoverClientGone(c *gin.Context) bool { + if c == nil || c.Request == nil || c.Request.Context().Err() == nil { + return false + } + // 先停 compact 心跳（接管 ResponseWriter，建立 happens-before），与 + // handleStreamingAwareError/errorResponse 等终结路径对齐，避免心跳 + // goroutine 与下面的状态标记并发触碰同一 writer。心跳已提交 200 时 + // 状态码已固化，不再标 499。 + if service.StopOpenAICompactSSEKeepaliveCommitted(c) { + return true + } + if !c.Writer.Written() { + c.Status(statusClientClosedRequest) + } + return true +} + // sleepWithContext 等待指定时长，返回 false 表示 context 已取消。 func sleepWithContext(ctx context.Context, d time.Duration) bool { if d <= 0 { diff --git a/backend/internal/handler/failover_loop_test.go b/backend/internal/handler/failover_loop_test.go index bcfc91e07a..8946eb5236 100644 --- a/backend/internal/handler/failover_loop_test.go +++ b/backend/internal/handler/failover_loop_test.go @@ -3,11 +3,13 @@ package handler import ( "context" "net/http" + "net/http/httptest" "testing" "time" "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" ) @@ -489,7 +491,28 @@ func TestHandleFailoverError_ContextCanceled(t *testing.T) { err := newTestFailoverErr(400, true, false) ctx, cancel := context.WithCancel(context.Background()) - cancel() // 立即取消 + go func() { + time.Sleep(30 * time.Millisecond) + cancel() // 通过入口检查后、sleep 期间取消 + }() + + start := time.Now() + action := fs.HandleFailoverError(ctx, mock, 100, "openai", maxSameAccountRetries, err) + elapsed := time.Since(start) + + require.Equal(t, FailoverCanceled, action) + require.Less(t, elapsed, 400*time.Millisecond, "sleep 应被取消打断") + // 进入重试分支后才取消：重试计数已递增 + require.Equal(t, 1, fs.SameAccountRetryCount[100]) + }) + + t.Run("入口即已取消_不改动任何failover状态", func(t *testing.T) { + mock := &mockTempUnscheduler{} + fs := NewFailoverState(3, false) + err := newTestFailoverErr(520, false, false) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // 调用前客户端已断开 start := time.Now() action := fs.HandleFailoverError(ctx, mock, 100, "openai", maxSameAccountRetries, err) @@ -497,8 +520,12 @@ func TestHandleFailoverError_ContextCanceled(t *testing.T) { require.Equal(t, FailoverCanceled, action) require.Less(t, elapsed, 100*time.Millisecond, "应立即返回") - // 入口已取消时不得改变任何重试状态。 - require.Zero(t, fs.SameAccountRetryCount[100]) + // 入口已取消时不得改变任何 failover 状态。 + require.Equal(t, 0, fs.SwitchCount, "取消的请求不应计入切换") + require.Equal(t, 0, fs.SameAccountRetryCount[100], "取消的请求不应改动重试计数") + require.NotContains(t, fs.FailedAccountIDs, int64(100)) + require.Nil(t, fs.LastFailoverErr) + require.Empty(t, mock.calls, "不应触发 TempUnschedule") }) t.Run("Antigravity延迟期间context取消", func(t *testing.T) { @@ -800,6 +827,29 @@ func TestHandleSelectionExhausted(t *testing.T) { require.Less(t, elapsed, 100*time.Millisecond, "应立即返回") }) + t.Run("context已取消_非503也返回Canceled而非Exhausted", func(t *testing.T) { + // #4257 核心场景：客户端断开后选号失败源于 context canceled， + // 不应被当成账号耗尽转成 502。 + fs := NewFailoverState(3, false) + fs.LastFailoverErr = newTestFailoverErr(520, false, false) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + action := fs.HandleSelectionExhausted(ctx) + require.Equal(t, FailoverCanceled, action) + }) + + t.Run("context已取消_无LastFailoverErr也返回Canceled", func(t *testing.T) { + fs := NewFailoverState(3, false) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + action := fs.HandleSelectionExhausted(ctx) + require.Equal(t, FailoverCanceled, action) + }) + t.Run("503且SwitchCount等于MaxSwitches_仍可重试", func(t *testing.T) { fs := NewFailoverState(2, false) fs.LastFailoverErr = newTestFailoverErr(503, false, false) @@ -809,3 +859,47 @@ func TestHandleSelectionExhausted(t *testing.T) { require.Equal(t, FailoverContinue, action) }) } + +// --------------------------------------------------------------------------- +// failoverClientGone 测试 +// --------------------------------------------------------------------------- + +func TestFailoverClientGone(t *testing.T) { + gin.SetMode(gin.TestMode) + + t.Run("活跃请求返回false", func(t *testing.T) { + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + + require.False(t, failoverClientGone(c)) + require.Equal(t, http.StatusOK, c.Writer.Status(), "不应改动状态码") + }) + + t.Run("客户端已断开_返回true并标记499", func(t *testing.T) { + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil).WithContext(ctx) + + require.True(t, failoverClientGone(c)) + require.Equal(t, statusClientClosedRequest, c.Writer.Status()) + }) + + t.Run("响应已提交_不改状态码", func(t *testing.T) { + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil).WithContext(ctx) + c.String(http.StatusOK, "partial") + + require.True(t, failoverClientGone(c)) + require.Equal(t, http.StatusOK, c.Writer.Status(), "已提交的状态码不应被覆盖") + }) + + t.Run("nil安全", func(t *testing.T) { + require.False(t, failoverClientGone(nil)) + }) +} diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index fb07b098d1..1c51713656 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -327,6 +327,7 @@ func (h *GatewayHandler) Messages(c *gin.Context) { c.Request = c.Request.WithContext(ctx) continue case FailoverCanceled: + failoverClientGone(c) return default: // FailoverExhausted if fs.LastFailoverErr != nil { @@ -456,6 +457,7 @@ func (h *GatewayHandler) Messages(c *gin.Context) { h.handleFailoverExhausted(c, fs.LastFailoverErr, service.PlatformGemini, streamStarted) return case FailoverCanceled: + failoverClientGone(c) return } } @@ -613,6 +615,7 @@ func (h *GatewayHandler) Messages(c *gin.Context) { c.Request = c.Request.WithContext(ctx) continue case FailoverCanceled: + failoverClientGone(c) return default: // FailoverExhausted if fs.LastFailoverErr != nil { @@ -876,6 +879,7 @@ func (h *GatewayHandler) Messages(c *gin.Context) { h.handleFailoverExhausted(c, fs.LastFailoverErr, account.Platform, streamStarted) return case FailoverCanceled: + failoverClientGone(c) return } } diff --git a/backend/internal/handler/gateway_handler_chat_completions.go b/backend/internal/handler/gateway_handler_chat_completions.go index d7893b6cd4..cc9ff14282 100644 --- a/backend/internal/handler/gateway_handler_chat_completions.go +++ b/backend/internal/handler/gateway_handler_chat_completions.go @@ -181,6 +181,7 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) { case FailoverContinue: continue case FailoverCanceled: + failoverClientGone(c) return default: if fs.LastFailoverErr != nil { @@ -265,6 +266,7 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) { h.handleCCFailoverExhausted(c, fs.LastFailoverErr, streamStarted) return case FailoverCanceled: + failoverClientGone(c) return } } diff --git a/backend/internal/handler/gateway_handler_responses.go b/backend/internal/handler/gateway_handler_responses.go index 844de47669..a5060b7263 100644 --- a/backend/internal/handler/gateway_handler_responses.go +++ b/backend/internal/handler/gateway_handler_responses.go @@ -179,6 +179,7 @@ func (h *GatewayHandler) Responses(c *gin.Context) { case FailoverContinue: continue case FailoverCanceled: + failoverClientGone(c) return default: if fs.LastFailoverErr != nil { @@ -244,6 +245,7 @@ func (h *GatewayHandler) Responses(c *gin.Context) { h.handleResponsesFailoverExhausted(c, fs.LastFailoverErr, streamStarted) return case FailoverCanceled: + failoverClientGone(c) return } } diff --git a/backend/internal/handler/gemini_v1beta_handler.go b/backend/internal/handler/gemini_v1beta_handler.go index b1653c6b3f..c595abfe55 100644 --- a/backend/internal/handler/gemini_v1beta_handler.go +++ b/backend/internal/handler/gemini_v1beta_handler.go @@ -368,6 +368,7 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) { c.Request = c.Request.WithContext(ctx) continue case FailoverCanceled: + failoverClientGone(c) return default: // FailoverExhausted h.handleGeminiFailoverExhausted(c, fs.LastFailoverErr) @@ -490,6 +491,7 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) { h.handleGeminiFailoverExhausted(c, fs.LastFailoverErr) return case FailoverCanceled: + failoverClientGone(c) return } } diff --git a/backend/internal/handler/grok_media.go b/backend/internal/handler/grok_media.go index 6b10f7bcbe..01928028e6 100644 --- a/backend/internal/handler/grok_media.go +++ b/backend/internal/handler/grok_media.go @@ -175,7 +175,7 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service. routingStart := time.Now() for { - if requestCtx.Err() != nil { + if failoverClientGone(c) { return } selection, scheduleDecision, err := h.gatewayService.SelectAccountWithSchedulerForCapability( @@ -192,6 +192,10 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service. service.PlatformGrok, ) if err != nil { + if failoverClientGone(c) { + reqLog.Info("grok_media.account_select_aborted_client_disconnected", zap.Error(err)) + return + } reqLog.Warn("grok_media.account_select_failed", zap.Error(err), zap.Int("excluded_account_count", len(failedAccountIDs)), @@ -261,7 +265,11 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service. if err != nil { var failoverErr *service.UpstreamFailoverError if errors.As(err, &failoverErr) { - if requestCtx.Err() != nil { + if failoverClientGone(c) { + reqLog.Info("grok_media.failover_aborted_client_disconnected", + zap.Int64("account_id", account.ID), + zap.Int("upstream_status", failoverErr.StatusCode), + ) return } if failoverErr.ShouldReportAccountScheduleFailure() { diff --git a/backend/internal/handler/openai_alpha_search.go b/backend/internal/handler/openai_alpha_search.go index 4f5c07d8c5..a1d185a1cb 100644 --- a/backend/internal/handler/openai_alpha_search.go +++ b/backend/internal/handler/openai_alpha_search.go @@ -124,6 +124,10 @@ func (h *OpenAIGatewayHandler) AlphaSearch(c *gin.Context) { service.PlatformOpenAI, ) if err != nil || selection == nil || selection.Account == nil { + if failoverClientGone(c) { + reqLog.Info("openai_alpha_search.account_select_aborted_client_disconnected", zap.Error(err)) + return + } if len(failedAccountIDs) == 0 { cls := classifyNoAccountErrorFromGin(c, h.gatewayService, apiKey, requestedModel, requestedModel, service.PlatformOpenAI) if !cls.ModelNotFound { @@ -181,6 +185,13 @@ func (h *OpenAIGatewayHandler) AlphaSearch(c *gin.Context) { h.handleFailoverExhausted(c, failoverErr, true) return } + if failoverClientGone(c) { + reqLog.Info("openai_alpha_search.failover_aborted_client_disconnected", + zap.Int64("account_id", account.ID), + zap.Int("upstream_status", failoverErr.StatusCode), + ) + return + } h.gatewayService.RecordOpenAIAccountSwitch() failedAccountIDs[account.ID] = struct{}{} lastFailoverErr = failoverErr diff --git a/backend/internal/handler/openai_chat_completions.go b/backend/internal/handler/openai_chat_completions.go index 004aff2bb7..62a2b70b59 100644 --- a/backend/internal/handler/openai_chat_completions.go +++ b/backend/internal/handler/openai_chat_completions.go @@ -136,7 +136,7 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) { var oauth429FailoverState service.OpenAIOAuth429FailoverState for { - if c.Request.Context().Err() != nil { + if failoverClientGone(c) { return } reqLog.Debug("openai_chat_completions.account_selecting", zap.Int("excluded_account_count", len(failedAccountIDs))) @@ -154,6 +154,10 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) { requestPlatform, ) if err != nil { + if failoverClientGone(c) { + reqLog.Info("openai_chat_completions.account_select_aborted_client_disconnected", zap.Error(err)) + return + } reqLog.Warn("openai_chat_completions.account_select_failed", zap.Error(openAICompatibleSelectionErrorForLog(err, requestPlatform)), zap.Int("excluded_account_count", len(failedAccountIDs)), @@ -235,7 +239,11 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) { } else { var failoverErr *service.UpstreamFailoverError if errors.As(err, &failoverErr) { - if c.Request.Context().Err() != nil { + if failoverClientGone(c) { + reqLog.Info("openai_chat_completions.failover_aborted_client_disconnected", + zap.Int64("account_id", account.ID), + zap.Int("upstream_status", failoverErr.StatusCode), + ) return } if c.Writer.Size() != writerSizeBeforeForward { diff --git a/backend/internal/handler/openai_embeddings.go b/backend/internal/handler/openai_embeddings.go index 56d775eb7c..99f869f14c 100644 --- a/backend/internal/handler/openai_embeddings.go +++ b/backend/internal/handler/openai_embeddings.go @@ -121,6 +121,10 @@ func (h *OpenAIGatewayHandler) Embeddings(c *gin.Context) { false, ) if err != nil { + if failoverClientGone(c) { + reqLog.Info("openai_embeddings.account_select_aborted_client_disconnected", zap.Error(err)) + return + } reqLog.Warn("openai_embeddings.account_select_failed", zap.Error(err), zap.Int("excluded_account_count", len(failedAccountIDs)), @@ -189,6 +193,13 @@ func (h *OpenAIGatewayHandler) Embeddings(c *gin.Context) { return } h.gatewayService.ReportOpenAIAccountScheduleResult(account.ID, false, nil) + if failoverClientGone(c) { + reqLog.Info("openai_embeddings.failover_aborted_client_disconnected", + zap.Int64("account_id", account.ID), + zap.Int("upstream_status", failoverErr.StatusCode), + ) + return + } h.gatewayService.RecordOpenAIAccountSwitch() failedAccountIDs[account.ID] = struct{}{} lastFailoverErr = failoverErr diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 33bef42cef..4f01872d4d 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -335,7 +335,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { var oauth429FailoverState service.OpenAIOAuth429FailoverState for { - if c.Request.Context().Err() != nil { + if failoverClientGone(c) { return } // Select account supporting the requested model @@ -354,6 +354,10 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { requestPlatform, ) if err != nil { + if failoverClientGone(c) { + reqLog.Info("openai.account_select_aborted_client_disconnected", zap.Error(err)) + return + } reqLog.Warn("openai.account_select_failed", zap.Error(openAICompatibleSelectionErrorForLog(err, requestPlatform)), zap.Int("excluded_account_count", len(failedAccountIDs)), @@ -447,7 +451,11 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { } else { var failoverErr *service.UpstreamFailoverError if errors.As(err, &failoverErr) { - if c.Request.Context().Err() != nil { + if failoverClientGone(c) { + reqLog.Info("openai.failover_aborted_client_disconnected", + zap.Int64("account_id", account.ID), + zap.Int("upstream_status", failoverErr.StatusCode), + ) return } if service.OpenAICompactKeepaliveAdjustedWrittenSize(c) != writerSizeBeforeForward { @@ -849,7 +857,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { effectiveMappedModel := preferredMappedModel for { - if c.Request.Context().Err() != nil { + if failoverClientGone(c) { return } currentRoutingModel := routingModel @@ -871,6 +879,10 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { requestPlatform, ) if err != nil { + if failoverClientGone(c) { + reqLog.Info("openai_messages.account_select_aborted_client_disconnected", zap.Error(err)) + return + } reqLog.Warn("openai_messages.account_select_failed", zap.Error(openAICompatibleSelectionErrorForLog(err, requestPlatform)), zap.Int("excluded_account_count", len(failedAccountIDs)), @@ -952,7 +964,11 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { } else { var failoverErr *service.UpstreamFailoverError if errors.As(err, &failoverErr) { - if c.Request.Context().Err() != nil { + if failoverClientGone(c) { + reqLog.Info("openai_messages.failover_aborted_client_disconnected", + zap.Int64("account_id", account.ID), + zap.Int("upstream_status", failoverErr.StatusCode), + ) return } if c.Writer.Size() != writerSizeBeforeForward { diff --git a/backend/internal/handler/openai_images.go b/backend/internal/handler/openai_images.go index d18af28966..f5709225d1 100644 --- a/backend/internal/handler/openai_images.go +++ b/backend/internal/handler/openai_images.go @@ -158,6 +158,10 @@ func (h *OpenAIGatewayHandler) Images(c *gin.Context) { parsed.RequiredCapability, ) if err != nil { + if failoverClientGone(c) { + reqLog.Info("openai.images.account_select_aborted_client_disconnected", zap.Error(err)) + return + } reqLog.Warn("openai.images.account_select_failed", zap.Error(err), zap.Int("excluded_account_count", len(failedAccountIDs)), @@ -274,6 +278,13 @@ func (h *OpenAIGatewayHandler) Images(c *gin.Context) { h.handleFailoverExhausted(c, failoverErr, true) return } + if failoverClientGone(c) { + reqLog.Info("openai.images.failover_aborted_client_disconnected", + zap.Int64("account_id", account.ID), + zap.Int("upstream_status", failoverErr.StatusCode), + ) + return + } if failoverErr.RetryableOnSameAccount { retryLimit := account.GetPoolModeRetryCount() if sameAccountRetryCount[account.ID] < retryLimit { diff --git a/backend/internal/handler/openai_responses_failover_cancel_test.go b/backend/internal/handler/openai_responses_failover_cancel_test.go new file mode 100644 index 0000000000..0d806e19d0 --- /dev/null +++ b/backend/internal/handler/openai_responses_failover_cancel_test.go @@ -0,0 +1,194 @@ +//go:build unit + +package handler + +import ( + "bytes" + "context" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/config" + middleware2 "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" + "github.com/tidwall/gjson" +) + +// openAIResponsesFailoverCancelUpstream 固定返回 HTTP 520，可在首次上游调用时 +// 触发回调（用于模拟“上游在途期间客户端断开”）。 +type openAIResponsesFailoverCancelUpstream struct { + service.HTTPUpstream + mu sync.Mutex + accountIDs []int64 + onFirstDo func() +} + +func (u *openAIResponsesFailoverCancelUpstream) Do(_ *http.Request, _ string, accountID int64, _ int) (*http.Response, error) { + u.mu.Lock() + u.accountIDs = append(u.accountIDs, accountID) + first := len(u.accountIDs) == 1 + u.mu.Unlock() + if first && u.onFirstDo != nil { + u.onFirstDo() + } + return &http.Response{ + StatusCode: 520, + Header: http.Header{"Content-Type": []string{"text/html"}}, + Body: io.NopCloser(bytes.NewBufferString("520: unknown error")), + }, nil +} + +func (u *openAIResponsesFailoverCancelUpstream) calls() []int64 { + u.mu.Lock() + defer u.mu.Unlock() + return append([]int64(nil), u.accountIDs...) +} + +func newOpenAIResponsesFailoverTestHandler(t *testing.T, upstream service.HTTPUpstream) *OpenAIGatewayHandler { + t.Helper() + accounts := []service.Account{ + { + ID: 1, + Name: "responses-account-1", + Platform: service.PlatformOpenAI, + Type: service.AccountTypeOAuth, + Status: service.StatusActive, + Schedulable: true, + Concurrency: 0, + Priority: 0, + Credentials: map[string]any{"access_token": "token-1"}, + }, + { + ID: 2, + Name: "responses-account-2", + Platform: service.PlatformOpenAI, + Type: service.AccountTypeOAuth, + Status: service.StatusActive, + Schedulable: true, + Concurrency: 0, + Priority: 1, + Credentials: map[string]any{"access_token": "token-2"}, + }, + } + accountRepo := openAIImagesFailoverAccountRepo{accounts: accounts} + cfg := &config.Config{RunMode: config.RunModeSimple} + gatewayService := service.NewOpenAIGatewayService( + accountRepo, + nil, + nil, + nil, + nil, + nil, + nil, + cfg, + nil, + nil, + nil, + nil, + nil, + upstream, + nil, + nil, + nil, + nil, + nil, + nil, + nil, + nil, + ) + billingService := service.NewBillingCacheService(nil, nil, nil, nil, nil, nil, cfg, nil) + t.Cleanup(billingService.Stop) + concurrencyService := service.NewConcurrencyService(nil) + handler := NewOpenAIGatewayHandler( + gatewayService, + concurrencyService, + billingService, + service.NewAPIKeyService(nil, nil, nil, nil, nil, nil, cfg), + nil, + nil, + nil, + nil, + cfg, + ) + handler.maxAccountSwitches = 10 + return handler +} + +func newOpenAIResponsesFailoverTestContext(t *testing.T, ctx context.Context) (*gin.Context, *httptest.ResponseRecorder) { + t.Helper() + groupID := int64(3131) + body := []byte(`{"model":"gpt-5.1","stream":false,"input":"hello"}`) + req := httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + if ctx != nil { + req = req.WithContext(ctx) + } + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = req + c.Set(string(middleware2.ContextKeyAPIKey), &service.APIKey{ + ID: 99, + GroupID: &groupID, + Group: &service.Group{ + ID: groupID, + Platform: service.PlatformOpenAI, + }, + User: &service.User{ID: 100}, + }) + c.Set(string(middleware2.ContextKeyUser), middleware2.AuthSubject{UserID: 100, Concurrency: 0}) + return c, rec +} + +// TestOpenAIGatewayHandlerResponses_FailoverAbortsWhenClientDisconnected 复现 +// #4257：客户端在上游请求在途期间断开，上游随后返回可 failover 的 520。 +// 期望：不再用已取消的 context 重新选号（不触达账号 2）、不把取消误报成 +// 502 账号耗尽、请求按 499 归类。 +func TestOpenAIGatewayHandlerResponses_FailoverAbortsWhenClientDisconnected(t *testing.T) { + gin.SetMode(gin.TestMode) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + upstream := &openAIResponsesFailoverCancelUpstream{onFirstDo: cancel} + handler := newOpenAIResponsesFailoverTestHandler(t, upstream) + c, rec := newOpenAIResponsesFailoverTestContext(t, ctx) + + handler.Responses(c) + + require.Equal(t, []int64{1}, upstream.calls(), "客户端断开后不应再切换到账号 2") + require.Equal(t, statusClientClosedRequest, c.Writer.Status(), "应按 499 归类") + require.Zero(t, rec.Body.Len(), "不应写入 502 错误响应体") + + _, hasFinalUpstreamErr := c.Get(service.OpsUpstreamStatusCodeKey) + require.False(t, hasFinalUpstreamErr, "不应记录 failover 耗尽的上游错误终态") + + // 真实发生过的 520 应保留 failover 事件（service 层在返回 failover 错误前记录） + rawEvents, ok := c.Get(service.OpsUpstreamErrorsKey) + require.True(t, ok) + events, ok := rawEvents.([]*service.OpsUpstreamErrorEvent) + require.True(t, ok) + require.Len(t, events, 1) + require.Equal(t, "failover", events[0].Kind) + require.Equal(t, 520, events[0].UpstreamStatusCode) +} + +// TestOpenAIGatewayHandlerResponses_FailoverContinuesForConnectedClient 回归 +// 守卫：客户端在线时 failover 行为不变——切换到账号 2，两个账号都 520 后按 +// 耗尽返回 502。 +func TestOpenAIGatewayHandlerResponses_FailoverContinuesForConnectedClient(t *testing.T) { + gin.SetMode(gin.TestMode) + + upstream := &openAIResponsesFailoverCancelUpstream{} + handler := newOpenAIResponsesFailoverTestHandler(t, upstream) + c, rec := newOpenAIResponsesFailoverTestContext(t, nil) + + handler.Responses(c) + + require.Equal(t, []int64{1, 2}, upstream.calls(), "在线客户端应正常切换账号") + require.Equal(t, http.StatusBadGateway, rec.Code) + require.Equal(t, "upstream_error", gjson.GetBytes(rec.Body.Bytes(), "error.type").String()) +}