fix: sticky session upstream restriction bypass + 429 over-aggressive credits exhaustion

1. Sticky session now checks upstream channel restriction:
   - Add isStickyAccountUpstreamRestricted() helper
   - Apply to both routing sticky (selectAccountForModelWithPlatform)
     and normal sticky (selectAccountWithMixedScheduling)

2. shouldMarkCreditsExhausted: only mark on 429 with credits keywords:
   - Node-level rate limiting / transient 429 no longer triggers 5h circuit break
   - Extract containsCreditsExhaustedKeyword() to unify 429 and 403 paths
This commit is contained in:
erio
2026-04-02 20:57:13 +08:00
parent 5a4340e49b
commit 259506b429
3 changed files with 32 additions and 6 deletions
@@ -279,9 +279,10 @@ func resolveCreditsOveragesModelKey(ctx context.Context, account *Account, upstr
// shouldMarkCreditsExhausted 判断一次 credits 请求失败是否应标记为 credits 耗尽。
// 此函数在积分注入后失败时调用(预检查注入 + attemptCreditsOveragesRetry 两条路径)。
// - 429 + 非单模型限流:积分注入后仍 429 → 标记耗尽。
// - 429 + 单模型限流("exhausted your capacity on this model"):该模型免费配额用完,
// 积分注入对此无效,但账号积分对其他模型可能仍可用 → 不标记积分耗尽。
// - 429 + 积分关键词(如 "insufficient credits"):积分确实不足 → 标记耗尽。
// - 429 + 无积分关键词(节点限流、瞬时 rate limit 等):与积分无关 → 不标记。
// - 403 等其他 4xx:检查 body 是否包含积分不足的关键词。
//
// clearCreditsExhausted 会在后续成功时自动清除。
@@ -299,9 +300,16 @@ func shouldMarkCreditsExhausted(resp *http.Response, respBody []byte, reqErr err
if strings.Contains(bodyLower, "exhausted your capacity on this model") {
return false
}
return true
// 仅当 body 包含积分相关关键词时才标记耗尽;
// 节点限流、瞬时 rate limit 等非积分 429 不应触发 5 小时熔断。
return containsCreditsExhaustedKeyword(bodyLower)
}
// 其他 4xx:关键词匹配(如 403 + "Insufficient credits")
return containsCreditsExhaustedKeyword(bodyLower)
}
// containsCreditsExhaustedKeyword 检查响应体是否包含积分耗尽的关键词。
func containsCreditsExhaustedKeyword(bodyLower string) bool {
for _, keyword := range creditsExhaustedKeywords {
if strings.Contains(bodyLower, keyword) {
return true
@@ -424,9 +424,15 @@ func TestShouldMarkCreditsExhausted(t *testing.T) {
require.False(t, shouldMarkCreditsExhausted(resp, body, nil))
})
t.Run("429 结构化限流也标记(积分注入后仍 429 即为耗尽)", func(t *testing.T) {
t.Run("429 结构化限流不标记(节点限流/瞬时 rate limit 与积分无关)", func(t *testing.T) {
resp := &http.Response{StatusCode: http.StatusTooManyRequests}
body := []byte(`{"error":{"status":"RESOURCE_EXHAUSTED","details":[{"@type":"type.googleapis.com/google.rpc.ErrorInfo","reason":"RATE_LIMIT_EXCEEDED"},{"@type":"type.googleapis.com/google.rpc.RetryInfo","retryDelay":"0.5s"}]}}`)
require.False(t, shouldMarkCreditsExhausted(resp, body, nil))
})
t.Run("429 含积分关键词时标记", func(t *testing.T) {
resp := &http.Response{StatusCode: http.StatusTooManyRequests}
body := []byte(`{"error":{"message":"Insufficient credits to complete this request"}}`)
require.True(t, shouldMarkCreditsExhausted(resp, body, nil))
})
+15 -3
View File
@@ -3006,7 +3006,7 @@ func (s *GatewayService) selectAccountForModelWithPlatform(ctx context.Context,
if clearSticky {
_ = s.cache.DeleteSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
}
if !clearSticky && s.isAccountInGroup(account, groupID) && account.Platform == platform && (requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, account, requestedModel)) && s.isAccountSchedulableForModelSelection(ctx, account, requestedModel) && s.isAccountSchedulableForQuota(account) && s.isAccountSchedulableForWindowCost(ctx, account, true) && s.isAccountSchedulableForRPM(ctx, account, true) {
if !clearSticky && s.isAccountInGroup(account, groupID) && account.Platform == platform && (requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, account, requestedModel)) && s.isAccountSchedulableForModelSelection(ctx, account, requestedModel) && s.isAccountSchedulableForQuota(account) && s.isAccountSchedulableForWindowCost(ctx, account, true) && s.isAccountSchedulableForRPM(ctx, account, true) && !s.isStickyAccountUpstreamRestricted(ctx, groupID, account, requestedModel) {
if s.debugModelRoutingEnabled() {
logger.LegacyPrintf("service.gateway", "[ModelRoutingDebug] legacy routed sticky hit: group_id=%v model=%s session=%s account=%d", derefGroupID(groupID), requestedModel, shortSessionHash(sessionHash), accountID)
}
@@ -3361,7 +3361,7 @@ func (s *GatewayService) selectAccountWithMixedScheduling(ctx context.Context, g
if clearSticky {
_ = s.cache.DeleteSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
}
if !clearSticky && s.isAccountInGroup(account, groupID) && (requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, account, requestedModel)) && s.isAccountSchedulableForModelSelection(ctx, account, requestedModel) && s.isAccountSchedulableForQuota(account) && s.isAccountSchedulableForWindowCost(ctx, account, true) && s.isAccountSchedulableForRPM(ctx, account, true) {
if !clearSticky && s.isAccountInGroup(account, groupID) && (requestedModel == "" || s.isModelSupportedByAccountWithContext(ctx, account, requestedModel)) && s.isAccountSchedulableForModelSelection(ctx, account, requestedModel) && s.isAccountSchedulableForQuota(account) && s.isAccountSchedulableForWindowCost(ctx, account, true) && s.isAccountSchedulableForRPM(ctx, account, true) && !s.isStickyAccountUpstreamRestricted(ctx, groupID, account, requestedModel) {
if account.Platform == nativePlatform || (account.Platform == PlatformAntigravity && account.IsMixedSchedulingEnabled()) {
return account, nil
}
@@ -3385,7 +3385,6 @@ func (s *GatewayService) selectAccountWithMixedScheduling(ctx context.Context, g
ctx = s.withRPMPrefetch(ctx, accounts)
// 3. 按优先级+最久未用选择(考虑模型支持和混合调度)
// needsUpstreamCheck 仅在主选择循环中使用;粘性会话命中时跳过此检查。
needsUpstreamCheck := s.needsUpstreamChannelRestrictionCheck(ctx, groupID)
var selected *Account
for i := range accounts {
@@ -8455,6 +8454,19 @@ func (s *GatewayService) needsUpstreamChannelRestrictionCheck(ctx context.Contex
return ch.BillingModelSource == BillingModelSourceUpstream
}
// isStickyAccountUpstreamRestricted 检查粘性会话命中的账号是否受 upstream 渠道限制。
// 合并 needsUpstreamChannelRestrictionCheck + isUpstreamModelRestrictedByChannel 两步调用,
// 供 sticky session 条件链使用,避免内联多个函数调用导致行过长。
func (s *GatewayService) isStickyAccountUpstreamRestricted(ctx context.Context, groupID *int64, account *Account, requestedModel string) bool {
if groupID == nil {
return false
}
if !s.needsUpstreamChannelRestrictionCheck(ctx, groupID) {
return false
}
return s.isUpstreamModelRestrictedByChannel(ctx, *groupID, account, requestedModel)
}
// ForwardCountTokens 转发 count_tokens 请求到上游 API
// 特点:不记录使用量、仅支持非流式响应
func (s *GatewayService) ForwardCountTokens(ctx context.Context, c *gin.Context, account *Account, parsed *ParsedRequest) error {