diff --git a/internal/dispatchadapter/config_test.go b/internal/dispatchadapter/config_test.go index 63b252f..e9d8531 100644 --- a/internal/dispatchadapter/config_test.go +++ b/internal/dispatchadapter/config_test.go @@ -24,6 +24,14 @@ func TestValidateRoutingPolicyJSON(t *testing.T) { assert.Error(t, ValidateRoutingPolicyJSON(`{"health":{"alpha":-1}}`)) assert.Error(t, ValidateRoutingPolicyJSON(`{"retry":{"failoverBudget":-1}}`)) assert.Error(t, ValidateRoutingPolicyJSON(`{"tierFactors":{"vip":1}}`), "未知层级") + assert.Error(t, ValidateRoutingPolicyJSON(`{"breaker":{"modelFailThreshold":0}}`), "模型级阈值必须 > 0") + assert.NoError(t, ValidateRoutingPolicyJSON(`{"breaker":{"modelFailThreshold":6}}`)) + + // 存量 JSON 缺 modelFailThreshold → 内置默认 4(默认值 + 浅覆盖机制兼容) + pol, err := buildPolicy(`{"breaker":{"failThreshold":5}}`) + assert.NoError(t, err) + assert.Equal(t, 5, pol.Breaker.FailThreshold) + assert.Equal(t, 4, pol.Breaker.ModelFailThreshold) } func TestBuildTenantPolicy(t *testing.T) { diff --git a/internal/dispatchadapter/state_local.go b/internal/dispatchadapter/state_local.go index f285166..03773d5 100644 --- a/internal/dispatchadapter/state_local.go +++ b/internal/dispatchadapter/state_local.go @@ -7,12 +7,18 @@ import ( "github.com/qianfree/team-api/relaykit/dispatch" ) +// probeKey 本地探测限流的复合键:model 空串 = 渠道级,非空 = 渠道×模型级。 +type probeKey struct { + ch int64 + model string +} + // localState Redis 故障时的实例本地降级镜像(基线方案 §13): // 保护仍在、只是不再全局协同。Redis 恢复后自然弃用(无需回填)。 type localState struct { mu sync.Mutex - probeMs map[int64]int64 // channelID → 最近探测放行时间(本地探测限流) + probeMs map[probeKey]int64 // 渠道/渠道×模型 → 最近探测放行时间(本地探测限流) leases map[int64]map[string]int64 // channelID → requestID → 过期时间戳 ms(严格容量渠道) @@ -28,7 +34,7 @@ type localHealth struct { func newLocalState() *localState { return &localState{ - probeMs: make(map[int64]int64), + probeMs: make(map[probeKey]int64), leases: make(map[int64]map[string]int64), credCD: make(map[int64]time.Time), health: make(map[int64]*localHealth), @@ -36,14 +42,16 @@ func newLocalState() *localState { } // tryProbe 本地探测限流:每窗口本实例只放行一个(失去全局协同,仍防打爆)。 -func (l *localState) tryProbe(channelID int64, windowMs int64) bool { +// model 语义与 TryProbeToken 一致:空串 = 渠道级,非空 = 渠道×模型级。 +func (l *localState) tryProbe(channelID int64, model string, windowMs int64) bool { l.mu.Lock() defer l.mu.Unlock() now := time.Now().UnixMilli() - if now-l.probeMs[channelID] < windowMs { + k := probeKey{ch: channelID, model: model} + if now-l.probeMs[k] < windowMs { return false } - l.probeMs[channelID] = now + l.probeMs[k] = now return true } diff --git a/internal/dispatchadapter/state_redis.go b/internal/dispatchadapter/state_redis.go index 9394ca3..29eff59 100644 --- a/internal/dispatchadapter/state_redis.go +++ b/internal/dispatchadapter/state_redis.go @@ -42,8 +42,8 @@ func healthDecayFor(class dispatch.ErrorClass) float64 { switch class { case dispatch.ErrClassRateLimit: return 0.97 // 429 主要喂 softLimit 估计器,基本不伤健康 - case dispatch.ErrClassChannelFatal: - return 0.70 // 重罚,加速权重坍缩 + case dispatch.ErrClassChannelFatal, dispatch.ErrClassModelFatal: + return 0.70 // 重罚,加速权重坍缩(模型级只伤该模型健康分) default: // TRANSIENT / TIMEOUT return 0.93 // 轻罚,瞬时抖动不重伤 } @@ -186,60 +186,114 @@ redis.call('HSET', KEYS[1], 'succ_ewma', succ, 'lat_ewma', lat, 'updated_ms', AR redis.call('PEXPIRE', KEYS[1], ARGV[5]) return 1` -// luaBreakerFail 熔断失败转移(与 dispatch/breaker.go 纯函数规则保持一致,改动需同步)。 -// first_opened_ms 记录本轮故障期的起点(探测失败重置 opened_ms 但不重置它),供自动禁用判定。 -// ARGV: nowMs, windowMs, failThreshold, fatal(0/1), cooldownBaseMs, cooldownMaxMs, keyTtlMs -const luaBreakerFail = ` -local h = KEYS[1] -local state = tonumber(redis.call('HGET', h, 'state') or '0') -local opened = tonumber(redis.call('HGET', h, 'opened_ms') or '0') -local cd = tonumber(redis.call('HGET', h, 'cooldown_ms') or '0') -local now = tonumber(ARGV[1]) -if state == 1 then - if now - opened >= cd then - -- 有效态 HALF_OPEN:探测失败 → 回 OPEN,冷却翻倍封顶 - cd = cd * 2 - if cd <= 0 then cd = tonumber(ARGV[5]) end - if cd > tonumber(ARGV[6]) then cd = tonumber(ARGV[6]) end - redis.call('HSET', h, 'state', 1, 'opened_ms', now, 'cooldown_ms', cd, 'fail_count', 0) +// luaBreakerOutcome 渠道级+模型级熔断转移(单脚本双 key 原子执行,与 dispatch/breaker.go +// 纯函数规则保持一致,改动需同步)。 +// +// 错误按作用域分流(classCode): +// +// 0=成功 双 key 复位(HALF_OPEN 探测成功 → CLOSED / 清窗口计数) +// 1=渠道致命 只喂渠道级,fatal 一次直达 OPEN(渠道级信号:余额尽/Key 耗尽/网络全断) +// 2=模型致命 模型级 fatal 直达;渠道级仅窗口计 1 票(404/模型映射错误是单模型信号) +// 3=瞬时/超时 双 key 窗口计数 +// +// 去重守卫(pre-event):在执行转移前,先读取模型级**旧状态**(state==1 含冷却期满的 +// 有效 HALF_OPEN)。若旧状态为 OPEN,本事件跳过渠道级计数;若旧状态为 CLOSED,本事件 +// 仍给渠道级计数(即使本事件会触发模型级 CLOSED→OPEN 转移)。配合 modelFailThreshold +// < failThreshold,单模型故障对渠道级的贡献封顶在 modelFailThreshold 次,渠道级 OPEN +// 只能由多模型同时恶化达成(跨模型共识)。 +// +// 关键时序:首次打开模型级的事件(第 modelFailThreshold 次故障)本身仍给渠道级贡献 1 票 +// (因守卫判断用的是转移前的旧状态 CLOSED),之后的事件(第 modelFailThreshold+1 次起) +// 才被守卫拦截(旧状态已是 OPEN)。这保证了整渠道故障(如 8 个模型各 404 一次)渠道级能 +// 累积到 8 票而非 0 票。 +// +// KEYS[1]=渠道级 breaker key,KEYS[2]=模型级 breaker key +// ARGV: nowMs, windowMs, chFailThreshold, modelFailThreshold, +// +// cooldownBaseMs, cooldownMaxMs, keyTtlMs, classCode +// +// 返回按位打包:bit0(1)=渠道级 CLOSED→OPEN,bit1(2)=模型级 CLOSED→OPEN(探测失败回 OPEN 不计) +const luaBreakerOutcome = ` +local function success_key(h, now, ttl) + local state = tonumber(redis.call('HGET', h, 'state') or '0') + local opened = tonumber(redis.call('HGET', h, 'opened_ms') or '0') + local cd = tonumber(redis.call('HGET', h, 'cooldown_ms') or '0') + if state == 1 and now - opened >= cd then + redis.call('HSET', h, 'state', 0, 'opened_ms', 0, 'cooldown_ms', 0, 'fail_count', 0, 'recovered_ms', now) + redis.call('HDEL', h, 'first_opened_ms') + elseif state == 0 then + redis.call('HSET', h, 'fail_count', 0) + redis.call('HDEL', h, 'first_opened_ms') end - redis.call('PEXPIRE', h, ARGV[7]) - return -1 + redis.call('PEXPIRE', h, ttl) end --- CLOSED:滑动窗口计数 -local ws = tonumber(redis.call('HGET', h, 'window_start_ms') or '0') -local fc = tonumber(redis.call('HGET', h, 'fail_count') or '0') -if now - ws > tonumber(ARGV[2]) then ws = now; fc = 0 end -fc = fc + 1 -if ARGV[4] == '1' or fc >= tonumber(ARGV[3]) then - redis.call('HSET', h, 'state', 1, 'opened_ms', now, 'cooldown_ms', ARGV[5], 'fail_count', 0, 'window_start_ms', ws) - if redis.call('HEXISTS', h, 'first_opened_ms') == 0 then - redis.call('HSET', h, 'first_opened_ms', now) - end -else - redis.call('HSET', h, 'fail_count', fc, 'window_start_ms', ws) -end -redis.call('PEXPIRE', h, ARGV[7]) -return fc` -// luaBreakerSuccess 熔断成功转移:HALF_OPEN 探测成功 → CLOSED(记录 recovered_ms 供爬坡, -// 清除 first_opened_ms 结束本轮故障期)。 -// ARGV: nowMs, keyTtlMs -const luaBreakerSuccess = ` -local h = KEYS[1] -local state = tonumber(redis.call('HGET', h, 'state') or '0') -local opened = tonumber(redis.call('HGET', h, 'opened_ms') or '0') -local cd = tonumber(redis.call('HGET', h, 'cooldown_ms') or '0') -local now = tonumber(ARGV[1]) -if state == 1 and now - opened >= cd then - redis.call('HSET', h, 'state', 0, 'opened_ms', 0, 'cooldown_ms', 0, 'fail_count', 0, 'recovered_ms', now) - redis.call('HDEL', h, 'first_opened_ms') -elseif state == 0 then - redis.call('HSET', h, 'fail_count', 0) - redis.call('HDEL', h, 'first_opened_ms') +-- 返回 1 表示发生 CLOSED→OPEN 转移(供指标),HALF_OPEN 探测失败回 OPEN 返回 0。 +-- first_opened_ms 记录本轮故障期起点(探测失败不重置),供自动禁用判定。 +local function fail_key(h, threshold, fatal, now, windowMs, cdBase, cdMax, ttl) + local state = tonumber(redis.call('HGET', h, 'state') or '0') + local opened = tonumber(redis.call('HGET', h, 'opened_ms') or '0') + local cd = tonumber(redis.call('HGET', h, 'cooldown_ms') or '0') + if state == 1 then + if now - opened >= cd then + -- 有效态 HALF_OPEN:探测失败 → 回 OPEN,冷却翻倍封顶 + cd = cd * 2 + if cd <= 0 then cd = cdBase end + if cd > cdMax then cd = cdMax end + redis.call('HSET', h, 'state', 1, 'opened_ms', now, 'cooldown_ms', cd, 'fail_count', 0) + end + redis.call('PEXPIRE', h, ttl) + return 0 + end + -- CLOSED:滑动窗口计数 + local ws = tonumber(redis.call('HGET', h, 'window_start_ms') or '0') + local fc = tonumber(redis.call('HGET', h, 'fail_count') or '0') + if now - ws > windowMs then ws = now; fc = 0 end + fc = fc + 1 + if fatal == 1 or fc >= threshold then + redis.call('HSET', h, 'state', 1, 'opened_ms', now, 'cooldown_ms', cdBase, 'fail_count', 0, 'window_start_ms', ws) + if redis.call('HEXISTS', h, 'first_opened_ms') == 0 then + redis.call('HSET', h, 'first_opened_ms', now) + end + redis.call('PEXPIRE', h, ttl) + return 1 + end + redis.call('HSET', h, 'fail_count', fc, 'window_start_ms', ws) + redis.call('PEXPIRE', h, ttl) + return 0 end -redis.call('PEXPIRE', h, ARGV[2]) -return 1` + +local now = tonumber(ARGV[1]) +local windowMs = tonumber(ARGV[2]) +local chTh = tonumber(ARGV[3]) +local mdlTh = tonumber(ARGV[4]) +local cdBase = tonumber(ARGV[5]) +local cdMax = tonumber(ARGV[6]) +local ttl = ARGV[7] +local cls = tonumber(ARGV[8]) + +if cls == 0 then + success_key(KEYS[1], now, ttl) + success_key(KEYS[2], now, ttl) + return 0 +end + +local mOpen = tonumber(redis.call('HGET', KEYS[2], 'state') or '0') == 1 +local ret = 0 +if cls == 1 then + ret = fail_key(KEYS[1], chTh, 1, now, windowMs, cdBase, cdMax, ttl) +elseif cls == 2 then + if not mOpen then + ret = fail_key(KEYS[1], chTh, 0, now, windowMs, cdBase, cdMax, ttl) + end + ret = ret + 2 * fail_key(KEYS[2], mdlTh, 1, now, windowMs, cdBase, cdMax, ttl) +else + if not mOpen then + ret = ret + fail_key(KEYS[1], chTh, 0, now, windowMs, cdBase, cdMax, ttl) + end + ret = ret + 2 * fail_key(KEYS[2], mdlTh, 0, now, windowMs, cdBase, cdMax, ttl) +end +return ret` // luaLimit429Observe softLimit 自动估计(基线方案 §8.2):记录收到 429 时的 inflight 水位, // onset_ewma = onset×0.8 + 当前水位×0.2(首次直接取当前水位)。 @@ -281,30 +335,31 @@ func (s *RedisState) processOutcome(ctx context.Context, o dispatch.Outcome) { keyLimit429+chStr, keyInflight+chStr, now, stateKeyTTLMs) } - breakerKeys := []string{keyBreaker + chStr} - // 模型级熔断只喂模型相关的致命错误(如 404 模型不存在),避免误伤其它模型 - if o.Class == dispatch.ErrClassChannelFatal || o.Success { - breakerKeys = append(breakerKeys, keyBreaker+chStr+":"+o.Model) - } - for _, bk := range breakerKeys { - if o.Success { - _, _ = g.Redis().Do(ctx, "EVAL", luaBreakerSuccess, 1, bk, now, stateKeyTTLMs) - continue - } - if o.Class == dispatch.ErrClassRateLimit { - continue - } - fatal := "0" - if o.Class == dispatch.ErrClassChannelFatal { - fatal = "1" + // 熔断上报(429 属容量信号不喂熔断,改喂上方 softLimit 估计器): + // 渠道级 + 模型级双 key 单脚本原子转移,错误按作用域分流,见 luaBreakerOutcome 注释。 + if o.Success || o.Class != dispatch.ErrClassRateLimit { + classCode := int64(3) // 瞬时/超时 + switch { + case o.Success: + classCode = 0 + case o.Class == dispatch.ErrClassChannelFatal: + classCode = 1 + case o.Class == dispatch.ErrClassModelFatal: + classCode = 2 } b := pol.Breaker - ret, err := g.Redis().Do(ctx, "EVAL", luaBreakerFail, 1, bk, - now, int64(b.WindowSeconds)*1000, b.FailThreshold, fatal, - int64(b.CooldownSeconds)*1000, int64(b.CooldownMaxSeconds)*1000, stateKeyTTLMs) - // CLOSED→OPEN 转移(返回值为窗口失败数且达到阈值/致命直达)计入熔断打开指标 - if err == nil && ret.Int() >= 0 && (fatal == "1" || ret.Int() >= b.FailThreshold) && BreakerOpenHook != nil { - BreakerOpenHook() + ret, err := g.Redis().Do(ctx, "EVAL", luaBreakerOutcome, 2, + keyBreaker+chStr, keyBreaker+chStr+":"+o.Model, + now, int64(b.WindowSeconds)*1000, int64(b.FailThreshold), int64(b.ModelFailThreshold), + int64(b.CooldownSeconds)*1000, int64(b.CooldownMaxSeconds)*1000, stateKeyTTLMs, classCode) + // 按位触发熔断打开指标:bit0=渠道级 CLOSED→OPEN,bit1=模型级(HALF_OPEN 探测失败回 OPEN 不计) + if err == nil && BreakerOpenHook != nil { + if ret.Int()&1 != 0 { + BreakerOpenHook() + } + if ret.Int()&2 != 0 { + BreakerOpenHook() + } } } } @@ -317,8 +372,15 @@ redis.call('HDEL', KEYS[1], 'first_opened_ms') redis.call('PEXPIRE', KEYS[1], ARGV[2]) return 1` -// MarkChannelRecovered 渠道被运营手动启用/恢复时调用:复位熔断并开启爬坡窗口, +// MarkChannelRecovered 渠道被运营手动启用/恢复时调用:复位渠道级熔断并开启爬坡窗口, // 避免恢复瞬间被 HRW 一次性灌满流量(基线方案 §4.5 rampFactor)。 +// +// 只复位渠道级,不动模型级熔断:调用方(管理后台启用/auto-test 恢复)只验证了 test_model +// 一个模型,不能证明其它模型已恢复,模型级隔离是真实流量失败挣来的证据,保留更安全。 +// +// 运维影响:启用渠道后,已打开的模型级熔断需自然恢复(冷却封顶 300s + 探测窗口 10s, +// 最长 5 分钟)。期间该模型请求仍需等待探测窗口放行,可能出现「渠道已启用但部分模型 +// 仍不可用」的现象。需要立即全复位(包括所有模型级熔断 + 健康 EWMA)时,使用 ResetHealth。 func (s *RedisState) MarkChannelRecovered(ctx context.Context, channelID int64) { if _, err := g.Redis().Do(ctx, "EVAL", luaBreakerManualReset, 1, keyBreaker+strconv.FormatInt(channelID, 10), time.Now().UnixMilli(), stateKeyTTLMs); err != nil { @@ -382,14 +444,18 @@ if now - ps < tonumber(ARGV[2]) then return 0 end redis.call('HSET', h, 'probe_ms', now) return 1` -// TryProbeToken 尝试获取 HALF_OPEN 探测令牌。Redis 故障 → 实例本地限流兜底。 -func (s *RedisState) TryProbeToken(ctx context.Context, channelID int64) bool { +// TryProbeToken 尝试获取 HALF_OPEN 探测令牌(model 空串 = 渠道级熔断 key, +// 非空 = 渠道×模型级熔断 key)。Redis 故障 → 实例本地限流兜底。 +func (s *RedisState) TryProbeToken(ctx context.Context, channelID int64, model string) bool { pol := s.policy() + key := keyBreaker + strconv.FormatInt(channelID, 10) + if model != "" { + key += ":" + model + } v, err := g.Redis().Do(ctx, "EVAL", luaProbeToken, 1, - keyBreaker+strconv.FormatInt(channelID, 10), - time.Now().UnixMilli(), int64(pol.Breaker.ProbeWindowSeconds)*1000) + key, time.Now().UnixMilli(), int64(pol.Breaker.ProbeWindowSeconds)*1000) if err != nil { - return s.local.tryProbe(channelID, int64(pol.Breaker.ProbeWindowSeconds)*1000) + return s.local.tryProbe(channelID, model, int64(pol.Breaker.ProbeWindowSeconds)*1000) } return v.Int() == 1 } diff --git a/internal/dispatchadapter/state_redis_test.go b/internal/dispatchadapter/state_redis_test.go index 1b2470d..349d908 100644 --- a/internal/dispatchadapter/state_redis_test.go +++ b/internal/dispatchadapter/state_redis_test.go @@ -50,7 +50,8 @@ func newTestState(t *testing.T, mutate func(*dispatch.RoutingPolicy)) *RedisStat func Test并发健康EWMA原子性(t *testing.T) { ctx := context.Background() s := newTestState(t, func(p *dispatch.RoutingPolicy) { - p.Breaker.FailThreshold = 1000 // 阈值调高,只验证计数不触发熔断 + p.Breaker.FailThreshold = 1000 // 阈值调高,只验证计数不触发熔断 + p.Breaker.ModelFailThreshold = 1000 // 模型级阈值同步调高,避免打开后渠道级停止计数 }) var wg sync.WaitGroup @@ -88,6 +89,7 @@ func Test健康衰减分档(t *testing.T) { {2, dispatch.ErrClassTimeout, 0.93}, {3, dispatch.ErrClassRateLimit, 0.97}, {4, dispatch.ErrClassChannelFatal, 0.70}, + {5, dispatch.ErrClassModelFatal, 0.70}, } for _, c := range cases { s.processOutcome(ctx, dispatch.Outcome{ChannelID: c.channel, Model: "m", Success: false, Class: c.class}) @@ -119,7 +121,9 @@ func Test限流不喂熔断窗口(t *testing.T) { func Test熔断转移与探测令牌(t *testing.T) { ctx := context.Background() - s := newTestState(t, nil) // failThreshold=8 + s := newTestState(t, func(p *dispatch.RoutingPolicy) { + p.Breaker.ModelFailThreshold = 100 // 调高模型级阈值,聚焦验证渠道级 8 次语义 + }) // 8 次瞬时失败 → 渠道级熔断 OPEN for range 8 { @@ -129,7 +133,7 @@ func Test熔断转移与探测令牌(t *testing.T) { assert.Equal(t, dispatch.BreakerOpen, rt.Breaker) // 冷却期内探测令牌不放行 - assert.False(t, s.TryProbeToken(ctx, 7)) + assert.False(t, s.TryProbeToken(ctx, 7, "")) // 手动把 opened_ms 拨回 31s 前(冷却 30s)→ 有效态 HALF_OPEN _, err := g.Redis().Do(ctx, "HSET", keyBreaker+"7", "opened_ms", time.Now().UnixMilli()-31_000) @@ -138,8 +142,8 @@ func Test熔断转移与探测令牌(t *testing.T) { assert.Equal(t, dispatch.BreakerHalfOpen, rt.Breaker) // 探测令牌:每窗口全局只放行一个 - assert.True(t, s.TryProbeToken(ctx, 7)) - assert.False(t, s.TryProbeToken(ctx, 7), "同窗口第二个探测必须拒绝") + assert.True(t, s.TryProbeToken(ctx, 7, "")) + assert.False(t, s.TryProbeToken(ctx, 7, ""), "同窗口第二个探测必须拒绝") // 探测成功 → CLOSED + recovered_ms(供爬坡因子) s.processOutcome(ctx, dispatch.Outcome{ChannelID: 7, Model: "m", Success: true, LatencyMs: 100}) @@ -168,6 +172,131 @@ func Test熔断探测失败冷却翻倍(t *testing.T) { assert.Equal(t, int64(60_000), vals[1].Int64(), "冷却翻倍 30s→60s") } +// --------------------------------------------------------------------------- +// 模型级隔离(错误作用域分流 + 去重守卫) +// --------------------------------------------------------------------------- + +// Test模型致命一次直达模型级 404/模型不存在:模型级 fatal 直达 OPEN, +// 渠道级仅窗口计 1 票;模型级已 OPEN 后同模型后续失败不再污染渠道级。 +func Test模型致命一次直达模型级(t *testing.T) { + ctx := context.Background() + s := newTestState(t, nil) + + s.processOutcome(ctx, dispatch.Outcome{ChannelID: 62, Model: "gone", Success: false, Class: dispatch.ErrClassModelFatal}) + rt := s.ReadRuntime(ctx, 62, "gone") + assert.Equal(t, dispatch.BreakerOpen, rt.ModelBreaker, "模型级 fatal 一次直达") + assert.Equal(t, dispatch.BreakerClosed, rt.Breaker, "渠道级不被拉黑") + + v, err := g.Redis().Do(ctx, "HGET", keyBreaker+"62", "fail_count") + require.NoError(t, err) + assert.Equal(t, 1, v.Int(), "渠道级仅窗口计 1 票") + + // 去重守卫:模型级 OPEN(含有效 HALF_OPEN)后,本模型失败不再喂渠道级 + for range 5 { + s.processOutcome(ctx, dispatch.Outcome{ChannelID: 62, Model: "gone", Success: false, Class: dispatch.ErrClassModelFatal}) + } + v, _ = g.Redis().Do(ctx, "HGET", keyBreaker+"62", "fail_count") + assert.Equal(t, 1, v.Int(), "模型 OPEN 后渠道级不再计数") +} + +// Test模型级窗口阈值与渠道级隔离 单模型慢性故障(瞬时类):模型级阈值 4 先打开, +// 渠道级贡献封顶在 4 票(< 8),渠道级永不熔断——单模型故障不误伤其它模型。 +func Test模型级窗口阈值与渠道级隔离(t *testing.T) { + ctx := context.Background() + s := newTestState(t, nil) // failThreshold=8, modelFailThreshold=4 + + for range 3 { + s.processOutcome(ctx, dispatch.Outcome{ChannelID: 60, Model: "m", Success: false, Class: dispatch.ErrClassTransient}) + } + v, err := g.Redis().Do(ctx, "HGET", keyBreaker+"60", "fail_count") + require.NoError(t, err) + assert.Equal(t, 3, v.Int(), "渠道级窗口计数 3") + v, _ = g.Redis().Do(ctx, "HGET", keyBreaker+"60:m", "fail_count") + assert.Equal(t, 3, v.Int(), "模型级窗口计数 3") + + // 第 4 次触发模型级打开(先渠道计数后模型转移,本事件贡献第 4 票),此后不再计数 + for range 10 { + s.processOutcome(ctx, dispatch.Outcome{ChannelID: 60, Model: "m", Success: false, Class: dispatch.ErrClassTransient}) + } + rt := s.ReadRuntime(ctx, 60, "m") + assert.Equal(t, dispatch.BreakerOpen, rt.ModelBreaker, "模型级达阈值 4 打开") + assert.Equal(t, dispatch.BreakerClosed, rt.Breaker, "渠道级停在 4 票不熔断") + v, _ = g.Redis().Do(ctx, "HGET", keyBreaker+"60", "fail_count") + assert.Equal(t, 4, v.Int(), "单模型对渠道级贡献封顶在 modelFailThreshold") +} + +// Test跨模型共识渠道级熔断 两个模型各恶化 4 次(4+4=8)→ 渠道级 OPEN: +// 渠道级熔断保留给真正的多模型同时恶化(整渠道故障)。 +func Test跨模型共识渠道级熔断(t *testing.T) { + ctx := context.Background() + s := newTestState(t, nil) + + for _, m := range []string{"m1", "m2"} { + for range 4 { + s.processOutcome(ctx, dispatch.Outcome{ChannelID: 61, Model: m, Success: false, Class: dispatch.ErrClassTransient}) + } + } + rt := s.ReadRuntime(ctx, 61, "m1") + assert.Equal(t, dispatch.BreakerOpen, rt.Breaker, "跨模型 4+4=8 达成共识") + assert.Equal(t, dispatch.BreakerOpen, rt.ModelBreaker, "两模型各自也达阈值 4") +} + +// Test渠道致命不喂模型级 402/Key 耗尽等渠道级信号:fatal 只打渠道级,模型级 key 不被触碰 +// (渠道恢复后模型无残留熔断)。 +func Test渠道致命不喂模型级(t *testing.T) { + ctx := context.Background() + s := newTestState(t, nil) + + s.processOutcome(ctx, dispatch.Outcome{ChannelID: 63, Model: "m", Success: false, Class: dispatch.ErrClassChannelFatal}) + rt := s.ReadRuntime(ctx, 63, "m") + assert.Equal(t, dispatch.BreakerOpen, rt.Breaker) + + v, err := g.Redis().Do(ctx, "EXISTS", keyBreaker+"63:m") + require.NoError(t, err) + assert.Equal(t, 0, v.Int(), "渠道级致命不创建/触碰模型级 key") +} + +// Test模型级探测成功复位 模型级 OPEN 冷却期满 → HALF_OPEN → 成功 → CLOSED + recovered_ms +// (模型级熔断可靠流量探测恢复,不再死锁到 24h TTL)。 +func Test模型级探测成功复位(t *testing.T) { + ctx := context.Background() + s := newTestState(t, nil) + + s.processOutcome(ctx, dispatch.Outcome{ChannelID: 64, Model: "m", Success: false, Class: dispatch.ErrClassModelFatal}) + _, _ = g.Redis().Do(ctx, "HSET", keyBreaker+"64:m", "opened_ms", time.Now().UnixMilli()-31_000) + + s.processOutcome(ctx, dispatch.Outcome{ChannelID: 64, Model: "m", Success: true, LatencyMs: 100}) + rt := s.ReadRuntime(ctx, 64, "m") + assert.Equal(t, dispatch.BreakerClosed, rt.ModelBreaker) + v, _ := g.Redis().Do(ctx, "HGET", keyBreaker+"64:m", "recovered_ms") + assert.Greater(t, v.Int64(), int64(0), "模型级恢复时间必须记录") +} + +// Test模型级探测令牌与探测失败翻倍 模型级令牌按渠道×模型 key 独立限流; +// 冷却期满后的失败走 HALF_OPEN 探测失败分支(模型级回 OPEN 冷却翻倍,渠道级不计数)。 +func Test模型级探测令牌与探测失败翻倍(t *testing.T) { + ctx := context.Background() + s := newTestState(t, nil) + + s.processOutcome(ctx, dispatch.Outcome{ChannelID: 65, Model: "m", Success: false, Class: dispatch.ErrClassModelFatal}) + _, _ = g.Redis().Do(ctx, "HSET", keyBreaker+"65:m", "opened_ms", time.Now().UnixMilli()-31_000) + + // 模型级令牌:每窗口放行一个;渠道级 CLOSED 时渠道令牌不放行(不再误拒探测) + assert.True(t, s.TryProbeToken(ctx, 65, "m")) + assert.False(t, s.TryProbeToken(ctx, 65, "m"), "同窗口第二个模型级探测必须拒绝") + assert.False(t, s.TryProbeToken(ctx, 65, ""), "渠道级 CLOSED 不放行渠道令牌") + + // 探测失败:模型级回 OPEN 冷却翻倍,渠道级不新增计数(保持首次 ModelFatal 贡献的 1 票) + s.processOutcome(ctx, dispatch.Outcome{ChannelID: 65, Model: "m", Success: false, Class: dispatch.ErrClassTransient}) + v, err := g.Redis().Do(ctx, "HMGET", keyBreaker+"65:m", "state", "cooldown_ms") + require.NoError(t, err) + vals := v.Vars() + assert.Equal(t, 1, vals[0].Int(), "模型级探测失败回 OPEN") + assert.Equal(t, int64(60_000), vals[1].Int64(), "冷却翻倍 30s→60s") + v, _ = g.Redis().Do(ctx, "HGET", keyBreaker+"65", "fail_count") + assert.Equal(t, 1, v.Int(), "渠道级不因探测失败新增计数") +} + func Test绑定CAS与反向索引(t *testing.T) { ctx := context.Background() s := newTestState(t, nil) @@ -334,9 +463,10 @@ func TestLocal严格容量租约(t *testing.T) { func TestLocal探测限流(t *testing.T) { l := newLocalState() - assert.True(t, l.tryProbe(1, 10_000)) - assert.False(t, l.tryProbe(1, 10_000), "同窗口本实例只放行一个") - assert.True(t, l.tryProbe(2, 10_000), "不同渠道互不影响") + assert.True(t, l.tryProbe(1, "", 10_000)) + assert.False(t, l.tryProbe(1, "", 10_000), "同窗口本实例只放行一个") + assert.True(t, l.tryProbe(2, "", 10_000), "不同渠道互不影响") + assert.True(t, l.tryProbe(1, "gpt-4o", 10_000), "同渠道的模型级令牌与渠道级互不影响") } func TestLocal凭证冷却过期(t *testing.T) { @@ -393,4 +523,11 @@ func Test手动恢复复位熔断并开启爬坡(t *testing.T) { assert.Equal(t, dispatch.BreakerClosed, rt.Breaker) assert.Greater(t, rt.RecoveredMs, int64(0)) assert.Zero(t, rt.FirstOpenedMs) + + // 模型级熔断不受 MarkChannelRecovered 影响(只验证了 test_model,不能证明其它模型恢复; + // 模型级靠流量探测自行恢复,决策见 MarkChannelRecovered 注释) + s.processOutcome(ctx, dispatch.Outcome{ChannelID: 50, Model: "m2", Success: false, Class: dispatch.ErrClassModelFatal}) + rt = s.ReadRuntime(ctx, 50, "m2") + assert.Equal(t, dispatch.BreakerOpen, rt.ModelBreaker, "模型级熔断必须保留") + assert.Equal(t, dispatch.BreakerClosed, rt.Breaker, "渠道级保持复位后状态") } diff --git a/relaykit/dispatch/classify.go b/relaykit/dispatch/classify.go index db8d2f3..ee0c6ed 100644 --- a/relaykit/dispatch/classify.go +++ b/relaykit/dispatch/classify.go @@ -18,6 +18,7 @@ const ( ErrClassCredential // 凭证错误,先冷却当前 Key 并轮换,耗尽后升级 CHANNEL_FATAL ErrClassChannelFatal // 渠道致命错误:零原地重试,立即 failover + 熔断计数直达 ErrClassTimeout // 超时(含 504):不原地重试,按可重放性决定 failover + ErrClassModelFatal // 模型级致命(404/模型不存在/模型映射失败):立即 failover + 模型级熔断直达,不归因渠道整体 ) // String 实现 fmt.Stringer,用于日志与指标标签。 @@ -33,6 +34,8 @@ func (c ErrorClass) String() string { return "credential" case ErrClassChannelFatal: return "channel_fatal" + case ErrClassModelFatal: + return "model_fatal" case ErrClassTimeout: return "timeout" default: @@ -97,11 +100,14 @@ func classifyNewAPIError(e *types.NewAPIError) (ErrorClass, bool) { case types.ErrorCodeChannelNoAvailableKey, types.ErrorCodeChannelParamOverrideInvalid, types.ErrorCodeChannelHeaderOverrideInvalid, - types.ErrorCodeChannelModelMappedError, - types.ErrorCodeChannelAwsClientError, - types.ErrorCodeModelNotFound: + types.ErrorCodeChannelAwsClientError: return ErrClassChannelFatal, true + // 模型级致命:只影响渠道下的单个模型,仅喂渠道×模型级熔断(不归因渠道整体) + case types.ErrorCodeChannelModelMappedError, + types.ErrorCodeModelNotFound: + return ErrClassModelFatal, true + // 超时 case types.ErrorCodeChannelResponseTimeExceeded: return ErrClassTimeout, true @@ -136,8 +142,8 @@ func classifyStatusCode(code int) ErrorClass { // 上游余额耗尽 return ErrClassChannelFatal case 404: - // 模型在该渠道不存在:换渠道可能有效(喂给渠道×模型级熔断) - return ErrClassChannelFatal + // 模型在该渠道不存在:换渠道可能有效(喂给渠道×模型级熔断,不伤渠道其它模型) + return ErrClassModelFatal case 408: return ErrClassTimeout case 429: diff --git a/relaykit/dispatch/classify_test.go b/relaykit/dispatch/classify_test.go index 3916d7f..6c5e5a8 100644 --- a/relaykit/dispatch/classify_test.go +++ b/relaykit/dispatch/classify_test.go @@ -20,7 +20,7 @@ func TestClassify_状态码全表(t *testing.T) { {401, ErrClassCredential}, {402, ErrClassChannelFatal}, {403, ErrClassCredential}, - {404, ErrClassChannelFatal}, + {404, ErrClassModelFatal}, // 模型不存在:单模型信号,喂模型级熔断 {408, ErrClassTimeout}, {409, ErrClassClient}, {413, ErrClassClient}, @@ -50,7 +50,8 @@ func TestClassify_NewAPIError错误码(t *testing.T) { {"invalid_key 归凭证", types.ErrorCodeChannelInvalidKey, ErrClassCredential}, {"no_available_key 归渠道级", types.ErrorCodeChannelNoAvailableKey, ErrClassChannelFatal}, {"protocol_mismatch 归渠道级(驱动换渠道)", types.ErrorCodeChannelProtocolMismatch, ErrClassChannelFatal}, - {"model_not_found 归渠道级", types.ErrorCodeModelNotFound, ErrClassChannelFatal}, + {"model_not_found 归模型级", types.ErrorCodeModelNotFound, ErrClassModelFatal}, + {"model_mapped_error 归模型级", types.ErrorCodeChannelModelMappedError, ErrClassModelFatal}, {"response_time_exceeded 归超时", types.ErrorCodeChannelResponseTimeExceeded, ErrClassTimeout}, {"aws_client_error 归渠道级", types.ErrorCodeChannelAwsClientError, ErrClassChannelFatal}, {"prompt_blocked 归客户端", types.ErrorCodePromptBlocked, ErrClassClient}, @@ -98,5 +99,6 @@ func TestClassify_无错误(t *testing.T) { func TestErrorClass_String(t *testing.T) { assert.Equal(t, "credential", ErrClassCredential.String()) assert.Equal(t, "timeout", ErrClassTimeout.String()) + assert.Equal(t, "model_fatal", ErrClassModelFatal.String()) assert.Equal(t, "none", ErrClassNone.String()) } diff --git a/relaykit/dispatch/coordinator.go b/relaykit/dispatch/coordinator.go index 8e3e9af..20ec2ee 100644 --- a/relaykit/dispatch/coordinator.go +++ b/relaykit/dispatch/coordinator.go @@ -286,9 +286,11 @@ func (s *RouteSession) selectChannel(ctx context.Context) *Decision { return nil } - // HALF_OPEN 渠道被选中:取探测令牌,每窗口全局只放行一个 - if _, isHalfOpen := halfOpen[chosen.ID]; isHalfOpen { - if !s.co.state.TryProbeToken(ctx, chosen.ID) { + // HALF_OPEN 渠道被选中:按熔断级别取探测令牌,每窗口全局只放行一个。 + // 先取更具体的模型级令牌(被拒不消耗渠道级令牌),再取渠道级;任一被拒即排除重选。 + if need, isHalfOpen := halfOpen[chosen.ID]; isHalfOpen { + if (need[1] && !s.co.state.TryProbeToken(ctx, chosen.ID, s.profile.Model)) || + (need[0] && !s.co.state.TryProbeToken(ctx, chosen.ID, "")) { roundExcluded[chosen.ID] = struct{}{} probeDenied++ continue @@ -336,8 +338,8 @@ func (s *RouteSession) selectChannel(ctx context.Context) *Decision { // // 注意:Channel.Breaker / ModelBreaker 由目录适配层提供,已做过冷却期的 // 惰性 OPEN→HALF_OPEN 判定(EffectiveBreakerState)。 -func (s *RouteSession) buildCandidates(snapshot []Channel, roundExcluded map[int64]struct{}, pol *RoutingPolicy) ([]ScoredChannel, map[int64]struct{}, ExclusionStats) { - halfOpen := make(map[int64]struct{}) +func (s *RouteSession) buildCandidates(snapshot []Channel, roundExcluded map[int64]struct{}, pol *RoutingPolicy) ([]ScoredChannel, map[int64][2]bool, ExclusionStats) { + halfOpen := make(map[int64][2]bool) excl := ExclusionStats{} var scored []ScoredChannel var zeroTier []Channel // 因 tierFactor=0 被排除的渠道,供扩组兜底 @@ -355,7 +357,8 @@ func (s *RouteSession) buildCandidates(snapshot []Channel, roundExcluded map[int continue } if ch.Breaker == BreakerHalfOpen || ch.ModelBreaker == BreakerHalfOpen { - halfOpen[ch.ID] = struct{}{} + // [0]=渠道级半开、[1]=模型级半开(快照按模型行展开,每渠道至多一行) + halfOpen[ch.ID] = [2]bool{ch.Breaker == BreakerHalfOpen, ch.ModelBreaker == BreakerHalfOpen} } if tierFactor(ch.Tier, pol) <= 0 { zeroTier = append(zeroTier, ch) diff --git a/relaykit/dispatch/coordinator_test.go b/relaykit/dispatch/coordinator_test.go index fdeed58..0e2897b 100644 --- a/relaykit/dispatch/coordinator_test.go +++ b/relaykit/dispatch/coordinator_test.go @@ -22,13 +22,19 @@ func (f *fakeCatalog) Snapshot(_ context.Context, _ int64, _ string, _ []int64) return out } +// probeTokenKey fakeState 探测令牌的复合键:model 空串 = 渠道级,非空 = 渠道×模型级。 +type probeTokenKey struct { + ch int64 + model string +} + type fakeState struct { bindings map[string]int64 setBinds int touchBinds int invalidated []int64 outcomes []Outcome - probeGrant map[int64]bool // 默认拒绝 + probeGrant map[probeTokenKey]bool // 默认拒绝 leaseDeny map[int64]bool acquired map[int64]int released map[int64]int @@ -39,7 +45,7 @@ type fakeState struct { func newFakeState() *fakeState { return &fakeState{ bindings: map[string]int64{}, - probeGrant: map[int64]bool{}, + probeGrant: map[probeTokenKey]bool{}, leaseDeny: map[int64]bool{}, acquired: map[int64]int{}, released: map[int64]int{}, @@ -60,8 +66,10 @@ func (f *fakeState) TouchBinding(_ context.Context, _ string, _ time.Duration) { func (f *fakeState) InvalidateChannelBindings(_ context.Context, channelID int64) { f.invalidated = append(f.invalidated, channelID) } -func (f *fakeState) ReportOutcome(o Outcome) { f.outcomes = append(f.outcomes, o) } -func (f *fakeState) TryProbeToken(_ context.Context, id int64) bool { return f.probeGrant[id] } +func (f *fakeState) ReportOutcome(o Outcome) { f.outcomes = append(f.outcomes, o) } +func (f *fakeState) TryProbeToken(_ context.Context, id int64, model string) bool { + return f.probeGrant[probeTokenKey{id, model}] +} func (f *fakeState) AcquireLease(_ context.Context, id int64, _ int, _ string) bool { if f.leaseDeny[id] { return false @@ -352,7 +360,7 @@ func TestCoordinator_熔断排除与探测放行(t *testing.T) { // 探测令牌放行:HALF_OPEN 渠道作为探测请求 state2 := newFakeState() - state2.probeGrant[2] = true + state2.probeGrant[probeTokenKey{2, ""}] = true co2, _ := newTestCoordinator(state2, open, half, normal) s2 := co2.Route(ctx, testProfile()) d2 := s2.Next(ctx) @@ -375,6 +383,66 @@ func TestCoordinator_模型级熔断排除(t *testing.T) { assert.Equal(t, int64(2), d.Channel.ID) } +// TestCoordinator_模型级半开探测 模型级熔断冷却期满(HALF_OPEN)走模型级探测令牌: +// 渠道级熔断 CLOSED 时不再被渠道级令牌误拒(修复模型级熔断无法靠流量恢复的死锁)。 +func TestCoordinator_模型级半开探测(t *testing.T) { + ctx := context.Background() + + modelHalf := healthyChannel(1, TierPrimary, 100) + modelHalf.ModelBreaker = BreakerHalfOpen // 仅模型级半开,渠道级 CLOSED + normal := healthyChannel(2, TierPrimary, 0.001) + + // 模型级令牌放行 → 作为探测请求选中 + state := newFakeState() + state.probeGrant[probeTokenKey{1, "gpt-4o"}] = true + co, _ := newTestCoordinator(state, modelHalf, normal) + s := co.Route(ctx, testProfile()) + d := s.Next(ctx) + require.NotNil(t, d) + assert.Equal(t, int64(1), d.Channel.ID) + assert.Equal(t, ReasonProbe, d.Reason) + + // 模型级令牌被拒 → 排除落到普通渠道(每探测窗口只放行一个) + state2 := newFakeState() + co2, _ := newTestCoordinator(state2, modelHalf, normal) + s2 := co2.Route(ctx, testProfile()) + d2 := s2.Next(ctx) + require.NotNil(t, d2) + assert.Equal(t, int64(2), d2.Channel.ID) + assert.Equal(t, 1, d2.Excluded.Breaker, "模型级探测拒绝计入 Breaker 排除") +} + +// TestCoordinator_双级别半开组合 渠道级与模型级同时 HALF_OPEN:两枚令牌都取, +// 任一被拒即排除;模型级令牌先取(被拒不消耗渠道级令牌)。 +func TestCoordinator_双级别半开组合(t *testing.T) { + ctx := context.Background() + + both := healthyChannel(1, TierPrimary, 100) + both.Breaker = BreakerHalfOpen + both.ModelBreaker = BreakerHalfOpen + normal := healthyChannel(2, TierPrimary, 0.001) + + // 模型级放行 + 渠道级拒绝 → 排除 + state := newFakeState() + state.probeGrant[probeTokenKey{1, "gpt-4o"}] = true + co, _ := newTestCoordinator(state, both, normal) + s := co.Route(ctx, testProfile()) + d := s.Next(ctx) + require.NotNil(t, d) + assert.Equal(t, int64(2), d.Channel.ID, "渠道级令牌被拒时整体排除") + + // 双放行 → 探测请求 + state2 := newFakeState() + state2.probeGrant[probeTokenKey{1, "gpt-4o"}] = true + state2.probeGrant[probeTokenKey{1, ""}] = true + co2, _ := newTestCoordinator(state2, both, normal) + s2 := co2.Route(ctx, testProfile()) + d2 := s2.Next(ctx) + require.NotNil(t, d2) + assert.Equal(t, int64(1), d2.Channel.ID) + assert.Equal(t, ReasonProbe, d2.Reason) +} + func TestCoordinator_tier扩组兜底(t *testing.T) { ctx := context.Background() state := newFakeState() diff --git a/relaykit/dispatch/policy.go b/relaykit/dispatch/policy.go index 9ea39e2..6eeec82 100644 --- a/relaykit/dispatch/policy.go +++ b/relaykit/dispatch/policy.go @@ -100,7 +100,8 @@ type RetryPolicy struct { // BreakerPolicy 熔断策略。 type BreakerPolicy struct { WindowSeconds int `json:"windowSeconds"` // 滑动窗口,默认 60 - FailThreshold int `json:"failThreshold"` // 窗口失败阈值,默认 8 + FailThreshold int `json:"failThreshold"` // 渠道级窗口失败阈值,默认 8 + ModelFailThreshold int `json:"modelFailThreshold"` // 模型级窗口失败阈值,默认 4(低于渠道级,单模型故障先在模型级熔断隔离) CooldownSeconds int `json:"cooldownSeconds"` // OPEN 冷却起始,默认 30 CooldownMaxSeconds int `json:"cooldownMaxSeconds"` // 冷却上限,默认 300 ProbeWindowSeconds int `json:"probeWindowSeconds"` // HALF_OPEN 探测窗口,默认 10 @@ -151,6 +152,7 @@ func DefaultRoutingPolicy() *RoutingPolicy { p.Breaker = BreakerPolicy{ WindowSeconds: 60, FailThreshold: 8, + ModelFailThreshold: 4, CooldownSeconds: 30, CooldownMaxSeconds: 300, ProbeWindowSeconds: 10, @@ -218,8 +220,8 @@ func (p *RoutingPolicy) Validate() error { return fmt.Errorf("retry 时间参数非法") } b := p.Breaker - if b.WindowSeconds <= 0 || b.FailThreshold <= 0 || b.CooldownSeconds <= 0 || - b.CooldownMaxSeconds < b.CooldownSeconds || b.ProbeWindowSeconds <= 0 { + if b.WindowSeconds <= 0 || b.FailThreshold <= 0 || b.ModelFailThreshold <= 0 || + b.CooldownSeconds <= 0 || b.CooldownMaxSeconds < b.CooldownSeconds || b.ProbeWindowSeconds <= 0 { return fmt.Errorf("breaker 参数非法") } if p.Degrade.MaxReplicas <= 0 { @@ -335,7 +337,7 @@ func Decide(class ErrorClass, delivery DeliveryState, replay Replayability, retr } return failoverOrAbort() - case ErrClassChannelFatal: + case ErrClassChannelFatal, ErrClassModelFatal: return failoverOrAbort() } return DecisionAbort, 0 diff --git a/relaykit/dispatch/policy_test.go b/relaykit/dispatch/policy_test.go index 74b66db..aa05d8a 100644 --- a/relaykit/dispatch/policy_test.go +++ b/relaykit/dispatch/policy_test.go @@ -14,7 +14,7 @@ func TestDecide_硬规则优先(t *testing.T) { p := retryPol() // 已向客户端写出响应:任何类别都终止 - for _, cls := range []ErrorClass{ErrClassTransient, ErrClassRateLimit, ErrClassCredential, ErrClassChannelFatal, ErrClassTimeout} { + for _, cls := range []ErrorClass{ErrClassTransient, ErrClassRateLimit, ErrClassCredential, ErrClassChannelFatal, ErrClassModelFatal, ErrClassTimeout} { d, _ := Decide(cls, DeliveryResponseStarted, ReplaySafe, 0, AttemptState{}, p) assert.Equal(t, DecisionAbort, d, "ResponseStarted 必须 Abort: %s", cls) @@ -23,7 +23,7 @@ func TestDecide_硬规则优先(t *testing.T) { } // ReplayUnsafe + MaybeSent 无条件 Abort(状态码/类别无关) - for _, cls := range []ErrorClass{ErrClassTransient, ErrClassRateLimit, ErrClassCredential, ErrClassChannelFatal, ErrClassTimeout} { + for _, cls := range []ErrorClass{ErrClassTransient, ErrClassRateLimit, ErrClassCredential, ErrClassChannelFatal, ErrClassModelFatal, ErrClassTimeout} { d, _ := Decide(cls, DeliveryMaybeSent, ReplayUnsafe, 0, AttemptState{HasAlternateKey: true}, p) assert.Equal(t, DecisionAbort, d, "ReplayUnsafe+MaybeSent 必须 Abort: %s", cls) } @@ -126,6 +126,17 @@ func TestDecide_渠道致命零原地(t *testing.T) { assert.Equal(t, DecisionFailover, d) } +func TestDecide_模型致命直接failover(t *testing.T) { + p := retryPol() + // 模型致命与渠道致命同语义:零原地重试、立即 failover + d, _ := Decide(ErrClassModelFatal, DeliveryResponseReceived, ReplayCostly, 0, AttemptState{}, p) + assert.Equal(t, DecisionFailover, d) + + // failover 预算耗尽 → Abort + d, _ = Decide(ErrClassModelFatal, DeliveryResponseReceived, ReplayCostly, 0, AttemptState{FailoverUsed: 2}, p) + assert.Equal(t, DecisionAbort, d) +} + func TestDecide_ReplayUnsafe预算收紧(t *testing.T) { p := retryPol() // unsafe 预算默认 0/0 @@ -149,7 +160,7 @@ func TestDecide_ReplayUnsafe预算收紧(t *testing.T) { // 用随机错误序列驱动 FSM,断言任何序列都不会突破预算且必然终止。 func TestDecide_属性_任意错误序列不突破预算(t *testing.T) { p := retryPol() - classes := []ErrorClass{ErrClassClient, ErrClassTransient, ErrClassRateLimit, ErrClassCredential, ErrClassChannelFatal, ErrClassTimeout} + classes := []ErrorClass{ErrClassClient, ErrClassTransient, ErrClassRateLimit, ErrClassCredential, ErrClassChannelFatal, ErrClassModelFatal, ErrClassTimeout} deliveries := []DeliveryState{DeliveryNotSent, DeliveryMaybeSent, DeliveryResponseReceived} replays := []Replayability{ReplaySafe, ReplayCostly, ReplayUnsafe} @@ -230,6 +241,7 @@ func TestRoutingPolicy_校验拒绝非法配置(t *testing.T) { {"负预算", mutate(func(p *RoutingPolicy) { p.Retry.FailoverBudget = -1 })}, {"退避上限小于基数", mutate(func(p *RoutingPolicy) { p.Retry.BackoffMaxMs = 10 })}, {"熔断阈值非正", mutate(func(p *RoutingPolicy) { p.Breaker.FailThreshold = 0 })}, + {"模型级熔断阈值非正", mutate(func(p *RoutingPolicy) { p.Breaker.ModelFailThreshold = 0 })}, {"冷却上限小于起始", mutate(func(p *RoutingPolicy) { p.Breaker.CooldownMaxSeconds = 1 })}, {"副本数非正", mutate(func(p *RoutingPolicy) { p.Degrade.MaxReplicas = 0 })}, } diff --git a/relaykit/dispatch/ports.go b/relaykit/dispatch/ports.go index 4947764..72682bc 100644 --- a/relaykit/dispatch/ports.go +++ b/relaykit/dispatch/ports.go @@ -24,7 +24,8 @@ type StatePort interface { ReportOutcome(o Outcome) // 熔断探测令牌:HALF_OPEN 时每窗口全局只放行一个真实请求。 - TryProbeToken(ctx context.Context, channelID int64) bool + // model 为空串 = 渠道级熔断 key;非空 = 渠道×模型级熔断 key。 + TryProbeToken(ctx context.Context, channelID int64, model string) bool // 容量租约(多副本安全,实例崩溃后自愈)。 AcquireLease(ctx context.Context, channelID int64, softLimit int, requestID string) bool