合并上游 main,保留 5925 的 Grok 重试上限与 5888 的协议兼容修复

同账号重试采用次数上限加 deadline;畸形 tools 在出站前删除;compaction 422 与结构化错误扫描一并保留。
This commit is contained in:
IanShaw027 committed 2026-08-21 08:49:44 +08:00
commit 7c53a842e4
51 files changed
+1414 -240

No files matched your search

+41 -7
View File
@@ -56,10 +56,13 @@ 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.SameAccountRetryDelay > 0 {
if failoverErr == nil {
return sameAccountRetryDelay
}
if failoverErr.SameAccountRetryDelay > 0 {
return failoverErr.SameAccountRetryDelay
}
if failoverErr == nil || !failoverErr.RequestScopedTransient || retryCount <= 1 {
if !failoverErr.RequestScopedTransient || retryCount <= 1 {
return sameAccountRetryDelay
}
@@ -77,10 +80,37 @@ func sameAccountRetryAllowed(failoverErr *service.UpstreamFailoverError, retryCo
if failoverErr == nil || !failoverErr.RetryableOnSameAccount {
return false
}
if !failoverErr.SameAccountRetryDeadline.IsZero() {
return time.Now().Before(failoverErr.SameAccountRetryDeadline)
if !sameAccountRetryDeadlineAllows(failoverErr) {
return false
}
return retryCount < retryLimit
// Deadline-window retries (OAuth 429) may pass retryLimit=0 and are not
// bound to pool_mode_retry_count. Pool-mode callers pass a positive limit.
if !failoverErr.SameAccountRetryDeadline.IsZero() && retryLimit <= 0 {
return true
}
if failoverErr.SameAccountRetryMax > 0 && (retryLimit <= 0 || failoverErr.SameAccountRetryMax < retryLimit) {
retryLimit = failoverErr.SameAccountRetryMax
}
return retryLimit > 0 && retryCount < retryLimit
}
// sameAccountRetryDeadlineAllows prevents a retry from starting after the
// service-provided same-account retry window has elapsed.
func sameAccountRetryDeadlineAllows(failoverErr *service.UpstreamFailoverError) bool {
return failoverErr == nil || failoverErr.SameAccountRetryDeadline.IsZero() || time.Now().Before(failoverErr.SameAccountRetryDeadline)
}
// effectiveSameAccountRetryLimit applies an error-specific cap without
// overriding an explicit account setting of zero (which disables retries).
func effectiveSameAccountRetryLimit(failoverErr *service.UpstreamFailoverError, account *service.Account) int {
if account == nil {
return 0
}
limit := account.GetPoolModeRetryCount()
if limit > 0 && failoverErr != nil && failoverErr.SameAccountRetryMax > 0 && failoverErr.SameAccountRetryMax < limit {
return failoverErr.SameAccountRetryMax
}
return limit
}
// FailoverState 跨循环迭代共享的 failover 状态
@@ -171,14 +201,18 @@ func (s *FailoverState) HandleFailoverError(
}
// 同账号重试不算切换账号,粘性会话仅在实际切换时强制缓存计费。
sameAccountRetry := sameAccountRetryAllowed(failoverErr, s.SameAccountRetryCount[accountID], retryLimit)
retryCount := s.SameAccountRetryCount[accountID]
if failoverErr.SameAccountRetryMax > 0 && failoverErr.SameAccountRetryMax < retryLimit {
retryLimit = failoverErr.SameAccountRetryMax
}
sameAccountRetry := failoverErr.RetryableOnSameAccount && retryLimit > 0 && retryCount < retryLimit && sameAccountRetryDeadlineAllows(failoverErr)
if needForceCacheBilling(s.hasBoundSession, failoverErr, sameAccountRetry) {
s.ForceCacheBilling = true
}
// 同账号重试:对 RetryableOnSameAccount 的临时性错误,先在同一账号上重试。
// 重试次数上限 retryLimit 由调用方传入(账号级 pool_mode_retry_count 配置)。
if sameAccountRetryAllowed(failoverErr, s.SameAccountRetryCount[accountID], retryLimit) {
if sameAccountRetry {
s.SameAccountRetryCount[accountID]++
retryDelay := sameAccountRetryDelayFor(failoverErr, s.SameAccountRetryCount[accountID])
logger.FromContext(ctx).Warn("gateway.failover_same_account_retry",
+20 -3
View File
@@ -75,6 +75,23 @@ func TestSameAccountRetryAllowedUsesDeadlineInsteadOfPoolCount(t *testing.T) {
require.False(t, sameAccountRetryAllowed(err, 0, 100))
}
func TestSameAccountRetryDeadlineAllows(t *testing.T) {
require.True(t, sameAccountRetryDeadlineAllows(&service.UpstreamFailoverError{}))
require.True(t, sameAccountRetryDeadlineAllows(&service.UpstreamFailoverError{
SameAccountRetryDeadline: time.Now().Add(time.Second),
}))
require.False(t, sameAccountRetryDeadlineAllows(&service.UpstreamFailoverError{
SameAccountRetryDeadline: time.Now().Add(-time.Second),
}))
}
func TestEffectiveSameAccountRetryLimitHonorsErrorCapAndDisabledAccount(t *testing.T) {
account := &service.Account{Type: service.AccountTypeAPIKey, Credentials: map[string]any{"pool_mode": true, "pool_mode_retry_count": float64(3)}}
require.Equal(t, 1, effectiveSameAccountRetryLimit(&service.UpstreamFailoverError{SameAccountRetryMax: 1}, account))
account.Credentials["pool_mode_retry_count"] = float64(0)
require.Equal(t, 0, effectiveSameAccountRetryLimit(&service.UpstreamFailoverError{SameAccountRetryMax: 1}, account))
}
// ---------------------------------------------------------------------------
// Helper
// ---------------------------------------------------------------------------
@@ -335,7 +352,7 @@ func TestHandleFailoverError_CacheBilling(t *testing.T) {
require.Zero(t, fs.SwitchCount)
})
t.Run("deadline允许超过计数上限时仍不强制缓存计费", func(t *testing.T) {
t.Run("deadline存在但计数已耗尽时按切换处理并强制缓存计费", func(t *testing.T) {
mock := &mockTempUnscheduler{}
fs := NewFailoverState(3, true)
fs.SameAccountRetryCount[100] = maxSameAccountRetries
@@ -345,8 +362,8 @@ func TestHandleFailoverError_CacheBilling(t *testing.T) {
fs.HandleFailoverError(context.Background(), mock, 100, "openai", maxSameAccountRetries, err)
require.False(t, fs.ForceCacheBilling)
require.Zero(t, fs.SwitchCount)
require.True(t, fs.ForceCacheBilling)
require.Equal(t, 1, fs.SwitchCount)
})
t.Run("同账号重试耗尽并实际切换时设置ForceCacheBilling", func(t *testing.T) {
+75 -34
View File
@@ -44,39 +44,84 @@ func (h *OpenAIGatewayHandler) GrokRealtime(c *gin.Context) {
return
}
selection, _, err := h.gatewayService.SelectAccountWithSchedulerForCapability(
c.Request.Context(),
apiKey.GroupID,
"",
"",
"grok-4.5",
nil,
service.OpenAIUpstreamTransportHTTPSSE,
// Grok only advertises chat_completions + media capabilities on HEAD.
service.OpenAIEndpointCapabilityChatCompletions,
false,
false,
false,
service.PlatformGrok,
)
if err != nil || selection == nil || selection.Account == nil {
h.errorResponse(c, http.StatusServiceUnavailable, "api_error", "No available Grok accounts")
return
}
var streamStarted bool
reqLog := requestLogger(c, "handler.openai_gateway.grok_realtime")
release, slotStatus := h.acquireResponsesAccountSlot(c, apiKey.GroupID, "", selection, true, &streamStarted, reqLog)
if slotStatus != openAISlotAcquireOK {
model := c.Query("model")
if strings.TrimSpace(model) == "" {
model = "grok-voice-latest"
}
// Keep the HTTP response uncommitted while selecting and probing an account.
// Realtime is not an HTTP streaming response; using reqStream=true here would
// let the wait queue flush an SSE ping before the WebSocket handshake succeeds.
failed := map[int64]struct{}{}
var selection *service.AccountSelectionResult
var release func()
var token string
var upstream *service.GrokRealtimeUpstream
var candidateSeen bool
for attempts := 0; attempts < 4; attempts++ {
// Realtime's voice model is not a text-model capability. Passing a
// concrete text model here would reject accounts mapped only to an
// older/default text model before the upstream handshake can decide.
// An empty requested model keeps account selection capability-based;
// the actual voice model remains in the upstream WS query below.
candidate, _, selectErr := h.gatewayService.SelectAccountWithSchedulerForCapability(
c.Request.Context(), apiKey.GroupID, "", "", "", failed,
service.OpenAIUpstreamTransportHTTPSSE,
service.OpenAIEndpointCapabilityChatCompletions,
false, false, false, service.PlatformGrok,
)
if selectErr != nil || candidate == nil || candidate.Account == nil {
break
}
candidateSeen = true
account := candidate.Account
var streamStarted bool
var slotStatus openAISlotAcquireResult
release, slotStatus = h.acquireResponsesAccountSlot(c, apiKey.GroupID, "", candidate, false, &streamStarted, reqLog)
if slotStatus != openAISlotAcquireOK {
if slotStatus == openAISlotAcquireFailed {
return
}
failed[account.ID] = struct{}{}
continue
}
var credErr error
token, _, credErr = h.gatewayService.GetRequestCredential(c.Request.Context(), c, account)
if credErr != nil {
release()
release = nil
failed[account.ID] = struct{}{}
continue
}
probeCtx, cancelProbe := context.WithTimeout(c.Request.Context(), service.DefaultGrokRealtimeDialTimeout)
candidateUpstream, openErr := h.gatewayService.OpenGrokRealtime(probeCtx, account, token, model)
cancelProbe()
if openErr != nil {
reqLog.Warn("grok_realtime.pre_accept_failed", zap.Int64("account_id", account.ID), zap.Error(openErr))
statusCode := http.StatusBadGateway
var dialErr *service.GrokRealtimeDialError
if errors.As(openErr, &dialErr) && dialErr.StatusCode > 0 {
statusCode = dialErr.StatusCode
}
h.gatewayService.HandleGrokRealtimeUpstreamError(c.Request.Context(), account, statusCode, []byte(openErr.Error()))
release()
release = nil
failed[account.ID] = struct{}{}
continue
}
selection, upstream = candidate, candidateUpstream
break
}
if selection == nil || selection.Account == nil || release == nil || upstream == nil {
if !candidateSeen {
h.errorResponse(c, http.StatusServiceUnavailable, "api_error", "No available Grok accounts")
} else {
h.errorResponse(c, http.StatusBadGateway, "upstream_error", "Grok realtime upstream unavailable")
}
return
}
defer release()
token, _, err := h.gatewayService.GetRequestCredential(c.Request.Context(), c, selection.Account)
if err != nil {
h.errorResponse(c, http.StatusBadGateway, "upstream_error", "Grok credential unavailable")
return
}
defer func() { _ = upstream.Close() }()
conn, err := coderws.Accept(c.Writer, c.Request, &coderws.AcceptOptions{CompressionMode: coderws.CompressionContextTakeover})
if err != nil {
@@ -84,12 +129,8 @@ func (h *OpenAIGatewayHandler) GrokRealtime(c *gin.Context) {
}
defer func() { _ = conn.CloseNow() }()
model := c.Query("model")
if strings.TrimSpace(model) == "" {
model = "grok-voice-latest"
}
started := time.Now()
audioObserved, proxyErr := h.gatewayService.ProxyGrokRealtime(c.Request.Context(), c, conn, selection.Account, token, model)
audioObserved, proxyErr := h.gatewayService.ProxyGrokRealtimeConn(c.Request.Context(), c, conn, upstream)
elapsed := time.Since(started)
if proxyErr != nil {
reqLog.Info("grok_realtime.proxy_failed", zap.Error(proxyErr))
+5 -3
View File
@@ -361,19 +361,21 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service.
return
}
if failoverErr.RetryableOnSameAccount {
retryLimit := account.GetPoolModeRetryCount()
if sameAccountRetryCount[account.ID] < retryLimit {
retryLimit := effectiveSameAccountRetryLimit(failoverErr, account)
if sameAccountRetryCount[account.ID] < retryLimit && sameAccountRetryDeadlineAllows(failoverErr) {
sameAccountRetryCount[account.ID]++
retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID])
reqLog.Warn("grok_media.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 <-requestCtx.Done():
return
case <-time.After(sameAccountRetryDelay):
case <-time.After(retryDelay):
}
continue
}
@@ -330,8 +330,8 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
}
// Pool mode: retry on the same account
if failoverErr.RetryableOnSameAccount {
retryLimit := account.GetPoolModeRetryCount()
if sameAccountRetryAllowed(failoverErr, sameAccountRetryCount[account.ID], retryLimit) {
retryLimit := effectiveSameAccountRetryLimit(failoverErr, account)
if sameAccountRetryCount[account.ID] < retryLimit && sameAccountRetryDeadlineAllows(failoverErr) {
sameAccountRetryCount[account.ID]++
retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID])
reqLog.Warn("openai_chat_completions.pool_mode_same_account_retry",
@@ -754,8 +754,8 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
}
// 池模式:同账号重试
if failoverErr.RetryableOnSameAccount {
retryLimit := account.GetPoolModeRetryCount()
if sameAccountRetryAllowed(failoverErr, sameAccountRetryCount[account.ID], retryLimit) {
retryLimit := effectiveSameAccountRetryLimit(failoverErr, account)
if sameAccountRetryCount[account.ID] < retryLimit && sameAccountRetryDeadlineAllows(failoverErr) {
sameAccountRetryCount[account.ID]++
retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID])
reqLog.Warn("openai.pool_mode_same_account_retry",
@@ -1309,8 +1309,8 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
}
// 池模式:同账号重试
if failoverErr.RetryableOnSameAccount {
retryLimit := account.GetPoolModeRetryCount()
if sameAccountRetryAllowed(failoverErr, sameAccountRetryCount[account.ID], retryLimit) {
retryLimit := effectiveSameAccountRetryLimit(failoverErr, account)
if sameAccountRetryCount[account.ID] < retryLimit && sameAccountRetryDeadlineAllows(failoverErr) {
sameAccountRetryCount[account.ID]++
retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID])
reqLog.Warn("openai_messages.pool_mode_same_account_retry",
@@ -672,7 +672,7 @@ func TestResolveOpenAIMessagesDispatchMappedModel(t *testing.T) {
Platform: service.PlatformGrok,
},
}
require.Equal(t, "grok-4.5", resolveOpenAIMessagesDispatchMappedModel(nil, apiKey, "claude-sonnet-4-5"))
require.Equal(t, "grok-4.6", resolveOpenAIMessagesDispatchMappedModel(nil, apiKey, "claude-sonnet-4-5"))
require.Empty(t, resolveOpenAIMessagesDispatchMappedModel(nil, apiKey, "grok"))
})
+2 -2
View File
@@ -308,8 +308,8 @@ func (h *OpenAIGatewayHandler) Images(c *gin.Context) {
return
}
if failoverErr.RetryableOnSameAccount {
retryLimit := account.GetPoolModeRetryCount()
if sameAccountRetryAllowed(failoverErr, sameAccountRetryCount[account.ID], retryLimit) {
retryLimit := effectiveSameAccountRetryLimit(failoverErr, account)
if sameAccountRetryCount[account.ID] < retryLimit && sameAccountRetryDeadlineAllows(failoverErr) {
sameAccountRetryCount[account.ID]++
retryDelay := sameAccountRetryDelayFor(failoverErr, sameAccountRetryCount[account.ID])
reqLog.Warn("openai.images.pool_mode_same_account_retry",