diff --git a/internal/dispatchadapter/catalog.go b/internal/dispatchadapter/catalog.go index abb9ce7..0e91ec6 100644 --- a/internal/dispatchadapter/catalog.go +++ b/internal/dispatchadapter/catalog.go @@ -259,6 +259,20 @@ func (c *Catalog) subscribeInvalidate(ctx context.Context) { } } +// healthOf 从快照中查渠道×模型的健康读值（Redis 降级时沿用 last-known）。 +// byModel 每个模型下同一渠道至多一行，线性扫描只发生在降级路径，不影响正常刷新。 +func (idx *snapshotIndex) healthOf(channelID int64, model string) (succ, lat float64, ok bool) { + if idx == nil { + return 0, 0, false + } + for _, ch := range idx.byModel[model] { + if ch.ID == channelID { + return ch.SuccEwma, ch.LatEwmaMs, true + } + } + return 0, 0, false +} + // Rebuild 全量重建快照：DB 目录 + Redis 运行时读值合并。 // 加载失败保留上一份快照（last-known，基线方案 §13）。 func (c *Catalog) Rebuild(ctx context.Context) { @@ -268,6 +282,7 @@ func (c *Catalog) Rebuild(ctx context.Context) { return } + prev := c.current.Load() pol := c.policy() now := time.Now().UnixMilli() rampWindowMs := int64(pol.Ramp.WindowSeconds) * 1000 @@ -284,6 +299,7 @@ func (c *Catalog) Rebuild(ctx context.Context) { model string } rtCache := make(map[rtKey]RuntimeReadout) + degraded, carried := 0, 0 for _, row := range data.rows { rk := rtKey{row.ChannelID, row.ModelName} @@ -294,6 +310,16 @@ func (c *Catalog) Rebuild(ctx context.Context) { } else { rt = RuntimeReadout{SuccEwma: 1} } + // Redis 健康读取失败：沿用上一份快照的健康读值。若让乐观默认值 succ=1 落进 + // 新快照，Redis 抖一下就会把全渠道的健康记忆清零（healthFactor 集体回满分）， + // 与注释声称的 last-known 语义相悖。 + if rt.Degraded { + degraded++ + if succ, lat, found := prev.healthOf(row.ChannelID, row.ModelName); found { + rt.SuccEwma, rt.LatEwmaMs = succ, lat + carried++ + } + } rtCache[rk] = rt } @@ -344,6 +370,10 @@ func (c *Catalog) Rebuild(ctx context.Context) { } } c.current.Store(idx) + if degraded > 0 { + g.Log().Warningf(ctx, "[Dispatch] 目录刷新时 Redis 健康读取降级: %d 个渠道×模型读失败，其中 %d 个已沿用上一份快照的健康读值（其余无历史，按满分处理）", + degraded, carried) + } } // effectiveSoftLimit softLimit 双来源（基线方案 §8.2）： diff --git a/internal/dispatchadapter/catalog_test.go b/internal/dispatchadapter/catalog_test.go index dafb0d8..79a0149 100644 --- a/internal/dispatchadapter/catalog_test.go +++ b/internal/dispatchadapter/catalog_test.go @@ -133,6 +133,48 @@ func TestCatalog_加载失败保留上一份快照(t *testing.T) { assert.Len(t, c.Snapshot(ctx, 0, "m", nil), 1, "last-known 快照必须保留") } +// TestCatalog_Redis降级沿用上一份健康读值 Redis 读失败时若让乐观默认值 succ=1 落进 +// 新快照，一次 Redis 抖动就会把全渠道的健康记忆清零、healthFactor 集体回满分， +// 与 Rebuild 声称的 last-known 语义相悖。 +func TestCatalog_Redis降级沿用上一份健康读值(t *testing.T) { + rows := []catalogRow{ + {ChannelID: 1, ChannelName: "A", ModelName: "gpt-4o", Weight: 10, Tier: "primary", CostRatio: 1}, + } + degraded := false + c := testCatalog(rows, nil, func(_ context.Context, _ int64, _ string) RuntimeReadout { + if degraded { + return RuntimeReadout{SuccEwma: 1, Degraded: true} + } + return RuntimeReadout{SuccEwma: 0.3, LatEwmaMs: 4000, HasHealth: true} + }) + + got := c.Snapshot(context.Background(), 0, "gpt-4o", nil) + require.Len(t, got, 1) + assert.Equal(t, 0.3, got[0].SuccEwma) + + degraded = true + c.Rebuild(context.Background()) + got = c.Snapshot(context.Background(), 0, "gpt-4o", nil) + require.Len(t, got, 1) + assert.Equal(t, 0.3, got[0].SuccEwma, "Redis 降级必须沿用 last-known，不得回落满分") + assert.Equal(t, 4000.0, got[0].LatEwmaMs, "延迟读值同样沿用") +} + +// TestCatalog_首次刷新即降级无历史可沿用 新实例启动时 Redis 就不可用：没有 last-known +// 可沿用，保持乐观默认满分（不能让新实例启动即把全渠道判死）。 +func TestCatalog_首次刷新即降级无历史可沿用(t *testing.T) { + rows := []catalogRow{ + {ChannelID: 1, ChannelName: "A", ModelName: "gpt-4o", Weight: 10, Tier: "primary", CostRatio: 1}, + } + c := testCatalog(rows, nil, func(_ context.Context, _ int64, _ string) RuntimeReadout { + return RuntimeReadout{SuccEwma: 1, Degraded: true} + }) + + got := c.Snapshot(context.Background(), 0, "gpt-4o", nil) + require.Len(t, got, 1) + assert.Equal(t, 1.0, got[0].SuccEwma, "无历史时保持乐观默认") +} + func TestRampElapsed(t *testing.T) { now := time.Now().UnixMilli() window := int64(120_000) diff --git a/internal/dispatchadapter/state_redis.go b/internal/dispatchadapter/state_redis.go index 707c810..6c48d11 100644 --- a/internal/dispatchadapter/state_redis.go +++ b/internal/dispatchadapter/state_redis.go @@ -636,16 +636,28 @@ type RuntimeReadout struct { RecoveredMs int64 // 渠道级熔断最近恢复时间（0 = 无记录） FirstOpenedMs int64 // 渠道级本轮故障期起点（0 = 非故障期），供自动禁用判定 Onset429Ewma float64 // 429 起始水位 EWMA（0 = 无 429 历史），供 softLimit 自动估计 + + // HasHealth 健康 hash 中读到了可用的 succ_ewma（该渠道×模型有过真实上报）。 + // false 表示 SuccEwma/LatEwmaMs 是乐观默认值而非实测值——聚合展示分时必须排除， + // 否则大量无流量模型的"满分"会把真正故障模型的低分稀释掉。 + HasHealth bool + // Degraded Redis 健康读取失败，本次读值完全不可信（既非实测也非"确实无数据"）。 + // 调用方必须区分对待：调度侧沿用上一份快照，展示侧跳过本轮落盘。 + Degraded bool } // ReadRuntime 读取渠道×模型的运行时状态（目录刷新循环调用，非请求热路径）。 -// Redis 故障 → 返回乐观默认值（健康满分 + CLOSED），由 last-known 快照语义兜底。 +// Redis 故障 → 返回乐观默认值（健康满分 + CLOSED）并置 Degraded，由调用方决定兜底策略。 func (s *RedisState) ReadRuntime(ctx context.Context, channelID int64, model string) RuntimeReadout { out := RuntimeReadout{SuccEwma: 1} chStr := strconv.FormatInt(channelID, 10) now := time.Now().UnixMilli() - if v, err := g.Redis().Do(ctx, "HMGET", keyHealth+chStr+":"+model, "succ_ewma", "lat_ewma"); err == nil { + if v, err := g.Redis().Do(ctx, "HMGET", keyHealth+chStr+":"+model, "succ_ewma", "lat_ewma"); err != nil { + // 读失败与"字段不存在"必须区分：前者读值不可信（Degraded），后者是确实没有 + // 真实上报（HasHealth=false）。二者都不能当成"这个渠道很健康"落盘或重建快照。 + out.Degraded = true + } else { vals := v.Vars() if len(vals) == 2 { // 必须按字符串显式判空：HMGET 对不存在的 key/字段返回 Redis nil，但 gredis 把它 @@ -656,6 +668,7 @@ func (s *RedisState) ReadRuntime(ctx context.Context, channelID int64, model str if s := strings.TrimSpace(vals[0].String()); s != "" { if f, perr := strconv.ParseFloat(s, 64); perr == nil && f > 0 && f <= 1 { out.SuccEwma = f + out.HasHealth = true } } if s := strings.TrimSpace(vals[1].String()); s != "" { diff --git a/internal/dispatchadapter/state_redis_test.go b/internal/dispatchadapter/state_redis_test.go index ddca776..1c4adb3 100644 --- a/internal/dispatchadapter/state_redis_test.go +++ b/internal/dispatchadapter/state_redis_test.go @@ -533,6 +533,51 @@ func TestReadRuntime异常值防御(t *testing.T) { assert.Equal(t, 0.0, rt.LatEwmaMs) } +// TestReadRuntime区分无数据与读失败 「没读到」与「读失败」必须可区分：前者是真实的 +// 「该模型没有流量」，后者是 Redis 不可用。二者都会回落成默认满分，若调用方无法区分， +// Redis 抖动期间维护快照会把全渠道健康分刷成 100，事故现场被销毁。 +func TestReadRuntime区分无数据与读失败(t *testing.T) { + ctx := context.Background() + s := newTestState(t, nil) + + // 从未上报：读值是乐观默认满分，但必须标记为「无真实数据」 + rt := s.ReadRuntime(ctx, 77, "cold") + assert.Equal(t, 1.0, rt.SuccEwma) + assert.False(t, rt.HasHealth, "无上报的默认满分不得冒充实测值") + assert.False(t, rt.Degraded) + + // 有真实上报 + _, err := g.Redis().Do(ctx, "HSET", keyHealth+"77:hot", "succ_ewma", "0.42", "lat_ewma", "900") + require.NoError(t, err) + rt = s.ReadRuntime(ctx, 77, "hot") + assert.Equal(t, 0.42, rt.SuccEwma) + assert.True(t, rt.HasHealth) + assert.False(t, rt.Degraded) + + // Redis 故障：同样回落满分，但标记 Degraded + mr.SetError("LOADING Redis is loading the dataset in memory") + defer mr.SetError("") + rt = s.ReadRuntime(ctx, 77, "hot") + assert.True(t, rt.Degraded, "读失败必须显式标记") + assert.False(t, rt.HasHealth, "读失败不是「有数据」") + assert.Equal(t, 1.0, rt.SuccEwma) +} + +// TestReadRuntime脏值不算有效数据 精确 0 / 非法值视为无数据，HasHealth 保持 false， +// 避免脏数据以「实测值」身份参与渠道聚合（2026-08-27 全渠道健康分坍缩事故的读侧防线）。 +func TestReadRuntime脏值不算有效数据(t *testing.T) { + ctx := context.Background() + s := newTestState(t, nil) + + for _, bad := range []string{"", "abc", "0"} { + _, err := g.Redis().Do(ctx, "HSET", keyHealth+"78:dirty", "succ_ewma", bad, "lat_ewma", "100") + require.NoError(t, err) + rt := s.ReadRuntime(ctx, 78, "dirty") + assert.False(t, rt.HasHealth, "succ_ewma=%q 不得算作有效实测数据", bad) + assert.Equal(t, 1.0, rt.SuccEwma) + } +} + // --------------------------------------------------------------------------- // 本地降级（修订 R4 + 基线方案 §13） // ---------------------------------------------------------------------------