feat: add runner failover policies and traces

2026-05-12 02:16:42 +08:00
parent be31923e74
commit 05632172d0
26 changed files with 2033 additions and 140 deletions
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"log/slog"
+	"strconv"
 	"strings"
 	"time"

@@ -51,6 +52,7 @@ func (s *Service) ExecuteStream(ctx context.Context, task store.GatewayTask, use
 }

 func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *auth.User, onDelta clients.StreamDelta) (Result, error) {
+	executeStartedAt := time.Now()
 	body := normalizeRequest(task.Kind, task.Request)
 	modelType := modelTypeFromKind(task.Kind, body)
 	if err := validateRequest(task.Kind, body); err != nil {
@@ -88,64 +90,159 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 		return Result{}, err
 	}

-	maxAttempts := maxAttemptsForCandidates(candidates)
+	runnerPolicy, err := s.store.GetActiveRunnerPolicy(ctx)
+	if err != nil {
+		return Result{}, err
+	}
+	maxPlatforms := maxPlatformsForCandidates(candidates, runnerPolicy)
+	maxFailoverDuration := maxFailoverDurationForCandidates(candidates, runnerPolicy)
+	attemptNo := 0
 	var lastErr error
 	for index, candidate := range candidates {
-		if index >= maxAttempts {
+		if index >= maxPlatforms {
 			break
 		}
-		attemptNo := index + 1
-		response, err := s.runCandidate(ctx, task, user, body, candidate, attemptNo, onDelta)
-		if err == nil {
-			billings := s.billings(ctx, user, task.Kind, body, candidate, response, isSimulation(task, candidate))
-			record := buildSuccessRecord(task, user, body, candidate, response, billings, isSimulation(task, candidate))
-			record.Metrics = s.withAttemptHistory(ctx, task.ID, record.Metrics)
-			finished, finishErr := s.store.FinishTaskSuccess(ctx, store.FinishTaskSuccessInput{
-				TaskID:             task.ID,
-				Result:             response.Result,
-				Billings:           billings,
-				RequestID:          record.RequestID,
-				ResolvedModel:      record.ResolvedModel,
-				Usage:              record.Usage,
-				Metrics:            record.Metrics,
-				BillingSummary:     record.BillingSummary,
-				FinalChargeAmount:  record.FinalChargeAmount,
-				ResponseStartedAt:  record.ResponseStartedAt,
-				ResponseFinishedAt: record.ResponseFinishedAt,
-				ResponseDurationMS: record.ResponseDurationMS,
-			})
-			if finishErr != nil {
-				return Result{}, finishErr
-			}
-			if settleErr := s.store.SettleTaskBilling(ctx, finished); settleErr != nil {
-				return Result{}, settleErr
-			}
-			if finished.FinalChargeAmount > 0 {
-				if err := s.emit(ctx, task.ID, "task.billing.settled", "succeeded", "billing", 0.98, "task billing settled", map[string]any{
-					"amount":   finished.FinalChargeAmount,
-					"currency": stringFromAny(record.BillingSummary["currency"]),
+		clientAttempts := clientAttemptsForCandidate(candidate)
+		var candidateErr error
+		for clientAttempt := 1; clientAttempt <= clientAttempts; clientAttempt++ {
+			attemptNo++
+			response, err := s.runCandidate(ctx, task, user, body, candidate, attemptNo, onDelta)
+			if err == nil {
+				billings := s.billings(ctx, user, task.Kind, body, candidate, response, isSimulation(task, candidate))
+				record := buildSuccessRecord(task, user, body, candidate, response, billings, isSimulation(task, candidate))
+				record.Metrics = s.withAttemptHistory(ctx, task.ID, record.Metrics)
+				finished, finishErr := s.store.FinishTaskSuccess(ctx, store.FinishTaskSuccessInput{
+					TaskID:             task.ID,
+					Result:             response.Result,
+					Billings:           billings,
+					RequestID:          record.RequestID,
+					ResolvedModel:      record.ResolvedModel,
+					Usage:              record.Usage,
+					Metrics:            record.Metrics,
+					BillingSummary:     record.BillingSummary,
+					FinalChargeAmount:  record.FinalChargeAmount,
+					ResponseStartedAt:  record.ResponseStartedAt,
+					ResponseFinishedAt: record.ResponseFinishedAt,
+					ResponseDurationMS: record.ResponseDurationMS,
+				})
+				if finishErr != nil {
+					return Result{}, finishErr
+				}
+				if settleErr := s.store.SettleTaskBilling(ctx, finished); settleErr != nil {
+					return Result{}, settleErr
+				}
+				if finished.FinalChargeAmount > 0 {
+					if err := s.emit(ctx, task.ID, "task.billing.settled", "succeeded", "billing", 0.98, "task billing settled", map[string]any{
+						"amount":   finished.FinalChargeAmount,
+						"currency": stringFromAny(record.BillingSummary["currency"]),
+					}, isSimulation(task, candidate)); err != nil {
+						return Result{}, err
+					}
+				}
+				if err := s.emit(ctx, task.ID, "task.completed", "succeeded", "completed", 1, "task completed", map[string]any{
+					"result":         response.Result,
+					"billings":       billings,
+					"usage":          record.Usage,
+					"metrics":        record.Metrics,
+					"billingSummary": record.BillingSummary,
+					"requestId":      record.RequestID,
 				}, isSimulation(task, candidate)); err != nil {
 					return Result{}, err
 				}
+				return Result{Task: finished, Output: response.Result}, nil
 			}
-			if err := s.emit(ctx, task.ID, "task.completed", "succeeded", "completed", 1, "task completed", map[string]any{
-				"result":         response.Result,
-				"billings":       billings,
-				"usage":          record.Usage,
-				"metrics":        record.Metrics,
-				"billingSummary": record.BillingSummary,
-				"requestId":      record.RequestID,
+			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(retryDecision, retryAction, clientAttempt, clientAttempts))
+			if !retryDecision.Retry {
+				break
+			}
+			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,
+				"scope":         "same_client",
+			}, retryDecision.Match, retryDecision.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, 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)}
+					}
+				}
+				s.recordAttemptTrace(ctx, task.ID, attemptNo, failoverTraceEntry(decision))
+			}
+			break
+		}
+		s.applyPriorityDemotePolicy(ctx, task.ID, attemptNo, runnerPolicy, candidate, 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)))
+			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
 			}
