fix(gemini): Vertex 共享容量不足的 429 不冷却账号,改为同号退避重试

- Vertex 按量付费返回的 "Resource exhausted ... try again later"
  (RESOURCE_EXHAUSTED、无错误明细)表示共享资源池当时没有容量,与账号额度
  无关;此前一律冷却账号 1 分钟,冷却期间账号被调度快照剔除,绑定显式缓存
  的请求因无法换号直接 503
- 这类 429 只记日志、账号保持可调度,由请求内同号 1/2/4/8s 退避重试消化,
  重试用尽后未绑定缓存的请求照常换号
- 带 retryDelay、文案重试时长、配额/预留吞吐字样或其他形态的 429 仍按账号
  自身限速冷却;非 Vertex 账号行为不变
This commit is contained in:
feeeei
2026-10-13 23:51:43 +08:00
parent 4d9ac3e482
commit 89ad4c2caa
3 changed files with 237 additions and 14 deletions
@@ -631,30 +631,159 @@ func TestHandleGeminiUpstreamError_VertexRetryInfoUsesParsedDelay(t *testing.T)
require.WithinDuration(t, before.Add(39*time.Second), repo.lastRateLimitReset, 2*time.Second)
}
// Vertex(service_account)429 响应不带任何可解析的重置时间时,走短冷却兜底,
// 不再冷却到 PST 午夜(按量付费没有每日配额,长冷却会导致整组不可用)。
func TestHandleGeminiUpstreamError_VertexFallbackUsesShortCooldown(t *testing.T) {
// Vertex(service_account)共享容量不足的 429 不写账号级限流:账号保持可调度,
// 由请求内的同号退避重试与换号消化。
func TestHandleGeminiUpstreamError_VertexSharedCapacityKeepsAccountSchedulable(t *testing.T) {
tests := []struct {
name string
body string
}{
{
name: "object",
body: `{"error":{"code":429,"message":"Resource exhausted. Please try again later. Please refer to https://cloud.google.com/vertex-ai/generative-ai/docs/error-code-429 for more details.","status":"RESOURCE_EXHAUSTED"}}`,
},
{
name: "array_wrapped",
body: `[{"error":{"code":429,"message":"Resource exhausted. Please try again later.","status":"RESOURCE_EXHAUSTED"}}]`,
},
{
name: "empty_details",
body: `{"error":{"code":429,"message":"Resource exhausted. Please try again later.","status":"RESOURCE_EXHAUSTED","details":[]}}`,
},
{
name: "documented_comma_wording",
body: `{"error":{"code":429,"message":"Resource exhausted, please try again later.","status":"RESOURCE_EXHAUSTED"}}`,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
repo := &rateLimit429AccountRepoStub{}
svc := &GeminiMessagesCompatService{accountRepo: repo}
account := &Account{
ID: 602,
Platform: PlatformGemini,
Type: AccountTypeServiceAccount,
Credentials: map[string]any{
"project_id": "my-vertex-project",
},
}
svc.handleGeminiUpstreamError(context.Background(), account, http.StatusTooManyRequests, http.Header{}, []byte(tt.body))
require.Zero(t, repo.rateLimitCalls)
})
}
}
// Vertex(service_account)项目配额、预留吞吐等账号自身的 429 不带可解析的重置时间时,
// 走短冷却兜底,不冷却到 PST 午夜(按量付费没有每日配额,长冷却会导致整组不可用)。
func TestHandleGeminiUpstreamError_VertexAccountScoped429UsesShortCooldown(t *testing.T) {
tests := []struct {
name string
body string
}{
{
name: "project_quota",
body: `{"error":{"code":429,"message":"Quota exceeded for aiplatform.googleapis.com/generate_content_requests_per_minute_per_project_per_base_model with base model: gemini-2.5-pro. Please submit a quota increase request.","status":"RESOURCE_EXHAUSTED"}}`,
},
{
name: "provisioned_throughput",
body: `{"error":{"code":429,"message":"Too many requests. Exceeded the Provisioned Throughput.","status":"RESOURCE_EXHAUSTED"}}`,
},
{
name: "quota_failure_details",
body: `{"error":{"code":429,"message":"Resource exhausted. Please try again later.","status":"RESOURCE_EXHAUSTED","details":[{"@type":"type.googleapis.com/google.rpc.QuotaFailure","violations":[{"subject":"project"}]}]}}`,
},
{
name: "other_status",
body: `{"error":{"code":429,"message":"Too many requests.","status":"UNAVAILABLE"}}`,
},
{
name: "unknown_wording",
body: `{"error":{"code":429,"message":"Rate limit exceeded for requests per minute per project.","status":"RESOURCE_EXHAUSTED"}}`,
},
{
name: "missing_message",
body: `{"error":{"code":429,"status":"RESOURCE_EXHAUSTED"}}`,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
repo := &rateLimit429AccountRepoStub{}
svc := &GeminiMessagesCompatService{accountRepo: repo}
account := &Account{
ID: 604,
Platform: PlatformGemini,
Type: AccountTypeServiceAccount,
Credentials: map[string]any{
"project_id": "my-vertex-project",
},
}
before := time.Now()
svc.handleGeminiUpstreamError(context.Background(), account, http.StatusTooManyRequests, http.Header{}, []byte(tt.body))
require.Equal(t, 1, repo.rateLimitCalls)
require.Equal(t, int64(604), repo.lastRateLimitID)
require.WithinDuration(t, before.Add(geminiVertexFallbackCooldown), repo.lastRateLimitReset, 2*time.Second)
// 显式守护:不应被冷却超过 15 分钟(更不应到次日 PST 午夜)
require.True(t, repo.lastRateLimitReset.Before(before.Add(15*time.Minute)))
})
}
}
// Vertex 共享容量形态的 429 只对 Vertex 账号豁免冷却:AI Studio 官方 API Key 与 Code Assist OAuth
// 收到同样的响应体仍按各自语义冷却。
func TestHandleGeminiUpstreamError_SharedCapacityBodyStillCoolsNonVertexAccounts(t *testing.T) {
body := []byte(`{"error":{"code":429,"message":"Resource exhausted. Please try again later. Please refer to https://cloud.google.com/vertex-ai/generative-ai/docs/error-code-429 for more details.","status":"RESOURCE_EXHAUSTED"}}`)
tests := []struct {
name string
account *Account
}{
{
name: "ai_studio_api_key",
account: &Account{ID: 605, Platform: PlatformGemini, Type: AccountTypeAPIKey, Credentials: map[string]any{"api_key": "k"}},
},
{
name: "code_assist_oauth",
account: &Account{ID: 606, Platform: PlatformGemini, Type: AccountTypeOAuth, Credentials: map[string]any{
"oauth_type": "code_assist", "project_id": "my-code-assist-project",
}},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
repo := &rateLimit429AccountRepoStub{}
svc := &GeminiMessagesCompatService{accountRepo: repo}
svc.handleGeminiUpstreamError(context.Background(), tt.account, http.StatusTooManyRequests, http.Header{}, body)
require.Equal(t, 1, repo.rateLimitCalls)
require.Equal(t, tt.account.ID, repo.lastRateLimitID)
})
}
}
// Vertex 429 文案里给出了重试时长时按该时长冷却,不归为共享容量不足。
func TestHandleGeminiUpstreamError_VertexMessageRetryHintUsesParsedDelay(t *testing.T) {
repo := &rateLimit429AccountRepoStub{}
svc := &GeminiMessagesCompatService{accountRepo: repo}
account := &Account{
ID: 602,
ID: 607,
Platform: PlatformGemini,
Type: AccountTypeServiceAccount,
Credentials: map[string]any{
"project_id": "my-vertex-project",
},
}
body := []byte(`{"error":{"code":429,"message":"Resource exhausted. Please try again later.","status":"RESOURCE_EXHAUSTED"}}`)
body := []byte(`{"error":{"code":429,"message":"Resource exhausted. Please retry in 5s","status":"RESOURCE_EXHAUSTED"}}`)
before := time.Now()
svc.handleGeminiUpstreamError(context.Background(), account, http.StatusTooManyRequests, http.Header{}, body)
require.Equal(t, 1, repo.rateLimitCalls)
require.Equal(t, int64(602), repo.lastRateLimitID)
require.WithinDuration(t, before.Add(geminiVertexFallbackCooldown), repo.lastRateLimitReset, 2*time.Second)
// 显式守护:不应被冷却超过 15 分钟(更不应到次日 PST 午夜)
require.True(t, repo.lastRateLimitReset.Before(before.Add(15*time.Minute)))
require.Equal(t, int64(607), repo.lastRateLimitID)
require.WithinDuration(t, before.Add(5*time.Second), repo.lastRateLimitReset, 2*time.Second)
}
// API Key(AI Studio)无法解析重置时间时仍冷却到 PST 午夜,行为保持不变。
@@ -45,8 +45,8 @@ const (
geminiRetryInfoTypeURL = "type.googleapis.com/google.rpc.RetryInfo"
// RetryInfo.retryDelay 的上限,防止异常值导致账号长时间不可用
geminiRetryInfoMaxDelay = 15 * time.Minute
// Vertex(service_account,按量付费)429 无法解析重置时间时的兜底冷却:
// 按量付费限速是短窗口 RPM/TPM,没有每日配额,不适用 PST 午夜重置
// Vertex(service_account)账号自身的 429(按量付费的项目配额、预留吞吐超额等)无法解析重置时间时的兜底冷却:
// 这类限速是短窗口 RPM/TPM,没有每日配额,不适用 PST 午夜重置;共享容量不足的 429 不冷却
geminiVertexFallbackCooldown = time.Minute
)
@@ -820,7 +820,7 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex
break
}
if resp.StatusCode == 429 {
// Mark as rate-limited early so concurrent requests avoid this account.
// Apply account-scoped 429 handling early so concurrent requests avoid a rate-limited account.
s.handleGeminiUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody)
}
if attempt < geminiMaxRetries {
@@ -3026,7 +3026,13 @@ func (s *GeminiMessagesCompatService) handleGeminiUpstreamError(ctx context.Cont
logger.LegacyPrintf("service.gemini_messages_compat", "[Gemini 429] Account %d (Google One OAuth, tier=%s, project=%s) rate limited, cooldown=%v", account.ID, tierID, projectID, time.Until(ra).Truncate(time.Second))
}
} else if account.IsVertexServiceAccount() {
// Vertex 按量付费:短窗限速,无每日配额,短冷却后即可恢复
// 共享容量不足描述的是请求当时的上游容量,不是账号额度:账号保持可调度,
// 由请求内的同号退避重试与换号消化。
if isVertexSharedCapacity429(body) {
logger.LegacyPrintf("service.gemini_messages_compat", "[Gemini 429] Account %d (Vertex service account, project=%s) shared capacity exhausted, account stays schedulable", account.ID, projectID)
return
}
// 项目配额、预留吞吐等账号自身的限速是短窗口 RPM/TPM,无每日配额,短冷却后即可恢复
ra = time.Now().Add(geminiVertexFallbackCooldown)
logger.LegacyPrintf("service.gemini_messages_compat", "[Gemini 429] Account %d (Vertex service account, project=%s) rate limited, cooldown=%v", account.ID, projectID, time.Until(ra).Truncate(time.Second))
} else {
@@ -3051,6 +3057,30 @@ func (s *GeminiMessagesCompatService) handleGeminiUpstreamError(ctx context.Cont
account.ID, resetTime, oauthType, tierID)
}
// isVertexSharedCapacity429 判断 Vertex 429 是否为按量付费共享容量不足:RESOURCE_EXHAUSTED、
// 不带任何错误明细(RetryInfo / QuotaFailure / ErrorInfo),文案为 "Resource exhausted ... try again later"
// 且不涉及项目配额或预留吞吐。其他形态的 429 按账号自身限速处理。
func isVertexSharedCapacity429(body []byte) bool {
errObj := gjson.GetBytes(body, "error")
if !errObj.Exists() {
errObj = gjson.GetBytes(body, "0.error")
}
if !errObj.IsObject() {
return false
}
if !strings.EqualFold(strings.TrimSpace(errObj.Get("status").String()), "RESOURCE_EXHAUSTED") {
return false
}
if len(errObj.Get("details").Array()) > 0 {
return false
}
msg := strings.ToLower(errObj.Get("message").String())
if !strings.Contains(msg, "resource exhausted") || !strings.Contains(msg, "try again later") {
return false
}
return !strings.Contains(msg, "quota") && !strings.Contains(msg, "provisioned throughput")
}
// handleGeminiThirdParty429 处理第三方上游的 429:不套用 AI Studio 每日配额语义,
// 只认响应里显式给出的重试时间,否则走全局 429 兜底冷却。
func (s *GeminiMessagesCompatService) handleGeminiThirdParty429(ctx context.Context, account *Account, body []byte) {
@@ -12,6 +12,7 @@ import (
"testing"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
)
@@ -99,3 +100,66 @@ func TestGeminiForwardAsChatCompletions_VertexAPIKeyUsesAPIKeyHeader(t *testing.
require.Equal(t, "https://aiplatform.googleapis.com/v1/publishers/google/models/gemini-3-flash-preview:generateContent", httpStub.lastReq.URL.String())
requireVertexAPIKeyAuth(t, httpStub.lastReq)
}
type geminiVertex429SequenceUpstream struct {
responses []*http.Response
calls int
}
func (s *geminiVertex429SequenceUpstream) Do(req *http.Request, proxyURL string, accountID int64, accountConcurrency int) (*http.Response, error) {
resp := s.responses[min(s.calls, len(s.responses)-1)]
s.calls++
return resp, nil
}
func (s *geminiVertex429SequenceUpstream) DoWithTLS(req *http.Request, proxyURL string, accountID int64, accountConcurrency int, _ *tlsfingerprint.Profile) (*http.Response, error) {
return s.Do(req, proxyURL, accountID, accountConcurrency)
}
// Vertex 共享容量不足的 429 在同一账号上退避重试,成功后账号状态不变;绑定显式缓存的请求同样如此。
func TestGeminiForwardNative_VertexSharedCapacity429RetriesSameAccountWithoutCooldown(t *testing.T) {
gin.SetMode(gin.TestMode)
sharedCapacity := `{"error":{"code":429,"message":"Resource exhausted. Please try again later.","status":"RESOURCE_EXHAUSTED"}}`
tests := []struct {
name string
ctx context.Context
}{
{name: "unbound", ctx: context.Background()},
{name: "bound_cached_content", ctx: WithGeminiCachedContentBound(context.Background())},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
upstream := &geminiVertex429SequenceUpstream{responses: []*http.Response{
{
StatusCode: http.StatusTooManyRequests,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(sharedCapacity)),
},
{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"application/json"}},
Body: io.NopCloser(strings.NewReader(geminiVertexAPIKeyTestResponse)),
},
}}
repo := &geminiBoundCacheRateLimitRepo{}
svc := &GeminiMessagesCompatService{
httpUpstream: upstream,
cfg: &config.Config{},
accountRepo: repo,
rateLimitService: NewRateLimitService(repo, nil, &config.Config{}, nil, nil),
}
c, rec := newGeminiNativeTestContext(t)
result, err := svc.ForwardNative(tt.ctx, c, geminiVertexAPIKeyTestAccount("my-project"),
"gemini-3.7-flash", "generateContent", false, []byte(`{"contents":[{"role":"user","parts":[{"text":"hi"}]}]}`))
require.NoError(t, err)
require.NotNil(t, result)
require.Equal(t, http.StatusOK, rec.Code)
require.Equal(t, 2, upstream.calls, "429 后在同一账号上重试")
require.Zero(t, repo.rateLimitedCalls, "不写账号级限流")
require.Zero(t, repo.tempCalls)
require.Zero(t, repo.setErrCalls)
require.Empty(t, repo.modelRateLimitCalls)
})
}
}