From 2ac05a20b6b60aa1dec05b84cc941db17eacfcae Mon Sep 17 00:00:00 2001 From: iomgaa Date: Mon, 14 Sep 2026 01:52:10 -0400 Subject: [PATCH] feat: add opt-in mid-stream stall hedging for chat HEDGE__STALL_AFTER_S triggers a cross-source hedge when the primary leg stalls mid-stream (independent of HEDGE__AFTER_S); two-phase hedge window with event recheck discipline; watchdog-killed primary converges on the surviving hedge leg. STREAM__STALL_THRESHOLD_MS (1000ms default) adjustable via env/settings/constructor. Assembly guards: ValueError on never-opening windows, embedding/OCR scope warning (B4). Batch D 14 tests red-to-green, 1688 unit+contracts. --- src/polygateway/client.py | 23 +- src/polygateway/config.py | 180 ++++++++--- src/polygateway/middleware/retry.py | 101 +++++- tests/unit/test_config.py | 190 ++++++++++- tests/unit/test_stall_hedge.py | 479 ++++++++++++++++++++++++++++ 5 files changed, 899 insertions(+), 74 deletions(-) create mode 100644 tests/unit/test_stall_hedge.py diff --git a/src/polygateway/client.py b/src/polygateway/client.py index 9b2ea10..dd8ee18 100644 --- a/src/polygateway/client.py +++ b/src/polygateway/client.py @@ -239,6 +239,8 @@ class GatewayClient: call_deadline_s: float | None = None, hedge_after_s: float | None = None, hedge_max_extra: int = 1, + hedge_stall_after_s: float | None = None, + stall_threshold_ms: int | None = None, now: Any = time.monotonic, sleep: Any = asyncio.sleep, rng: Any = random.random, @@ -247,15 +249,25 @@ class GatewayClient: self._call_deadline_s = ensure_call_deadline( call_deadline_s, "GatewayClient(call_deadline_s=...)" ) - # 对冲守卫与 GatewaySettings 共用同一份(issue #24 H4): 直接构造这条路 - # 不经过 settings,值域/交叉守卫若只挂在 settings 上就会被它绕过 - self._hedge_after_s = check_hedge_assembly( + # 对冲守卫与 GatewaySettings 共用同一份(issue #24 H4 / 1.3.8 #28): 直接构造这条路 + # 不经过 settings,值域/交叉守卫若只挂在 settings 上就会被它绕过。 + # 返回契约(1.3.8 审查 B2): tuple(归一化 after_s, 归一化 stall_after_s) + self._hedge_after_s, self._hedge_stall_after_s = check_hedge_assembly( hedge_after_s=hedge_after_s, hedge_max_extra=hedge_max_extra, sources=sources, call_deadline_s=self._call_deadline_s, origin="GatewayClient(hedge_after_s=...)", + stall_after_s=hedge_stall_after_s, + stall_origin="GatewayClient(hedge_stall_after_s=...)", ) + # 沉默计量阈值分发(1.3.8 设计 §6): None = 保留各 SourceConfig 自带值 + # (不 stomp 直构造调用方的 per-source 设置);非 None 经 replace 写进每个源 + # (replace 触发 SourceConfig.__post_init__ 重跑,值域校验复用同一处) + if stall_threshold_ms is not None: + sources = [ + dataclasses.replace(s, stall_threshold_ms=stall_threshold_ms) for s in sources + ] emitter = ( TelemetryEmitter(telemetry, scope=scope, pricing=pricing, text_cap=text_cap) if telemetry is not None @@ -280,6 +292,8 @@ class GatewayClient: emitter=emitter, # max_extra 不下传(issue #24 H5): v1 编排固定单路对冲,无消费者 hedge_after_s=self._hedge_after_s, + # 中段沉默对冲阈值(1.3.8 #28): 独立于 hedge_after_s 生效,守卫归一化值 + stall_after_s=self._hedge_stall_after_s, now=now, sleep=sleep, rng=rng, @@ -524,6 +538,9 @@ class GatewayClient: call_deadline_s=settings.call_deadline_s, hedge_after_s=settings.hedge_after_s, hedge_max_extra=settings.hedge_max_extra, + hedge_stall_after_s=settings.hedge_stall_after_s, + # settings 值为权威,恒分发(1.3.8 §3.5): 含缺省 1000 在内 + stall_threshold_ms=settings.stall_threshold_ms, ) _mark_owned_components(client, limiter=limiter, breaker=breaker, telemetry=telemetry) client._owns_cache = cache is None # 缓存后端可以是 None(backend=none),helper 会跳过 diff --git a/src/polygateway/config.py b/src/polygateway/config.py index f2ff270..fef2a9d 100644 --- a/src/polygateway/config.py +++ b/src/polygateway/config.py @@ -54,7 +54,7 @@ _SOURCE_FIELDS: dict[str, tuple[str, str]] = { "TRUST_ENV": ("trust_env", "bool"), "EXTRA_BODY": ("extra_body", "json"), } -_RESERVED_SEGMENTS = frozenset({"GLOBAL", "RETRY", "BREAKER", "BACKPRESSURE", "HEDGE"}) +_RESERVED_SEGMENTS = frozenset({"GLOBAL", "RETRY", "BREAKER", "BACKPRESSURE", "HEDGE", "STREAM"}) _SELECTORS = frozenset({"round_robin", "least_inflight", "health_aware"}) _QUOTA_FULL = frozenset({"wait", "fail_fast"}) # 熔断全拒时的处置(issue #14);值域与 _QUOTA_FULL 相同但语义不同——配额满是 @@ -194,6 +194,16 @@ class GatewaySettings: # 由 `check_hedge_assembly` 单一定义点把关,直接构造/replace/env 三路一致 hedge_after_s: float | None = None hedge_max_extra: int = 1 + # 中段沉默对冲(1.3.8 issue #28): 相邻增量 chunk 间隔超阈值 → 并发异源副路。 + # 缺省 None = 关闭;独立于 hedge_after_s 生效(设计 §9-④);仅对流式 chat 请求 + # 生效,非流式请求运行期静默退化(入口门)。值域与交叉守卫同归 + # `check_hedge_assembly` 单一定义点 + hedge_stall_after_s: float | None = None + # 沉默计量阈值(1.3.8 issue #27): 相邻增量 chunk 间隔**严格大于**它才计入 + # stall_ms 并从 generation_ms 扣减。默认 1000ms(设计 §9-③ 拍板,三层可调); + # 只影响计量口径,不改任何行为。值域(正整数)归 `SourceConfig.__post_init__` + # 一处——分发经 `dataclasses.replace` 时重跑,本字段不重复校验 + stall_threshold_ms: int = 1000 def __post_init__(self) -> None: self._normalize() @@ -378,17 +388,17 @@ class GatewaySettings: env 路的值域错误已在 `_load_hedge` 里带真实键名报过;此处对合法值是幂等 空操作,交叉守卫(阈值 vs timeout/deadline/ttft、单源)只在这里有一处。 """ - object.__setattr__( - self, - "hedge_after_s", - check_hedge_assembly( - hedge_after_s=self.hedge_after_s, - hedge_max_extra=self.hedge_max_extra, - sources=self.sources, - call_deadline_s=self.call_deadline_s, - origin="GatewaySettings.hedge_after_s", - ), + after_s, stall_after_s = check_hedge_assembly( + hedge_after_s=self.hedge_after_s, + hedge_max_extra=self.hedge_max_extra, + sources=self.sources, + call_deadline_s=self.call_deadline_s, + origin="GatewaySettings.hedge_after_s", + stall_after_s=self.hedge_stall_after_s, + stall_origin="GatewaySettings.hedge_stall_after_s", ) + object.__setattr__(self, "hedge_after_s", after_s) + object.__setattr__(self, "hedge_stall_after_s", stall_after_s) @classmethod def from_env( @@ -420,6 +430,7 @@ class GatewaySettings: circuit_open=_load_choice(env, f"{scope_u}__CIRCUIT_OPEN", _CIRCUIT_OPEN, "fail_fast"), call_deadline_s=_load_call_deadline(scope_u, env), **_load_hedge(scope_u, env), + **_load_stream(scope_u, env), **_load_pgw(env), ) @@ -742,8 +753,29 @@ def _load_call_deadline(scope: str, env: Mapping[str, str]) -> float | None: return ensure_call_deadline(_cast(found[1], "float", found[0]), found[0]) +def _load_stream(scope: str, env: Mapping[str, str]) -> dict[str, object]: + """读 `{SCOPE}__STREAM__STALL_THRESHOLD_MS`(1.3.8 issue #27 设计 §6)。 + + 3 段键(`split("__")` 长度 3 ≠ 4),`_load_sources` 的段数判据天然跳过; + `STREAM` 已进 `_RESERVED_SEGMENTS`,4 段的 `{SCOPE}__STREAM__{N}__*` 也不会 + 被当成 provider 段造出源(1.3.7 `HEDGE` 同款)。未设 = 1000(设计 §9-③ 拍板); + 值域(正整数)归 `SourceConfig.__post_init__` 一处——字段分发经 + `dataclasses.replace` 时重跑,本函数只负责解析。 + + Args: + scope: 已大写的 scope 名。 + env: 已合并的环境映射。 + + Returns: + `{"stall_threshold_ms": int}`,直传构造器。 + """ + found = _first(env, f"{scope}__STREAM__STALL_THRESHOLD_MS") + threshold = int(_cast(found[1], "int", found[0])) if found else 1000 + return {"stall_threshold_ms": threshold} + + def _load_hedge(scope: str, env: Mapping[str, str]) -> dict[str, object]: - """读 `{SCOPE}__HEDGE__AFTER_S`/`{SCOPE}__HEDGE__MAX_EXTRA`(issue #24 H4)。 + """读 `{SCOPE}__HEDGE__AFTER_S`/`MAX_EXTRA`/`STALL_AFTER_S`(issue #24 H4 + 1.3.8 #28)。 两键均为 3 段键(`split("__")` 长度 3 ≠ 4),`_load_sources` 的段数判据天然 跳过它们;`HEDGE` 已进 `_RESERVED_SEGMENTS`,4 段的 `{SCOPE}__HEDGE__{N}__*` @@ -766,7 +798,13 @@ def _load_hedge(scope: str, env: Mapping[str, str]) -> dict[str, object]: ) found_extra = _first(env, f"{scope}__HEDGE__MAX_EXTRA") extra = int(_cast(found_extra[1], "int", found_extra[0])) if found_extra else 1 - return {"hedge_after_s": after, "hedge_max_extra": extra} + found_stall = _first(env, f"{scope}__HEDGE__STALL_AFTER_S") + stall = ( + ensure_call_deadline(_cast(found_stall[1], "float", found_stall[0]), found_stall[0]) + if found_stall + else None + ) + return {"hedge_after_s": after, "hedge_max_extra": extra, "hedge_stall_after_s": stall} def check_hedge_assembly( @@ -776,26 +814,32 @@ def check_hedge_assembly( sources: Sequence[SourceConfig], call_deadline_s: float | None, origin: str, -) -> float | None: - """对冲装配守卫的唯一事实源(issue #24 设计 §5 全表,H4 批准)。 + stall_after_s: float | None = None, + stall_origin: str = "hedge_stall_after_s", +) -> tuple[float | None, float | None]: + """对冲装配守卫的唯一事实源(issue #24 设计 §5 全表,H4 批准 + 1.3.8 issue #28 §6)。 `GatewaySettings.__post_init__` 与 `GatewayClient.__init__` 调同一份,两条装配 - 路的值域/交叉守卫不漂移。返回归一化后的 `hedge_after_s`(None 或有限正数, - 复用 `ensure_call_deadline` 的值域校验);`hedge_after_s is None`(未启用)时 - 值域归一化后直接返回,交叉守卫不查——它们没有可校验的对象。 + 路的值域/交叉守卫不漂移。返回契约(1.3.8 审查 B2): + `tuple(归一化 hedge_after_s, 归一化 stall_after_s)`(各为 None 或有限正数, + 复用 `ensure_call_deadline` 的值域校验);**两阈值皆 None 才早退**——只对 + `hedge_after_s is None` 早退会让仅 stall 配置绕过交叉守卫(审查 B2 钉死)。 Args: - hedge_after_s: 对冲触发阈值(秒);None = 关闭。 + hedge_after_s: 首 token/整体耗时对冲触发阈值(秒);None = 关闭。 hedge_max_extra: 每次逻辑调用最多并发对冲路数;v1 仅单路生效(H5)。 - sources: 本 scope 的源集合(交叉守卫要读 timeout_s/ttft_timeout_s)。 + sources: 本 scope 的源集合(交叉守卫要读 timeout_s/ttft/inter_token)。 call_deadline_s: 调用期限(秒);与对冲的组合守卫见设计 §6。 origin: after_s 值域报错的定位串(env 键名 / `GatewaySettings.hedge_after_s` / `GatewayClient(hedge_after_s=...)`);跨字段守卫与 warning 沿用本模块先例, 消息自带字段名与 env 键型,不挂 origin。 + stall_after_s: 中段沉默对冲触发阈值(秒,1.3.8 issue #28);None = 关闭。 + stall_origin: stall_after_s 值域报错的定位串(同 origin 的三路纪律)。 Raises: - ValueError: max_extra 非 int/bool 或出 [1,3];阈值 ≥ 最小源 timeout_s(永不 - 可能触发);阈值 ≥ call_deadline_s(期限先于对冲触发,对冲形同虚设)。 + ValueError: max_extra 非 int/bool 或出 [1,3];after_s ≥ 最小源 timeout_s(永不 + 可能触发);stall_after_s ≥ 最小源有效 inter-token 超时(回落同款,窗永不开); + 阈值 ≥ call_deadline_s(期限先于对冲触发,对冲形同虚设)。 """ if isinstance(hedge_max_extra, bool) or not isinstance(hedge_max_extra, int): raise ValueError( @@ -807,45 +851,80 @@ def check_hedge_assembly( "每次逻辑调用最多并发对冲路数,v1 仅单路生效" ) after = ensure_call_deadline(hedge_after_s, origin) - if after is None: - return None + stall = ensure_call_deadline(stall_after_s, stall_origin) + if after is None and stall is None: + return None, None if not sources: - # 直传路(GatewayClient(sources=[], hedge_after_s=...))没有 settings 的 - # 非空守卫先行拦截;报错必须定位到 hedge,而不是裸 min() 空序列异常 + # 直传路(GatewayClient(sources=[], ...))没有 settings 的 + # 非空守卫先行拦截;报错必须定位到 hedge 键族,而不是裸 min() 空序列异常。 + # 消息保留 "hedge_after_s" 字样: 既有用例据此定位键族(1.3.8 扩含 stall) raise ValueError( - "hedge_after_s({SCOPE}__HEDGE__AFTER_S)的交叉守卫要求 sources 不能为空;" - "对冲已启用但没有可校验的源" + "hedge_after_s/hedge_stall_after_s({SCOPE}__HEDGE__AFTER_S/STALL_AFTER_S)" + "的交叉守卫要求 sources 不能为空;对冲已启用但没有可校验的源" ) min_timeout = min(s.timeout_s for s in sources) - if after >= min_timeout: - raise ValueError( - f"hedge_after_s({{SCOPE}}__HEDGE__AFTER_S={after})须 < 最小源 timeout_s" - f"({min_timeout});对冲永不可能触发,配置即错误" - ) - if call_deadline_s is not None and after >= call_deadline_s: - raise ValueError( - f"hedge_after_s({after})须 < call_deadline_s({call_deadline_s});" - "期限会先于对冲触发,对冲形同虚设(设计 §6)" - ) - ttfts = [s.ttft_timeout_s for s in sources if s.ttft_timeout_s is not None] - if ttfts and after >= min(ttfts): - logger.warning( - "hedge_after_s({}) ≥ 最小源 ttft_timeout_s({}): 流式挂起会被 TTFT 看门狗" - "先行切断,对冲对流式形同虚设(非流式仍有效)", - after, - min(ttfts), - ) + if after is not None: + if after >= min_timeout: + raise ValueError( + f"hedge_after_s({{SCOPE}}__HEDGE__AFTER_S={after})须 < 最小源 timeout_s" + f"({min_timeout});对冲永不可能触发,配置即错误" + ) + if call_deadline_s is not None and after >= call_deadline_s: + raise ValueError( + f"hedge_after_s({after})须 < call_deadline_s({call_deadline_s});" + "期限会先于对冲触发,对冲形同虚设(设计 §6)" + ) + ttfts = [s.ttft_timeout_s for s in sources if s.ttft_timeout_s is not None] + if ttfts and after >= min(ttfts): + logger.warning( + "hedge_after_s({}) ≥ 最小源 ttft_timeout_s({}): 流式挂起会被 TTFT 看门狗" + "先行切断,对冲对流式形同虚设(非流式仍有效)", + after, + min(ttfts), + ) + if stall is not None: + # 有效最小值与 openai_compat.py:573 看门狗装配的回落同款: + # 配了 watchdog 的源按 inter_token_timeout_s,未配的按 timeout_s(设计 §11-N5) + min_inter = min((s.inter_token_timeout_s or s.timeout_s) for s in sources) + if stall >= min_inter: + raise ValueError( + f"hedge_stall_after_s({{SCOPE}}__HEDGE__STALL_AFTER_S={stall})须 < 最小源" + f"有效 inter-token 超时({min_inter};inter_token_timeout_s 缺省回落" + f"timeout_s,与看门狗装配同款);中段对冲窗永不开,配置即错误" + ) + if call_deadline_s is not None and stall >= call_deadline_s: + raise ValueError( + f"hedge_stall_after_s({stall})须 < call_deadline_s({call_deadline_s});" + "期限会先于中段对冲触发,对冲形同虚设(设计 §6)" + ) if len(sources) == 1: logger.warning( - "单源 scope 配置了对冲阈值 hedge_after_s={};运行期拿不到异源候选,对冲自然静默", + "单源 scope 配置了对冲阈值 hedge_after_s={}/hedge_stall_after_s={};" + "运行期拿不到异源候选,对冲自然静默", after, + stall, ) if hedge_max_extra > 1: logger.warning( "hedge_max_extra={} 已接受,但 v1 仅单路对冲生效(梯次追加为 H5 预留)", hedge_max_extra, ) - return after + return after, stall + + +def _warn_hedge_keys_on_non_stream_scope(gateway: GatewaySettings, scope_kind: str) -> None: + """B4(1.3.8 设计 §11): 装配期可证明纯非流式的 scope 配了对冲键 → warning。 + + chat scope 不做(`stream` 是 per-call 参数,装配期无信号,B4 裁决);对冲键在 + Embedding/OCR 链路没有任何消费者,静默退化必须说出来。`hedge_after_s` 一并 + 检测(1.3.7 同病顺带补齐,缺陷修复不扩公共面)。 + """ + if gateway.hedge_after_s is not None or gateway.hedge_stall_after_s is not None: + logger.warning( + "对冲键(HEDGE__AFTER_S/HEDGE__STALL_AFTER_S)仅对 chat scope 生效: " + "{} scope 非流式,对冲配置静默退化(1.3.8 B4)", + scope_kind, + ) @dataclass(frozen=True) @@ -868,6 +947,7 @@ class EmbeddingSettings: raise ValueError(f"EmbeddingSettings.batch_size 必须 ≥ 1: {self.batch_size}") if self.expected_dim is not None and self.expected_dim < 1: raise ValueError(f"EmbeddingSettings.expected_dim 必须 ≥ 1: {self.expected_dim}") + _warn_hedge_keys_on_non_stream_scope(self.gateway, "embedding") @classmethod def from_env( @@ -911,6 +991,10 @@ class OcrSettings: gateway: GatewaySettings + def __post_init__(self) -> None: + """内嵌 gateway 由 `GatewaySettings.__post_init__` 自己把关;此处只补 B4 对冲键 warning。""" + _warn_hedge_keys_on_non_stream_scope(self.gateway, "ocr") + @classmethod def from_env( cls, diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index 5c01c29..8fbbbaa 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -201,6 +201,7 @@ class RetryMW: pacer: AdaptivePacer | None = None, emitter: TelemetryEmitter | None = None, hedge_after_s: float | None = None, + stall_after_s: float | None = None, now: Callable[[], float] = time.monotonic, sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, rng: Callable[[], float] = random.random, @@ -222,6 +223,9 @@ class RetryMW: # 值域/交叉守卫在装配层(config.check_hedge_assembly),本类不重复校验; # hedge_max_extra 不下传——v1 编排固定单路对冲(H5) self._hedge_after_s = hedge_after_s + # 中段沉默对冲触发阈值(1.3.8 issue #28): None = 关闭;独立于 _hedge_after_s + # 生效(设计 §9-④),仅对流式请求开窗。守卫同在装配层,本类不重复校验 + self._stall_after_s = stall_after_s self._now = now self._sleep = sleep self._rng = rng @@ -272,8 +276,20 @@ class RetryMW: # (单路 = 成功那轮;对冲 = 赢家那一路),`_attempt` 只负责把本次 # transport 耗时与沉默观测投进来 generation_sink: list[_GenerationSample] = [] - if self._hedge_after_s is None: - # 默认关闭: 逐字 1.3.6 单路路径(默认关闭回归门据此成立) + if self._hedge_after_s is None and self._stall_after_s is None: + # 两阈值均未设: 逐字 1.3.7 单路路径(默认关闭回归门据此成立) + outcome = await self._attempt( + request, + *picked, + reasons, + attempt_fails, + first_token_event=None, + content_event=None, + generation_sink=generation_sink, + ) + elif self._hedge_after_s is None and not request.stream: + # 仅 stall 已设且请求非流式: 中段沉默无非流式概念,逐字走单路 + # (等价未配置,设计 §11-B3 入口门) outcome = await self._attempt( request, *picked, @@ -374,7 +390,15 @@ class RetryMW: self._feed_outcome(source.name, ok=True) self._pacer.on_success(source.name) response = self._build_response(source, result, call_id, started) - await self._emit(request, source, call_id, started, response=response) + await self._emit( + request, + source, + call_id, + started, + response=response, + stall_ms=stall_ms_i, + stall_count=stall_count_i, + ) return response except RequestRejectedError as exc: actual, settlement_known = 0, True # 逐字保住 1.3.5: 坏请求全额退还 @@ -441,6 +465,11 @@ class RetryMW: # Phase 1 启动原路: 事件与 sink 每轮新建(局部状态,严禁实例属性) source, permit, entry = picked first_token: asyncio.Event = asyncio.Event() + # 中段沉默观测(1.3.8): 仅"流式 + stall 阈值已设"的原路携带真事件; + # 其余恒 None = 不观测(对冲路亦恒 None,Phase 4,N3③) + content_event = ( + asyncio.Event() if (request.stream and self._stall_after_s is not None) else None + ) sink_p: list[_GenerationSample] = [] primary = asyncio.create_task( self._attempt( @@ -451,14 +480,15 @@ class RetryMW: reasons, attempt_fails, first_token_event=first_token, - content_event=None, + content_event=content_event, generation_sink=sink_p, ) ) hedge: asyncio.Task[LLMResponse | _Failed] | None = None try: - # Phase 2 触发窗: 只认"阈值到 + 首 token 未至 + 原路在途"(loop 相对时长) - if not await self._past_hedge_window(primary, first_token): + # Phase 2 触发窗: 两阶段(1.3.8)——首 token 前走 AFTER_S,首 token 后 + # 走 STALL_AFTER_S;只认 loop 相对时长,绝不读注入钟(设计 §4.1) + if not await self._past_hedge_window(primary, first_token, content_event): # 原路已了结/首 token 已至: 等价于未配置对冲 outcome = await primary if isinstance(outcome, LLMResponse): @@ -558,19 +588,32 @@ class RetryMW: await asyncio.gather(*tasks, return_exceptions=True) raise - async def _past_hedge_window(self, primary: asyncio.Task, first_token: asyncio.Event) -> bool: - """对冲触发窗: 阈值到 + 首 token 未至 + 原路在途,三者齐备才放行对冲。 + async def _past_hedge_window( + self, + primary: asyncio.Task, + first_token: asyncio.Event, + content_event: asyncio.Event | None, + ) -> bool: + """对冲触发窗两阶段(1.3.8 设计 §5.2): 首 token 前走 AFTER_S,首 token 后走 STALL_AFTER_S。 只用事件循环相对时长(`asyncio.wait` timeout),绝不读注入 `now`——测试 伪造注入钟跳变不得触发对冲(设计 §4.1)。waiter 任务必须收口,且**不得吞 外部取消**(finally 里 await 已取消的 waiter 会接住一个 CancelledError, 须靠 cancelling() 区分它是 waiter 自己的还是外面打进来的)。 """ + # 阶段 1 首 token 前: AFTER_S 已设才开窗(设计 §5.2);仅 stall 时此阶段不 + # 触发,但仍须等首 token 到达才能进阶段 2——否则 pre-TTFT 沉默会被 stall + # 窗误吞(计划 §3.6) waiter = asyncio.create_task(first_token.wait()) try: - await asyncio.wait( - {primary, waiter}, timeout=self._hedge_after_s, return_when=asyncio.FIRST_COMPLETED - ) + if self._hedge_after_s is not None: + await asyncio.wait( + {primary, waiter}, + timeout=self._hedge_after_s, + return_when=asyncio.FIRST_COMPLETED, + ) + else: + await asyncio.wait({primary, waiter}, return_when=asyncio.FIRST_COMPLETED) finally: waiter.cancel() try: @@ -578,7 +621,35 @@ class RetryMW: except asyncio.CancelledError: if asyncio.current_task().cancelling(): # 外部取消,穿透 raise - return not primary.done() and not first_token.is_set() + if self._hedge_after_s is not None and not primary.done() and not first_token.is_set(): + return True # AFTER_S 触发: 阈值到 + 首 token 未至 + 原路在途(1.3.7 原语义) + if primary.done(): + return False + # 阶段 2 首 token 后: STALL_AFTER_S 循环等待;非流式/未设 stall → 不开窗 + if self._stall_after_s is None or content_event is None: + return False + while not primary.done(): + waiter = asyncio.create_task(content_event.wait()) # 每轮自建 waiter(N3④) + try: + await asyncio.wait( + {primary, waiter}, + timeout=self._stall_after_s, + return_when=asyncio.FIRST_COMPLETED, + ) + finally: + waiter.cancel() + try: + await waiter + except asyncio.CancelledError: + if asyncio.current_task().cancelling(): # 外部取消,穿透 + raise + if primary.done(): + return False + if content_event.is_set(): # N3① 复检: chunk 与超时同 pass 到达 → 防误触发 + content_event.clear() # N3② check→clear 之间无 await + continue + return True # 复检仍未置位 = 真沉默 → 触发对冲 + return False @staticmethod def _succeeded(task: asyncio.Task, done: set[asyncio.Task]) -> bool: @@ -700,11 +771,15 @@ class RetryMW: *, response: LLMResponse | None = None, error: PolyGatewayError | str | None = None, + stall_ms: int | None = None, + stall_count: int | None = None, ) -> None: """逐次遥测(经注入的单一 Emitter);遥测失败不得影响调用(铁律)。 异常**对象原样下传**而非先 `str()` 压平(1.3.5 设计 §5): 状态码、底层异常 类型与网关响应体已经在异常上了,在这里压平就是把它们丢掉。 + stall 两值(1.3.8)只在成功分支由 `_attempt` 显式传入(计划 §3.4); + 失败/取消分支恒 None——被取消输家无 TransportResult(设计 §11-B2)。 """ if self._emitter is None: return @@ -720,6 +795,8 @@ class RetryMW: reasoning_applies=True, # 公开方法四值之一;本中间件只服务 chat 洋葱 operation="chat", + stall_ms=stall_ms, + stall_count=stall_count, ) except asyncio.CancelledError: raise diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 396442e..e1920cb 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -1179,17 +1179,16 @@ class TestHedgeConfig: 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 - ) + # 未启用对冲(两阈值皆 None)时空 sources 直接放行: 交叉守卫没有可校验的对象; + # 返回契约(1.3.8 审查 B2): tuple(归一化 after_s, 归一化 stall_after_s), + # 两阈值皆 None 才早退 + assert check_hedge_assembly( + hedge_after_s=None, + hedge_max_extra=1, + sources=[], + call_deadline_s=None, + origin="test", + ) == (None, None) def test_hedge_guard_below_min_timeout_raises(self): """阈值 ≥ 最小源 timeout_s = 对冲永不可能触发,装配期炸掉(ValueError)。""" @@ -1283,3 +1282,172 @@ class TestHedgeConfig: client = GatewayClient.from_settings(s) assert client._hedge_after_s == 8.0 assert client._terminal._hedge_after_s == 8.0 + + +class TestStallConfig: + """1.3.8 两新键: `{SCOPE}__STREAM__STALL_THRESHOLD_MS`(计量,默认 1000 三层可调) + 与 `{SCOPE}__HEDGE__STALL_AFTER_S`(中段对冲,未设=关闭)及其装配守卫(设计 §6, + 计划批次 D;单一定义点仍是 `config.check_hedge_assembly`)。""" + + def _two_source_env(self, **overrides): + """双源 env(同 provider 避免注册表依赖): 隔离单源 warning 的干扰。""" + return _env( + **{ + "LLM__QWEN__2__BASE_URL": "https://gw-b.example/v1", + "LLM__QWEN__2__API_KEY": "sk-b", + "LLM__QWEN__2__MODEL": "qwen-plus", + "LLM__QWEN__2__TIMEOUT_S": "90", + **overrides, + } + ) + + def test_stall_threshold_key_loads_three_layers(self): + """计量阈值三层调整接口(§9-③ 人类要求): env 键 / settings 字段 / client kwarg。 + + 分发语义: from_settings 恒以 settings 值为权威(覆盖源自带值);client 直传 + None = 保留各 SourceConfig 自带值,非 None 经 dataclasses.replace 分发。""" + from tests.unit.test_client import _client + + # 层 1: env 键 → settings 字段 → from_settings 分发进每个 SourceConfig + env = self._two_source_env(**{"LLM__STREAM__STALL_THRESHOLD_MS": "2500"}) + s = GatewaySettings.from_env("LLM", env=env) + assert s.stall_threshold_ms == 2500 + client = GatewayClient.from_settings(s) + assert {src.stall_threshold_ms for src in client._terminal._sources} == {2500} + # 层 2: settings 字段直接构造(dataclasses.replace 同路),settings 值权威: + # 源自带 500 也被 scope 值覆盖(恒分发,计划 §3.5) + base = GatewaySettings.from_env("LLM", env=_env()) + custom_source = dataclasses.replace(base.sources[0], stall_threshold_ms=500) + s2 = dataclasses.replace(base, sources=(custom_source,), stall_threshold_ms=800) + client2 = GatewayClient.from_settings(s2) + assert client2._terminal._sources[0].stall_threshold_ms == 800 + # 层 3: client kwarg——非 None 分发,None(缺省)保留源自带值 + client3 = _client(stall_threshold_ms=600) + assert client3._terminal._sources[0].stall_threshold_ms == 600 + client4 = _client(sources=[custom_source]) + assert client4._terminal._sources[0].stall_threshold_ms == 500 # 不 stomp + # 未设恒 1000(SourceConfig 缺省) + assert GatewaySettings.from_env("LLM", env=_env()).stall_threshold_ms == 1000 + assert _client()._terminal._sources[0].stall_threshold_ms == 1000 + + def test_stream_segment_reserved(self): + """`STREAM` 进保留段: 四段 `LLM__STREAM__1__*` 不造源(1.3.7 HEDGE 同款)。""" + env_collision = _env( + **{ + "LLM__STREAM__1__BASE_URL": "https://gw-c.example/v1", + "LLM__STREAM__1__API_KEY": "sk-c", + "LLM__STREAM__1__MODEL": "m-c", + "LLM__STREAM__1__TIMEOUT_S": "60", + } + ) + s = GatewaySettings.from_env("LLM", env=env_collision) + assert [src.name for src in s.sources] == ["qwen_1"] + # 三段键天然跳过 `_load_sources` 的段数判据,也不造源 + env = _env(**{"LLM__STREAM__STALL_THRESHOLD_MS": "2000"}) + s2 = GatewaySettings.from_env("LLM", env=env) + assert [src.name for src in s2.sources] == ["qwen_1"] + + def test_stall_after_s_guard_effective_minimum_raises(self): + """`stall_after_s ≥ min(inter_token_timeout_s or timeout_s)` → ValueError。 + + 有效值回落与 openai_compat.py:573 看门狗装配同款: 配 watchdog 的源用 + inter_token_timeout_s,未配的源用 timeout_s;对冲窗永不开装配期直接拒绝。""" + # 源 1 配 watchdog(inter_token 15)、源 2 纯 timeout(90): 有效最小值 = 15 + env = self._two_source_env( + **{ + "LLM__QWEN__1__TTFT_TIMEOUT_S": "30", + "LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S": "15", + "LLM__HEDGE__STALL_AFTER_S": "15", # ≥ 15 → 拒绝(边界含等号) + } + ) + with pytest.raises(ValueError, match="STALL_AFTER_S"): + GatewaySettings.from_env("LLM", env=env) + s = GatewaySettings.from_env( + "LLM", + env=self._two_source_env( + **{ + "LLM__QWEN__1__TTFT_TIMEOUT_S": "30", + "LLM__QWEN__1__INTER_TOKEN_TIMEOUT_S": "15", + "LLM__HEDGE__STALL_AFTER_S": "14", + } + ), + ) + assert s.hedge_stall_after_s == 14.0 + # 全源未配 watchdog: 有效最小值 = min(timeout_s) = 90 + with pytest.raises(ValueError, match="STALL_AFTER_S"): + GatewaySettings.from_env( + "LLM", env=self._two_source_env(**{"LLM__HEDGE__STALL_AFTER_S": "90"}) + ) + + def test_stall_after_s_guard_deadline_conflict_raises(self): + """`stall_after_s ≥ call_deadline_s` → ValueError(与 after_s 同款,§11-N5)。""" + env = self._two_source_env( + **{"LLM__HEDGE__STALL_AFTER_S": "30", "LLM__CALL_DEADLINE_S": "30"} + ) + with pytest.raises(ValueError, match="call_deadline_s"): + GatewaySettings.from_env("LLM", env=env) + s = GatewaySettings.from_env( + "LLM", + env=self._two_source_env( + **{"LLM__HEDGE__STALL_AFTER_S": "29", "LLM__CALL_DEADLINE_S": "30"} + ), + ) + assert s.hedge_stall_after_s == 29.0 and s.call_deadline_s == 30.0 + + def test_stall_after_s_single_source_warns(self): + """单源 scope 设 stall 阈值: 装配期 warning 放行(1.3.7 §5 同款)。""" + with _captured_warnings() as warnings: + s = GatewaySettings.from_env( + "LLM", env=_env(**{"LLM__HEDGE__STALL_AFTER_S": "8"}) + ) + assert s.hedge_stall_after_s == 8.0 + assert any("单源" in m for m in warnings) + + def test_embedding_ocr_scope_hedge_keys_warn(self): + """B4: Embedding/OCR scope 配对冲键(AFTER_S 或 STALL_AFTER_S)→ 装配 warning; + chat scope 无此 warning(per-call stream,装配期无信号,B4 裁决)。""" + # Embedding: env 路(AFTER_S 与 STALL_AFTER_S 各验一次) + embed_env = { + "EMBED__QWEN__1__BASE_URL": "https://gw.example/v1", + "EMBED__QWEN__1__API_KEY": "sk-a", + "EMBED__QWEN__1__MODEL": "text-embedding-v3", + "EMBED__QWEN__1__TIMEOUT_S": "60", + "EMBED__RETRY__MAX_ATTEMPTS": "3", + "EMBED__RETRY__BACKOFF_BASE_S": "1.0", + "EMBED__RETRY__BACKOFF_MAX_S": "10.0", + "EMBED__BREAKER__FAIL_THRESHOLD": "5", + "EMBED__BREAKER__COOLDOWN_S": "60", + "PGW_CACHE_BACKEND": "none", + "PGW_TELEMETRY_BACKEND": "none", + "EMBED__BATCH_SIZE": "64", + } + with _captured_warnings() as warnings: + EmbeddingSettings.from_env( + "EMBED", env={**embed_env, "EMBED__HEDGE__AFTER_S": "8"} + ) + assert any("仅对 chat scope 生效" in m for m in warnings) + with _captured_warnings() as warnings: + EmbeddingSettings.from_env( + "EMBED", env={**embed_env, "EMBED__HEDGE__STALL_AFTER_S": "8"} + ) + assert any("仅对 chat scope 生效" in m for m in warnings) + # Embedding: 直构造路(__post_init__ 那道,非 from_env 专属) + gw = GatewaySettings.from_env("EMBED", env=embed_env) + gw_hedged = dataclasses.replace(gw, hedge_stall_after_s=8.0) + with _captured_warnings() as warnings: + EmbeddingSettings(gateway=gw_hedged, batch_size=64) + assert any("仅对 chat scope 生效" in m for m in warnings) + # OCR: env 路(1.3.7 AFTER_S 同病顺带补齐 + 新 STALL_AFTER_S) + ocr_env = dict(TestOcrSettings._OCR_ENV) + with _captured_warnings() as warnings: + OcrSettings.from_env("OCR", env={**ocr_env, "OCR__HEDGE__AFTER_S": "8"}) + assert any("仅对 chat scope 生效" in m for m in warnings) + with _captured_warnings() as warnings: + OcrSettings.from_env("OCR", env={**ocr_env, "OCR__HEDGE__STALL_AFTER_S": "8"}) + assert any("仅对 chat scope 生效" in m for m in warnings) + # chat 侧: 同样的对冲键不产生 B4 warning(双源 env 隔离单源 warning 干扰) + with _captured_warnings() as warnings: + GatewaySettings.from_env( + "LLM", env=self._two_source_env(**{"LLM__HEDGE__STALL_AFTER_S": "8"}) + ) + assert not any("仅对 chat scope 生效" in m for m in warnings) diff --git a/tests/unit/test_stall_hedge.py b/tests/unit/test_stall_hedge.py new file mode 100644 index 0000000..85f17bc --- /dev/null +++ b/tests/unit/test_stall_hedge.py @@ -0,0 +1,479 @@ +"""中段沉默对冲编排测试(1.3.8 issue #28 设计 §5.2 / 计划批次 D)。 + +设施纪律(计划 §5): 复用 test_hedge.py 范式——两源 scope、事件驱动假 transport、 +**真实 loop 钟**(不注入 FakeClock 驱动触发;注入钟仅用于赢家净口径精确断言, +触发路径结构性不读它)、阈值 0.05s、断言容差 4–10×;取消窗口用 entered Event +栅栏钉死,禁 sleep 撞窗口;不标 slow。 + +命名区隔(审查 N5): 本文件的 `stall_after_s` 是**对冲触发阈值**(相邻增量 chunk +间隔超过它 → 触发对冲);test_hedge.py:150 的 `stall_window_s` 是 **StallClock +预算**(非生产性等待的调用级上限)——同名词不同概念,勿混。 +""" + +import asyncio +import dataclasses +import time + +import pytest + +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_hedge import RecordingEmitter +from tests.unit.test_retry import 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) +_STALL_AFTER_S = 0.05 # 真实 loop 钟对冲触发阈值(≠ StallClock 预算 stall_window_s,N5) + + +class StallHedgeTransport: + """事件驱动假 transport(中段沉默剧本): 逐步骤控制 chunk 事件/挂起/成败。 + + 剧本 = 每源一串**步骤**,每次 `complete` 调用从上次位置继续执行到终态步骤 + (`ok`/`fail`)为止;多轮调用(重试)顺序消费后续步骤,耗尽即 AssertionError。 + + 步骤: + ("chunk",) — 置 first_token_event 与 content_event(首个增量 chunk) + ("chunk_after", delay) — 睡 delay 后只置 content_event(后续增量 chunk) + ("hang",) — 置位该源 entered,挂起直到该源 release(或被取消) + ("ok", content) — 终态: 立即成功(带已登记的 stall/clock 效果) + ("ok_after", delay, content) — 终态: 真实 loop 钟睡 delay 后成功 + ("fail_after", delay, factory) — 终态: 睡 delay 后抛 `factory()` 新造的异常 + ("stall", ms, count) — 非终态: 给本剧本终态结果打自报 stall 值(假 + transport 自己报数,不经真实计时,计划 §3.3) + ("advance", secs) — 非终态: 推进注入钟(仅配 FakeClock 的用例) + """ + + def __init__(self, scripts: dict[str, list[tuple]], clock: FakeClock | None = None): + self._scripts = {name: list(steps) for name, steps in scripts.items()} + self._pos = dict.fromkeys(scripts, 0) + self._clock = clock + self.calls: list[str] = [] + # 逐次记录收到的 first_token_event/content_event 身份(对冲路恒 None 断言原料) + self.ft_events: list[object] = [] + self.ce_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, + content_event, + ): + self.calls.append(source.name) + self.ft_events.append(first_token_event) + self.ce_events.append(content_event) + steps = self._scripts[source.name] + stall: tuple[float, float] | None = None + while self._pos[source.name] < len(steps): + action = steps[self._pos[source.name]] + self._pos[source.name] += 1 + kind = action[0] + if kind == "chunk": + if first_token_event is not None: + first_token_event.set() + if content_event is not None: + content_event.set() + elif kind == "chunk_after": + await asyncio.sleep(action[1]) + if content_event is not None: + content_event.set() + elif kind == "hang": + self.entered[source.name].set() + await self.release[source.name].wait() + elif kind == "stall": + stall = (action[1], action[2]) + elif kind == "advance": + assert self._clock is not None, "advance 步骤需要构造时注入 FakeClock" + self._clock.advance(action[1]) + elif kind == "ok": + result = _ok(action[1]) + if stall is not None: + result = dataclasses.replace(result, stall_ms=stall[0], stall_count=stall[1]) + return result + elif kind == "ok_after": + await asyncio.sleep(action[1]) + result = _ok(action[2]) + if stall is not None: + result = dataclasses.replace(result, stall_ms=stall[0], stall_count=stall[1]) + return result + elif kind == "fail_after": + await asyncio.sleep(action[1]) + raise action[2]() + else: + raise AssertionError(f"未知剧本步骤: {action!r}") + raise AssertionError(f"{source.name} 剧本耗尽且无终态步骤") + + +def _harness( + sources, + transport, + *, + hedge_after_s=None, + stall_after_s=None, + max_attempts=3, + emitter=None, + selector=None, + limiter=None, + gate=None, + stall_window_s=300.0, + now=None, +): + """真实 loop 钟装配(同 test_hedge.py 范式;`now` 仅供净口径精确断言注入 FakeClock)。 + + `stall_window_s` 是 StallClock 预算(N5 命名区隔),与 `stall_after_s` 无关。 + """ + 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, + stall_after_s=stall_after_s, + **({"now": now} if now is not None else {}), + ) + return mw, limiter, gate + + +def _req(*, stream=True, ctx=None): + """中段沉默是流式概念,本文件默认 stream=True(test_hedge.py 默认 False 相反)。""" + return ChatRequest( + messages=[{"role": "user", "content": "hi"}], stream=stream, call_context=ctx + ) + + +def _ctx(now=None): + return _CallContext(now=now if now is not None else time.monotonic) + + +class TestStallHedgeTrigger: + """阶段 2 触发(设计 §5.2): 首 token 已至后相邻增量 chunk 间隔超阈值 → 对冲。""" + + async def test_stall_triggers_hedge_after_first_token(self): + """s1 首 chunk 后沉默、s2 即时成功: stall 窗触发对冲,赢家为对冲路。 + + 净口径顺延断言(计划批次 D 首行): 注入 FakeClock + s2 自报 stall + (1500ms/2 段)、advance 2.0s → 赢家 `generation_ms == 500` 精确值, + `generation_ms + stall_ms == total_latency_ms`(同钟还原公式)。 + 遥测行(T3 缺口补齐钉): 赢家行 stall 两值为自报值,输家行恒 None。 + """ + clock = FakeClock() + emitter = RecordingEmitter() + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport( + { + "s1": [("chunk",), ("hang",), ("ok", "late")], + "s2": [("stall", 1500.0, 2), ("advance", 2.0), ("ok", "fast")], + }, + clock=clock, + ) + mw, _, _ = _harness( + [s1, s2], transport, stall_after_s=_STALL_AFTER_S, emitter=emitter, now=clock + ) + ctx = _ctx(now=clock) + started = time.monotonic() + resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5) + elapsed = time.monotonic() - started + assert resp.content == "fast" and resp.source_name == "s2" + assert elapsed < 10 * _STALL_AFTER_S # 沉默路被对冲截断,而非等到看门狗 + assert transport.calls == ["s1", "s2"] + stats = ctx.snapshot() + assert stats.hedges == 1 and stats.hedge_won is True and stats.attempts == 2 + # 赢家路净口径: 墙钟 2000(advance 精确推进)− 沉默 1500 = 500 + assert stats.generation_ms == 500 + assert stats.stall_ms == 1500 and stats.stall_count == 2 + assert stats.generation_ms + stats.stall_ms == stats.total_latency_ms # 还原公式 + # 逐次遥测: 赢家行带 stall 归因,被取消输家行恒 None(设计 §11-B2) + winner = next(r for r in emitter.rows if r["source"] == "s2") + loser = next(r for r in emitter.rows if r["source"] == "s1") + assert winner["stall_ms"] == 1500 and winner["stall_count"] == 2 + assert winner["error"] is None + assert loser["error"] == "hedge_cancelled" + assert loser["stall_ms"] is None and loser["stall_count"] is None + + async def test_single_leg_attempt_row_carries_stall(self): + """T3 缺口补齐钉: 未启用对冲的 chat 成功行也须经 `_emit` 落 stall 两值。 + + 计划 §3.4 指定 `retry._emit` 显式转发(归 T3);批次 C 落行断言只覆盖了 + emitter 直调,chat 路径漏传无人把守——本用例钉住端到端路径。 + """ + clock = FakeClock() + emitter = RecordingEmitter() + transport = StallHedgeTransport( + {"s1": [("stall", 300.0, 1), ("advance", 1.0), ("ok", "x")]}, clock=clock + ) + mw, _, _ = _harness([_src("s1")], transport, emitter=emitter, now=clock) + ctx = _ctx(now=clock) + resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5) + assert resp.content == "x" + assert len(emitter.rows) == 1 + row = emitter.rows[0] + assert row["error"] is None + assert row["stall_ms"] == 300 and row["stall_count"] == 1 + stats = ctx.snapshot() + assert stats.generation_ms == 700 and stats.stall_ms == 300 + + async def test_stall_recheck_prevents_false_trigger(self): + """chunk 在超时到期前一刻到达(0.049s 置位,阈值 0.05s)→ 不触发(N3① 复检)。 + + 只断"不触发 + attempts==1"结果,不断精确时序(计划 §7 抖动纪律)。 + """ + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport( + { + "s1": [("chunk",), ("chunk_after", 0.049), ("ok", "done")], + "s2": [("ok", "other")], + } + ) + mw, _, _ = _harness([s1, s2], transport, stall_after_s=_STALL_AFTER_S) + ctx = _ctx() + resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5) + assert resp.content == "done" and resp.source_name == "s1" + assert transport.calls == ["s1"] # 对冲从未发出 + stats = ctx.snapshot() + assert stats.hedges == 0 and stats.attempts == 1 and stats.hedge_won is False + + async def test_stall_and_after_s_independent_matrix(self): + """参数化四格(设计 §5.2 逐字): 两阈值独立生效,各格触发语义不串。""" + # 格 1: 仅 AFTER_S + 流式 pre-TTFT 挂起 → AFTER_S 触发(1.3.7 语义不变) + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport( + {"s1": [("hang",), ("ok", "late")], "s2": [("ok", "fast")]} + ) + mw, _, _ = _harness([s1, s2], transport, hedge_after_s=_STALL_AFTER_S) + ctx = _ctx() + resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5) + assert resp.source_name == "s2" + assert ctx.snapshot().hedges == 1 + + # 格 2: 仅 STALL_AFTER_S + 流式 pre-TTFT 沉默 → **不触发**(阶段 1 不开窗) + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport( + {"s1": [("hang",), ("ok", "s1ok")], "s2": [("ok", "other")]} + ) + mw, _, _ = _harness([s1, s2], transport, stall_after_s=_STALL_AFTER_S) + ctx = _ctx() + task = asyncio.ensure_future(mw(_req(ctx=ctx))) + await transport.entered["s1"].wait() + await asyncio.sleep(4 * _STALL_AFTER_S) # 4× 余量确认 stall 窗管不到 pre-TTFT + 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 + + # 格 3: 两者皆设 + 首 chunk 后沉默 → 阶段 2 stall 窗触发 + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport( + {"s1": [("chunk",), ("hang",), ("ok", "late")], "s2": [("ok", "fast")]} + ) + mw, _, _ = _harness( + [s1, s2], transport, hedge_after_s=_STALL_AFTER_S, stall_after_s=_STALL_AFTER_S + ) + ctx = _ctx() + resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5) + assert resp.source_name == "s2" + assert ctx.snapshot().hedges == 1 + + # 格 3b: 两者皆设 + pre-TTFT 挂起 → 阶段 1 AFTER_S 触发(两阶段各管一段) + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport( + {"s1": [("hang",), ("ok", "late")], "s2": [("ok", "fast")]} + ) + mw, _, _ = _harness( + [s1, s2], transport, hedge_after_s=_STALL_AFTER_S, stall_after_s=_STALL_AFTER_S + ) + ctx = _ctx() + resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5) + assert resp.source_name == "s2" + assert ctx.snapshot().hedges == 1 + + # 格 4: 皆无 → 逐字单路(默认关闭回归门的单元格形态) + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport({"s1": [("hang",), ("ok", "solo")]}) + mw, _, _ = _harness([s1, s2], transport) + ctx = _ctx() + task = asyncio.ensure_future(mw(_req(ctx=ctx))) + await transport.entered["s1"].wait() + transport.release["s1"].set() + resp = await asyncio.wait_for(task, timeout=5) + assert resp.content == "solo" and resp.source_name == "s1" + stats = ctx.snapshot() + assert stats.hedges == 0 and stats.attempts == 1 + assert transport.ce_events == [None] # 未配置 = 不观测增量 chunk + + async def test_stall_only_non_stream_degrades_to_single_leg(self): + """仅 stall 已设 + `stream=False` → 逐字单路(中段沉默无非流式概念,入口门)。""" + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport( + {"s1": [("hang",), ("ok", "x")], "s2": [("ok", "other")]} + ) + mw, _, _ = _harness([s1, s2], transport, stall_after_s=_STALL_AFTER_S) + ctx = _ctx() + task = asyncio.ensure_future(mw(_req(stream=False, ctx=ctx))) + await transport.entered["s1"].wait() + await asyncio.sleep(4 * _STALL_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 + assert transport.ce_events == [None] # 非流式不建 content_event + + +class TestStallHedgeWatchdogAndFailure: + """与看门狗的交互(设计 §5.2): 主路被掐死 ≠ 本轮失败,副路已在跑则等副路。""" + + async def test_watchdog_kills_primary_hedge_leg_survives(self): + """主路首 chunk 后被看门狗掐死(Transient)、副路在途 → 副路结果裁定。 + + attempt 级记账照常(N9): 主路源熔断失败计数 +1、遥测主路 error 行 + (TransientError 对象)+ 副路正常行;"先了结的是失败则等另一路"逐字复用。 + """ + emitter = RecordingEmitter() + gate = InMemoryGate(config=_BREAKER) + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport( + { + "s1": [ + ("chunk",), + ( + "fail_after", + 0.12, + lambda: TransientError("inter_token timeout", source_name="s1"), + ), + ], + "s2": [("ok_after", 0.2, "rescued")], + } + ) + mw, _, _ = _harness( + [s1, s2], transport, stall_after_s=_STALL_AFTER_S, emitter=emitter, gate=gate + ) + ctx = _ctx() + resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5) + assert resp.content == "rescued" and resp.source_name == "s2" + stats = ctx.snapshot() + assert stats.hedges == 1 and stats.hedge_won is True + # 掐死是 attempt 级失败: 熔断失败样本 +1(对冲输家取消不喂,此处是喂的——N9 不对称) + gate_s1 = gate._gates["s1"] + assert gate_s1.a0 + gate_s1.a1 == 1 + # 遥测: 主路 error 行携 TransientError 对象(掐死),副路正常行 + primary_row = next(r for r in emitter.rows if r["source"] == "s1") + hedge_row = next(r for r in emitter.rows if r["source"] == "s2") + assert isinstance(primary_row["error"], TransientError) + assert hedge_row["error"] is None + + async def test_both_fail_counts_budget_once(self): + """主路看门狗掐死 + 副路 Transient: 两败只计一次预算(_combine_failures 复用)。 + + max_attempts=2 时恰进第二轮: 若两败计两次,第一轮即耗尽,第二轮的两路 + 调用根本不会发生。两败轮次无赢家 → generation 不记、stall 快照恒 None。 + """ + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport( + { + "s1": [ + ("chunk",), + ("fail_after", 0.1, lambda: TransientError("wd", source_name="s1")), + ("chunk",), + ("fail_after", 0.1, lambda: TransientError("wd", source_name="s1")), + ], + "s2": [ + ("fail_after", 0.12, lambda: TransientError("h", source_name="s2")), + ("fail_after", 0.12, lambda: TransientError("h", source_name="s2")), + ], + } + ) + mw, _, _ = _harness([s1, s2], transport, stall_after_s=_STALL_AFTER_S, max_attempts=2) + ctx = _ctx() + with pytest.raises(AllSourcesExhausted) as ei: + await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5) + assert ei.value.reason == "retry_exhausted" + assert transport.calls == ["s1", "s2", "s1", "s2"] # 两败计一次才有第二轮 + stats = ctx.snapshot() + assert stats.hedges == 2 # 两败轮次路数照登(实际并发发出即计) + assert stats.hedge_won is False + assert stats.generation_ms == 0 # 无赢家: 裸生成时间无归属不记 + assert stats.stall_ms is None and stats.stall_count is None + + +class TestStallHedgeCancellation: + """取消穿透(计划取消铁律): stall 等待中外部取消,waiter 收口无残留。""" + + async def test_external_cancel_during_stall_wait_penetrates(self): + """阶段 2 循环内外部 `task.cancel()`: CancelledError 穿透、无残留任务。 + + 阈值取 5.0s 宽窗: 取消确定落在 stall 等待内(阶段 2 循环),不依赖时序。 + 对冲尚未发出(宽窗未到期)→ 只有原路被取消,permit 释放。 + """ + before = asyncio.all_tasks() + emitter = RecordingEmitter() + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport( + {"s1": [("chunk",), ("hang",), ("ok", "x")], "s2": [("ok", "y")]} + ) + mw, limiter, _ = _harness([s1, s2], transport, stall_after_s=5.0, emitter=emitter) + task = asyncio.ensure_future(mw(_req(ctx=_ctx()))) + await transport.entered["s1"].wait() # 原路已交付首 chunk 并挂起: 编排处于阶段 2 + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert transport.calls == ["s1"] # 宽窗未到期,对冲从未发出 + assert {r["source"]: r["error"] for r in emitter.rows} == {"s1": "cancelled"} + assert (await limiter.source_stats("s1")).inflight == 0 + assert asyncio.all_tasks() == before # waiter/原路全收口,无残留 + + +class TestStallHedgeEventDiscipline: + """Event 并发纪律(N3③): 对冲路 content_event 恒 None(v1 不梯次)。""" + + async def test_hedge_leg_content_event_always_none(self): + """原路(流式 + stall 已设)收真事件,对冲路恒 None——逐次身份记录断言。""" + s1, s2 = _src("s1"), _src("s2") + transport = StallHedgeTransport( + {"s1": [("chunk",), ("hang",), ("ok", "late")], "s2": [("ok", "win")]} + ) + mw, _, _ = _harness([s1, s2], transport, stall_after_s=_STALL_AFTER_S) + ctx = _ctx() + resp = await asyncio.wait_for(mw(_req(ctx=ctx)), timeout=5) + assert resp.source_name == "s2" + assert transport.calls == ["s1", "s2"] + # 原路携带首 token + content 双观测;对冲路两事件恒 None(N3③,同 H5 首 token 约定) + assert [e is not None for e in transport.ft_events] == [True, False] + assert [e is not None for e in transport.ce_events] == [True, False]