diff --git a/src/polygateway/backends/redis/breaker.py b/src/polygateway/backends/redis/breaker.py new file mode 100644 index 0000000..9e6e5c5 --- /dev/null +++ b/src/polygateway/backends/redis/breaker.py @@ -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()