fix(channel):修复渠道调度问题，单模型异常不再影响整个渠道
This commit is contained in:
chengqian committed
1 parent
98ff538587
commit
949957755e
11 files changed
+432
-119
No files matched your search
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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, "渠道级保持复位后状态")
|
||||
}
|
||||
@@ -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:
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 })},
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user