From bc4683d1f5e01dd23cfbdbcbf395a514f4debd56 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Thu, 6 Aug 2026 10:36:40 -0400 Subject: [PATCH] test: make the per-call clock invariant actually testable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The concurrency case used two RetryMW instances, so instance-level sharing was hidden by object isolation and a clock promoted to an instance attribute passed all seven cases. Both cases now reuse one mw, and a new one idles past the window between two calls on that instance — the shape that would expose _entered_at pinned to process start. Mutation-checked: promoting the clock fails the new case. --- tests/unit/test_backpressure.py | 39 +++++++++++++++++++++++++++------ 1 file changed, 32 insertions(+), 7 deletions(-) diff --git a/tests/unit/test_backpressure.py b/tests/unit/test_backpressure.py index 4bd4ee2..eb0fea8 100644 --- a/tests/unit/test_backpressure.py +++ b/tests/unit/test_backpressure.py @@ -130,7 +130,11 @@ class TestStallQuadrants: assert resp.content == "ok" async def test_global_stale_but_local_fresh_keeps_waiting(self): - """仅全局超窗(从未出餐 age=inf): 本地才刚开始等 → 不判死。""" + """仅全局超窗(从未出餐 age=inf): 本地才刚开始等 → 不判死。 + + `inf` 语义在 issue #8 后未变;变的是"本地"的口径——它现在度量的是 + 非生产性等待累计,不再是墙钟总耗时(见 TestStallBudget)。 + """ clock = FakeClock() src, limiter = _blocked_limiter(clock) held = await limiter.try_acquire("s1", 0) @@ -338,18 +342,39 @@ class TestStallBudget: await task assert (await limiter.source_stats("s1")).inflight == 0 # permit 在 finally 释放 + async def test_clock_is_per_call_not_per_instance(self): + """StallClock 必须是**调用级**局部状态,不得提升为 RetryMW 实例属性。 + + 生产形态是一个长寿命 RetryMW 跑成千上万次调用。若 clock 成了实例属性, + `_entered_at` 会固定在进程启动时刻, 每次调用的 stalled_s() 随进程运行 + 时长单调增长, 最终所有调用被误判 stalled——这是本用例要拦的灾难。 + + 判别力的关键是**复用同一个 mw**: 两个 mw 实例天然隔离, 抓不到实例共享。 + """ + clock = FakeClock() + src = make_source() + limiter = InMemoryLimiter( + scope="llm", sources={"s1": src}, global_limits=_NO_GLOBAL, now=clock + ) + transport = ClockAdvancingTransport([(0.0, _ok("first")), (0.0, _ok("second"))], clock) + mw = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(), transport=transport) + first = await mw(_REQ) + clock.advance(_STALL + 100) # 两次调用之间进程空转远超窗 + second = await mw(_REQ) + assert (first.content, second.content) == ("first", "second") + async def test_concurrent_calls_do_not_share_clock(self): - """StallClock 必须是调用级局部状态: 一路长尝试不得污染另一路的 stall 账。""" + """并发两路共用同一个 mw: 一路长尝试不得污染另一路的 stall 账。""" clock = FakeClock() src = make_source(max_concurrency=2) limiter = InMemoryLimiter( scope="llm", sources={"s1": src}, global_limits=_NO_GLOBAL, now=clock ) - slow = ClockAdvancingTransport([(_STALL + 100, _ok("slow"))], clock) - fast = ClockAdvancingTransport([(0.0, _ok("fast"))], clock) - mw_slow = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(), transport=slow) - mw_fast = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(), transport=fast) - results = await asyncio.gather(mw_slow(_REQ), mw_fast(_REQ)) + transport = ClockAdvancingTransport( + [(_STALL + 100, _ok("slow")), (0.0, _ok("fast"))], clock + ) + mw = _mw([src], limiter, [], clock=clock, sleep=BoundedSleep(), transport=transport) + results = await asyncio.gather(mw(_REQ), mw(_REQ)) assert {r.content for r in results} == {"slow", "fast"}