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), "")) +}