iomgaa
8edd3fb2cd
retry_after_s never had a written definition, so each backend improvised and they drifted apart. It now answers exactly one question: how long until a retry is *certainly* worth attempting. OPEN has such a moment (the cooldown deadline); HALF_OPEN does not, because the probe can come back at any time -- so it reports 0.0, which already means "retry now" elsewhere in the library. Six exits are brought in line. The half-open rejection is the one issue 14 reported: it returned the probe lease remainder, a deadlock-guard value derived from 2x the slowest timeout, so a 60s cooldown told callers to wait 600s. Worse, retry.py fed that number into the source cooldown memo, whose set_until only moves forward -- a source stayed skipped in-process for the whole lease even after its probe succeeded and the gate closed. That now writes an already-expired deadline, so the memo goes back to recording only real OPEN cooldowns. The other five were pre-existing memory/redis divergences hidden by a contract-test blind spot (the suite pinned that a second caller gets rejected, never what number it got): redis reported the probe TTL on grant and the lease remainder on fenced-out writes, where memory has always reported 0. Contract cases now pin all four half-open exits on both backends, with 1:1 real-wait variants for redis since the fake-clock ones skip there.
442 lines
17 KiB
Python
442 lines
17 KiB
Python
"""跨进程熔断门: CHS provider_gate 五操作 Lua 逐字移植(D3 双后端;M2 设计 §3)。
|
||
|
||
语义蓝本 `reference/CHSAnalyzer/app/coordination/{scripts,provider_gate}.py`:
|
||
每源一个 HASH(state/epoch/failures/open_until/probe_until/probe_owner),
|
||
每操作单条 Lua 原子。保真点: epoch **仅在 record_failure 开断时 +1**
|
||
(scripts.py:180);探针/force_open failures 顶格(:173-178);fencing 判据
|
||
两分支(:119-123/:157-161);release_probe 置 open_until=now 让下家立即
|
||
接管(:210-216);探针租约 = probe_until 绝对 ms + 惰性重发,无看门狗;
|
||
时钟回拨经 math.max clamp。已声明偏离: key 前缀 `provider_gate:` →
|
||
`pgw:gate:`;`LimiterError` → `GovernanceBackendError`。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from typing import TYPE_CHECKING
|
||
|
||
from redis.exceptions import RedisError
|
||
|
||
from polygateway.errors import GovernanceBackendError
|
||
from polygateway.ports import GateDecision, GateState, GateUpdate
|
||
|
||
if TYPE_CHECKING:
|
||
from redis.asyncio import Redis
|
||
|
||
from polygateway.types import BreakerConfig
|
||
|
||
# KEYS: provider_hash ; ARGV: owner probe_ttl_ms(CHS scripts.py:82-106)
|
||
# 返回: allowed state epoch is_probe probe_owner retry_after_ms
|
||
TRY_ENTER = """
|
||
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'
|
||
local epoch = tonumber(redis.call('HGET', KEYS[1], 'epoch') or '0')
|
||
local open_until = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
|
||
local probe_until = tonumber(redis.call('HGET', KEYS[1], 'probe_until') or '0')
|
||
|
||
if state == 'closed' then
|
||
return {1, state, epoch, 0, '', 0}
|
||
end
|
||
if state == 'open' and now < open_until then
|
||
return {0, state, epoch, 0, '', open_until - now}
|
||
end
|
||
if state == 'half_open' and now < probe_until then
|
||
-- 探针在途: 无确定的可再试时刻 → 0(issue #14,与 memory `_remaining` 同口径)
|
||
return {0, state, epoch, 0, '', 0}
|
||
end
|
||
|
||
local next_probe_until = now + tonumber(ARGV[2])
|
||
redis.call('HSET', KEYS[1],
|
||
'state', 'half_open',
|
||
'probe_owner', ARGV[1],
|
||
'probe_until', next_probe_until)
|
||
return {1, 'half_open', epoch, 1, ARGV[1], 0}
|
||
"""
|
||
|
||
# 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 = (
|
||
_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'
|
||
local epoch = tonumber(redis.call('HGET', KEYS[1], 'epoch') or '0')
|
||
local failures = tonumber(redis.call('HGET', KEYS[1], 'failures') or '0')
|
||
local owner = redis.call('HGET', KEYS[1], 'probe_owner') or ''
|
||
local is_probe = tonumber(ARGV[2])
|
||
local matches = false
|
||
if is_probe == 1 then
|
||
matches = state == 'half_open' and epoch == tonumber(ARGV[1]) and owner == ARGV[3]
|
||
else
|
||
matches = state == 'closed' and epoch == tonumber(ARGV[1])
|
||
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,
|
||
'epoch', epoch,
|
||
'open_until', 0,
|
||
'probe_owner', '',
|
||
'probe_until', 0)
|
||
return {1, 'closed', epoch, 0, 0}
|
||
end
|
||
|
||
local deadline = 0
|
||
if state == 'open' then
|
||
deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
|
||
end
|
||
return {0, state, epoch, failures, math.max(deadline - now, 0)}
|
||
"""
|
||
)
|
||
|
||
# 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 = (
|
||
_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'
|
||
local epoch = tonumber(redis.call('HGET', KEYS[1], 'epoch') or '0')
|
||
local failures = tonumber(redis.call('HGET', KEYS[1], 'failures') or '0')
|
||
local owner = redis.call('HGET', KEYS[1], 'probe_owner') or ''
|
||
local is_probe = tonumber(ARGV[2])
|
||
local matches = false
|
||
if is_probe == 1 then
|
||
matches = state == 'half_open' and epoch == tonumber(ARGV[1]) and owner == ARGV[3]
|
||
else
|
||
matches = state == 'closed' and epoch == tonumber(ARGV[1])
|
||
end
|
||
|
||
if not matches then
|
||
local deadline = 0
|
||
if state == 'open' then
|
||
deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
|
||
end
|
||
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])
|
||
local open_via_rate = false
|
||
local window_healthy = 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]) then
|
||
if fails_w / attempts >= tonumber(ARGV[9]) then
|
||
open_via_rate = true
|
||
else
|
||
-- 窗口证据充足且健康: 连败是噪声,抑制连续通道(迭代 6)
|
||
window_healthy = true
|
||
end
|
||
end
|
||
end
|
||
|
||
if force_open == 1 or is_probe == 1 then
|
||
failures = threshold
|
||
else
|
||
failures = failures + 1
|
||
end
|
||
local open_via_streak = failures >= threshold and not window_healthy
|
||
if force_open == 1 or is_probe == 1 or open_via_rate or open_via_streak 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 eff = cooldown_eff(KEYS[1], tonumber(ARGV[6]), tonumber(ARGV[11]))
|
||
redis.call('HSET', KEYS[1],
|
||
'state', 'open',
|
||
'failures', failures,
|
||
'epoch', epoch,
|
||
'open_until', now + eff,
|
||
'probe_owner', '',
|
||
'probe_until', 0)
|
||
redis.call('HDEL', KEYS[1], 'closed_since')
|
||
return {1, 'open', epoch, failures, eff}
|
||
end
|
||
redis.call('HSET', KEYS[1],
|
||
'state', 'closed',
|
||
'failures', failures,
|
||
'epoch', epoch,
|
||
'open_until', 0,
|
||
'probe_owner', '',
|
||
'probe_until', 0)
|
||
return {1, 'closed', epoch, failures, 0}
|
||
"""
|
||
)
|
||
|
||
# KEYS: provider_hash ; ARGV: expected_epoch probe_owner(CHS :203-225)
|
||
# 返回: applied state epoch failures retry_after_ms
|
||
RELEASE_PROBE = """
|
||
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'
|
||
local epoch = tonumber(redis.call('HGET', KEYS[1], 'epoch') or '0')
|
||
local failures = tonumber(redis.call('HGET', KEYS[1], 'failures') or '0')
|
||
local owner = redis.call('HGET', KEYS[1], 'probe_owner') or ''
|
||
if state == 'half_open' and epoch == tonumber(ARGV[1]) and owner == ARGV[2] then
|
||
redis.call('HSET', KEYS[1],
|
||
'state', 'open',
|
||
'open_until', now,
|
||
'probe_owner', '',
|
||
'probe_until', 0)
|
||
return {1, 'open', epoch, failures, 0}
|
||
end
|
||
local deadline = 0
|
||
if state == 'open' then
|
||
deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
|
||
end
|
||
return {0, state, epoch, failures, math.max(deadline - now, 0)}
|
||
"""
|
||
|
||
# KEYS: 若干 provider hash;返回最早可尝试等待毫秒(CHS :228-245)
|
||
RETRY_AFTER = """
|
||
local t = redis.call('TIME')
|
||
local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)
|
||
local minimum = nil
|
||
for _, key in ipairs(KEYS) do
|
||
local state = redis.call('HGET', key, 'state') or 'closed'
|
||
local remaining = 0
|
||
if state == 'open' then
|
||
local deadline = tonumber(redis.call('HGET', key, 'open_until') or '0')
|
||
remaining = math.max(deadline - now, 0)
|
||
end
|
||
if minimum == nil or remaining < minimum then minimum = remaining end
|
||
end
|
||
return minimum or 0
|
||
"""
|
||
|
||
|
||
class RedisGate:
|
||
"""以 Redis HASH + Lua 原子维护每源健康状态;同 scope 多进程共享。"""
|
||
|
||
def __init__(self, *, config: BreakerConfig, redis: Redis, scope: str) -> None:
|
||
if not scope.strip():
|
||
raise ValueError("scope 不能为空")
|
||
self._scope = scope.lower()
|
||
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)
|
||
self._success_lua = redis.register_script(RECORD_SUCCESS)
|
||
self._failure_lua = redis.register_script(RECORD_FAILURE)
|
||
self._release_lua = redis.register_script(RELEASE_PROBE)
|
||
self._retry_after_lua = redis.register_script(RETRY_AFTER)
|
||
|
||
@classmethod
|
||
def from_url(cls, url: str, **kwargs) -> RedisGate:
|
||
"""自建并持有 Redis 客户端(aclose 时代关);共享后端请直接注入 redis。"""
|
||
import redis.asyncio as aioredis
|
||
|
||
gate = cls(redis=aioredis.from_url(url), **kwargs)
|
||
gate._owns_client = True
|
||
return gate
|
||
|
||
def _key(self, source_name: str) -> str:
|
||
if not source_name.strip():
|
||
raise ValueError("source_name 不能为空")
|
||
return f"pgw:gate:{self._scope}:{source_name}"
|
||
|
||
@staticmethod
|
||
def _state(value: object) -> GateState:
|
||
if isinstance(value, bytes):
|
||
value = value.decode("utf-8")
|
||
return GateState(str(value))
|
||
|
||
@classmethod
|
||
def _decision(cls, source_name: str, result: list[object]) -> GateDecision:
|
||
"""解析 try_enter 六元返回;GateDecision 自带不变式校验兜底移植正确性。"""
|
||
is_probe = int(result[3]) == 1
|
||
owner = result[4]
|
||
if isinstance(owner, bytes):
|
||
owner = owner.decode("utf-8")
|
||
return GateDecision(
|
||
source_name=source_name,
|
||
allowed=int(result[0]) == 1,
|
||
state=cls._state(result[1]),
|
||
epoch=int(result[2]),
|
||
is_probe=is_probe,
|
||
probe_owner=str(owner) if is_probe else None,
|
||
retry_after_s=int(result[5]) / 1000.0,
|
||
)
|
||
|
||
@classmethod
|
||
def _update(cls, result: list[object]) -> GateUpdate:
|
||
return GateUpdate(
|
||
applied=int(result[0]) == 1,
|
||
state=cls._state(result[1]),
|
||
epoch=int(result[2]),
|
||
failure_count=int(result[3]),
|
||
retry_after_s=int(result[4]) / 1000.0,
|
||
)
|
||
|
||
@staticmethod
|
||
def _entry_args(entry: GateDecision) -> list[object]:
|
||
"""提取 fencing 令牌;拒绝用被拒决定伪造写回(CHS provider_gate.py:94-103)。"""
|
||
if not entry.allowed:
|
||
raise ValueError("被拒决定不得写回")
|
||
return [entry.epoch, 1 if entry.is_probe else 0, entry.probe_owner or ""]
|
||
|
||
async def try_enter(self, source_name: str, owner: str) -> GateDecision:
|
||
"""健康普通准入;冷却到期/探针租约过期时原子授予唯一探针。"""
|
||
if not owner.strip():
|
||
raise ValueError("owner 不能为空")
|
||
try:
|
||
result = await self._try_enter_lua(
|
||
keys=[self._key(source_name)], args=[owner, self._probe_ttl_ms]
|
||
)
|
||
except RedisError as exc:
|
||
raise GovernanceBackendError(
|
||
f"熔断后端 try_enter 失败: {exc}", scope=self._scope
|
||
) from exc
|
||
return self._decision(source_name, result)
|
||
|
||
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=args)
|
||
except RedisError as exc:
|
||
raise GovernanceBackendError(
|
||
f"熔断后端 record_success 失败: {exc}", scope=self._scope
|
||
) from exc
|
||
return self._update(result)
|
||
|
||
async def record_failure(
|
||
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,
|
||
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:
|
||
raise GovernanceBackendError(
|
||
f"熔断后端 record_failure 失败: {exc}", scope=self._scope
|
||
) from exc
|
||
return self._update(result)
|
||
|
||
async def release_probe(self, entry: GateDecision) -> GateUpdate:
|
||
"""探针无果归还: 源保持 OPEN 且 open_until=now,下家立即可接管;幂等。"""
|
||
if not (entry.allowed and entry.is_probe and entry.probe_owner is not None):
|
||
raise ValueError("release_probe 只接受在途探针决定")
|
||
try:
|
||
result = await self._release_lua(
|
||
keys=[self._key(entry.source_name)], args=[entry.epoch, entry.probe_owner]
|
||
)
|
||
except RedisError as exc:
|
||
raise GovernanceBackendError(
|
||
f"熔断后端 release_probe 失败: {exc}", scope=self._scope
|
||
) from exc
|
||
return self._update(result)
|
||
|
||
async def retry_after_s(self, sources: tuple[str, ...]) -> float:
|
||
"""集合中最早可尝试等待秒数;健康/到期返回 0;空集合 ValueError(契约)。"""
|
||
if not sources:
|
||
raise ValueError("sources 不能为空")
|
||
try:
|
||
result = await self._retry_after_lua(keys=[self._key(s) for s in sources])
|
||
except RedisError as exc:
|
||
raise GovernanceBackendError(
|
||
f"熔断后端 retry_after_s 失败: {exc}", scope=self._scope
|
||
) from exc
|
||
return int(result) / 1000.0
|
||
|
||
async def aclose(self) -> None:
|
||
"""幂等释放自建客户端;注入的客户端归注入方管理。"""
|
||
if self._owns_client:
|
||
self._owns_client = False
|
||
await self._redis.aclose()
|