iomgaa
c7ccb5798c
Dual-channel opening: consecutive failures (CHS-compatible, no streak bump) plus windowed failure rate (two 30s buckets, min_calls guard). 429s bypass both channels as backpressure, and a probe hitting 429 releases instead of holding the lease. Open duration doubles per rate/probe reopen up to max_cooldown_s, decaying after stable CLOSED. record_success gains count_attempt so bad-result successes stay out of the window.
293 lines
13 KiB
Python
293 lines
13 KiB
Python
"""熔断后端契约测试(状态机 + 半开单探针租约 + epoch fencing)。
|
||
|
||
蓝本: VT adapters/breaker.py 状态机语义 + CHS provider_gate.py 契约;
|
||
M2 的 Redis 实现复用本套件。
|
||
"""
|
||
|
||
import pytest
|
||
|
||
from polygateway.ports import GateState
|
||
from polygateway.types import BreakerConfig
|
||
|
||
_CFG = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
|
||
|
||
|
||
async def _open_gate(gate, source="s1"):
|
||
"""连续失败到阈值,打开熔断,返回最后一次 Update。"""
|
||
update = None
|
||
for _ in range(_CFG.fail_threshold):
|
||
entry = await gate.try_enter(source, "w")
|
||
assert entry.allowed
|
||
update = await gate.record_failure(entry, "network_error", False)
|
||
assert update.state is GateState.OPEN
|
||
return update
|
||
|
||
|
||
class TestStateMachine:
|
||
async def test_closed_admits_and_success_resets(self, gate_factory):
|
||
gate = gate_factory(_CFG)
|
||
entry = await gate.try_enter("s1", "w")
|
||
assert entry.allowed and entry.state is GateState.CLOSED and not entry.is_probe
|
||
update = await gate.record_success(entry)
|
||
assert update.applied and update.failure_count == 0
|
||
|
||
async def test_threshold_opens_and_rejects(self, gate_factory):
|
||
gate = gate_factory(_CFG)
|
||
await _open_gate(gate)
|
||
entry = await gate.try_enter("s1", "w")
|
||
assert not entry.allowed and entry.state is GateState.OPEN
|
||
assert entry.retry_after_s > 0
|
||
|
||
async def test_success_before_threshold_resets_count(self, gate_factory):
|
||
gate = gate_factory(_CFG)
|
||
for _ in range(_CFG.fail_threshold - 1):
|
||
entry = await gate.try_enter("s1", "w")
|
||
await gate.record_failure(entry, "timeout", False)
|
||
entry = await gate.try_enter("s1", "w")
|
||
update = await gate.record_success(entry)
|
||
assert update.applied and update.failure_count == 0
|
||
assert (await gate.try_enter("s1", "w")).allowed
|
||
|
||
async def test_force_open_single_strike(self, gate_factory):
|
||
gate = gate_factory(_CFG)
|
||
entry = await gate.try_enter("s1", "w")
|
||
update = await gate.record_failure(entry, "source_dead", True)
|
||
assert update.state is GateState.OPEN
|
||
assert not (await gate.try_enter("s1", "w")).allowed
|
||
|
||
|
||
class TestHalfOpenProbe:
|
||
async def test_cooldown_grants_single_probe(self, gate_factory, clock):
|
||
gate = gate_factory(_CFG)
|
||
await _open_gate(gate)
|
||
clock.advance(_CFG.cooldown_s + 1)
|
||
probe = await gate.try_enter("s1", "w1")
|
||
assert probe.allowed and probe.is_probe and probe.probe_owner == "w1"
|
||
# 第二个进入者被拒(防惊群)
|
||
second = await gate.try_enter("s1", "w2")
|
||
assert not second.allowed and second.state is GateState.HALF_OPEN
|
||
|
||
async def test_probe_success_closes(self, gate_factory, clock):
|
||
gate = gate_factory(_CFG)
|
||
await _open_gate(gate)
|
||
clock.advance(_CFG.cooldown_s + 1)
|
||
probe = await gate.try_enter("s1", "w1")
|
||
update = await gate.record_success(probe)
|
||
assert update.applied and update.state is GateState.CLOSED
|
||
assert (await gate.try_enter("s1", "w2")).allowed
|
||
|
||
async def test_probe_failure_reopens(self, gate_factory, clock):
|
||
gate = gate_factory(_CFG)
|
||
await _open_gate(gate)
|
||
clock.advance(_CFG.cooldown_s + 1)
|
||
probe = await gate.try_enter("s1", "w1")
|
||
update = await gate.record_failure(probe, "network_error", False)
|
||
assert update.state is GateState.OPEN
|
||
assert not (await gate.try_enter("s1", "w2")).allowed
|
||
|
||
async def test_probe_lease_expiry_allows_takeover(self, gate_factory, clock):
|
||
"""探针持有者死亡 → 租约过期后新 caller 接管探针,防死锁。"""
|
||
gate = gate_factory(_CFG)
|
||
await _open_gate(gate)
|
||
clock.advance(_CFG.cooldown_s + 1)
|
||
stale = await gate.try_enter("s1", "dead-worker")
|
||
assert stale.is_probe
|
||
clock.advance(_CFG.probe_ttl_s + 1)
|
||
takeover = await gate.try_enter("s1", "w2")
|
||
assert takeover.allowed and takeover.is_probe and takeover.probe_owner == "w2"
|
||
|
||
async def test_release_probe_hands_back_and_idempotent(self, gate_factory, clock):
|
||
gate = gate_factory(_CFG)
|
||
await _open_gate(gate)
|
||
clock.advance(_CFG.cooldown_s + 1)
|
||
probe = await gate.try_enter("s1", "w1")
|
||
update = await gate.release_probe(probe)
|
||
assert update.applied
|
||
# 幂等: 第二次释放不再生效
|
||
assert not (await gate.release_probe(probe)).applied
|
||
# 源保持可接管状态: 下一 caller 拿到探针
|
||
nxt = await gate.try_enter("s1", "w2")
|
||
assert nxt.allowed and nxt.is_probe
|
||
|
||
|
||
class TestEpochFencing:
|
||
async def test_stale_epoch_write_rejected(self, gate_factory, clock):
|
||
"""entry 取得后世代已推进(他人触发开路)→ 迟到写回被 fencing 拒绝。"""
|
||
gate = gate_factory(_CFG)
|
||
stale_entry = await gate.try_enter("s1", "slow-worker")
|
||
await _open_gate(gate) # 他人连续失败 → 开路,epoch 推进
|
||
late = await gate.record_success(stale_entry)
|
||
assert not late.applied
|
||
# 门仍是 OPEN,未被迟到的成功污染
|
||
assert not (await gate.try_enter("s1", "w")).allowed
|
||
|
||
async def test_stale_probe_owner_rejected(self, gate_factory, clock):
|
||
gate = gate_factory(_CFG)
|
||
await _open_gate(gate)
|
||
clock.advance(_CFG.cooldown_s + 1)
|
||
stale_probe = await gate.try_enter("s1", "dead-worker")
|
||
clock.advance(_CFG.probe_ttl_s + 1)
|
||
takeover = await gate.try_enter("s1", "w2")
|
||
assert takeover.is_probe
|
||
# 死亡探针的迟到写回被拒(owner 已易主)
|
||
assert not (await gate.record_success(stale_probe)).applied
|
||
|
||
|
||
_RATE_CFG = BreakerConfig(
|
||
fail_threshold=100, # 连续通道抬高失声,单测率通道
|
||
cooldown_s=60.0,
|
||
probe_ttl_s=120.0,
|
||
min_calls=4,
|
||
fail_rate=0.9,
|
||
window_s=60.0,
|
||
max_cooldown_s=240.0,
|
||
)
|
||
|
||
|
||
async def _fail(gate, source="s1", reason="timeout", n=1):
|
||
update = None
|
||
for _ in range(n):
|
||
entry = await gate.try_enter(source, "w")
|
||
assert entry.allowed
|
||
update = await gate.record_failure(entry, reason, False)
|
||
return update
|
||
|
||
|
||
class TestRateChannel:
|
||
"""M2.5 失败率通道(设计 §3.1;全部确定性序列)。"""
|
||
|
||
async def test_rate_opens_at_min_calls(self, gate_factory):
|
||
# 病灶 1 回归: 高失败率源在 min_calls 样本处开路(连续通道静默)
|
||
gate = gate_factory(_RATE_CFG)
|
||
await _fail(gate, n=3) # attempts 3 < min_calls 4 → 仍 CLOSED
|
||
assert (await gate.try_enter("s1", "w")).allowed
|
||
update = await _fail(gate, n=1) # attempts 4, 失败率 1.0 ≥ 0.9
|
||
assert update.state is GateState.OPEN
|
||
assert not (await gate.try_enter("s1", "w")).allowed
|
||
|
||
async def test_below_min_calls_never_rate_opens(self, gate_factory):
|
||
gate = gate_factory(_RATE_CFG)
|
||
await _fail(gate, n=3)
|
||
assert (await gate.try_enter("s1", "w")).allowed
|
||
|
||
async def test_real_success_dilutes_window(self, gate_factory):
|
||
# 真实成功计入 attempts: 3 失败 + 1 成功 + 1 失败 = 4/5 = 0.8 < 0.9 不开
|
||
gate = gate_factory(_RATE_CFG)
|
||
await _fail(gate, n=3)
|
||
entry = await gate.try_enter("s1", "w")
|
||
await gate.record_success(entry)
|
||
update = await _fail(gate, n=1)
|
||
assert update.state is GateState.CLOSED
|
||
assert (await gate.try_enter("s1", "w")).allowed
|
||
|
||
async def test_result_invalid_not_counted(self, gate_factory):
|
||
# 坏结果 ≠ 坏服务: count_attempt=False 完全不动窗口 →
|
||
# 3 失败 + 1 不计成功 + 1 失败 = attempts 4, 失败率 1.0 → 开路
|
||
gate = gate_factory(_RATE_CFG)
|
||
await _fail(gate, n=3)
|
||
entry = await gate.try_enter("s1", "w")
|
||
await gate.record_success(entry, count_attempt=False)
|
||
update = await _fail(gate, n=1)
|
||
assert update.state is GateState.OPEN
|
||
|
||
async def test_rate_limited_bypasses_both_channels(self, gate_factory):
|
||
# C1 对抗: 429 是背压不是故障——连续通道(阈值 3)与率通道都不吃
|
||
cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0, min_calls=4)
|
||
gate = gate_factory(cfg)
|
||
await _fail(gate, reason="rate_limited", n=10)
|
||
assert (await gate.try_enter("s1", "w")).allowed
|
||
# timeout 脉冲照常走连续通道开路
|
||
update = await _fail(gate, reason="timeout", n=3)
|
||
assert update.state is GateState.OPEN
|
||
|
||
async def test_probe_rate_limited_releases_not_hangs(self, gate_factory, clock):
|
||
# 探针撞 429: 非故障证据也非成功——按无果归还语义放下家接管,
|
||
# 不得把探针租约挂到 TTL(否则源被锁死 probe_ttl_s)
|
||
gate = gate_factory(_RATE_CFG)
|
||
await _fail(gate, n=4)
|
||
clock.advance(61)
|
||
probe = await gate.try_enter("s1", "w1")
|
||
assert probe.is_probe
|
||
update = await gate.record_failure(probe, "rate_limited", False)
|
||
assert update.applied
|
||
nxt = await gate.try_enter("s1", "w2")
|
||
assert nxt.allowed and nxt.is_probe # 立即可再探,而非等 probe_ttl
|
||
|
||
async def test_rate_open_failure_count_capped(self, gate_factory):
|
||
gate = gate_factory(_RATE_CFG)
|
||
await _fail(gate, n=3)
|
||
update = await _fail(gate, n=1)
|
||
assert update.state is GateState.OPEN
|
||
assert update.failure_count == _RATE_CFG.fail_threshold # 顶格语义沿用
|
||
|
||
|
||
class TestReopenBackoff:
|
||
"""开路时长指数递增与衰减(设计 §3.1;memory 假时钟,redis 走真实等待变体)。"""
|
||
|
||
async def test_backoff_doubles_and_caps(self, gate_factory, clock):
|
||
gate = gate_factory(_RATE_CFG)
|
||
await _fail(gate, n=4) # 率通道首开: cooldown_eff = 60
|
||
assert 0 < await gate.retry_after_s(("s1",)) <= 60.0
|
||
clock.advance(61)
|
||
probe = await gate.try_enter("s1", "w1")
|
||
assert probe.is_probe
|
||
await gate.record_failure(probe, "timeout", False) # 探针失败重开: streak 2 → 120
|
||
wait = await gate.retry_after_s(("s1",))
|
||
assert 60.0 < wait <= 120.0
|
||
clock.advance(121)
|
||
probe = await gate.try_enter("s1", "w1")
|
||
await gate.record_failure(probe, "timeout", False) # streak 3 → 240(封顶)
|
||
wait = await gate.retry_after_s(("s1",))
|
||
assert 120.0 < wait <= 240.0
|
||
|
||
async def test_streak_survives_close_then_decays(self, gate_factory, clock):
|
||
gate = gate_factory(_RATE_CFG)
|
||
await _fail(gate, n=4)
|
||
clock.advance(61)
|
||
probe = await gate.try_enter("s1", "w1")
|
||
await gate.record_failure(probe, "timeout", False) # streak 2
|
||
clock.advance(121)
|
||
probe = await gate.try_enter("s1", "w1")
|
||
await gate.record_success(probe) # 转 CLOSED,streak 不清零
|
||
# 窗口含探针成功 1 次,9 失败 → 9/10 = 0.9 率开;streak 递增至封顶
|
||
await _fail(gate, n=9)
|
||
assert await gate.retry_after_s(("s1",)) > 60.0
|
||
# 衰减: CLOSED 稳定 2×cooldown_eff 后首次 record_success 归零
|
||
clock.advance(241)
|
||
probe = await gate.try_enter("s1", "w1")
|
||
await gate.record_success(probe)
|
||
clock.advance(2 * 240.0 + 1)
|
||
entry = await gate.try_enter("s1", "w")
|
||
await gate.record_success(entry) # 触发衰减
|
||
await _fail(gate, n=9) # 再开(9/10)回到基础档
|
||
assert 0 < await gate.retry_after_s(("s1",)) <= 60.0
|
||
|
||
async def test_consecutive_open_does_not_bump_streak(self, gate_factory, clock):
|
||
# C1 对抗: 连续通道误熔健康源的代价封顶为单次 cooldown_s
|
||
cfg = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
|
||
gate = gate_factory(cfg)
|
||
await _fail(gate, reason="timeout", n=3) # 连续通道开路
|
||
clock.advance(61)
|
||
probe = await gate.try_enter("s1", "w1")
|
||
await gate.record_success(probe) # 恢复
|
||
await _fail(gate, reason="timeout", n=3) # 再次连续开路
|
||
assert 0 < await gate.retry_after_s(("s1",)) <= 60.0 # 无翻倍
|
||
|
||
|
||
class TestRetryAfter:
|
||
async def test_retry_after_semantics(self, gate_factory, clock):
|
||
gate = gate_factory(_CFG)
|
||
assert await gate.retry_after_s(("s1",)) == 0.0 # 健康 → 0
|
||
await _open_gate(gate)
|
||
wait = await gate.retry_after_s(("s1",))
|
||
assert 0 < wait <= _CFG.cooldown_s
|
||
clock.advance(_CFG.cooldown_s + 1)
|
||
assert await gate.retry_after_s(("s1",)) == 0.0 # 冷却到期 → 0
|
||
with pytest.raises(ValueError):
|
||
await gate.retry_after_s(())
|
||
|
||
async def test_retry_after_takes_min_across_sources(self, gate_factory, clock):
|
||
gate = gate_factory(_CFG)
|
||
await _open_gate(gate, "s1") # s1 开路;s2 健康
|
||
assert await gate.retry_after_s(("s1", "s2")) == 0.0
|