-			return Result{Task: finished, Output: response.Result}, nil
-		}
-		lastErr = err
-		retryable := clients.IsRetryable(err)
-		if !retryable || !retryEnabled(candidate) || attemptNo >= maxAttempts {
 			break
 		}
-		if err := s.emit(ctx, task.ID, "task.retrying", "running", "retry", 0.55, "retrying next client", map[string]any{"attempt": attemptNo, "error": err.Error()}, isSimulation(task, candidate)); err != nil {
+		decision := failoverDecisionForCandidate(runnerPolicy, candidate, candidateErr)
+		s.recordAttemptTrace(ctx, task.ID, attemptNo, failoverTraceEntry(decision))
+		if !decision.Retry {
+			break
+		}
+		s.applyFailoverAction(ctx, task.ID, candidate, decision, isSimulation(task, candidate))
+		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,
+			"error":   candidateErr.Error(),
+			"reason":  decision.Reason,
+			"scope":   "next_platform",
+		}, decision.Match, decision.Info), isSimulation(task, candidate)); err != nil {
 			return Result{}, err
 		}
 	}
@@ -184,15 +281,16 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 	reservations := s.rateLimitReservations(ctx, user, candidate, body)
 	limitResult, err := s.store.ReserveRateLimits(ctx, task.ID, reservations)
 	if err != nil {
+		clientErr := &clients.ClientError{Code: "rate_limit", Message: err.Error(), Retryable: false}
 		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
 			AttemptID:    attemptID,
 			Status:       "failed",
 			Retryable:    false,
-			Metrics:      mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), map[string]any{"error": err.Error(), "retryable": false}),
+			Metrics:      mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(clientErr, false)}}),
 			ErrorCode:    "rate_limit",
 			ErrorMessage: err.Error(),
 		})
-		return clients.Response{}, &clients.ClientError{Code: "rate_limit", Message: err.Error(), Retryable: false}
+		return clients.Response{}, clientErr
 	}
 	defer s.store.ReleaseConcurrencyLeases(context.WithoutCancel(ctx), limitResult.LeaseIDs)

