mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-19 16:17:56 +00:00
fix(gemini): 传输层错误转 failover,不再同账号退避重试
上游 Do 返回非 HTTP 错误(代理/DNS/TCP/TLS/响应头超时)时,Gemini 三条转发路径 (ForwardNative / Forward / ForwardAsChatCompletions)原先在同一账号上重试 5 次, 退避 1+2+4+8≈15s(time.Sleep 不感知 ctx),耗尽后直接向客户端写 502 且不换号: 单账号链路故障期间 TTFT 被整段拖高,同组健康账号闲置。 对齐 Anthropic/Bedrock 侧 handleUpstreamTransportError 的既有语义: - 传输层错误单次尝试后即返回 *UpstreamFailoverError(502),由 handler failover 循环换号,耗尽后才按端点格式渲染错误;service 不再写响应 - 持久性故障(connection refused/no route/DNS not found/代理认证失败)额外临时 摘除账号 10 分钟;瞬时故障(EOF/reset/超时)只换号不摘号 - context.Canceled 保持原样返回:不换号也不摘账号 - countTokens 传输失败时立即本地估算兜底,不换号 - 摘号实现抽为 tempUnscheduleAccountForTransportError 供两侧共用,日志事件名不变 HTTP 状态码(429/5xx)的服务内重试环未改动。
This commit is contained in:
feeeei
@@ -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),
|
||||
|
||||
@@ -125,24 +125,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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
// 错误策略优先：匹配则跳过重试直接处理。
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
@@ -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(), "换号场景不应写客户端响应")
|
||||
}
|
||||
Reference in New Issue
Block a user