diff --git a/backend/internal/service/error_policy_test.go b/backend/internal/service/error_policy_test.go index 2aa7a421d1..178cec9250 100644 --- a/backend/internal/service/error_policy_test.go +++ b/backend/internal/service/error_policy_test.go @@ -209,6 +209,50 @@ func TestCheckErrorPolicy(t *testing.T) { body: []byte(`unauthorized`), expected: ErrorPolicySkipped, }, + { + name: "pool_mode_temp_unschedulable_hit_returns_temp_unscheduled", + account: &Account{ + ID: 9, + Type: AccountTypeAPIKey, + Platform: PlatformOpenAI, + Credentials: map[string]any{ + "pool_mode": true, + "temp_unschedulable_enabled": true, + "temp_unschedulable_rules": []any{ + map[string]any{ + "error_code": float64(http.StatusServiceUnavailable), + "keywords": []any{"unavailable"}, + "duration_minutes": float64(30), + }, + }, + }, + }, + statusCode: http.StatusServiceUnavailable, + body: []byte(`Service temporarily unavailable`), + expected: ErrorPolicyTempUnscheduled, + }, + { + name: "pool_mode_temp_unschedulable_miss_returns_skipped", + account: &Account{ + ID: 10, + Type: AccountTypeAPIKey, + Platform: PlatformOpenAI, + Credentials: map[string]any{ + "pool_mode": true, + "temp_unschedulable_enabled": true, + "temp_unschedulable_rules": []any{ + map[string]any{ + "error_code": float64(http.StatusServiceUnavailable), + "keywords": []any{"maintenance"}, + "duration_minutes": float64(30), + }, + }, + }, + }, + statusCode: http.StatusServiceUnavailable, + body: []byte(`Service temporarily unavailable`), + expected: ErrorPolicySkipped, + }, } for _, tt := range tests { @@ -222,7 +266,7 @@ func TestCheckErrorPolicy(t *testing.T) { } } -func TestHandleUpstreamError_PoolModeCustomErrorCodesOverride(t *testing.T) { +func TestHandleUpstreamError_PoolModePolicies(t *testing.T) { t.Run("pool_mode_without_custom_error_codes_still_skips", func(t *testing.T) { repo := &errorPolicyRepoStub{} svc := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) @@ -262,6 +306,72 @@ func TestHandleUpstreamError_PoolModeCustomErrorCodesOverride(t *testing.T) { require.Equal(t, 1, repo.setErrCalls) require.Equal(t, 0, repo.tempCalls) }) + + t.Run("pool_mode_explicit_temp_rule_stops_scheduling", func(t *testing.T) { + repo := &errorPolicyRepoStub{} + svc := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + account := &Account{ + ID: 32, + Type: AccountTypeAPIKey, + Platform: PlatformOpenAI, + Credentials: map[string]any{ + "pool_mode": true, + "temp_unschedulable_enabled": true, + "temp_unschedulable_rules": []any{ + map[string]any{ + "error_code": float64(http.StatusServiceUnavailable), + "keywords": []any{"unavailable"}, + "duration_minutes": float64(30), + }, + }, + }, + } + + shouldDisable := svc.HandleUpstreamError( + context.Background(), + account, + http.StatusServiceUnavailable, + http.Header{}, + []byte("Service temporarily unavailable"), + ) + + require.True(t, shouldDisable) + require.Equal(t, 0, repo.setErrCalls) + require.Equal(t, 1, repo.tempCalls) + }) + + t.Run("pool_mode_temp_rule_miss_still_skips", func(t *testing.T) { + repo := &errorPolicyRepoStub{} + svc := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + account := &Account{ + ID: 33, + Type: AccountTypeAPIKey, + Platform: PlatformOpenAI, + Credentials: map[string]any{ + "pool_mode": true, + "temp_unschedulable_enabled": true, + "temp_unschedulable_rules": []any{ + map[string]any{ + "error_code": float64(http.StatusServiceUnavailable), + "keywords": []any{"maintenance"}, + "duration_minutes": float64(30), + }, + }, + }, + } + + shouldDisable := svc.HandleUpstreamError( + context.Background(), + account, + http.StatusServiceUnavailable, + http.Header{}, + []byte("Service temporarily unavailable"), + ) + + require.False(t, shouldDisable) + require.Equal(t, 0, repo.setErrCalls) + require.Equal(t, 0, repo.tempCalls) + }) } // --------------------------------------------------------------------------- diff --git a/backend/internal/service/openai_account_runtime_block_fastpath_test.go b/backend/internal/service/openai_account_runtime_block_fastpath_test.go index 7d7cfbda50..6a17584196 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath_test.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath_test.go @@ -83,6 +83,58 @@ func TestOpenAIRuntimeBlocker_IgnoresNonOpenAIFromRateLimitService(t *testing.T) require.False(t, gateway.isOpenAIAccountRuntimeBlocked(account)) } +func TestOpenAIPoolModeTempRule_StopsSameAccountRetryAndBlocksAcrossModels(t *testing.T) { + repo := &errorPolicyRepoStub{} + rateLimitService := NewRateLimitService(repo, nil, &config.Config{}, nil, nil) + gateway := &OpenAIGatewayService{ + cfg: &config.Config{}, + rateLimitService: rateLimitService, + } + rateLimitService.SetAccountRuntimeBlocker(gateway) + account := &Account{ + ID: 46, + Platform: PlatformOpenAI, + Type: AccountTypeAPIKey, + Status: StatusActive, + Schedulable: true, + Credentials: map[string]any{ + "pool_mode": true, + "pool_mode_retry_status_codes": []any{float64(http.StatusServiceUnavailable)}, + "temp_unschedulable_enabled": true, + "temp_unschedulable_rules": []any{ + map[string]any{ + "error_code": float64(http.StatusServiceUnavailable), + "keywords": []any{"unavailable"}, + "duration_minutes": float64(30), + }, + }, + }, + } + body := []byte(`{"error":{"message":"Service temporarily unavailable"}}`) + resp := &http.Response{ + StatusCode: http.StatusServiceUnavailable, + Header: http.Header{}, + } + + failoverErr := gateway.failoverOpenAIUpstreamHTTPError( + context.Background(), + nil, + account, + resp, + body, + "Service temporarily unavailable", + "gpt-5.4", + ) + + require.NotNil(t, failoverErr) + require.False(t, failoverErr.RetryableOnSameAccount) + require.Equal(t, 1, repo.tempCalls) + require.Equal(t, 0, repo.setErrCalls) + require.Equal(t, StatusActive, account.Status) + require.True(t, gateway.isOpenAIAccountRequestRuntimeBlocked(account, "gpt-5.4")) + require.True(t, gateway.isOpenAIAccountRequestRuntimeBlocked(account, "gpt-5.5")) +} + func TestOpenAIModelNotFound_DoesNotRuntimeBlockWholeAccount(t *testing.T) { repo := &modelNotFoundAccountRepoStub{} svc := &OpenAIGatewayService{ diff --git a/backend/internal/service/openai_alpha_search.go b/backend/internal/service/openai_alpha_search.go index f11a21958f..488db038e0 100644 --- a/backend/internal/service/openai_alpha_search.go +++ b/backend/internal/service/openai_alpha_search.go @@ -95,13 +95,14 @@ func (s *OpenAIGatewayService) ForwardAlphaSearch(ctx context.Context, c *gin.Co // 被永久标记为 error；历史导入且缺少 auth_mode 标记的 at- token 也会 // 漏过 PAT 类型判断。这里仍允许本次请求换号，但不修改任何账号状态； // 真正的凭据失效由普通 Responses 请求或 whoami 校验判定。 + shouldDisable := false if shouldApplyOpenAIAlphaSearchAccountErrorSideEffects(resp.StatusCode) { - s.handleFailoverSideEffects(ctx, resp, account, respBody, upstreamModel) + shouldDisable = s.handleFailoverSideEffects(ctx, resp, account, respBody, upstreamModel) } return nil, &UpstreamFailoverError{ StatusCode: resp.StatusCode, ResponseBody: respBody, - RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), + RetryableOnSameAccount: !shouldDisable && account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), } } } @@ -169,13 +170,14 @@ func (s *OpenAIGatewayService) forwardAlphaSearchViaResponsesWebSearch( if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMessage, respBody) { resp.Body = io.NopCloser(bytes.NewReader(respBody)) // 仍按 alpha/search 工具请求处理：PAT 的工具链路失败不能直接永久置错。 + shouldDisable := false if shouldApplyOpenAIAlphaSearchAccountErrorSideEffects(resp.StatusCode) { - s.handleFailoverSideEffects(ctx, resp, account, respBody, upstreamModel) + shouldDisable = s.handleFailoverSideEffects(ctx, resp, account, respBody, upstreamModel) } return nil, &UpstreamFailoverError{ StatusCode: resp.StatusCode, ResponseBody: respBody, - RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), + RetryableOnSameAccount: !shouldDisable && account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), } } } diff --git a/backend/internal/service/openai_embeddings.go b/backend/internal/service/openai_embeddings.go index fa821d8f31..c0c5900297 100644 --- a/backend/internal/service/openai_embeddings.go +++ b/backend/internal/service/openai_embeddings.go @@ -132,11 +132,11 @@ func (s *OpenAIGatewayService) ForwardEmbeddings( Message: upstreamMsg, Detail: upstreamDetail, }) - s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, upstreamModel) + shouldDisable := s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, upstreamModel) return nil, &UpstreamFailoverError{ StatusCode: resp.StatusCode, ResponseBody: respBody, - RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), + RetryableOnSameAccount: !shouldDisable && account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), } } writeOpenAIEmbeddingsUpstreamResponse(c, resp, respBody, s.responseHeaderFilter) diff --git a/backend/internal/service/openai_gateway_cc_pipeline.go b/backend/internal/service/openai_gateway_cc_pipeline.go index 52658b2729..533f2ad383 100644 --- a/backend/internal/service/openai_gateway_cc_pipeline.go +++ b/backend/internal/service/openai_gateway_cc_pipeline.go @@ -112,15 +112,16 @@ func (s *OpenAIGatewayService) failoverOpenAIUpstreamHTTPError( Message: upstreamMsg, Detail: upstreamDetail, }) + shouldDisable := false if account.Platform != PlatformGrok { - s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, upstreamModel) + shouldDisable = s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody, upstreamModel) } return newOpenAIUpstreamFailoverError( resp.StatusCode, resp.Header, respBody, upstreamMsg, - account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)), + !shouldDisable && account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)), ) } diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 16c034f19d..93697be774 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -877,13 +877,13 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco Detail: upstreamDetail, }) - s.handleFailoverSideEffects(ctx, resp, account, respBody, upstreamModel) + shouldDisable := s.handleFailoverSideEffects(ctx, resp, account, respBody, upstreamModel) return nil, newOpenAIUpstreamFailoverError( resp.StatusCode, resp.Header, respBody, upstreamMsg, - account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)), + !shouldDisable && account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)), ) } return s.handleErrorResponse(ctx, resp, c, account, body, billingModel) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index a17d050afe..1fb281b4a2 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -580,7 +580,7 @@ func (s *OpenAIGatewayService) handleFailoverErrorResponsePassthrough( logOpenAIInstructionsRequiredDebug(ctx, c, account, resp.StatusCode, upstreamMsg, requestBody, body) reqModel, _, _ := extractOpenAIRequestMetaFromBody(requestBody) canonicalModel := canonicalOpenAIAccountSchedulingModel(account, reqModel) - _ = s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, body, canonicalModel) + shouldDisable := s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, body, canonicalModel) appendOpsUpstreamError(c, OpsUpstreamErrorEvent{ Platform: account.Platform, AccountID: account.ID, @@ -598,7 +598,7 @@ func (s *OpenAIGatewayService) handleFailoverErrorResponsePassthrough( resp.Header, body, upstreamMsg, - account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), + !shouldDisable && account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), ) } diff --git a/backend/internal/service/openai_gateway_upstream_errors.go b/backend/internal/service/openai_gateway_upstream_errors.go index a21775308d..ff97e43fc3 100644 --- a/backend/internal/service/openai_gateway_upstream_errors.go +++ b/backend/internal/service/openai_gateway_upstream_errors.go @@ -305,12 +305,11 @@ func (s *OpenAIGatewayService) readUpstreamErrorBody(resp *http.Response) []byte return body } -func (s *OpenAIGatewayService) handleFailoverSideEffects(ctx context.Context, resp *http.Response, account *Account, responseBody []byte, canonicalModel ...string) { +func (s *OpenAIGatewayService) handleFailoverSideEffects(ctx context.Context, resp *http.Response, account *Account, responseBody []byte, canonicalModel ...string) bool { if len(canonicalModel) > 0 { - s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, responseBody, canonicalModel[0]) - return + return s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, responseBody, canonicalModel[0]) } - s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, responseBody) + return s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, responseBody) } func (s *OpenAIGatewayService) handleErrorResponse( @@ -470,7 +469,7 @@ func (s *OpenAIGatewayService) handleErrorResponse( return nil, &UpstreamFailoverError{ StatusCode: resp.StatusCode, ResponseBody: body, - RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), + RetryableOnSameAccount: false, } } @@ -640,7 +639,7 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse( return nil, &UpstreamFailoverError{ StatusCode: resp.StatusCode, ResponseBody: body, - RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), + RetryableOnSameAccount: false, } } diff --git a/backend/internal/service/openai_images.go b/backend/internal/service/openai_images.go index fa0b7f8b69..f32fc56159 100644 --- a/backend/internal/service/openai_images.go +++ b/backend/internal/service/openai_images.go @@ -652,11 +652,11 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesAPIKey( Kind: "failover", Message: upstreamMsg, }) - s.handleFailoverSideEffects(upstreamCtx, resp, account, respBody, upstreamModel) + shouldDisable := s.handleFailoverSideEffects(upstreamCtx, resp, account, respBody, upstreamModel) return nil, &UpstreamFailoverError{ StatusCode: resp.StatusCode, ResponseBody: respBody, - RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), + RetryableOnSameAccount: !shouldDisable && account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), } } return s.handleOpenAIImagesErrorResponse(upstreamCtx, resp, c, account, upstreamModel) diff --git a/backend/internal/service/openai_images_responses.go b/backend/internal/service/openai_images_responses.go index f5b99ac9ca..7924de4fc3 100644 --- a/backend/internal/service/openai_images_responses.go +++ b/backend/internal/service/openai_images_responses.go @@ -920,7 +920,7 @@ func (s *OpenAIGatewayService) handleOpenAIImagesErrorResponse( return nil, &UpstreamFailoverError{ StatusCode: resp.StatusCode, ResponseBody: body, - RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), + RetryableOnSameAccount: false, } } @@ -1752,11 +1752,11 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesOAuth( Kind: "failover", Message: upstreamMsg, }) - s.handleFailoverSideEffects(upstreamCtx, resp, account, respBody, requestModel) + shouldDisable := s.handleFailoverSideEffects(upstreamCtx, resp, account, respBody, requestModel) return nil, &UpstreamFailoverError{ StatusCode: resp.StatusCode, ResponseBody: respBody, - RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), + RetryableOnSameAccount: !shouldDisable && account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode), } } return s.handleOpenAIImagesErrorResponse(upstreamCtx, resp, c, account, requestModel) @@ -1885,11 +1885,11 @@ func (s *OpenAIGatewayService) handleOpenAIImagesOAuthResponseError( } responseBody := openAIImagesUpstreamErrorResponseBody(upstreamErr) - s.handleOpenAIAccountUpstreamError(ctx, account, upstreamErr.StatusCode, headers, responseBody, requestedModel) + shouldDisable := s.handleOpenAIAccountUpstreamError(ctx, account, upstreamErr.StatusCode, headers, responseBody, requestedModel) return &UpstreamFailoverError{ StatusCode: upstreamErr.StatusCode, ResponseBody: responseBody, ResponseHeaders: headers, - RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(upstreamErr.StatusCode), + RetryableOnSameAccount: !shouldDisable && account.IsPoolMode() && account.IsPoolModeRetryableStatus(upstreamErr.StatusCode), } } diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index 507e02d877..559ee2e05e 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -159,6 +159,11 @@ func (s *RateLimitService) CheckErrorPolicy(ctx context.Context, account *Accoun return ErrorPolicySkipped } if account.IsPoolMode() { + // 池模式只跳过默认账号状态处理；管理员显式配置的临时不可调度规则仍应生效。 + // 401 保留现有认证错误语义，避免改变重复 401 的升级行为。 + if statusCode != http.StatusUnauthorized && s.tryTempUnschedulable(ctx, account, statusCode, responseBody) { + return ErrorPolicyTempUnscheduled + } return ErrorPolicySkipped } if s.tryTempUnschedulable(ctx, account, statusCode, responseBody) { @@ -172,8 +177,12 @@ func (s *RateLimitService) CheckErrorPolicy(ctx context.Context, account *Accoun func (s *RateLimitService) HandleUpstreamError(ctx context.Context, account *Account, statusCode int, headers http.Header, responseBody []byte, requestedModel ...string) (shouldDisable bool) { customErrorCodesEnabled := account.IsCustomErrorCodesEnabled() - // 池模式默认不标记本地账号状态；仅当用户显式配置自定义错误码时按本地策略处理。 + // 池模式默认不标记本地账号状态；但管理员显式配置的临时不可调度规则优先。 + // 401 保留现有认证错误语义，不在这里改变池模式的认证处理。 if account.IsPoolMode() && !customErrorCodesEnabled { + if statusCode != http.StatusUnauthorized && s.tryTempUnschedulable(ctx, account, statusCode, responseBody) { + return true + } slog.Info("pool_mode_error_skipped", "account_id", account.ID, "status_code", statusCode) return false }