mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-14 12:02:27 +08:00
fix(openai): preserve exponential capacity backoff
This commit is contained in:
@@ -36,6 +36,9 @@ const (
|
||||
maxSameAccountRetries = 3
|
||||
// sameAccountRetryDelay 同账号重试间隔
|
||||
sameAccountRetryDelay = 500 * time.Millisecond
|
||||
// maxRequestScopedRetryDelay 限制请求级瞬时错误的指数退避上限,避免高重试配置
|
||||
// 将单次请求拖入分钟级等待。
|
||||
maxRequestScopedRetryDelay = 8 * time.Second
|
||||
// singleAccountBackoffDelay 单账号分组 503 退避重试固定延时。
|
||||
// Service 层在 SingleAccountRetry 模式下已做充分原地重试(最多 3 次、总等待 30s),
|
||||
// Handler 层只需短暂间隔后重新进入 Service 层即可。
|
||||
@@ -52,6 +55,21 @@ const (
|
||||
// 语义上等同于「无可用账号」:候选账号都不满足分组的利润约束。
|
||||
const profitVetoExhaustedMessage = "No available accounts: all candidates rejected by group profit control"
|
||||
|
||||
func sameAccountRetryDelayFor(failoverErr *service.UpstreamFailoverError, retryCount int) time.Duration {
|
||||
if failoverErr == nil || !failoverErr.RequestScopedTransient || retryCount <= 1 {
|
||||
return sameAccountRetryDelay
|
||||
}
|
||||
|
||||
delay := sameAccountRetryDelay
|
||||
for i := 1; i < retryCount; i++ {
|
||||
if delay >= maxRequestScopedRetryDelay/2 {
|
||||
return maxRequestScopedRetryDelay
|
||||
}
|
||||
delay *= 2
|
||||
}
|
||||
return delay
|
||||
}
|
||||
|
||||
// FailoverState 跨循环迭代共享的 failover 状态
|
||||
type FailoverState struct {
|
||||
SwitchCount int
|
||||
@@ -149,13 +167,15 @@ func (s *FailoverState) HandleFailoverError(
|
||||
// 重试次数上限 retryLimit 由调用方传入(账号级 pool_mode_retry_count 配置)。
|
||||
if failoverErr.RetryableOnSameAccount && s.SameAccountRetryCount[accountID] < retryLimit {
|
||||
s.SameAccountRetryCount[accountID]++
|
||||
retryDelay := sameAccountRetryDelayFor(failoverErr, s.SameAccountRetryCount[accountID])
|
||||
logger.FromContext(ctx).Warn("gateway.failover_same_account_retry",
|
||||
zap.Int64("account_id", accountID),
|
||||
zap.Int("upstream_status", failoverErr.StatusCode),
|
||||
zap.Int("same_account_retry_count", s.SameAccountRetryCount[accountID]),
|
||||
zap.Int("same_account_retry_max", retryLimit),
|
||||
zap.Duration("retry_delay", retryDelay),
|
||||
)
|
||||
if !sleepWithContext(ctx, sameAccountRetryDelay) {
|
||||
if !sleepWithContext(ctx, retryDelay) {
|
||||
return FailoverCanceled
|
||||
}
|
||||
return FailoverContinue
|
||||
|
||||
@@ -31,6 +31,35 @@ func (m *mockTempUnscheduler) TempUnscheduleRetryableError(_ context.Context, ac
|
||||
m.calls = append(m.calls, tempUnscheduleCall{accountID: accountID, failoverErr: failoverErr})
|
||||
}
|
||||
|
||||
func TestSameAccountRetryDelayFor(t *testing.T) {
|
||||
capacityErr := &service.UpstreamFailoverError{RequestScopedTransient: true}
|
||||
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
retryCount int
|
||||
want time.Duration
|
||||
}{
|
||||
{name: "first retry", retryCount: 1, want: 500 * time.Millisecond},
|
||||
{name: "second retry", retryCount: 2, want: time.Second},
|
||||
{name: "third retry", retryCount: 3, want: 2 * time.Second},
|
||||
{name: "fourth retry", retryCount: 4, want: 4 * time.Second},
|
||||
{name: "fifth retry", retryCount: 5, want: 8 * time.Second},
|
||||
{name: "capped retry", retryCount: 10, want: 8 * time.Second},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
require.Equal(t, tc.want, sameAccountRetryDelayFor(capacityErr, tc.retryCount))
|
||||
})
|
||||
}
|
||||
|
||||
t.Run("non request scoped errors keep fixed delay", func(t *testing.T) {
|
||||
require.Equal(t, 500*time.Millisecond, sameAccountRetryDelayFor(&service.UpstreamFailoverError{}, 10))
|
||||
})
|
||||
|
||||
t.Run("nil error keeps fixed delay", func(t *testing.T) {
|
||||
require.Equal(t, 500*time.Millisecond, sameAccountRetryDelayFor(nil, 10))
|
||||
})
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Helper
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -288,16 +288,18 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
|
||||
retryLimit := account.GetPoolModeRetryCount()
|
||||
if sameAccountRetryCount[account.ID] < retryLimit {
|
||||
sameAccountRetryCount[account.ID]++
|
||||
retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID])
|
||||
reqLog.Warn("openai_chat_completions.pool_mode_same_account_retry",
|
||||
zap.Int64("account_id", account.ID),
|
||||
zap.Int("upstream_status", failoverErr.StatusCode),
|
||||
zap.Int("retry_limit", retryLimit),
|
||||
zap.Int("retry_count", sameAccountRetryCount[account.ID]),
|
||||
zap.Duration("retry_delay", retryDelay),
|
||||
)
|
||||
select {
|
||||
case <-c.Request.Context().Done():
|
||||
return
|
||||
case <-time.After(sameAccountRetryDelay):
|
||||
case <-time.After(retryDelay):
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -610,16 +610,18 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
|
||||
retryLimit := account.GetPoolModeRetryCount()
|
||||
if sameAccountRetryCount[account.ID] < retryLimit {
|
||||
sameAccountRetryCount[account.ID]++
|
||||
retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID])
|
||||
reqLog.Warn("openai.pool_mode_same_account_retry",
|
||||
zap.Int64("account_id", account.ID),
|
||||
zap.Int("upstream_status", failoverErr.StatusCode),
|
||||
zap.Int("retry_limit", retryLimit),
|
||||
zap.Int("retry_count", sameAccountRetryCount[account.ID]),
|
||||
zap.Duration("retry_delay", retryDelay),
|
||||
)
|
||||
select {
|
||||
case <-c.Request.Context().Done():
|
||||
return
|
||||
case <-time.After(sameAccountRetryDelay):
|
||||
case <-time.After(retryDelay):
|
||||
}
|
||||
continue
|
||||
}
|
||||
@@ -1158,16 +1160,18 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
|
||||
retryLimit := account.GetPoolModeRetryCount()
|
||||
if sameAccountRetryCount[account.ID] < retryLimit {
|
||||
sameAccountRetryCount[account.ID]++
|
||||
retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID])
|
||||
reqLog.Warn("openai_messages.pool_mode_same_account_retry",
|
||||
zap.Int64("account_id", account.ID),
|
||||
zap.Int("upstream_status", failoverErr.StatusCode),
|
||||
zap.Int("retry_limit", retryLimit),
|
||||
zap.Int("retry_count", sameAccountRetryCount[account.ID]),
|
||||
zap.Duration("retry_delay", retryDelay),
|
||||
)
|
||||
select {
|
||||
case <-c.Request.Context().Done():
|
||||
return
|
||||
case <-time.After(sameAccountRetryDelay):
|
||||
case <-time.After(retryDelay):
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user