From 8362d6d27aed83422234ed018f18efef38471262 Mon Sep 17 00:00:00 2001 From: wangbo Date: Mon, 3 Aug 2026 17:42:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(routing):=20=E5=AE=8C=E5=96=84=E5=B9=B3?= =?UTF-8?q?=E5=8F=B0=E6=BB=A1=E8=BD=BD=E9=81=BF=E8=AE=A9=E4=B8=8E=E6=95=85?= =?UTF-8?q?=E9=9A=9C=E8=BD=AE=E8=BD=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为未配置并发上限的平台增加基于运行和等待任务数的软负载，并按非满载、有效优先级、缓存亲和与负载稳定排序。\n\n平台模型 RPM、TPM 和并发额度竞争失败时只轮转候选，不触发冷却、禁用或降级；用户组额度保持不可绕过。补齐异步冷却排队恢复、满载原因、选择原因与低基数指标。\n\n验证：go test ./...；go vet ./...；PostgreSQL 原子额度、准入队列、Worker 容量回收及故障策略 HTTP acceptance。 --- ...policy_http_acceptance_integration_test.go | 6 +- apps/api/internal/httpapi/handlers.go | 12 +- apps/api/internal/runner/admission.go | 31 ++-- apps/api/internal/runner/limits.go | 36 +++-- apps/api/internal/runner/pricing.go | 3 + apps/api/internal/runner/recording.go | 40 ++--- apps/api/internal/runner/retry_decision.go | 9 ++ .../internal/runner/retry_decision_test.go | 51 ++++--- apps/api/internal/runner/service.go | 137 ++++++++++------- apps/api/internal/securityevents/metrics.go | 35 +++++ .../internal/securityevents/metrics_test.go | 12 ++ apps/api/internal/store/candidates.go | 68 ++++++++- apps/api/internal/store/candidates_test.go | 138 ++++++++++++++++++ apps/api/internal/store/runtime_policies.go | 38 +++-- apps/api/internal/store/runtime_types.go | 4 + 15 files changed, 468 insertions(+), 152 deletions(-) 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 index 3dca1a6..1c74048 100644 --- a/apps/api/internal/httpapi/failure_policy_http_acceptance_integration_test.go +++ b/apps/api/internal/httpapi/failure_policy_http_acceptance_integration_test.go @@ -774,7 +774,6 @@ func (f *failurePolicyAcceptanceFixture) testAllCoolingSynchronousContract(t *te 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) @@ -801,9 +800,8 @@ func (f *failurePolicyAcceptanceFixture) testAllCoolingSynchronousContract(t *te 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) + if stubA.calls.Load() != 1 || stubB.calls.Load() != 1 { + t.Fatalf("candidate selection cooldown must not create provider attempts: calls=%d/%d", stubA.calls.Load(), stubB.calls.Load()) } } diff --git a/apps/api/internal/httpapi/handlers.go b/apps/api/internal/httpapi/handlers.go index 453989f..0bfaec3 100644 --- a/apps/api/internal/httpapi/handlers.go +++ b/apps/api/internal/httpapi/handlers.go @@ -1202,14 +1202,18 @@ func (s *Server) createTask(kind string, compatible bool) http.Handler { return } if err := s.runner.ValidateModelAccess(r.Context(), kind, model, prepared.Body, user); err != nil { - if errors.Is(err, store.ErrNoModelCandidate) { + if responsePlan.asyncMode && store.ModelCandidateRetryAfter(err) > 0 { + // Persist recoverable asynchronous requests so the shared queue can + // wake them when the earliest candidate cooldown expires. + } else if errors.Is(err, store.ErrNoModelCandidate) { applyRunErrorHeaders(w, err) writeTaskError(statusFromRunError(err), runErrorMessage(err), runErrorDetails(err), runErrorCode(err)) return + } else { + s.logger.Error("validate model access failed", "kind", kind, "error_category", "model_access_validation_failed", "error", err) + writeTaskError(http.StatusInternalServerError, "validate model access failed", nil, "model_access_validation_failed") + return } - s.logger.Error("validate model access failed", "kind", kind, "error_category", "model_access_validation_failed", "error", err) - writeTaskError(http.StatusInternalServerError, "validate model access failed", nil, "model_access_validation_failed") - return } runMode, err := s.admittedTaskRunMode(admission, prepared.Body) diff --git a/apps/api/internal/runner/admission.go b/apps/api/internal/runner/admission.go index b876e19..5b63cd0 100644 --- a/apps/api/internal/runner/admission.go +++ b/apps/api/internal/runner/admission.go @@ -123,11 +123,12 @@ func (s *Service) buildTaskAdmissionPlanForCurrentBinding( candidates, _ = pinCandidatesToTaskAdmission(candidates, admission) var candidateRateLimitErr error for _, candidate := range candidates { - available, availabilityErr := s.store.RuntimeCandidateAvailable(ctx, candidate.PlatformID, candidate.PlatformModelID) + available, unavailableReason, availabilityErr := s.store.RuntimeCandidateAvailability(ctx, candidate.PlatformID, candidate.PlatformModelID) if availabilityErr != nil { return taskAdmissionPlan{}, availabilityErr } if !available { + s.observeCandidateRouting(unavailableReason) continue } scopes, groupID, scopeErr := s.taskAdmissionScopes(ctx, task, user, candidate) @@ -141,13 +142,6 @@ func (s *Service) buildTaskAdmissionPlanForCurrentBinding( break } } - if !hasConcurrentLimit { - return taskAdmissionPlan{ - Candidate: candidate, - Body: body, - ModelType: modelType, - }, nil - } reservations := acceptanceInfrastructureReservations( task, s.rateLimitReservations(ctx, user, candidate, body), @@ -162,6 +156,13 @@ func (s *Service) buildTaskAdmissionPlanForCurrentBinding( } return taskAdmissionPlan{}, err } + if !hasConcurrentLimit { + return taskAdmissionPlan{ + Candidate: candidate, + Body: body, + ModelType: modelType, + }, nil + } return taskAdmissionPlan{ Candidate: candidate, Body: body, @@ -172,6 +173,7 @@ func (s *Service) buildTaskAdmissionPlanForCurrentBinding( }, nil } if candidateRateLimitErr != nil { + s.observeCandidateRouting("all_full_queued") return taskAdmissionPlan{}, candidateRateLimitErr } return taskAdmissionPlan{}, store.ErrNoModelCandidate @@ -475,12 +477,6 @@ func (s *Service) ensureCandidateAdmission( break } } - if !hasConcurrentLimit { - if err := s.store.DeleteTaskAdmission(context.WithoutCancel(ctx), task.ID); err != nil && !errors.Is(err, pgx.ErrNoRows) { - return store.TaskAdmissionResult{}, false, err - } - return store.TaskAdmissionResult{}, false, nil - } reservations := acceptanceInfrastructureReservations( task, s.rateLimitReservations(ctx, user, candidate, body), @@ -488,6 +484,12 @@ func (s *Service) ensureCandidateAdmission( if err := s.store.CheckRateLimits(ctx, reservations); err != nil { return store.TaskAdmissionResult{}, true, err } + if !hasConcurrentLimit { + if err := s.store.DeleteTaskAdmission(context.WithoutCancel(ctx), task.ID); err != nil && !errors.Is(err, pgx.ErrNoRows) { + return store.TaskAdmissionResult{}, false, err + } + return store.TaskAdmissionResult{}, false, nil + } plan := taskAdmissionPlan{ Candidate: candidate, Body: body, @@ -809,6 +811,7 @@ func (s *Service) dispatchWaitingAsyncTasks(ctx context.Context, admissions []st return false, outcome.Err } if !outcome.Result.Admitted { + s.observeCandidateRouting("quota_race_rotated") platformModelID := outcome.Result.Admission.PlatformModelID if platformModelID == "" { for _, input := range inputs { diff --git a/apps/api/internal/runner/limits.go b/apps/api/internal/runner/limits.go index bb113d4..8f49455 100644 --- a/apps/api/internal/runner/limits.go +++ b/apps/api/internal/runner/limits.go @@ -50,21 +50,18 @@ func isLocalRateLimitError(err error) bool { return errors.As(err, &limitErr) } +func platformModelRateLimitError(err error) (*store.RateLimitExceededError, bool) { + var limitErr *store.RateLimitExceededError + if !errors.As(err, &limitErr) || limitErr.ScopeType != "platform_model" { + return nil, false + } + return limitErr, true +} + func (s *Service) rateLimitReservations(ctx context.Context, user *auth.User, candidate store.RuntimeModelCandidate, body map[string]any) []store.RateLimitReservation { out := make([]store.RateLimitReservation, 0) - out = append(out, reservationsFromPolicy( - "platform_model", - candidate.PlatformModelID, - firstNonEmptyString(candidate.DisplayName, candidate.ModelAlias, candidate.ModelName), - map[string]any{ - "platformId": candidate.PlatformID, - "platformName": candidate.PlatformName, - "modelAlias": candidate.ModelAlias, - "modelName": candidate.ModelName, - }, - effectiveRateLimitPolicy(candidate), - body, - )...) + // Cross-platform scopes must be evaluated first so a user-group limit can + // never be bypassed by rotating to another platform model. if group, err := s.store.ResolveUserGroupPolicy(ctx, user); err == nil && group.ID != "" { out = append(out, reservationsFromPolicy( "user_group", @@ -78,6 +75,19 @@ func (s *Service) rateLimitReservations(ctx context.Context, user *auth.User, ca body, )...) } + out = append(out, reservationsFromPolicy( + "platform_model", + candidate.PlatformModelID, + firstNonEmptyString(candidate.DisplayName, candidate.ModelAlias, candidate.ModelName), + map[string]any{ + "platformId": candidate.PlatformID, + "platformName": candidate.PlatformName, + "modelAlias": candidate.ModelAlias, + "modelName": candidate.ModelName, + }, + effectiveRateLimitPolicy(candidate), + body, + )...) return out } diff --git a/apps/api/internal/runner/pricing.go b/apps/api/internal/runner/pricing.go index 9641d5a..d7b463c 100644 --- a/apps/api/internal/runner/pricing.go +++ b/apps/api/internal/runner/pricing.go @@ -50,6 +50,9 @@ func (s *Service) Estimate(ctx context.Context, kind string, model string, body // using stale routing or capacity state. func (s *Service) ValidateModelAccess(ctx context.Context, kind string, model string, body map[string]any, user *auth.User) error { _, _, err := s.candidatesForRequest(ctx, kind, model, body, user) + if store.ModelCandidateRetryAfter(err) > 0 { + s.observeCandidateRouting("cooldown_skipped") + } return err } diff --git a/apps/api/internal/runner/recording.go b/apps/api/internal/runner/recording.go index 6cc4905..8507ac3 100644 --- a/apps/api/internal/runner/recording.go +++ b/apps/api/internal/runner/recording.go @@ -126,6 +126,8 @@ func attemptMetrics(candidate store.RuntimeModelCandidate, attemptNo int, simula "currentPriority": candidate.PlatformPriority, "loadRatio": candidate.LoadRatio, "loadAvoided": candidate.LoadAvoided, + "fullReasons": candidate.FullReasons, + "selectionReason": candidate.SelectionReason, "simulated": simulated, } if candidate.ResponseProtocol != "" { @@ -138,25 +140,25 @@ func attemptMetrics(candidate store.RuntimeModelCandidate, attemptNo int, simula metrics["responseConverted"] = true } } - if candidate.LoadLimited { - metrics["loadMetrics"] = map[string]any{ - "rpm": map[string]any{ - "current": candidate.LoadMetrics.RPMCurrent, - "limit": candidate.LoadMetrics.RPMLimit, - "ratio": candidate.LoadMetrics.RPMRatio, - }, - "tpm": map[string]any{ - "current": candidate.LoadMetrics.TPMCurrent, - "limit": candidate.LoadMetrics.TPMLimit, - "ratio": candidate.LoadMetrics.TPMRatio, - }, - "concurrent": map[string]any{ - "current": candidate.LoadMetrics.ConcurrentCurrent, - "limit": candidate.LoadMetrics.ConcurrentLimit, - "ratio": candidate.LoadMetrics.ConcurrentRatio, - }, - "queued": candidate.LoadMetrics.QueuedCount, - } + metrics["loadMetrics"] = map[string]any{ + "rpm": map[string]any{ + "current": candidate.LoadMetrics.RPMCurrent, + "limit": candidate.LoadMetrics.RPMLimit, + "ratio": candidate.LoadMetrics.RPMRatio, + }, + "tpm": map[string]any{ + "current": candidate.LoadMetrics.TPMCurrent, + "limit": candidate.LoadMetrics.TPMLimit, + "ratio": candidate.LoadMetrics.TPMRatio, + }, + "concurrent": map[string]any{ + "current": candidate.LoadMetrics.ConcurrentCurrent, + "limit": candidate.LoadMetrics.ConcurrentLimit, + "ratio": candidate.LoadMetrics.ConcurrentRatio, + "softCurrent": candidate.LoadMetrics.SoftCurrent, + "softRatio": candidate.LoadMetrics.SoftRatio, + }, + "queued": candidate.LoadMetrics.QueuedCount, } if candidate.CacheAffinity.Key != "" { metrics["cacheAffinityKey"] = candidate.CacheAffinity.Key diff --git a/apps/api/internal/runner/retry_decision.go b/apps/api/internal/runner/retry_decision.go index c854dd7..9968d06 100644 --- a/apps/api/internal/runner/retry_decision.go +++ b/apps/api/internal/runner/retry_decision.go @@ -160,6 +160,15 @@ func resolveCandidateFailure(input resolveCandidateFailureInput) failureDecision } } if errors.Is(input.Err, store.ErrRateLimited) { + if _, platformLimited := platformModelRateLimitError(input.Err); platformLimited && input.HasNextCandidate { + return failureDecision{ + Route: "next", + Effect: "none", + Reason: "quota_race_rotated", + Match: policyRuleMatch{Source: "gateway_rate_limits", Policy: "rateLimitPolicy", Rule: "platformModelCapacity", Value: "saturated"}, + Info: info, + } + } route := "stop" if input.Async && store.RateLimitRetryable(input.Err) { route = "requeue" diff --git a/apps/api/internal/runner/retry_decision_test.go b/apps/api/internal/runner/retry_decision_test.go index 0788a02..ac59568 100644 --- a/apps/api/internal/runner/retry_decision_test.go +++ b/apps/api/internal/runner/retry_decision_test.go @@ -80,24 +80,6 @@ func TestFailoverTimeBudgetExceeded(t *testing.T) { } } -func TestLoadAvoidanceFallbackContinuesToAvoidedCandidate(t *testing.T) { - candidates := []store.RuntimeModelCandidate{ - {PlatformID: "available-candidate"}, - {PlatformID: "avoided-full-candidate", LoadAvoided: true}, - } - - if !hasLoadAvoidanceFallback(candidates, 0, 99) { - t.Fatal("expected non-avoided candidate to fall back to later avoided candidate") - } - if hasLoadAvoidanceFallback(candidates, 1, 99) { - t.Fatal("avoided candidate should not force another load-avoidance fallback") - } - decision := loadAvoidanceFallbackDecision(&clients.ClientError{Code: "bad_request", StatusCode: 400, Retryable: false}) - if !decision.Retry || decision.Reason != "load_avoidance_fallback" || decision.Action != "next" { - t.Fatalf("expected active load avoidance fallback to force next candidate, got %+v", decision) - } -} - func TestFailoverHardStopBeatsModelOverride(t *testing.T) { runnerPolicy := store.RunnerPolicy{ Status: "active", @@ -148,7 +130,7 @@ func TestFailoverPolicyAllowsModelOverride(t *testing.T) { func TestLocalRateLimitWaitsInQueueWithoutRetryOrFailover(t *testing.T) { err := &localRateLimitError{ clientErr: &clients.ClientError{Code: "rate_limit", Message: "local capacity exceeded", Retryable: true}, - cause: &store.RateLimitExceededError{Metric: "concurrent", Retryable: true}, + cause: &store.RateLimitExceededError{ScopeType: "user_group", Metric: "concurrent", Retryable: true}, } retryDecision := retryDecisionForCandidate(store.RuntimeModelCandidate{}, err) if retryDecision.Retry || retryDecision.Reason != "local_rate_limit_wait_queue" { @@ -164,6 +146,37 @@ func TestLocalRateLimitWaitsInQueueWithoutRetryOrFailover(t *testing.T) { } } +func TestPlatformModelRateLimitRaceRotatesWithoutFailureEffect(t *testing.T) { + err := &localRateLimitError{ + clientErr: &clients.ClientError{Code: "rate_limit", Message: "platform capacity raced", Retryable: true}, + cause: &store.RateLimitExceededError{ + ScopeType: "platform_model", + Metric: "concurrent", + Retryable: true, + RetryAfter: time.Second, + }, + } + + decision := resolveCandidateFailure(resolveCandidateFailureInput{ + RunnerPolicy: store.RunnerPolicy{Status: "active"}, + Err: err, + HasNextCandidate: true, + Async: true, + }) + if decision.Route != "next" || decision.Effect != "none" || decision.Reason != "quota_race_rotated" { + t.Fatalf("platform quota race should rotate without mutation: %+v", decision) + } + + last := resolveCandidateFailure(resolveCandidateFailureInput{ + RunnerPolicy: store.RunnerPolicy{Status: "active"}, + Err: err, + Async: true, + }) + if last.Route != "requeue" || last.Effect != "none" { + t.Fatalf("last saturated platform should requeue: %+v", last) + } +} + func TestProviderAuthErrorsFailOverInsteadOfHardStop(t *testing.T) { runnerPolicy := store.RunnerPolicy{ Status: "active", diff --git a/apps/api/internal/runner/service.go b/apps/api/internal/runner/service.go index fb25b3a..64f40dd 100644 --- a/apps/api/internal/runner/service.go +++ b/apps/api/internal/runner/service.go @@ -382,6 +382,7 @@ func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask, err = responseChainUnavailableError() } if task.AsyncMode && store.ModelCandidateRetryAfter(err) > 0 { + s.observeCandidateRouting("cooldown_skipped") queued, delay, queueErr := s.requeueModelCoolingTask(ctx, task, err) if queueErr != nil { return Result{}, queueErr @@ -389,6 +390,7 @@ func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask, return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay} } if store.ModelCandidateRetryAfter(err) > 0 { + s.observeCandidateRouting("cooldown_skipped") failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err) if finishErr != nil { return Result{}, finishErr @@ -481,6 +483,11 @@ func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask, return Result{Task: failed, Output: failed.Result}, err } } + for _, candidate := range candidates { + if candidate.LoadAvoided { + s.observeCandidateRouting("full_avoided") + } + } var asyncAdmission *store.TaskAdmission if distributedAdmission && task.AsyncMode { asyncAdmission, err = s.loadAsyncTaskAdmission(ctx, task) @@ -614,25 +621,35 @@ func (s *Service) executeWithToken(ctx context.Context, task store.GatewayTask, break } } - if hasConcurrentLimit { - reservations := acceptanceInfrastructureReservations( - task, - s.rateLimitReservations(ctx, user, candidates[0], body), - ) - if err := s.store.CheckRateLimits(ctx, reservations); err != nil { - if task.AsyncMode && errors.Is(err, store.ErrRateLimited) && store.RateLimitRetryable(err) { - queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, err, candidates[0]) - if queueErr != nil { - return Result{}, queueErr + deferInitialAdmission := false + reservations := acceptanceInfrastructureReservations( + task, + s.rateLimitReservations(ctx, user, candidates[0], body), + ) + if err := s.store.CheckRateLimits(ctx, reservations); err != nil { + if _, platformLimited := platformModelRateLimitError(err); platformLimited && len(candidates) > 1 { + if task.AsyncMode { + if deleteErr := s.store.DeleteTaskAdmission(context.WithoutCancel(ctx), task.ID); deleteErr != nil { + return Result{}, deleteErr } - return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay} + asyncAdmission = nil } + deferInitialAdmission = true + } else if task.AsyncMode && errors.Is(err, store.ErrRateLimited) && store.RateLimitRetryable(err) { + queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, err, candidates[0]) + if queueErr != nil { + return Result{}, queueErr + } + return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay} + } else { failed, finishErr := s.failTask(ctx, task.ID, task.ExecutionToken, clients.ErrorCode(err), err.Error(), task.RunMode == "simulation", err) if finishErr != nil { return Result{}, finishErr } return Result{Task: failed, Output: failed.Result}, err } + } + if hasConcurrentLimit && !deferInitialAdmission { plan := taskAdmissionPlan{ Candidate: candidates[0], Body: body, @@ -799,13 +816,17 @@ candidatesLoop: if platformsVisited >= maxPlatforms { break } - available, availabilityErr := s.store.RuntimeCandidateAvailable(ctx, candidate.PlatformID, candidate.PlatformModelID) + available, unavailableReason, availabilityErr := s.store.RuntimeCandidateAvailability(ctx, candidate.PlatformID, candidate.PlatformModelID) if availabilityErr != nil { return Result{}, availabilityErr } if !available { + s.observeCandidateRouting(unavailableReason) continue } + if candidate.SelectionReason == "normal_rotation" || candidate.SelectionReason == "" { + s.observeCandidateRouting("normal_rotation") + } if distributedAdmission && candidate.PlatformModelID != admittedPlatformModelID { admissionResult, candidateLimited, admissionErr := s.ensureCandidateAdmission(ctx, task, user, body, candidate) if admissionErr != nil { @@ -814,9 +835,17 @@ candidatesLoop: if errors.Is(admissionErr, store.ErrQueueTimeout) { break } - if candidateLimited && errors.Is(admissionErr, store.ErrRateLimited) { + if _, platformLimited := platformModelRateLimitError(admissionErr); platformLimited { + s.observeCandidateRouting("quota_race_rotated") continue } + if task.AsyncMode && errors.Is(admissionErr, store.ErrRateLimited) && store.RateLimitRetryable(admissionErr) { + queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, admissionErr, candidate) + if queueErr != nil { + return Result{}, queueErr + } + return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay} + } return Result{}, admissionErr } if candidateLimited { @@ -1062,6 +1091,9 @@ candidatesLoop: DownstreamStarted: downstreamStarted.Load(), }) if candidateDecision.Route == "requeue" { + if _, platformLimited := platformModelRateLimitError(err); platformLimited { + s.observeCandidateRouting("all_full_queued") + } queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, err, candidate) if queueErr != nil { return Result{}, queueErr @@ -1069,21 +1101,35 @@ candidatesLoop: return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay} } attemptNo = s.recordFailedAttempt(ctx, failedAttemptRecord{ - Task: task, - Body: candidateBody, - Candidate: &candidate, - AttemptNo: nextAttemptNo, - Code: clients.ErrorCode(err), - Cause: err, - Simulated: isSimulation(task, candidate), - Scope: "rate_limit", - Reason: "local_rate_limit_blocked", - ExtraMetrics: []map[string]any{parameterPreprocessingMetrics(preprocessing.Log)}, - ModelType: candidate.ModelType, + Task: task, + Body: candidateBody, + Candidate: &candidate, + AttemptNo: nextAttemptNo, + Code: clients.ErrorCode(err), + Cause: err, + Simulated: isSimulation(task, candidate), + Scope: "rate_limit", + Reason: "local_rate_limit_blocked", + ExtraMetrics: []map[string]any{ + parameterPreprocessingMetrics(preprocessing.Log), + {"selectionReason": candidateDecision.Reason}, + }, + ModelType: candidate.ModelType, }) candidateDecisionAttempt = attemptNo candidateClientAttempt = clientAttempt s.recordAttemptTrace(ctx, task.ID, attemptNo, failureDecisionTraceEntry(candidateDecision, candidate, clientAttempt, clientAttempts, false, "")) + if candidateDecision.Route == "next" { + s.observeCandidateRouting("quota_race_rotated") + if admittedPlatformModelID == candidate.PlatformModelID { + if deleteErr := s.store.DeleteTaskAdmission(context.WithoutCancel(ctx), task.ID); deleteErr != nil { + return Result{}, deleteErr + } + admittedPlatformModelID = "" + admittedLeases = nil + } + break + } break candidatesLoop } attemptNo = nextAttemptNo @@ -1179,6 +1225,9 @@ candidatesLoop: return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: 0} } if task.AsyncMode && errors.Is(lastErr, store.ErrRateLimited) && store.RateLimitRetryable(lastErr) { + if _, platformLimited := platformModelRateLimitError(lastErr); platformLimited { + s.observeCandidateRouting("all_full_queued") + } queued, delay, queueErr := s.requeueRateLimitedTask(ctx, task, lastErr, lastCandidate) if queueErr != nil { return Result{}, queueErr @@ -1754,6 +1803,15 @@ func (s *Service) observeProviderQuotaWait(metric string) { } } +func (s *Service) observeCandidateRouting(reason string) { + observer, ok := s.billingMetrics.(interface { + ObserveCandidateRouting(string) + }) + if ok { + observer.ObserveCandidateRouting(reason) + } +} + func minimalRemoteTaskCheckpoint(provider string, specType string, payload map[string]any) map[string]any { const maxBytes = 8192 provider = strings.ToLower(strings.TrimSpace(provider)) @@ -2545,37 +2603,6 @@ func failoverTimeBudgetExceeded(start time.Time, maxDuration time.Duration) bool return maxDuration > 0 && time.Since(start) >= maxDuration } -func hasLoadAvoidanceFallback(candidates []store.RuntimeModelCandidate, index int, maxPlatforms int) bool { - if index < 0 || index >= len(candidates) || candidates[index].LoadAvoided { - return false - } - limit := len(candidates) - if maxPlatforms > 0 && maxPlatforms < limit { - limit = maxPlatforms - } - for next := index + 1; next < limit; next++ { - if candidates[next].LoadAvoided { - return true - } - } - return false -} - -func loadAvoidanceFallbackDecision(err error) failoverDecision { - return failoverDecision{ - Retry: true, - Action: "next", - Reason: "load_avoidance_fallback", - Match: policyRuleMatch{ - Source: "runtime_client_load", - Policy: "loadAvoidance", - Rule: "fallback", - Value: "loadRatio>=1", - }, - Info: failureInfoFromError(err), - } -} - func normalizeRequest(kind string, body map[string]any) map[string]any { out := cloneMap(body) return out diff --git a/apps/api/internal/securityevents/metrics.go b/apps/api/internal/securityevents/metrics.go index 0c4bc5b..8fb7e68 100644 --- a/apps/api/internal/securityevents/metrics.go +++ b/apps/api/internal/securityevents/metrics.go @@ -79,6 +79,13 @@ type Metrics struct { taskAdmissionCancelled atomic.Uint64 taskAdmissionExpired atomic.Uint64 taskAdmissionMigrated atomic.Uint64 + candidateNormalRotation atomic.Uint64 + candidateFullAvoided atomic.Uint64 + candidateQuotaRaceRotated atomic.Uint64 + candidateCooldownSkipped atomic.Uint64 + candidateDisabledSkipped atomic.Uint64 + candidateAllFullQueued atomic.Uint64 + candidateRoutingOther atomic.Uint64 taskAdmissionWaitBuckets [11]atomic.Uint64 taskAdmissionWaitMicros atomic.Uint64 } @@ -239,6 +246,25 @@ func (m *Metrics) ObserveTaskAdmission(event string) { } } +func (m *Metrics) ObserveCandidateRouting(reason string) { + switch reason { + case "normal_rotation": + m.candidateNormalRotation.Add(1) + case "full_avoided": + m.candidateFullAvoided.Add(1) + case "quota_race_rotated": + m.candidateQuotaRaceRotated.Add(1) + case "cooldown_skipped": + m.candidateCooldownSkipped.Add(1) + case "disabled_skipped": + m.candidateDisabledSkipped.Add(1) + case "all_full_queued": + m.candidateAllFullQueued.Add(1) + default: + m.candidateRoutingOther.Add(1) + } +} + func (m *Metrics) ObserveTaskAdmissionWait(wait time.Duration) { if wait < 0 { wait = 0 @@ -436,6 +462,15 @@ func (m *Metrics) Handler(provider MetricsSnapshotProvider, issuer, audience str {"concurrent", m.providerQuotaWaitConcurrent.Load()}, {"other", m.providerQuotaWaitOther.Load()}, }) + outcomeCounters(w, "easyai_gateway_candidate_routing_total", "Candidate routing decisions by bounded reason.", []outcomeValue{ + {"normal_rotation", m.candidateNormalRotation.Load()}, + {"full_avoided", m.candidateFullAvoided.Load()}, + {"quota_race_rotated", m.candidateQuotaRaceRotated.Load()}, + {"cooldown_skipped", m.candidateCooldownSkipped.Load()}, + {"disabled_skipped", m.candidateDisabledSkipped.Load()}, + {"all_full_queued", m.candidateAllFullQueued.Load()}, + {"other", m.candidateRoutingOther.Load()}, + }) platformModelRateLimitUtilizationGauges(w, modelRateLimits) plainGauge(w, "easyai_gateway_postgres_pool_max_connections", "Maximum PostgreSQL connections in this process pool.", int64(postgresPool.MaxConnections)) plainGauge(w, "easyai_gateway_postgres_pool_total_connections", "Current PostgreSQL connections in this process pool.", int64(postgresPool.TotalConnections)) diff --git a/apps/api/internal/securityevents/metrics_test.go b/apps/api/internal/securityevents/metrics_test.go index 8d5a324..8fc0221 100644 --- a/apps/api/internal/securityevents/metrics_test.go +++ b/apps/api/internal/securityevents/metrics_test.go @@ -46,6 +46,12 @@ func TestMetricsExposeBoundedOutcomesAndState(t *testing.T) { metrics.SetWorkerLoad(12, 3, 8, 2, 5, 1, "busy") metrics.ObserveProviderQuotaWait("rpm") metrics.ObserveProviderQuotaWait("tpm_total") + metrics.ObserveCandidateRouting("normal_rotation") + metrics.ObserveCandidateRouting("full_avoided") + metrics.ObserveCandidateRouting("quota_race_rotated") + metrics.ObserveCandidateRouting("cooldown_skipped") + metrics.ObserveCandidateRouting("disabled_skipped") + metrics.ObserveCandidateRouting("all_full_queued") metrics.ObserveAsyncWorkerResize("success") metrics.ObserveConcurrencyLeaseRenewal("success") metrics.ObserveConcurrencyLeaseRenewal("lost") @@ -86,6 +92,12 @@ func TestMetricsExposeBoundedOutcomesAndState(t *testing.T) { `easyai_gateway_worker_pressure_state 1`, `easyai_gateway_provider_quota_waits_total{outcome="rpm"} 1`, `easyai_gateway_provider_quota_waits_total{outcome="tpm"} 1`, + `easyai_gateway_candidate_routing_total{outcome="normal_rotation"} 1`, + `easyai_gateway_candidate_routing_total{outcome="full_avoided"} 1`, + `easyai_gateway_candidate_routing_total{outcome="quota_race_rotated"} 1`, + `easyai_gateway_candidate_routing_total{outcome="cooldown_skipped"} 1`, + `easyai_gateway_candidate_routing_total{outcome="disabled_skipped"} 1`, + `easyai_gateway_candidate_routing_total{outcome="all_full_queued"} 1`, `easyai_gateway_platform_model_rate_limit_utilization{platform_model_id="platform-model-1",metric="concurrent"} 0.750000`, `easyai_gateway_async_worker_resizes_total{outcome="success"} 1`, `easyai_gateway_concurrency_lease_renewals_total{outcome="success"} 1`, diff --git a/apps/api/internal/store/candidates.go b/apps/api/internal/store/candidates.go index 085b139..170ce9e 100644 --- a/apps/api/internal/store/candidates.go +++ b/apps/api/internal/store/candidates.go @@ -699,7 +699,14 @@ func applyRuntimeCandidateLoad(candidate *RuntimeModelCandidate, input runtimeCa concurrentLimit := rateLimitForMetric(input.Policy, "concurrent") rpmCurrent := input.RPMUsed + input.RPMReserved tpmCurrent := input.TPMUsed + input.TPMReserved - concurrentCurrent := input.ConcurrentActive + input.QueuedWaiting + activeCurrent := maxFloat(input.ConcurrentActive, input.StateRunningCount) + waitingCurrent := maxFloat(input.QueuedWaiting, input.StateWaitingCount) + concurrentCurrent := activeCurrent + waitingCurrent + softRatio := unboundedLoadRatio(concurrentCurrent) + concurrentRatio := softRatio + if concurrentLimit > 0 { + concurrentRatio = ratioIfLimited(concurrentCurrent, concurrentLimit) + } metrics := RuntimeCandidateLoadMetrics{ RPMCurrent: rpmCurrent, RPMLimit: rpmLimit, @@ -709,15 +716,20 @@ func applyRuntimeCandidateLoad(candidate *RuntimeModelCandidate, input runtimeCa TPMRatio: ratioIfLimited(tpmCurrent, tpmLimitValue), ConcurrentCurrent: concurrentCurrent, ConcurrentLimit: concurrentLimit, - ConcurrentRatio: ratioIfLimited(concurrentCurrent, concurrentLimit), - QueuedCount: input.QueuedWaiting, + ConcurrentRatio: concurrentRatio, + SoftCurrent: concurrentCurrent, + SoftRatio: softRatio, + QueuedCount: waitingCurrent, StateRunningCount: input.StateRunningCount, StateWaitingCount: input.StateWaitingCount, StateLimiterRatio: input.StateLimiterRatio, } + candidate.RunningCount = activeCurrent + candidate.WaitingCount = waitingCurrent candidate.LoadMetrics = metrics candidate.LoadLimited = rpmLimit > 0 || tpmLimitValue > 0 || concurrentLimit > 0 candidate.LoadRatio = maxFloat(metrics.RPMRatio, metrics.TPMRatio, metrics.ConcurrentRatio) + candidate.FullReasons = runtimeCandidateFullReasons(*candidate) } func ratioIfLimited(current float64, limit float64) float64 { @@ -727,11 +739,20 @@ func ratioIfLimited(current float64, limit float64) float64 { return current / limit } +func unboundedLoadRatio(current float64) float64 { + if current <= 0 { + return 0 + } + return current / (current + 1) +} + func sortRuntimeModelCandidates(items []RuntimeModelCandidate) { hasFull := false hasNonFull := false for index := range items { items[index].LoadAvoided = false + items[index].SelectionReason = "normal_rotation" + items[index].FullReasons = runtimeCandidateFullReasons(items[index]) if !items[index].CacheAffinity.Applied && items[index].CacheAffinity.AdjustedPriority == 0 { items[index].CacheAffinity.AdjustedPriority = float64(items[index].PlatformPriority) } @@ -744,6 +765,13 @@ func sortRuntimeModelCandidates(items []RuntimeModelCandidate) { if hasFull && hasNonFull { for index := range items { items[index].LoadAvoided = runtimeCandidateFull(items[index]) + if items[index].LoadAvoided { + items[index].SelectionReason = "full_avoided" + } + } + } else if hasFull { + for index := range items { + items[index].SelectionReason = "all_full_fallback" } } sort.SliceStable(items, func(i, j int) bool { @@ -778,16 +806,16 @@ func sortRuntimeModelCandidates(items []RuntimeModelCandidate) { if items[i].LoadRatio != items[j].LoadRatio { return items[i].LoadRatio < items[j].LoadRatio } - if items[i].RunningCount != items[j].RunningCount { - return items[i].RunningCount < items[j].RunningCount - } if items[i].WaitingCount != items[j].WaitingCount { return items[i].WaitingCount < items[j].WaitingCount } + if items[i].RunningCount != items[j].RunningCount { + return items[i].RunningCount < items[j].RunningCount + } if items[i].LastAssignedUnix != items[j].LastAssignedUnix { return items[i].LastAssignedUnix < items[j].LastAssignedUnix } - return false + return items[i].PlatformModelID < items[j].PlatformModelID }) appliedCount := 0 for index := range items { @@ -818,7 +846,31 @@ func sortRuntimeModelCandidates(items []RuntimeModelCandidate) { } func runtimeCandidateFull(candidate RuntimeModelCandidate) bool { - return candidate.LoadLimited && candidate.LoadRatio >= 1 + return len(runtimeCandidateFullReasons(candidate)) > 0 +} + +func runtimeCandidateFullReasons(candidate RuntimeModelCandidate) []string { + if len(candidate.FullReasons) > 0 { + return append([]string(nil), candidate.FullReasons...) + } + metrics := candidate.LoadMetrics + reasons := make([]string, 0, 4) + if metrics.ConcurrentLimit > 0 && metrics.ConcurrentRatio >= 1 { + reasons = append(reasons, "concurrent") + } + if metrics.RPMLimit > 0 && metrics.RPMRatio >= 1 { + reasons = append(reasons, "rpm") + } + if metrics.TPMLimit > 0 && metrics.TPMRatio >= 1 { + reasons = append(reasons, "tpm") + } + if metrics.QueuedCount > 0 { + reasons = append(reasons, "waiting") + } + if len(reasons) == 0 && candidate.LoadLimited && candidate.LoadRatio >= 1 { + reasons = append(reasons, "limited") + } + return reasons } func (s *Store) modelCandidateCooldownError(ctx context.Context, model string, modelType string) (error, error) { diff --git a/apps/api/internal/store/candidates_test.go b/apps/api/internal/store/candidates_test.go index c11689f..cbb3f18 100644 --- a/apps/api/internal/store/candidates_test.go +++ b/apps/api/internal/store/candidates_test.go @@ -86,6 +86,144 @@ func TestRuntimeCandidateLoadUsesMaxLimitedMetric(t *testing.T) { } } +func TestRuntimeCandidateLoadUsesSoftConcurrencyWithoutLimit(t *testing.T) { + tests := []struct { + name string + active float64 + waiting float64 + stateRun float64 + stateWait float64 + wantCurrent float64 + wantRatio float64 + }{ + {name: "idle", wantCurrent: 0, wantRatio: 0}, + {name: "one active", active: 1, wantCurrent: 1, wantRatio: 0.5}, + {name: "two from runtime state", stateRun: 2, wantCurrent: 2, wantRatio: 2.0 / 3.0}, + {name: "uses maximum per phase without double counting", active: 1, waiting: 2, stateRun: 3, stateWait: 1, wantCurrent: 5, wantRatio: 5.0 / 6.0}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + candidate := RuntimeModelCandidate{} + applyRuntimeCandidateLoad(&candidate, runtimeCandidateLoadInput{ + ConcurrentActive: test.active, + QueuedWaiting: test.waiting, + StateRunningCount: test.stateRun, + StateWaitingCount: test.stateWait, + }) + if candidate.LoadLimited { + t.Fatal("soft concurrency must not become a hard limit") + } + if candidate.LoadMetrics.SoftCurrent != test.wantCurrent { + t.Fatalf("soft current=%v, want %v", candidate.LoadMetrics.SoftCurrent, test.wantCurrent) + } + if diff := candidate.LoadRatio - test.wantRatio; diff < -1e-9 || diff > 1e-9 { + t.Fatalf("load ratio=%v, want %v", candidate.LoadRatio, test.wantRatio) + } + }) + } +} + +func TestRuntimeCandidateLoadUsesSoftConcurrencyAlongsideRPM(t *testing.T) { + candidate := RuntimeModelCandidate{} + applyRuntimeCandidateLoad(&candidate, runtimeCandidateLoadInput{ + Policy: map[string]any{"rules": []any{map[string]any{"metric": "rpm", "limit": 100}}}, + RPMUsed: 20, + StateRunningCount: 4, + }) + + if !candidate.LoadLimited { + t.Fatal("rpm policy should remain a hard limit") + } + if candidate.LoadMetrics.ConcurrentLimit != 0 || candidate.LoadMetrics.ConcurrentRatio != 0.8 { + t.Fatalf("unexpected soft concurrent metric: %+v", candidate.LoadMetrics) + } + if candidate.LoadRatio != 0.8 { + t.Fatalf("load ratio=%v, want 0.8", candidate.LoadRatio) + } +} + +func TestRuntimeCandidateWaitingMarksCandidateFull(t *testing.T) { + candidates := []RuntimeModelCandidate{ + { + PlatformID: "higher-priority-waiting", + PlatformModelID: "model-a", + PlatformPriority: 10, + LoadRatio: 0.5, + LoadMetrics: RuntimeCandidateLoadMetrics{QueuedCount: 1}, + }, + { + PlatformID: "lower-priority-idle", + PlatformModelID: "model-b", + PlatformPriority: 20, + }, + } + + sortRuntimeModelCandidates(candidates) + + if candidates[0].PlatformID != "lower-priority-idle" { + t.Fatalf("non-full lower-priority candidate should receive overflow: %+v", candidates) + } + if !candidates[1].LoadAvoided || !containsString(candidates[1].FullReasons, "waiting") { + t.Fatalf("waiting candidate should be marked full and avoided: %+v", candidates[1]) + } +} + +func TestRuntimeCandidateSortingUsesStableIDAfterLoadTies(t *testing.T) { + candidates := []RuntimeModelCandidate{ + {PlatformID: "second", PlatformModelID: "model-b", PlatformPriority: 10}, + {PlatformID: "first", PlatformModelID: "model-a", PlatformPriority: 10}, + } + + sortRuntimeModelCandidates(candidates) + + if candidates[0].PlatformModelID != "model-a" { + t.Fatalf("stable candidate order=%+v", candidates) + } +} + +func TestRuntimeCandidateSortingKeepsDeterministicFallbackWhenAllFull(t *testing.T) { + candidates := []RuntimeModelCandidate{ + { + PlatformID: "second", + PlatformModelID: "model-b", + PlatformPriority: 10, + LoadMetrics: RuntimeCandidateLoadMetrics{ + ConcurrentLimit: 1, + ConcurrentRatio: 1, + }, + }, + { + PlatformID: "first", + PlatformModelID: "model-a", + PlatformPriority: 10, + LoadMetrics: RuntimeCandidateLoadMetrics{ + ConcurrentLimit: 1, + ConcurrentRatio: 1, + }, + }, + } + + sortRuntimeModelCandidates(candidates) + + if candidates[0].PlatformModelID != "model-a" || candidates[0].LoadAvoided || candidates[1].LoadAvoided { + t.Fatalf("all-full fallback should be stable without pretending a candidate was avoidable: %+v", candidates) + } + for _, candidate := range candidates { + if candidate.SelectionReason != "all_full_fallback" { + t.Fatalf("all-full candidate reason=%q, want all_full_fallback", candidate.SelectionReason) + } + } +} + +func containsString(values []string, expected string) bool { + for _, value := range values { + if value == expected { + return true + } + } + return false +} + func TestRuntimeCandidateSortingAvoidsFullCandidatesButKeepsFallback(t *testing.T) { candidates := []RuntimeModelCandidate{ { diff --git a/apps/api/internal/store/runtime_policies.go b/apps/api/internal/store/runtime_policies.go index 071ccbf..6c73289 100644 --- a/apps/api/internal/store/runtime_policies.go +++ b/apps/api/internal/store/runtime_policies.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "encoding/json" + "errors" "strings" "time" @@ -200,25 +201,30 @@ WHERE id = $1::uuid`, platformModelID, cooldownSeconds) return err } -func (s *Store) RuntimeCandidateAvailable(ctx context.Context, platformID string, platformModelID string) (bool, error) { +func (s *Store) RuntimeCandidateAvailability(ctx context.Context, platformID string, platformModelID string) (bool, string, error) { if strings.TrimSpace(platformID) == "" || strings.TrimSpace(platformModelID) == "" { - return false, nil + return false, "disabled_skipped", nil } - var available bool + var reason string 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 +SELECT CASE + WHEN platform.deleted_at IS NOT NULL OR platform.status <> 'enabled' OR model.enabled = false + THEN 'disabled_skipped' + WHEN platform.cooldown_until > now() OR model.cooldown_until > now() + THEN 'cooldown_skipped' + ELSE '' +END +FROM platform_models model +JOIN integration_platforms platform ON platform.id = model.platform_id +WHERE platform.id = $1::uuid + AND model.id = $2::uuid`, platformID, platformModelID).Scan(&reason) + if errors.Is(err, pgx.ErrNoRows) { + return false, "disabled_skipped", nil + } + if err != nil { + return false, "", err + } + return reason == "", reason, nil } func (s *Store) ApplyCandidateFailureEffect(ctx context.Context, input CandidateFailureEffectInput) (CandidateFailureEffectResult, error) { diff --git a/apps/api/internal/store/runtime_types.go b/apps/api/internal/store/runtime_types.go index 76a63db..fac8757 100644 --- a/apps/api/internal/store/runtime_types.go +++ b/apps/api/internal/store/runtime_types.go @@ -194,6 +194,8 @@ type RuntimeModelCandidate struct { LoadRatio float64 LoadLimited bool LoadAvoided bool + FullReasons []string + SelectionReason string LoadMetrics RuntimeCandidateLoadMetrics CacheAffinity RuntimeCandidateCacheAffinity RunningCount float64 @@ -230,6 +232,8 @@ type RuntimeCandidateLoadMetrics struct { ConcurrentCurrent float64 ConcurrentLimit float64 ConcurrentRatio float64 + SoftCurrent float64 + SoftRatio float64 QueuedCount float64 StateRunningCount float64 StateWaitingCount float64