iomgaa
a57a5cea72
Independent verification caught that the split shipped in the previous commit did not actually hold on the only path production uses. The gate wrappers re-raise GovernanceBackendError but nothing else, so SourceNotConfiguredError fell into the following `except Exception` and came back out as a governance_backend_down failure with retry_after_s=5.0. A misconfigured source name would still retry forever and never surface. The existing tests missed it because both of them call the private _cfg() directly, one layer below the wrapper the governance loops actually go through. The regression test goes through QuotaGate. telemetry.py has to widen its terminal catch in the same commit: once the wrapper stops relabeling the error, it is no longer a GovernanceBackendError, and it is raised before any attempt exists, so the path would have recorded no telemetry at all. Also corrects the leak path count from three to five. QuotaGate.stats and BreakerGate.retry_after_s are not wrapped by _record_quietly either.
75 lines
2.8 KiB
Python
75 lines
2.8 KiB
Python
"""熔断门封装: 后端故障必须报错而非放行(降级方向铁律,与 QuotaGate 同构)。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from polygateway.errors import GovernanceBackendError, SourceNotConfiguredError
|
|
|
|
if TYPE_CHECKING:
|
|
from polygateway.ports import GateDecision, GateUpdate, ProviderGate
|
|
from polygateway.types import SourceConfig
|
|
|
|
|
|
class BreakerGate:
|
|
"""RetryMW 面向熔断后端的唯一入口;包装一切后端异常。"""
|
|
|
|
def __init__(self, gate: ProviderGate, *, scope: str) -> None:
|
|
self._gate = gate
|
|
# 后端故障即 scope 级不可用,异常须携 scope 供调用方定位(issue #7 §3.3)
|
|
self._scope = scope
|
|
|
|
async def try_enter(self, source: SourceConfig, owner: str) -> GateDecision:
|
|
try:
|
|
return await self._gate.try_enter(source.name, owner)
|
|
except (GovernanceBackendError, SourceNotConfiguredError):
|
|
raise
|
|
except Exception as exc:
|
|
raise GovernanceBackendError(
|
|
f"熔断后端故障(try_enter): {exc}", scope=self._scope
|
|
) from exc
|
|
|
|
async def record_success(
|
|
self, entry: GateDecision, *, count_attempt: bool = True
|
|
) -> GateUpdate:
|
|
try:
|
|
return await self._gate.record_success(entry, count_attempt=count_attempt)
|
|
except (GovernanceBackendError, SourceNotConfiguredError):
|
|
raise
|
|
except Exception as exc:
|
|
raise GovernanceBackendError(
|
|
f"熔断后端故障(record_success): {exc}", scope=self._scope
|
|
) from exc
|
|
|
|
async def record_failure(
|
|
self, entry: GateDecision, reason: str, force_open: bool
|
|
) -> GateUpdate:
|
|
try:
|
|
return await self._gate.record_failure(entry, reason, force_open)
|
|
except (GovernanceBackendError, SourceNotConfiguredError):
|
|
raise
|
|
except Exception as exc:
|
|
raise GovernanceBackendError(
|
|
f"熔断后端故障(record_failure): {exc}", scope=self._scope
|
|
) from exc
|
|
|
|
async def release_probe(self, entry: GateDecision) -> GateUpdate:
|
|
try:
|
|
return await self._gate.release_probe(entry)
|
|
except (GovernanceBackendError, SourceNotConfiguredError):
|
|
raise
|
|
except Exception as exc:
|
|
raise GovernanceBackendError(
|
|
f"熔断后端故障(release_probe): {exc}", scope=self._scope
|
|
) from exc
|
|
|
|
async def retry_after_s(self, sources: tuple[str, ...]) -> float:
|
|
try:
|
|
return await self._gate.retry_after_s(sources)
|
|
except (GovernanceBackendError, SourceNotConfiguredError):
|
|
raise
|
|
except Exception as exc:
|
|
raise GovernanceBackendError(
|
|
f"熔断后端故障(retry_after_s): {exc}", scope=self._scope
|
|
) from exc
|