feat: add redis circuit breaker gate with epoch fencing
This commit is contained in:
@@ -0,0 +1,326 @@
|
||||
"""跨进程熔断门: 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
|
||||
return {0, state, epoch, 0, '', probe_until - now}
|
||||
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], tonumber(ARGV[2])}
|
||||
"""
|
||||
|
||||
# KEYS: provider_hash ; ARGV: expected_epoch is_probe probe_owner(CHS :110-143)
|
||||
# 返回: applied state epoch failures retry_after_ms
|
||||
RECORD_SUCCESS = """
|
||||
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
|
||||
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')
|
||||
elseif state == 'half_open' then
|
||||
deadline = tonumber(redis.call('HGET', KEYS[1], 'probe_until') or '0')
|
||||
end
|
||||
return {0, state, epoch, failures, math.max(deadline - now, 0)}
|
||||
"""
|
||||
|
||||
# KEYS: provider_hash(CHS :148-199)
|
||||
# ARGV: expected_epoch is_probe probe_owner force_open threshold cooldown_ms
|
||||
# 返回: applied state epoch failures retry_after_ms
|
||||
RECORD_FAILURE = """
|
||||
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')
|
||||
elseif state == 'half_open' then
|
||||
deadline = tonumber(redis.call('HGET', KEYS[1], 'probe_until') or '0')
|
||||
end
|
||||
return {0, state, epoch, failures, math.max(deadline - now, 0)}
|
||||
end
|
||||
|
||||
local threshold = tonumber(ARGV[5])
|
||||
if tonumber(ARGV[4]) == 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
|
||||
epoch = epoch + 1
|
||||
local open_until = now + tonumber(ARGV[6])
|
||||
redis.call('HSET', KEYS[1],
|
||||
'state', 'open',
|
||||
'failures', failures,
|
||||
'epoch', epoch,
|
||||
'open_until', open_until,
|
||||
'probe_owner', '',
|
||||
'probe_until', 0)
|
||||
return {1, 'open', epoch, failures, tonumber(ARGV[6])}
|
||||
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')
|
||||
elseif state == 'half_open' then
|
||||
deadline = tonumber(redis.call('HGET', KEYS[1], 'probe_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)
|
||||
elseif state == 'half_open' then
|
||||
local deadline = tonumber(redis.call('HGET', key, 'probe_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._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}") from exc
|
||||
return self._decision(source_name, result)
|
||||
|
||||
async def record_success(self, entry: GateDecision) -> GateUpdate:
|
||||
try:
|
||||
result = await self._success_lua(
|
||||
keys=[self._key(entry.source_name)], args=self._entry_args(entry)
|
||||
)
|
||||
except RedisError as exc:
|
||||
raise GovernanceBackendError(f"熔断后端 record_success 失败: {exc}") 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])
|
||||
try:
|
||||
result = await self._failure_lua(keys=[self._key(entry.source_name)], args=args)
|
||||
except RedisError as exc:
|
||||
raise GovernanceBackendError(f"熔断后端 record_failure 失败: {exc}") 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}") 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}") from exc
|
||||
return int(result) / 1000.0
|
||||
|
||||
async def aclose(self) -> None:
|
||||
"""幂等释放自建客户端;注入的客户端归注入方管理。"""
|
||||
if self._owns_client:
|
||||
self._owns_client = False
|
||||
await self._redis.aclose()
|
||||
Reference in New Issue
Block a user