diff --git a/backend/internal/service/gateway_upstream_transport_error.go b/backend/internal/service/gateway_upstream_transport_error.go index 66763b1199..7781d5cedd 100644 --- a/backend/internal/service/gateway_upstream_transport_error.go +++ b/backend/internal/service/gateway_upstream_transport_error.go @@ -81,7 +81,17 @@ func (s *GatewayService) handleUpstreamTransportError(ctx context.Context, c *gi // - "gateway.account_temp_unschedule_transport_failed" — DB write attempted // but returned an error (the account remains schedulable). func (s *GatewayService) tempUnscheduleTransportError(ctx context.Context, account *Account, safeErr string) { - if s == nil || account == nil || s.accountRepo == nil { + if s == nil { + return + } + tempUnscheduleAccountForTransportError(ctx, s.accountRepo, account, safeErr) +} + +// tempUnscheduleAccountForTransportError is the repo-level implementation +// shared by every forward path whose scheduler reads the persisted +// temp-unschedulable state (Anthropic/Bedrock and Gemini). +func tempUnscheduleAccountForTransportError(ctx context.Context, repo AccountRepository, account *Account, safeErr string) { + if account == nil || repo == nil { return } until := time.Now().Add(gatewayTransportErrorTempUnschedDuration) @@ -89,7 +99,7 @@ func (s *GatewayService) tempUnscheduleTransportError(ctx context.Context, accou bgCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), openAIAccountStateUpdateTimeout) defer cancel() - if err := s.accountRepo.SetTempUnschedulable(bgCtx, account.ID, until, reason); err != nil { + if err := repo.SetTempUnschedulable(bgCtx, account.ID, until, reason); err != nil { logger.L().With(zap.String("component", "service.gateway")).Warn( "gateway.account_temp_unschedule_transport_failed", zap.Int64("account_id", account.ID), diff --git a/backend/internal/service/gemini_chat_completions_compat_service.go b/backend/internal/service/gemini_chat_completions_compat_service.go index 6c5418516a..a98c94cf93 100644 --- a/backend/internal/service/gemini_chat_completions_compat_service.go +++ b/backend/internal/service/gemini_chat_completions_compat_service.go @@ -124,24 +124,7 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions( resp, err = s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency) if err != nil { - safeErr := sanitizeUpstreamErrorMessage(err.Error()) - appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ - ProxyID: opsUpstreamProxyID(account), - ProxyName: opsUpstreamProxyName(account), - Platform: account.Platform, - AccountID: account.ID, - AccountName: account.Name, - UpstreamStatusCode: 0, - Kind: "request_error", - Message: safeErr, - }) - if attempt < geminiMaxRetries { - logger.LegacyPrintf("service.gemini_chat_completions", "Gemini account %d: upstream request failed, retry %d/%d: %v", account.ID, attempt, geminiMaxRetries, err) - sleepGeminiBackoff(attempt) - continue - } - setOpsUpstreamError(c, 0, safeErr, "") - return nil, s.writeChatCompletionsError(c, http.StatusBadGateway, "upstream_error", "Upstream request failed after retries: "+safeErr) + return nil, s.handleUpstreamTransportError(ctx, c, account, err) } if matched, rebuilt := s.checkErrorPolicyInLoop(ctx, account, resp, mappedModel); matched { diff --git a/backend/internal/service/gemini_messages_compat_service.go b/backend/internal/service/gemini_messages_compat_service.go index 0b7c154fec..28de64e77c 100644 --- a/backend/internal/service/gemini_messages_compat_service.go +++ b/backend/internal/service/gemini_messages_compat_service.go @@ -810,24 +810,7 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex resp, err = s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency) if err != nil { - safeErr := sanitizeUpstreamErrorMessage(err.Error()) - appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ - ProxyID: opsUpstreamProxyID(account), - ProxyName: opsUpstreamProxyName(account), - Platform: account.Platform, - AccountID: account.ID, - AccountName: account.Name, - UpstreamStatusCode: 0, - Kind: "request_error", - Message: safeErr, - }) - if attempt < geminiMaxRetries { - logger.LegacyPrintf("service.gemini_messages_compat", "Gemini account %d: upstream request failed, retry %d/%d: %v", account.ID, attempt, geminiMaxRetries, err) - sleepGeminiBackoff(attempt) - continue - } - setOpsUpstreamError(c, 0, safeErr, "") - return nil, s.writeClaudeError(c, http.StatusBadGateway, "upstream_error", "Upstream request failed after retries: "+safeErr) + return nil, s.handleUpstreamTransportError(ctx, c, account, err) } // Special-case: signature/thought_signature validation errors are not transient, but may be fixed by @@ -1362,23 +1345,10 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin. resp, err = s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency) if err != nil { - safeErr := sanitizeUpstreamErrorMessage(err.Error()) - appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ - ProxyID: opsUpstreamProxyID(account), - ProxyName: opsUpstreamProxyName(account), - Platform: account.Platform, - AccountID: account.ID, - AccountName: account.Name, - UpstreamStatusCode: 0, - Kind: "request_error", - Message: safeErr, - }) - if attempt < geminiMaxRetries { - logger.LegacyPrintf("service.gemini_messages_compat", "Gemini account %d: upstream request failed, retry %d/%d: %v", account.ID, attempt, geminiMaxRetries, err) - sleepGeminiBackoff(attempt) - continue - } - if action == "countTokens" { + transportErr := s.handleUpstreamTransportError(ctx, c, account, err) + // countTokens 不因上游链路故障而失败：本地估算兜底，不换号。 + var failoverErr *UpstreamFailoverError + if action == "countTokens" && errors.As(transportErr, &failoverErr) { estimated := estimateGeminiCountTokens(body) c.JSON(http.StatusOK, map[string]any{"totalTokens": estimated}) return &ForwardResult{ @@ -1391,8 +1361,7 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin. FirstTokenMs: nil, }, nil } - setOpsUpstreamError(c, 0, safeErr, "") - return nil, s.writeGoogleError(c, http.StatusBadGateway, "Upstream request failed after retries: "+safeErr) + return nil, transportErr } // 错误策略优先：匹配则跳过重试直接处理。 diff --git a/backend/internal/service/gemini_upstream_transport_error.go b/backend/internal/service/gemini_upstream_transport_error.go new file mode 100644 index 0000000000..ea3f7ebc66 --- /dev/null +++ b/backend/internal/service/gemini_upstream_transport_error.go @@ -0,0 +1,52 @@ +package service + +import ( + "context" + "errors" + "net/http" + + "github.com/gin-gonic/gin" +) + +// geminiTransportFailoverBody 是传输层失败时挂在 failover 错误上的 Google 格式错误体， +// 供 handler 在换号耗尽后做透传规则匹配与错误信息提取。 +var geminiTransportFailoverBody = []byte(`{"error":{"code":502,"message":"Upstream request failed","status":"INTERNAL"}}`) + +// handleUpstreamTransportError 处理 Gemini 三条转发路径（原生 / Claude 兼容 / +// Chat Completions 兼容）上的传输层失败：Do 返回非 HTTP 错误（代理 / DNS / TCP / TLS / +// 响应头超时），没有拿到任何上游状态码。 +// +// - 记录 Ops 错误事件（status 0，kind=request_error）； +// - 持久性故障（代理凭据失效、端点拒绝连接、DNS/路由不可达）临时摘除账号； +// 瞬时故障（EOF / reset / 超时）账号保持可调度； +// - 客户端已断开（context.Canceled）原样返回：不换号、不摘号； +// - 其余一律返回 *UpstreamFailoverError，由 handler 的 failover 循环换号。 +// +// 本函数不写响应：响应归 handler 所有（换号，或耗尽后按端点格式渲染错误）。 +func (s *GeminiMessagesCompatService) handleUpstreamTransportError(ctx context.Context, c *gin.Context, account *Account, err error) error { + safeErr := sanitizeUpstreamErrorMessage(err.Error()) + setOpsUpstreamError(c, 0, safeErr, "") + event := OpsUpstreamErrorEvent{ + Platform: account.Platform, + AccountID: account.ID, + AccountName: account.Name, + UpstreamStatusCode: 0, + Kind: "request_error", + Message: safeErr, + } + event.ProxyID, event.ProxyName = opsUpstreamProxyAttribution(account) + appendOpsUpstreamError(c, event) + + if errors.Is(err, context.Canceled) || (errors.Is(err, context.DeadlineExceeded) && errors.Is(ctx.Err(), context.DeadlineExceeded)) { + return err + } + + if classifyUpstreamTransportError(err).Persistent { + tempUnscheduleAccountForTransportError(ctx, s.accountRepo, account, safeErr) + } + + return &UpstreamFailoverError{ + StatusCode: http.StatusBadGateway, + ResponseBody: geminiTransportFailoverBody, + } +} diff --git a/backend/internal/service/gemini_upstream_transport_error_test.go b/backend/internal/service/gemini_upstream_transport_error_test.go new file mode 100644 index 0000000000..03e5c24983 --- /dev/null +++ b/backend/internal/service/gemini_upstream_transport_error_test.go @@ -0,0 +1,159 @@ +//go:build unit + +package service + +import ( + "bytes" + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +// --------------------------------------------------------------------------- +// Gemini 传输层失败（Do 返回非 HTTP 错误）的契约，三条转发路径一致： +// - 不在同账号上重试，单次尝试后立即返回 *UpstreamFailoverError(502) 交给 handler 换号； +// - service 不写客户端响应； +// - 瞬时故障不摘号，持久性故障临时摘号； +// - 客户端断开原样返回，不换号、不摘号； +// - countTokens 以本地估算兜底。 +// --------------------------------------------------------------------------- + +const geminiTransportTestNativeBody = `{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}` + +func newGeminiTransportErrorService(upstreamErr error) (*GeminiMessagesCompatService, *geminiCompatHTTPUpstreamStub, *transportTempUnschedRepoStub) { + httpStub := &geminiCompatHTTPUpstreamStub{err: upstreamErr} + repo := &transportTempUnschedRepoStub{} + svc := &GeminiMessagesCompatService{ + accountRepo: repo, + httpUpstream: httpStub, + cfg: &config.Config{}, + } + return svc, httpStub, repo +} + +func requireGeminiTransportFailover(t *testing.T, err error) *UpstreamFailoverError { + t.Helper() + var failoverErr *UpstreamFailoverError + require.True(t, errors.As(err, &failoverErr), "传输层错误应换号，got %T: %v", err, err) + require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode) + require.Equal(t, string(geminiTransportFailoverBody), string(failoverErr.ResponseBody)) + require.True(t, failoverErr.ShouldRetryNextAccount()) + require.False(t, failoverErr.RetryableOnSameAccount, "传输层错误不做同账号重试") + return failoverErr +} + +func TestGeminiForwardNative_TransientTransportErrorFailsOverWithoutRetry(t *testing.T) { + gin.SetMode(gin.TestMode) + svc, httpStub, repo := newGeminiTransportErrorService(errors.New(`Post "https://upstream/v1beta/models/gemini-2.5-flash:generateContent": EOF`)) + c, rec := newGeminiNativeTestContext(t) + + start := time.Now() + result, err := svc.ForwardNative(context.Background(), c, geminiPoolModeAPIKeyAccount(), + "gemini-2.5-flash", "generateContent", false, []byte(geminiTransportTestNativeBody)) + + require.Nil(t, result) + requireGeminiTransportFailover(t, err) + require.Equal(t, 1, httpStub.calls, "不应在同账号上重试") + require.Less(t, time.Since(start), geminiRetryBaseDelay, "不应进入退避等待") + require.Zero(t, repo.calls, "瞬时故障不摘号") + require.False(t, c.Writer.Written(), "换号场景不应写客户端响应") + require.Zero(t, rec.Body.Len()) + + raw, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok, "应记录 ops 上游错误事件") + events := raw.([]*OpsUpstreamErrorEvent) + require.Len(t, events, 1) + require.Equal(t, "request_error", events[0].Kind) + require.Equal(t, int64(700), events[0].AccountID) + require.Zero(t, events[0].UpstreamStatusCode) +} + +func TestGeminiForwardNative_PersistentTransportErrorEvictsAccount(t *testing.T) { + gin.SetMode(gin.TestMode) + svc, httpStub, repo := newGeminiTransportErrorService(errors.New(`dial tcp 1.2.3.4:443: connect: connection refused`)) + c, _ := newGeminiNativeTestContext(t) + account := geminiPoolModeAPIKeyAccount() + + before := time.Now() + _, err := svc.ForwardNative(context.Background(), c, account, + "gemini-2.5-flash", "streamGenerateContent", true, []byte(geminiTransportTestNativeBody)) + + requireGeminiTransportFailover(t, err) + require.Equal(t, 1, httpStub.calls) + require.Equal(t, 1, repo.calls, "持久性故障应临时摘号") + require.Equal(t, account.ID, repo.lastID) + require.WithinDuration(t, before.Add(gatewayTransportErrorTempUnschedDuration), repo.lastUntil, 5*time.Second) + require.False(t, c.Writer.Written()) +} + +func TestGeminiForwardNative_ClientCanceledReturnsOriginalError(t *testing.T) { + gin.SetMode(gin.TestMode) + svc, httpStub, repo := newGeminiTransportErrorService(context.Canceled) + c, _ := newGeminiNativeTestContext(t) + + _, err := svc.ForwardNative(context.Background(), c, geminiPoolModeAPIKeyAccount(), + "gemini-2.5-flash", "generateContent", false, []byte(geminiTransportTestNativeBody)) + + require.ErrorIs(t, err, context.Canceled) + var failoverErr *UpstreamFailoverError + require.False(t, errors.As(err, &failoverErr), "客户端断开不应换号") + require.Equal(t, 1, httpStub.calls) + require.Zero(t, repo.calls, "客户端断开不摘号") + require.False(t, c.Writer.Written()) +} + +func TestGeminiForwardNative_CountTokensTransportErrorFallsBackToEstimate(t *testing.T) { + gin.SetMode(gin.TestMode) + svc, httpStub, _ := newGeminiTransportErrorService(errors.New("EOF")) + c, rec := newGeminiNativeTestContext(t) + + result, err := svc.ForwardNative(context.Background(), c, geminiPoolModeAPIKeyAccount(), + "gemini-2.5-flash", "countTokens", false, []byte(geminiTransportTestNativeBody)) + + require.NoError(t, err) + require.NotNil(t, result) + require.Equal(t, 1, httpStub.calls) + require.Equal(t, http.StatusOK, rec.Code) + require.Contains(t, rec.Body.String(), `"totalTokens"`) +} + +func TestGeminiForward_TransportErrorFailsOverWithoutRetry(t *testing.T) { + gin.SetMode(gin.TestMode) + svc, httpStub, repo := newGeminiTransportErrorService(errors.New("read tcp 10.0.0.1:1234->1.2.3.4:443: read: connection reset by peer")) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + body := []byte(`{"model":"gemini-2.5-flash","max_tokens":16,"messages":[{"role":"user","content":"hello"}]}`) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + + result, err := svc.Forward(context.Background(), c, geminiPoolModeAPIKeyAccount(), body) + + require.Nil(t, result) + requireGeminiTransportFailover(t, err) + require.Equal(t, 1, httpStub.calls, "不应在同账号上重试") + require.Zero(t, repo.calls) + require.False(t, c.Writer.Written(), "换号场景不应写客户端响应") +} + +func TestGeminiForwardAsChatCompletions_TransportErrorFailsOverWithoutRetry(t *testing.T) { + gin.SetMode(gin.TestMode) + svc, httpStub, repo := newGeminiTransportErrorService(errors.New("EOF")) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + body := []byte(`{"model":"gemini-2.5-flash","messages":[{"role":"user","content":"hi"}]}`) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + + result, err := svc.ForwardAsChatCompletions(context.Background(), c, geminiPoolModeAPIKeyAccount(), body) + + require.Nil(t, result) + requireGeminiTransportFailover(t, err) + require.Equal(t, 1, httpStub.calls, "不应在同账号上重试") + require.Zero(t, repo.calls) + require.False(t, c.Writer.Written(), "换号场景不应写客户端响应") +}