diff --git a/controller/channel.go b/controller/channel.go index 386b27ac..b2ec9464 100644 --- a/controller/channel.go +++ b/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 } diff --git a/controller/channel_routing_health_test.go b/controller/channel_routing_health_test.go new file mode 100644 index 00000000..dbf739c3 --- /dev/null +++ b/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) +} diff --git a/dto/channel_routing_health.go b/dto/channel_routing_health.go new file mode 100644 index 00000000..2bf6e70c --- /dev/null +++ b/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"` +} diff --git a/pkg/channelhealth/runtime_display.go b/pkg/channelhealth/runtime_display.go new file mode 100644 index 00000000..f2869ebd --- /dev/null +++ b/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 +} diff --git a/pkg/channelhealth/runtime_display_test.go b/pkg/channelhealth/runtime_display_test.go new file mode 100644 index 00000000..c4dc26e6 --- /dev/null +++ b/pkg/channelhealth/runtime_display_test.go @@ -0,0 +1,139 @@ +package channelhealth + +import ( + "context" + "io" + "maps" + "net" + "testing" + "time" + + maxcommon "github.com/MAX-API-Next/MAX-API/common" + "github.com/go-redis/redis/v8" + "github.com/stretchr/testify/require" +) + +type displayReadHook struct { + keys []string +} + +func (hook *displayReadHook) BeforeProcess(ctx context.Context, cmd redis.Cmder) (context.Context, error) { + hook.keys = append(hook.keys, cmd.Name()) + return ctx, nil +} + +func (hook *displayReadHook) AfterProcess(context.Context, redis.Cmder) error { return nil } + +func (hook *displayReadHook) BeforeProcessPipeline(ctx context.Context, commands []redis.Cmder) (context.Context, error) { + for _, command := range commands { + hook.keys = append(hook.keys, command.Args()[1].(string)) + } + return ctx, nil +} + +func (hook *displayReadHook) AfterProcessPipeline(context.Context, []redis.Cmder) error { return nil } + +func TestLoadRuntimeStatesStrictIsReadOnlyAndKeepsRoutingCompatibility(t *testing.T) { + withRuntimeRedis(t) + ctx := context.Background() + now := time.Now().UTC() + state := emptyRuntimeState(1) + state.Penalty = 1 + state.PenaltyUntil = now.Add(time.Minute) + state.LastObservedAt = now + require.NoError(t, maxcommon.RDB.HSet(ctx, RuntimeStateKey(1), encodeRuntimeState(state)).Err()) + require.NoError(t, maxcommon.RDB.HSet(ctx, RuntimeStateKey(3), map[string]interface{}{ + "penalty": "broken", "runtime_disabled": "false", + }).Err()) + before, err := maxcommon.RDB.HGetAll(ctx, RuntimeStateKey(1)).Result() + require.NoError(t, err) + hook := &displayReadHook{} + maxcommon.RDB.AddHook(hook) + + results, err := LoadRuntimeStatesStrict(ctx, []int{1, 2, 3, 1, 0, -1}) + require.NoError(t, err) + require.Len(t, results, 3) + require.ElementsMatch(t, []string{RuntimeStateKey(1), RuntimeStateKey(2), RuntimeStateKey(3)}, hook.keys) + require.NoError(t, results[1].Err) + require.Equal(t, int64(1), results[1].State.Penalty) + require.Equal(t, state.PenaltyUntil, results[1].State.PenaltyUntil) + require.NoError(t, results[2].Err) + require.Zero(t, results[2].State.Penalty) + require.ErrorIs(t, results[3].Err, ErrInvalidRuntimeState) + + // The routing loader intentionally retains its pre-existing tolerant behavior. + routing, err := LoadRuntimeStates(ctx, []int{3}) + require.NoError(t, err) + require.Zero(t, routing[3].Penalty) + after, err := maxcommon.RDB.HGetAll(ctx, RuntimeStateKey(1)).Result() + require.NoError(t, err) + require.Equal(t, before, after) + require.EqualValues(t, 0, maxcommon.RDB.Exists(ctx, RuntimeStateKey(2)).Val()) +} + +func TestLoadRuntimeStatesStrictValidatesCorruptionPerChannel(t *testing.T) { + withRuntimeRedis(t) + valid := map[string]string{"penalty": "1", "runtime_disabled": "false"} + for name, change := range map[string]func(map[string]string){ + "missing penalty": func(fields map[string]string) { delete(fields, "penalty") }, + "missing flag": func(fields map[string]string) { delete(fields, "runtime_disabled") }, + "invalid flag": func(fields map[string]string) { fields["runtime_disabled"] = "TRUE" }, + "negative": func(fields map[string]string) { fields["penalty"] = "-1" }, + "overflow": func(fields map[string]string) { fields["penalty"] = "9223372036854775808" }, + "bad deadline": func(fields map[string]string) { fields["penalty_until"] = "tomorrow" }, + "bad observation": func(fields map[string]string) { fields["last_observed_at"] = "broken" }, + "bad counter": func(fields map[string]string) { fields["timeout_count_streaming"] = "broken" }, + "wrong channel": func(fields map[string]string) { fields["channel_id"] = "99" }, + "bad mode": func(fields map[string]string) { fields["last_timeout_mode"] = "unknown" }, + } { + t.Run(name, func(t *testing.T) { + fields := maps.Clone(valid) + change(fields) + require.NoError(t, maxcommon.RDB.Del(context.Background(), RuntimeStateKey(1)).Err()) + require.NoError(t, maxcommon.RDB.HSet(context.Background(), RuntimeStateKey(1), fields).Err()) + results, err := LoadRuntimeStatesStrict(context.Background(), []int{1, 2}) + require.NoError(t, err) + require.ErrorIs(t, results[1].Err, ErrInvalidRuntimeState) + require.NoError(t, results[2].Err) + }) + } + // Legacy states may lack fields added later; unknown fields are forward compatible. + valid["future_field"] = "opaque" + require.NoError(t, maxcommon.RDB.Del(context.Background(), RuntimeStateKey(1)).Err()) + require.NoError(t, maxcommon.RDB.HSet(context.Background(), RuntimeStateKey(1), valid).Err()) + results, err := LoadRuntimeStatesStrict(context.Background(), []int{1}) + require.NoError(t, err) + require.NoError(t, results[1].Err) +} + +func TestLoadRuntimeStatesStrictDoesNotReturnPartialBatchOnRedisError(t *testing.T) { + withRuntimeRedis(t) + require.NoError(t, maxcommon.RDB.Set(context.Background(), RuntimeStateKey(2), "wrong-type", 0).Err()) + results, err := LoadRuntimeStatesStrict(context.Background(), []int{1, 2}) + require.Error(t, err) + require.Empty(t, results) +} + +func TestLoadRuntimeStatesStrictHonorsDeadline(t *testing.T) { + oldClient, oldEnabled := maxcommon.RDB, maxcommon.RedisEnabled + clientConn, serverConn := net.Pipe() + defer serverConn.Close() + go func() { _, _ = io.Copy(io.Discard, serverConn) }() + client := redis.NewClient(&redis.Options{ + Dialer: func(context.Context, string, string) (net.Conn, error) { return clientConn, nil }, + MaxRetries: -1, + ReadTimeout: time.Second, + }) + maxcommon.RDB, maxcommon.RedisEnabled = client, true + t.Cleanup(func() { + _ = client.Close() + maxcommon.RDB, maxcommon.RedisEnabled = oldClient, oldEnabled + }) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + started := time.Now() + results, err := LoadRuntimeStatesStrict(ctx, []int{1}) + require.Error(t, err) + require.Empty(t, results) + require.Less(t, time.Since(started), time.Second) +} diff --git a/service/channel_routing_health_view.go b/service/channel_routing_health_view.go new file mode 100644 index 00000000..52b44c80 --- /dev/null +++ b/service/channel_routing_health_view.go @@ -0,0 +1,174 @@ +package service + +import ( + "context" + "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" +) + +const ( + channelRoutingHealthReadTimeout = 100 * time.Millisecond + channelRoutingHealthBatchSize = 200 + channelRoutingHealthMaxChannels = 1000 +) + +// ChannelRoutingHealthView preserves the existing sanitized channel response. +// The additional snapshot never enters the persistence model or routing cache. +type ChannelRoutingHealthView struct { + *model.Channel + RoutingHealth dto.ChannelRoutingHealth `json:"routing_health"` +} + +// BuildChannelRoutingHealthViews enriches only channels already selected by a +// management query. It does not query SQL, mutate channels, or publish transitions. +func BuildChannelRoutingHealthViews(ctx context.Context, channels []*model.Channel) []*ChannelRoutingHealthView { + common.OptionMapRWMutex.RLock() + setting := *operation_setting.GetMonitorSetting() + common.OptionMapRWMutex.RUnlock() + if ctx == nil { + ctx = context.Background() + } + ctx, cancel := context.WithTimeout(ctx, channelRoutingHealthReadTimeout) + defer cancel() + return buildChannelRoutingHealthViews(ctx, channels, setting, channelhealth.LoadRuntimeStatesStrict) +} + +type channelRoutingHealthReader func(context.Context, []int) (map[int]channelhealth.RuntimeStateReadResult, error) + +func buildChannelRoutingHealthViews(ctx context.Context, channels []*model.Channel, setting operation_setting.MonitorSetting, read channelRoutingHealthReader) []*ChannelRoutingHealthView { + views := make([]*ChannelRoutingHealthView, len(channels)) + enabled := setting.AutoPriorityDemotionEnabled || setting.TimeoutAutoDisableEnabled + ids := make([]int, 0, min(len(channels), channelRoutingHealthMaxChannels)) + selected := make(map[int]bool) + if enabled { + for _, channel := range channels { + if channel == nil || channel.Id <= 0 || selected[channel.Id] || len(ids) >= channelRoutingHealthMaxChannels { + continue + } + selected[channel.Id] = true + ids = append(ids, channel.Id) + } + } + states := make(map[int]channelhealth.RuntimeStateReadResult, len(ids)) + failureReason := "projection_budget_exceeded" + if enabled && (!common.RedisEnabled || common.RDB == nil) { + failureReason = "redis_unconfigured" + } else { + for start := 0; start < len(ids); start += channelRoutingHealthBatchSize { + if ctx.Err() != nil { + break + } + end := min(start+channelRoutingHealthBatchSize, len(ids)) + loaded, err := read(ctx, ids[start:end]) + if err != nil { + failureReason = "redis_read_failed" + break + } + for _, id := range ids[start:end] { + if state, ok := loaded[id]; ok { + states[id] = state + } + } + } + } + now := time.Now().UTC() + for index, channel := range channels { + if channel == nil { + continue + } + view := &ChannelRoutingHealthView{Channel: channel} + view.RoutingHealth = unknownChannelRoutingHealth(channel.GetPriority(), now, failureReason) + switch { + case !enabled: + view.RoutingHealth = dto.ChannelRoutingHealth{ + SchemaVersion: 1, + Status: dto.ChannelRoutingHealthPolicyOff, + Source: dto.ChannelRoutingHealthSourcePolicyOff, + BasePriority: channel.GetPriority(), + EffectivePriority: common.GetPointer(channel.GetPriority()), + ActivePenalty: common.GetPointer(int64(0)), + RuntimeDisabled: common.GetPointer(false), + EvaluatedAt: now, + } + case channel.Id <= 0: + view.RoutingHealth.UnavailableReason = common.GetPointer("invalid_channel") + case !selected[channel.Id]: + view.RoutingHealth.UnavailableReason = common.GetPointer("projection_budget_exceeded") + default: + if result, ok := states[channel.Id]; ok { + if result.Err != nil { + view.RoutingHealth.UnavailableReason = common.GetPointer("redis_invalid_state") + } else { + view.RoutingHealth = projectChannelRoutingHealth(channel.GetPriority(), result.State, setting, now) + } + } + } + views[index] = view + } + return views +} + +func unknownChannelRoutingHealth(base int64, now time.Time, reason string) dto.ChannelRoutingHealth { + return dto.ChannelRoutingHealth{ + SchemaVersion: 1, + Status: dto.ChannelRoutingHealthUnknown, + Source: dto.ChannelRoutingHealthSourceUnavailable, + BasePriority: base, + EvaluatedAt: now, + UnavailableReason: common.GetPointer(reason), + } +} + +func projectChannelRoutingHealth(base int64, state channelhealth.RuntimeState, setting operation_setting.MonitorSetting, now time.Time) dto.ChannelRoutingHealth { + disabled := state.RuntimeDisabled && setting.TimeoutAutoDisableEnabled + view := dto.ChannelRoutingHealth{ + SchemaVersion: 1, + Status: dto.ChannelRoutingHealthNormal, + Source: dto.ChannelRoutingHealthSourceRedis, + BasePriority: base, + StoredPenalty: common.GetPointer(state.Penalty), + ActivePenalty: common.GetPointer(int64(0)), + StoredRuntimeDisabled: common.GetPointer(state.RuntimeDisabled), + RuntimeDisabled: common.GetPointer(disabled), + PenaltyUntil: channelRoutingHealthTime(state.PenaltyUntil), + DisabledUntil: channelRoutingHealthTime(state.DisabledUntil), + LastObservedAt: channelRoutingHealthTime(state.LastObservedAt), + EvaluatedAt: now, + } + disableRecoveryMode := operation_setting.TimeoutAutoDisableRecoveryProbe + if setting.TimeoutAutoDisableRecoveryMode == operation_setting.TimeoutAutoDisableRecoveryManual { + disableRecoveryMode = operation_setting.TimeoutAutoDisableRecoveryManual + } + view.DisableRecoveryMode = &disableRecoveryMode + if state.Penalty > 0 { + recoveryMode := operation_setting.RecoveryModeAutomatic + if state.PenaltyUntil.IsZero() { + recoveryMode = operation_setting.RecoveryModeManual + } + view.PenaltyRecoveryMode = &recoveryMode + } + if disabled { + view.Status = dto.ChannelRoutingHealthRuntimeDisabled + return view + } + view.EffectivePriority = common.GetPointer(channelhealth.EffectivePriority(base, state, now)) + // Mirror both routing paths: an existing penalty can remain effective when + // only auto-disable is enabled; a stored disabled flag suppresses its deduction. + if !state.RuntimeDisabled && state.Penalty > 0 && (state.PenaltyUntil.IsZero() || now.Before(state.PenaltyUntil)) { + view.ActivePenalty = common.GetPointer(state.Penalty) + view.Status = dto.ChannelRoutingHealthPenalized + } + return view +} + +func channelRoutingHealthTime(value time.Time) *time.Time { + if value.IsZero() { + return nil + } + return common.GetPointer(value.UTC()) +} diff --git a/service/channel_routing_health_view_test.go b/service/channel_routing_health_view_test.go new file mode 100644 index 00000000..cd86ab45 --- /dev/null +++ b/service/channel_routing_health_view_test.go @@ -0,0 +1,239 @@ +package service + +import ( + "context" + "errors" + "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/go-redis/redis/v8" + "github.com/stretchr/testify/require" +) + +func withChannelRoutingHealthRedis(t *testing.T) *miniredis.Miniredis { + t.Helper() + server := miniredis.RunT(t) + oldClient, oldEnabled := common.RDB, common.RedisEnabled + oldSetting := *operation_setting.GetMonitorSetting() + client := redis.NewClient(&redis.Options{Addr: server.Addr()}) + common.RDB, common.RedisEnabled = client, true + t.Cleanup(func() { + _ = client.Close() + common.RDB, common.RedisEnabled = oldClient, oldEnabled + *operation_setting.GetMonitorSetting() = oldSetting + }) + return server +} + +func TestBuildChannelRoutingHealthViewsObservesPenaltyWithoutMutations(t *testing.T) { + server := withChannelRoutingHealthRedis(t) + setting := operation_setting.MonitorSetting{ + AutoPriorityDemotionEnabled: true, StreamingFirstResultTimeoutSeconds: 20, + PriorityDeduction: 1, PenaltyCooldownSeconds: 600, RecoveryMode: operation_setting.RecoveryModeAutomatic, + } + *operation_setting.GetMonitorSetting() = setting + observed := time.Now().UTC() + result, err := channelhealth.EvaluateTimeoutGuard(context.Background(), channelhealth.AttemptEvidence{ + ChannelID: 1086, AttemptID: "display-integration", RequestMode: channelhealth.RequestModeStreaming, + TimeoutKind: channelhealth.TimeoutKindStreamingFirstResult, TimeoutEligible: true, ObservedAt: observed, + }, &setting) + require.NoError(t, err) + require.Equal(t, "priority_demotion_firing", result.Transition) + before := server.Dump() + base := int64(10) + channel := &model.Channel{Id: 1086, Priority: &base} + views := BuildChannelRoutingHealthViews(context.Background(), []*model.Channel{channel, {Id: 2}, nil}) + require.Len(t, views, 3) + require.Same(t, channel, views[0].Channel) + require.Equal(t, int64(10), channel.GetPriority()) + require.Equal(t, before, server.Dump()) + health := views[0].RoutingHealth + require.Equal(t, dto.ChannelRoutingHealthPenalized, health.Status) + require.EqualValues(t, 9, *health.EffectivePriority) + require.EqualValues(t, 1, *health.ActivePenalty) + require.Equal(t, result.State.PenaltyUntil, *health.PenaltyUntil) + require.Equal(t, operation_setting.RecoveryModeAutomatic, *health.PenaltyRecoveryMode) + require.Equal(t, dto.ChannelRoutingHealthNormal, views[1].RoutingHealth.Status) + require.EqualValues(t, 0, *views[1].RoutingHealth.EffectivePriority) + require.False(t, *views[1].RoutingHealth.RuntimeDisabled) + require.Equal(t, health.EvaluatedAt, views[1].RoutingHealth.EvaluatedAt) + require.Nil(t, views[2]) + + encoded, err := common.Marshal(views) + require.NoError(t, err) + var decoded []map[string]interface{} + require.NoError(t, common.Unmarshal(encoded, &decoded)) + require.EqualValues(t, 10, decoded[0]["priority"]) + known := decoded[1]["routing_health"].(map[string]interface{}) + require.EqualValues(t, 0, known["effective_priority"]) + require.Equal(t, false, known["runtime_disabled"]) + require.Contains(t, known, "penalty_until") + require.Nil(t, known["penalty_until"]) +} + +func TestChannelRoutingHealthProjectionMatchesExistingRoutingRules(t *testing.T) { + now := time.Now().UTC() + for _, test := range []struct { + name string + base int64 + state channelhealth.RuntimeState + autoBan bool + status string + priority *int64 + active int64 + }{ + {"active", 10, channelhealth.RuntimeState{Penalty: 1, PenaltyUntil: now.Add(time.Minute)}, false, dto.ChannelRoutingHealthPenalized, common.GetPointer(int64(9)), 1}, + {"floor", 0, channelhealth.RuntimeState{Penalty: 1}, false, dto.ChannelRoutingHealthPenalized, common.GetPointer(int64(0)), 1}, + {"negative base", -3, channelhealth.RuntimeState{Penalty: 1}, false, dto.ChannelRoutingHealthPenalized, common.GetPointer(int64(-4)), 1}, + {"expired", 10, channelhealth.RuntimeState{Penalty: 1, PenaltyUntil: now}, false, dto.ChannelRoutingHealthNormal, common.GetPointer(int64(10)), 0}, + {"manual", 10, channelhealth.RuntimeState{Penalty: 1}, false, dto.ChannelRoutingHealthPenalized, common.GetPointer(int64(9)), 1}, + {"disabled waiting", 10, channelhealth.RuntimeState{RuntimeDisabled: true, DisabledUntil: now.Add(time.Minute)}, true, dto.ChannelRoutingHealthRuntimeDisabled, nil, 0}, + {"disabled due", 10, channelhealth.RuntimeState{RuntimeDisabled: true, DisabledUntil: now.Add(-time.Minute)}, true, dto.ChannelRoutingHealthRuntimeDisabled, nil, 0}, + {"stored disable inactive", 10, channelhealth.RuntimeState{RuntimeDisabled: true, Penalty: 1}, false, dto.ChannelRoutingHealthNormal, common.GetPointer(int64(10)), 0}, + } { + t.Run(test.name, func(t *testing.T) { + view := projectChannelRoutingHealth(test.base, test.state, operation_setting.MonitorSetting{ + AutoPriorityDemotionEnabled: true, TimeoutAutoDisableEnabled: test.autoBan, + TimeoutAutoDisableRecoveryMode: operation_setting.TimeoutAutoDisableRecoveryManual, + }, now) + require.Equal(t, test.status, view.Status) + require.Equal(t, test.priority, view.EffectivePriority) + require.Equal(t, test.active, *view.ActivePenalty) + require.Equal(t, test.state.RuntimeDisabled, *view.StoredRuntimeDisabled) + if test.priority != nil { + require.Equal(t, channelhealth.EffectivePriority(test.base, test.state, now), *view.EffectivePriority) + } + if test.name == "manual" { + require.Equal(t, operation_setting.RecoveryModeManual, *view.PenaltyRecoveryMode) + require.Nil(t, view.PenaltyUntil) + } + require.Equal(t, operation_setting.TimeoutAutoDisableRecoveryManual, *view.DisableRecoveryMode) + }) + } +} + +func TestBuildChannelRoutingHealthViewsHonorsAllPolicyCombinations(t *testing.T) { + withChannelRoutingHealthRedis(t) + for _, demotion := range []bool{false, true} { + for _, disable := range []bool{false, true} { + calls := 0 + views := buildChannelRoutingHealthViews(context.Background(), []*model.Channel{{Id: 1, Priority: common.GetPointer(int64(10))}}, + operation_setting.MonitorSetting{AutoPriorityDemotionEnabled: demotion, TimeoutAutoDisableEnabled: disable}, + func(context.Context, []int) (map[int]channelhealth.RuntimeStateReadResult, error) { + calls++ + return map[int]channelhealth.RuntimeStateReadResult{1: {State: channelhealth.RuntimeState{Penalty: 1}}}, nil + }) + health := views[0].RoutingHealth + if !demotion && !disable { + require.Zero(t, calls) + require.Equal(t, dto.ChannelRoutingHealthPolicyOff, health.Status) + require.EqualValues(t, 10, *health.EffectivePriority) + require.Nil(t, health.StoredPenalty) + } else { + require.Equal(t, 1, calls) + require.Equal(t, dto.ChannelRoutingHealthPenalized, health.Status) + require.EqualValues(t, 9, *health.EffectivePriority) + } + } + } +} + +func TestBuildChannelRoutingHealthViewsUnknownIsNotKnownZero(t *testing.T) { + withChannelRoutingHealthRedis(t) + *operation_setting.GetMonitorSetting() = operation_setting.MonitorSetting{AutoPriorityDemotionEnabled: true} + common.RedisEnabled = false + views := BuildChannelRoutingHealthViews(context.Background(), []*model.Channel{{Id: 1}}) + health := views[0].RoutingHealth + require.Equal(t, dto.ChannelRoutingHealthUnknown, health.Status) + require.Equal(t, "redis_unconfigured", *health.UnavailableReason) + encoded, err := common.Marshal(health) + require.NoError(t, err) + var decoded map[string]interface{} + require.NoError(t, common.Unmarshal(encoded, &decoded)) + for _, field := range []string{"effective_priority", "active_penalty", "stored_penalty", "runtime_disabled", "stored_runtime_disabled"} { + require.Contains(t, decoded, field) + require.Nil(t, decoded[field]) + } + common.RedisEnabled = true + for _, test := range []struct { + name string + result map[int]channelhealth.RuntimeStateReadResult + err error + reason string + }{ + {"transport", nil, errors.New("private connection detail"), "redis_read_failed"}, + {"corrupt", map[int]channelhealth.RuntimeStateReadResult{1: {Err: channelhealth.ErrInvalidRuntimeState}}, nil, "redis_invalid_state"}, + } { + t.Run(test.name, func(t *testing.T) { + views := buildChannelRoutingHealthViews(context.Background(), []*model.Channel{{Id: 1}}, *operation_setting.GetMonitorSetting(), + func(context.Context, []int) (map[int]channelhealth.RuntimeStateReadResult, error) { + return test.result, test.err + }) + require.Equal(t, dto.ChannelRoutingHealthUnknown, views[0].RoutingHealth.Status) + require.Equal(t, test.reason, *views[0].RoutingHealth.UnavailableReason) + require.Nil(t, views[0].RoutingHealth.EffectivePriority) + }) + } +} + +func TestBuildChannelRoutingHealthViewsBoundsAndDeduplicatesReads(t *testing.T) { + withChannelRoutingHealthRedis(t) + channels := make([]*model.Channel, 1003) + for index := range channels { + channels[index] = &model.Channel{Id: index + 1} + } + channels[1002] = channels[0] + calls := 0 + seen := make(map[int]bool) + views := buildChannelRoutingHealthViews(context.Background(), channels, operation_setting.MonitorSetting{AutoPriorityDemotionEnabled: true}, + func(_ context.Context, ids []int) (map[int]channelhealth.RuntimeStateReadResult, error) { + calls++ + require.LessOrEqual(t, len(ids), 200) + results := make(map[int]channelhealth.RuntimeStateReadResult) + for _, id := range ids { + require.False(t, seen[id]) + seen[id] = true + results[id] = channelhealth.RuntimeStateReadResult{State: channelhealth.RuntimeState{ChannelID: id}} + } + return results, nil + }) + require.Equal(t, 5, calls) + require.Len(t, seen, 1000) + require.Len(t, views, len(channels)) + require.Equal(t, dto.ChannelRoutingHealthUnknown, views[1000].RoutingHealth.Status) + require.Equal(t, "projection_budget_exceeded", *views[1000].RoutingHealth.UnavailableReason) + require.Equal(t, dto.ChannelRoutingHealthNormal, views[1002].RoutingHealth.Status) +} + +func TestBuildChannelRoutingHealthViewsPreservesSuccessfulBatchesOnTimeout(t *testing.T) { + withChannelRoutingHealthRedis(t) + channels := make([]*model.Channel, 201) + for index := range channels { + channels[index] = &model.Channel{Id: index + 1} + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + calls := 0 + views := buildChannelRoutingHealthViews(ctx, channels, operation_setting.MonitorSetting{AutoPriorityDemotionEnabled: true}, + func(ctx context.Context, ids []int) (map[int]channelhealth.RuntimeStateReadResult, error) { + calls++ + if calls == 1 { + results := make(map[int]channelhealth.RuntimeStateReadResult) + for _, id := range ids { + results[id] = channelhealth.RuntimeStateReadResult{} + } + return results, nil + } + <-ctx.Done() + return nil, ctx.Err() + }) + require.Equal(t, dto.ChannelRoutingHealthNormal, views[0].RoutingHealth.Status) + require.Equal(t, dto.ChannelRoutingHealthUnknown, views[200].RoutingHealth.Status) + require.Nil(t, views[200].RoutingHealth.EffectivePriority) +} diff --git a/web/default/src/features/channels/api.ts b/web/default/src/features/channels/api.ts index 7ea7872e..ce34a923 100644 --- a/web/default/src/features/channels/api.ts +++ b/web/default/src/features/channels/api.ts @@ -95,28 +95,90 @@ export type CodexCredentialRefreshResponse = { * Get paginated list of channels */ export async function getChannels( - params: GetChannelsParams = {} -): Promise { - const res = await api.get('/api/channel', { params }) - return res.data + params: GetChannelsParams = {}, + config: ApiRequestConfig = {} +): Promise { + return observeChannelRead(async () => { + const res = await api.get('/api/channel', { ...config, params }) + return res.data + }) } /** * Search channels with filters */ export async function searchChannels( - params: SearchChannelsParams -): Promise { - const res = await api.get('/api/channel/search', { params }) - return res.data + params: SearchChannelsParams, + config: ApiRequestConfig = {} +): Promise { + return observeChannelRead(async () => { + const res = await api.get('/api/channel/search', { ...config, params }) + return res.data + }) } /** * Get single channel by ID */ -export async function getChannel(id: number): Promise { - const res = await api.get(`/api/channel/${id}`) - return res.data +export async function getChannel( + id: number, + config: ApiRequestConfig = {} +): Promise { + return observeChannelRead(async () => { + const res = await api.get(`/api/channel/${id}`, config) + return res.data + }) +} + +// Browser-local metadata, never sent to the API or used as a distributed version. +export type ChannelReadObservation = { + requestOrder: number + receivedAt: number +} +let channelReadOrder = 0 +async function observeChannelRead(read: () => Promise) { + const requestOrder = ++channelReadOrder + const response = await read() + return { + ...response, + observation: { requestOrder, receivedAt: performance.now() }, + } +} + +// Keep the polling cache limited to the saved routing projection. Independent +// cancellation must not cancel a concurrent configuration-details request. +export type ChannelRoutingSnapshot = Pick< + Channel, + 'id' | 'priority' | 'routing_health' +> & { observation: ChannelReadObservation } + +export function toChannelRoutingSnapshot( + channel: Channel, + observation: ChannelReadObservation +): ChannelRoutingSnapshot { + return { + id: channel.id, + priority: channel.priority, + routing_health: channel.routing_health, + observation, + } +} + +export async function getChannelRoutingSnapshot( + id: number, + signal?: AbortSignal +): Promise { + const response = await getChannel(id, { + signal, + disableDuplicate: true, + timeout: 8_000, + skipBusinessError: true, + skipErrorHandler: true, + }) + if (!response.success || !response.data || response.data.id !== id) { + throw new Error('Channel routing snapshot unavailable') + } + return toChannelRoutingSnapshot(response.data, response.observation) } /** diff --git a/web/default/src/features/channels/components/channel-health-refresh.runtime.test.tsx b/web/default/src/features/channels/components/channel-health-refresh.runtime.test.tsx new file mode 100644 index 00000000..984a0209 --- /dev/null +++ b/web/default/src/features/channels/components/channel-health-refresh.runtime.test.tsx @@ -0,0 +1,1115 @@ +/* +Copyright (C) 2023-2026 MAX-API-Next + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as +published by the Free Software Foundation, either version 3 of the +License, or (at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . + +For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues +*/ +import { act, useState, type ReactElement } from 'react' +import { QueryClient, QueryClientProvider } from '@tanstack/react-query' +import { + createMemoryHistory, + createRootRoute, + createRoute, + createRouter, + RouterProvider, +} from '@tanstack/react-router' +import { createReactTestEnvironment } from '@/test/react' +import assert from 'node:assert/strict' +import { after, test } from 'node:test' +import { api } from '@/lib/api' +import type { ChannelRoutingHealth } from '../lib/channel-routing-health' +import { channelSchema, type Channel } from '../types' + +const env = createReactTestEnvironment() +await env.setup() +const globals = new Map() +for (const [key, value] of Object.entries({ + localStorage: env.dom.window.localStorage, + matchMedia: env.dom.window.matchMedia.bind(env.dom.window), + customElements: env.dom.window.customElements, + scrollTo: () => {}, +})) { + globals.set(key, Object.getOwnPropertyDescriptor(globalThis, key)) + Object.defineProperty(globalThis, key, { configurable: true, value }) +} +Object.defineProperty(env.dom.window.Element.prototype, 'getAnimations', { + configurable: true, + value: () => [], +}) +Object.defineProperty(env.dom.window.HTMLElement.prototype, 'scrollIntoView', { + configurable: true, + value: () => {}, +}) +Object.defineProperty(env.dom.window.document, 'visibilityState', { + configurable: true, + value: 'visible', +}) +const { within, waitFor, fireEvent } = + await import('@testing-library/react/pure') +const { ChannelRoutingHealthCard } = + await import('./channel-routing-health-card') +const { ChannelRoutingHealthCell } = await import('./channel-routing-health') +const { ChannelHealthSnapshotContext } = + await import('./channel-health-snapshot-context') +const { ChannelMutateDrawer } = await import('./drawers/channel-mutate-drawer') +const { ChannelsProvider, useChannels } = await import('./channels-provider') +const { ChannelsTable } = await import('./channels-table') +const { useChannelHealthRefresh, CHANNEL_HEALTH_REFRESH_MS } = + await import('../hooks/use-channel-health-refresh') +const { channelsQueryKeys, invalidateChannelConfiguration } = + await import('../lib/channel-actions') +const { getChannel } = await import('../api') + +after(() => { + env.teardown() + for (const [key, descriptor] of globals) { + if (descriptor) Object.defineProperty(globalThis, key, descriptor) + else Reflect.deleteProperty(globalThis, key) + } +}) + +function health( + overrides: Partial = {} +): ChannelRoutingHealth { + return { + schema_version: 1, + status: 'penalized', + source: 'redis', + base_priority: 10, + effective_priority: 9, + stored_penalty: 1, + active_penalty: 1, + stored_runtime_disabled: false, + runtime_disabled: false, + penalty_until: null, + disabled_until: null, + penalty_recovery_mode: 'manual', + disable_recovery_mode: 'probe_then_automatic', + last_observed_at: null, + evaluated_at: new Date(Date.now()).toISOString(), + unavailable_reason: null, + ...overrides, + } +} +function channel( + id: number, + snapshot: unknown = health(), + priority = 10 +): Channel { + return channelSchema.parse({ + id, + priority, + routing_health: snapshot, + type: 1, + key: '', + status: 1, + name: `Synthetic channel ${id}`, + models: 'synthetic-model', + group: 'default', + created_time: 0, + test_time: 0, + response_time: 0, + balance_updated_time: 0, + }) +} +function client() { + return new QueryClient({ + defaultOptions: { queries: { retry: false, gcTime: 0 } }, + }) +} +function observation(requestOrder = 0) { + return { requestOrder, receivedAt: performance.now() } +} +async function renderPage(content: ReactElement, queryClient: QueryClient) { + const root = createRootRoute() + const auth = createRoute({ getParentRoute: () => root, id: '_authenticated' }) + const page = createRoute({ + getParentRoute: () => auth, + path: 'channels/', + component: () => content, + validateSearch: (search) => search, + }) + const router = createRouter({ + routeTree: root.addChildren([auth.addChildren([page])]), + history: createMemoryHistory({ initialEntries: ['/channels/'] }), + }) + await router.load() + const view = await env.render( + + + + + + ) + return { ...view, router } +} + +test( + 'runtime polling updates the real editor card while draft and detail invalidations remain isolated', + // This exercises the complete drawer. Full form interactions can exceed 20s + // on Windows; keep the budget local to this test. + { timeout: 60_000 }, + async () => { + const queryClient = client(), + adapter = api.defaults.adapter + let savedPriority = 10, + penalty = 1, + reads = 0 + const writes: Record[] = [] + api.defaults.adapter = async (config) => { + let data: unknown = [] + if (config.method === 'put') writes.push(JSON.parse(String(config.data))) + if (config.url === '/api/channel/7') { + reads += 1 + data = channel( + 7, + health({ + base_priority: savedPriority, + effective_priority: savedPriority - penalty, + active_penalty: penalty, + stored_penalty: penalty, + }), + savedPriority + ) + } + return { + config, + status: 200, + statusText: 'OK', + headers: {}, + data: { success: true, data }, + } + } + function Editor() { + const [open, setOpen] = useState(true) + return ( + + ) + } + const view = await renderPage(, queryClient) + const body = within(env.dom.window.document.body) + try { + await waitFor(() => + assert.ok(body.queryByLabelText('Configured priority')) + ) + const input = body.getByLabelText( + 'Configured priority' + ) as HTMLInputElement + await waitFor(() => assert.equal(input.value, '10')) + const card = within(body.getByLabelText('Routing state snapshot')) + assert.equal( + reads, + 1, + 'fresh details seed the first runtime snapshot without a duplicate GET' + ) + await act(async () => + fireEvent.change(input, { target: { value: '20' } }) + ) + penalty = 2 + await view.click( + card.getByRole('button', { name: 'Refresh runtime state' }) + ) + await waitFor(() => assert.ok(card.queryByText('Timeout penalty: 2'))) + assert.equal(input.value, '20') + savedPriority = 11 + await act(async () => { + await invalidateChannelConfiguration(queryClient, 7) + }) + await waitFor(() => + assert.ok(card.queryByText('Configured priority: 11')) + ) + assert.equal( + input.value, + '20', + 'even detail-cache invalidation preserves the draft' + ) + const cached = queryClient.getQueryData>( + channelsQueryKeys.runtime(7) + )! + assert.deepEqual(Object.keys(cached).sort(), [ + 'id', + 'observation', + 'priority', + 'routing_health', + ]) + await view.click(body.getByText('Update Channel', { selector: 'button' })) + await waitFor(() => assert.equal(writes.length, 1)) + assert.equal(writes[0].priority, 20) + assert.equal('routing_health' in writes[0], false) + assert.equal('observation' in writes[0], false) + } finally { + await view.unmount() + queryClient.clear() + api.defaults.adapter = adapter + } + } +) + +test( + 'cached configuration stays blocked until a successful read and new channels have no runtime card', + { timeout: 20_000 }, + async () => { + const queryClient = client(), + adapter = api.defaults.adapter + queryClient.setQueryData( + channelsQueryKeys.detail(7), + { success: true, data: channel(7) }, + { updatedAt: Date.now() - 60_000 } + ) + let failure = true + api.defaults.adapter = async (config) => { + const isDetails = config.url === '/api/channel/7' + return { + config, + status: 200, + statusText: 'OK', + headers: {}, + data: + isDetails && failure + ? { success: false } + : { + success: true, + data: isDetails + ? channel( + 7, + health({ base_priority: 11, effective_priority: 10 }), + 11 + ) + : [], + }, + } + } + function Editor() { + const [row, setRow] = useState(channel(7)) + return ( + <> + + {}} currentRow={row} /> + + ) + } + const view = await renderPage(, queryClient), + body = within(document.body) + try { + await waitFor(() => + assert.ok( + body.queryByText( + 'Unable to load channel configuration. Retry before editing.' + ) + ) + ) + assert.equal(body.queryByLabelText('Configured priority'), null) + assert.equal( + ( + body.getByRole('button', { + name: 'Update Channel', + }) as HTMLButtonElement + ).disabled, + true + ) + failure = false + await view.click(body.getByRole('button', { name: 'Retry' })) + await waitFor(() => + assert.equal( + (body.getByLabelText('Configured priority') as HTMLInputElement) + .value, + '11' + ) + ) + await view.click( + within(view.container).getByText('Create another channel') + ) + await waitFor(() => + assert.equal( + body.queryByRole('region', { name: 'Routing state snapshot' }), + null + ) + ) + } finally { + await view.unmount() + queryClient.clear() + api.defaults.adapter = adapter + } + } +) + +test('failed runtime refresh preserves the old value and a later retry obtains a new snapshot', async () => { + const queryClient = client(), + adapter = api.defaults.adapter + let failure = true + api.defaults.adapter = async (config) => ({ + config, + status: 200, + statusText: 'OK', + headers: {}, + data: failure + ? { success: false } + : { + success: true, + data: channel( + 7, + health({ + status: 'normal', + effective_priority: 10, + active_penalty: 0, + stored_penalty: 0, + penalty_recovery_mode: null, + }) + ), + }, + }) + const view = await renderPage( + , + queryClient + ) + const fields = within(view.container) + try { + await view.click( + fields.getByRole('button', { name: 'Refresh runtime state' }) + ) + await waitFor(() => + assert.ok( + fields.queryByText( + 'Runtime refresh failed. The previous snapshot may be outdated.' + ) + ) + ) + assert.ok(fields.getByText('9', { exact: true })) + failure = false + await view.click( + fields.getByRole('button', { name: 'Refresh runtime state' }) + ) + await waitFor(() => assert.ok(fields.queryByText('No runtime restriction'))) + assert.equal( + fields.queryByText( + 'Runtime refresh failed. The previous snapshot may be outdated.' + ), + null + ) + } finally { + await view.unmount() + queryClient.clear() + api.defaults.adapter = adapter + } +}) + +test('late configuration responses cannot replace a newer runtime projection', async () => { + const queryClient = client() + const timestamp = Date.now() + queryClient.setQueryData( + channelsQueryKeys.runtime(7), + { + id: 7, + priority: 20, + routing_health: health({ + base_priority: 20, + effective_priority: 19, + evaluated_at: new Date(timestamp).toISOString(), + }), + observation: observation(1), + }, + { updatedAt: timestamp - 100 } + ) + const view = await renderPage( + , + queryClient + ) + try { + assert.ok(within(view.container).getByText('Configured priority: 20')) + assert.ok(within(view.container).getByText('19', { exact: true })) + } finally { + await view.unmount() + queryClient.clear() + } +}) + +test( + 'a late polling response cannot undo accepted details, but a subsequent read can update them', + { timeout: 20_000 }, + async () => { + const queryClient = client(), + adapter = api.defaults.adapter + let finishOld: (() => void) | undefined + let reads = 0 + api.defaults.adapter = async (config) => { + reads += 1 + const read = reads + if (read === 1) + await new Promise((resolve) => { + finishOld = resolve + }) + return { + config, + status: 200, + statusText: 'OK', + headers: {}, + data: { + success: true, + data: channel( + 7, + health({ + effective_priority: read === 2 ? 8 : 9, + active_penalty: read === 2 ? 2 : 1, + stored_penalty: read === 2 ? 2 : 1, + }) + ), + }, + } + } + function Card() { + const [details, setDetails] = useState({ + data: channel(7), + observation: { requestOrder: 0, receivedAt: performance.now() }, + }) + return ( + <> + + + + ) + } + const view = await renderPage(, queryClient), + fields = within(view.container) + try { + await view.click( + fields.getByRole('button', { name: 'Refresh runtime state' }) + ) + await waitFor(() => assert.ok(finishOld)) + await view.click(fields.getByText('Receive newer details')) + await waitFor(() => assert.ok(fields.queryByText('8', { exact: true }))) + const accepted = queryClient.getQueryData<{ observation: unknown }>( + channelsQueryKeys.runtime(7) + )!.observation + await act(async () => { + finishOld!() + }) + await waitFor(() => assert.equal(queryClient.isFetching(), 0)) + assert.ok( + fields.getByText('8', { exact: true }), + 'old request must not restore 9' + ) + assert.equal(fields.queryByText('9', { exact: true }), null) + assert.deepEqual( + queryClient.getQueryData<{ observation: unknown }>( + channelsQueryKeys.runtime(7) + )!.observation, + accepted, + 'a rejected response must not reset observation age' + ) + await view.click( + fields.getByRole('button', { name: 'Refresh runtime state' }) + ) + await waitFor(() => assert.ok(fields.queryByText('9', { exact: true }))) + } finally { + finishOld?.() + await view.unmount() + queryClient.clear() + api.defaults.adapter = adapter + } + } +) + +test( + 'node clocks in either direction do not override local request order', + { timeout: 20_000 }, + async () => { + for (const skew of [-120_000, 120_000]) { + const queryClient = client(), + timestamp = Date.now() + queryClient.setQueryData(channelsQueryKeys.runtime(7), { + id: 7, + priority: 10, + routing_health: health({ + evaluated_at: new Date(timestamp + skew).toISOString(), + }), + observation: { requestOrder: 0, receivedAt: performance.now() }, + }) + const view = await renderPage( + , + queryClient + ) + try { + assert.equal( + queryClient.getQueryData<{ routing_health: ChannelRoutingHealth }>( + channelsQueryKeys.runtime(7) + )!.routing_health.effective_priority, + 8 + ) + // Cache notifications are batched; wait for their visible result. + await waitFor(() => + assert.ok(within(view.container).queryByText('8', { exact: true })) + ) + assert.equal( + within(view.container).queryByText('9', { exact: true }), + null + ) + } finally { + await view.unmount() + queryClient.clear() + } + } + } +) + +test('snapshot age and recovery hints use elapsed observation time despite node clock skew', async () => { + for (const skew of [-120_000, 120_000]) { + const evaluatedAt = Date.now() + skew + const source = channel( + 7, + health({ + penalty_recovery_mode: 'automatic', + evaluated_at: new Date(evaluatedAt).toISOString(), + penalty_until: new Date(evaluatedAt + 30_000).toISOString(), + }) + ) + function Snapshot() { + const [elapsed, setElapsed] = useState(0) + return ( + <> + + + + + + ) + } + const view = await env.render(), + fields = within(view.container) + try { + assert.equal( + fields.queryByText('Runtime snapshot may be outdated.'), + null + ) + assert.equal( + fields.queryByText('Waiting for a fresh runtime snapshot'), + null + ) + await view.click(fields.getByText('Advance elapsed time')) + assert.ok(fields.getByText('Runtime snapshot may be outdated.')) + assert.ok(fields.getByText('Waiting for a fresh runtime snapshot')) + assert.ok( + fields.getByText('9', { exact: true }), + 'elapsed time must not restore priority' + ) + } finally { + await view.unmount() + } + } +}) + +test('deadline scheduling uses server-relative duration and ignores early timer callbacks for either node clock skew', async () => { + const originalNow = Object.getOwnPropertyDescriptor(performance, 'now') + const originalTimeout = globalThis.setTimeout + const delays: number[] = [] + const timers: Array<{ callback: () => void; delay: number }> = [] + let timestamp = 1000 + Object.defineProperty(performance, 'now', { + configurable: true, + value: () => timestamp, + }) + globalThis.setTimeout = ((callback: () => void, delay?: number) => { + delays.push(delay ?? 0) + if (delay === 30_000 || delay === 1) { + timers.push({ callback, delay }) + // Drive deadline callbacks explicitly; React's zero-delay tasks still run. + return originalTimeout(() => {}, 120_000) + } + return originalTimeout(callback, delay) + }) as typeof setTimeout + try { + for (const skew of [-120_000, 120_000]) { + delays.length = 0 + timers.length = 0 + timestamp = 1000 + const evaluatedAt = Date.now() + skew + const snapshots = [ + channel( + 7, + health({ + penalty_recovery_mode: 'automatic', + evaluated_at: new Date(evaluatedAt).toISOString(), + penalty_until: new Date(evaluatedAt + 30_000).toISOString(), + }) + ), + ] + let calls = 0 + const refetch = async () => { + calls += 1 + } + function Clock() { + useChannelHealthRefresh({ snapshots, receivedAt: 1000, refetch }) + return null + } + const view = await env.render() + try { + assert.ok( + delays.includes(30_000), + '30 seconds on the server must schedule 30 seconds locally' + ) + assert.equal( + calls, + 0, + 'a slow server clock must not trigger immediate expiry' + ) + timestamp = 30_999.5 + await act(async () => timers[0].callback()) + assert.equal( + calls, + 0, + 'an early timer cannot request expiry confirmation' + ) + assert.equal( + timers.at(-1)!.delay, + 1, + 'remaining fractional milliseconds round up' + ) + timestamp = 31_000 + await act(async () => timers.at(-1)!.callback()) + assert.equal(calls, 1, 'a reached deadline requests one confirmation') + } finally { + await view.unmount() + } + } + } finally { + globalThis.setTimeout = originalTimeout + if (originalNow) Object.defineProperty(performance, 'now', originalNow) + else Reflect.deleteProperty(performance, 'now') + } +}) + +test( + 'switching channels cancels the previous runtime request and closing stops foreground refresh', + { timeout: 20_000 }, + async () => { + const queryClient = client(), + adapter = api.defaults.adapter + let completeOld: (() => void) | undefined + let previousSignal: { readonly aborted: boolean } | undefined + let reads = 0 + api.defaults.adapter = async (config) => { + reads += 1 + previousSignal = config.signal as AbortSignal + await new Promise((resolve) => { + completeOld = resolve + }) + return { + config, + status: 200, + statusText: 'OK', + headers: {}, + data: { + success: true, + data: channel(7, health({ effective_priority: -77 })), + }, + } + } + function Cards() { + const [id, setId] = useState(7), + [open, setOpen] = useState(true) + return ( + <> + + + {open && ( + + )} + + ) + } + const view = await renderPage(, queryClient), + fields = within(view.container) + try { + await act(async () => { + void queryClient.refetchQueries({ + queryKey: channelsQueryKeys.runtime(7), + }) + }) + await waitFor(() => assert.ok(completeOld)) + await view.click(fields.getByText('Next channel')) + assert.equal(previousSignal?.aborted, true) + await act(async () => { + completeOld!() + }) + assert.ok(fields.getByText('Configured priority: 11')) + assert.equal(fields.queryByText('-77', { exact: true }), null) + await view.click(fields.getByText('Close card')) + await act(async () => { + window.dispatchEvent(new Event('focus')) + }) + assert.equal(reads, 1) + assert.equal( + fields.queryByRole('region', { name: 'Routing state snapshot' }), + null + ) + } finally { + completeOld?.() + await view.unmount() + queryClient.clear() + api.defaults.adapter = adapter + } + } +) + +test('visible polling pauses for background and editing, resumes on focus, and stops after unmount', async () => { + const originalInterval = globalThis.setInterval, + originalNow = Object.getOwnPropertyDescriptor(performance, 'now') + const callbacks: Array<() => void> = [] + let timestamp = performance.now(), + calls = 0 + Object.defineProperty(performance, 'now', { + configurable: true, + value: () => timestamp, + }) + globalThis.setInterval = ((callback: () => void, delay?: number) => { + if (delay === CHANNEL_HEALTH_REFRESH_MS) callbacks.push(callback) + return originalInterval(callback, delay) + }) as typeof setInterval + const refetch = async () => { + calls += 1 + } + const snapshots = [channel(7)] + function Clock() { + const [paused, setPaused] = useState(false) + useChannelHealthRefresh({ snapshots, paused, refetch }) + return ( + + ) + } + const view = await env.render() + const visible = (value: string) => + Object.defineProperty(document, 'visibilityState', { + configurable: true, + value, + }) + try { + await act(async () => { + timestamp += 30_000 + callbacks.at(-1)!() + }) + assert.equal(calls, 1) + visible('hidden') + await act(async () => { + timestamp += 30_000 + callbacks.at(-1)!() + document.dispatchEvent(new Event('visibilitychange')) + }) + assert.equal(calls, 1) + visible('visible') + await act(async () => { + window.dispatchEvent(new Event('focus')) + document.dispatchEvent(new Event('visibilitychange')) + }) + assert.equal(calls, 2, 'one foreground transition is deduplicated') + await view.click(within(view.container).getByText('Toggle editing')) + await act(async () => { + timestamp += 30_000 + window.dispatchEvent(new Event('focus')) + }) + assert.equal(calls, 2) + await view.click(within(view.container).getByText('Toggle editing')) + assert.equal(calls, 3) + await view.unmount() + await act(async () => { + timestamp += 30_000 + window.dispatchEvent(new Event('focus')) + }) + assert.equal(calls, 3) + } finally { + await view.unmount() + visible('visible') + globalThis.setInterval = originalInterval + if (originalNow) Object.defineProperty(performance, 'now', originalNow) + else Reflect.deleteProperty(performance, 'now') + } +}) + +test('a deadline triggers one refresh per snapshot and retains the penalized value until server confirmation', async () => { + let calls = 0 + const receivedAt = performance.now() + const snapshots = [ + channel( + 7, + health({ + penalty_recovery_mode: 'automatic', + penalty_until: new Date(Date.now() + 200).toISOString(), + }) + ), + ] + const refetch = async () => { + calls += 1 + throw new Error('synthetic read failure') + } + function Deadline() { + const now = useChannelHealthRefresh({ snapshots, receivedAt, refetch }) + return ( + + + + ) + } + const view = await env.render(), + fields = within(view.container) + try { + await act(async () => window.dispatchEvent(new Event('focus'))) + assert.equal(calls, 1) + await waitFor(() => assert.equal(calls, 2)) + assert.ok(fields.getByText('Waiting for a fresh runtime snapshot')) + assert.ok(fields.getByText('9', { exact: true })) + await new Promise((resolve) => setTimeout(resolve, 30)) + assert.equal( + calls, + 2, + 'foreground deduplication must not suppress the later deadline refresh' + ) + } finally { + await view.unmount() + } +}) + +test('list and search share refresh protection, hide placeholder health, and retain failed-refresh data', async () => { + const queryClient = client(), + adapter = api.defaults.adapter + let reads = 0, + fail = false + let finishSearch: (() => void) | undefined + api.defaults.adapter = async (config) => { + let data: unknown = [] + if (config.url === '/api/channel' || config.url === '/api/channel/search') { + reads += 1 + if (config.url.endsWith('/search')) + await new Promise((resolve) => { + finishSearch = resolve + }) + data = { items: [channel(7)], total: 1, type_counts: { '1': 1 } } + if (fail) + return { + config, + status: 200, + statusText: 'OK', + headers: {}, + data: { success: false }, + } + } + return { + config, + status: 200, + statusText: 'OK', + headers: {}, + data: { success: true, data }, + } + } + function List() { + const { refreshPaused } = useChannels() + return ( + <> +

