"""跨进程熔断门: 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])} """ # 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') 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 + 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') 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 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') 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._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()