diff --git a/apps/api/internal/httpapi/compat_protocol.go b/apps/api/internal/httpapi/compat_protocol.go index 603afbc..eb82700 100644 --- a/apps/api/internal/httpapi/compat_protocol.go +++ b/apps/api/internal/httpapi/compat_protocol.go @@ -123,6 +123,17 @@ func writeOpenAIError(w http.ResponseWriter, status int, message string, details if strings.TrimSpace(code) != "" { payload["code"] = code } + if len(details) > 0 { + publicDetails := map[string]any{} + for key, value := range details { + if key != "param" { + publicDetails[key] = value + } + } + if len(publicDetails) > 0 { + payload["details"] = publicDetails + } + } writeJSON(w, status, map[string]any{"error": payload}) } diff --git a/apps/api/internal/httpapi/failure_policy_http_acceptance_integration_test.go b/apps/api/internal/httpapi/failure_policy_http_acceptance_integration_test.go new file mode 100644 index 0000000..c6bf2d7 --- /dev/null +++ b/apps/api/internal/httpapi/failure_policy_http_acceptance_integration_test.go @@ -0,0 +1,911 @@ +package httpapi + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/config" + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" + "github.com/jackc/pgx/v5/pgxpool" +) + +type failurePolicyAcceptanceFixture struct { + ctx context.Context + cancel context.CancelFunc + db *store.Store + pool *pgxpool.Pool + server *httptest.Server + adminToken string + apiKey string + suffix string +} + +type controlledProviderStub struct { + server *httptest.Server + calls atomic.Int64 +} + +type acceptancePlatform struct { + ID string `json:"id"` +} + +type acceptancePlatformModel struct { + ID string `json:"id"` +} + +type acceptanceTaskDetail struct { + ID string `json:"id"` + Status string `json:"status"` + Attempts []struct { + AttemptNo int `json:"attemptNo"` + PlatformID string `json:"platformId"` + PlatformName string `json:"platformName"` + PlatformModel string `json:"platformModelId"` + Status string `json:"status"` + ErrorCode string `json:"errorCode"` + DynamicMetrics map[string]any `json:"metrics"` + Trace []map[string]any `json:"trace"` + } `json:"attempts"` +} + +func TestFailurePolicyHTTPAcceptance(t *testing.T) { + fixture := newFailurePolicyAcceptanceFixture(t) + defer fixture.close() + + originalPolicy, err := fixture.db.GetActiveRunnerPolicy(fixture.ctx) + if err != nil { + t.Fatalf("read original runner policy: %v", err) + } + t.Cleanup(func() { + _, _ = fixture.db.UpsertDefaultRunnerPolicy(context.Background(), store.RunnerPolicyInput{ + PolicyKey: originalPolicy.PolicyKey, + Name: originalPolicy.Name, + Description: originalPolicy.Description, + FailoverPolicy: originalPolicy.FailoverPolicy, + HardStopPolicy: originalPolicy.HardStopPolicy, + PriorityDemotePolicy: originalPolicy.PriorityDemotePolicy, + SingleSourcePolicy: originalPolicy.SingleSourcePolicy, + CacheAffinityPolicy: originalPolicy.CacheAffinityPolicy, + Metadata: originalPolicy.Metadata, + Status: originalPolicy.Status, + }) + }) + fixture.setRunnerPolicy(t, true) + + t.Run("Gemini 429 后轮转且只冷却模型一次", fixture.testGemini429ImageFailover) + t.Run("5xx 同平台重试耗尽后降权并轮转", fixture.testServerErrorRetryThenFailover) + t.Run("401 立即禁用平台并轮转", fixture.testAuthFailureDisableThenFailover) + t.Run("400 参数错误硬停止", fixture.testBadRequestStops) + t.Run("旧 degradePolicy 仅转换一次", fixture.testLegacyDegradeDoesNotDoubleApply) + t.Run("并发单源保护至少保留一个来源", fixture.testConcurrentSingleSourceProtection) + t.Run("全部冷却同步返回脱敏 429", fixture.testAllCoolingSynchronousContract) + t.Run("全部冷却异步零 attempt 重排并恢复", fixture.testAllCoolingAsyncRecovery) + t.Run("流式首包边界控制轮转", fixture.testStreamingBoundary) + t.Run("管理接口严格校验且不覆盖原策略", fixture.testRunnerPolicyHTTPValidation) +} + +func newFailurePolicyAcceptanceFixture(t *testing.T) *failurePolicyAcceptanceFixture { + t.Helper() + databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL")) + if databaseURL == "" { + t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run failure-policy HTTP acceptance") + } + ctx, cancel := context.WithCancel(context.Background()) + applyMigration(t, ctx, databaseURL) + db, err := store.Connect(ctx, databaseURL) + if err != nil { + cancel() + t.Fatalf("connect acceptance store: %v", err) + } + pool, err := pgxpool.New(ctx, databaseURL) + if err != nil { + db.Close() + cancel() + t.Fatalf("connect acceptance pool: %v", err) + } + handler := NewServerWithContext(ctx, config.Config{ + AppEnv: "test", + HTTPAddr: ":0", + DatabaseURL: databaseURL, + IdentityMode: "hybrid", + JWTSecret: "failure-policy-http-acceptance-secret", + BillingEngineMode: "observe", + CORSAllowedOrigin: "*", + }, db, slog.New(slog.NewTextHandler(io.Discard, nil))) + server := httptest.NewServer(handler) + fixture := &failurePolicyAcceptanceFixture{ + ctx: ctx, + cancel: cancel, + db: db, + pool: pool, + server: server, + suffix: strconv.FormatInt(time.Now().UnixNano(), 10), + } + + username := "failure_acceptance_" + fixture.suffix + password := "password123" + doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/register", "", map[string]any{ + "username": username, + "email": username + "@example.com", + "password": password, + }, http.StatusCreated, &map[string]any{}) + if _, err := pool.Exec(ctx, `UPDATE gateway_users SET roles = '["admin"]'::jsonb WHERE username = $1`, username); err != nil { + fixture.close() + t.Fatalf("promote acceptance admin: %v", err) + } + var login struct { + AccessToken string `json:"accessToken"` + } + doJSON(t, server.URL, http.MethodPost, "/api/v1/auth/login", "", map[string]any{ + "account": username, "password": password, + }, http.StatusOK, &login) + fixture.adminToken = login.AccessToken + var gatewayUserID string + if err := pool.QueryRow(ctx, `SELECT id::text FROM gateway_users WHERE username=$1`, username).Scan(&gatewayUserID); err != nil { + fixture.close() + t.Fatalf("read acceptance user: %v", err) + } + doJSON(t, server.URL, http.MethodPatch, "/api/admin/users/"+gatewayUserID+"/wallet", fixture.adminToken, map[string]any{ + "currency": "resource", + "balance": 1000000, + "reason": "seed isolated failure-policy acceptance wallet", + }, http.StatusOK, &map[string]any{}) + var key struct { + Secret string `json:"secret"` + } + doJSON(t, server.URL, http.MethodPost, "/api/v1/api-keys", fixture.adminToken, map[string]any{ + "name": "failure policy acceptance", + }, http.StatusCreated, &key) + fixture.apiKey = key.Secret + return fixture +} + +func (f *failurePolicyAcceptanceFixture) close() { + if f.server != nil { + f.server.Close() + } + if f.cancel != nil { + f.cancel() + } + if f.pool != nil { + f.pool.Close() + } + if f.db != nil { + f.db.Close() + } +} + +func (f *failurePolicyAcceptanceFixture) setRunnerPolicy(t *testing.T, singleSource bool) { + t.Helper() + doJSON(t, f.server.URL, http.MethodPatch, "/api/admin/runtime/runner-policy", f.adminToken, standardFailureRunnerPolicy(singleSource), http.StatusOK, &map[string]any{}) +} + +func standardFailureRunnerPolicy(singleSource bool) map[string]any { + return map[string]any{ + "policyKey": "default-runner-v1", + "name": "失败策略 HTTP 验收", + "status": "active", + "description": "真实 provider HTTP client 的失败策略验收", + "failoverPolicy": map[string]any{ + "enabled": true, + "maxPlatforms": 99, + "maxDurationSeconds": 600, + "actionRules": []any{ + map[string]any{"categories": []any{"rate_limit"}, "statusCodes": []any{429}, "action": "cooldown_and_next", "target": "model", "cooldownSeconds": 30}, + map[string]any{"categories": []any{"auth_error"}, "statusCodes": []any{401, 403}, "action": "disable_and_next", "target": "platform"}, + map[string]any{"categories": []any{"timeout", "provider_5xx", "provider_overloaded"}, "statusCodes": []any{408, 500, 502, 503, 504}, "action": "demote_and_next", "target": "platform", "demoteSteps": 1}, + map[string]any{"categories": []any{"network", "stream_error"}, "action": "next"}, + map[string]any{"categories": []any{"request_error", "unsupported_model", "user_permission", "insufficient_balance"}, "statusCodes": []any{400, 404, 409, 422}, "action": "stop"}, + }, + }, + "singleSourcePolicy": map[string]any{"enabled": singleSource}, + "cacheAffinityPolicy": map[string]any{"enabled": false}, + } +} + +func (f *failurePolicyAcceptanceFixture) createBaseModel(t *testing.T, model string, modelType string) string { + t.Helper() + var created struct { + ID string `json:"id"` + } + doJSON(t, f.server.URL, http.MethodPost, "/api/admin/catalog/base-models", f.adminToken, map[string]any{ + "providerKey": "openai", + "canonicalModelKey": "openai:" + model, + "invocationName": model, + "providerModelName": model, + "modelType": []string{modelType}, + "displayName": model, + "status": "active", + "capabilities": map[string]any{}, + "baseBillingConfig": map[string]any{}, + }, http.StatusCreated, &created) + return created.ID +} + +func (f *failurePolicyAcceptanceFixture) createPlatformModel( + t *testing.T, + name string, + provider string, + baseURL string, + priority int, + baseModelID string, + providerModelName string, + modelType string, + retryAttempts int, + runtimePolicySetID string, + runtimeOverride map[string]any, +) (acceptancePlatform, acceptancePlatformModel) { + t.Helper() + var platform acceptancePlatform + doJSON(t, f.server.URL, http.MethodPost, "/api/admin/platforms", f.adminToken, map[string]any{ + "provider": provider, + "platformKey": "acceptance-" + name + "-" + f.suffix, + "name": "acceptance-" + name, + "baseUrl": strings.TrimRight(baseURL, "/"), + "authType": "bearer", + "credentials": map[string]any{"apiKey": "controlled-upstream-test-key"}, + "priority": priority, + "status": "enabled", + }, http.StatusCreated, &platform) + var model acceptancePlatformModel + payload := map[string]any{ + "baseModelId": baseModelID, + "providerModelName": providerModelName, + "modelType": []string{modelType}, + "retryPolicy": map[string]any{"enabled": true, "maxAttempts": retryAttempts}, + } + if runtimePolicySetID != "" { + payload["runtimePolicySetId"] = runtimePolicySetID + } + if len(runtimeOverride) > 0 { + payload["runtimePolicyOverride"] = runtimeOverride + } + doJSON(t, f.server.URL, http.MethodPost, "/api/admin/platforms/"+platform.ID+"/models", f.adminToken, payload, http.StatusCreated, &model) + return platform, model +} + +func newControlledProviderStub(handler func(call int, w http.ResponseWriter, r *http.Request)) *controlledProviderStub { + stub := &controlledProviderStub{} + stub.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + call := int(stub.calls.Add(1)) + handler(call, w, r) + })) + return stub +} + +func (s *controlledProviderStub) close() { + if s != nil && s.server != nil { + s.server.Close() + } +} + +func writeStubJSON(w http.ResponseWriter, status int, payload any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(payload) +} + +func writeOpenAIChatSuccess(w http.ResponseWriter, content string) { + writeStubJSON(w, http.StatusOK, map[string]any{ + "id": "chatcmpl-controlled", "object": "chat.completion", "created": time.Now().Unix(), "model": "controlled-model", + "choices": []any{map[string]any{ + "index": 0, "message": map[string]any{"role": "assistant", "content": content}, "finish_reason": "stop", + }}, + "usage": map[string]any{"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, + }) +} + +func writeOpenAIImageSuccess(w http.ResponseWriter) { + writeStubJSON(w, http.StatusOK, map[string]any{ + "created": time.Now().Unix(), + "data": []any{map[string]any{"url": "https://example.invalid/controlled-test.png"}}, + }) +} + +func (f *failurePolicyAcceptanceFixture) postJSON(t *testing.T, path string, payload any, headers map[string]string) (int, http.Header, []byte) { + t.Helper() + raw, err := json.Marshal(payload) + if err != nil { + t.Fatalf("marshal acceptance payload: %v", err) + } + request, err := http.NewRequest(http.MethodPost, f.server.URL+path, bytes.NewReader(raw)) + if err != nil { + t.Fatalf("build acceptance request: %v", err) + } + request.Header.Set("Authorization", "Bearer "+f.apiKey) + request.Header.Set("Content-Type", "application/json") + for key, value := range headers { + request.Header.Set(key, value) + } + response, err := http.DefaultClient.Do(request) + if err != nil { + t.Fatalf("acceptance request %s: %v", path, err) + } + defer response.Body.Close() + body, _ := io.ReadAll(response.Body) + return response.StatusCode, response.Header.Clone(), body +} + +func (f *failurePolicyAcceptanceFixture) taskIDForModel(t *testing.T, model string, startedAt time.Time) string { + t.Helper() + var taskID string + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + err := f.pool.QueryRow(f.ctx, ` +SELECT id::text +FROM gateway_tasks +WHERE model = $1 + AND created_at >= $2 +ORDER BY created_at DESC +LIMIT 1`, model, startedAt.Add(-time.Second)).Scan(&taskID) + if err == nil && taskID != "" { + return taskID + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("task for model %q was not persisted", model) + return "" +} + +func (f *failurePolicyAcceptanceFixture) loadTask(t *testing.T, taskID string) acceptanceTaskDetail { + t.Helper() + var detail acceptanceTaskDetail + doJSON(t, f.server.URL, http.MethodGet, "/api/v1/tasks/"+taskID, f.apiKey, nil, http.StatusOK, &detail) + return detail +} + +func countFailureEffects(detail acceptanceTaskDetail, effect string) int { + count := 0 + for _, attempt := range detail.Attempts { + trace := attempt.Trace + for _, raw := range anySlice(attempt.DynamicMetrics["trace"]) { + if entry, ok := raw.(map[string]any); ok { + trace = append(trace, entry) + } + } + for _, entry := range trace { + if entry["event"] == "failure_decision" && entry["effect"] == effect { + count++ + } + } + } + return count +} + +func anySlice(value any) []any { + items, _ := value.([]any) + return items +} + +func (f *failurePolicyAcceptanceFixture) testGemini429ImageFailover(t *testing.T) { + model := "acceptance-gemini-image-" + f.suffix + baseModelID := f.createBaseModel(t, model, "image_generate") + gemini := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + writeStubJSON(w, http.StatusTooManyRequests, map[string]any{ + "error": map[string]any{"code": 429, "message": "RESOURCE_EXHAUSTED controlled failure", "status": "RESOURCE_EXHAUSTED"}, + }) + }) + defer gemini.close() + openai := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIImageSuccess(w) }) + defer openai.close() + platformA, modelA := f.createPlatformModel(t, "gemini-image-a", "gemini", gemini.server.URL, 10, baseModelID, "gemini-image-controlled", "image_generate", 3, "", nil) + platformB, _ := f.createPlatformModel(t, "openai-image-b", "openai", openai.server.URL+"/v1", 20, baseModelID, "openai-image-controlled", "image_generate", 1, "", nil) + + startedAt := time.Now() + status, _, body := f.postJSON(t, "/api/v1/images/generations", map[string]any{"model": model, "prompt": "controlled image"}, nil) + if status != http.StatusOK { + t.Fatalf("status=%d body=%s", status, body) + } + if gemini.calls.Load() != 1 || openai.calls.Load() != 1 { + t.Fatalf("upstream calls gemini=%d openai=%d, want 1/1", gemini.calls.Load(), openai.calls.Load()) + } + var modelCooling, platformCooling bool + if err := f.pool.QueryRow(f.ctx, ` +SELECT COALESCE(model.cooldown_until > now(), false), COALESCE(platform.cooldown_until > now(), false) +FROM platform_models model +JOIN integration_platforms platform ON platform.id = model.platform_id +WHERE model.id = $1::uuid`, modelA.ID).Scan(&modelCooling, &platformCooling); err != nil { + t.Fatalf("read Gemini cooldown state: %v", err) + } + if !modelCooling || platformCooling { + t.Fatalf("cooldown state model=%v platform=%v, want true/false for platform %s", modelCooling, platformCooling, platformA.ID) + } + detail := f.loadTask(t, f.taskIDForModel(t, model, startedAt)) + if len(detail.Attempts) != 2 || detail.Attempts[0].PlatformID != platformA.ID || detail.Attempts[1].PlatformID != platformB.ID || countFailureEffects(detail, "cooldown") != 1 { + t.Fatalf("unexpected task attempts/trace: %+v", detail) + } +} + +func (f *failurePolicyAcceptanceFixture) testServerErrorRetryThenFailover(t *testing.T) { + model := "acceptance-5xx-" + f.suffix + baseModelID := f.createBaseModel(t, model, "text_generate") + failed := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + writeStubJSON(w, http.StatusInternalServerError, map[string]any{"error": map[string]any{"message": "controlled 500"}}) + }) + defer failed.close() + success := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIChatSuccess(w, "fallback ok") }) + defer success.close() + platformA, _ := f.createPlatformModel(t, "5xx-a", "openai", failed.server.URL+"/v1", 10, baseModelID, "controlled-5xx", "text_generate", 2, "", nil) + platformB, _ := f.createPlatformModel(t, "5xx-b", "openai", success.server.URL+"/v1", 20, baseModelID, "controlled-success", "text_generate", 1, "", nil) + + startedAt := time.Now() + status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{ + "model": model, "messages": []any{map[string]any{"role": "user", "content": "retry"}}, + }, nil) + if status != http.StatusOK { + t.Fatalf("status=%d body=%s", status, body) + } + if failed.calls.Load() != 2 || success.calls.Load() != 1 { + t.Fatalf("upstream calls failed=%d success=%d, want 2/1", failed.calls.Load(), success.calls.Load()) + } + detail := f.loadTask(t, f.taskIDForModel(t, model, startedAt)) + if len(detail.Attempts) != 3 || + detail.Attempts[0].PlatformID != platformA.ID || + detail.Attempts[1].PlatformID != platformA.ID || + detail.Attempts[2].PlatformID != platformB.ID || + countFailureEffects(detail, "demote") != 1 { + t.Fatalf("unexpected retry/failover chain: %+v", detail) + } + var dynamicPriority *int + if err := f.pool.QueryRow(f.ctx, `SELECT dynamic_priority FROM integration_platforms WHERE id=$1::uuid`, platformA.ID).Scan(&dynamicPriority); err != nil { + t.Fatalf("read demoted priority: %v", err) + } + if dynamicPriority == nil { + t.Fatal("failed platform should be demoted exactly once after retry exhaustion") + } +} + +func (f *failurePolicyAcceptanceFixture) testAuthFailureDisableThenFailover(t *testing.T) { + model := "acceptance-auth-" + f.suffix + baseModelID := f.createBaseModel(t, model, "text_generate") + failed := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + writeStubJSON(w, http.StatusUnauthorized, map[string]any{"error": map[string]any{"message": "invalid api key"}}) + }) + defer failed.close() + success := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIChatSuccess(w, "auth fallback") }) + defer success.close() + platformA, _ := f.createPlatformModel(t, "auth-a", "openai", failed.server.URL+"/v1", 10, baseModelID, "controlled-auth", "text_generate", 3, "", nil) + _, _ = f.createPlatformModel(t, "auth-b", "openai", success.server.URL+"/v1", 20, baseModelID, "controlled-success", "text_generate", 1, "", nil) + + startedAt := time.Now() + status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{ + "model": model, "messages": []any{map[string]any{"role": "user", "content": "auth"}}, + }, nil) + if status != http.StatusOK { + t.Fatalf("status=%d body=%s", status, body) + } + if failed.calls.Load() != 1 || success.calls.Load() != 1 { + t.Fatalf("upstream calls failed=%d success=%d, want 1/1", failed.calls.Load(), success.calls.Load()) + } + var statusValue string + if err := f.pool.QueryRow(f.ctx, `SELECT status FROM integration_platforms WHERE id=$1::uuid`, platformA.ID).Scan(&statusValue); err != nil { + t.Fatalf("read disabled platform: %v", err) + } + detail := f.loadTask(t, f.taskIDForModel(t, model, startedAt)) + if statusValue != "disabled" || countFailureEffects(detail, "disable") != 1 || len(detail.Attempts) != 2 { + t.Fatalf("unexpected auth disable result status=%s detail=%+v", statusValue, detail) + } +} + +func (f *failurePolicyAcceptanceFixture) testBadRequestStops(t *testing.T) { + model := "acceptance-400-" + f.suffix + baseModelID := f.createBaseModel(t, model, "text_generate") + failed := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + writeStubJSON(w, http.StatusBadRequest, map[string]any{"error": map[string]any{"message": "controlled invalid parameter", "code": "invalid_parameter"}}) + }) + defer failed.close() + success := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIChatSuccess(w, "must not run") }) + defer success.close() + platformA, modelA := f.createPlatformModel(t, "400-a", "openai", failed.server.URL+"/v1", 10, baseModelID, "controlled-bad", "text_generate", 3, "", nil) + _, _ = f.createPlatformModel(t, "400-b", "openai", success.server.URL+"/v1", 20, baseModelID, "controlled-success", "text_generate", 1, "", nil) + + status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{ + "model": model, "messages": []any{map[string]any{"role": "user", "content": "bad"}}, + }, nil) + if status != http.StatusBadRequest || success.calls.Load() != 0 || failed.calls.Load() != 1 { + t.Fatalf("status=%d calls=%d/%d body=%s", status, failed.calls.Load(), success.calls.Load(), body) + } + var platformStatus string + var modelEnabled bool + var platformCooldown, modelCooldown *time.Time + if err := f.pool.QueryRow(f.ctx, ` +SELECT platform.status, model.enabled, platform.cooldown_until, model.cooldown_until +FROM integration_platforms platform +JOIN platform_models model ON model.platform_id=platform.id +WHERE platform.id=$1::uuid AND model.id=$2::uuid`, platformA.ID, modelA.ID).Scan(&platformStatus, &modelEnabled, &platformCooldown, &modelCooldown); err != nil { + t.Fatalf("read hard-stop state: %v", err) + } + if platformStatus != "enabled" || !modelEnabled || platformCooldown != nil || modelCooldown != nil { + t.Fatalf("hard stop mutated runtime state: status=%s enabled=%v platformCooldown=%v modelCooldown=%v", platformStatus, modelEnabled, platformCooldown, modelCooldown) + } +} + +func (f *failurePolicyAcceptanceFixture) testLegacyDegradeDoesNotDoubleApply(t *testing.T) { + model := "acceptance-legacy-" + f.suffix + baseModelID := f.createBaseModel(t, model, "text_generate") + var policySet struct { + ID string `json:"id"` + } + doJSON(t, f.server.URL, http.MethodPost, "/api/admin/runtime/policy-sets", f.adminToken, map[string]any{ + "policyKey": "acceptance-legacy-" + f.suffix, + "name": "acceptance legacy degrade", + "degradePolicy": map[string]any{ + "enabled": true, "keywords": []any{"rate_limit"}, "cooldownSeconds": 45, + }, + }, http.StatusCreated, &policySet) + failed := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "rate_limit controlled"}}) + }) + defer failed.close() + success := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIChatSuccess(w, "legacy fallback") }) + defer success.close() + _, modelA := f.createPlatformModel(t, "legacy-a", "openai", failed.server.URL+"/v1", 10, baseModelID, "legacy-failed", "text_generate", 3, policySet.ID, nil) + _, _ = f.createPlatformModel(t, "legacy-b", "openai", success.server.URL+"/v1", 20, baseModelID, "legacy-success", "text_generate", 1, "", nil) + + startedAt := time.Now() + status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{ + "model": model, "messages": []any{map[string]any{"role": "user", "content": "legacy"}}, + }, nil) + if status != http.StatusOK || failed.calls.Load() != 1 || success.calls.Load() != 1 { + t.Fatalf("status=%d calls=%d/%d body=%s", status, failed.calls.Load(), success.calls.Load(), body) + } + detail := f.loadTask(t, f.taskIDForModel(t, model, startedAt)) + if countFailureEffects(detail, "cooldown") != 1 { + t.Fatalf("legacy and global policy must not both execute: %+v", detail) + } + var remaining float64 + if err := f.pool.QueryRow(f.ctx, `SELECT EXTRACT(EPOCH FROM cooldown_until-now())::float8 FROM platform_models WHERE id=$1::uuid`, modelA.ID).Scan(&remaining); err != nil { + t.Fatalf("read legacy cooldown: %v", err) + } + if remaining < 35 || remaining > 50 { + t.Fatalf("legacy cooldown should be applied once with 45 seconds, remaining=%f", remaining) + } +} + +func (f *failurePolicyAcceptanceFixture) testConcurrentSingleSourceProtection(t *testing.T) { + f.setRunnerPolicy(t, true) + model := "acceptance-concurrent-" + f.suffix + baseModelID := f.createBaseModel(t, model, "text_generate") + stubs := make([]*controlledProviderStub, 0, 3) + modelIDs := make([]string, 0, 3) + for index := 0; index < 3; index++ { + stub := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "controlled 429"}}) + }) + stubs = append(stubs, stub) + _, platformModel := f.createPlatformModel(t, fmt.Sprintf("concurrent-%d", index), "openai", stub.server.URL+"/v1", 10+index*10, baseModelID, fmt.Sprintf("concurrent-%d", index), "text_generate", 1, "", nil) + modelIDs = append(modelIDs, platformModel.ID) + } + defer func() { + for _, stub := range stubs { + stub.close() + } + }() + + var wait sync.WaitGroup + errorsFound := make(chan error, 12) + for index := 0; index < 12; index++ { + wait.Add(1) + go func() { + defer wait.Done() + status, _, body, err := f.postJSONNoFail("/api/v1/chat/completions", map[string]any{ + "model": model, "messages": []any{map[string]any{"role": "user", "content": "concurrent"}}, + }) + if err != nil { + errorsFound <- err + return + } + if status != http.StatusTooManyRequests { + errorsFound <- fmt.Errorf("status=%d body=%s", status, body) + } + }() + } + wait.Wait() + close(errorsFound) + for err := range errorsFound { + t.Error(err) + } + + var available int + if err := f.pool.QueryRow(f.ctx, ` +SELECT count(*) +FROM platform_models model +JOIN integration_platforms platform ON platform.id=model.platform_id +WHERE model.id = ANY($1::uuid[]) + AND model.enabled=true + AND platform.status='enabled' + AND (model.cooldown_until IS NULL OR model.cooldown_until <= now()) + AND (platform.cooldown_until IS NULL OR platform.cooldown_until <= now())`, modelIDs).Scan(&available); err != nil { + t.Fatalf("count available sources: %v", err) + } + if available < 1 { + t.Fatal("concurrent cooldown decisions disabled or cooled every source") + } + before := make([]int64, len(stubs)) + for index, stub := range stubs { + before[index] = stub.calls.Load() + } + status, _, _, err := f.postJSONNoFail("/api/v1/chat/completions", map[string]any{ + "model": model, "messages": []any{map[string]any{"role": "user", "content": "post-cooldown"}}, + }) + if err != nil || status != http.StatusTooManyRequests { + t.Fatalf("post-cooldown request status=%d err=%v", status, err) + } + for index, modelID := range modelIDs { + var cooled bool + if err := f.pool.QueryRow(f.ctx, `SELECT COALESCE(cooldown_until > now(), false) FROM platform_models WHERE id=$1::uuid`, modelID).Scan(&cooled); err != nil { + t.Fatalf("read source cooldown: %v", err) + } + if cooled && stubs[index].calls.Load() != before[index] { + t.Fatalf("cooled source %d was called again: before=%d after=%d", index, before[index], stubs[index].calls.Load()) + } + } +} + +func (f *failurePolicyAcceptanceFixture) postJSONNoFail(path string, payload any) (int, http.Header, []byte, error) { + raw, err := json.Marshal(payload) + if err != nil { + return 0, nil, nil, err + } + request, err := http.NewRequest(http.MethodPost, f.server.URL+path, bytes.NewReader(raw)) + if err != nil { + return 0, nil, nil, err + } + request.Header.Set("Authorization", "Bearer "+f.apiKey) + request.Header.Set("Content-Type", "application/json") + response, err := http.DefaultClient.Do(request) + if err != nil { + return 0, nil, nil, err + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + return response.StatusCode, response.Header.Clone(), body, err +} + +func (f *failurePolicyAcceptanceFixture) testAllCoolingSynchronousContract(t *testing.T) { + f.setRunnerPolicy(t, false) + defer f.setRunnerPolicy(t, true) + model := "acceptance-all-cooling-" + f.suffix + baseModelID := f.createBaseModel(t, model, "image_generate") + stubA := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "private-a-raw-error"}}) + }) + defer stubA.close() + stubB := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "private-b-raw-error"}}) + }) + defer stubB.close() + override := func(seconds int) map[string]any { + return map[string]any{"failoverPolicy": map[string]any{"actionRules": []any{ + map[string]any{"statusCodes": []any{429}, "action": "cooldown_and_next", "target": "model", "cooldownSeconds": seconds}, + }}} + } + _, modelA := f.createPlatformModel(t, "cooling-a", "openai", stubA.server.URL+"/v1", 10, baseModelID, "cooling-a", "image_generate", 3, "", override(30)) + _, modelB := f.createPlatformModel(t, "cooling-b", "openai", stubB.server.URL+"/v1", 20, baseModelID, "cooling-b", "image_generate", 3, "", override(60)) + defer func() { + _, _ = f.pool.Exec(context.Background(), `UPDATE platform_models SET cooldown_until=NULL WHERE id IN ($1::uuid,$2::uuid)`, modelA.ID, modelB.ID) + }() + + status, _, _ := f.postJSON(t, "/api/v1/images/generations", map[string]any{"model": model, "prompt": "cool both"}, nil) + if status != http.StatusTooManyRequests || stubA.calls.Load() != 1 || stubB.calls.Load() != 1 { + t.Fatalf("initial cooling request status=%d calls=%d/%d", status, stubA.calls.Load(), stubB.calls.Load()) + } + startedAt := time.Now() + status, headers, body := f.postJSON(t, "/api/v1/images/generations", map[string]any{"model": model, "prompt": "all cooling"}, nil) + if status != http.StatusTooManyRequests { + t.Fatalf("cooldown contract status=%d body=%s", status, body) + } + retryAfter, err := strconv.Atoi(headers.Get("Retry-After")) + if err != nil || retryAfter < 20 || retryAfter > 31 { + t.Fatalf("Retry-After=%q, want earliest recovery near 30 seconds", headers.Get("Retry-After")) + } + bodyText := string(body) + for _, forbidden := range []string{"acceptance-cooling-a", "acceptance-cooling-b", stubA.server.URL, stubB.server.URL, "private-a-raw-error", "private-b-raw-error"} { + if strings.Contains(bodyText, forbidden) { + t.Fatalf("cooldown response leaked %q: %s", forbidden, bodyText) + } + } + var envelope struct { + Error struct { + Code string `json:"code"` + Details map[string]any `json:"details"` + } `json:"error"` + } + if err := json.Unmarshal(body, &envelope); err != nil { + t.Fatalf("decode cooldown response: %v body=%s", err, body) + } + if envelope.Error.Code != "model_cooling_down" || len(envelope.Error.Details) != 2 { + t.Fatalf("unexpected cooldown envelope: %+v", envelope) + } + detail := f.loadTask(t, f.taskIDForModel(t, model, startedAt)) + if len(detail.Attempts) != 0 { + t.Fatalf("candidate selection cooldown must not create provider attempts: %+v", detail.Attempts) + } +} + +func (f *failurePolicyAcceptanceFixture) testAllCoolingAsyncRecovery(t *testing.T) { + f.setRunnerPolicy(t, false) + defer f.setRunnerPolicy(t, true) + model := "acceptance-async-cooling-" + f.suffix + baseModelID := f.createBaseModel(t, model, "image_generate") + stubA := newControlledProviderStub(func(call int, w http.ResponseWriter, _ *http.Request) { + if call == 1 { + writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "controlled first 429"}}) + return + } + writeOpenAIImageSuccess(w) + }) + defer stubA.close() + stubB := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + writeStubJSON(w, http.StatusTooManyRequests, map[string]any{"error": map[string]any{"message": "controlled second 429"}}) + }) + defer stubB.close() + override := func(seconds int) map[string]any { + return map[string]any{"failoverPolicy": map[string]any{"actionRules": []any{ + map[string]any{"statusCodes": []any{429}, "action": "cooldown_and_next", "target": "model", "cooldownSeconds": seconds}, + }}} + } + _, modelA := f.createPlatformModel(t, "async-cooling-a", "openai", stubA.server.URL+"/v1", 10, baseModelID, "async-a", "image_generate", 1, "", override(4)) + _, modelB := f.createPlatformModel(t, "async-cooling-b", "openai", stubB.server.URL+"/v1", 20, baseModelID, "async-b", "image_generate", 1, "", override(8)) + defer func() { + _, _ = f.pool.Exec(context.Background(), `UPDATE platform_models SET cooldown_until=NULL WHERE id IN ($1::uuid,$2::uuid)`, modelA.ID, modelB.ID) + }() + status, _, _ := f.postJSON(t, "/api/v1/images/generations", map[string]any{"model": model, "prompt": "prime cooldown"}, nil) + if status != http.StatusTooManyRequests { + t.Fatalf("prime cooldown status=%d", status) + } + + startedAt := time.Now() + status, _, body := f.postJSON(t, "/api/v1/images/generations", map[string]any{"model": model, "prompt": "async recovery"}, map[string]string{"X-Async": "true"}) + if status != http.StatusAccepted { + t.Fatalf("async submit status=%d body=%s", status, body) + } + taskID := f.taskIDForModel(t, model, startedAt) + deadline := time.Now().Add(2 * time.Second) + queuedObserved := false + for time.Now().Before(deadline) { + detail := f.loadTask(t, taskID) + if detail.Status == "queued" && len(detail.Attempts) == 0 { + queuedObserved = true + break + } + time.Sleep(25 * time.Millisecond) + } + if !queuedObserved { + t.Fatal("async cooling task was not observed queued with zero provider attempts") + } + deadline = time.Now().Add(12 * time.Second) + var completed acceptanceTaskDetail + for time.Now().Before(deadline) { + completed = f.loadTask(t, taskID) + if completed.Status == "succeeded" { + break + } + time.Sleep(100 * time.Millisecond) + } + if completed.Status != "succeeded" || len(completed.Attempts) != 1 || stubA.calls.Load() != 2 { + t.Fatalf("async task did not recover on first restored candidate: task=%+v callsA=%d callsB=%d", completed, stubA.calls.Load(), stubB.calls.Load()) + } +} + +func (f *failurePolicyAcceptanceFixture) testStreamingBoundary(t *testing.T) { + t.Run("首个 delta 前允许轮转", func(t *testing.T) { + model := "acceptance-stream-before-" + f.suffix + baseModelID := f.createBaseModel(t, model, "text_generate") + failed := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + writeStubJSON(w, http.StatusInternalServerError, map[string]any{"error": map[string]any{"message": "before delta"}}) + }) + defer failed.close() + success := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, "data: {\"id\":\"chunk-ok\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"stream fallback\"},\"finish_reason\":null}]}\n\n") + _, _ = io.WriteString(w, "data: {\"id\":\"chunk-ok\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n") + _, _ = io.WriteString(w, "data: [DONE]\n\n") + }) + defer success.close() + _, _ = f.createPlatformModel(t, "stream-before-a", "openai", failed.server.URL+"/v1", 10, baseModelID, "stream-failed", "text_generate", 1, "", nil) + _, _ = f.createPlatformModel(t, "stream-before-b", "openai", success.server.URL+"/v1", 20, baseModelID, "stream-success", "text_generate", 1, "", nil) + + status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{ + "model": model, "stream": true, "messages": []any{map[string]any{"role": "user", "content": "stream"}}, + }, nil) + if status != http.StatusOK || failed.calls.Load() != 1 || success.calls.Load() != 1 || strings.Count(string(body), "stream fallback") != 1 || !strings.Contains(string(body), "[DONE]") { + t.Fatalf("unexpected pre-delta failover status=%d calls=%d/%d body=%s", status, failed.calls.Load(), success.calls.Load(), body) + } + }) + + t.Run("已输出 delta 后禁止轮转", func(t *testing.T) { + model := "acceptance-stream-after-" + f.suffix + baseModelID := f.createBaseModel(t, model, "text_generate") + broken := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { + hijacker, ok := w.(http.Hijacker) + if !ok { + http.Error(w, "hijacking unavailable", http.StatusInternalServerError) + return + } + connection, buffer, err := hijacker.Hijack() + if err != nil { + return + } + defer connection.Close() + _, _ = buffer.WriteString("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: 4096\r\n\r\n") + _, _ = buffer.WriteString("data: {\"id\":\"chunk-broken\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"first platform only\"},\"finish_reason\":null}]}\n\n") + _ = buffer.Flush() + }) + defer broken.close() + fallback := newControlledProviderStub(func(_ int, w http.ResponseWriter, _ *http.Request) { writeOpenAIChatSuccess(w, "must not mix") }) + defer fallback.close() + _, _ = f.createPlatformModel(t, "stream-after-a", "openai", broken.server.URL+"/v1", 10, baseModelID, "stream-broken", "text_generate", 2, "", nil) + _, _ = f.createPlatformModel(t, "stream-after-b", "openai", fallback.server.URL+"/v1", 20, baseModelID, "stream-fallback", "text_generate", 1, "", nil) + + status, _, body := f.postJSON(t, "/api/v1/chat/completions", map[string]any{ + "model": model, "stream": true, "messages": []any{map[string]any{"role": "user", "content": "stream"}}, + }, nil) + if status != http.StatusOK || broken.calls.Load() != 1 || fallback.calls.Load() != 0 { + t.Fatalf("post-delta failure rotated unexpectedly status=%d calls=%d/%d body=%s", status, broken.calls.Load(), fallback.calls.Load(), body) + } + bodyText := string(body) + if !strings.Contains(bodyText, "first platform only") || strings.Contains(bodyText, "must not mix") || strings.Count(bodyText, "event: error") != 1 { + t.Fatalf("stream response mixed providers or missed terminal error: %s", bodyText) + } + }) +} + +func (f *failurePolicyAcceptanceFixture) testRunnerPolicyHTTPValidation(t *testing.T) { + valid := standardFailureRunnerPolicy(true) + var saved map[string]any + doJSON(t, f.server.URL, http.MethodPatch, "/api/admin/runtime/runner-policy", f.adminToken, valid, http.StatusOK, &saved) + var before map[string]any + doJSON(t, f.server.URL, http.MethodGet, "/api/admin/runtime/runner-policy", f.adminToken, nil, http.StatusOK, &before) + + invalidRules := []map[string]any{ + {"categories": []any{"rate_limit"}, "action": "rotate"}, + {"categories": []any{"rate_limit"}, "action": "next", "target": "client"}, + {"action": "next"}, + {"statusCodes": []any{429}, "action": "cooldown_and_next", "target": "model", "cooldownSeconds": 0}, + {"statusCodes": []any{500}, "action": "demote_and_next", "target": "platform", "demoteSteps": -1}, + } + for index, rule := range invalidRules { + payload := standardFailureRunnerPolicy(true) + payload["failoverPolicy"].(map[string]any)["actionRules"] = []any{rule} + raw, _ := json.Marshal(payload) + request, _ := http.NewRequest(http.MethodPatch, f.server.URL+"/api/admin/runtime/runner-policy", bytes.NewReader(raw)) + request.Header.Set("Authorization", "Bearer "+f.adminToken) + request.Header.Set("Content-Type", "application/json") + response, err := http.DefaultClient.Do(request) + if err != nil { + t.Fatalf("invalid policy request %d: %v", index, err) + } + body, _ := io.ReadAll(response.Body) + response.Body.Close() + if response.StatusCode != http.StatusBadRequest { + t.Fatalf("invalid policy %d status=%d body=%s", index, response.StatusCode, body) + } + } + var after map[string]any + doJSON(t, f.server.URL, http.MethodGet, "/api/admin/runtime/runner-policy", f.adminToken, nil, http.StatusOK, &after) + beforeFailover, _ := before["failoverPolicy"].(map[string]any) + afterFailover, _ := after["failoverPolicy"].(map[string]any) + if !jsonObjectsEqual(beforeFailover["actionRules"], afterFailover["actionRules"]) { + t.Fatalf("invalid PATCH overwrote runner action rules: before=%+v after=%+v", beforeFailover["actionRules"], afterFailover["actionRules"]) + } +} + +func jsonObjectsEqual(left any, right any) bool { + leftRaw, _ := json.Marshal(left) + rightRaw, _ := json.Marshal(right) + return bytes.Equal(leftRaw, rightRaw) +} diff --git a/apps/api/internal/httpapi/gemini_compat.go b/apps/api/internal/httpapi/gemini_compat.go index 2fb5238..98ce2b0 100644 --- a/apps/api/internal/httpapi/gemini_compat.go +++ b/apps/api/internal/httpapi/gemini_compat.go @@ -218,6 +218,7 @@ func (s *Server) geminiGenerateContent(w http.ResponseWriter, r *http.Request) { if !requestStillConnected(r) { return } + applyRunErrorHeaders(w, runErr) if wire := clients.ErrorWireResponse(runErr); wireResponseMatches(wire, clients.ProtocolGeminiGenerateContent) { writeWireResponse(w, wire) return @@ -270,6 +271,7 @@ func (s *Server) writeGeminiGenerateContentStream(runCtx context.Context, w http if !requestStillConnected(r) { return } + applyRunErrorHeaders(w, runErr) if !nativePassthrough && !convertedFrames { if wire := clients.ErrorWireResponse(runErr); wireResponseMatches(wire, clients.ProtocolGeminiGenerateContent) { writeWireResponse(w, wire) diff --git a/apps/api/internal/httpapi/handlers.go b/apps/api/internal/httpapi/handlers.go index 2c70dd1..14fd893 100644 --- a/apps/api/internal/httpapi/handlers.go +++ b/apps/api/internal/httpapi/handlers.go @@ -980,6 +980,7 @@ func (s *Server) estimatePricing(w http.ResponseWriter, r *http.Request) { return } if errors.Is(err, store.ErrNoModelCandidate) { + applyRunErrorHeaders(w, err) writeErrorWithDetails(w, statusFromRunError(err), runErrorMessage(err), runErrorDetails(err), store.ModelCandidateErrorCode(err)) return } @@ -1421,6 +1422,7 @@ func writeProtocolCompatibleTaskResponse(runCtx context.Context, w http.Response if !requestStillConnected(r) { return } + applyRunErrorHeaders(w, runErr) errorPayload := map[string]any{"message": runErrorMessage(runErr), "type": "server_error", "param": nil, "code": runErrorCode(runErr)} if status := statusFromRunError(runErr); status >= 400 && status < 500 { errorPayload["type"] = "invalid_request_error" @@ -1458,6 +1460,7 @@ func writeProtocolCompatibleTaskResponse(runCtx context.Context, w http.Response if !requestStillConnected(r) { return } + applyRunErrorHeaders(w, runErr) if wire := clients.ErrorWireResponse(runErr); wireResponseMatches(wire, targetProtocol) { writeWireResponse(w, wire) return @@ -1694,11 +1697,21 @@ func runErrorDetails(err error) map[string]any { return map[string]any{"rateLimit": detail} } if detail := store.ModelCandidateErrorDetails(err); len(detail) > 0 { - return map[string]any{"modelCandidate": detail} + return detail } return nil } +func applyRunErrorHeaders(w http.ResponseWriter, err error) { + if retryAfter := store.ModelCandidateRetryAfter(err); retryAfter > 0 { + seconds := int((retryAfter + time.Second - 1) / time.Second) + if seconds < 1 { + seconds = 1 + } + w.Header().Set("Retry-After", strconv.Itoa(seconds)) + } +} + func rateLimitErrorSummary(err error) string { var limitErr *store.RateLimitExceededError if !errors.As(err, &limitErr) { diff --git a/apps/api/internal/httpapi/model_cooldown_error_test.go b/apps/api/internal/httpapi/model_cooldown_error_test.go new file mode 100644 index 0000000..e0d72bd --- /dev/null +++ b/apps/api/internal/httpapi/model_cooldown_error_test.go @@ -0,0 +1,39 @@ +package httpapi + +import ( + "net/http/httptest" + "testing" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +func TestModelCooldownErrorPublicContract(t *testing.T) { + recoveryAt := time.Now().UTC().Add(17 * time.Second).Truncate(time.Second) + err := &store.ModelCandidateUnavailableError{ + Code: "model_cooling_down", + Message: "请求的模型暂时不可用，请稍后重试", + RetryAfter: 17 * time.Second, + RecoveryAt: recoveryAt, + Details: map[string]any{ + "retryAfterSeconds": 17, + "recoveryAt": recoveryAt.Format(time.RFC3339), + }, + } + recorder := httptest.NewRecorder() + applyRunErrorHeaders(recorder, err) + + if got := statusFromRunError(err); got != 429 { + t.Fatalf("status = %d, want 429", got) + } + if got := runErrorCode(err); got != "model_cooling_down" { + t.Fatalf("code = %q, want model_cooling_down", got) + } + if got := recorder.Header().Get("Retry-After"); got != "17" { + t.Fatalf("Retry-After = %q, want 17", got) + } + details := runErrorDetails(err) + if len(details) != 2 || details["retryAfterSeconds"] != 17 || details["recoveryAt"] != recoveryAt.Format(time.RFC3339) { + t.Fatalf("unexpected public details: %#v", details) + } +} diff --git a/apps/api/internal/httpapi/runtime_policy_handlers.go b/apps/api/internal/httpapi/runtime_policy_handlers.go index 50f37bb..6f0c2e4 100644 --- a/apps/api/internal/httpapi/runtime_policy_handlers.go +++ b/apps/api/internal/httpapi/runtime_policy_handlers.go @@ -3,6 +3,7 @@ package httpapi import ( "encoding/json" "errors" + "fmt" "net/http" "strings" @@ -71,6 +72,10 @@ func (s *Server) updateRunnerPolicy(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "invalid json body") return } + if err := validateRunnerPolicyInput(input); err != nil { + writeError(w, http.StatusBadRequest, err.Error(), "invalid_parameter") + return + } item, err := s.store.UpsertDefaultRunnerPolicy(r.Context(), input) if err != nil { s.logger.Error("update runner policy failed", "error", err) @@ -80,6 +85,89 @@ func (s *Server) updateRunnerPolicy(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, item) } +func validateRunnerPolicyInput(input store.RunnerPolicyInput) error { + rawRules, exists := input.FailoverPolicy["actionRules"] + if !exists { + return nil + } + rules, ok := rawRules.([]any) + if !ok { + return errors.New("failoverPolicy.actionRules must be an array") + } + validActions := map[string]bool{ + "stop": true, "next": true, "cooldown_and_next": true, "demote_and_next": true, "disable_and_next": true, + } + for index, rawRule := range rules { + rule, ok := rawRule.(map[string]any) + if !ok { + return fmt.Errorf("failoverPolicy.actionRules[%d] must be an object", index) + } + action := strings.TrimSpace(stringValue(rule["action"])) + if !validActions[action] { + return fmt.Errorf("failoverPolicy.actionRules[%d].action is invalid", index) + } + if target, present := rule["target"]; present { + targetValue := strings.TrimSpace(stringValue(target)) + if targetValue != "model" && targetValue != "platform" { + return fmt.Errorf("failoverPolicy.actionRules[%d].target is invalid", index) + } + } + if !runnerActionRuleHasMatch(rule) { + return fmt.Errorf("failoverPolicy.actionRules[%d] must include at least one match condition", index) + } + if action == "cooldown_and_next" && positiveJSONInteger(rule["cooldownSeconds"]) <= 0 { + return fmt.Errorf("failoverPolicy.actionRules[%d].cooldownSeconds must be a positive integer", index) + } + if action == "demote_and_next" && positiveJSONInteger(rule["demoteSteps"]) <= 0 { + return fmt.Errorf("failoverPolicy.actionRules[%d].demoteSteps must be a positive integer", index) + } + } + return nil +} + +func runnerActionRuleHasMatch(rule map[string]any) bool { + for _, key := range []string{"categories", "codes", "errorCodes", "statusCodes", "keywords"} { + switch values := rule[key].(type) { + case []any: + for _, value := range values { + if strings.TrimSpace(fmt.Sprint(value)) != "" { + return true + } + } + case []string: + for _, value := range values { + if strings.TrimSpace(value) != "" { + return true + } + } + } + } + return false +} + +func positiveJSONInteger(value any) int { + switch typed := value.(type) { + case int: + if typed > 0 { + return typed + } + case int64: + if typed > 0 { + return int(typed) + } + case float64: + if typed > 0 && typed == float64(int(typed)) { + return int(typed) + } + case json.Number: + parsed, err := typed.Int64() + if err == nil && parsed > 0 { + return int(parsed) + } + } + return 0 +} + type updatePlatformDynamicPriorityRequest struct { DynamicPriority *int `json:"dynamicPriority" example:"10"` Reset bool `json:"reset" example:"false"` diff --git a/apps/api/internal/httpapi/runtime_policy_validation_test.go b/apps/api/internal/httpapi/runtime_policy_validation_test.go new file mode 100644 index 0000000..3fd4a9f --- /dev/null +++ b/apps/api/internal/httpapi/runtime_policy_validation_test.go @@ -0,0 +1,57 @@ +package httpapi + +import ( + "strings" + "testing" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +func TestValidateRunnerPolicyInputActionRules(t *testing.T) { + validRule := func() map[string]any { + return map[string]any{ + "categories": []any{"rate_limit"}, + "action": "cooldown_and_next", + "target": "model", + "cooldownSeconds": float64(60), + } + } + tests := []struct { + name string + mutate func(map[string]any) + wantErr string + }{ + {name: "valid"}, + {name: "invalid action", mutate: func(rule map[string]any) { rule["action"] = "rotate" }, wantErr: ".action is invalid"}, + {name: "invalid target", mutate: func(rule map[string]any) { rule["target"] = "client" }, wantErr: ".target is invalid"}, + {name: "empty match", mutate: func(rule map[string]any) { rule["categories"] = []any{} }, wantErr: "at least one match condition"}, + {name: "missing cooldown", mutate: func(rule map[string]any) { delete(rule, "cooldownSeconds") }, wantErr: ".cooldownSeconds must be a positive integer"}, + {name: "zero cooldown", mutate: func(rule map[string]any) { rule["cooldownSeconds"] = float64(0) }, wantErr: ".cooldownSeconds must be a positive integer"}, + {name: "fractional cooldown", mutate: func(rule map[string]any) { rule["cooldownSeconds"] = 1.5 }, wantErr: ".cooldownSeconds must be a positive integer"}, + {name: "missing demote steps", mutate: func(rule map[string]any) { + rule["action"] = "demote_and_next" + delete(rule, "cooldownSeconds") + }, wantErr: ".demoteSteps must be a positive integer"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + rule := validRule() + if test.mutate != nil { + test.mutate(rule) + } + err := validateRunnerPolicyInput(store.RunnerPolicyInput{ + FailoverPolicy: map[string]any{"actionRules": []any{rule}}, + }) + if test.wantErr == "" { + if err != nil { + t.Fatalf("unexpected validation error: %v", err) + } + return + } + if err == nil || !strings.Contains(err.Error(), test.wantErr) { + t.Fatalf("validation error = %v, want substring %q", err, test.wantErr) + } + }) + } +} diff --git a/apps/api/internal/httpapi/voice_clone_handlers.go b/apps/api/internal/httpapi/voice_clone_handlers.go index 33a6ca1..7f570df 100644 --- a/apps/api/internal/httpapi/voice_clone_handlers.go +++ b/apps/api/internal/httpapi/voice_clone_handlers.go @@ -61,6 +61,7 @@ func (s *Server) deleteClonedVoice(w http.ResponseWriter, r *http.Request) { } result, err := s.runner.DeleteClonedVoice(r.Context(), user, r.PathValue("voiceID")) if err != nil { + applyRunErrorHeaders(w, err) writeErrorWithDetails(w, statusFromRunError(err), runErrorMessage(err), runErrorDetails(err), runErrorCode(err)) return } diff --git a/apps/api/internal/runner/retry_decision.go b/apps/api/internal/runner/retry_decision.go index f312d81..c854dd7 100644 --- a/apps/api/internal/runner/retry_decision.go +++ b/apps/api/internal/runner/retry_decision.go @@ -42,6 +42,29 @@ type failoverDecision struct { Info failureInfo } +type failureDecision struct { + Route string + Effect string + Target string + CooldownSeconds int + DemoteSteps int + Reason string + Match policyRuleMatch + Info failureInfo +} + +type resolveCandidateFailureInput struct { + RunnerPolicy store.RunnerPolicy + Candidate store.RuntimeModelCandidate + Err error + ClientAttempt int + MaxClientAttempts int + HasNextCandidate bool + FailoverExpired bool + Async bool + DownstreamStarted bool +} + type priorityDemoteDecision struct { Demote bool Reason string @@ -104,7 +127,7 @@ func failoverDecisionForCandidate(runnerPolicy store.RunnerPolicy, candidate sto if errors.Is(err, store.ErrRateLimited) { return failoverDecision{Retry: false, Action: "queue", Reason: "local_rate_limit_wait_queue", Match: policyRuleMatch{Source: "gateway_rate_limits", Policy: "rateLimitPolicy", Rule: "localCapacity", Value: "exceeded"}, Info: info} } - if ruleDecision, ok := failoverActionRuleDecisionWithSources(runnerPolicy.FailoverPolicy, overridePolicy, info); ok { + if ruleDecision, ok := candidateFailoverActionRuleDecision(runnerPolicy.FailoverPolicy, overridePolicy, candidate, info); ok { return ruleDecision } if match, ok := failoverDenyMatchWithSources(runnerPolicy.FailoverPolicy, overridePolicy, info); ok { @@ -125,6 +148,122 @@ func failoverDecisionForCandidate(runnerPolicy store.RunnerPolicy, candidate sto return failoverDecision{Retry: false, Action: "stop", Reason: "client_non_retryable", Match: policyRuleMatch{Source: "provider_client", Policy: "ClientError", Rule: "Retryable", Value: "false"}, Info: info} } +func resolveCandidateFailure(input resolveCandidateFailureInput) failureDecision { + info := failureInfoFromError(input.Err) + if isResultPersistenceFailure(input.Err) { + return failureDecision{ + Route: "stop", + Effect: "none", + Reason: "result_persistence_failed", + Match: policyRuleMatch{Source: "gateway_result_storage", Policy: "persistence", Rule: "providerFailover", Value: "disabled"}, + Info: info, + } + } + if errors.Is(input.Err, store.ErrRateLimited) { + route := "stop" + if input.Async && store.RateLimitRetryable(input.Err) { + route = "requeue" + } + return failureDecision{ + Route: route, + Effect: "none", + Reason: "local_rate_limit_wait_queue", + Match: policyRuleMatch{Source: "gateway_rate_limits", Policy: "rateLimitPolicy", Rule: "localCapacity", Value: "exceeded"}, + Info: info, + } + } + + failover := failoverDecisionForCandidate(input.RunnerPolicy, input.Candidate, input.Err) + effect := failoverEffect(failover.Action) + if input.DownstreamStarted { + return failureDecision{ + Route: "stop", + Effect: effect, + Target: failover.Target, + CooldownSeconds: failover.CooldownSeconds, + DemoteSteps: failover.DemoteSteps, + Reason: "downstream_response_started", + Match: policyRuleMatch{Source: "gateway_stream", Policy: "streamSafety", Rule: "downstreamStarted", Value: "true"}, + Info: info, + } + } + explicitStop := failover.Action == "stop" && (failover.Reason == "failover_action_rule" || + failover.Reason == "hard_stop_policy" || + failover.Reason == "failover_deny_policy" || + failover.Reason == "failover_disabled" || + failover.Reason == "runner_policy_disabled") + if explicitStop { + return failureDecision{ + Route: "stop", + Effect: effect, + Target: failover.Target, + CooldownSeconds: failover.CooldownSeconds, + DemoteSteps: failover.DemoteSteps, + Reason: failover.Reason, + Match: failover.Match, + Info: failover.Info, + } + } + + immediateCandidateInvalidation := failover.Action == "cooldown_and_next" || failover.Action == "disable_and_next" + retry := retryDecisionForCandidate(input.Candidate, input.Err) + if !immediateCandidateInvalidation && retry.Retry && input.ClientAttempt < input.MaxClientAttempts { + return failureDecision{ + Route: "retry_same", + Effect: "none", + Reason: retry.Reason, + Match: retry.Match, + Info: retry.Info, + } + } + + route := "next" + reason := failover.Reason + match := failover.Match + if input.FailoverExpired { + route = "stop" + reason = "failover_time_budget_exceeded" + match = policyRuleMatch{ + Source: "gateway_runner_policies.failover_policy", + Policy: "failoverPolicy", + Rule: "maxDurationSeconds", + Value: "exceeded", + } + } else if !input.HasNextCandidate { + route = "stop" + reason = "no_next_platform" + match = policyRuleMatch{ + Source: "runner_candidates", + Policy: "candidateSelection", + Rule: "candidateCount", + Value: "exhausted", + } + } + return failureDecision{ + Route: route, + Effect: effect, + Target: failover.Target, + CooldownSeconds: failover.CooldownSeconds, + DemoteSteps: failover.DemoteSteps, + Reason: reason, + Match: match, + Info: failover.Info, + } +} + +func failoverEffect(action string) string { + switch action { + case "cooldown_and_next": + return "cooldown" + case "demote_and_next": + return "demote" + case "disable_and_next": + return "disable" + default: + return "none" + } +} + func shouldDemoteCandidatePriority(runnerPolicy store.RunnerPolicy, err error) bool { decision := priorityDemoteDecisionForCandidate(runnerPolicy, err) return decision.Demote @@ -346,6 +485,46 @@ func failoverActionRuleDecisionWithSources(base map[string]any, override map[str return failoverActionRuleDecision(actionRulesFromPolicy(base), "gateway_runner_policies.failover_policy", info) } +func candidateFailoverActionRuleDecision(base map[string]any, override map[string]any, candidate store.RuntimeModelCandidate, info failureInfo) (failoverDecision, bool) { + if rules := actionRulesFromPolicy(override); len(rules) > 0 { + if decision, ok := failoverActionRuleDecision(rules, "runtime_policy_override.failoverPolicy", info); ok { + return decision, true + } + } + autoDisablePolicy := effectiveRuntimePolicy(candidate.AutoDisablePolicy, candidate.RuntimePolicyOverride, "autoDisablePolicy") + if boolFromPolicy(autoDisablePolicy, "enabled", false) && intFromPolicy(autoDisablePolicy, "threshold") <= 1 { + if value, ok := matchingKeywordValue(autoDisablePolicy, "keywords", info.Target); ok { + return failoverDecision{ + Retry: true, + Action: "disable_and_next", + Target: "platform", + Reason: "legacy_auto_disable_policy", + Match: policyRuleMatch{Source: "model_runtime_policy_sets.auto_disable_policy", Policy: "autoDisablePolicy", Rule: "keywords", Value: value}, + Info: info, + }, true + } + } + degradePolicy := effectiveRuntimePolicy(candidate.DegradePolicy, candidate.RuntimePolicyOverride, "degradePolicy") + if boolFromPolicy(degradePolicy, "enabled", false) { + if value, ok := matchingKeywordValue(degradePolicy, "keywords", info.Target); ok { + cooldownSeconds := intFromPolicy(degradePolicy, "cooldownSeconds") + if cooldownSeconds <= 0 { + cooldownSeconds = 300 + } + return failoverDecision{ + Retry: true, + Action: "cooldown_and_next", + Target: "model", + Reason: "legacy_degrade_policy", + CooldownSeconds: cooldownSeconds, + Match: policyRuleMatch{Source: "model_runtime_policy_sets.degrade_policy", Policy: "degradePolicy", Rule: "keywords", Value: value}, + Info: info, + }, true + } + } + return failoverActionRuleDecision(actionRulesFromPolicy(base), "gateway_runner_policies.failover_policy", info) +} + func failoverActionRuleDecision(rules []map[string]any, source string, info failureInfo) (failoverDecision, bool) { for index, rule := range rules { if !boolFromPolicy(rule, "enabled", true) { diff --git a/apps/api/internal/runner/retry_decision_test.go b/apps/api/internal/runner/retry_decision_test.go index aebc1f1..0788a02 100644 --- a/apps/api/internal/runner/retry_decision_test.go +++ b/apps/api/internal/runner/retry_decision_test.go @@ -334,3 +334,151 @@ func TestPriorityDemotePolicyIsSkippedWhenFailoverActionRulesExist(t *testing.T) t.Fatalf("legacy priority demotion should be skipped when actionRules are active, got %+v", decision) } } + +func TestResolveCandidateFailureCooldownSkipsSameClientRetry(t *testing.T) { + decision := resolveCandidateFailure(resolveCandidateFailureInput{ + RunnerPolicy: store.RunnerPolicy{ + Status: "active", + FailoverPolicy: map[string]any{ + "enabled": true, + "actionRules": []any{ + map[string]any{ + "statusCodes": []any{429}, + "action": "cooldown_and_next", + "target": "model", + "cooldownSeconds": 60, + }, + }, + }, + }, + Candidate: store.RuntimeModelCandidate{ + ModelRetryPolicy: map[string]any{"enabled": true, "maxAttempts": 3}, + }, + Err: &clients.ClientError{Code: "resource_exhausted", StatusCode: 429, Retryable: true}, + ClientAttempt: 1, + MaxClientAttempts: 3, + HasNextCandidate: true, + }) + + if decision.Route != "next" || decision.Effect != "cooldown" || decision.Target != "model" || decision.CooldownSeconds != 60 { + t.Fatalf("429 cooldown rule should invalidate the candidate immediately, got %+v", decision) + } +} + +func TestResolveCandidateFailureDemoteRetriesBeforeNext(t *testing.T) { + input := resolveCandidateFailureInput{ + RunnerPolicy: store.RunnerPolicy{ + Status: "active", + FailoverPolicy: map[string]any{ + "enabled": true, + "actionRules": []any{ + map[string]any{ + "statusCodes": []any{500}, + "action": "demote_and_next", + "target": "platform", + "demoteSteps": 2, + }, + }, + }, + }, + Candidate: store.RuntimeModelCandidate{ + ModelRetryPolicy: map[string]any{"enabled": true, "maxAttempts": 2}, + }, + Err: &clients.ClientError{Code: "server_error", StatusCode: 500, Retryable: true}, + ClientAttempt: 1, + MaxClientAttempts: 2, + HasNextCandidate: true, + } + + first := resolveCandidateFailure(input) + if first.Route != "retry_same" || first.Effect != "none" { + t.Fatalf("demotion must wait until same-client retries are exhausted, got %+v", first) + } + + input.ClientAttempt = 2 + second := resolveCandidateFailure(input) + if second.Route != "next" || second.Effect != "demote" || second.DemoteSteps != 2 { + t.Fatalf("exhausted retry should demote once and rotate, got %+v", second) + } +} + +func TestResolveCandidateFailureStopsAfterDownstreamStarts(t *testing.T) { + decision := resolveCandidateFailure(resolveCandidateFailureInput{ + RunnerPolicy: store.RunnerPolicy{ + Status: "active", + FailoverPolicy: map[string]any{ + "enabled": true, + "actionRules": []any{ + map[string]any{ + "categories": []any{"stream_error"}, + "action": "next", + }, + }, + }, + }, + Candidate: store.RuntimeModelCandidate{ModelRetryPolicy: map[string]any{"enabled": true, "maxAttempts": 3}}, + Err: &clients.ClientError{Code: "stream_read_error", Retryable: true}, + ClientAttempt: 1, + MaxClientAttempts: 3, + HasNextCandidate: true, + DownstreamStarted: true, + }) + + if decision.Route != "stop" || decision.Reason != "downstream_response_started" { + t.Fatalf("stream output must prevent retries and failover, got %+v", decision) + } +} + +func TestResolveCandidateFailureLegacyPolicyPrecedence(t *testing.T) { + candidate := store.RuntimeModelCandidate{ + DegradePolicy: map[string]any{ + "enabled": true, + "keywords": []any{"resource_exhausted"}, + "cooldownSeconds": 45, + }, + } + decision := resolveCandidateFailure(resolveCandidateFailureInput{ + RunnerPolicy: store.RunnerPolicy{ + Status: "active", + FailoverPolicy: map[string]any{ + "enabled": true, + "actionRules": []any{ + map[string]any{ + "statusCodes": []any{429}, + "action": "cooldown_and_next", + "target": "platform", + "cooldownSeconds": 90, + }, + }, + }, + }, + Candidate: candidate, + Err: &clients.ClientError{Code: "resource_exhausted", Message: "resource_exhausted", StatusCode: 429, Retryable: true}, + ClientAttempt: 1, + MaxClientAttempts: 3, + HasNextCandidate: true, + }) + + if decision.Match.Policy != "degradePolicy" || decision.Target != "model" || decision.CooldownSeconds != 45 { + t.Fatalf("legacy compatibility rule should precede global action rules, got %+v", decision) + } +} + +func TestResolveCandidateFailureUnmatchedRetryableUsesSameThenNext(t *testing.T) { + input := resolveCandidateFailureInput{ + RunnerPolicy: store.RunnerPolicy{Status: "active", FailoverPolicy: map[string]any{"enabled": true, "actionRules": []any{}}}, + Candidate: store.RuntimeModelCandidate{ModelRetryPolicy: map[string]any{"enabled": true, "maxAttempts": 2}}, + Err: &clients.ClientError{Code: "unknown_transient", Retryable: true}, + ClientAttempt: 1, + MaxClientAttempts: 2, + HasNextCandidate: true, + } + + if decision := resolveCandidateFailure(input); decision.Route != "retry_same" { + t.Fatalf("unmatched retryable error should retry the same client first, got %+v", decision) + } + input.ClientAttempt = 2 + if decision := resolveCandidateFailure(input); decision.Route != "next" || decision.Effect != "none" { + t.Fatalf("unmatched exhausted error should rotate without side effects, got %+v", decision) + } +} diff --git a/apps/api/internal/runner/runtime_policy.go b/apps/api/internal/runner/runtime_policy.go index 131a90e..41c41b0 100644 --- a/apps/api/internal/runner/runtime_policy.go +++ b/apps/api/internal/runner/runtime_policy.go @@ -2,153 +2,59 @@ package runner import ( "context" - "errors" "strings" - "github.com/easyai/easyai-ai-gateway/apps/api/internal/clients" "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" ) -func (s *Service) applyCandidateFailurePolicies(ctx context.Context, taskID string, candidate store.RuntimeModelCandidate, cause error, simulated bool, singleSourceProtected bool) { - code := clients.ErrorCode(cause) - message := "" - if cause != nil { - message = cause.Error() +func (s *Service) applyFailureDecision(ctx context.Context, taskID string, requestedModel string, candidate store.RuntimeModelCandidate, decision failureDecision, singleSourceProtection bool, simulated bool) (store.CandidateFailureEffectResult, error) { + result, err := s.store.ApplyCandidateFailureEffect(ctx, store.CandidateFailureEffectInput{ + Effect: decision.Effect, + Target: decision.Target, + PlatformID: candidate.PlatformID, + PlatformModelID: candidate.PlatformModelID, + RequestedModel: requestedModel, + ModelType: candidate.ModelType, + CooldownSeconds: decision.CooldownSeconds, + DemoteSteps: decision.DemoteSteps, + SingleSourceProtection: singleSourceProtection, + }) + if err != nil { + return result, err } - - autoDisablePolicy := effectiveRuntimePolicy(candidate.AutoDisablePolicy, candidate.RuntimePolicyOverride, "autoDisablePolicy") - if failurePolicyMatches(autoDisablePolicy, code, message) && intFromPolicy(autoDisablePolicy, "threshold") <= 1 { - if singleSourceProtected { - s.emitSingleSourceProtected(ctx, taskID, candidate, "auto_disable", code, simulated) - } else if err := s.store.DisableCandidatePlatform(ctx, candidate.PlatformID); err == nil { - _ = s.emit(ctx, taskID, "task.policy.auto_disabled", "running", "auto_disable", 0.48, "candidate platform disabled by failure policy", map[string]any{ - "platformId": candidate.PlatformID, - "platformModelId": candidate.PlatformModelID, - "code": code, - }, simulated) - } - } - - degradePolicy := effectiveRuntimePolicy(candidate.DegradePolicy, candidate.RuntimePolicyOverride, "degradePolicy") - if failurePolicyMatches(degradePolicy, code, message) { - cooldownSeconds := intFromPolicy(degradePolicy, "cooldownSeconds") - if singleSourceProtected { - s.emitSingleSourceProtected(ctx, taskID, candidate, "degrade_cooldown", code, simulated) - } else if err := s.store.CooldownCandidatePlatformModel(ctx, candidate.PlatformModelID, cooldownSeconds); err == nil { - _ = s.emit(ctx, taskID, "task.policy.degraded", "running", "degrade", 0.5, "candidate model cooled down by failure policy", map[string]any{ - "platformId": candidate.PlatformID, - "platformModelId": candidate.PlatformModelID, - "cooldownSeconds": cooldownSeconds, - "code": code, - }, simulated) - } - } -} - -func (s *Service) applyFailoverAction(ctx context.Context, taskID string, candidate store.RuntimeModelCandidate, requestedModel string, decision failoverDecision, simulated bool, singleSourceProtected bool) { - switch decision.Action { - case "disable_and_next": - if singleSourceProtected { - s.emitSingleSourceProtected(ctx, taskID, candidate, decision.Action, decision.Info.Code, simulated) - return - } - target := decision.Target - if target == "" { - target = "platform" - } - var err error - if target == "model" { - err = s.store.DisableCandidatePlatformModel(ctx, candidate.PlatformModelID) - } else { - err = s.store.DisableCandidatePlatform(ctx, candidate.PlatformID) - } - if err == nil { - _ = s.emit(ctx, taskID, "task.policy.failover_disabled", "running", "failover", 0.51, "candidate platform disabled by runner failover policy", addPolicyTracePayload(map[string]any{ - "platformId": candidate.PlatformID, - "platformModelId": candidate.PlatformModelID, - "action": decision.Action, - "target": target, - "reason": decision.Reason, - }, decision.Match, decision.Info), simulated) - } - case "cooldown_and_next": - if singleSourceProtected { - s.emitSingleSourceProtected(ctx, taskID, candidate, decision.Action, decision.Info.Code, simulated) - return - } - target := decision.Target - if target == "" { - target = "model" - } - var err error - if target == "platform" { - err = s.store.CooldownCandidatePlatform(ctx, candidate.PlatformID, decision.CooldownSeconds) - } else { - err = s.store.CooldownCandidatePlatformModel(ctx, candidate.PlatformModelID, decision.CooldownSeconds) - } - if err == nil { - _ = s.emit(ctx, taskID, "task.policy.failover_cooled_down", "running", "failover", 0.51, "candidate model cooled down by runner failover policy", addPolicyTracePayload(map[string]any{ - "platformId": candidate.PlatformID, - "platformModelId": candidate.PlatformModelID, - "cooldownSeconds": decision.CooldownSeconds, - "action": decision.Action, - "target": target, - "reason": decision.Reason, - }, decision.Match, decision.Info), simulated) - } - case "demote_and_next": - demoteSteps := decision.DemoteSteps - if demoteSteps <= 0 { - demoteSteps = 1 - } - if dynamicPriority, err := s.store.DemoteCandidatePlatformPriorityBySteps(ctx, candidate.PlatformID, candidate.PlatformModelID, requestedModel, candidate.ModelType, demoteSteps); err == nil { - _ = s.emit(ctx, taskID, "task.policy.priority_demoted", "running", "priority_demote", 0.52, "candidate platform priority demoted by runner failover policy", addPolicyTracePayload(map[string]any{ - "platformId": candidate.PlatformID, - "platformModelId": candidate.PlatformModelID, - "dynamicPriority": dynamicPriority, - "demoteSteps": demoteSteps, - "action": decision.Action, - "target": "platform", - "reason": decision.Reason, - "errorMessage": decision.Info.Message, - }, decision.Match, decision.Info), simulated) - } - } -} - -func (s *Service) emitSingleSourceProtected(ctx context.Context, taskID string, candidate store.RuntimeModelCandidate, action string, code string, simulated bool) { - _ = s.emit(ctx, taskID, "task.policy.single_source_protected", "running", "policy_guard", 0.5, "single source protected from disable or cooldown", map[string]any{ - "platformId": candidate.PlatformID, - "platformModelId": candidate.PlatformModelID, - "action": action, - "code": code, - }, simulated) -} - -func (s *Service) applyPriorityDemotePolicy(ctx context.Context, taskID string, attemptNo int, runnerPolicy store.RunnerPolicy, candidate store.RuntimeModelCandidate, requestedModel string, cause error, simulated bool) { - if errors.Is(cause, store.ErrRateLimited) { - return - } - decision := priorityDemoteDecisionForCandidate(runnerPolicy, cause) - if !decision.Demote { - return - } - demoteSteps := decision.DemoteSteps - if demoteSteps <= 0 { - demoteSteps = 99 - } - if dynamicPriority, err := s.store.DemoteCandidatePlatformPriorityBySteps(ctx, candidate.PlatformID, candidate.PlatformModelID, requestedModel, candidate.ModelType, demoteSteps); err == nil { - s.recordAttemptTrace(ctx, taskID, attemptNo, priorityDemoteTraceEntry(decision, candidate.PlatformID, candidate.PlatformModelID, dynamicPriority)) - _ = s.emit(ctx, taskID, "task.policy.priority_demoted", "running", "priority_demote", 0.52, "candidate platform priority demoted by runner policy", addPolicyTracePayload(map[string]any{ + if result.GuardReason == "single_source" { + _ = s.emit(ctx, taskID, "task.policy.single_source_protected", "running", "policy_guard", 0.5, "single source protected from disable or cooldown", addPolicyTracePayload(map[string]any{ "platformId": candidate.PlatformID, "platformModelId": candidate.PlatformModelID, - "dynamicPriority": dynamicPriority, - "demoteSteps": demoteSteps, - "code": clients.ErrorCode(cause), - "reason": decision.Reason, - "errorMessage": messageFromError(cause), + "effect": decision.Effect, + "target": decision.Target, + "guardReason": result.GuardReason, }, decision.Match, decision.Info), simulated) + return result, nil } + if !result.Applied { + return result, nil + } + + payload := addPolicyTracePayload(map[string]any{ + "platformId": candidate.PlatformID, + "platformModelId": candidate.PlatformModelID, + "effect": decision.Effect, + "target": decision.Target, + "reason": decision.Reason, + }, decision.Match, decision.Info) + switch decision.Effect { + case "disable": + _ = s.emit(ctx, taskID, "task.policy.failover_disabled", "running", "failover", 0.51, "candidate disabled by runner failover policy", payload, simulated) + case "cooldown": + payload["cooldownSeconds"] = decision.CooldownSeconds + _ = s.emit(ctx, taskID, "task.policy.failover_cooled_down", "running", "failover", 0.51, "candidate cooled down by runner failover policy", payload, simulated) + case "demote": + payload["demoteSteps"] = decision.DemoteSteps + payload["dynamicPriority"] = result.DynamicPriority + _ = s.emit(ctx, taskID, "task.policy.priority_demoted", "running", "priority_demote", 0.52, "candidate platform priority demoted by runner failover policy", payload, simulated) + } + return result, nil } func messageFromError(err error) string { diff --git a/apps/api/internal/runner/service.go b/apps/api/internal/runner/service.go index b78e751..cfdd3c1 100644 --- a/apps/api/internal/runner/service.go +++ b/apps/api/internal/runner/service.go @@ -11,6 +11,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/easyai/easyai-ai-gateway/apps/api/internal/auth" @@ -269,6 +270,20 @@ func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask, if task.Kind == "responses" && responseExecution.PreviousChain != nil { err = responseChainUnavailableError() } + if task.AsyncMode && store.ModelCandidateRetryAfter(err) > 0 { + queued, delay, queueErr := s.requeueModelCoolingTask(ctx, task, err) + if queueErr != nil { + return Result{}, queueErr + } + return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay} + } + if store.ModelCandidateRetryAfter(err) > 0 { + failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err) + if finishErr != nil { + return Result{}, finishErr + } + return Result{Task: failed, Output: failed.Result}, err + } s.recordFailedAttempt(ctx, failedAttemptRecord{ Task: task, Body: body, @@ -539,18 +554,38 @@ func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask, maxPlatforms := maxPlatformsForCandidates(candidates, runnerPolicy) maxFailoverDuration := maxFailoverDurationForCandidates(candidates, runnerPolicy) - singleSourceProtected := singleSourceProtectionActive(candidates, maxPlatforms, runnerPolicy) + singleSourceProtection := runnerSingleSourceProtectionEnabled(runnerPolicy) + var downstreamStarted atomic.Bool + candidateDelta := onDelta + if onDelta != nil { + candidateDelta = func(event clients.StreamDeltaEvent) error { + downstreamStarted.Store(true) + return onDelta(event) + } + } var lastErr error var lastCandidate store.RuntimeModelCandidate var lastPreprocessing *parameterPreprocessingLog + platformsVisited := 0 candidatesLoop: for index, candidate := range candidates { - if index >= maxPlatforms { + if platformsVisited >= maxPlatforms { break } + available, availabilityErr := s.store.RuntimeCandidateAvailable(ctx, candidate.PlatformID, candidate.PlatformModelID) + if availabilityErr != nil { + return Result{}, availabilityErr + } + if !available { + continue + } + platformsVisited++ lastCandidate = candidate clientAttempts := clientAttemptsForCandidate(candidate) var candidateErr error + var candidateDecision failureDecision + var candidateDecisionAttempt int + var candidateClientAttempt int for clientAttempt := 1; clientAttempt <= clientAttempts; clientAttempt++ { nextAttemptNo := attemptNo + 1 preprocessing, ok := preprocessingByCandidate[pricingCandidateKey(candidate)] @@ -579,7 +614,7 @@ candidatesLoop: } candidateBody := preprocessing.Body candidatePricing := pricingByCandidate[pricingCandidateKey(candidate)] - response, err := s.runCandidate(ctx, task, user, candidateBody, preprocessing.Log, candidate, candidatePricing, nextAttemptNo, onDelta, responseExecution, singleSourceProtected, runnerPolicy.CacheAffinityPolicy, cacheAffinityKeys.Record) + response, err := s.runCandidate(ctx, task, user, candidateBody, preprocessing.Log, candidate, candidatePricing, nextAttemptNo, candidateDelta, responseExecution, runnerPolicy.CacheAffinityPolicy, cacheAffinityKeys.Record) if err != nil && isVolcesRemoteTaskCancellation(candidate, err) { cancelled, changed, cancelErr := s.store.CancelSubmittedTask(ctx, task.ID, task.ExecutionToken, "任务已由火山引擎取消") if cancelErr != nil { @@ -748,7 +783,18 @@ candidatesLoop: if isLocalRateLimitError(err) { lastErr = err candidateErr = err - if task.AsyncMode && store.RateLimitRetryable(err) { + candidateDecision = resolveCandidateFailure(resolveCandidateFailureInput{ + RunnerPolicy: runnerPolicy, + Candidate: candidate, + Err: err, + ClientAttempt: clientAttempt, + MaxClientAttempts: clientAttempts, + HasNextCandidate: platformsVisited < maxPlatforms && index+1 < len(candidates), + FailoverExpired: failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration), + Async: task.AsyncMode, + DownstreamStarted: downstreamStarted.Load(), + }) + if candidateDecision.Route == "requeue" { queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, err, candidate) if queueErr != nil { return Result{}, queueErr @@ -768,104 +814,88 @@ candidatesLoop: ExtraMetrics: []map[string]any{parameterPreprocessingMetrics(preprocessing.Log)}, ModelType: candidate.ModelType, }) + candidateDecisionAttempt = attemptNo + candidateClientAttempt = clientAttempt + s.recordAttemptTrace(ctx, task.ID, attemptNo, failureDecisionTraceEntry(candidateDecision, candidate, clientAttempt, clientAttempts, false, "")) break candidatesLoop } attemptNo = nextAttemptNo lastErr = err candidateErr = err - retryDecision := retryDecisionForCandidate(candidate, err) - retryAction := "retry" - if !retryDecision.Retry { - retryAction = "stop" - } - if clientAttempt >= clientAttempts { - retryDecision.Retry = false - retryDecision.Reason = "same_client_max_attempts" - retryDecision.Match = policyRuleMatch{ - Source: "model_runtime_policy_sets.retry_policy", - Policy: "retryPolicy", - Rule: "maxAttempts", - Value: strconv.Itoa(clientAttempts), - } - retryDecision.Info = failureInfoFromError(err) - retryAction = "stop" - } - if failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration) { - retryDecision.Retry = false - retryDecision.Reason = "failover_time_budget_exceeded" - retryDecision.Match = policyRuleMatch{ - Source: "gateway_runner_policies.failover_policy", - Policy: "failoverPolicy", - Rule: "maxDurationSeconds", - Value: strconv.Itoa(int(maxFailoverDuration.Seconds())), - } - retryDecision.Info = failureInfoFromError(err) - retryAction = "stop" - } - s.recordAttemptTrace(ctx, task.ID, attemptNo, retryTraceEntry(candidate, retryDecision, retryAction, clientAttempt, clientAttempts)) - if !retryDecision.Retry { + candidateDecision = resolveCandidateFailure(resolveCandidateFailureInput{ + RunnerPolicy: runnerPolicy, + Candidate: candidate, + Err: err, + ClientAttempt: clientAttempt, + MaxClientAttempts: clientAttempts, + HasNextCandidate: platformsVisited < maxPlatforms && index+1 < len(candidates), + FailoverExpired: failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration), + Async: task.AsyncMode, + DownstreamStarted: downstreamStarted.Load(), + }) + candidateDecisionAttempt = attemptNo + candidateClientAttempt = clientAttempt + if candidateDecision.Route != "retry_same" { break } + s.recordAttemptTrace(ctx, task.ID, attemptNo, failureDecisionTraceEntry(candidateDecision, candidate, clientAttempt, clientAttempts, false, "")) if err := s.emit(ctx, task.ID, "task.retrying", "running", "retry", 0.45, "retrying same client", addPolicyTracePayload(map[string]any{ "attempt": attemptNo, "clientAttempt": clientAttempt, "clientId": candidate.ClientID, "error": err.Error(), - "reason": retryDecision.Reason, + "reason": candidateDecision.Reason, "scope": "same_client", - }, retryDecision.Match, retryDecision.Info), isSimulation(task, candidate)); err != nil { + }, candidateDecision.Match, candidateDecision.Info), isSimulation(task, candidate)); err != nil { return Result{}, err } } - if candidateErr == nil || index+1 >= len(candidates) || index+1 >= maxPlatforms { - if candidateErr != nil { - s.applyPriorityDemotePolicy(ctx, task.ID, attemptNo, runnerPolicy, candidate, task.Model, candidateErr, isSimulation(task, candidate)) - decision := failoverDecisionForCandidate(runnerPolicy, candidate, candidateErr) - if decision.Retry { - decision.Retry = false - decision.Action = "stop" - decision.Reason = "no_next_platform" - decision.Match = policyRuleMatch{Source: "runner_candidates", Policy: "candidateSelection", Rule: "candidateCount", Value: strconv.Itoa(len(candidates))} - if index+1 >= maxPlatforms { - decision.Reason = "max_platforms_reached" - decision.Match = policyRuleMatch{Source: "gateway_runner_policies.failover_policy", Policy: "failoverPolicy", Rule: "maxPlatforms", Value: strconv.Itoa(maxPlatforms)} - } + if candidateErr == nil { + break + } + effectResult, effectErr := s.applyFailureDecision( + context.WithoutCancel(ctx), + task.ID, + task.Model, + candidate, + candidateDecision, + singleSourceProtection, + isSimulation(task, candidate), + ) + if effectErr != nil { + s.logger.Warn("apply candidate failure effect failed", "taskID", task.ID, "effect", candidateDecision.Effect, "error", effectErr) + effectResult.GuardReason = "effect_error" + } + s.recordAttemptTrace(ctx, task.ID, candidateDecisionAttempt, failureDecisionTraceEntry( + candidateDecision, + candidate, + candidateClientAttempt, + clientAttempts, + effectResult.Applied, + effectResult.GuardReason, + )) + if candidateDecision.Route != "next" { + if candidateDecision.Reason == "failover_time_budget_exceeded" { + elapsedSeconds := int(time.Since(executeStartedAt).Seconds()) + if err := s.emit(ctx, task.ID, "task.failover.stopped", "running", "retry", 0.55, "failover time budget exceeded", map[string]any{ + "elapsedSeconds": elapsedSeconds, + "maxDurationSeconds": int(maxFailoverDuration.Seconds()), + "scope": "next_platform", + "statusCode": clients.ErrorResponseMetadata(candidateErr).StatusCode, + }, isSimulation(task, candidate)); err != nil { + return Result{}, err } - s.recordAttemptTrace(ctx, task.ID, attemptNo, failoverTraceEntry(decision, candidate)) } break } - s.applyPriorityDemotePolicy(ctx, task.ID, attemptNo, runnerPolicy, candidate, task.Model, candidateErr, isSimulation(task, candidate)) - if failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration) { - elapsedSeconds := int(time.Since(executeStartedAt).Seconds()) - maxDurationSeconds := int(maxFailoverDuration.Seconds()) - s.recordAttemptTrace(ctx, task.ID, attemptNo, failoverTimeBudgetTraceEntry(elapsedSeconds, maxDurationSeconds, failureInfoFromError(candidateErr), candidate)) - if err := s.emit(ctx, task.ID, "task.failover.stopped", "running", "retry", 0.55, "failover time budget exceeded", map[string]any{ - "elapsedSeconds": elapsedSeconds, - "maxDurationSeconds": maxDurationSeconds, - "scope": "next_platform", - "statusCode": clients.ErrorResponseMetadata(candidateErr).StatusCode, - }, isSimulation(task, candidate)); err != nil { - return Result{}, err - } - break - } - decision := failoverDecisionForCandidate(runnerPolicy, candidate, candidateErr) - if !decision.Retry && !isResultPersistenceFailure(candidateErr) && hasLoadAvoidanceFallback(candidates, index, maxPlatforms) { - decision = loadAvoidanceFallbackDecision(candidateErr) - } - s.recordAttemptTrace(ctx, task.ID, attemptNo, failoverTraceEntry(decision, candidate)) - if !decision.Retry { - break - } - s.applyFailoverAction(ctx, task.ID, candidate, task.Model, decision, isSimulation(task, candidate), singleSourceProtected) if err := s.emit(ctx, task.ID, "task.retrying", "running", "retry", 0.55, "retrying next client", addPolicyTracePayload(map[string]any{ "attempt": attemptNo, - "action": decision.Action, + "route": candidateDecision.Route, + "effect": candidateDecision.Effect, "error": candidateErr.Error(), - "reason": decision.Reason, + "reason": candidateDecision.Reason, "scope": "next_platform", - }, decision.Match, decision.Info), isSimulation(task, candidate)); err != nil { + }, candidateDecision.Match, candidateDecision.Info), isSimulation(task, candidate)); err != nil { return Result{}, err } } @@ -951,7 +981,7 @@ func billingItemsFixedTotal(items []any) fixedAmount { return total } -func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user *auth.User, body map[string]any, preprocessing parameterPreprocessingLog, candidate store.RuntimeModelCandidate, pricing resolvedPricing, attemptNo int, onDelta clients.StreamDelta, responseExecution responseExecutionContext, singleSourceProtected bool, cacheAffinityPolicy map[string]any, cacheAffinityRecordKeys []string) (clients.Response, error) { +func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user *auth.User, body map[string]any, preprocessing parameterPreprocessingLog, candidate store.RuntimeModelCandidate, pricing resolvedPricing, attemptNo int, onDelta clients.StreamDelta, responseExecution responseExecutionContext, cacheAffinityPolicy map[string]any, cacheAffinityRecordKeys []string) (clients.Response, error) { var err error candidate, err = candidateWithEnvironmentCredentials(candidate) if err != nil { @@ -1230,7 +1260,6 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user if !simulated && submissionStatus == "submitting" { return clients.Response{}, &upstreamSubmissionUnknownError{AttemptID: attemptID, Cause: err} } - s.applyCandidateFailurePolicies(ctx, task.ID, candidate, err, simulated, singleSourceProtected) return clients.Response{}, err } uploadedResult, err := s.uploadGeneratedAssets(ctx, task.ID, task.Kind, response.Result) @@ -1669,6 +1698,34 @@ func (s *Service) requeueRateLimitedTask(ctx context.Context, task store.Gateway return queued, delay, nil } +func (s *Service) requeueModelCoolingTask(ctx context.Context, task store.GatewayTask, cause error) (store.GatewayTask, time.Duration, error) { + delay := store.ModelCandidateRetryAfter(cause) + recoveryAt := store.ModelCandidateRecoveryAt(cause) + if recoveryAt.IsZero() { + recoveryAt = time.Now().Add(delay) + } + if delay <= 0 { + delay = time.Until(recoveryAt) + } + if delay < time.Second { + delay = time.Second + recoveryAt = time.Now().Add(delay) + } + queued, err := s.store.RequeueTaskUntil(ctx, task.ID, task.ExecutionToken, recoveryAt, "") + if err != nil { + return store.GatewayTask{}, 0, err + } + payload := map[string]any{ + "code": store.ModelCandidateErrorCode(cause), + "retryAfterSeconds": int((delay + time.Second - 1) / time.Second), + "recoveryAt": recoveryAt.UTC().Format(time.RFC3339), + } + if eventErr := s.emit(ctx, task.ID, "task.queued", "queued", "model_cooling_down", 0.2, "task queued until a model candidate recovers", payload, task.RunMode == "simulation"); eventErr != nil { + return store.GatewayTask{}, 0, eventErr + } + return queued, delay, nil +} + func taskRetryJitter(taskID string) time.Duration { var sum uint32 for _, value := range []byte(taskID) { diff --git a/apps/api/internal/runner/trace.go b/apps/api/internal/runner/trace.go index d585bcf..7a8fa40 100644 --- a/apps/api/internal/runner/trace.go +++ b/apps/api/internal/runner/trace.go @@ -48,6 +48,29 @@ func failoverTraceEntry(decision failoverDecision, candidate store.RuntimeModelC return entry } +func failureDecisionTraceEntry(decision failureDecision, candidate store.RuntimeModelCandidate, clientAttempt int, maxAttempts int, applied bool, guardReason string) map[string]any { + entry := policyTraceEntry("failure_decision", "candidate_failure", decision.Route, decision.Reason, decision.Match, decision.Info) + entry["route"] = decision.Route + entry["effect"] = decision.Effect + entry["applied"] = applied + entry["guardReason"] = guardReason + entry["clientAttempt"] = clientAttempt + entry["maxAttempts"] = maxAttempts + entry["rule"] = decision.Match.Rule + entry["value"] = decision.Match.Value + if decision.Target != "" { + entry["target"] = decision.Target + } + if decision.CooldownSeconds > 0 { + entry["cooldownSeconds"] = decision.CooldownSeconds + } + if decision.DemoteSteps > 0 { + entry["demoteSteps"] = decision.DemoteSteps + } + addCandidatePriorityTraceFields(entry, candidate) + return entry +} + func priorityDemoteTraceEntry(decision priorityDemoteDecision, platformID string, platformModelID string, dynamicPriority int) map[string]any { entry := policyTraceEntry("priority_demoted", "priority_demote", "demote", decision.Reason, decision.Match, decision.Info) entry["demote"] = decision.Demote diff --git a/apps/api/internal/store/candidates.go b/apps/api/internal/store/candidates.go index d43e8e1..d532cfb 100644 --- a/apps/api/internal/store/candidates.go +++ b/apps/api/internal/store/candidates.go @@ -2,12 +2,15 @@ package store import ( "context" + "errors" "fmt" "math" "sort" "strings" + "time" "github.com/easyai/easyai-ai-gateway/apps/api/internal/auth" + "github.com/jackc/pgx/v5" ) type ListModelCandidatesOptions struct { @@ -810,13 +813,20 @@ func runtimeCandidateFull(candidate RuntimeModelCandidate) bool { func (s *Store) modelCandidateCooldownError(ctx context.Context, model string, modelType string) (error, error) { exactModel := strings.TrimSpace(model) - rows, err := s.pool.Query(ctx, ` -SELECT p.name, - COALESCE(NULLIF(m.display_name, ''), NULLIF(m.model_alias, ''), m.model_name), - COALESCE(to_char(p.cooldown_until AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'), ''), - GREATEST(COALESCE(EXTRACT(EPOCH FROM p.cooldown_until - now()), 0), 0)::float8, - COALESCE(to_char(m.cooldown_until AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'), ''), - GREATEST(COALESCE(EXTRACT(EPOCH FROM m.cooldown_until - now()), 0), 0)::float8 + var code string + var recoveryAt time.Time + var remainingSeconds float64 + err := s.pool.QueryRow(ctx, ` +SELECT CASE + WHEN COALESCE(m.cooldown_until, to_timestamp(0)) >= COALESCE(p.cooldown_until, to_timestamp(0)) + THEN 'model_cooling_down' + ELSE 'platform_cooling_down' + END, + GREATEST(COALESCE(p.cooldown_until, to_timestamp(0)), COALESCE(m.cooldown_until, to_timestamp(0))) AS recovery_at, + GREATEST( + EXTRACT(EPOCH FROM GREATEST(COALESCE(p.cooldown_until, to_timestamp(0)), COALESCE(m.cooldown_until, to_timestamp(0))) - now()), + 0 + )::float8 FROM platform_models m JOIN integration_platforms p ON p.id = m.platform_id LEFT JOIN base_model_catalog b ON b.id = m.base_model_id @@ -836,64 +846,39 @@ WHERE p.status = 'enabled' AND compatibility_alias.active = true AND (compatibility_alias.expires_at IS NULL OR compatibility_alias.expires_at > now()) AND (compatibility_alias.base_model_id = b.id OR alias_base.invocation_name = b.invocation_name) - ) - ) -ORDER BY GREATEST(COALESCE(p.cooldown_until, to_timestamp(0)), COALESCE(m.cooldown_until, to_timestamp(0))) DESC, - p.priority ASC, - m.created_at ASC`, exactModel, modelType) + ) + ) + AND ( + p.cooldown_until > now() + OR m.cooldown_until > now() + ) +ORDER BY GREATEST(COALESCE(p.cooldown_until, to_timestamp(0)), COALESCE(m.cooldown_until, to_timestamp(0))) ASC, + p.priority ASC, + m.created_at ASC +LIMIT 1`, exactModel, modelType).Scan(&code, &recoveryAt, &remainingSeconds) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } if err != nil { return nil, err } - defer rows.Close() - - for rows.Next() { - var platformName string - var displayName string - var platformCooldownUntil string - var platformRemainingSeconds float64 - var modelCooldownUntil string - var modelRemainingSeconds float64 - if err := rows.Scan( - &platformName, - &displayName, - &platformCooldownUntil, - &platformRemainingSeconds, - &modelCooldownUntil, - &modelRemainingSeconds, - ); err != nil { - return nil, err - } - if modelRemainingSeconds > 0 { - return &ModelCandidateUnavailableError{ - Code: "model_cooling_down", - Message: cooldownErrorMessage("模型", displayName, modelRemainingSeconds, modelCooldownUntil), - }, nil - } - if platformRemainingSeconds > 0 { - return &ModelCandidateUnavailableError{ - Code: "platform_cooling_down", - Message: cooldownErrorMessage("平台", platformName, platformRemainingSeconds, platformCooldownUntil), - }, nil - } + retryAfterSeconds := int(math.Ceil(remainingSeconds)) + if retryAfterSeconds < 1 { + retryAfterSeconds = 1 } - if err := rows.Err(); err != nil { - return nil, err + message := "请求的模型暂时不可用，请稍后重试" + if code == "platform_cooling_down" { + message = "可用平台暂时不可用，请稍后重试" } - return nil, nil -} - -func cooldownErrorMessage(scope string, name string, remainingSeconds float64, cooldownUntil string) string { - name = strings.TrimSpace(name) - if name == "" { - name = "候选" - } - remainingMinutes := remainingSeconds / 60 - if remainingMinutes < 0.1 { - remainingMinutes = 0.1 - } - message := fmt.Sprintf("%s %s 冷却中，剩余 %.1f 分钟", scope, name, remainingMinutes) - if strings.TrimSpace(cooldownUntil) != "" { - message += "，预计恢复时间 " + cooldownUntil - } - return message + recoveryAt = recoveryAt.UTC() + return &ModelCandidateUnavailableError{ + Code: code, + Message: message, + RetryAfter: time.Duration(retryAfterSeconds) * time.Second, + RecoveryAt: recoveryAt, + Details: map[string]any{ + "retryAfterSeconds": retryAfterSeconds, + "recoveryAt": recoveryAt.Format(time.RFC3339), + }, + }, nil } diff --git a/apps/api/internal/store/runtime_policies.go b/apps/api/internal/store/runtime_policies.go index 3b04730..a2dfe11 100644 --- a/apps/api/internal/store/runtime_policies.go +++ b/apps/api/internal/store/runtime_policies.go @@ -8,6 +8,7 @@ import ( "time" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" ) const runtimePolicyColumns = ` @@ -38,6 +39,24 @@ type PlatformDynamicPriorityState struct { UpdatedAt time.Time `json:"updatedAt"` } +type CandidateFailureEffectInput struct { + Effect string + Target string + PlatformID string + PlatformModelID string + RequestedModel string + ModelType string + CooldownSeconds int + DemoteSteps int + SingleSourceProtection bool +} + +type CandidateFailureEffectResult struct { + Applied bool + GuardReason string + DynamicPriority int +} + func (s *Store) ListRuntimePolicySets(ctx context.Context) ([]RuntimePolicySet, error) { rows, err := s.pool.Query(ctx, `SELECT `+runtimePolicyColumns+` FROM model_runtime_policy_sets ORDER BY policy_key ASC`) if err != nil { @@ -171,6 +190,164 @@ WHERE id = $1::uuid`, platformModelID, cooldownSeconds) return err } +func (s *Store) RuntimeCandidateAvailable(ctx context.Context, platformID string, platformModelID string) (bool, error) { + if strings.TrimSpace(platformID) == "" || strings.TrimSpace(platformModelID) == "" { + return false, nil + } + var available bool + err := s.pool.QueryRow(ctx, ` +SELECT EXISTS ( + SELECT 1 + FROM platform_models model + JOIN integration_platforms platform ON platform.id = model.platform_id + WHERE platform.id = $1::uuid + AND model.id = $2::uuid + AND platform.status = 'enabled' + AND platform.deleted_at IS NULL + AND (platform.cooldown_until IS NULL OR platform.cooldown_until <= now()) + AND model.enabled = true + AND (model.cooldown_until IS NULL OR model.cooldown_until <= now()) +)`, platformID, platformModelID).Scan(&available) + return available, err +} + +func (s *Store) ApplyCandidateFailureEffect(ctx context.Context, input CandidateFailureEffectInput) (CandidateFailureEffectResult, error) { + if input.Effect == "" || input.Effect == "none" { + return CandidateFailureEffectResult{}, nil + } + if input.Effect == "demote" { + steps := input.DemoteSteps + if steps <= 0 { + steps = 1 + } + priority, err := s.DemoteCandidatePlatformPriorityBySteps( + ctx, + input.PlatformID, + input.PlatformModelID, + input.RequestedModel, + input.ModelType, + steps, + ) + return CandidateFailureEffectResult{Applied: err == nil, DynamicPriority: priority}, err + } + if input.Effect != "cooldown" && input.Effect != "disable" { + return CandidateFailureEffectResult{}, nil + } + + tx, err := s.pool.Begin(ctx) + if err != nil { + return CandidateFailureEffectResult{}, err + } + defer func() { + _ = tx.Rollback(ctx) + }() + + lockKey := strings.TrimSpace(input.RequestedModel) + "\x1f" + strings.TrimSpace(input.ModelType) + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1::text, 0))`, lockKey); err != nil { + return CandidateFailureEffectResult{}, err + } + if input.SingleSourceProtection { + alternativeAvailable, err := candidateAlternativeAvailable(ctx, tx, input) + if err != nil { + return CandidateFailureEffectResult{}, err + } + if !alternativeAvailable { + if err := tx.Commit(ctx); err != nil { + return CandidateFailureEffectResult{}, err + } + return CandidateFailureEffectResult{GuardReason: "single_source"}, nil + } + } + + var commandTag pgconn.CommandTag + switch { + case input.Effect == "disable" && input.Target == "model": + commandTag, err = tx.Exec(ctx, ` +UPDATE platform_models +SET enabled = false, + updated_at = now() +WHERE id = $1::uuid + AND enabled = true`, input.PlatformModelID) + case input.Effect == "disable": + commandTag, err = tx.Exec(ctx, ` +UPDATE integration_platforms +SET status = 'disabled', + updated_at = now() +WHERE id = $1::uuid + AND status = 'enabled' + AND deleted_at IS NULL`, input.PlatformID) + case input.Effect == "cooldown" && input.Target == "platform": + commandTag, err = tx.Exec(ctx, ` +UPDATE integration_platforms +SET cooldown_until = GREATEST( + COALESCE(cooldown_until, to_timestamp(0)), + now() + ($2::int * interval '1 second') + ), + updated_at = now() +WHERE id = $1::uuid + AND status = 'enabled' + AND deleted_at IS NULL`, input.PlatformID, positiveCooldownSeconds(input.CooldownSeconds)) + default: + commandTag, err = tx.Exec(ctx, ` +UPDATE platform_models +SET cooldown_until = GREATEST( + COALESCE(cooldown_until, to_timestamp(0)), + now() + ($2::int * interval '1 second') + ), + updated_at = now() +WHERE id = $1::uuid + AND enabled = true`, input.PlatformModelID, positiveCooldownSeconds(input.CooldownSeconds)) + } + if err != nil { + return CandidateFailureEffectResult{}, err + } + applied := commandTag.RowsAffected() > 0 + if err := tx.Commit(ctx); err != nil { + return CandidateFailureEffectResult{}, err + } + return CandidateFailureEffectResult{Applied: applied}, nil +} + +func candidateAlternativeAvailable(ctx context.Context, tx pgx.Tx, input CandidateFailureEffectInput) (bool, error) { + var available bool + err := tx.QueryRow(ctx, ` +SELECT EXISTS ( + SELECT 1 + FROM platform_models model + JOIN integration_platforms platform ON platform.id = model.platform_id + LEFT JOIN base_model_catalog base ON base.id = model.base_model_id + WHERE platform.id <> $1::uuid + AND platform.status = 'enabled' + AND platform.deleted_at IS NULL + AND (platform.cooldown_until IS NULL OR platform.cooldown_until <= now()) + AND model.enabled = true + AND (model.cooldown_until IS NULL OR model.cooldown_until <= now()) + AND (NULLIF($3::text, '') IS NULL OR model.model_type @> jsonb_build_array($3::text)) + AND ( + base.invocation_name = $2::text + OR (base.id IS NULL AND model.model_name = $2::text) + OR EXISTS ( + SELECT 1 + FROM model_compatibility_aliases compatibility_alias + JOIN base_model_catalog alias_base ON alias_base.id = compatibility_alias.base_model_id + WHERE compatibility_alias.alias = $2::text + AND compatibility_alias.model_type = $3::text + AND compatibility_alias.active = true + AND (compatibility_alias.expires_at IS NULL OR compatibility_alias.expires_at > now()) + AND (compatibility_alias.base_model_id = base.id OR alias_base.invocation_name = base.invocation_name) + ) + ) +)`, input.PlatformID, strings.TrimSpace(input.RequestedModel), strings.TrimSpace(input.ModelType)).Scan(&available) + return available, err +} + +func positiveCooldownSeconds(value int) int { + if value <= 0 { + return 300 + } + return value +} + func (s *Store) DemoteCandidatePlatformPriority(ctx context.Context, platformID string, platformModelID string, requestedModel string, modelType string) (int, error) { return s.DemoteCandidatePlatformPriorityBySteps(ctx, platformID, platformModelID, requestedModel, modelType, 99) } diff --git a/apps/api/internal/store/runtime_types.go b/apps/api/internal/store/runtime_types.go index 6162353..cdf7640 100644 --- a/apps/api/internal/store/runtime_types.go +++ b/apps/api/internal/store/runtime_types.go @@ -13,9 +13,11 @@ var ( ) type ModelCandidateUnavailableError struct { - Code string - Message string - Details map[string]any + Code string + Message string + Details map[string]any + RetryAfter time.Duration + RecoveryAt time.Time } func (e *ModelCandidateUnavailableError) Error() string { @@ -42,6 +44,22 @@ func ModelCandidateErrorDetails(err error) map[string]any { return nil } +func ModelCandidateRetryAfter(err error) time.Duration { + var candidateErr *ModelCandidateUnavailableError + if errors.As(err, &candidateErr) && candidateErr.RetryAfter > 0 { + return candidateErr.RetryAfter + } + return 0 +} + +func ModelCandidateRecoveryAt(err error) time.Time { + var candidateErr *ModelCandidateUnavailableError + if errors.As(err, &candidateErr) { + return candidateErr.RecoveryAt + } + return time.Time{} +} + type RateLimitExceededError struct { ScopeType string ScopeKey string diff --git a/apps/api/internal/store/tasks_runtime.go b/apps/api/internal/store/tasks_runtime.go index 48dfe38..03f9139 100644 --- a/apps/api/internal/store/tasks_runtime.go +++ b/apps/api/internal/store/tasks_runtime.go @@ -470,6 +470,13 @@ func (s *Store) RequeueTask(ctx context.Context, taskID string, executionToken s delay = 10 * time.Minute } nextRunAt := time.Now().Add(delay) + return s.RequeueTaskUntil(ctx, taskID, executionToken, nextRunAt, queueKey) +} + +func (s *Store) RequeueTaskUntil(ctx context.Context, taskID string, executionToken string, nextRunAt time.Time, queueKey string) (GatewayTask, error) { + if nextRunAt.Before(time.Now().Add(time.Second)) { + nextRunAt = time.Now().Add(time.Second) + } return scanGatewayTask(s.pool.QueryRow(ctx, ` UPDATE gateway_tasks SET status = 'queued', @@ -820,6 +827,29 @@ ORDER BY COALESCE(attempt_no, 0), created_at`, taskID) } func (s *Store) AppendTaskAttemptTrace(ctx context.Context, taskID string, attemptNo int, entry map[string]any) error { + if strings.TrimSpace(taskID) == "" || attemptNo <= 0 || len(entry) == 0 { + return nil + } + encoded, err := json.Marshal(entry) + if err != nil { + return err + } + result, err := s.pool.Exec(ctx, ` +UPDATE gateway_task_attempts +SET metrics = jsonb_set( + COALESCE(metrics, '{}'::jsonb), + '{trace}', + COALESCE(metrics->'trace', '[]'::jsonb) || jsonb_build_array($3::jsonb), + true + ) +WHERE task_id = $1::uuid + AND attempt_no = $2::int`, taskID, attemptNo, encoded) + if err != nil { + return err + } + if result.RowsAffected() == 0 { + return pgx.ErrNoRows + } return nil } diff --git a/apps/web/src/pages/admin/RuntimePoliciesPanel.tsx b/apps/web/src/pages/admin/RuntimePoliciesPanel.tsx index d90ac79..e4af25b 100644 --- a/apps/web/src/pages/admin/RuntimePoliciesPanel.tsx +++ b/apps/web/src/pages/admin/RuntimePoliciesPanel.tsx @@ -1,13 +1,18 @@ import { useEffect, useState, type FormEvent } from 'react'; import { Select as AntSelect } from 'antd'; import { Gauge, Pencil, Plus, RotateCcw, Route, Save, ShieldCheck, Trash2 } from 'lucide-react'; -import type { GatewayRunnerPolicy, GatewayRunnerPolicyUpsertRequest, RuntimePolicySet, RuntimePolicySetUpsertRequest } from '@easyai-ai-gateway/contracts'; +import type { + GatewayRunnerPolicy, + GatewayRunnerPolicyUpsertRequest, + RunnerFailoverAction, + RunnerFailoverTarget, + RuntimePolicySet, + RuntimePolicySetUpsertRequest, +} from '@easyai-ai-gateway/contracts'; import { Badge, Button, Card, CardContent, CardHeader, CardTitle, ConfirmDialog, FormDialog, Input, Label, Select, Tabs, Textarea } from '../../components/ui'; import type { LoadState } from '../../types'; type RuntimePanelTab = 'model' | 'runner'; -type RunnerFailoverAction = 'next' | 'cooldown_and_next' | 'demote_and_next' | 'disable_and_next' | 'stop'; -type RunnerFailoverTarget = 'platform' | 'model'; type RunnerActionRule = { id: string; enabled: boolean; @@ -159,6 +164,9 @@ export function RuntimePoliciesPanel(props: { event.preventDefault(); setLocalError(''); try { + if (runnerForm.actionRules.some((rule) => !hasRunnerActionRuleCondition(rule))) { + throw new Error('每条故障动作规则至少需要一个匹配条件'); + } await props.onSaveRunnerPolicy(runnerFormToPayload(runnerForm)); } catch (err) { setLocalError(err instanceof Error ? err.message : '全局调度策略保存失败'); @@ -171,7 +179,7 @@ export function RuntimePoliciesPanel(props: {
运行策略 -

模型运行策略维护限流、平台内重试、自动禁用和降级；全局调度策略控制平台间切换、硬拒绝和失败平台优先级降级。

+

模型运行策略维护限流和平台内重试；旧自动禁用、冷却字段仅兼容读取，全局 action rule 统一控制平台间动作。

{activeTab === 'model' &&