diff --git a/src/polygateway/config.py b/src/polygateway/config.py index 511d2fc..f2ff270 100644 --- a/src/polygateway/config.py +++ b/src/polygateway/config.py @@ -809,6 +809,13 @@ def check_hedge_assembly( after = ensure_call_deadline(hedge_after_s, origin) if after is None: return None + if not sources: + # 直传路(GatewayClient(sources=[], hedge_after_s=...))没有 settings 的 + # 非空守卫先行拦截;报错必须定位到 hedge,而不是裸 min() 空序列异常 + raise ValueError( + "hedge_after_s({SCOPE}__HEDGE__AFTER_S)的交叉守卫要求 sources 不能为空;" + "对冲已启用但没有可校验的源" + ) min_timeout = min(s.timeout_s for s in sources) if after >= min_timeout: raise ValueError( diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index 022f817..6540679 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -456,6 +456,22 @@ class RetryMW: if isinstance(outcome, LLMResponse): self._record_generation(request, sink_p) return outcome + # Phase 3.5 准入后复查: 原路可能在对冲准入的 await 期间已了结——此时 + # 一个对冲请求都不发(那是白付一次真实计费请求 + 一份 est 滞留 + 一条 + # hedge_cancelled 行)。按既有语义释放刚拿到的对冲准入(probe 须 + # release_probe,顺序同 _attempt 取消分支),直接裁定原路结果 + if primary.done(): + hedge_source, hedge_permit, hedge_entry = hedge_picked + try: + if hedge_entry.is_probe: + await self._record_quietly(self._breaker.release_probe(hedge_entry)) + finally: + self._pacer.leave(hedge_source.name) + await settle_and_release(hedge_permit, 0) + outcome = await primary + if isinstance(outcome, LLMResponse): + self._record_generation(request, sink_p) + return outcome # Phase 4 启动对冲路(v1 单路,H5): 对冲路恒传 first_token_event=None, # 不再触发梯次对冲 sink_h: list[int] = [] diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 71ac727..396442e 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -1166,6 +1166,31 @@ class TestHedgeConfig: with pytest.raises(ValueError, match=r"GatewayClient\(hedge_after_s"): _client(hedge_after_s=0) + def test_hedge_guard_empty_sources_raises_with_hedge_location(self): + """直传路空 sources + 设阈值: ValueError 且消息定位到 hedge(不是裸 min() 报错)。""" + from polygateway.config import check_hedge_assembly + + with pytest.raises(ValueError, match="sources") as ei: + check_hedge_assembly( + hedge_after_s=8, + hedge_max_extra=1, + sources=[], + call_deadline_s=None, + origin="GatewayClient(hedge_after_s=8)", + ) + assert "hedge_after_s" in str(ei.value) # 定位得到是哪个键 + # 未启用对冲(None)时空 sources 直接放行: 交叉守卫没有可校验的对象 + assert ( + check_hedge_assembly( + hedge_after_s=None, + hedge_max_extra=1, + sources=[], + call_deadline_s=None, + origin="test", + ) + is None + ) + def test_hedge_guard_below_min_timeout_raises(self): """阈值 ≥ 最小源 timeout_s = 对冲永不可能触发,装配期炸掉(ValueError)。""" env = self._two_source_env(**{"LLM__HEDGE__AFTER_S": "90"}) # min(timeout)=90 diff --git a/tests/unit/test_hedge.py b/tests/unit/test_hedge.py index 06fe8af..d419414 100644 --- a/tests/unit/test_hedge.py +++ b/tests/unit/test_hedge.py @@ -25,6 +25,7 @@ from polygateway.types import ( 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) @@ -80,6 +81,38 @@ class HedgeTransport: 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。""" @@ -121,8 +154,13 @@ def _harness( limiter=None, gate=None, stall_window_s=300.0, + now=None, ): - """真实 loop 钟装配(对冲计时纪律: 只用 loop 相对时长,不注入 FakeClock)。""" + """真实 loop 钟装配(对冲计时纪律: 只用 loop 相对时长,不注入 FakeClock)。 + + `now` 仅供"注入钟与对冲触发正交"用例注入 FakeClock——触发路径结构性不读 + 它,注入只是为了证明这一点。 + """ limiter = limiter or InMemoryLimiter( scope="llm", sources={s.name: s for s in sources}, @@ -144,6 +182,7 @@ def _harness( cooldown_memo=SourceCooldownMemo(), emitter=emitter, hedge_after_s=hedge_after_s, + **({"now": now} if now is not None else {}), ) return mw, limiter, gate @@ -209,6 +248,30 @@ class TestHedgeTrigger: 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: """异源排除与静默放弃(验收矩阵 ②③)。""" @@ -254,6 +317,81 @@ class TestHedgeRouting: 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",)]})