Cscitech

controller/channel.go

@@ -178,7 +178,7 @@ func GetAllChannels(c *gin.Context) {
 		typeCounts[r.Type] = r.Count
 	}
 	common.ApiSuccess(c, gin.H{
-		"items":       channelData,
+		"items":       service.BuildChannelRoutingHealthViews(c.Request.Context(), channelData),
 		"total":       total,
 		"page":        pageInfo.GetPage(),
 		"page_size":   pageInfo.GetPageSize(),
@@ -373,7 +373,7 @@ func SearchChannels(c *gin.Context) {
 		"success": true,
 		"message": "",
 		"data": gin.H{
-			"items":       pagedData,
+			"items":       service.BuildChannelRoutingHealthViews(c.Request.Context(), pagedData),
 			"total":       total,
 			"type_counts": typeCounts,
 		},
@@ -398,7 +398,7 @@ func GetChannel(c *gin.Context) {
 	c.JSON(http.StatusOK, gin.H{
 		"success": true,
 		"message": "",
-		"data":    channel,
+		"data":    service.BuildChannelRoutingHealthViews(c.Request.Context(), []*model.Channel{channel})[0],
 	})
 	return
 }

controller/channel_routing_health_test.go

@@ -0,0 +1,203 @@
+package controller
+
+import (
+	"context"
+	"net/http"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/MAX-API-Next/MAX-API/common"
+	"github.com/MAX-API-Next/MAX-API/dto"
+	"github.com/MAX-API-Next/MAX-API/model"
+	"github.com/MAX-API-Next/MAX-API/pkg/channelhealth"
+	"github.com/MAX-API-Next/MAX-API/setting/operation_setting"
+	"github.com/alicebob/miniredis/v2"
+	"github.com/gin-gonic/gin"
+	"github.com/go-redis/redis/v8"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm"
+)
+
+func setupChannelRoutingHealthController(t *testing.T) (*gorm.DB, *model.Channel, *miniredis.Miniredis) {
+	t.Helper()
+	oldDB, oldLogDB := model.DB, model.LOG_DB
+	oldClient, oldEnabled := common.RDB, common.RedisEnabled
+	oldSQLite, oldMySQL, oldPostgres := common.UsingSQLite, common.UsingMySQL, common.UsingPostgreSQL
+	oldSetting := *operation_setting.GetMonitorSetting()
+	t.Cleanup(func() {
+		model.DB, model.LOG_DB = oldDB, oldLogDB
+		common.RDB, common.RedisEnabled = oldClient, oldEnabled
+		common.UsingSQLite, common.UsingMySQL, common.UsingPostgreSQL = oldSQLite, oldMySQL, oldPostgres
+		*operation_setting.GetMonitorSetting() = oldSetting
+	})
+	initModelListColumnNames(t)
+	db, channel := setupChannelUpdateControllerTestDB(t)
+	channel.ChannelInfo.MultiKeyDisabledReason = map[int]string{0: "fixture reason"}
+	channel.ChannelInfo.MultiKeyDisabledTime = map[int]int64{0: 123}
+	require.NoError(t, db.Save(channel).Error)
+	server := miniredis.RunT(t)
+	client := redis.NewClient(&redis.Options{Addr: server.Addr()})
+	common.RDB, common.RedisEnabled = client, true
+	t.Cleanup(func() { _ = client.Close() })
+	setting := operation_setting.MonitorSetting{
+		AutoPriorityDemotionEnabled: true, StreamingFirstResultTimeoutSeconds: 20,
+		PriorityDeduction: 1, PenaltyCooldownSeconds: 600, RecoveryMode: operation_setting.RecoveryModeAutomatic,
+	}
+	*operation_setting.GetMonitorSetting() = setting
+	result, err := channelhealth.EvaluateTimeoutGuard(context.Background(), channelhealth.AttemptEvidence{
+		ChannelID: channel.Id, AttemptID: t.Name(), RequestMode: channelhealth.RequestModeStreaming,
+		TimeoutKind: channelhealth.TimeoutKindStreamingFirstResult, TimeoutEligible: true, ObservedAt: time.Now().UTC(),
+	}, &setting)
+	require.NoError(t, err)
+	require.True(t, result.Applied)
+	return db, channel, server
+}
+
+func TestChannelRoutingHealthManagementReadEndpointsPreserveConfiguration(t *testing.T) {
+	db, primary, server := setupChannelRoutingHealthController(t)
+	secondary := &model.Channel{
+		Name: "secondary-channel", Type: 1, Key: "fixture-secondary", Status: common.ChannelStatusEnabled,
+		Models: "gpt-origin", Group: "default", Tag: primary.Tag, Priority: common.GetPointer(int64(9)),
+	}
+	require.NoError(t, db.Create(secondary).Error)
+	before := server.Dump()
+	for _, test := range []struct {
+		name    string
+		url     string
+		handler gin.HandlerFunc
+		detail  bool
+		count   int
+		total   int
+	}{
+		{"list page", "/api/channel/?page_size=1&sort_by=priority&sort_order=desc", GetAllChannels, false, 1, 2},
+		{"search page", "/api/channel/search?keyword=channel&page_size=1&sort_by=priority&sort_order=desc", SearchChannels, false, 1, 2},
+		{"tag children", "/api/channel/?tag_mode=true&page_size=1&sort_by=priority&sort_order=desc", GetAllChannels, false, 2, 1},
+		{"tag search", "/api/channel/search?tag_mode=true&keyword=channel&page_size=20&sort_by=priority&sort_order=desc", SearchChannels, false, 2, 2},
+		{"detail", "/api/channel/" + strconv.Itoa(primary.Id), GetChannel, true, 1, 0},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			ctx, recorder := newAuthenticatedContext(t, http.MethodGet, test.url, nil, 1)
+			ctx.Params = gin.Params{{Key: "id", Value: strconv.Itoa(primary.Id)}}
+			test.handler(ctx)
+			require.Equal(t, http.StatusOK, recorder.Code)
+			response := decodeAPIResponse(t, recorder)
+			require.True(t, response.Success, response.Message)
+			var item map[string]interface{}
+			if test.detail {
+				require.NoError(t, common.Unmarshal(response.Data, &item))
+			} else {
+				var page struct {
+					Items []map[string]interface{} `json:"items"`
+					Total int                      `json:"total"`
+				}
+				require.NoError(t, common.Unmarshal(response.Data, &page))
+				require.Len(t, page.Items, test.count)
+				require.Equal(t, test.total, page.Total)
+				item = page.Items[0]
+				if len(page.Items) > 1 {
+					require.Equal(t, dto.ChannelRoutingHealthNormal, page.Items[1]["routing_health"].(map[string]interface{})["status"])
+				}
+			}
+			require.EqualValues(t, primary.Id, item["id"])
+			require.EqualValues(t, 10, item["priority"])
+			health := item["routing_health"].(map[string]interface{})
+			require.Equal(t, dto.ChannelRoutingHealthPenalized, health["status"])
+			require.EqualValues(t, 9, health["effective_priority"])
+			require.EqualValues(t, 1, health["active_penalty"])
+			require.Empty(t, item["key"])
+			require.NotContains(t, item["channel_info"], "multi_key_disabled_reason")
+			require.NotContains(t, item["channel_info"], "multi_key_disabled_time")
+			delete(item, "routing_health")
+			original, err := model.GetChannelById(primary.Id, false)
+			require.NoError(t, err)
+			clearChannelInfo(original)
+			encoded, err := common.Marshal(original)
+			require.NoError(t, err)
+			var expected map[string]interface{}
+			require.NoError(t, common.Unmarshal(encoded, &expected))
+			require.Equal(t, expected, item)
+			require.Equal(t, before, server.Dump())
+		})
+	}
+	stored, err := model.GetChannelById(primary.Id, true)
+	require.NoError(t, err)
+	require.EqualValues(t, 10, stored.GetPriority())
+	require.Equal(t, primary.Key, stored.Key)
+	require.Equal(t, primary.ChannelInfo.MultiKeyDisabledReason, stored.ChannelInfo.MultiKeyDisabledReason)
+	var logCount int64
+	require.NoError(t, db.Model(&model.Log{}).Count(&logCount).Error)
+	require.Zero(t, logCount)
+}
+
+func TestChannelRoutingHealthManagementReadSurvivesRedisFailure(t *testing.T) {
+	_, channel, server := setupChannelRoutingHealthController(t)
+	server.SetError("ERR private-diagnostic")
+	ctx, recorder := newAuthenticatedContext(t, http.MethodGet, "/api/channel/", nil, 1)
+	GetAllChannels(ctx)
+	response := decodeAPIResponse(t, recorder)
+	require.True(t, response.Success, response.Message)
+	var page struct {
+		Items []struct {
+			ID            int                      `json:"id"`
+			Priority      int64                    `json:"priority"`
+			RoutingHealth dto.ChannelRoutingHealth `json:"routing_health"`
+		} `json:"items"`
+	}
+	require.NoError(t, common.Unmarshal(response.Data, &page))
+	require.Len(t, page.Items, 1)
+	require.Equal(t, channel.Id, page.Items[0].ID)
+	require.EqualValues(t, 10, page.Items[0].Priority)
+	require.Equal(t, dto.ChannelRoutingHealthUnknown, page.Items[0].RoutingHealth.Status)
+	require.Nil(t, page.Items[0].RoutingHealth.EffectivePriority)
+	require.Equal(t, "redis_read_failed", *page.Items[0].RoutingHealth.UnavailableReason)
+	require.NotContains(t, recorder.Body.String(), "private-diagnostic")
+}
+
+func TestChannelRoutingHealthCannotCreatePhantomChannelOrBeMassAssigned(t *testing.T) {
+	db, channel, server := setupChannelRoutingHealthController(t)
+	// Keep the existing update audit on a cache hit so this test does not leave
+	// unrelated asynchronous user-cache refill work after its Redis is closed.
+	require.NoError(t, db.AutoMigrate(&model.CacheInvalidationTask{}))
+	require.NoError(t, common.RedisHSetObj("user:1", &model.UserBase{
+		Id: 1, Username: "fixture-admin", Role: common.RoleRootUser, Status: common.UserStatusEnabled,
+	}, time.Hour))
+	require.NoError(t, common.RDB.HSet(context.Background(), channelhealth.RuntimeStateKey(99999), map[string]interface{}{
+		"penalty": "1", "runtime_disabled": "false",
+	}).Err())
+	before := server.Dump()
+	runtimeBefore, err := common.RDB.HGetAll(context.Background(), channelhealth.RuntimeStateKey(channel.Id)).Result()
+	require.NoError(t, err)
+	ctx, recorder := newAuthenticatedContext(t, http.MethodGet, "/api/channel/99999", nil, 1)
+	ctx.Params = gin.Params{{Key: "id", Value: "99999"}}
+	GetChannel(ctx)
+	response := decodeAPIResponse(t, recorder)
+	require.False(t, response.Success)
+	require.Equal(t, before, server.Dump())
+
+	ctx, recorder = newAuthenticatedContext(t, http.MethodPut, "/api/channel/", map[string]interface{}{
+		"id": channel.Id, "priority": 11,
+		"routing_health": map[string]interface{}{
+			"effective_priority": -999, "stored_penalty": 0, "runtime_disabled": true,
+		},
+	}, 1)
+	UpdateChannel(ctx)
+	response = decodeAPIResponse(t, recorder)
+	require.True(t, response.Success, response.Message)
+	stored, err := model.GetChannelById(channel.Id, false)
+	require.NoError(t, err)
+	require.EqualValues(t, 11, stored.GetPriority())
+	runtimeAfter, err := common.RDB.HGetAll(context.Background(), channelhealth.RuntimeStateKey(channel.Id)).Result()
+	require.NoError(t, err)
+	require.Equal(t, runtimeBefore, runtimeAfter)
+	ctx, recorder = newAuthenticatedContext(t, http.MethodGet, "/api/channel/"+strconv.Itoa(channel.Id), nil, 1)
+	ctx.Params = gin.Params{{Key: "id", Value: strconv.Itoa(channel.Id)}}
+	GetChannel(ctx)
+	response = decodeAPIResponse(t, recorder)
+	var view struct {
+		RoutingHealth dto.ChannelRoutingHealth `json:"routing_health"`
+	}
+	require.NoError(t, common.Unmarshal(response.Data, &view))
+	require.EqualValues(t, 10, *view.RoutingHealth.EffectivePriority)
+	require.False(t, *view.RoutingHealth.RuntimeDisabled)
+}

dto/channel_routing_health.go

@@ -0,0 +1,36 @@
+package dto
+
+import "time"
+
+const (
+	ChannelRoutingHealthNormal          = "normal"
+	ChannelRoutingHealthPenalized       = "penalized"
+	ChannelRoutingHealthRuntimeDisabled = "runtime_disabled"
+	ChannelRoutingHealthPolicyOff       = "policy_off"
+	ChannelRoutingHealthUnknown         = "unknown"
+
+	ChannelRoutingHealthSourceRedis       = "redis"
+	ChannelRoutingHealthSourcePolicyOff   = "policy_off"
+	ChannelRoutingHealthSourceUnavailable = "unavailable"
+)
+
+// ChannelRoutingHealth is a read-only management snapshot, not channel configuration.
+// Nullable fields distinguish an unavailable observation from a known zero or false.
+type ChannelRoutingHealth struct {
+	SchemaVersion         int        `json:"schema_version"`
+	Status                string     `json:"status"`
+	Source                string     `json:"source"`
+	BasePriority          int64      `json:"base_priority"`
+	EffectivePriority     *int64     `json:"effective_priority"`
+	StoredPenalty         *int64     `json:"stored_penalty"`
+	ActivePenalty         *int64     `json:"active_penalty"`
+	StoredRuntimeDisabled *bool      `json:"stored_runtime_disabled"`
+	RuntimeDisabled       *bool      `json:"runtime_disabled"`
+	PenaltyUntil          *time.Time `json:"penalty_until"`
+	DisabledUntil         *time.Time `json:"disabled_until"`
+	PenaltyRecoveryMode   *string    `json:"penalty_recovery_mode"`
+	DisableRecoveryMode   *string    `json:"disable_recovery_mode"`
+	LastObservedAt        *time.Time `json:"last_observed_at"`
+	EvaluatedAt           time.Time  `json:"evaluated_at"`
+	UnavailableReason     *string    `json:"unavailable_reason"`
+}

pkg/channelhealth/runtime_display.go

@@ -0,0 +1,119 @@
+package channelhealth
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	maxcommon "github.com/MAX-API-Next/MAX-API/common"
+	"github.com/go-redis/redis/v8"
+)
+
+var ErrInvalidRuntimeState = errors.New("invalid channel timeout runtime state")
+
+// RuntimeStateReadResult preserves per-channel validation failures without
+// converting corrupt state into an apparently healthy empty observation.
+type RuntimeStateReadResult struct {
+	State RuntimeState
+	Err   error
+}
+
+// LoadRuntimeStatesStrict only reads the requested hashes. Routing continues to
+// use the existing tolerant loaders; this stricter adapter is for management views.
+func LoadRuntimeStatesStrict(ctx context.Context, channelIDs []int) (map[int]RuntimeStateReadResult, error) {
+	results := make(map[int]RuntimeStateReadResult, len(channelIDs))
+	if len(channelIDs) == 0 {
+		return results, nil
+	}
+	client := maxcommon.RDB
+	if !maxcommon.RedisEnabled || client == nil {
+		return results, ErrRuntimeUnavailable
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if err := ctx.Err(); err != nil {
+		return results, err
+	}
+	pipe := client.Pipeline()
+	defer pipe.Close()
+	commands := make(map[int]*redis.StringStringMapCmd, len(channelIDs))
+	for _, channelID := range channelIDs {
+		if channelID <= 0 {
+			continue
+		}
+		if _, ok := commands[channelID]; ok {
+			continue
+		}
+		commands[channelID] = pipe.HGetAll(ctx, runtimeStateKey(channelID))
+	}
+	if len(commands) == 0 {
+		return results, nil
+	}
+	if _, err := pipe.Exec(ctx); err != nil {
+		return results, err
+	}
+	for channelID, command := range commands {
+		fields, err := command.Result()
+		if err != nil {
+			return nil, err
+		}
+		if err := validateRuntimeStateFields(channelID, fields); err != nil {
+			results[channelID] = RuntimeStateReadResult{Err: err}
+			continue
+		}
+		results[channelID] = RuntimeStateReadResult{State: decodeRuntimeState(channelID, fields)}
+	}
+	return results, nil
+}
+
+func validateRuntimeStateFields(channelID int, fields map[string]string) error {
+	if len(fields) == 0 {
+		return nil // A successful read of a missing hash is a known empty state.
+	}
+	invalid := func(field string) error {
+		return fmt.Errorf("%w: %s", ErrInvalidRuntimeState, field)
+	}
+	// Both fields have always been written by the runtime encoder. Other fields
+	// remain optional for older states, including states predating recovery watermarks.
+	if _, ok := fields["penalty"]; !ok {
+		return invalid("penalty")
+	}
+	if value := fields["runtime_disabled"]; value != "true" && value != "false" {
+		return invalid("runtime_disabled")
+	}
+	if value, ok := fields["channel_id"]; ok {
+		id, err := strconv.ParseInt(value, 10, 64)
+		if err != nil || id != int64(channelID) {
+			return invalid("channel_id")
+		}
+	}
+	intFields := []string{"penalty", "generation", "recovery_probe_success_count"}
+	timeFields := []string{"penalty_until", "disabled_until", "last_observed_at", "recovery_watermark"}
+	for _, mode := range []string{RequestModeStreaming, RequestModeNonStreaming, "combined"} {
+		intFields = append(intFields, "timeout_count_"+mode, "sample_count_"+mode)
+		timeFields = append(timeFields, "window_start_"+mode)
+	}
+	for _, field := range intFields {
+		if value, ok := fields[field]; ok {
+			parsed, err := strconv.ParseInt(value, 10, 64)
+			if err != nil || parsed < 0 {
+				return invalid(field)
+			}
+		}
+	}
+	for _, field := range timeFields {
+		if value := fields[field]; value != "" {
+			if _, err := time.Parse(time.RFC3339Nano, value); err != nil {
+				return invalid(field)
+			}
+		}
+	}
+	if mode := fields["last_timeout_mode"]; mode != "" &&
+		mode != RequestModeStreaming && mode != RequestModeNonStreaming {
+		return invalid("last_timeout_mode")
+	}
+	return nil
+}