{refreshPaused ? 'Refresh paused' : 'Refresh active'}

+ + + ) + } + const view = await renderPage(, queryClient), + fields = within(view.container) + try { + await waitFor(() => assert.ok(fields.queryByText('Timeout penalty: 1'))) + await view.click(fields.getByRole('button', { name: '10' })) + await waitFor(() => assert.ok(fields.queryByText('Refresh paused'))) + await act(async () => { + await queryClient.invalidateQueries({ + queryKey: channelsQueryKeys.lists(), + }) + }) + assert.equal(reads, 1) + const input = view.container.querySelector( + 'input.font-mono' + ) as HTMLInputElement + await act(async () => { + fireEvent.change(input, { target: { value: '20' } }) + fireEvent.keyDown(input, { key: 'Escape' }) + }) + await waitFor(() => assert.ok(fields.queryByText('Refresh active'))) + fail = true + await act(async () => { + await queryClient.refetchQueries({ queryKey: channelsQueryKeys.lists() }) + }) + await waitFor(() => + assert.ok( + fields.queryByText( + 'Channel refresh failed. Displayed results may be outdated.' + ) + ) + ) + assert.ok(fields.getByText('Timeout penalty: 1')) + fail = false + await act(async () => { + void view.router.navigate({ + to: '/channels', + search: { filter: 'synthetic' }, + }) + }) + await waitFor(() => assert.ok(finishSearch)) + assert.equal(fields.queryByText('Timeout penalty: 1'), null) + assert.ok( + fields.getAllByText( + 'Waiting for the current results. Previous runtime values are hidden.' + ).length > 0 + ) + await act(async () => { + finishSearch!() + }) + await waitFor(() => assert.ok(fields.queryByText('Timeout penalty: 1'))) + } finally { + finishSearch?.() + await view.unmount() + queryClient.clear() + api.defaults.adapter = adapter + } +}) + +test( + 'tag priority confirmation keeps its pending value and pauses refresh until cancellation', + { timeout: 20_000 }, + async () => { + const queryClient = client(), + adapter = api.defaults.adapter + const previousTagMode = localStorage.getItem('enable-tag-mode') + localStorage.setItem('enable-tag-mode', 'true') + let reads = 0, + writes = 0 + api.defaults.adapter = async (config) => { + if (config.method !== 'get') writes += 1 + let data: unknown = [] + if (config.url === '/api/channel') { + reads += 1 + data = { + items: [ + { ...channel(7), tag: 'shared' }, + { ...channel(8), tag: 'shared' }, + ], + total: 2, + type_counts: { '1': 2 }, + } + } + return { + config, + status: 200, + statusText: 'OK', + headers: {}, + data: { success: true, data }, + } + } + const view = await renderPage(, queryClient), + fields = within(view.container), + body = within(document.body) + try { + await waitFor(() => + assert.ok(fields.queryByRole('button', { name: '10' })) + ) + await view.click(fields.getByRole('button', { name: '10' })) + const input = view.container.querySelector( + 'input.font-mono' + ) as HTMLInputElement + await act(async () => { + fireEvent.change(input, { target: { value: '20' } }) + fireEvent.keyDown(input, { key: 'Enter' }) + }) + await waitFor(() => assert.ok(body.queryByRole('alertdialog'))) + await act(async () => { + await queryClient.invalidateQueries({ + queryKey: channelsQueryKeys.lists(), + }) + window.dispatchEvent(new Event('focus')) + }) + assert.equal(reads, 1) + assert.ok( + body.getByText( + 'This will update the priority to 20 for all 2 channel(s) with tag "shared". Continue?' + ) + ) + await view.click(body.getByRole('button', { name: 'Cancel' })) + await waitFor(() => assert.ok(reads >= 2)) + assert.equal(writes, 0) + } finally { + await view.unmount() + queryClient.clear() + api.defaults.adapter = adapter + if (previousTagMode === null) localStorage.removeItem('enable-tag-mode') + else localStorage.setItem('enable-tag-mode', previousTagMode) + } + } +) diff --git a/web/default/src/features/channels/components/channel-health-snapshot-context.ts b/web/default/src/features/channels/components/channel-health-snapshot-context.ts new file mode 100644 index 00000000..07c33edf --- /dev/null +++ b/web/default/src/features/channels/components/channel-health-snapshot-context.ts @@ -0,0 +1,25 @@ +/* +Copyright (C) 2023-2026 MAX-API-Next + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as +published by the Free Software Foundation, either version 3 of the +License, or (at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . + +For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues +*/ +import { createContext } from 'react' + +export const ChannelHealthSnapshotContext = createContext<{ + now: number | null + receivedAt?: number + placeholder: boolean +}>({ now: null, placeholder: false }) diff --git a/web/default/src/features/channels/components/channel-routing-health-card.tsx b/web/default/src/features/channels/components/channel-routing-health-card.tsx new file mode 100644 index 00000000..27dc6141 --- /dev/null +++ b/web/default/src/features/channels/components/channel-routing-health-card.tsx @@ -0,0 +1,145 @@ +/* +Copyright (C) 2023-2026 MAX-API-Next + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as +published by the Free Software Foundation, either version 3 of the +License, or (at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . + +For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues +*/ +import { useCallback, useMemo, useEffect } from 'react' +import { useQuery, useQueryClient } from '@tanstack/react-query' +import { Link } from '@tanstack/react-router' +import { useTranslation } from 'react-i18next' +import { Button } from '@/components/ui/button' +import { + getChannelRoutingSnapshot, + toChannelRoutingSnapshot, + type ChannelReadObservation, + type ChannelRoutingSnapshot, +} from '../api' +import { + useChannelHealthRefresh, + CHANNEL_HEALTH_REFRESH_MS, +} from '../hooks/use-channel-health-refresh' +import { channelsQueryKeys } from '../lib/channel-actions' +import type { Channel } from '../types' +import { ChannelHealthSnapshotContext } from './channel-health-snapshot-context' +import { ChannelRoutingHealthDetails } from './channel-routing-health' + +// React Query uses this for BOTH setQueryData (details) and query completion +// (polling). Ordering only at the seeding effect leaves the reverse race open. +function acceptSnapshot(previous: unknown, incoming: unknown) { + const current = previous as ChannelRoutingSnapshot | undefined + const next = incoming as ChannelRoutingSnapshot + if ( + current?.id === next.id && + current.observation?.requestOrder >= next.observation.requestOrder + ) + return current + return next +} + +export function ChannelRoutingHealthCard(props: { + initialChannel: Channel + initialUpdatedAt: number + initialObservation: ChannelReadObservation +}) { + const { t } = useTranslation() + const id = props.initialChannel.id + const queryClient = useQueryClient() + const query = useQuery({ + queryKey: channelsQueryKeys.runtime(id), + queryFn: ({ signal }) => getChannelRoutingSnapshot(id, signal), + initialData: () => + toChannelRoutingSnapshot(props.initialChannel, props.initialObservation), + initialDataUpdatedAt: props.initialUpdatedAt, + structuralSharing: acceptSnapshot, + staleTime: CHANNEL_HEALTH_REFRESH_MS, + refetchOnWindowFocus: false, + refetchOnReconnect: false, + retry: false, + }) + useEffect(() => { + const key = channelsQueryKeys.runtime(id) + const cached = queryClient.getQueryData(key) + const incoming = toChannelRoutingSnapshot( + props.initialChannel, + props.initialObservation + ) + if (acceptSnapshot(cached, incoming) === incoming) { + queryClient.setQueryData(key, incoming, { + updatedAt: props.initialUpdatedAt, + }) + } + }, [ + id, + props.initialChannel, + props.initialObservation, + props.initialUpdatedAt, + queryClient, + ]) + const { refetch } = query + const refresh = useCallback( + () => refetch({ cancelRefetch: false }), + [refetch] + ) + const snapshots = useMemo(() => [query.data], [query.data]) + const receivedAt = query.data.observation.receivedAt + const now = useChannelHealthRefresh({ + snapshots, + receivedAt, + refetch: refresh, + }) + return ( +
+
+

