Merge pull request #117 from xunxun1982/xxdev

feat: 优化429限速错误的健康度惩罚机制

Author: xunxun1982

internal/centralizedmgmt/hub_service.go

@@ -548,7 +548,9 @@ func (s *HubService) calculateAggregateGroupHealthScore(aggregateGroupID uint) f
 // calculateAggregateGroupHealthScoreWithVisited calculates health score with cycle detection.
 // Uses path-scoped visited set to prevent infinite recursion on circular aggregate group references.
 // The visited set is scoped to the current recursion path, allowing shared sub-groups in DAG structures.
-func (s *HubService) calculateAggregateGroupHealthScoreWithVisited(aggregateGroupID uint, visited map[uint]struct{}) float64 {
+// NOTE: Currently nested aggregates are not supported (validated at sub-group creation time),
+// so the visited parameter is not actively used but kept for future extensibility.
+func (s *HubService) calculateAggregateGroupHealthScoreWithVisited(aggregateGroupID uint, _ map[uint]struct{}) float64 {
 	// Get sub-group relationships
 	var subGroupRels []models.GroupSubGroup
 	if err := s.db.Where("group_id = ? AND weight > 0", aggregateGroupID).

internal/centralizedmgmt/hub_service_test.go

@@ -134,7 +134,7 @@ func createTestSubGroup(t *testing.T, db *gorm.DB, aggregateGroupID, subGroupID
 // setupHubService creates a HubService with test dependencies
 // Note: GroupManager is set to nil for most tests since HubService queries DB directly.
 // For tests that need SelectGroupForModel, we need to set up GroupManager properly.
-func setupHubService(t *testing.T, db *gorm.DB) *HubService {
+func setupHubService(_ *testing.T, db *gorm.DB) *HubService {
 	// Create a mock store
 	mockStore := store.NewMemoryStore()
 
@@ -150,7 +150,7 @@ func setupHubService(t *testing.T, db *gorm.DB) *HubService {
 // This is needed for tests that use SelectGroupForModel
 // Note: This requires the global db.DB to be set, which is complex in unit tests.
 // For now, we skip tests that require GroupManager if setup fails.
-func setupHubServiceWithGroupManager(t *testing.T, db *gorm.DB) *HubService {
+func setupHubServiceWithGroupManager(t *testing.T, _ *gorm.DB) *HubService {
 	t.Skip("Skipping test that requires GroupManager - complex setup with global db.DB")
 	return nil
 }
@@ -2396,7 +2396,7 @@ func TestCalculateGroupHealthScore(t *testing.T) {
 			hubService.dynamicWeightManager.RecordGroupSuccess(group.ID)
 		}
 		for i := 0; i < 3; i++ {
-			hubService.dynamicWeightManager.RecordGroupFailure(group.ID)
+			hubService.dynamicWeightManager.RecordGroupFailure(group.ID, false)
 		}
 
 		score := hubService.calculateGroupHealthScore(group)
@@ -2413,7 +2413,7 @@ func TestCalculateGroupHealthScore(t *testing.T) {
 	t.Run("consecutive_failures_reduce_health", func(t *testing.T) {
 		// Record 5 consecutive failures
 		for i := 0; i < 5; i++ {
-			hubService.dynamicWeightManager.RecordGroupFailure(group.ID)
+			hubService.dynamicWeightManager.RecordGroupFailure(group.ID, false)
 		}
 
 		score := hubService.calculateGroupHealthScore(group)
@@ -2458,15 +2458,15 @@ func TestGroupHealthScoreVsSubGroupMetrics(t *testing.T) {
 	for i := 0; i < 9; i++ {
 		hubService.dynamicWeightManager.RecordGroupSuccess(standardGroup.ID)
 	}
-	hubService.dynamicWeightManager.RecordGroupFailure(standardGroup.ID)
+	hubService.dynamicWeightManager.RecordGroupFailure(standardGroup.ID, false)
 
 	// Record poor performance in aggregate group (used for aggregate health score)
 	// Record 1 success first, then 8 failures to get low success rate + consecutive failures
 	for i := 0; i < 1; i++ {
 		hubService.dynamicWeightManager.RecordSubGroupSuccess(aggGroup.ID, standardGroup.ID)
 	}
 	for i := 0; i < 8; i++ {
-		hubService.dynamicWeightManager.RecordSubGroupFailure(aggGroup.ID, standardGroup.ID)
+		hubService.dynamicWeightManager.RecordSubGroupFailure(aggGroup.ID, standardGroup.ID, false)
 	}
 
 	// Hub health score should reflect group-level metrics (90% success rate)
@@ -2545,7 +2545,7 @@ func TestCalculateAggregateGroupHealthScore(t *testing.T) {
 			hubService.dynamicWeightManager.RecordSubGroupSuccess(aggGroup.ID, subGroup2.ID)
 		}
 		for i := 0; i < 3; i++ {
-			hubService.dynamicWeightManager.RecordSubGroupFailure(aggGroup.ID, subGroup2.ID)
+			hubService.dynamicWeightManager.RecordSubGroupFailure(aggGroup.ID, subGroup2.ID, false)
 		}
 
 		// Calculate aggregate health score (weighted average of sub-groups)
@@ -2588,7 +2588,7 @@ func TestHealthScoreInModelPool(t *testing.T) {
 		hubService.dynamicWeightManager.RecordGroupSuccess(group.ID)
 	}
 	for i := 0; i < 2; i++ {
-		hubService.dynamicWeightManager.RecordGroupFailure(group.ID)
+		hubService.dynamicWeightManager.RecordGroupFailure(group.ID, false)
 	}
 
 	// Manually test health score calculation
@@ -2670,7 +2670,7 @@ func TestAggregateHealthWithSkewedUsage(t *testing.T) {
 		hubService.dynamicWeightManager.RecordSubGroupSuccess(aggGroup.ID, subGroupB.ID)
 	}
 	for i := 0; i < 8; i++ {
-		hubService.dynamicWeightManager.RecordSubGroupFailure(aggGroup.ID, subGroupB.ID)
+		hubService.dynamicWeightManager.RecordSubGroupFailure(aggGroup.ID, subGroupB.ID, false)
 	}
 
 	// Sub-group C: 0 requests (never used)

internal/proxy/codex_cc_support_test.go

@@ -149,7 +149,7 @@ func TestCodexCCThinkingBlockWindowsPathConversion(t *testing.T) {
 }
 
 // Helper function to convert Codex response map to Claude response
-func convertCodexToClaudeResponseFromMap(codexResp map[string]interface{}, reverseToolNameMap map[string]string) *ClaudeResponse {
+func convertCodexToClaudeResponseFromMap(codexResp map[string]interface{}, _ map[string]string) *ClaudeResponse {
 	// This is a simplified version for testing
 	claudeResp := &ClaudeResponse{
 		ID:      "msg_test",

internal/proxy/gemini_cc_support.go

@@ -400,9 +400,10 @@ func convertClaudeMessageToGemini(msg ClaudeMessage, toolNameShortMap map[string
 		var userParts []GeminiPart
 		frIndex := 0 // Track which function response to use (matches order of tool_result blocks)
 		for _, block := range blocks {
-			if block.Type == "text" {
+			switch block.Type {
+			case "text":
 				userParts = append(userParts, GeminiPart{Text: block.Text})
-			} else if block.Type == "tool_result" {
+			case "tool_result":
 				// Flush accumulated user text before function response
 				if len(userParts) > 0 {
 					result = append(result, GeminiContent{
@@ -421,6 +422,8 @@ func convertClaudeMessageToGemini(msg ClaudeMessage, toolNameShortMap map[string
 						Parts: []GeminiPart{{FunctionResponse: &fr}},
 					})
 				}
+			default:
+				// Ignore unknown block types
 			}
 		}
 		// Flush any remaining user text

internal/proxy/server.go

@@ -757,8 +757,10 @@ func (ps *ProxyServer) HandleProxy(c *gin.Context) {
 			// direct API calls that should preserve their original query parameters.
 			// Also check /v1beta/messages for Gemini CC conversion (path may be rewritten to /v1beta/messages)
 			if isCCSupportEnabled(group) && (strings.HasSuffix(c.Request.URL.Path, "/v1/messages") || strings.HasSuffix(c.Request.URL.Path, "/v1beta/messages")) {
-				// Handle Codex channel CC support (Claude -> Codex/Responses API)
-				if group.ChannelType == "codex" {
+				// Handle channel-specific CC support conversions
+				switch group.ChannelType {
+				case "codex":
+					// Handle Codex channel CC support (Claude -> Codex/Responses API)
 					convertedBody, converted, ccErr := ps.applyCodexCCRequestConversion(c, group, finalBodyBytes)
 					if ccErr != nil {
 						logrus.WithError(ccErr).WithFields(logrus.Fields{
@@ -784,7 +786,7 @@ func (ps *ProxyServer) HandleProxy(c *gin.Context) {
 							"new_path":     c.Request.URL.Path,
 						}).Debug("Codex CC support: converted Claude request to Codex format")
 					}
-				} else if group.ChannelType == "gemini" {
+				case "gemini":
 					// Handle Gemini channel CC support (Claude -> Gemini API)
 					convertedBody, converted, ccErr := ps.applyGeminiCCRequestConversion(c, group, finalBodyBytes)
 					if ccErr != nil {
@@ -809,7 +811,7 @@ func (ps *ProxyServer) HandleProxy(c *gin.Context) {
 							"new_path":     c.Request.URL.Path,
 						}).Debug("Gemini CC support: converted Claude request to Gemini format")
 					}
-				} else {
+				default:
 					// Handle OpenAI channel CC support (Claude -> OpenAI Chat Completions)
 					convertedBody, converted, ccErr := ps.applyCCRequestConversionDirect(c, group, finalBodyBytes)
 					if ccErr != nil {
@@ -1377,8 +1379,10 @@ func (ps *ProxyServer) executeRequestWithAggregateRetry(
 	isMessagesEndpoint := strings.HasSuffix(c.Request.URL.Path, "/v1/messages") ||
 		strings.HasSuffix(c.Request.URL.Path, "/v1beta/messages")
 	if isCCSupportEnabled(group) && isMessagesEndpoint {
-		// Handle Codex channel CC support (Claude -> Codex/Responses API)
-		if group.ChannelType == "codex" {
+		// Handle channel-specific CC support conversions
+		switch group.ChannelType {
+		case "codex":
+			// Handle Codex channel CC support (Claude -> Codex/Responses API)
 			// Sanitize query parameters for Codex CC (remove Claude-specific params like beta=true)
 			// This is needed even if path wasn't /claude/ since Anthropic aggregate may send directly to /v1/messages
 			sanitizeCCQueryParams(c.Request.URL)
@@ -1440,7 +1444,7 @@ func (ps *ProxyServer) executeRequestWithAggregateRetry(
 					"new_path":        c.Request.URL.Path,
 				}).Debug("Codex CC support: converted Claude request for sub-group")
 			}
-		} else if group.ChannelType == "gemini" {
+		case "gemini":
 			// Handle Gemini channel CC support (Claude -> Gemini API)
 			sanitizeCCQueryParams(c.Request.URL)
 
@@ -1469,7 +1473,7 @@ func (ps *ProxyServer) executeRequestWithAggregateRetry(
 					"new_path":        c.Request.URL.Path,
 				}).Debug("Gemini CC support: converted Claude request for sub-group")
 			}
-		} else {
+		default:
 			// Handle OpenAI channel CC support (Claude -> OpenAI Chat Completions)
 			convertedBody, converted, ccErr := ps.applyCCRequestConversionDirect(c, group, finalBodyBytes)
 			if ccErr != nil {
@@ -2198,7 +2202,7 @@ func (ps *ProxyServer) logRequest(
 	// This ensures that failed sub-group attempts are reflected in health scores,
 	// even when the overall aggregate request succeeds via retry to another sub-group.
 	if ps.dynamicWeightManager != nil {
-		ps.recordDynamicWeightMetrics(c, originalGroup, group, logEntry.IsSuccess)
+		ps.recordDynamicWeightMetrics(c, originalGroup, group, logEntry.IsSuccess, statusCode)
 	}
 }
 
@@ -2209,23 +2213,30 @@ func (ps *ProxyServer) logRequest(
 // or unavailable, it can add tail latency or reduce throughput. The current implementation
 // prioritizes simplicity and correctness. For production deployments with strict latency SLAs,
 // consider async/buffering and ensure strict client timeouts in the store implementation.
-func (ps *ProxyServer) recordDynamicWeightMetrics(c *gin.Context, originalGroup, group *models.Group, isSuccess bool) {
+func (ps *ProxyServer) recordDynamicWeightMetrics(c *gin.Context, originalGroup, group *models.Group, isSuccess bool, statusCode int) {
 	if ps.dynamicWeightManager == nil {
 		return
 	}
 
+	// Determine if this is a rate limit error (429)
+	// Rate limit errors receive lighter penalties as they indicate temporary throttling
+	// rather than service unavailability
+	isRateLimit := !isSuccess && statusCode == 429
+
 	// Record sub-group metrics for aggregate groups
 	if originalGroup != nil && originalGroup.GroupType == "aggregate" && originalGroup.ID != group.ID {
 		if isSuccess {
 			ps.dynamicWeightManager.RecordSubGroupSuccess(originalGroup.ID, group.ID)
 		} else {
-			ps.dynamicWeightManager.RecordSubGroupFailure(originalGroup.ID, group.ID)
+			ps.dynamicWeightManager.RecordSubGroupFailure(originalGroup.ID, group.ID, isRateLimit)
 		}
 
 		logrus.WithFields(logrus.Fields{
 			"aggregate_group_id": originalGroup.ID,
 			"sub_group_id":       group.ID,
 			"is_success":         isSuccess,
+			"is_rate_limit":      isRateLimit,
+			"status_code":        statusCode,
 		}).Debug("Recorded dynamic weight metrics for sub-group")
 	}
 
@@ -2235,12 +2246,14 @@ func (ps *ProxyServer) recordDynamicWeightMetrics(c *gin.Context, originalGroup,
 		if isSuccess {
 			ps.dynamicWeightManager.RecordGroupSuccess(group.ID)
 		} else {
-			ps.dynamicWeightManager.RecordGroupFailure(group.ID)
+			ps.dynamicWeightManager.RecordGroupFailure(group.ID, isRateLimit)
 		}
 
 		logrus.WithFields(logrus.Fields{
-			"group_id":   group.ID,
-			"is_success": isSuccess,
+			"group_id":      group.ID,
+			"is_success":    isSuccess,
+			"is_rate_limit": isRateLimit,
+			"status_code":   statusCode,
 		}).Debug("Recorded dynamic weight metrics for standard group")
 	}
 
@@ -2264,15 +2277,17 @@ func (ps *ProxyServer) recordDynamicWeightMetrics(c *gin.Context, originalGroup,
 						if isSuccess {
 							ps.dynamicWeightManager.RecordModelRedirectSuccess(group.ID, originalModelStr, targetModel)
 						} else {
-							ps.dynamicWeightManager.RecordModelRedirectFailure(group.ID, originalModelStr, targetModel)
+							ps.dynamicWeightManager.RecordModelRedirectFailure(group.ID, originalModelStr, targetModel, isRateLimit)
 						}
 
 						logrus.WithFields(logrus.Fields{
-							"group_id":     group.ID,
-							"source_model": originalModelStr,
-							"target_model": targetModel,
-							"target_index": targetIdxInt,
-							"is_success":   isSuccess,
+							"group_id":      group.ID,
+							"source_model":  originalModelStr,
+							"target_model":  targetModel,
+							"target_index":  targetIdxInt,
+							"is_success":    isSuccess,
+							"is_rate_limit": isRateLimit,
+							"status_code":   statusCode,
 						}).Debug("Recorded dynamic weight metrics for model redirect")
 					}
 				}