iomgaa
9f7d407120
Address branch review findings 1-4 on feature/1.3.7-hedged-requests: - src/polygateway/middleware/retry.py: recheck primary.done() after hedge admission in _attempt_hedged; release the hedge permit via settle_and_release(permit, 0) (release_probe for probe entries) and adjudicate the primary directly instead of firing a billable HTTP request that would be cancelled immediately - src/polygateway/config.py: check_hedge_assembly raises a hedge-located ValueError for empty sources instead of a bare min() error - tests/unit/test_hedge.py: pin that an injected FakeClock jump of 10^6 seconds does not trigger hedging (design section 8); pin pick(exclude) counting no gate_rejections and leaving reasons untouched; pin silent hedge abandonment when the candidate circuit is open; deterministic regression for the admission-window race (BlockingLimiter harness) - tests/unit/test_config.py: assert the empty-sources guard message locates the hedge key Red-to-green evidence in tests/outputs/137/review-fixes/
595 lines
28 KiB
Python
595 lines
28 KiB
Python
"""对冲编排测试(issue #24 设计 §4.6 / 计划批次 G)。
|
||
|
||
设施纪律(计划 §5): 两源 scope、事件驱动假 transport(每源一对 entered/release
|
||
Event + 可脚本化"先置首 token 再挂起")、**真实 loop 钟**(不注入 FakeClock)、
|
||
`hedge_after_s=0.05`、断言容差 4–10×;取消窗口用 entered 双 Event 栅栏钉死,
|
||
禁 sleep 撞窗口;计时只断言下界与相对比较,不断言精确值。
|
||
"""
|
||
|
||
import asyncio
|
||
import time
|
||
|
||
import pytest
|
||
|
||
from polygateway import CallDeadlineExceeded
|
||
from polygateway.backends.memory.breaker import InMemoryGate
|
||
from polygateway.backends.memory.limiter import InMemoryLimiter
|
||
from polygateway.errors import AllSourcesExhausted, TransientError
|
||
from polygateway.middleware.retry import RetryMW
|
||
from polygateway.sources import SourceCooldownMemo
|
||
from polygateway.types import (
|
||
BackpressurePolicy,
|
||
BreakerConfig,
|
||
ChatRequest,
|
||
GlobalLimits,
|
||
RetryPolicy,
|
||
_CallContext,
|
||
)
|
||
from tests.contracts.conftest import FakeClock
|
||
from tests.unit.test_retry import RecordingSelector, StaticSelector, _ok, _src
|
||
|
||
_BREAKER = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
|
||
_NO_GLOBAL = GlobalLimits(max_concurrency=0, rpm=0, tpm=0)
|
||
_HEDGE_AFTER_S = 0.05 # 真实 loop 钟阈值;断言上界取 10×(0.5s)
|
||
|
||
|
||
class HedgeTransport:
|
||
"""事件驱动假 transport: 按源剧本精确控制首 token 置位与完成时刻。
|
||
|
||
剧本动作(每源一条队列,耗尽后重复最后一项——429 风暴用例需无限供应):
|
||
("hang",) — 置位该源 entered,挂起直到该源 release(或被取消)
|
||
("token_then_hang",) — 先置 first_token_event 再 hang(首 token 已至)
|
||
("succeed", content) — 立即成功
|
||
("succeed_after", delay, content)— 真实 loop 钟睡 delay 后成功
|
||
("fail_after", delay, factory) — 睡 delay 后抛 `factory()` 新造的异常
|
||
"""
|
||
|
||
def __init__(self, scripts: dict[str, list[tuple]]):
|
||
self._scripts = {name: list(actions) for name, actions in scripts.items()}
|
||
self.calls: list[str] = []
|
||
# 逐次记录收到的 first_token_event 身份(H5: 对冲路恒为 None,不再梯次)
|
||
self.ft_events: list[object] = []
|
||
self.entered = {name: asyncio.Event() for name in scripts}
|
||
self.release = {name: asyncio.Event() for name in scripts}
|
||
|
||
async def complete(
|
||
self, *, messages, source, stream, overlay, call_id, reasoning_effort, first_token_event
|
||
):
|
||
self.calls.append(source.name)
|
||
self.ft_events.append(first_token_event)
|
||
actions = self._scripts[source.name]
|
||
action = actions.pop(0) if len(actions) > 1 else actions[0]
|
||
kind = action[0]
|
||
if kind == "hang":
|
||
self.entered[source.name].set()
|
||
await self.release[source.name].wait()
|
||
return _ok(f"ok-{source.name}")
|
||
if kind == "token_then_hang":
|
||
if first_token_event is not None:
|
||
first_token_event.set()
|
||
self.entered[source.name].set()
|
||
await self.release[source.name].wait()
|
||
return _ok(f"ok-{source.name}")
|
||
if kind == "succeed":
|
||
return _ok(action[1])
|
||
if kind == "succeed_after":
|
||
await asyncio.sleep(action[1])
|
||
return _ok(action[2])
|
||
if kind == "fail_after":
|
||
await asyncio.sleep(action[1])
|
||
raise action[2]()
|
||
raise AssertionError(f"未知剧本动作: {action!r}")
|
||
|
||
|
||
class BlockingLimiter:
|
||
"""限流包装: 挂起指定源的 try_acquire 直到测试放行(确定性复现"对冲准入挂起")。
|
||
|
||
`acquiring` 置位 = 对冲准入已停在该源闸内;`allow` 置位后才继续。只拦对冲
|
||
会走到的源,原路准入不受影响;其余方法逐字委托内层 InMemoryLimiter。
|
||
"""
|
||
|
||
def __init__(self, inner: InMemoryLimiter, block_source: str):
|
||
self._inner = inner
|
||
self._block_source = block_source
|
||
self.acquiring = asyncio.Event()
|
||
self.allow = asyncio.Event()
|
||
|
||
async def try_acquire(self, source_key, est_tokens):
|
||
if source_key == self._block_source:
|
||
self.acquiring.set()
|
||
await self.allow.wait()
|
||
return await self._inner.try_acquire(source_key, est_tokens)
|
||
|
||
async def acquire(self, source_key, est_tokens):
|
||
return await self._inner.acquire(source_key, est_tokens)
|
||
|
||
async def source_stats(self, source_key):
|
||
return await self._inner.source_stats(source_key)
|
||
|
||
async def mark_progress(self):
|
||
return await self._inner.mark_progress()
|
||
|
||
async def progress_age_s(self):
|
||
return await self._inner.progress_age_s()
|
||
|
||
|
||
class RecordingEmitter:
|
||
"""逐次遥测假 emitter: 记录每行的源/错误标签/逻辑调用 ID/attempt call_id。"""
|
||
|
||
def __init__(self):
|
||
self.rows: list[dict] = []
|
||
|
||
async def emit_attempt(
|
||
self,
|
||
*,
|
||
request,
|
||
source,
|
||
call_id,
|
||
latency_ms,
|
||
response,
|
||
error,
|
||
reasoning_applies,
|
||
operation,
|
||
):
|
||
self.rows.append(
|
||
{
|
||
"source": source.name,
|
||
"call_id": call_id,
|
||
"logical_call_id": request.call_context.logical_call_id
|
||
if request.call_context is not None
|
||
else None,
|
||
"error": error,
|
||
}
|
||
)
|
||
|
||
|
||
def _harness(
|
||
sources,
|
||
transport,
|
||
*,
|
||
hedge_after_s=_HEDGE_AFTER_S,
|
||
max_attempts=3,
|
||
emitter=None,
|
||
selector=None,
|
||
limiter=None,
|
||
gate=None,
|
||
stall_window_s=300.0,
|
||
now=None,
|
||
):
|
||
"""真实 loop 钟装配(对冲计时纪律: 只用 loop 相对时长,不注入 FakeClock)。
|
||
|
||
`now` 仅供"注入钟与对冲触发正交"用例注入 FakeClock——触发路径结构性不读
|
||
它,注入只是为了证明这一点。
|
||
"""
|
||
limiter = limiter or InMemoryLimiter(
|
||
scope="llm",
|
||
sources={s.name: s for s in sources},
|
||
global_limits=_NO_GLOBAL,
|
||
lease_ttl_s=100.0,
|
||
)
|
||
gate = gate or InMemoryGate(config=_BREAKER)
|
||
mw = RetryMW(
|
||
scope="llm",
|
||
sources=sources,
|
||
# 固定配置序: 原路恒为 s1、对冲路恒为 s2,断言不依赖选源器内部状态
|
||
selector=selector if selector is not None else StaticSelector(),
|
||
limiter=limiter,
|
||
gate=gate,
|
||
transport=transport,
|
||
retry=RetryPolicy(max_attempts=max_attempts, backoff_base_s=0.01, backoff_max_s=0.05),
|
||
backpressure=BackpressurePolicy(stall_window_s=stall_window_s, poll_interval_s=0.01),
|
||
quota_full="wait",
|
||
cooldown_memo=SourceCooldownMemo(),
|
||
emitter=emitter,
|
||
hedge_after_s=hedge_after_s,
|
||
**({"now": now} if now is not None else {}),
|
||
)
|
||
return mw, limiter, gate
|
||
|
||
|
||
def _req(*, stream=False, ctx=None):
|
||
return ChatRequest(
|
||
messages=[{"role": "user", "content": "hi"}], stream=stream, call_context=ctx
|
||
)
|
||
|
||
|
||
def _ctx():
|
||
return _CallContext(now=time.monotonic)
|
||
|
||
|
||
class TestHedgeTrigger:
|
||
"""触发两形态(验收矩阵 ①): 非流式纯时间阈值;流式以首 token 未至为判据。"""
|
||
|
||
async def test_non_stream_triggers_hedge_and_fast_leg_wins(self):
|
||
"""s1 挂起、s2 即时成功: 对冲截断长尾,赢家为对冲路(⑩: 计时不含触发前等待)。"""
|
||
s1, s2 = _src("s1"), _src("s2")
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed_after", 0.02, "fast")]})
|
||
mw, _, _ = _harness([s1, s2], transport)
|
||
ctx = _ctx()
|
||
started = time.monotonic()
|
||
# wait_for 是防挂安全带(红相位无实现时 5s 判负),不是计时断言
|
||
resp = await asyncio.wait_for(mw(_req(stream=False, ctx=ctx)), timeout=5)
|
||
elapsed = time.monotonic() - started
|
||
assert resp.content == "fast" and resp.source_name == "s2"
|
||
assert elapsed < 10 * _HEDGE_AFTER_S # 挂起路被对冲截断,而非等到释放
|
||
stats = ctx.snapshot()
|
||
assert stats.hedges == 1 and stats.hedge_won is True
|
||
# 赢家裸生成时间: 下界 = s2 实际 transport 耗时(0.02s 留截断余量),
|
||
# 且严格小于总时长(不含 0.05s 触发窗等待)
|
||
assert stats.generation_ms >= 15
|
||
assert stats.generation_ms < stats.total_latency_ms
|
||
|
||
async def test_stream_triggers_only_when_first_token_absent(self):
|
||
"""两例(①): 首 token 未至 → 触发;先置首 token 再挂起 → 不触发。"""
|
||
# 例一: 流式但首 token 未至,阈值到 → 对冲触发
|
||
s1, s2 = _src("s1"), _src("s2")
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "fast")]})
|
||
mw, _, _ = _harness([s1, s2], transport)
|
||
ctx = _ctx()
|
||
resp = await asyncio.wait_for(mw(_req(stream=True, ctx=ctx)), timeout=5)
|
||
assert resp.source_name == "s2"
|
||
stats = ctx.snapshot()
|
||
assert stats.hedges == 1 and stats.hedge_won is True and stats.attempts == 2
|
||
|
||
# 例二: 首 token 已至(假 transport 先 set 再挂起)→ 不触发,误杀慢生成即此处
|
||
transport2 = HedgeTransport({"s1": [("token_then_hang",)], "s2": [("succeed", "other")]})
|
||
mw2, _, _ = _harness([_src("s1"), _src("s2")], transport2)
|
||
ctx2 = _ctx()
|
||
|
||
async def release_later():
|
||
await asyncio.sleep(4 * _HEDGE_AFTER_S) # 4× 余量确认窗口已过
|
||
transport2.release["s1"].set()
|
||
|
||
releaser = asyncio.create_task(release_later())
|
||
resp2 = await asyncio.wait_for(mw2(_req(stream=True, ctx=ctx2)), timeout=5)
|
||
await releaser
|
||
assert resp2.source_name == "s1"
|
||
assert transport2.calls == ["s1"] # 对冲从未发出
|
||
stats2 = ctx2.snapshot()
|
||
assert stats2.hedges == 0 and stats2.hedge_won is False and stats2.attempts == 1
|
||
|
||
async def test_injected_clock_jump_does_not_trigger_hedge(self):
|
||
"""对冲触发只认真实 loop 钟: 注入钟跳 10^6 秒不得触发对冲(设计 §8 验收矩阵)。
|
||
|
||
s1 挂起剧本 + 触发窗内注入钟拨快 10^6 秒: 若触发路径误读注入钟,对冲会
|
||
**立即**发出;断言对冲实际发出时刻不早于真实 loop 阈值(下界断言,不断
|
||
精确值),形态同 test_client.py:1974 deadline 的注入钟对应用例。
|
||
"""
|
||
clock = FakeClock()
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "hedged")]})
|
||
mw, _, _ = _harness([_src("s1"), _src("s2")], transport, now=clock)
|
||
ctx = _ctx()
|
||
task = asyncio.ensure_future(mw(_req(ctx=ctx)))
|
||
await transport.entered["s1"].wait() # 原路在途,触发窗计时中
|
||
clock.advance(1_000_000.0) # 跳变落在窗内: 误读注入钟即立刻触发
|
||
started = time.monotonic()
|
||
resp = await asyncio.wait_for(task, timeout=5)
|
||
elapsed = time.monotonic() - started
|
||
assert resp.source_name == "s2" # 对冲确由真实 loop 阈值触发并截断长尾
|
||
assert transport.calls == ["s1", "s2"]
|
||
# 下界留 20% 调度余量;误读注入钟的触发是毫秒级,与此差一个数量级以上
|
||
assert elapsed >= _HEDGE_AFTER_S * 0.8
|
||
stats = ctx.snapshot()
|
||
assert stats.hedges == 1 and stats.hedge_won is True
|
||
|
||
|
||
class TestHedgeRouting:
|
||
"""异源排除与静默放弃(验收矩阵 ②③)。"""
|
||
|
||
async def test_hedge_goes_to_other_source(self):
|
||
"""对冲请求落在另一源;两 attempt 行共享同一 logical_call_id(②⑤)。"""
|
||
emitter = RecordingEmitter()
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "hedged")]})
|
||
mw, _, _ = _harness([_src("s1"), _src("s2")], transport, emitter=emitter)
|
||
ctx = _ctx()
|
||
resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
|
||
assert resp.source_name == "s2"
|
||
assert transport.calls == ["s1", "s2"] # 第二请求落在异源
|
||
# H5 接缝: 原路携带首 token 观测,对冲路恒 None(v1 单路,不再梯次)
|
||
assert [e is not None for e in transport.ft_events] == [True, False]
|
||
assert len(emitter.rows) == 2
|
||
assert {r["logical_call_id"] for r in emitter.rows} == {ctx.logical_call_id}
|
||
assert emitter.rows[0]["call_id"] != emitter.rows[1]["call_id"] # 各 attempt 独立 ID
|
||
|
||
async def test_hedge_silent_when_no_candidate(self):
|
||
"""异源配额被占满 → 准入失败静默放弃: 不对冲、不抛错、原请求照等(③)。"""
|
||
s1 = _src("s1")
|
||
s2 = _src("s2", max_concurrency=1)
|
||
limiter = InMemoryLimiter(
|
||
scope="llm", sources={"s1": s1, "s2": s2}, global_limits=_NO_GLOBAL, lease_ttl_s=100.0
|
||
)
|
||
held = await limiter.try_acquire("s2", 0) # 外部预占满 s2 并发
|
||
assert held is not None
|
||
try:
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "x")]})
|
||
mw, _, _ = _harness([s1, s2], transport, limiter=limiter)
|
||
ctx = _ctx()
|
||
task = asyncio.ensure_future(mw(_req(ctx=ctx)))
|
||
await transport.entered["s1"].wait()
|
||
# 4× 余量: 给对冲窗与那次注定失败的准入留足发生时间
|
||
await asyncio.sleep(4 * _HEDGE_AFTER_S)
|
||
assert transport.calls == ["s1"] # 对冲静默未发出
|
||
transport.release["s1"].set()
|
||
resp = await asyncio.wait_for(task, timeout=5)
|
||
assert resp.source_name == "s1"
|
||
stats = ctx.snapshot()
|
||
assert stats.hedges == 0 and stats.attempts == 1 and stats.hedge_won is False
|
||
finally:
|
||
await held.release()
|
||
|
||
async def test_pick_exclude_all_is_not_a_rejection(self):
|
||
"""exclude 覆盖全源 → 返回 None 且 gate_rejections==0、reasons 不写(排除 ≠ 拒绝)。
|
||
|
||
admission 级直接钉(admission.py:192 `continue` 语义): 若未来重构把排除计入
|
||
gate_rejections,`on_no_runnable` 的"全源熔断类拒绝"判据会被污染,此钉当场报警。
|
||
"""
|
||
transport = HedgeTransport({"s1": [("succeed", "x")], "s2": [("succeed", "y")]})
|
||
mw, _, _ = _harness([_src("s1"), _src("s2")], transport)
|
||
reasons = {"prior": "rate_limited"} # 既有原因须原样保留
|
||
picked, gate_rejections = await mw._admission.pick(
|
||
reasons, {}, exclude=frozenset({"s1", "s2"})
|
||
)
|
||
assert picked is None
|
||
assert gate_rejections == 0
|
||
assert reasons == {"prior": "rate_limited"}
|
||
|
||
async def test_hedge_silent_when_candidate_circuit_open(self):
|
||
"""对冲候选被熔断开路 → 静默放弃: 不对冲、不抛错、原请求照等(②③的另一形态)。
|
||
|
||
现有限流闸用例只钉了"配额占满"一条静默路径;开路/pacer 拒绝走 pick 的另一
|
||
分支(gate_rejections 计数、reasons 写 circuit_open、settle_and_release 后
|
||
返回 None),同样不得发出对冲请求。
|
||
"""
|
||
gate = InMemoryGate(config=_BREAKER)
|
||
entry = await gate.try_enter("s2", "test-owner")
|
||
assert entry.allowed
|
||
await gate.record_failure(entry, "source_dead", True) # SourceDead 一击即熔
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "x")]})
|
||
mw, _, _ = _harness([_src("s1"), _src("s2")], transport, gate=gate)
|
||
ctx = _ctx()
|
||
task = asyncio.ensure_future(mw(_req(ctx=ctx)))
|
||
await transport.entered["s1"].wait()
|
||
# 4× 余量: 给对冲窗与那次注定被开路拒绝的准入留足发生时间
|
||
await asyncio.sleep(4 * _HEDGE_AFTER_S)
|
||
assert transport.calls == ["s1"] # 对冲静默未发出
|
||
transport.release["s1"].set()
|
||
resp = await asyncio.wait_for(task, timeout=5)
|
||
assert resp.source_name == "s1"
|
||
stats = ctx.snapshot()
|
||
assert stats.hedges == 0 and stats.attempts == 1 and stats.hedge_won is False
|
||
|
||
async def test_primary_done_during_hedge_admission_sends_no_hedge(self):
|
||
"""原路在对冲准入 await 期间已完成: 释放对冲准入直接裁定,一个对冲请求都不发。
|
||
|
||
剧本钉死窗口(禁 sleep 猜): s2 的 try_acquire 挂起(对冲准入停在闸内) →
|
||
放行 s1 → 轮询 s1 inflight 归零(_attempt finally 结算完,primary 必 done)
|
||
→ 此刻才放行对冲准入。pick 返回时原路已了结,编排必须不落 create_task。
|
||
"""
|
||
s1 = _src("s1", tpm=1000, est_tokens=400)
|
||
s2 = _src("s2", tpm=1000, est_tokens=400)
|
||
inner = InMemoryLimiter(
|
||
scope="llm",
|
||
sources={"s1": s1, "s2": s2},
|
||
global_limits=_NO_GLOBAL,
|
||
lease_ttl_s=100.0,
|
||
)
|
||
limiter = BlockingLimiter(inner, "s2")
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "hedge")]})
|
||
mw, _, _ = _harness([s1, s2], transport, limiter=limiter)
|
||
ctx = _ctx()
|
||
task = asyncio.ensure_future(mw(_req(ctx=ctx)))
|
||
await transport.entered["s1"].wait() # 原路在途
|
||
await limiter.acquiring.wait() # 对冲准入停在 s2 闸内(触发窗已过)
|
||
transport.release["s1"].set() # 原路放行完成
|
||
while (await inner.source_stats("s1")).inflight != 0:
|
||
await asyncio.sleep(0.001) # 结算完 = primary 已 done(同一任务步内返回)
|
||
limiter.allow.set() # pick 此刻才返回: primary.done() 已成立
|
||
resp = await asyncio.wait_for(task, timeout=5)
|
||
assert resp.content == "ok-s1" and resp.source_name == "s1"
|
||
assert transport.calls == ["s1"] # 对冲 HTTP 从未发出(未修前这里会看到 s2)
|
||
stats = ctx.snapshot()
|
||
assert stats.hedges == 0 and stats.attempts == 1 and stats.hedge_won is False
|
||
s2_stats = await inner.source_stats("s2")
|
||
assert s2_stats.inflight == 0 and s2_stats.tpm_used == 0 # 对冲准入按 0 结算释放
|
||
|
||
async def test_hedge_silent_when_single_source(self):
|
||
"""单源 scope: 运行期拿不到异源候选自然静默,行为与不配阈值逐字相同(②)。"""
|
||
transport = HedgeTransport({"s1": [("hang",)]})
|
||
mw, _, _ = _harness([_src("s1")], transport)
|
||
ctx = _ctx()
|
||
task = asyncio.ensure_future(mw(_req(ctx=ctx)))
|
||
await transport.entered["s1"].wait()
|
||
await asyncio.sleep(4 * _HEDGE_AFTER_S) # 窗口已过,仍无候选
|
||
assert transport.calls == ["s1"]
|
||
transport.release["s1"].set()
|
||
resp = await asyncio.wait_for(task, timeout=5)
|
||
assert resp.content == "ok-s1"
|
||
stats = ctx.snapshot()
|
||
assert stats.hedges == 0 and stats.attempts == 1 and stats.hedge_won is False
|
||
|
||
|
||
class TestHedgeSettlementAndSignals:
|
||
"""赢输记账与熔断/健康信号(验收矩阵 ④⑤;设计 §3 关键判断: 挂起 ≠ 源死亡)。"""
|
||
|
||
async def test_winner_settles_actual_loser_keeps_est(self):
|
||
"""赢家按真实 usage 结算;输家取消落 1.3.6 S3 格: est 预扣保留(④)。"""
|
||
s1 = _src("s1", tpm=1000, est_tokens=400)
|
||
s2 = _src("s2", tpm=1000, est_tokens=400)
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "win")]})
|
||
mw, limiter, _ = _harness([s1, s2], transport)
|
||
resp = await asyncio.wait_for(mw(_req()), timeout=5)
|
||
assert resp.source_name == "s2"
|
||
winner_stats = await limiter.source_stats("s2")
|
||
loser_stats = await limiter.source_stats("s1")
|
||
assert winner_stats.tpm_used == 15 # 预扣 400,实测 10+5 → settle 后只记 15
|
||
assert loser_stats.tpm_used == 400 # 输家 est 保留(可能被上游计费,保守下限)
|
||
assert winner_stats.inflight == 0 and loser_stats.inflight == 0
|
||
|
||
async def test_loser_row_labelled_hedge_cancelled(self):
|
||
"""输家 attempt 行 error=='hedge_cancelled',赢家行无 error,同行逻辑调用(④⑤)。
|
||
|
||
终态行是 client 级语义且成功调用本就不写终态行(emit_terminal_once 只在
|
||
异常/取消路径触发),MW 级可观测面即这两条 attempt 行。
|
||
"""
|
||
emitter = RecordingEmitter()
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "win")]})
|
||
mw, _, _ = _harness([_src("s1"), _src("s2")], transport, emitter=emitter)
|
||
ctx = _ctx()
|
||
await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
|
||
assert len(emitter.rows) == 2
|
||
loser = next(r for r in emitter.rows if r["source"] == "s1")
|
||
winner = next(r for r in emitter.rows if r["source"] == "s2")
|
||
assert loser["error"] == "hedge_cancelled"
|
||
assert winner["error"] is None
|
||
assert loser["logical_call_id"] == winner["logical_call_id"] == ctx.logical_call_id
|
||
|
||
async def test_loser_does_not_feed_breaker(self):
|
||
"""输家取消不喂熔断失败计数、不喂健康分;赢家照常 record_success(④)。"""
|
||
selector = RecordingSelector()
|
||
gate = InMemoryGate(config=_BREAKER)
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "win")]})
|
||
mw, _, _ = _harness([_src("s1"), _src("s2")], transport, selector=selector, gate=gate)
|
||
await asyncio.wait_for(mw(_req()), timeout=5)
|
||
gate_s1 = gate._gates["s1"]
|
||
assert gate_s1.a0 + gate_s1.a1 == 0 # 熔断失败率窗口无样本
|
||
assert selector.outcomes == [("s2", True)] # 健康喂数只有赢家的成功
|
||
assert (await gate.try_enter("s1", "w")).allowed # 挂起源未被标记
|
||
|
||
async def test_attempts_two_and_no_task_leak(self):
|
||
"""快照 attempts==2(含输家);返回后无本调用残留任务(⑤)。"""
|
||
before = asyncio.all_tasks()
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("succeed", "win")]})
|
||
mw, _, _ = _harness([_src("s1"), _src("s2")], transport)
|
||
ctx = _ctx()
|
||
await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
|
||
assert ctx.snapshot().attempts == 2
|
||
assert asyncio.all_tasks() == before
|
||
|
||
|
||
class TestHedgeCancellation:
|
||
"""取消穿透(⑦)与期限组合(⑧): 两任务同消、不 shield、不留后台任务。"""
|
||
|
||
async def test_external_cancel_cancels_both_legs(self):
|
||
"""两路均在途时外部取消: CancelledError 上抛,两 permit 释放。
|
||
|
||
输家标记只在赢家产生后才置位——外部取消下没有赢家,两行都是普通
|
||
"cancelled"(⑦;竞速误贴属设计 §4.5 已批准残留)。
|
||
"""
|
||
emitter = RecordingEmitter()
|
||
s1, s2 = _src("s1"), _src("s2")
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("hang",)]})
|
||
mw, limiter, _ = _harness([s1, s2], transport, emitter=emitter)
|
||
task = asyncio.ensure_future(mw(_req()))
|
||
# 双 Event 栅栏: 确认对冲已触发、两路均在途,再取消(禁 sleep 猜窗口)
|
||
await transport.entered["s1"].wait()
|
||
await transport.entered["s2"].wait()
|
||
task.cancel()
|
||
with pytest.raises(asyncio.CancelledError):
|
||
await task
|
||
assert {r["source"]: r["error"] for r in emitter.rows} == {
|
||
"s1": "cancelled",
|
||
"s2": "cancelled",
|
||
}
|
||
assert (await limiter.source_stats("s1")).inflight == 0
|
||
assert (await limiter.source_stats("s2")).inflight == 0
|
||
|
||
async def test_deadline_cuts_hedged_tree(self):
|
||
"""client 级 call_deadline_s=0.2 + 两路挂起 → CallDeadlineExceeded,permit 全释放(⑧)。"""
|
||
from tests.unit.test_client import _client
|
||
|
||
s1, s2 = _src("s1"), _src("s2")
|
||
limiter = InMemoryLimiter(
|
||
scope="llm", sources={"s1": s1, "s2": s2}, global_limits=_NO_GLOBAL
|
||
)
|
||
transport = HedgeTransport({"s1": [("hang",)], "s2": [("hang",)]})
|
||
client = _client(
|
||
sources=[s1, s2],
|
||
transport=transport,
|
||
limiter=limiter,
|
||
call_deadline_s=0.2,
|
||
hedge_after_s=_HEDGE_AFTER_S,
|
||
)
|
||
async with client:
|
||
with pytest.raises(CallDeadlineExceeded):
|
||
await client.chat([{"role": "user", "content": "hi"}], stream=False)
|
||
assert transport.calls == ["s1", "s2"] # 期限截止前对冲确已触发
|
||
assert (await limiter.source_stats("s1")).inflight == 0
|
||
assert (await limiter.source_stats("s2")).inflight == 0
|
||
|
||
|
||
class TestHedgeWinnerAdjudication:
|
||
"""赢家裁定(⑩): 取快者,含原路后发先至的对称面。"""
|
||
|
||
async def test_primary_late_success_wins_back(self):
|
||
"""s1 挂 0.3s(6× 阈值)后成功、s2 对冲路在途: 原路先完成 → 原路赢。"""
|
||
emitter = RecordingEmitter()
|
||
transport = HedgeTransport({"s1": [("succeed_after", 0.3, "late")], "s2": [("hang",)]})
|
||
mw, _, _ = _harness([_src("s1"), _src("s2")], transport, emitter=emitter)
|
||
ctx = _ctx()
|
||
resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5)
|
||
assert resp.content == "late" and resp.source_name == "s1"
|
||
stats = ctx.snapshot()
|
||
assert stats.hedges == 1 and stats.hedge_won is False
|
||
# 裸生成时间为原路那次 transport 时长(≈300ms,只断言下界与相对关系)
|
||
assert 250 <= stats.generation_ms <= stats.total_latency_ms
|
||
loser = next(r for r in emitter.rows if r["source"] == "s2")
|
||
assert loser["error"] == "hedge_cancelled" # 在途对冲路被裁为输家
|
||
|
||
|
||
class TestHedgeFailureCombination:
|
||
"""两败汇合(H6): 只计一次重试预算;429 分账逐字沿用 attempt 级机制。"""
|
||
|
||
async def test_both_fail_counts_budget_once(self):
|
||
"""两路 Transient: max_attempts=2 时恰进第二轮(两败只计一次),第二轮两败后才耗尽。"""
|
||
transport = HedgeTransport(
|
||
{
|
||
"s1": [("fail_after", 0.1, lambda: TransientError("p", source_name="s1"))],
|
||
"s2": [("fail_after", 0.12, lambda: TransientError("h", source_name="s2"))],
|
||
}
|
||
)
|
||
mw, _, _ = _harness([_src("s1"), _src("s2")], transport, max_attempts=2)
|
||
ctx = _ctx()
|
||
with pytest.raises(AllSourcesExhausted) as ei:
|
||
await mw(_req(ctx=ctx))
|
||
assert ei.value.reason == "retry_exhausted"
|
||
# 若两败计两次预算,第一轮即耗尽,这些调用根本不会发生
|
||
assert transport.calls == ["s1", "s2", "s1", "s2"]
|
||
# 两败轮次同样登记对冲路数(设计 §4.5: 实际并发发出即计)
|
||
assert ctx.snapshot().hedges == 2
|
||
|
||
async def test_both_429_refund_no_budget(self):
|
||
"""两路皆 429: 免预算且耗时退 stall 账——小 stall 窗下终局 stalled 而非耗尽。"""
|
||
|
||
def _429():
|
||
return TransientError("throttled", status_code=429, retry_after_s=0.01)
|
||
|
||
transport = HedgeTransport(
|
||
{"s1": [("fail_after", 0.1, _429)], "s2": [("fail_after", 0.12, _429)]}
|
||
)
|
||
mw, _, _ = _harness([_src("s1"), _src("s2")], transport, max_attempts=1, stall_window_s=0.3)
|
||
with pytest.raises(AllSourcesExhausted) as ei:
|
||
await mw(_req())
|
||
# max_attempts=1: 任一路计预算都会当场 retry_exhausted;
|
||
# 两 429 免预算 → 循环到 stall 窗口判死
|
||
assert ei.value.reason == "stalled"
|
||
|
||
async def test_mixed_429_and_failure_counts_budget(self):
|
||
"""一路 429 一路 Transient → 计一次预算、不退还 stall 账(_combine_failures)。"""
|
||
transport = HedgeTransport(
|
||
{
|
||
"s1": [
|
||
(
|
||
"fail_after",
|
||
0.1,
|
||
lambda: TransientError("rl", status_code=429, retry_after_s=0.01),
|
||
)
|
||
],
|
||
"s2": [("fail_after", 0.12, lambda: TransientError("boom", source_name="s2"))],
|
||
}
|
||
)
|
||
mw, _, _ = _harness([_src("s1"), _src("s2")], transport, max_attempts=1, stall_window_s=0.3)
|
||
with pytest.raises(AllSourcesExhausted) as ei:
|
||
await mw(_req())
|
||
assert ei.value.reason == "retry_exhausted"
|
||
assert transport.calls == ["s1", "s2"] # 恰一轮两路: 计一次预算即耗尽
|