feat(routing): 完善平台满载避让与故障轮转

为未配置并发上限的平台增加基于运行和等待任务数的软负载，并按非满载、有效优先级、缓存亲和与负载稳定排序。\n\n平台模型 RPM、TPM 和并发额度竞争失败时只轮转候选，不触发冷却、禁用或降级；用户组额度保持不可绕过。补齐异步冷却排队恢复、满载原因、选择原因与低基数指标。\n\n验证：go test ./...；go vet ./...；PostgreSQL 原子额度、准入队列、Worker 容量回收及故障策略 HTTP acceptance。
2026-08-03 17:42:34 +08:00
parent b125599354
commit 8362d6d27a
15 changed files with 468 additions and 152 deletions
@@ -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