mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-15 23:22:59 +08:00
fix(account): address review on OpenAI quota auto-pause
- gate previous_response_id sticky path with quota auto-pause check at both the snapshot and DB-recheck stages (previously bypassed, #1) - skip pausing when the usage window already reset to avoid a stale stuck-pause; carry codex_*_reset_at / reset_after_seconds / codex_usage_updated_at through the scheduler snapshot whitelist (#2) - remove the incomplete limit mode; percentage threshold only (#3) - add global default 5h/7d threshold inputs to the Ops settings dialog with validation and en/zh i18n (#4) - downgrade account_auto_paused_by_quota log from Info to Debug; it fires per-candidate on the scheduling hot path (#5) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
ead471d64b
commit
8b7a822706
@@ -550,10 +550,13 @@ func filterSchedulerExtra(extra map[string]any) map[string]any {
|
||||
"openai_responses_supported",
|
||||
"codex_5h_used_percent",
|
||||
"codex_7d_used_percent",
|
||||
"codex_5h_reset_at",
|
||||
"codex_7d_reset_at",
|
||||
"codex_5h_reset_after_seconds",
|
||||
"codex_7d_reset_after_seconds",
|
||||
"codex_usage_updated_at",
|
||||
"auto_pause_5h_threshold",
|
||||
"auto_pause_7d_threshold",
|
||||
"auto_pause_5h_limit",
|
||||
"auto_pause_7d_limit",
|
||||
}
|
||||
filtered := make(map[string]any)
|
||||
for _, key := range keys {
|
||||
|
||||
@@ -80,10 +80,15 @@ func TestBuildSchedulerMetadataAccount_KeepsQuotaAutoPauseFields(t *testing.T) {
|
||||
account := service.Account{
|
||||
ID: 88,
|
||||
Extra: map[string]any{
|
||||
"codex_5h_used_percent": 12.34,
|
||||
"codex_7d_used_percent": 56.78,
|
||||
"auto_pause_5h_threshold": 0.95,
|
||||
"auto_pause_7d_threshold": 0.96,
|
||||
"codex_5h_used_percent": 12.34,
|
||||
"codex_7d_used_percent": 56.78,
|
||||
"codex_5h_reset_at": "2026-05-29T10:00:00Z",
|
||||
"codex_7d_reset_at": "2026-06-01T10:00:00Z",
|
||||
"codex_5h_reset_after_seconds": 300,
|
||||
"codex_7d_reset_after_seconds": 600,
|
||||
"codex_usage_updated_at": "2026-05-29T09:00:00Z",
|
||||
"auto_pause_5h_threshold": 0.95,
|
||||
"auto_pause_7d_threshold": 0.96,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -91,6 +96,11 @@ func TestBuildSchedulerMetadataAccount_KeepsQuotaAutoPauseFields(t *testing.T) {
|
||||
|
||||
require.Equal(t, 12.34, got.Extra["codex_5h_used_percent"])
|
||||
require.Equal(t, 56.78, got.Extra["codex_7d_used_percent"])
|
||||
require.Equal(t, "2026-05-29T10:00:00Z", got.Extra["codex_5h_reset_at"])
|
||||
require.Equal(t, "2026-06-01T10:00:00Z", got.Extra["codex_7d_reset_at"])
|
||||
require.Equal(t, 300, got.Extra["codex_5h_reset_after_seconds"])
|
||||
require.Equal(t, 600, got.Extra["codex_7d_reset_after_seconds"])
|
||||
require.Equal(t, "2026-05-29T09:00:00Z", got.Extra["codex_usage_updated_at"])
|
||||
require.Equal(t, 0.95, got.Extra["auto_pause_5h_threshold"])
|
||||
require.Equal(t, 0.96, got.Extra["auto_pause_7d_threshold"])
|
||||
}
|
||||
|
||||
@@ -704,7 +704,6 @@ func TestOpenAIGatewayService_SelectAccountForModelWithExclusions_AutoPauseBy5hT
|
||||
Extra: map[string]any{
|
||||
"codex_5h_used_percent": 95.0,
|
||||
"auto_pause_5h_threshold": 0.95,
|
||||
"auto_pause_5h_limit": 100,
|
||||
},
|
||||
}
|
||||
secondary := Account{ID: 35002, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5}
|
||||
@@ -729,7 +728,6 @@ func TestOpenAIGatewayService_SelectAccountForModelWithExclusions_AllowsBelow5hT
|
||||
Extra: map[string]any{
|
||||
"codex_5h_used_percent": 80.0,
|
||||
"auto_pause_5h_threshold": 0.95,
|
||||
"auto_pause_5h_limit": 100,
|
||||
},
|
||||
}
|
||||
secondary := Account{ID: 35102, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5}
|
||||
@@ -754,7 +752,6 @@ func TestOpenAIGatewayService_SelectAccountForModelWithExclusions_AutoPauseBy7dT
|
||||
Extra: map[string]any{
|
||||
"codex_7d_used_percent": 95.0,
|
||||
"auto_pause_7d_threshold": 0.95,
|
||||
"auto_pause_7d_limit": 200,
|
||||
},
|
||||
}
|
||||
secondary := Account{ID: 35202, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5}
|
||||
@@ -790,7 +787,6 @@ func TestOpenAIGatewayService_SelectAccountForModelWithExclusions_UsesGlobalDefa
|
||||
Priority: 0,
|
||||
Extra: map[string]any{
|
||||
"codex_5h_used_percent": 95.0,
|
||||
"auto_pause_5h_limit": 100,
|
||||
},
|
||||
}
|
||||
secondary := Account{ID: 35402, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5}
|
||||
@@ -802,6 +798,60 @@ func TestOpenAIGatewayService_SelectAccountForModelWithExclusions_UsesGlobalDefa
|
||||
require.Equal(t, int64(35402), account.ID)
|
||||
}
|
||||
|
||||
func TestOpenAIGatewayService_SelectAccountForModelWithExclusions_StaleUsageWindowResetSkipsPause(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
// Usage is over threshold but the window's reset time has already passed, so the
|
||||
// cached percentage is stale (the real window rolled over) and the account must NOT
|
||||
// stay paused — otherwise it could be skipped forever with no traffic to refresh it.
|
||||
primary := Account{
|
||||
ID: 35501,
|
||||
Platform: PlatformOpenAI,
|
||||
Type: AccountTypeAPIKey,
|
||||
Status: StatusActive,
|
||||
Schedulable: true,
|
||||
Concurrency: 1,
|
||||
Priority: 0,
|
||||
Extra: map[string]any{
|
||||
"codex_5h_used_percent": 99.0,
|
||||
"auto_pause_5h_threshold": 0.95,
|
||||
"codex_5h_reset_at": time.Now().Add(-time.Minute).Format(time.RFC3339),
|
||||
},
|
||||
}
|
||||
secondary := Account{ID: 35502, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5}
|
||||
svc := &OpenAIGatewayService{accountRepo: schedulerTestOpenAIAccountRepo{accounts: []Account{primary, secondary}}, cfg: &config.Config{}}
|
||||
|
||||
account, err := svc.SelectAccountForModelWithExclusions(ctx, nil, "", "gpt-5.1", nil)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, account)
|
||||
require.Equal(t, int64(35501), account.ID)
|
||||
}
|
||||
|
||||
func TestOpenAIGatewayService_SelectAccountForModelWithExclusions_FreshUsageWindowStillPauses(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
// Same as above but the window has not reset yet, so the account stays paused.
|
||||
primary := Account{
|
||||
ID: 35601,
|
||||
Platform: PlatformOpenAI,
|
||||
Type: AccountTypeAPIKey,
|
||||
Status: StatusActive,
|
||||
Schedulable: true,
|
||||
Concurrency: 1,
|
||||
Priority: 0,
|
||||
Extra: map[string]any{
|
||||
"codex_5h_used_percent": 99.0,
|
||||
"auto_pause_5h_threshold": 0.95,
|
||||
"codex_5h_reset_at": time.Now().Add(time.Hour).Format(time.RFC3339),
|
||||
},
|
||||
}
|
||||
secondary := Account{ID: 35602, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Status: StatusActive, Schedulable: true, Concurrency: 1, Priority: 5}
|
||||
svc := &OpenAIGatewayService{accountRepo: schedulerTestOpenAIAccountRepo{accounts: []Account{primary, secondary}}, cfg: &config.Config{}}
|
||||
|
||||
account, err := svc.SelectAccountForModelWithExclusions(ctx, nil, "", "gpt-5.1", nil)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, account)
|
||||
require.Equal(t, int64(35602), account.ID)
|
||||
}
|
||||
|
||||
func TestOpenAIGatewayService_SelectAccountForModelWithExclusions_SkipsFreshlyRateLimitedSnapshotCandidate(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
groupID := int64(10102)
|
||||
|
||||
@@ -1328,13 +1328,12 @@ func isOpenAIAccountEligibleForRequest(ctx context.Context, account *Account, re
|
||||
return false
|
||||
}
|
||||
if paused, reason := shouldAutoPauseOpenAIAccountByQuota(ctx, account); paused {
|
||||
slog.Info("account_auto_paused_by_quota",
|
||||
// Debug level: this fires per-candidate on the scheduling hot path, so Info
|
||||
// would amplify into log spam once several accounts cross the threshold.
|
||||
slog.Debug("account_auto_paused_by_quota",
|
||||
"account_id", account.ID,
|
||||
"usage_5h_percent", readOpenAIQuotaUsedPercent(account.Extra, "5h"),
|
||||
"usage_7d_percent", readOpenAIQuotaUsedPercent(account.Extra, "7d"),
|
||||
"threshold_type", reason.window,
|
||||
"window", reason.window,
|
||||
"threshold", reason.threshold,
|
||||
"limit", reason.limit,
|
||||
"utilization", reason.utilization,
|
||||
)
|
||||
return false
|
||||
@@ -1354,7 +1353,6 @@ func isOpenAIAccountEligibleForRequest(ctx context.Context, account *Account, re
|
||||
type openAIQuotaAutoPauseDecision struct {
|
||||
window string
|
||||
threshold float64
|
||||
limit float64
|
||||
utilization float64
|
||||
}
|
||||
|
||||
@@ -1363,18 +1361,15 @@ func shouldAutoPauseOpenAIAccountByQuota(ctx context.Context, account *Account)
|
||||
return false, openAIQuotaAutoPauseDecision{}
|
||||
}
|
||||
threshold5h, threshold7d := resolveOpenAIQuotaAutoPauseThresholds(ctx, account)
|
||||
now := time.Now()
|
||||
if threshold5h > 0 {
|
||||
if utilization, limit, ok := resolveOpenAIQuotaUtilization(account.Extra, "5h"); ok {
|
||||
if utilization >= threshold5h {
|
||||
return true, openAIQuotaAutoPauseDecision{window: "5h", threshold: threshold5h, limit: limit, utilization: utilization}
|
||||
}
|
||||
if utilization, ok := resolveOpenAIQuotaUtilization(account.Extra, "5h", now); ok && utilization >= threshold5h {
|
||||
return true, openAIQuotaAutoPauseDecision{window: "5h", threshold: threshold5h, utilization: utilization}
|
||||
}
|
||||
}
|
||||
if threshold7d > 0 {
|
||||
if utilization, limit, ok := resolveOpenAIQuotaUtilization(account.Extra, "7d"); ok {
|
||||
if utilization >= threshold7d {
|
||||
return true, openAIQuotaAutoPauseDecision{window: "7d", threshold: threshold7d, limit: limit, utilization: utilization}
|
||||
}
|
||||
if utilization, ok := resolveOpenAIQuotaUtilization(account.Extra, "7d", now); ok && utilization >= threshold7d {
|
||||
return true, openAIQuotaAutoPauseDecision{window: "7d", threshold: threshold7d, utilization: utilization}
|
||||
}
|
||||
}
|
||||
return false, openAIQuotaAutoPauseDecision{}
|
||||
@@ -1431,18 +1426,49 @@ func resolveAccountExtraNumber(extra map[string]any, keys ...string) (float64, b
|
||||
return 0, false
|
||||
}
|
||||
|
||||
func resolveOpenAIQuotaUtilization(extra map[string]any, window string) (float64, float64, bool) {
|
||||
limitKeys := []string{"auto_pause_" + window + "_limit", "quota_" + window + "_limit", window + "_limit"}
|
||||
if limit, ok := resolveAccountExtraNumber(extra, limitKeys...); ok && limit > 0 {
|
||||
if usage, ok := resolveAccountExtraNumber(extra, "usage_"+window); ok && usage >= 0 {
|
||||
return usage / limit, limit, true
|
||||
}
|
||||
}
|
||||
// resolveOpenAIQuotaUtilization returns the current utilization ratio (0..1) for the
|
||||
// given Codex usage window. ok=false means there is no usable signal to pause on:
|
||||
// either no snapshot exists, or the window has already rolled over so the cached
|
||||
// percentage is stale. The stale guard matters because a paused account stops
|
||||
// receiving requests, so its snapshot is never refreshed from upstream headers —
|
||||
// without this check an old used_percent would keep the account paused forever even
|
||||
// after the real window reset.
|
||||
func resolveOpenAIQuotaUtilization(extra map[string]any, window string, now time.Time) (float64, bool) {
|
||||
usedPercent := readOpenAIQuotaUsedPercent(extra, window)
|
||||
if usedPercent <= 0 {
|
||||
return 0, 0, false
|
||||
return 0, false
|
||||
}
|
||||
return usedPercent / 100, 100, true
|
||||
if openAIQuotaWindowReset(extra, window, now) {
|
||||
return 0, false
|
||||
}
|
||||
return usedPercent / 100, true
|
||||
}
|
||||
|
||||
// openAIQuotaWindowReset reports whether the Codex usage window's reset time has
|
||||
// already passed relative to now. It prefers the absolute codex_<window>_reset_at
|
||||
// timestamp and falls back to codex_<window>_reset_after_seconds anchored at
|
||||
// codex_usage_updated_at, mirroring AccountUsageService's window-progress logic.
|
||||
func openAIQuotaWindowReset(extra map[string]any, window string, now time.Time) bool {
|
||||
if len(extra) == 0 {
|
||||
return false
|
||||
}
|
||||
if resetAtRaw, ok := extra["codex_"+window+"_reset_at"]; ok {
|
||||
if resetAt, err := parseTime(fmt.Sprint(resetAtRaw)); err == nil {
|
||||
return !now.Before(resetAt)
|
||||
}
|
||||
}
|
||||
resetAfter := parseExtraInt(extra["codex_"+window+"_reset_after_seconds"])
|
||||
if resetAfter <= 0 {
|
||||
return false
|
||||
}
|
||||
base := now
|
||||
if updatedRaw, ok := extra["codex_usage_updated_at"]; ok {
|
||||
if updatedAt, err := parseTime(fmt.Sprint(updatedRaw)); err == nil {
|
||||
base = updatedAt
|
||||
}
|
||||
}
|
||||
resetAt := base.Add(time.Duration(resetAfter) * time.Second)
|
||||
return !now.Before(resetAt)
|
||||
}
|
||||
|
||||
func readOpenAIQuotaUsedPercent(extra map[string]any, window string) float64 {
|
||||
|
||||
@@ -48,6 +48,46 @@ func TestOpenAIGatewayService_SelectAccountByPreviousResponseID_Hit(t *testing.T
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenAIGatewayService_SelectAccountByPreviousResponseID_QuotaAutoPausedMiss(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
groupID := int64(23)
|
||||
account := Account{
|
||||
ID: 77,
|
||||
Platform: PlatformOpenAI,
|
||||
Type: AccountTypeAPIKey,
|
||||
Status: StatusActive,
|
||||
Schedulable: true,
|
||||
Concurrency: 2,
|
||||
Extra: map[string]any{
|
||||
"openai_apikey_responses_websockets_v2_enabled": true,
|
||||
"codex_5h_used_percent": 96.0,
|
||||
"auto_pause_5h_threshold": 0.95,
|
||||
},
|
||||
}
|
||||
cache := &stubGatewayCache{}
|
||||
store := NewOpenAIWSStateStore(cache)
|
||||
cfg := newOpenAIWSV2TestConfig()
|
||||
svc := &OpenAIGatewayService{
|
||||
accountRepo: stubOpenAIAccountRepo{accounts: []Account{account}},
|
||||
cache: cache,
|
||||
cfg: cfg,
|
||||
concurrencyService: NewConcurrencyService(stubConcurrencyCache{}),
|
||||
openaiWSStateStore: store,
|
||||
}
|
||||
|
||||
require.NoError(t, store.BindResponseAccount(ctx, groupID, "resp_prev_quota", account.ID, time.Hour))
|
||||
|
||||
selection, err := svc.SelectAccountByPreviousResponseID(ctx, &groupID, "resp_prev_quota", "gpt-5.1", nil, false)
|
||||
require.NoError(t, err)
|
||||
require.Nil(t, selection, "超过 5h 配额阈值的账号不应继续命中 previous_response_id 粘连")
|
||||
|
||||
// Auto-pause is transient, so the binding is preserved: the chain can resume on the
|
||||
// same account once the quota window resets.
|
||||
boundAccountID, getErr := store.GetResponseAccount(ctx, groupID, "resp_prev_quota")
|
||||
require.NoError(t, getErr)
|
||||
require.Equal(t, account.ID, boundAccountID)
|
||||
}
|
||||
|
||||
func TestOpenAIGatewayService_SelectAccountByPreviousResponseID_RateLimitedMiss(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
groupID := int64(23)
|
||||
|
||||
@@ -4045,6 +4045,13 @@ func (s *OpenAIGatewayService) selectAccountByPreviousResponseIDForCapability(
|
||||
if !account.SupportsOpenAIEndpointCapability(requiredCapability) {
|
||||
return nil, nil
|
||||
}
|
||||
// Quota auto-pause must also gate the previous_response_id sticky path; otherwise an
|
||||
// account over its 5h/7d threshold keeps serving the same response chain even though
|
||||
// normal scheduling skips it. Pause is transient, so fall through to normal scheduling
|
||||
// without deleting the binding (the window may reset before the next turn).
|
||||
if paused, _ := shouldAutoPauseOpenAIAccountByQuota(ctx, account); paused {
|
||||
return nil, nil
|
||||
}
|
||||
if s.schedulerSnapshot != nil && s.accountRepo != nil {
|
||||
latest, latestErr := s.accountRepo.GetByID(ctx, account.ID)
|
||||
if latestErr != nil || latest == nil {
|
||||
@@ -4061,6 +4068,9 @@ func (s *OpenAIGatewayService) selectAccountByPreviousResponseIDForCapability(
|
||||
if !latest.SupportsOpenAIEndpointCapability(requiredCapability) {
|
||||
return nil, nil
|
||||
}
|
||||
if paused, _ := shouldAutoPauseOpenAIAccountByQuota(ctx, latest); paused {
|
||||
return nil, nil
|
||||
}
|
||||
if s.isOpenAIAccountRuntimeBlocked(latest) {
|
||||
_ = store.DeleteResponseAccount(ctx, derefGroupID(groupID), responseID)
|
||||
return nil, nil
|
||||
|
||||
Reference in New Issue
Block a user