test: pin the expiry window and close settlement review gaps
This commit is contained in:
@@ -634,6 +634,7 @@ async def gather_bounded[T](aws: Iterable[Awaitable[T]], *, concurrency: int) ->
|
|||||||
"""有界并发 gather(D5 便利函数,替代 VT 手搓 semaphore+gather 样板)。
|
"""有界并发 gather(D5 便利函数,替代 VT 手搓 semaphore+gather 样板)。
|
||||||
|
|
||||||
语义与 `asyncio.gather` 默认一致: 结果保序、首个异常上抛;仅增加并发上限。
|
语义与 `asyncio.gather` 默认一致: 结果保序、首个异常上抛;仅增加并发上限。
|
||||||
|
期限计时从每次调用真正开始执行起算,信号量排队时长不在 `call_deadline_s` 之内。
|
||||||
"""
|
"""
|
||||||
if concurrency < 1:
|
if concurrency < 1:
|
||||||
raise ValueError("concurrency 必须 ≥ 1")
|
raise ValueError("concurrency 必须 ≥ 1")
|
||||||
|
|||||||
@@ -65,7 +65,7 @@ async def with_call_deadline[T](aw: Awaitable[T], *, deadline_s: float | None, s
|
|||||||
# 体内(含清理路径)自抛的 TimeoutError 的**身份**,唯一可靠的区分依据
|
# 体内(含清理路径)自抛的 TimeoutError 的**身份**,唯一可靠的区分依据
|
||||||
inner_timeout: BaseException | None = None
|
inner_timeout: BaseException | None = None
|
||||||
# 先建对象再进上下文: `as cm` 只在 `__aenter__` 返回后才绑定, 而 except 块无条件
|
# 先建对象再进上下文: `as cm` 只在 `__aenter__` 返回后才绑定, 而 except 块无条件
|
||||||
# 读 `cm`——进入阶段一旦异常就会变成 NameError 掩盖真实错误
|
# 读 `cm`——进入阶段一旦抛 TimeoutError 就会变成 NameError 掩盖真实错误
|
||||||
cm = asyncio.timeout(deadline_s)
|
cm = asyncio.timeout(deadline_s)
|
||||||
try:
|
try:
|
||||||
async with cm:
|
async with cm:
|
||||||
|
|||||||
@@ -5,6 +5,7 @@
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import time
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
@@ -124,14 +125,23 @@ async def test_narrow_success_returns_value_without_pending_cancellation():
|
|||||||
|
|
||||||
|
|
||||||
async def test_domain_error_inside_window_propagates():
|
async def test_domain_error_inside_window_propagates():
|
||||||
|
"""计时器已触发但取消尚未投递的窗口内,体内先抛领域异常 → 原样上抛,期限静默让位。
|
||||||
|
|
||||||
|
忙等超过期限: 计时器回调已在 loop 上触发,但任务不挂起取消就投递不进来,
|
||||||
|
此刻体内同步抛出的领域异常必须原样逃逸(设计 §5.2 形态四)。
|
||||||
|
"""
|
||||||
|
|
||||||
class BoomError(RuntimeError):
|
class BoomError(RuntimeError):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
async def body():
|
async def body():
|
||||||
|
end = time.monotonic() + 0.1
|
||||||
|
while time.monotonic() < end:
|
||||||
|
pass
|
||||||
raise BoomError("boom")
|
raise BoomError("boom")
|
||||||
|
|
||||||
with pytest.raises(BoomError):
|
with pytest.raises(BoomError):
|
||||||
await with_call_deadline(body(), deadline_s=0.05, scope="llm")
|
await with_call_deadline(body(), deadline_s=0.02, scope="llm")
|
||||||
|
|
||||||
|
|
||||||
async def test_none_deadline_takes_the_legacy_path():
|
async def test_none_deadline_takes_the_legacy_path():
|
||||||
|
|||||||
@@ -700,7 +700,7 @@ class TestEmbedCallDeadline:
|
|||||||
elapsed = loop.time() - started
|
elapsed = loop.time() - started
|
||||||
assert exc.value.scope == "embed"
|
assert exc.value.scope == "embed"
|
||||||
# 按批计的话 20 批全都能跑完(根本不会抛),共享一份则跑不到头
|
# 按批计的话 20 批全都能跑完(根本不会抛),共享一份则跑不到头
|
||||||
assert 2 <= len(transport.calls) < 20
|
assert 1 <= len(transport.calls) < 20
|
||||||
assert elapsed < 20 * 0.05, f"总时长疑似随批数放大: {elapsed}s"
|
assert elapsed < 20 * 0.05, f"总时长疑似随批数放大: {elapsed}s"
|
||||||
|
|
||||||
async def test_empty_input_is_exempt_from_the_deadline(self):
|
async def test_empty_input_is_exempt_from_the_deadline(self):
|
||||||
|
|||||||
@@ -585,6 +585,22 @@ class TestCancellationSettlement:
|
|||||||
await mw(_REQ)
|
await mw(_REQ)
|
||||||
assert (await limiter.source_stats("a")).tpm_used == 0
|
assert (await limiter.source_stats("a")).tpm_used == 0
|
||||||
|
|
||||||
|
async def test_circuit_open_rejection_settles_zero_end_to_end(self):
|
||||||
|
"""S1 端到端: 开路拒绝的 pick 预扣后按 0 结算, 不给 tpm_used 增加任何量。"""
|
||||||
|
clock = FakeClock()
|
||||||
|
script = [TransientError(str(i)) for i in range(9)]
|
||||||
|
mw, limiter, *_ = _harness(
|
||||||
|
[_src("a", max_concurrency=1, tpm=10000, est_tokens=400)],
|
||||||
|
script,
|
||||||
|
clock=clock,
|
||||||
|
max_attempts=99,
|
||||||
|
)
|
||||||
|
# 3 次瞬时失败后 a 开路 → 第 4 次 pick 被拒绝
|
||||||
|
with pytest.raises(CircuitOpenError):
|
||||||
|
await mw(_REQ)
|
||||||
|
# 三次瞬时失败各保留 est = 1200; 开路那次 pick 若漏了 settle(0) 会再 +400
|
||||||
|
assert (await limiter.source_stats("a")).tpm_used == 1200
|
||||||
|
|
||||||
|
|
||||||
class TestCancellation:
|
class TestCancellation:
|
||||||
async def test_cancel_mid_flight_releases_permit(self):
|
async def test_cancel_mid_flight_releases_permit(self):
|
||||||
|
|||||||
Reference in New Issue
Block a user