Merge pull request #7340 from ye4241/fix/kimi-cn-quota-403-rate-limit-pause

fix(ratelimit): CN coding-plan 账号配额耗尽的 403 按限流暂停调度，而非永久禁用

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 稳定（<prefix>_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
 }

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 次累计 + 临时冷却。

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)

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