@@ -207,7 +305,7 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 			AttemptID:    attemptID,
 			Status:       "failed",
 			Retryable:    false,
-			Metrics:      mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), map[string]any{"error": err.Error(), "retryable": false}),
+			Metrics:      mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(err, false)}}),
 			ErrorCode:    clients.ErrorCode(err),
 			ErrorMessage: err.Error(),
 		})
@@ -266,7 +364,7 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 			ErrorCode:          clients.ErrorCode(err),
 			ErrorMessage:       err.Error(),
 		})
-		_ = s.emit(ctx, task.ID, "task.attempt.failed", "running", "attempt_failed", 0.45, err.Error(), map[string]any{"attempt": attemptNo, "retryable": retryable, "requestId": requestID, "metrics": metrics}, simulated)
+		_ = s.emit(ctx, task.ID, "task.attempt.failed", "running", "attempt_failed", 0.45, err.Error(), map[string]any{"attempt": attemptNo, "retryable": retryable, "requestId": requestID, "statusCode": clients.ErrorResponseMetadata(err).StatusCode, "metrics": metrics}, simulated)
 		s.applyCandidateFailurePolicies(ctx, task.ID, candidate, err, simulated)
 		return clients.Response{}, err
 	}
@@ -275,6 +373,7 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 		metrics := mergeMetrics(taskMetrics(task, user, body, candidate, response, simulated), map[string]any{
 			"error":     err.Error(),
 			"retryable": clients.IsRetryable(err),
+			"trace":     []any{failureTraceEntry(err, clients.IsRetryable(err))},
 		})
 		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
 			AttemptID:          attemptID,
@@ -429,20 +528,54 @@ func retryEnabled(candidate store.RuntimeModelCandidate) bool {
 	return true
 }

-func maxAttemptsForCandidates(candidates []store.RuntimeModelCandidate) int {
+func clientAttemptsForCandidate(candidate store.RuntimeModelCandidate) int {
+	if !retryEnabled(candidate) {
+		return 1
+	}
+	if value := intFromPolicy(effectiveRetryPolicy(candidate), "maxAttempts"); value > 0 {
+		return value
+	}
+	return 1
+}
+
+func maxPlatformsForCandidates(candidates []store.RuntimeModelCandidate, runnerPolicy store.RunnerPolicy) int {
 	if len(candidates) == 0 {
 		return 0
 	}
-	maxAttempts := len(candidates)
+	maxPlatforms := len(candidates)
+	if value := intFromPolicy(runnerPolicy.FailoverPolicy, "maxPlatforms"); value > 0 {
+		if value < maxPlatforms {
+			maxPlatforms = value
+		}
+	} else if maxPlatforms > 99 {
+		maxPlatforms = 99
+	}
 	for _, candidate := range candidates {
-		if value := intFromPolicy(effectiveRetryPolicy(candidate), "maxAttempts"); value > 0 && value < maxAttempts {
-			maxAttempts = value
+		if value := intFromPolicy(effectiveFailoverPolicy(runnerPolicy.FailoverPolicy, candidate.RuntimePolicyOverride), "maxPlatforms"); value > 0 && value < maxPlatforms {
+			maxPlatforms = value
 		}
 	}
-	if maxAttempts <= 0 {
+	if maxPlatforms <= 0 {
 		return 1
 	}
-	return maxAttempts
+	return maxPlatforms
+}
+
+func maxFailoverDurationForCandidates(candidates []store.RuntimeModelCandidate, runnerPolicy store.RunnerPolicy) time.Duration {
+	seconds := intFromPolicy(runnerPolicy.FailoverPolicy, "maxDurationSeconds")
+	if seconds <= 0 {
+		seconds = 600
+	}
+	for _, candidate := range candidates {
+		if value := intFromPolicy(effectiveFailoverPolicy(runnerPolicy.FailoverPolicy, candidate.RuntimePolicyOverride), "maxDurationSeconds"); value > 0 && value < seconds {
+			seconds = value
+		}
+	}
+	return time.Duration(seconds) * time.Second
+}
+
+func failoverTimeBudgetExceeded(start time.Time, maxDuration time.Duration) bool {
+	return maxDuration > 0 && time.Since(start) >= maxDuration
 }

 func normalizeRequest(kind string, body map[string]any) map[string]any {