fix(dispatch): 健康读值区分「无数据」与「读失败」，Redis 降级不再清零健康记忆
ReadRuntime 此前对「该模型没有流量」和「Redis 读失败」返回同样的乐观默认值 succ=1，调用方无从分辨。后果是 catalog.Rebuild 会用这批默认值重建快照，把上一轮 真实的 EWMA 覆盖掉——注释声称的 last-known 语义实际没有兑现，Redis 抖一下全渠道 healthFactor 集体回满分。 RuntimeReadout 新增 HasHealth（确有真实上报，脏值/精确 0/字段缺失都不算）与 Degraded（读失败，读值整体不可信）两个标记；Rebuild 在 Degraded 时沿用上一份快照的 健康读值，无历史可沿用才保持乐观默认（新实例启动即遇 Redis 故障不应把全渠道判死）。
This commit is contained in:
chengqian authored and qian committed
1 parent
3035fa118c
commit
3f3fc43856
4 files changed
+132
-2
No files matched your search
@@ -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）：
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 != "" {
|
||||
|
||||
@@ -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）
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
Reference in new issue
Block a user