iomgaa
45073486a7
A fail-closed limiter or breaker backend means the scope cannot emit a single request, which is exactly scope-level unavailability. But the error sat directly under PolyGatewayError, so a caller writing only `except GatewayUnavailableError` dropped it into the catch-all branch: Redis blips once and a backlog of tasks burns its business failure budget into the dead letter queue, over a fault a restart would clear. Three gate paths leak to callers rather than being absorbed by _record_quietly (try_acquire, try_enter, progress_age_s); each is now pinned by a test, since none of them had one before. The two unknown-source sites move to SourceNotConfiguredError instead of following along. They report a misconfigured source name, not an outage, and letting them into the retryable family would be the mirror of the bug being fixed here: the task would retry forever and never surface.
260 lines
8.9 KiB
Python
260 lines
8.9 KiB
Python
"""跨连接共享治理状态验证(M2 设计 §11.3;双连接池 = 多 worker 等价,人类认可)。
|
|
|
|
移植 CHS tests/integration/test_redis_limiter.py 的 2 个跨连接用例
|
|
(:127 全局并发、:166 进度可见)并把 :105 的全局 RPM(CHS 原版单连接)
|
|
升级为跨连接;再加熔断共享、双 client 联合 RPM 不超配、取消释放与
|
|
Redis 掉线 fail-closed 方向。真多进程验证在 soak harness --workers。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from polygateway.backends.redis.breaker import RedisGate
|
|
from polygateway.backends.redis.limiter import RedisLimiter
|
|
from polygateway.client import GatewayClient
|
|
from polygateway.errors import (
|
|
AllSourcesExhausted,
|
|
GatewayUnavailableError,
|
|
GovernanceBackendError,
|
|
)
|
|
from polygateway.sources import RoundRobinSelector
|
|
from polygateway.types import (
|
|
BackpressurePolicy,
|
|
BreakerConfig,
|
|
GlobalLimits,
|
|
RetryPolicy,
|
|
TransportResult,
|
|
)
|
|
from tests.contracts.conftest import (
|
|
await_window_headroom,
|
|
make_source,
|
|
redis_url_from_env,
|
|
)
|
|
|
|
_CFG = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
|
|
_BP = BackpressurePolicy(stall_window_s=300.0, poll_interval_s=0.02)
|
|
_RETRY = RetryPolicy(max_attempts=2, backoff_base_s=0.01, backoff_max_s=0.05)
|
|
|
|
|
|
@pytest.fixture
|
|
async def clients():
|
|
"""两个独立连接池(模拟两进程)。"""
|
|
url = redis_url_from_env()
|
|
if url is None:
|
|
pytest.skip("REDIS_URL 未配置")
|
|
import redis.asyncio as aioredis
|
|
|
|
a, b = aioredis.from_url(url), aioredis.from_url(url)
|
|
try:
|
|
yield a, b
|
|
finally:
|
|
await a.aclose()
|
|
await b.aclose()
|
|
|
|
|
|
def _limiter(client, scope, sources, global_limits, **kwargs) -> RedisLimiter:
|
|
return RedisLimiter(
|
|
scope=scope,
|
|
sources={s.name: s for s in sources},
|
|
global_limits=global_limits,
|
|
redis=client,
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
class ScriptedTransport:
|
|
"""返回固定成功结果(或挂起)的 transport;记录调用数(参照 test_retry FakeTransport)。"""
|
|
|
|
def __init__(self, hang: bool = False):
|
|
self.hang = hang
|
|
self.calls: list[str] = []
|
|
|
|
async def complete(self, *, messages, source, stream, overlay, call_id):
|
|
self.calls.append(source.name)
|
|
if self.hang:
|
|
await asyncio.Event().wait()
|
|
return TransportResult(
|
|
content="ok",
|
|
thinking="",
|
|
prompt_tokens=1,
|
|
completion_tokens=1,
|
|
usage_source="measured",
|
|
ttft_ms=None,
|
|
max_inter_token_ms=None,
|
|
raw={},
|
|
)
|
|
|
|
|
|
def _client(scope, sources, limiter, gate, transport, *, quota_full="wait") -> GatewayClient:
|
|
return GatewayClient(
|
|
scope=scope,
|
|
sources=sources,
|
|
selector=RoundRobinSelector(),
|
|
limiter=limiter,
|
|
breaker=gate,
|
|
transport=transport,
|
|
retry=_RETRY,
|
|
backpressure=_BP,
|
|
quota_full=quota_full,
|
|
)
|
|
|
|
|
|
# —— CHS 移植 3 例 ——
|
|
|
|
|
|
async def test_cross_connection_global_concurrency(clients):
|
|
"""CHS :127: 两连接池共享全局并发闸,第三个 acquire 跨连接被拒。"""
|
|
a_cli, b_cli = clients
|
|
scope = f"t{uuid4().hex[:8]}"
|
|
sources = [make_source(max_concurrency=100)]
|
|
limits = GlobalLimits(max_concurrency=2, rpm=0, tpm=0)
|
|
a = _limiter(a_cli, scope, sources, limits)
|
|
b = _limiter(b_cli, scope, sources, limits)
|
|
p1 = await a.try_acquire("s1", 1)
|
|
p2 = await b.try_acquire("s1", 1)
|
|
assert p1 is not None and p2 is not None
|
|
assert await a.try_acquire("s1", 1) is None # 全局 2 满,跨连接生效
|
|
await p1.release()
|
|
await p2.release()
|
|
|
|
|
|
async def test_cross_connection_global_rpm(clients):
|
|
"""CHS :105 升级为跨连接: 两池各消费全局 RPM 名额,第三个在任一连接都被拒。"""
|
|
a_cli, b_cli = clients
|
|
await await_window_headroom(a_cli)
|
|
scope = f"t{uuid4().hex[:8]}"
|
|
sources = [make_source("s1", rpm=100), make_source("s2", rpm=100)]
|
|
limits = GlobalLimits(max_concurrency=0, rpm=2, tpm=0)
|
|
a = _limiter(a_cli, scope, sources, limits)
|
|
b = _limiter(b_cli, scope, sources, limits)
|
|
p1 = await a.try_acquire("s1", 0)
|
|
p2 = await b.try_acquire("s2", 0)
|
|
assert p1 is not None and p2 is not None
|
|
await p1.release()
|
|
await p2.release()
|
|
assert await a.try_acquire("s1", 0) is None # RPM 不随 release 归还
|
|
assert await b.try_acquire("s2", 0) is None
|
|
|
|
|
|
async def test_progress_visible_across_connections(clients):
|
|
"""CHS :166: A mark 后 B(另一连接)立即读到 fresh age——背压活性是跨进程的。"""
|
|
a_cli, b_cli = clients
|
|
scope = f"t{uuid4().hex[:8]}"
|
|
sources = [make_source()]
|
|
a = _limiter(a_cli, scope, sources, GlobalLimits(0, 0, 0))
|
|
b = _limiter(b_cli, scope, sources, GlobalLimits(0, 0, 0))
|
|
assert await b.progress_age_s() == float("inf")
|
|
await a.mark_progress()
|
|
assert await b.progress_age_s() < 5.0
|
|
|
|
|
|
# —— 联合验证 ——
|
|
|
|
|
|
async def test_breaker_state_shared_across_connections(clients):
|
|
"""A 连接把源打开路,B 连接的 try_enter 立即被拒(熔断状态共享)。"""
|
|
a_cli, b_cli = clients
|
|
scope = f"t{uuid4().hex[:8]}"
|
|
gate_a = RedisGate(config=_CFG, redis=a_cli, scope=scope)
|
|
gate_b = RedisGate(config=_CFG, redis=b_cli, scope=scope)
|
|
for _ in range(_CFG.fail_threshold):
|
|
entry = await gate_a.try_enter("s1", "wa")
|
|
await gate_a.record_failure(entry, "network_error", False)
|
|
decision = await gate_b.try_enter("s1", "wb")
|
|
assert not decision.allowed and decision.retry_after_s > 0
|
|
|
|
|
|
async def test_two_clients_global_rpm_not_exceeded(clients):
|
|
"""双 GatewayClient(双连接池、同 scope)并发打满: 真实通过数 ≤ 全局 RPM。"""
|
|
a_cli, b_cli = clients
|
|
await await_window_headroom(a_cli)
|
|
scope = f"t{uuid4().hex[:8]}"
|
|
sources = [make_source(rpm=100, est_tokens=1)]
|
|
limits = GlobalLimits(max_concurrency=0, rpm=4, tpm=0)
|
|
transport = ScriptedTransport()
|
|
ca = _client(
|
|
scope,
|
|
sources,
|
|
_limiter(a_cli, scope, sources, limits),
|
|
RedisGate(config=_CFG, redis=a_cli, scope=scope),
|
|
transport,
|
|
quota_full="fail_fast",
|
|
)
|
|
cb = _client(
|
|
scope,
|
|
sources,
|
|
_limiter(b_cli, scope, sources, limits),
|
|
RedisGate(config=_CFG, redis=b_cli, scope=scope),
|
|
transport,
|
|
quota_full="fail_fast",
|
|
)
|
|
msgs = [{"role": "user", "content": "hi"}]
|
|
results = await asyncio.gather(
|
|
*(c.chat(msgs) for c in (ca, cb) for _ in range(4)), return_exceptions=True
|
|
)
|
|
ok = [r for r in results if not isinstance(r, BaseException)]
|
|
rejected = [r for r in results if isinstance(r, AllSourcesExhausted)]
|
|
assert len(transport.calls) == 4 # 8 并发中恰 4 个穿过全局 RPM 闸
|
|
assert len(ok) == 4 and len(rejected) == 4
|
|
assert all(r.reason == "quota_exhausted" for r in rejected)
|
|
|
|
|
|
async def test_cancel_in_flight_releases_lease(clients):
|
|
"""in-flight 取消 → permit 在 finally 释放,租约不泄漏(取消穿透铁律)。"""
|
|
a_cli, _ = clients
|
|
scope = f"t{uuid4().hex[:8]}"
|
|
sources = [make_source(max_concurrency=1)]
|
|
limiter = _limiter(a_cli, scope, sources, GlobalLimits(0, 0, 0))
|
|
client = _client(
|
|
scope,
|
|
sources,
|
|
limiter,
|
|
RedisGate(config=_CFG, redis=a_cli, scope=scope),
|
|
ScriptedTransport(hang=True),
|
|
)
|
|
task = asyncio.create_task(client.chat([{"role": "user", "content": "hi"}]))
|
|
while not (await limiter.source_stats("s1")).inflight:
|
|
await asyncio.sleep(0.02)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
assert (await limiter.source_stats("s1")).inflight == 0
|
|
|
|
|
|
# —— 掉线方向(fail-closed 集成证据)——
|
|
|
|
|
|
async def test_redis_down_admission_fails_closed():
|
|
"""Redis 不可达 → 准入侧报错绝不放行(库铁律),且以 scope 级形态到达调用方。
|
|
|
|
issue #7: 调用方只写 `except GatewayUnavailableError` 就该覆盖后端故障——
|
|
真实 Redis 掉线是这条链路唯一的端到端证据,故断言收紧到 scope 级语义。
|
|
"""
|
|
import redis.asyncio as aioredis
|
|
|
|
dead = aioredis.from_url(
|
|
"redis://127.0.0.1:1/0", socket_connect_timeout=0.3, socket_timeout=0.3
|
|
)
|
|
try:
|
|
limiter = RedisLimiter(
|
|
scope="t-dead",
|
|
sources={"s1": make_source()},
|
|
global_limits=GlobalLimits(0, 0, 0),
|
|
redis=dead,
|
|
lease_ttl_s=30.0,
|
|
)
|
|
gate = RedisGate(config=_CFG, redis=dead, scope="t-dead")
|
|
for call in (limiter.try_acquire("s1", 0), gate.try_enter("s1", "w")):
|
|
with pytest.raises(GatewayUnavailableError) as ei:
|
|
await call
|
|
assert isinstance(ei.value, GovernanceBackendError)
|
|
assert ei.value.reason == "governance_backend_down"
|
|
assert ei.value.scope == "t-dead"
|
|
assert ei.value.retry_after_s > 0
|
|
finally:
|
|
await dead.aclose()
|