From 953028718d590dc64a79899f29f46c7c44f012cd Mon Sep 17 00:00:00 2001 From: IanShaw <131567472+IanShaw027@users.noreply.github.com> Date: Thu, 20 Aug 2026 01:02:40 -0700 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20Grok=20=E9=94=99=E8=AF=AF?= =?UTF-8?q?=E5=88=86=E7=B1=BB=E4=B8=8E=E5=AE=B9=E9=87=8F=E9=87=8D=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/handler/failover_loop.go | 17 ++- backend/internal/service/gateway_service.go | 14 +-- .../internal/service/grok_upstream_errors.go | 2 +- .../internal/service/grok_upstream_failure.go | 102 +++++++++++++++++- .../service/grok_upstream_failure_test.go | 72 +++++++++++++ 5 files changed, 192 insertions(+), 15 deletions(-) diff --git a/backend/internal/handler/failover_loop.go b/backend/internal/handler/failover_loop.go index edbe4e232a..669cfcf365 100644 --- a/backend/internal/handler/failover_loop.go +++ b/backend/internal/handler/failover_loop.go @@ -56,7 +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.RequestScopedTransient || retryCount <= 1 { + if failoverErr == nil { + return sameAccountRetryDelay + } + if failoverErr.SameAccountRetryDelay > 0 { + return failoverErr.SameAccountRetryDelay + } + if !failoverErr.RequestScopedTransient || retryCount <= 1 { return sameAccountRetryDelay } @@ -158,14 +164,19 @@ func (s *FailoverState) HandleFailoverError( } // 同账号重试不算切换账号，粘性会话仅在实际切换时强制缓存计费。 - sameAccountRetry := failoverErr.RetryableOnSameAccount && s.SameAccountRetryCount[accountID] < retryLimit + retryCount := s.SameAccountRetryCount[accountID] + sameAccountRetryAllowed := failoverErr.RetryableOnSameAccount && retryLimit > 0 && retryCount < retryLimit + if sameAccountRetryAllowed && !failoverErr.SameAccountRetryDeadline.IsZero() { + sameAccountRetryAllowed = time.Now().Before(failoverErr.SameAccountRetryDeadline) + } + sameAccountRetry := sameAccountRetryAllowed if needForceCacheBilling(s.hasBoundSession, failoverErr, sameAccountRetry) { s.ForceCacheBilling = true } // 同账号重试：对 RetryableOnSameAccount 的临时性错误，先在同一账号上重试。 // 重试次数上限 retryLimit 由调用方传入（账号级 pool_mode_retry_count 配置）。 - if failoverErr.RetryableOnSameAccount && s.SameAccountRetryCount[accountID] < retryLimit { + if sameAccountRetryAllowed { s.SameAccountRetryCount[accountID]++ retryDelay := sameAccountRetryDelayFor(failoverErr, s.SameAccountRetryCount[accountID]) logger.FromContext(ctx).Warn("gateway.failover_same_account_retry", diff --git a/backend/internal/service/gateway_service.go b/backend/internal/service/gateway_service.go index 20e7e5b9e6..6e1ebecbbb 100644 --- a/backend/internal/service/gateway_service.go +++ b/backend/internal/service/gateway_service.go @@ -671,12 +671,14 @@ type GatewayFailureReason string // source-compatible and preserves their legacy retry-next-account behavior. type UpstreamFailoverError struct { StatusCode int - ResponseBody []byte // 上游响应体，用于错误透传规则匹配 - ResponseHeaders http.Header // 上游响应头，用于透传 cf-ray/cf-mitigated/content-type 等诊断信息 - ForceCacheBilling bool // Antigravity 粘性会话切换时设为 true - RetryableOnSameAccount bool // 临时性错误（如 Google 间歇性 400、空响应），应在同一账号上重试 N 次再切换 - RequestScopedTransient bool // 故障因素与账号无关（如上游按客户端身份/模型容量降载）：可同账号重试，但不得据此对账号做临时封禁 - SafeToFailoverAfterWrite bool // 仅写出 SSE 注释等非语义字节时，仍可在同一客户端流中切换账号 + ResponseBody []byte // 上游响应体，用于错误透传规则匹配 + ResponseHeaders http.Header // 上游响应头，用于透传 cf-ray/cf-mitigated/content-type 等诊断信息 + ForceCacheBilling bool // Antigravity 粘性会话切换时设为 true + RetryableOnSameAccount bool // 临时性错误（如 Google 间歇性 400、空响应），应在同一账号上重试 N 次再切换 + SameAccountRetryDelay time.Duration // 同账号重试的最小间隔；零值使用 handler 默认值 + SameAccountRetryDeadline time.Time // 同账号重试截止时间；零值表示仅受 retryLimit 限制 + RequestScopedTransient bool // 故障因素与账号无关（如上游按客户端身份/模型容量降载）：可同账号重试，但不得据此对账号做临时封禁 + SafeToFailoverAfterWrite bool // 仅写出 SSE 注释等非语义字节时，仍可在同一客户端流中切换账号 Stage GatewayFailureStage Scope GatewayFailureScope Reason GatewayFailureReason diff --git a/backend/internal/service/grok_upstream_errors.go b/backend/internal/service/grok_upstream_errors.go index 0039ea4cdc..1d1bddc6d6 100644 --- a/backend/internal/service/grok_upstream_errors.go +++ b/backend/internal/service/grok_upstream_errors.go @@ -199,7 +199,7 @@ func (s *OpenAIGatewayService) shouldFailoverGrokUpstreamError(statusCode int, r } decision := classifyGrokUpstreamFailure(statusCode, responseBody, "") switch decision.Class { - case GrokFailureFreeUsage, GrokFailureEmptyUpstream, GrokFailureBilling, GrokFailureModelCapacity: + case GrokFailureFreeUsage, GrokFailureEmptyUpstream, GrokFailureBilling, GrokFailureModelCapacity, GrokFailureCompatibility: return decision.ShouldFailover } return s.shouldFailoverUpstreamError(statusCode) diff --git a/backend/internal/service/grok_upstream_failure.go b/backend/internal/service/grok_upstream_failure.go index e7bed08bf0..475c6cff16 100644 --- a/backend/internal/service/grok_upstream_failure.go +++ b/backend/internal/service/grok_upstream_failure.go @@ -26,6 +26,10 @@ const ( GrokFailureRateLimit GrokUpstreamFailureClass = "rate_limit" GrokFailureAuth GrokUpstreamFailureClass = "auth_error" GrokFailureServer GrokUpstreamFailureClass = "server_error" + // GrokFailureCompatibility represents a request-history/body-shape that is + // incompatible with the selected account or upstream replay contract. It + // is account-independent: fail over, but never quarantine the pool. + GrokFailureCompatibility GrokUpstreamFailureClass = "compatibility_error" ) // GrokUpstreamFailureDecision is a pure classification result. Callers map it @@ -111,6 +115,20 @@ func classifyGrokUpstreamFailure(statusCode int, responseBody []byte, requestedM } } + // Responses replay/compaction payloads can be rejected by one Grok + // deployment while the same request is valid on another account. Treat + // these precise decoder/content-shape failures as account compatibility + // errors, rather than durable account health failures. + if isGrokCompatibilityError(low, code) { + return GrokUpstreamFailureDecision{ + Class: GrokFailureCompatibility, + Model: model, + ShouldFailover: true, + ShouldCooldown: false, + Reason: firstNonEmpty(text, "grok response compatibility error"), + } + } + // Empty HTTP 200 / empty model output (often rewritten to synthetic 502). if isGrokEmptyModelOutputText(low) || isGrokEmptyModelOutputCode(code) { return GrokUpstreamFailureDecision{ @@ -355,7 +373,8 @@ func isGrokBillingQuotaText(low string) bool { if strings.Contains(low, "payment") && (strings.Contains(low, "required") || strings.Contains(low, "fail")) { return true } - if strings.Contains(low, "spending limit") || strings.Contains(low, "run out of credits") || strings.Contains(low, "out of credits") { + if strings.Contains(low, "spending limit") || strings.Contains(low, "run out of credits") || strings.Contains(low, "out of credits") || + (strings.Contains(low, "need a grok subscription") || strings.Contains(low, "need grok subscription")) { return true } if strings.Contains(low, "余额不足") || strings.Contains(low, "欠费") || strings.Contains(low, "需要付费") { @@ -364,6 +383,73 @@ func isGrokBillingQuotaText(low string) bool { return false } +// grokRetryableOnSameAccount marks transient 429 classes for the shared +// failover loop. Capacity and ordinary throttles are request/model pressure, +// not evidence that the credential is invalid, so a bounded retry on the same +// account is preferable before switching accounts. Free-usage and billing +// exhaustion deliberately skip same-account retry and fail over immediately. +func grokRetryableOnSameAccount(account *Account, statusCode int, responseBody []byte) bool { + if account == nil || !account.IsGrok() { + return false + } + decision := classifyGrokUpstreamFailure(statusCode, responseBody, "") + switch decision.Class { + case GrokFailureFreeUsage, GrokFailureBilling, GrokFailureCompatibility: + // Quota/entitlement exhaustion is account state, not transient + // pressure. Retrying the same account only repeats the failure. + return false + case GrokFailureModelCapacity: + if statusCode == http.StatusTooManyRequests { + return true + } + } + return account.IsPoolMode() && account.IsPoolModeRetryableStatus(statusCode) +} + +func grokSameAccountRetryMetadata(account *Account, statusCode int, responseBody []byte) (bool, time.Duration, time.Time) { + if !grokRetryableOnSameAccount(account, statusCode, responseBody) { + return false, 0, time.Time{} + } + decision := classifyGrokUpstreamFailure(statusCode, responseBody, "") + if decision.Class != GrokFailureModelCapacity { + return true, 0, time.Time{} + } + return true, 500 * time.Millisecond, time.Now().Add(30 * time.Second) +} + +func isGrokCompatibilityError(low, code string) bool { + combined := strings.ToLower(strings.TrimSpace(low + " " + code)) + // Compaction blobs are account/session-bound and frequently fail with 400 + // or 422 after a reconnect. Also cover xAI's JSON decoder shape errors. + for _, phrase := range []string{ + "could not decode the compaction blob", + "cannot decode the compaction blob", + "decode the compaction blob", + "ensure it is unmodified from the compact response", + "compaction blob", + "failed to deserialize the json body", + "data did not match any variant", + "untagged enum content", + "invalid content shape", + "invalid response history", + } { + if strings.Contains(combined, phrase) { + return true + } + } + for _, marker := range []string{ + "invalid_compaction", + "compaction_decode_error", + "response_history_incompatible", + "invalid_content_shape", + } { + if strings.Contains(combined, marker) { + return true + } + } + return false +} + func isGrokModelCapacityText(low string) bool { return strings.Contains(low, "capacity") || strings.Contains(low, "overloaded") || @@ -489,10 +575,11 @@ func (s *OpenAIGatewayService) applyGrokUpstreamFailureDecision( case GrokFailureEmptyUpstream: reason = "grok empty model output" case GrokFailureModelCapacity: - if persistGrokTransientModelCooldown(account, decision) { - return true - } - reason = "grok model capacity" + // Capacity is scoped to the requested model. Never persist an account-wide + // unschedulable state for this transient class; the failover loop performs + // a bounded same-account retry before selecting another account. + _ = persistGrokTransientModelCooldown(account, decision) + return true case GrokFailureRateLimit: // Pure 429 without free-usage language keeps the existing rate-limit // snapshot path (Retry-After / quota headers). Body-only rate-limit @@ -501,6 +588,11 @@ func (s *OpenAIGatewayService) applyGrokUpstreamFailureDecision( return false case GrokFailureServer: reason = "grok upstream temporary error" + case GrokFailureCompatibility: + // Deliberately no account mutation. The caller uses ShouldFailover to + // retry another account; cooling a pool for a request-shape mismatch + // would remove healthy accounts. + return true default: return false } diff --git a/backend/internal/service/grok_upstream_failure_test.go b/backend/internal/service/grok_upstream_failure_test.go index 0bd1063577..358c1595d3 100644 --- a/backend/internal/service/grok_upstream_failure_test.go +++ b/backend/internal/service/grok_upstream_failure_test.go @@ -61,6 +61,44 @@ func TestClassifyGrokUpstreamFailure_Billing(t *testing.T) { require.True(t, d.ShouldFailover) } +func TestClassifyGrokUpstreamFailure_GrokSubscriptionRequiredIsBilling(t *testing.T) { + d := classifyGrokUpstreamFailure(http.StatusPaymentRequired, + []byte(`{"error":{"message":"You have run out of credits or need a Grok subscription"}}`), "grok-4.6") + require.Equal(t, GrokFailureBilling, d.Class) + require.True(t, d.ShouldFailover) + require.True(t, d.ShouldCooldown) +} + +func TestGrokRetryableOnSameAccount_CapacityAndRateLimit(t *testing.T) { + account := &Account{ID: 9105, Platform: PlatformGrok, Type: AccountTypeOAuth} + require.True(t, grokRetryableOnSameAccount(account, http.StatusTooManyRequests, + []byte(`{"error":{"message":"The model is currently at capacity due to high demand"}}`))) + require.False(t, grokRetryableOnSameAccount(account, http.StatusTooManyRequests, + []byte(`{"error":{"message":"rate limit exceeded"}}`))) + require.False(t, grokRetryableOnSameAccount(account, http.StatusPaymentRequired, + []byte(`{"error":{"message":"You have run out of credits or need a Grok subscription"}}`))) + require.False(t, grokRetryableOnSameAccount(account, http.StatusBadRequest, + []byte(`{"error":{"message":"capacity field is invalid"}}`))) + nonGrok := &Account{ID: 9106, Platform: PlatformOpenAI, Type: AccountTypeOAuth} + require.False(t, grokRetryableOnSameAccount(nonGrok, http.StatusTooManyRequests, + []byte(`{"error":{"message":"model at capacity"}}`))) +} + +func TestGrokSameAccountRetryMetadata_CapacityDeadline(t *testing.T) { + account := &Account{ID: 9107, Platform: PlatformGrok, Type: AccountTypeOAuth} + retryable, delay, deadline := grokSameAccountRetryMetadata(account, http.StatusTooManyRequests, + []byte(`{"error":{"message":"model capacity exceeded"}}`)) + require.True(t, retryable) + require.Equal(t, 500*time.Millisecond, delay) + require.WithinDuration(t, time.Now().Add(30*time.Second), deadline, 2*time.Second) + + retryable, delay, deadline = grokSameAccountRetryMetadata(account, http.StatusTooManyRequests, + []byte(`{"error":{"message":"rate limit exceeded"}}`)) + require.False(t, retryable) + require.Zero(t, delay) + require.True(t, deadline.IsZero()) +} + func TestClassifyGrokUpstreamFailure_ValidationNoCool(t *testing.T) { d := classifyGrokUpstreamFailure(http.StatusBadRequest, []byte(`{"error":{"message":"invalid tool schema"}}`), "") require.Equal(t, GrokFailureNone, d.Class) @@ -75,12 +113,33 @@ func TestClassifyGrokUpstreamFailure_FreeUsageWinsOver5xx(t *testing.T) { require.NotEqual(t, GrokFailureServer, d.Class) } +func TestClassifyGrokUpstreamFailure_CompatibilityDoesNotCooldown(t *testing.T) { + cases := []string{ + `{"error":{"message":"Could not decode the compaction blob. Ensure it is unmodified from the compact response"}}`, + `{"error":{"message":"data did not match any variant of the untagged enum content"}}`, + `{"code":"compaction_decode_error","message":"invalid response history"}`, + } + for _, body := range cases { + d := classifyGrokUpstreamFailure(http.StatusUnprocessableEntity, []byte(body), "grok-4.6") + require.Equal(t, GrokFailureCompatibility, d.Class, body) + require.True(t, d.ShouldFailover, body) + require.False(t, d.ShouldCooldown, body) + require.Zero(t, d.Cooldown, body) + } +} + func TestShouldFailoverGrokUpstreamError_FreeUsageBody(t *testing.T) { svc := &OpenAIGatewayService{} body := []byte(`{"error":{"code":"subscription:free-usage-exhausted","message":"free usage exhausted"}}`) require.True(t, svc.shouldFailoverGrokUpstreamError(http.StatusBadRequest, body)) } +func TestShouldFailoverGrokUpstreamError_CompatibilityBody(t *testing.T) { + svc := &OpenAIGatewayService{} + body := []byte(`{"error":{"message":"Could not decode the compaction blob"}}`) + require.True(t, svc.shouldFailoverGrokUpstreamError(http.StatusUnprocessableEntity, body)) +} + func TestShouldFailoverGrokUpstreamError_ContentPolicyStillNoFailover(t *testing.T) { svc := &OpenAIGatewayService{} body := []byte(`{"error":{"code":"new_sensitive","message":"text is sensitive"}}`) @@ -149,6 +208,19 @@ func TestHandleGrokAccountUpstreamError_MultiAgentCapacityBlocksOnlyThatModel(t require.False(t, isGrokModelQuotaBlocked(account.ID, "grok-4.5", time.Now())) } +func TestHandleGrokAccountUpstreamError_CapacityNeverCoolsAccount(t *testing.T) { + repo := &grokQuotaAccountRepo{} + svc := &OpenAIGatewayService{accountRepo: repo} + account := &Account{ID: 9121, Platform: PlatformGrok, Type: AccountTypeOAuth} + ctx := withGrokTeamRateLimitModel(context.Background(), "grok-4.6") + + svc.handleGrokAccountUpstreamError(ctx, account, http.StatusTooManyRequests, nil, + []byte(`{"error":{"message":"The model is currently at capacity due to high demand"}}`)) + + require.Zero(t, repo.tempUnschedCalls) + require.False(t, svc.isOpenAIAccountRuntimeBlocked(account)) +} + func TestHandleGrokAccountUpstreamError_FreeUsageDoesNotCoolPoolMode(t *testing.T) { repo := &grokQuotaAccountRepo{} svc := &OpenAIGatewayService{accountRepo: repo}