From db8692d679d375b53ce157477d8e7b6b4d889140 Mon Sep 17 00:00:00 2001 From: ye4241 <6803102+ye4241@users.noreply.github.com> Date: Sat, 19 Sep 2026 00:00:57 +0800 Subject: [PATCH] fix(ratelimit): pause CN coding-plan accounts on quota-exhausted 403 Kimi reports Coding Plan window exhaustion (5h/weekly) as a 403 with error.type=access_terminated_error instead of 429. It fell into the generic handleOpenAI403 escalation, so three 403s within 180 minutes permanently disabled accounts that recover by themselves at the window reset. Detect the quota-exhausted 403 for CN coding-plan accounts and cool it down exactly like the 429 path: SetRateLimited until the earliest future window reset from the quota snapshot, falling back to a short temp-unschedulable window when the snapshot is missing. Non-coding-plan accounts keep the generic 403 escalation since the rolling-window recovery mechanism does not apply to them. Also extract cooldownCNProviderToQuotaSnapshotReset and setCNProviderTempUnschedulable so the 429 (OpenCodeGo / Coding Plan) and 403 branches stop carrying three copies of the same sequence. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../service/ratelimit_cn_providers.go | 135 ++++++++++---- backend/internal/service/ratelimit_service.go | 8 + .../service/ratelimit_service_401_test.go | 10 + .../ratelimit_service_cn_quota_403_test.go | 171 ++++++++++++++++++ 4 files changed, 288 insertions(+), 36 deletions(-) create mode 100644 backend/internal/service/ratelimit_service_cn_quota_403_test.go diff --git a/backend/internal/service/ratelimit_cn_providers.go b/backend/internal/service/ratelimit_cn_providers.go index 15ea009c2a..bb222a9f9b 100644 --- a/backend/internal/service/ratelimit_cn_providers.go +++ b/backend/internal/service/ratelimit_cn_providers.go @@ -6,6 +6,8 @@ import ( "net/http" "strings" "time" + + "github.com/tidwall/gjson" ) // 国产供应商（kimi/zhipu/deepseek）的响应式冷却辅助。 @@ -35,22 +37,65 @@ func isCNProviderConcurrencyLimit403(account *Account, upstreamMsg string) bool strings.TrimSpace(upstreamMsg) == kimiConcurrentRequestLimitMessage } +// cnQuotaExhausted403ErrorType 是 Kimi Coding Plan 配额窗口耗尽的 403 错误类型。 +// 与并发限制 403（见 isCNProviderConcurrencyLimit403）不同，这是 5h/weekly +// 窗口耗尽的限流信号：窗口到期后自动恢复，绝不能落入通用 403 升级计数 +// （连续 3 次永久 SetError），必须按 429 口径冷却到真实窗口重置点。 +const cnQuotaExhausted403ErrorType = "access_terminated_error" + +// cnQuotaExhaustedReasonPrefix 是配额耗尽 403 临时停调 reason 的稳定前缀。 +const cnQuotaExhaustedReasonPrefix = "cn_quota_exhausted" + +// isCNProviderQuotaExhausted403 识别 CN 供应商配额窗口耗尽的 403。 +// 先做廉价的文案子串匹配（usage limit / quota will reset），未命中再解析 +// 响应体匹配结构化错误类型（Kimi 的 access_terminated_error），以兼容同族 +// 供应商的文案变体。与并发限制文案互斥，两类信号不会误判。 +// 仅限 Coding Plan 账号：滚动窗口快照（含重置时间）只有 Coding Plan 账号才有， +// 非 Coding Plan 账号的同类 403 语义不明（可能是需要升级套餐的硬上限）， +// 应回退到通用 403 升级逻辑而非无限临时冷却。 +func isCNProviderQuotaExhausted403(account *Account, responseBody []byte, upstreamMsg string) bool { + if account == nil || !account.IsCNProvider() || !account.IsCodingPlan() { + return false + } + msg := strings.ToLower(strings.TrimSpace(upstreamMsg)) + if strings.Contains(msg, "usage limit") || strings.Contains(msg, "quota will reset") { + return true + } + return strings.EqualFold(strings.TrimSpace(gjson.GetBytes(responseBody, "error.type").String()), cnQuotaExhausted403ErrorType) +} + +// handleCNProviderQuotaExhausted403 把 CN 供应商配额窗口耗尽的 403 按 429 +// 同口径处理：有配额快照时冷却到最早的未来窗口重置点（SetRateLimited， +// 窗口过期后调度自动恢复）；快照缺失（额度探测尚未刷新）时兜底默认 403 +// 冷却时长的临时停调，避免过度停调到数天后的窗口。无论哪条路径都不写 +// 账号 error 状态。 +func (s *RateLimitService) handleCNProviderQuotaExhausted403( + ctx context.Context, + account *Account, + upstreamMsg string, +) { + if s.cooldownCNProviderToQuotaSnapshotReset(ctx, account, cnQuotaExhaustedReasonPrefix, "cn_quota_exhausted_rate_limited") != nil { + return + } + + // upstreamMsg 由调用方（HandleUpstreamError）从同一响应体提取，此处直接复用； + // 为空时仅存前缀，保证 reason 稳定可检索。 + reason := cnQuotaExhaustedReasonPrefix + if msg := strings.TrimSpace(upstreamMsg); msg != "" { + reason += ": " + msg + } + until := time.Now().Add(time.Duration(openAI403CooldownMinutesDefault) * time.Minute) + s.setCNProviderTempUnschedulable(ctx, account, until, cnQuotaExhaustedReasonPrefix, reason, "cn_quota_exhausted_temp_unschedulable") +} + func (s *RateLimitService) handleCNProviderConcurrencyLimit403( ctx context.Context, account *Account, ) { until := time.Now().Add(time.Duration(openAI403CooldownMinutesDefault) * time.Minute) - reason := cnConcurrencyLimitReasonPrefix + ": " + kimiConcurrentRequestLimitMessage - s.notifyAccountSchedulingBlocked(account, until, cnConcurrencyLimitReasonPrefix) - if err := s.accountRepo.SetTempUnschedulable(ctx, account.ID, until, reason); err != nil { - slog.Warn("cn_concurrency_limit_set_temp_unschedulable_failed", "account_id", account.ID, "error", err) - return - } - slog.Info("cn_provider_concurrency_limited", - "account_id", account.ID, - "platform", account.Platform, - "until", until.UTC(), - ) + s.setCNProviderTempUnschedulable(ctx, account, until, + cnConcurrencyLimitReasonPrefix, cnConcurrencyLimitReasonPrefix+": "+kimiConcurrentRequestLimitMessage, + "cn_provider_concurrency_limited") } // cnBalanceLowReason 构造余额不足临时停调的 reason（带稳定前缀）。 @@ -151,6 +196,45 @@ func cnProviderQuotaSnapshotReset(account *Account, now time.Time) *time.Time { return earliest } +// cooldownCNProviderToQuotaSnapshotReset 把账号冷却到配额快照中最早的未来窗口 +// 重置点（SetRateLimited，窗口过期后调度自动恢复），供 429（OpenCodeGo / +// Coding Plan）与配额耗尽 403 共用。仅在冷却成功持久化后返回非 nil； +// 无快照或写库失败时返回 nil，由调用方落入各自的兜底逻辑（避免写库失败 +// 时账号完全失去持久冷却）。 +func (s *RateLimitService) cooldownCNProviderToQuotaSnapshotReset(ctx context.Context, account *Account, reason, logEvent string) *time.Time { + until := cnProviderQuotaSnapshotReset(account, time.Now()) + if until == nil { + return nil + } + s.notifyAccountSchedulingBlocked(account, *until, reason) + if err := s.accountRepo.SetRateLimited(ctx, account.ID, *until); err != nil { + slog.Warn("rate_limit_set_failed", "account_id", account.ID, "error", err) + return nil + } + slog.Info(logEvent, + "account_id", account.ID, + "platform", account.Platform, + "reset_at", until.UTC(), + ) + return until +} + +// setCNProviderTempUnschedulable 写临时停调（含调度层通知与日志），供 CN 403 +// 各分支（并发限制、配额耗尽兜底）共用。告警 key 由 notifyReason 派生，保持 +// 各分支既有日志 key 稳定（_set_temp_unschedulable_failed）。 +func (s *RateLimitService) setCNProviderTempUnschedulable(ctx context.Context, account *Account, until time.Time, notifyReason, storeReason, logEvent string) { + s.notifyAccountSchedulingBlocked(account, until, notifyReason) + if err := s.accountRepo.SetTempUnschedulable(ctx, account.ID, until, storeReason); err != nil { + slog.Warn(notifyReason+"_set_temp_unschedulable_failed", "account_id", account.ID, "error", err) + return + } + slog.Info(logEvent, + "account_id", account.ID, + "platform", account.Platform, + "until", until.UTC(), + ) +} + // applyCNProviderReactive429 处理国产供应商的 429 响应。 // 返回 true 表示已处理（调用方应 return），false 表示未命中、继续走默认 429 逻辑。 func (s *RateLimitService) applyCNProviderReactive429( @@ -160,17 +244,7 @@ func (s *RateLimitService) applyCNProviderReactive429( responseBody []byte, ) bool { if account.IsOpenCodeGo() { - if until := cnProviderQuotaSnapshotReset(account, time.Now()); until != nil { - s.notifyAccountSchedulingBlocked(account, *until, "429") - if err := s.accountRepo.SetRateLimited(ctx, account.ID, *until); err != nil { - slog.Warn("rate_limit_set_failed", "account_id", account.ID, "error", err) - return true - } - slog.Info("opencode_go_rate_limited", - "account_id", account.ID, - "platform", account.Platform, - "reset_at", *until, - ) + if s.cooldownCNProviderToQuotaSnapshotReset(ctx, account, "429", "opencode_go_rate_limited") != nil { return true } if resetAt := parseOpenAIRateLimitResetTime(responseBody); resetAt != nil { @@ -199,20 +273,9 @@ func (s *RateLimitService) applyCNProviderReactive429( } // 2) Coding Plan 窗口耗尽：冷却到快照中最早的窗口重置点（见 // cnProviderQuotaSnapshotReset：429 多由 5h 窗口触发，取较早点避免过度停调）。 - if account.IsCodingPlan() { - if until := cnProviderQuotaSnapshotReset(account, time.Now()); until != nil { - s.notifyAccountSchedulingBlocked(account, *until, "429") - if err := s.accountRepo.SetRateLimited(ctx, account.ID, *until); err != nil { - slog.Warn("rate_limit_set_failed", "account_id", account.ID, "error", err) - return true - } - slog.Info("cn_coding_plan_rate_limited", - "account_id", account.ID, - "platform", account.Platform, - "reset_at", *until, - ) - return true - } + if account.IsCodingPlan() && + s.cooldownCNProviderToQuotaSnapshotReset(ctx, account, "429", "cn_coding_plan_rate_limited") != nil { + return true } return false } diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index 370546af99..6fc119f7ba 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -992,6 +992,14 @@ func (s *RateLimitService) handle403(ctx context.Context, account *Account, upst s.handleCNProviderConcurrencyLimit403(ctx, account) return true } + // Kimi 等 CN 供应商把 Coding Plan 配额窗口耗尽打成 403 + // （error.type=access_terminated_error），这是窗口到期后自动恢复的限流 + // 信号而非封禁：按 429 口径冷却到真实窗口重置点，避免落入下方通用 403 + // 升级计数后被永久 SetError。 + if isCNProviderQuotaExhausted403(account, responseBody, upstreamMsg) { + s.handleCNProviderQuotaExhausted403(ctx, account, upstreamMsg) + return true + } // 国产供应商与 openai 同口径:HTML 403(CDN/代理拦截页)不构成账号失效证据, // 且 403 在 failover 状态集里会被逐账号重放——直接 SetError 会让一个坏请求/ // 一层坏代理连环永久禁用整组账号。走 HTML 豁免 + N 次累计 + 临时冷却。 diff --git a/backend/internal/service/ratelimit_service_401_test.go b/backend/internal/service/ratelimit_service_401_test.go index c1f9f06954..0a3e04ee56 100644 --- a/backend/internal/service/ratelimit_service_401_test.go +++ b/backend/internal/service/ratelimit_service_401_test.go @@ -17,6 +17,7 @@ type rateLimitAccountRepoStub struct { mockAccountRepoForGemini setErrorCalls int tempCalls int + rateLimitedCalls int updateCredentialsCalls int updateExtraCalls int lastCredentials map[string]any @@ -25,6 +26,8 @@ type rateLimitAccountRepoStub struct { lastTempReason string lastErrorID int64 lastTempID int64 + lastRateLimitedID int64 + lastRateLimitedAt time.Time tempErr error } @@ -42,6 +45,13 @@ func (r *rateLimitAccountRepoStub) SetTempUnschedulable(ctx context.Context, id return r.tempErr } +func (r *rateLimitAccountRepoStub) SetRateLimited(ctx context.Context, id int64, resetAt time.Time) error { + r.rateLimitedCalls++ + r.lastRateLimitedID = id + r.lastRateLimitedAt = resetAt + return nil +} + func (r *rateLimitAccountRepoStub) UpdateCredentials(ctx context.Context, id int64, credentials map[string]any) error { r.updateCredentialsCalls++ r.lastCredentials = shallowCopyMap(credentials) diff --git a/backend/internal/service/ratelimit_service_cn_quota_403_test.go b/backend/internal/service/ratelimit_service_cn_quota_403_test.go new file mode 100644 index 0000000000..44062eb8a2 --- /dev/null +++ b/backend/internal/service/ratelimit_service_cn_quota_403_test.go @@ -0,0 +1,171 @@ +//go:build unit + +package service + +import ( + "context" + "fmt" + "net/http" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/stretchr/testify/require" +) + +// newCNQuotaExhaustedTestAccount 构造一个 Kimi Coding Plan 账号： +// account_mode=coding 使 IsCodingPlan() 为真，Extra 带 5h/weekly 窗口快照。 +func newCNQuotaExhaustedTestAccount(resetIn time.Duration) *Account { + extra := map[string]any{} + if resetIn > 0 { + extra[cnExtraKey(PlatformKimi, cnExtraSuffix5hUsed)] = 100 + extra[cnExtraKey(PlatformKimi, cnExtraSuffix5hReset)] = time.Now().Add(resetIn).UTC().Format(time.RFC3339) + extra[cnExtraKey(PlatformKimi, cnExtraSuffixWeeklyReset)] = time.Now().Add(7 * 24 * time.Hour).UTC().Format(time.RFC3339) + } + return &Account{ + ID: 401, + Platform: PlatformKimi, + Type: AccountTypeAPIKey, + Credentials: map[string]any{"account_mode": AccountModeCoding}, + Extra: extra, + } +} + +const kimiWeeklyQuotaExhaustedBody = `{"error":{"message":"You've reached your weekly (7-day) usage limit. Your quota will reset when the current 7-day window ends. To continue now, purchase extra usage or upgrade your plan: https://www.kimi.com/membership/subscription?tab=quota","type":"access_terminated_error"}}` + +// runKimi403 以 403 + 给定响应体驱动 HandleUpstreamError，返回 shouldDisable。 +func runKimi403(service *RateLimitService, account *Account, body string) bool { + return service.HandleUpstreamError( + context.Background(), + account, + http.StatusForbidden, + http.Header{}, + []byte(body), + ) +} + +func TestHandleUpstreamError_KimiQuotaExhausted403RateLimitedToWindowReset(t *testing.T) { + repo := &rateLimitAccountRepoStub{} + blocker := &runtimeBlockRecorder{} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + service.SetAccountRuntimeBlocker(blocker) + account := newCNQuotaExhaustedTestAccount(2 * time.Hour) + + shouldDisable := runKimi403(service, account, kimiWeeklyQuotaExhaustedBody) + + require.True(t, shouldDisable) + require.Equal(t, 0, repo.setErrorCalls, "quota exhaustion must never SetError") + require.Equal(t, 0, repo.tempCalls) + require.Equal(t, 1, repo.rateLimitedCalls) + require.Equal(t, account.ID, repo.lastRateLimitedID) + require.True(t, repo.lastRateLimitedAt.After(time.Now()), "cooldown should target the future window reset") + require.True(t, repo.lastRateLimitedAt.Before(time.Now().Add(3*time.Hour)), "should cool down to the nearer 5h reset, not weekly") + require.Len(t, blocker.accounts, 1) + require.Equal(t, cnQuotaExhaustedReasonPrefix, blocker.reasons[0]) +} + +func TestHandleUpstreamError_KimiQuotaExhausted403WithoutSnapshotFallsBackToTemp(t *testing.T) { + repo := &rateLimitAccountRepoStub{} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + account := newCNQuotaExhaustedTestAccount(0) + + shouldDisable := runKimi403(service, account, kimiWeeklyQuotaExhaustedBody) + + require.True(t, shouldDisable) + require.Equal(t, 0, repo.setErrorCalls) + require.Equal(t, 0, repo.rateLimitedCalls) + require.Equal(t, 1, repo.tempCalls) + require.Contains(t, repo.lastTempReason, cnQuotaExhaustedReasonPrefix) + require.Greater(t, len(repo.lastTempReason), len(cnQuotaExhaustedReasonPrefix)) +} + +func TestHandleUpstreamError_KimiQuotaExhausted403MatchedByMessageOnly(t *testing.T) { + repo := &rateLimitAccountRepoStub{} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + account := newCNQuotaExhaustedTestAccount(2 * time.Hour) + // 无 error.type 字段，仅靠文案兜底匹配。 + body := `{"error":{"message":"You've reached your weekly (7-day) usage limit. Your quota will reset when the current 7-day window ends."}}` + + shouldDisable := runKimi403(service, account, body) + + require.True(t, shouldDisable) + require.Equal(t, 0, repo.setErrorCalls) + require.Equal(t, 1, repo.rateLimitedCalls) +} + +func TestHandleUpstreamError_KimiConcurrencyLimit403StillUsesConcurrencyPath(t *testing.T) { + repo := &rateLimitAccountRepoStub{} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + account := newCNQuotaExhaustedTestAccount(2 * time.Hour) + body := fmt.Sprintf(`{"error":{"message":%q,"type":"access_terminated_error"}}`, kimiConcurrentRequestLimitMessage) + + shouldDisable := runKimi403(service, account, body) + + require.True(t, shouldDisable) + require.Equal(t, 0, repo.setErrorCalls) + require.Equal(t, 0, repo.rateLimitedCalls, "concurrency-limit message must keep its dedicated path") + require.Equal(t, 1, repo.tempCalls) + require.Contains(t, repo.lastTempReason, cnConcurrencyLimitReasonPrefix) +} + +func TestHandleUpstreamError_KimiGeneric403StillEscalates(t *testing.T) { + repo := &rateLimitAccountRepoStub{} + counter := &openAI403CounterCacheStub{counts: []int64{3}} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + service.SetOpenAI403CounterCache(counter) + account := newCNQuotaExhaustedTestAccount(2 * time.Hour) + + shouldDisable := runKimi403(service, account, `{"error":{"message":"policy rejected this request"}}`) + + require.True(t, shouldDisable) + require.Equal(t, 1, repo.setErrorCalls, "non-quota 403 keeps the generic escalation path") + require.Equal(t, 0, repo.rateLimitedCalls) +} + +func TestHandleUpstreamError_KimiNonCodingPlanQuotaMessageUsesGeneric403(t *testing.T) { + repo := &rateLimitAccountRepoStub{} + counter := &openAI403CounterCacheStub{counts: []int64{3}} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + service.SetOpenAI403CounterCache(counter) + account := newCNQuotaExhaustedTestAccount(2 * time.Hour) + account.Credentials = map[string]any{"account_mode": AccountModePayG} + + shouldDisable := runKimi403(service, account, kimiWeeklyQuotaExhaustedBody) + + require.True(t, shouldDisable) + require.Equal(t, 1, repo.setErrorCalls, "non-coding-plan account must keep the generic 403 escalation") + require.Equal(t, 0, repo.rateLimitedCalls) + require.Equal(t, 0, repo.tempCalls) +} + +func TestHandleUpstreamError_NonCNAccessTerminated403KeepsGenericPath(t *testing.T) { + repo := &rateLimitAccountRepoStub{} + counter := &openAI403CounterCacheStub{counts: []int64{1}} + service := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + service.SetOpenAI403CounterCache(counter) + account := &Account{ + ID: 402, + Platform: PlatformOpenAI, + Type: AccountTypeOAuth, + } + + shouldDisable := runKimi403(service, account, kimiWeeklyQuotaExhaustedBody) + + require.True(t, shouldDisable) + require.Equal(t, 0, repo.rateLimitedCalls, "non-CN platform must not enter the CN quota path") + require.Equal(t, 0, repo.setErrorCalls) + require.Equal(t, 1, repo.tempCalls) + require.Contains(t, repo.lastTempReason, "(1/3)") +} + +func TestIsCNProviderQuotaExhausted403_Classification(t *testing.T) { + kimi := newCNQuotaExhaustedTestAccount(0) + require.True(t, isCNProviderQuotaExhausted403(kimi, []byte(kimiWeeklyQuotaExhaustedBody), "")) + require.True(t, isCNProviderQuotaExhausted403(kimi, []byte(`{"error":{"message":"You've reached your weekly (7-day) usage limit."}}`), "You've reached your weekly (7-day) usage limit.")) + require.False(t, isCNProviderQuotaExhausted403(kimi, []byte(`{"error":{"message":"policy rejected"}}`), "policy rejected")) + require.False(t, isCNProviderQuotaExhausted403(kimi, nil, "")) + + openai := &Account{ID: 1, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + require.False(t, isCNProviderQuotaExhausted403(openai, []byte(kimiWeeklyQuotaExhaustedBody), "")) + require.False(t, isCNProviderQuotaExhausted403(nil, []byte(kimiWeeklyQuotaExhaustedBody), "")) +}