{t('Routing state snapshot')}

+ +
+

+ {t( + 'Based on saved configuration. Unsaved edits are not reflected here.' + )} +

+ {query.isError && ( +

+ {t('Runtime refresh failed. The previous snapshot may be outdated.')} +

+ )} + + + + + {t('Open channel health settings')} + +
+ ) +} diff --git a/web/default/src/features/channels/components/channel-routing-health.runtime.test.tsx b/web/default/src/features/channels/components/channel-routing-health.runtime.test.tsx new file mode 100644 index 00000000..c97c11c9 --- /dev/null +++ b/web/default/src/features/channels/components/channel-routing-health.runtime.test.tsx @@ -0,0 +1,339 @@ +/* +Copyright (C) 2023-2026 MAX-API-Next + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as +published by the Free Software Foundation, either version 3 of the +License, or (at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . + +For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues +*/ +import { QueryClient, QueryClientProvider } from '@tanstack/react-query' +import { getCoreRowModel, useReactTable } from '@tanstack/react-table' +import { createReactTestEnvironment } from '@/test/react' +import assert from 'node:assert/strict' +import { after, test } from 'node:test' +import type { ChannelRoutingHealth } from '../lib/channel-routing-health' +import { channelSchema, type Channel } from '../types' + +const env = createReactTestEnvironment() +await env.setup() +const previousStorage = Object.getOwnPropertyDescriptor( + globalThis, + 'localStorage' +) +const previousMatchMedia = Object.getOwnPropertyDescriptor( + globalThis, + 'matchMedia' +) +const previousCustomElements = Object.getOwnPropertyDescriptor( + globalThis, + 'customElements' +) +Object.defineProperty(globalThis, 'localStorage', { + configurable: true, + value: env.dom.window.localStorage, +}) +Object.defineProperty(globalThis, 'matchMedia', { + configurable: true, + value: env.dom.window.matchMedia.bind(env.dom.window), +}) +Object.defineProperty(globalThis, 'customElements', { + configurable: true, + value: env.dom.window.customElements, +}) +const { within, waitFor } = await import('@testing-library/react/pure') +const { ChannelRoutingHealthCell } = await import('./channel-routing-health') +const { useChannelsColumns } = await import('./channels-columns') +const { ChannelsProvider } = await import('./channels-provider') +const { MobileCardList } = + await import('@/components/data-table/mobile-card-list') +const { api } = await import('@/lib/api') +const { aggregateChannelsByTag } = await import('../lib/channel-utils') + +after(() => { + env.teardown() + if (previousStorage) + Object.defineProperty(globalThis, 'localStorage', previousStorage) + else Reflect.deleteProperty(globalThis, 'localStorage') + if (previousMatchMedia) + Object.defineProperty(globalThis, 'matchMedia', previousMatchMedia) + else Reflect.deleteProperty(globalThis, 'matchMedia') + if (previousCustomElements) + Object.defineProperty(globalThis, 'customElements', previousCustomElements) + else Reflect.deleteProperty(globalThis, 'customElements') +}) + +function health( + overrides: Partial = {} +): ChannelRoutingHealth { + return { + schema_version: 1, + status: 'normal', + source: 'redis', + base_priority: 10, + effective_priority: 10, + stored_penalty: 0, + active_penalty: 0, + stored_runtime_disabled: false, + runtime_disabled: false, + penalty_until: null, + disabled_until: null, + penalty_recovery_mode: null, + disable_recovery_mode: 'probe_then_automatic', + last_observed_at: null, + evaluated_at: new Date().toISOString(), + unavailable_reason: null, + ...overrides, + } +} + +function channel(id: number, snapshot?: unknown): Channel { + return channelSchema.parse({ + id, + priority: 10, + routing_health: snapshot, + name: `Channel ${id}`, + type: 1, + key: '', + status: 1, + tag: 'shared', + created_time: 0, + test_time: 0, + response_time: 0, + balance_updated_time: 0, + }) +} + +test('penalized priority and recovery details are visible without changing the configured value', async () => { + const source = channel( + 7, + health({ + status: 'penalized', + effective_priority: 9, + stored_penalty: 1, + active_penalty: 1, + penalty_recovery_mode: 'manual', + }) + ) + const view = await env.render() + try { + const fields = within(view.container) + assert.ok(fields.getByText('9', { exact: true })) + assert.ok(fields.getByText('Timeout penalty: 1')) + assert.ok(fields.getByText('Waiting for manual recovery')) + assert.equal(view.container.querySelectorAll('input').length, 0) + const details = view.container.querySelector('details')! + await view.click(fields.getByText('Routing snapshot details')) + assert.equal(details.open, true) + assert.ok(fields.getByText('Configured priority: 10')) + assert.equal(source.priority, 10) + } finally { + await view.unmount() + } +}) + +test('zero, unavailable snapshots and unsupported servers remain distinguishable', async () => { + for (const [snapshot, expected] of [ + [ + health({ base_priority: 0, effective_priority: 0 }), + 'No runtime restriction', + ], + [undefined, 'Runtime state display is not supported by this server.'], + [{ schema_version: 2 }, 'Unsupported runtime state version'], + [{ schema_version: 1 }, 'Runtime state unavailable'], + [ + health({ + status: 'unknown', + source: 'unavailable', + effective_priority: null, + stored_penalty: null, + active_penalty: null, + stored_runtime_disabled: null, + runtime_disabled: null, + unavailable_reason: 'private diagnostic', + }), + 'Runtime state unavailable', + ], + ] as const) { + const view = await env.render( + + ) + try { + assert.ok(within(view.container).getByText(expected)) + if (expected === 'No runtime restriction') { + assert.ok(within(view.container).getByText('0', { exact: true })) + } else { + assert.ok(within(view.container).getByText('—', { exact: true })) + } + assert.equal( + view.container.textContent?.includes('private diagnostic'), + false + ) + } finally { + await view.unmount() + } + } +}) + +test('a due disabled state still waits for a probe and never displays priority zero', async () => { + const view = await env.render( + + ) + try { + const fields = within(view.container) + assert.ok(fields.getByText('Paused for upstream timeouts')) + assert.ok(fields.getByText('Waiting for a successful recovery probe')) + assert.equal(fields.queryByText('0', { exact: true }), null) + } finally { + await view.unmount() + } +}) + +test('a penalty at its minimum and an expired snapshot are explained without recomputing routing', async () => { + const view = await env.render( + + ) + try { + const fields = within(view.container) + assert.ok(fields.getByText('0', { exact: true })) + assert.ok(fields.getByText('Priority is at its minimum')) + assert.ok(fields.getByText('Waiting for a fresh runtime snapshot')) + } finally { + await view.unmount() + } +}) + +test('tag rows report current-result counts and do not present the first child as the tag', async () => { + const rows = aggregateChannelsByTag([ + channel( + 1, + health({ + status: 'penalized', + effective_priority: 9, + stored_penalty: 1, + active_penalty: 1, + penalty_recovery_mode: 'manual', + }) + ), + channel( + 2, + health({ + status: 'runtime_disabled', + effective_priority: null, + stored_runtime_disabled: true, + runtime_disabled: true, + }) + ), + channel(3), + ]) + const view = await env.render() + try { + const fields = within(view.container) + assert.ok( + fields.getByText('Current results: 1 penalized / 1 paused / 1 unknown') + ) + assert.ok(fields.getByText('Runtime state is partially unknown')) + assert.ok(fields.getByText('Includes channels paused by timeout policy')) + assert.equal(fields.queryByText('Timeout penalty: 1'), null) + } finally { + await view.unmount() + } +}) + +function MobileHarness(props: { channel: Channel }) { + const columns = useChannelsColumns().filter((column) => { + const id = + column.id ?? ('accessorKey' in column ? column.accessorKey : undefined) + return id === 'name' || id === 'priority' || id === 'effective_priority' + }) + const table = useReactTable({ + data: [props.channel], + columns, + getCoreRowModel: getCoreRowModel(), + }) + return +} + +test('actual mobile columns show both priorities and editing sends only configured priority', async () => { + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }) + const originalAdapter = api.defaults.adapter + const requests: Record[] = [] + api.defaults.adapter = async (config) => { + assert.equal(config.method, 'put') + requests.push(JSON.parse(String(config.data)) as Record) + return { + data: { success: true }, + config, + status: 200, + statusText: 'OK', + headers: {}, + } + } + const view = await env.render( + + + + + + ) + try { + const fields = within(view.container) + assert.ok(fields.getByText('Configured priority', { exact: true })) + assert.ok(fields.getByText('Snapshot effective priority', { exact: true })) + assert.ok(fields.getByText('10', { exact: true })) + assert.ok(fields.getByText('9', { exact: true })) + await view.click(fields.getByRole('button', { name: 'Increment' })) + await waitFor(() => assert.equal(requests.length, 1)) + assert.deepEqual(requests[0], { id: 7, priority: 11 }) + } finally { + await view.unmount() + api.defaults.adapter = originalAdapter + client.clear() + } +}) diff --git a/web/default/src/features/channels/components/channel-routing-health.tsx b/web/default/src/features/channels/components/channel-routing-health.tsx new file mode 100644 index 00000000..8a068b1a --- /dev/null +++ b/web/default/src/features/channels/components/channel-routing-health.tsx @@ -0,0 +1,261 @@ +/* +Copyright (C) 2023-2026 MAX-API-Next + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as +published by the Free Software Foundation, either version 3 of the +License, or (at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . + +For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues +*/ +import { useContext } from 'react' +import { useTranslation } from 'react-i18next' +import { + readChannelRoutingHealth, + summarizeChannelRoutingHealth, + type ChannelRoutingHealth, +} from '../lib/channel-routing-health' +import { isTagAggregateRow, type TagRow } from '../lib/channel-utils' +import type { Channel } from '../types' +import { ChannelHealthSnapshotContext } from './channel-health-snapshot-context' + +function RecoveryNote(props: { health: ChannelRoutingHealth }) { + const { t, i18n } = useTranslation() + const health = props.health + const snapshot = useContext(ChannelHealthSnapshotContext) + const elapsed = + snapshot.now !== null && snapshot.receivedAt !== undefined + ? Math.max(0, snapshot.now - snapshot.receivedAt) + : 0 + const evaluatedAt = Date.parse(health.evaluated_at) + elapsed + const formatTime = (value: string) => + new Date(value).toLocaleString(i18n.resolvedLanguage || i18n.language) + if (health.status === 'runtime_disabled') { + if (health.disable_recovery_mode === 'manual') { + return {t('Waiting for manual recovery')} + } + if ( + health.disabled_until && + Date.parse(health.disabled_until) > evaluatedAt + ) { + return ( + + {t('Earliest recovery check: {{time}}', { + time: formatTime(health.disabled_until), + })} + {' · '} + {t('Recovery requires a successful probe.')} + + ) + } + return {t('Waiting for a successful recovery probe')} + } + if (health.status !== 'penalized') return null + if (health.penalty_recovery_mode === 'manual') { + return {t('Waiting for manual recovery')} + } + if ( + !health.penalty_until || + Date.parse(health.penalty_until) <= evaluatedAt + ) { + return {t('Waiting for a fresh runtime snapshot')} + } + return ( + + {t('Penalty expiry in this snapshot: {{time}}', { + time: formatTime(health.penalty_until), + })} + + ) +} + +export function ChannelRoutingHealthDetails(props: { routingHealth: unknown }) { + const { t, i18n } = useTranslation() + const snapshot = useContext(ChannelHealthSnapshotContext) + const observation = readChannelRoutingHealth(props.routingHealth) + if (observation.kind !== 'known') { + let message = t('Runtime state unavailable') + if (observation.kind === 'missing') { + message = t('Runtime state display is not supported by this server.') + } else if (observation.kind === 'unsupported') { + message = t('Unsupported runtime state version') + } + return ( +
+ — +

{message}

+

+ {t( + 'Unknown means the display could not confirm runtime state. Routing may use cached state or configured priority.' + )} +

+
+ ) + } + const health = observation.health + let label = t('No runtime restriction') + if (health.status === 'penalized') { + label = t('Timeout penalty: {{points}}', { points: health.active_penalty }) + } else if (health.status === 'runtime_disabled') { + label = t('Paused for upstream timeouts') + } else if (health.status === 'policy_off') { + label = t('Runtime policy is disabled') + } else if (health.status === 'unknown') { + label = t('Runtime state unavailable') + } + const formatTime = (value: string) => + new Date(value).toLocaleString(i18n.resolvedLanguage || i18n.language) + const atFloor = + health.status === 'penalized' && + health.base_priority >= 0 && + health.effective_priority === 0 + return ( +
+ {snapshot.now !== null && + snapshot.receivedAt !== undefined && + snapshot.now - snapshot.receivedAt > 60_000 && ( +

+ {t('Runtime snapshot may be outdated.')} +

+ )} +
+ + {health.effective_priority ?? '—'} + + {label} +
+ {health.status === 'unknown' && ( +

+ {t( + 'Unknown means the display could not confirm runtime state. Routing may use cached state or configured priority.' + )} +

+ )} + {atFloor && ( +

+ {t('Priority is at its minimum')} +

+ )} + {(health.status === 'penalized' || + health.status === 'runtime_disabled') && ( +

+ +

+ )} +
+ + {t('Routing snapshot details')} + +
+

+ {t('Configured priority')}: {health.base_priority} +

+ {health.stored_penalty !== null && ( +

+ {t('Recorded timeout penalty: {{points}}', { + points: health.stored_penalty, + })} +

+ )} + {health.stored_runtime_disabled && !health.runtime_disabled && ( +

+ {t('The recorded timeout disable is not currently enforced.')} +

+ )} + {health.last_observed_at && ( +

+ {t('Last runtime event')}: {formatTime(health.last_observed_at)} +

+ )} +

+ {t('Snapshot time')}:{' '} + +

+

+ {t( + 'This snapshot describes runtime restrictions, not upstream health.' + )} +

+

+ {t( + 'Ordinary priority routing only. Node configuration, policy and runtime caches must match this snapshot; group, model, retry and weights also affect selection.' + )} +

+

+ {t( + 'Snapshot time is the responding node’s calculation time, not a global state version or a record of a past request.' + )} +

+
+
+
+ ) +} + +function TagChannelHealth(props: { channel: TagRow }) { + const { t } = useTranslation() + const summary = summarizeChannelRoutingHealth(props.channel.children) + let value = '—' + if (summary.multiplePriorities) value = t('Multiple effective priorities') + else if (summary.effectivePriority !== null) { + value = String(summary.effectivePriority) + } else if (summary.disabled === summary.total && summary.total > 0) { + value = t('Paused for upstream timeouts') + } + return ( +
+

{value}

+

+ {t( + 'Current results: {{penalized}} penalized / {{disabled}} paused / {{unknown}} unknown', + { + penalized: summary.penalized, + disabled: summary.disabled, + unknown: summary.unknown, + } + )} +

+ {summary.unknown > 0 && summary.unknown < summary.total && ( +

{t('Runtime state is partially unknown')}

+ )} + {summary.disabled > 0 && summary.disabled < summary.total && ( +

{t('Includes channels paused by timeout policy')}

+ )} +
+ ) +} + +export function ChannelRoutingHealthCell(props: { channel: Channel }) { + const { t } = useTranslation() + const snapshot = useContext(ChannelHealthSnapshotContext) + if (snapshot.placeholder) { + return ( +

+ {t( + 'Waiting for the current results. Previous runtime values are hidden.' + )} +

+ ) + } + return ( +
+ {isTagAggregateRow(props.channel) ? ( + + ) : ( + + )} +
+ ) +} diff --git a/web/default/src/features/channels/components/channels-columns.tsx b/web/default/src/features/channels/components/channels-columns.tsx index f0a8ad81..f6c51378 100644 --- a/web/default/src/features/channels/components/channels-columns.tsx +++ b/web/default/src/features/channels/components/channels-columns.tsx @@ -17,9 +17,10 @@ along with this program. If not, see . For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues */ /* eslint-disable react-refresh/only-export-components */ -import { useState } from 'react' +import { useState, useMemo } from 'react' import { useQueryClient } from '@tanstack/react-query' import { type ColumnDef } from '@tanstack/react-table' +import type { TFunction } from 'i18next' import { AlertTriangle, ChevronDown, @@ -72,7 +73,8 @@ import { } from '../lib' import { parseUpstreamUpdateMeta } from '../lib/upstream-update-utils' import type { Channel } from '../types' -import { useChannels } from './channels-provider' +import { ChannelRoutingHealthCell } from './channel-routing-health' +import { useChannels, useChannelRefreshLock } from './channels-provider' import { DataTableRowActions } from './data-table-row-actions' import { DataTableTagRowActions } from './data-table-tag-row-actions' import { @@ -165,6 +167,30 @@ function PriorityCell({ channel }: { channel: Channel }) { const priority = channel.priority const [confirmOpen, setConfirmOpen] = useState(false) const [pendingValue, setPendingValue] = useState(null) + const [editing, setEditing] = useState(false) + const [saving, setSaving] = useState(false) + useChannelRefreshLock(editing || confirmOpen || saving) + const save = async (value: number) => { + setSaving(true) + try { + if (isTagRow) + await handleUpdateTagField( + channel.tag || '', + 'priority', + value, + queryClient + ) + else + await handleUpdateChannelField( + channel.id, + 'priority', + value, + queryClient + ) + } finally { + setSaving(false) + } + } // Tag row - editable with confirmation for all tag channels if (isTagRow) { @@ -175,12 +201,19 @@ function PriorityCell({ channel }: { channel: Channel }) { <> { setPendingValue(value) setConfirmOpen(true) }} min={-999} /> + {priority == null && ( +

+ {t('Multiple configured priorities')} +

+ )} { if (pendingValue !== null) { - handleUpdateTagField(tag, 'priority', pendingValue, queryClient) + void save(pendingValue) } setConfirmOpen(false) }} @@ -202,8 +235,10 @@ function PriorityCell({ channel }: { channel: Channel }) { return ( { - handleUpdateChannelField(channel.id, 'priority', value, queryClient) + void save(value) }} min={-999} /> @@ -220,6 +255,25 @@ function WeightCell({ channel }: { channel: Channel }) { const weight = channel.weight const [confirmOpen, setConfirmOpen] = useState(false) const [pendingValue, setPendingValue] = useState(null) + const [editing, setEditing] = useState(false) + const [saving, setSaving] = useState(false) + useChannelRefreshLock(editing || confirmOpen || saving) + const save = async (value: number) => { + setSaving(true) + try { + if (isTagRow) + await handleUpdateTagField( + channel.tag || '', + 'weight', + value, + queryClient + ) + else + await handleUpdateChannelField(channel.id, 'weight', value, queryClient) + } finally { + setSaving(false) + } + } // Tag row - editable with confirmation for all tag channels if (isTagRow) { @@ -230,6 +284,8 @@ function WeightCell({ channel }: { channel: Channel }) { <> { setPendingValue(value) setConfirmOpen(true) @@ -244,7 +300,7 @@ function WeightCell({ channel }: { channel: Channel }) { confirmText='Update' handleConfirm={() => { if (pendingValue !== null) { - handleUpdateTagField(tag, 'weight', pendingValue, queryClient) + void save(pendingValue) } setConfirmOpen(false) }} @@ -257,8 +313,10 @@ function WeightCell({ channel }: { channel: Channel }) { return ( { - handleUpdateChannelField(channel.id, 'weight', value, queryClient) + void save(value) }} min={0} /> @@ -421,8 +479,7 @@ function BalanceCell({ channel }: { channel: Channel }) { /** * Generate channels columns configuration */ -export function useChannelsColumns(): ColumnDef[] { - const { t } = useTranslation() +function createChannelsColumns(t: TFunction): ColumnDef[] { return [ // Checkbox column { @@ -913,12 +970,27 @@ export function useChannelsColumns(): ColumnDef[] { // Priority column { accessorKey: 'priority', - meta: { label: t('Priority'), mobileHidden: true }, + meta: { label: t('Configured priority') }, header: ({ column }) => ( - + ), cell: ({ row }) => , - size: 100, + size: 160, + }, + + { + id: 'effective_priority', + meta: { + label: t('Snapshot effective priority'), + cellClassName: 'whitespace-normal', + }, + header: t('Snapshot effective priority'), + cell: ({ row }) => , + enableSorting: false, + size: 220, }, // Weight column @@ -1029,3 +1101,8 @@ export function useChannelsColumns(): ColumnDef[] { }, ] } + +export function useChannelsColumns(): ColumnDef[] { + const { t } = useTranslation() + return useMemo(() => createChannelsColumns(t), [t]) +} diff --git a/web/default/src/features/channels/components/channels-primary-buttons.tsx b/web/default/src/features/channels/components/channels-primary-buttons.tsx index c8bddaf0..ccc5c24f 100644 --- a/web/default/src/features/channels/components/channels-primary-buttons.tsx +++ b/web/default/src/features/channels/components/channels-primary-buttons.tsx @@ -50,7 +50,7 @@ import { handleTestAllChannels, handleUpdateAllBalances, } from '../lib' -import { useChannels } from './channels-provider' +import { useChannels, useChannelRefreshLock } from './channels-provider' export function ChannelsPrimaryButtons() { const { t } = useTranslation() @@ -65,6 +65,7 @@ export function ChannelsPrimaryButtons() { } = useChannels() const queryClient = useQueryClient() const [showDeleteDialog, setShowDeleteDialog] = useState(false) + useChannelRefreshLock(showDeleteDialog) const handleTagModeToggle = (checked: boolean) => { localStorage.setItem('enable-tag-mode', String(checked)) diff --git a/web/default/src/features/channels/components/channels-provider.tsx b/web/default/src/features/channels/components/channels-provider.tsx index 27869c7a..1d55760e 100644 --- a/web/default/src/features/channels/components/channels-provider.tsx +++ b/web/default/src/features/channels/components/channels-provider.tsx @@ -17,7 +17,14 @@ along with this program. If not, see . For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues */ /* eslint-disable react-refresh/only-export-components */ -import React, { createContext, useContext, useState, useCallback } from 'react' +import React, { + createContext, + useContext, + useState, + useCallback, + useEffect, + useId, +} from 'react' import { useQueryClient } from '@tanstack/react-query' import { useChannelUpstreamUpdates } from '../hooks/use-channel-upstream-updates' import { channelsQueryKeys } from '../lib' @@ -54,6 +61,8 @@ type ChannelsContextType = { idSort: boolean setIdSort: (enabled: boolean) => void upstream: UpstreamUpdateState + refreshPaused: boolean + setRefreshLock: (id: string, paused: boolean) => void } // ============================================================================ @@ -72,6 +81,16 @@ export function ChannelsProvider({ children }: { children: React.ReactNode }) { const [open, setOpen] = useState(null) const [currentRow, setCurrentRow] = useState(null) const [currentTag, setCurrentTag] = useState(null) + const [refreshLocks, setRefreshLocks] = useState>(() => new Set()) + const setRefreshLock = useCallback((id: string, paused: boolean) => { + setRefreshLocks((previous) => { + if (previous.has(id) === paused) return previous + const next = new Set(previous) + if (paused) next.add(id) + else next.delete(id) + return next + }) + }, []) const [enableTagMode, setEnableTagMode] = useState(() => { return localStorage.getItem('enable-tag-mode') === 'true' }) @@ -99,6 +118,9 @@ export function ChannelsProvider({ children }: { children: React.ReactNode }) { idSort, setIdSort, upstream, + refreshPaused: + open !== null || upstream.showModal || refreshLocks.size > 0, + setRefreshLock, }} > {children} @@ -117,3 +139,12 @@ export function useChannels() { } return context } + +export function useChannelRefreshLock(paused: boolean) { + const { setRefreshLock } = useChannels() + const id = useId() + useEffect(() => { + setRefreshLock(id, paused) + return () => setRefreshLock(id, false) + }, [id, paused, setRefreshLock]) +} diff --git a/web/default/src/features/channels/components/channels-table.tsx b/web/default/src/features/channels/components/channels-table.tsx index ab4893b1..4db75d6d 100644 --- a/web/default/src/features/channels/components/channels-table.tsx +++ b/web/default/src/features/channels/components/channels-table.tsx @@ -16,8 +16,8 @@ along with this program. If not, see . For commercial licensing, please contact https://github.com/MAX-API-Next/MAX-API/issues */ -import { useState, useMemo, useEffect } from 'react' -import { useQuery } from '@tanstack/react-query' +import { useState, useMemo, useEffect, useCallback } from 'react' +import { useQuery, useQueryClient } from '@tanstack/react-query' import { getRouteApi } from '@tanstack/react-router' import { getCoreRowModel, @@ -45,6 +45,10 @@ import { CHANNEL_STATUS, CHANNEL_STATUS_OPTIONS, } from '../constants' +import { + useChannelHealthRefresh, + CHANNEL_HEALTH_REFRESH_MS, +} from '../hooks/use-channel-health-refresh' import { channelsQueryKeys, aggregateChannelsByTag, @@ -53,6 +57,7 @@ import { getChannelTypeLabel, } from '../lib' import type { Channel, ChannelSortBy } from '../types' +import { ChannelHealthSnapshotContext } from './channel-health-snapshot-context' import { useChannelsColumns } from './channels-columns' import { useChannels } from './channels-provider' import { DataTableBulkActions } from './data-table-bulk-actions' @@ -76,7 +81,8 @@ function isDisabledChannelRow(channel: Channel) { export function ChannelsTable() { const { t } = useTranslation() - const { enableTagMode, idSort } = useChannels() + const { enableTagMode, idSort, refreshPaused } = useChannels() + const queryClient = useQueryClient() const isMobile = useMediaQuery('(max-width: 640px)') // Table state @@ -198,7 +204,7 @@ export function ChannelsTable() { // Fetch channels data // eslint-disable-next-line @tanstack/query/exhaustive-deps - const { data, isLoading, isFetching } = useQuery({ + const query = useQuery({ queryKey: channelsQueryKeys.list({ keyword: globalFilter, model: modelFilter, @@ -220,53 +226,92 @@ export function ChannelsTable() { p: pagination.pageIndex + 1, page_size: pagination.pageSize, }), - queryFn: async () => { + queryFn: async ({ signal }) => { + const config = { + signal, + disableDuplicate: true, + skipBusinessError: true, + skipErrorHandler: true, + timeout: 10_000, + } if (shouldSearch) { - return searchChannels({ - keyword: globalFilter, - model: modelFilter, - group: - groupFilter.length > 0 && !groupFilter.includes('all') - ? groupFilter[0] - : undefined, - status: - statusFilter.length > 0 && !statusFilter.includes('all') - ? statusFilter[0] - : undefined, - type: - typeFilter.length > 0 && !typeFilter.includes('all') - ? Number(typeFilter[0]) - : undefined, - tag_mode: enableTagMode, - id_sort: idSort, - ...sortParams, - p: pagination.pageIndex + 1, - page_size: pagination.pageSize, - }) + const response = await searchChannels( + { + keyword: globalFilter, + model: modelFilter, + group: + groupFilter.length > 0 && !groupFilter.includes('all') + ? groupFilter[0] + : undefined, + status: + statusFilter.length > 0 && !statusFilter.includes('all') + ? statusFilter[0] + : undefined, + type: + typeFilter.length > 0 && !typeFilter.includes('all') + ? Number(typeFilter[0]) + : undefined, + tag_mode: enableTagMode, + id_sort: idSort, + ...sortParams, + p: pagination.pageIndex + 1, + page_size: pagination.pageSize, + }, + config + ) + if (!response.success) throw new Error('Channel search unavailable') + return response } else { - return getChannels({ - group: - groupFilter.length > 0 && !groupFilter.includes('all') - ? groupFilter[0] - : undefined, - status: - statusFilter.length > 0 && !statusFilter.includes('all') - ? statusFilter[0] - : undefined, - type: - typeFilter.length > 0 && !typeFilter.includes('all') - ? Number(typeFilter[0]) - : undefined, - tag_mode: enableTagMode, - id_sort: idSort, - ...sortParams, - p: pagination.pageIndex + 1, - page_size: pagination.pageSize, - }) + const response = await getChannels( + { + group: + groupFilter.length > 0 && !groupFilter.includes('all') + ? groupFilter[0] + : undefined, + status: + statusFilter.length > 0 && !statusFilter.includes('all') + ? statusFilter[0] + : undefined, + type: + typeFilter.length > 0 && !typeFilter.includes('all') + ? Number(typeFilter[0]) + : undefined, + tag_mode: enableTagMode, + id_sort: idSort, + ...sortParams, + p: pagination.pageIndex + 1, + page_size: pagination.pageSize, + }, + config + ) + if (!response.success) throw new Error('Channel list unavailable') + return response } }, placeholderData: (previousData) => previousData, + enabled: !refreshPaused, + staleTime: CHANNEL_HEALTH_REFRESH_MS, + refetchOnWindowFocus: false, + refetchOnReconnect: false, + retry: false, + }) + const { data, isLoading, isFetching, isPlaceholderData, isError, refetch } = + query + const refresh = useCallback( + () => refetch({ cancelRefetch: false }), + [refetch] + ) + const snapshots = useMemo(() => data?.data?.items ?? [], [data]) + const now = useChannelHealthRefresh({ + snapshots, + receivedAt: data?.observation.receivedAt, + paused: refreshPaused, + refetch: refresh, }) + useEffect(() => { + if (refreshPaused) + void queryClient.cancelQueries({ queryKey: channelsQueryKeys.lists() }) + }, [refreshPaused, queryClient]) // Apply tag aggregation if tag mode is enabled const channels = useMemo(() => { @@ -379,69 +424,106 @@ export function ChannelsTable() { ] return ( - handleModelFilterChange(e.target.value)} - onKeyDown={(event) => { - if (event.key === 'Enter' && !event.nativeEvent.isComposing) { - event.preventDefault() - handleSearch() - } - }} - className='w-full sm:w-[150px] lg:w-[180px]' - /> - ), - filters: [ - { - columnId: 'status', - title: t('Status'), - options: [...CHANNEL_STATUS_OPTIONS], - singleSelect: true, - }, - { - columnId: 'type', - title: t('Type'), - options: typeFilterOptions, - singleSelect: true, - }, - { - columnId: 'group', - title: t('Group'), - options: groupFilterOptions, - singleSelect: true, - }, - ], + - isDisabledChannelRow(row.original) - ? isMobile - ? DISABLED_ROW_MOBILE - : DISABLED_ROW_DESKTOP - : undefined - } - bulkActions={} - /> + > +
+ {refreshPaused && ( +

+ {t( + 'Automatic channel refresh is paused while editing or confirming changes.' + )} +

+ )} + {isPlaceholderData && ( +

+ {t( + 'Waiting for the current results. Previous runtime values are hidden.' + )} +

+ )} + {isError && ( +

+ {t('Channel refresh failed. Displayed results may be outdated.')}{' '} + +

+ )} + handleModelFilterChange(e.target.value)} + onKeyDown={(event) => { + if (event.key === 'Enter' && !event.nativeEvent.isComposing) { + event.preventDefault() + handleSearch() + } + }} + className='w-full sm:w-[150px] lg:w-[180px]' + /> + ), + filters: [ + { + columnId: 'status', + title: t('Status'), + options: [...CHANNEL_STATUS_OPTIONS], + singleSelect: true, + }, + { + columnId: 'type', + title: t('Type'), + options: typeFilterOptions, + singleSelect: true, + }, + { + columnId: 'group', + title: t('Group'), + options: groupFilterOptions, + singleSelect: true, + }, + ], + }} + getRowClassName={(row, { isMobile }) => + isDisabledChannelRow(row.original) + ? isMobile + ? DISABLED_ROW_MOBILE + : DISABLED_ROW_DESKTOP + : undefined + } + bulkActions={} + /> +
+ ) } diff --git a/web/default/src/features/channels/components/data-table-bulk-actions.tsx b/web/default/src/features/channels/components/data-table-bulk-actions.tsx index 132f9c14..9cfd8d5a 100644 --- a/web/default/src/features/channels/components/data-table-bulk-actions.tsx +++ b/web/default/src/features/channels/components/data-table-bulk-actions.tsx @@ -45,6 +45,7 @@ import { handleBatchSetTag, } from '../lib' import type { Channel } from '../types' +import { useChannelRefreshLock } from './channels-provider' interface DataTableBulkActionsProps { table: Table @@ -58,6 +59,7 @@ export function DataTableBulkActions({ const [showTagDialog, setShowTagDialog] = useState(false) const [showDeleteConfirm, setShowDeleteConfirm] = useState(false) const [tagValue, setTagValue] = useState('') + useChannelRefreshLock(showTagDialog || showDeleteConfirm) const selectedRows = table.getFilteredSelectedRowModel().rows const selectedIds = selectedRows.reduce((ids, row) => { diff --git a/web/default/src/features/channels/components/dialogs/edit-tag-dialog.tsx b/web/default/src/features/channels/components/dialogs/edit-tag-dialog.tsx index 6c0f6eda..76f6248e 100644 --- a/web/default/src/features/channels/components/dialogs/edit-tag-dialog.tsx +++ b/web/default/src/features/channels/components/dialogs/edit-tag-dialog.tsx @@ -52,7 +52,7 @@ import { getAllModels, getGroups, } from '../../api' -import { channelsQueryKeys } from '../../lib' +import { invalidateChannelConfiguration } from '../../lib' import type { TagOperationParams } from '../../types' import { useChannels } from '../channels-provider' @@ -206,7 +206,7 @@ export function EditTagDialog({ open, onOpenChange }: EditTagDialogProps) { if (response.success) { toast.success(t('Tag updated successfully')) - queryClient.invalidateQueries({ queryKey: channelsQueryKeys.lists() }) + void invalidateChannelConfiguration(queryClient) onOpenChange(false) } else { toast.error(response.message || t('Failed to update tag')) diff --git a/web/default/src/features/channels/components/dialogs/tag-batch-edit-dialog.tsx b/web/default/src/features/channels/components/dialogs/tag-batch-edit-dialog.tsx index 50499588..8b3e5888 100644 --- a/web/default/src/features/channels/components/dialogs/tag-batch-edit-dialog.tsx +++ b/web/default/src/features/channels/components/dialogs/tag-batch-edit-dialog.tsx @@ -37,7 +37,7 @@ import { Skeleton } from '@/components/ui/skeleton' import { Textarea } from '@/components/ui/textarea' import { MultiSelect } from '@/components/multi-select' import { getTagModels, editTagChannels, getGroups } from '../../api' -import { channelsQueryKeys } from '../../lib' +import { invalidateChannelConfiguration } from '../../lib' import type { TagOperationParams } from '../../types' import { useChannels } from '../channels-provider' import { ModelMappingEditor } from '../model-mapping-editor' @@ -173,7 +173,7 @@ function TagBatchEditDialogContent({ ) if (response.success) { toast.success(t('Tag updated successfully')) - queryClient.invalidateQueries({ queryKey: channelsQueryKeys.lists() }) + void invalidateChannelConfiguration(queryClient) handleClose() } else { toast.error(response.message || t('Failed to update tag')) diff --git a/web/default/src/features/channels/components/drawers/channel-mutate-drawer.tsx b/web/default/src/features/channels/components/drawers/channel-mutate-drawer.tsx index 47413587..c2494d53 100644 --- a/web/default/src/features/channels/components/drawers/channel-mutate-drawer.tsx +++ b/web/default/src/features/channels/components/drawers/channel-mutate-drawer.tsx @@ -116,6 +116,7 @@ import { CHANNEL_FORM_DEFAULT_VALUES, channelFormSchema, channelsQueryKeys, + invalidateChannelConfiguration, transformChannelToFormDefaults, type ChannelFormValues, deduplicateKeys, @@ -143,6 +144,7 @@ import { } from '../../lib/status-code-risk-guard' import type { Channel } from '../../types' import { ChannelCapabilityMatrix } from '../channel-capability-matrix' +import { ChannelRoutingHealthCard } from '../channel-routing-health-card' import { useChannels } from '../channels-provider' import { AdvancedCustomEditorDialog } from '../dialogs/advanced-custom-editor-dialog' import { CodexOAuthDialog } from '../dialogs/codex-oauth-dialog' @@ -239,12 +241,37 @@ export function ChannelMutateDrawer({ const isEditing = Boolean(currentRow) const channelId = currentRow?.id ?? null + const initializedEditorRef = useRef(null) + const [initializedEditorId, setInitializedEditorId] = useState< + number | 'new' | null + >(null) // Fetch channel details if editing - const { data: channelData, isLoading: isChannelLoading } = useQuery({ + const { + data: channelData, + dataUpdatedAt: channelDataUpdatedAt, + isError: isChannelError, + refetch: refetchChannelDetail, + isFetching: isChannelFetching, + } = useQuery({ queryKey: channelsQueryKeys.detail(currentRow?.id || 0), - queryFn: () => getChannel(currentRow!.id), - enabled: isEditing && Boolean(currentRow?.id), + queryFn: async ({ signal }) => { + const response = await getChannel(currentRow!.id, { + signal, + disableDuplicate: true, + timeout: 8_000, + skipBusinessError: true, + skipErrorHandler: true, + }) + if (!response.success || response.data?.id !== currentRow!.id) + throw new Error('Channel details unavailable') + return response + }, + enabled: open && isEditing && Boolean(currentRow?.id), + refetchOnWindowFocus: false, + refetchOnReconnect: false, + refetchOnMount: 'always', + retry: false, }) // Fetch available groups @@ -436,7 +463,7 @@ export function ChannelMutateDrawer({ // Helper computed values const isBatchMode = multiKeyMode === 'batch' || multiKeyMode === 'multi_to_single' - const isChannelDetailLoading = isEditing && isChannelLoading + const isChannelDetailLoading = isEditing && initializedEditorId !== channelId const advancedCustomStats = useMemo( () => getAdvancedCustomStats(currentAdvancedCustom), [currentAdvancedCustom] @@ -766,9 +793,29 @@ export function ChannelMutateDrawer({ isEditing, }) + useEffect(() => { + initializedEditorRef.current = null + setInitializedEditorId(null) + }, [open, channelId]) + // Load channel data into form when editing useEffect(() => { - if (isEditing && channelData?.data) { + if (!open) { + initializedEditorRef.current = null + return + } + // Initialize once per channel/open session. Broad cache invalidations and + // credential refreshes must not reset an in-progress configuration draft. + if ( + isEditing && + !isChannelError && + channelData?.success && + channelData.data?.id === channelId + ) { + if (initializedEditorRef.current === channelId || isChannelFetching) + return + initializedEditorRef.current = channelId + setInitializedEditorId(channelId) const defaults = transformChannelToFormDefaults(channelData.data) const shouldOpenAdvancedSettings = readAdvancedSettingsPreference() || @@ -788,7 +835,9 @@ export function ChannelMutateDrawer({ initialModelMappingRef.current = channelData.data.model_mapping || '' initialStatusCodeMappingRef.current = channelData.data.status_code_mapping || '' - } else if (!isEditing) { + } else if (!isEditing && initializedEditorRef.current !== 'new') { + initializedEditorRef.current = 'new' + setInitializedEditorId('new') form.reset(CHANNEL_FORM_DEFAULT_VALUES) setAdvancedSettingsOpen(false) setExpandedEditorNavItemId(undefined) @@ -797,7 +846,15 @@ export function ChannelMutateDrawer({ initialModelMappingRef.current = '' initialStatusCodeMappingRef.current = '' } - }, [isEditing, channelData, form]) + }, [ + open, + channelId, + isEditing, + channelData, + isChannelFetching, + isChannelError, + form, + ]) useEffect(() => { if (currentType !== CHANNEL_TYPE_DOUBAO_VIDEO) { @@ -971,12 +1028,7 @@ export function ChannelMutateDrawer({ // Handle successful submission const handleSuccess = useCallback(() => { - queryClient.invalidateQueries({ queryKey: channelsQueryKeys.lists() }) - if (channelId) { - queryClient.invalidateQueries({ - queryKey: channelsQueryKeys.detail(channelId), - }) - } + void invalidateChannelConfiguration(queryClient, channelId ?? undefined) onOpenChange(false) setOpen(null) }, [channelId, queryClient, onOpenChange, setOpen]) @@ -1346,7 +1398,27 @@ export function ChannelMutateDrawer({ className={sideDrawerFormClassName('gap-5')} > {isChannelDetailLoading ? ( - +
+ {isChannelError && ( + + + {t( + 'Unable to load channel configuration. Retry before editing.' + )}{' '} + + + + )} + {!isChannelError && } +
) : (
( - {t('Priority')} + + {t('Configured priority')} +
+ {open && + isEditing && + channelData?.success && + channelData.data?.id === channelId && ( + + )} + {t('Cancel')} -