iomgaa/PolyGateway
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases 9 Wiki Activity

feat: add failure-rate breaker channel with exponential reopen backoff

Browse Source 
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.
This commit is contained in:
iomgaa
2026-07-21 09:08:14 -04:00
parent 72b25724d8
commit c7ccb5798c
4 changed files with 426 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/polygateway/backends/memory/breaker.py
+84 -8
View File
@@ -4,6 +4,12 @@
契约形态承 CHS `provider_gate.py`: 半开探针是**带 TTL 的租约**(持有者
死亡后可被接管,防"探针永远在路上"死锁),写回经 epoch fencing 拒绝
旧世代污染。epoch 在每次进入 OPEN 时递增。时钟构造注入,纯确定性可测。
M2.5 双通道(设计 2026-07-21-m25 §3.1): 在连续失败通道之外加失败率
通道(双 30s 桶窗口,样本 ≥ min_calls 且失败率 ≥ fail_rate 即开路);
429(rate_limited)是背压不是故障,两通道均不计;开路时长按 reopen_streak
指数递增封顶 max_cooldown_s——连续通道触发的开路不递增 streak(误熔
健康源的代价封顶为单次 cooldown_s)。
"""
from __future__ import annotations
@@ -22,7 +28,7 @@ if TYPE_CHECKING:
@dataclass
class _SourceGate:
"""单源的门控可变状态。"""
"""单源的门控可变状态(窗口/退避域见设计 §3.1 状态机域清单)。"""
state: GateState = GateState.CLOSED
fails: int = 0
@@ -31,6 +37,14 @@ class _SourceGate:
probe_owner: str | None = None
probe_expires: float = 0.0
reasons: dict[str, str] = field(default_factory=dict)
# 失败率窗口: 双半窗桶轮换(win_id = now // (window_s/2))
win_id: int = -1
a0: int = 0 # 当前桶 attempts / failures
f0: int = 0
a1: int = 0 # 上一桶
f1: int = 0
reopen_streak: int = 0
closed_since: float | None = None
class InMemoryGate:
@@ -46,6 +60,38 @@ class InMemoryGate:
raise ValueError("source_name 不能为空")
return self._gates.setdefault(source_name, _SourceGate())
# —— M2.5 失败率窗口 ——
def _rotate_window(self, g: _SourceGate) -> None:
"""按半窗粒度轮换双桶;跨两桶以上的空窗直接清零。"""
half = self._cfg.window_s / 2.0
wid = int(self._now() // half)
if wid == g.win_id:
return
if wid == g.win_id + 1:
g.a1, g.f1 = g.a0, g.f0
else:
g.a1, g.f1 = 0, 0
g.a0, g.f0 = 0, 0
g.win_id = wid
def _window_add(self, g: _SourceGate, *, failed: bool) -> None:
self._rotate_window(g)
g.a0 += 1
if failed:
g.f0 += 1
def _rate_channel_open(self, g: _SourceGate) -> bool:
self._rotate_window(g)
attempts = g.a0 + g.a1
if attempts < self._cfg.min_calls:
return False
return (g.f0 + g.f1) / attempts >= self._cfg.fail_rate
def _cooldown_eff(self, g: _SourceGate) -> float:
streak = max(1, g.reopen_streak)
return min(self._cfg.cooldown_s * (2 ** (streak - 1)), self._cfg.max_cooldown_s)
def _grant_probe(self, g: _SourceGate, source_name: str, owner: str) -> GateDecision:
g.state = GateState.HALF_OPEN
g.probe_owner = owner
@@ -123,18 +169,35 @@ class InMemoryGate:
else 0.0,
)
def _open(self, g: _SourceGate, reason: str) -> None:
def _open(self, g: _SourceGate, reason: str, *, bump_streak: bool) -> None:
if bump_streak:
g.reopen_streak += 1
g.state = GateState.OPEN
g.epoch += 1 # 世代推进: 旧 entry 的迟到写回自此被 fencing 拒绝
g.open_until = self._now() + self._cfg.cooldown_s
g.open_until = self._now() + self._cooldown_eff(g)
g.fails = max(g.fails, self._cfg.fail_threshold)
g.probe_owner = None
g.probe_expires = 0.0
g.closed_since = None
async def record_success(self, entry: GateDecision) -> GateUpdate:
async def record_success(
self, entry: GateDecision, *, count_attempt: bool = True
) -> GateUpdate:
g = self._gate(entry.source_name)
if not self._fenced(g, entry):
return self._snapshot(g, applied=False)
was_probe = entry.is_probe
if count_attempt:
self._window_add(g, failed=False)
# streak 衰减: CLOSED 稳定满 2×cooldown_eff 后的首次成功归零(设计 §3.1)
if (
g.reopen_streak > 0
and g.closed_since is not None
and self._now() - g.closed_since >= 2 * self._cooldown_eff(g)
):
g.reopen_streak = 0
if was_probe:
g.closed_since = self._now() # 仅探针转 CLOSED 时写,普通成功不刷新
g.state = GateState.CLOSED
g.fails = 0
g.probe_owner = None
@@ -147,12 +210,25 @@ class InMemoryGate:
g = self._gate(entry.source_name)
if not self._fenced(g, entry):
return self._snapshot(g, applied=False)
if entry.is_probe or force_open:
self._open(g, reason) # 探针失败重开 / SourceDead 一击即熔
if reason == "rate_limited" and not force_open:
# 429 = 背压不是故障(设计 §3.1): 两通道均不计,选源层软处理;
# 探针撞 429 按无果归还语义放下家,不挂租约
if entry.is_probe:
g.state = GateState.OPEN
g.open_until = self._now()
g.probe_owner = None
g.probe_expires = 0.0
return self._snapshot(g, applied=True)
if entry.is_probe or force_open:
# 探针失败重开递增 streak;SourceDead 一击即熔不递增
self._open(g, reason, bump_streak=entry.is_probe)
return self._snapshot(g, applied=True)
self._window_add(g, failed=True)
g.fails += 1
if g.fails >= self._cfg.fail_threshold:
self._open(g, reason)
if self._rate_channel_open(g):
self._open(g, reason, bump_streak=True)
elif g.fails >= self._cfg.fail_threshold:
self._open(g, reason, bump_streak=False) # 连续通道不递增(C1 封顶)
return self._snapshot(g, applied=True)
async def release_probe(self, entry: GateDecision) -> GateUpdate:
src/polygateway/backends/redis/breaker.py
+109 -14
View File
@@ -53,9 +53,33 @@ redis.call('HSET', KEYS[1],
return {1, 'half_open', epoch, 1, ARGV[1], tonumber(ARGV[2])}
"""
# KEYS: provider_hash ; ARGV: expected_epoch is_probe probe_owner(CHS :110-143)
# M2.5 窗口/退避公共片段(拼接进 success/failure 脚本;Lua 脚本间无法共享函数)
_WINDOW_HELPERS = """
local function rotate_window(key, now, half_ms)
local wid = math.floor(now / half_ms)
local cur = tonumber(redis.call('HGET', key, 'win_id') or '-1')
if wid ~= cur then
if wid == cur + 1 then
redis.call('HSET', key,
'a1', redis.call('HGET', key, 'a0') or '0',
'f1', redis.call('HGET', key, 'f0') or '0')
else
redis.call('HSET', key, 'a1', 0, 'f1', 0)
end
redis.call('HSET', key, 'a0', 0, 'f0', 0, 'win_id', wid)
end
end
local function cooldown_eff(key, cooldown_ms, max_cooldown_ms)
local streak = math.max(1, math.min(
tonumber(redis.call('HGET', key, 'reopen_streak') or '0'), 16))
return math.floor(math.min(cooldown_ms * 2 ^ (streak - 1), max_cooldown_ms))
end
"""
# KEYS: provider_hash(CHS :110-143 + M2.5 窗口/衰减)
# ARGV: expected_epoch is_probe probe_owner count_attempt half_window_ms cooldown_ms max_cooldown_ms
# 返回: applied state epoch failures retry_after_ms
RECORD_SUCCESS = """
RECORD_SUCCESS = _WINDOW_HELPERS + """
local t = redis.call('TIME')
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
local state = redis.call('HGET', KEYS[1], 'state') or 'closed'
@@ -71,6 +95,20 @@ else
end
if matches then
if tonumber(ARGV[4]) == 1 then
rotate_window(KEYS[1], now, tonumber(ARGV[5]))
redis.call('HINCRBY', KEYS[1], 'a0', 1)
end
-- streak 衰减: CLOSED 稳定满 2×cooldown_eff 后的首次成功归零(设计 §3.1)
local streak = tonumber(redis.call('HGET', KEYS[1], 'reopen_streak') or '0')
local since = tonumber(redis.call('HGET', KEYS[1], 'closed_since') or '0')
if streak > 0 and since > 0
and now - since >= 2 * cooldown_eff(KEYS[1], tonumber(ARGV[6]), tonumber(ARGV[7])) then
redis.call('HSET', KEYS[1], 'reopen_streak', 0)
end
if is_probe == 1 then
redis.call('HSET', KEYS[1], 'closed_since', now)
end
redis.call('HSET', KEYS[1],
'state', 'closed',
'failures', 0,
@@ -90,10 +128,11 @@ end
return {0, state, epoch, failures, math.max(deadline - now, 0)}
"""
# KEYS: provider_hash(CHS :148-199)
# KEYS: provider_hash(CHS :148-199 + M2.5 双通道/退避)
# ARGV: expected_epoch is_probe probe_owner force_open threshold cooldown_ms
# rate_limited min_calls fail_rate half_window_ms max_cooldown_ms
# 返回: applied state epoch failures retry_after_ms
RECORD_FAILURE = """
RECORD_FAILURE = _WINDOW_HELPERS + """
local t = redis.call('TIME')
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
local state = redis.call('HGET', KEYS[1], 'state') or 'closed'
@@ -118,23 +157,60 @@ if not matches then
return {0, state, epoch, failures, math.max(deadline - now, 0)}
end
local force_open = tonumber(ARGV[4])
if tonumber(ARGV[7]) == 1 and force_open == 0 then
-- 429 = 背压不是故障(M2.5 设计 §3.1): 两通道均不计;
-- 探针撞 429 按无果归还语义放下家,不挂租约
if is_probe == 1 then
redis.call('HSET', KEYS[1],
'state', 'open',
'open_until', now,
'probe_owner', '',
'probe_until', 0)
return {1, 'open', epoch, failures, 0}
end
return {1, state, epoch, failures, 0}
end
local threshold = tonumber(ARGV[5])
if tonumber(ARGV[4]) == 1 or is_probe == 1 then
local open_via_rate = false
if force_open == 0 and is_probe == 0 then
rotate_window(KEYS[1], now, tonumber(ARGV[10]))
redis.call('HINCRBY', KEYS[1], 'a0', 1)
redis.call('HINCRBY', KEYS[1], 'f0', 1)
local attempts = tonumber(redis.call('HGET', KEYS[1], 'a0') or '0')
+ tonumber(redis.call('HGET', KEYS[1], 'a1') or '0')
local fails_w = tonumber(redis.call('HGET', KEYS[1], 'f0') or '0')
+ tonumber(redis.call('HGET', KEYS[1], 'f1') or '0')
if attempts >= tonumber(ARGV[8]) and fails_w / attempts >= tonumber(ARGV[9]) then
open_via_rate = true
end
end
if force_open == 1 or is_probe == 1 then
failures = threshold
else
failures = failures + 1
end
if tonumber(ARGV[4]) == 1 or is_probe == 1 or failures >= threshold then
if force_open == 1 or is_probe == 1 or open_via_rate or failures >= threshold then
-- 递增 streak 的只有率通道开路与探针失败重开(C1: 连续通道/force_open 不递增)
if open_via_rate or is_probe == 1 then
redis.call('HINCRBY', KEYS[1], 'reopen_streak', 1)
end
if open_via_rate then
failures = math.max(failures, threshold)
end
epoch = epoch + 1
local open_until = now + tonumber(ARGV[6])
local eff = cooldown_eff(KEYS[1], tonumber(ARGV[6]), tonumber(ARGV[11]))
redis.call('HSET', KEYS[1],
'state', 'open',
'failures', failures,
'epoch', epoch,
'open_until', open_until,
'open_until', now + eff,
'probe_owner', '',
'probe_until', 0)
return {1, 'open', epoch, failures, tonumber(ARGV[6])}
redis.call('HDEL', KEYS[1], 'closed_since')
return {1, 'open', epoch, failures, eff}
end
redis.call('HSET', KEYS[1],
'state', 'closed',
@@ -203,6 +279,10 @@ class RedisGate:
self._threshold = config.fail_threshold
self._cooldown_ms = math.ceil(config.cooldown_s * 1000)
self._probe_ttl_ms = math.ceil(config.probe_ttl_s * 1000)
self._max_cooldown_ms = math.ceil(config.max_cooldown_s * 1000)
self._half_window_ms = max(1, math.ceil(config.window_s * 500)) # 半窗
self._min_calls = config.min_calls
self._fail_rate = config.fail_rate
self._redis = redis
self._owns_client = False
self._try_enter_lua = redis.register_script(TRY_ENTER)
@@ -277,11 +357,15 @@ class RedisGate:
raise GovernanceBackendError(f"熔断后端 try_enter 失败: {exc}") from exc
return self._decision(source_name, result)
async def record_success(self, entry: GateDecision) -> GateUpdate:
async def record_success(
self, entry: GateDecision, *, count_attempt: bool = True
) -> GateUpdate:
args = self._entry_args(entry)
args.extend(
[1 if count_attempt else 0, self._half_window_ms, self._cooldown_ms, self._max_cooldown_ms]
)
try:
result = await self._success_lua(
keys=[self._key(entry.source_name)], args=self._entry_args(entry)
)
result = await self._success_lua(keys=[self._key(entry.source_name)], args=args)
except RedisError as exc:
raise GovernanceBackendError(f"熔断后端 record_success 失败: {exc}") from exc
return self._update(result)
@@ -290,7 +374,18 @@ class RedisGate:
self, entry: GateDecision, reason: str, force_open: bool
) -> GateUpdate:
args = self._entry_args(entry)
args.extend([1 if force_open else 0, self._threshold, self._cooldown_ms])
args.extend(
[
1 if force_open else 0,
self._threshold,
self._cooldown_ms,
1 if reason == "rate_limited" else 0,
self._min_calls,
self._fail_rate,
self._half_window_ms,
self._max_cooldown_ms,
]
)
try:
result = await self._failure_lua(keys=[self._key(entry.source_name)], args=args)
except RedisError as exc: