test(timing): 时序注入补上"竞争者占着 CPU"这条轴

.github/workflows/ci.yml

@@ -71,10 +71,15 @@ jobs:
   #   - 线程被 start() 了但迟迟跑不到第一行（等待被拉长管不到这种——它没有任何等待
   #     可拉长）。这条轴真的抓到过东西：它让 TunnelManager.connect 里 stderr drain
   #     线程的竞态（读已被清空的 self.process）从"偶尔"变成"每次"。
+  #   - 有竞争者在真的烧 CPU。前两条轴说的都是"线程晚了"，而且都只花墙钟——睡觉的线程
+  #     不占谁的 CPU，所以"这段预算里该算完多少活"这一类断言能从它们下面走过去。第三条
+  #     轴让工作线程每次 sleep/wait 都真的占一段 CPU，于是在 GIL 上留下真的竞争；代价与
+  #     工作线程自己的活动成比例，而不是一条 24/7 的 hog 线程——那种做法实测把 pytest 的
+  #     收集阶段从 0.43s 变成 30s，记录留在 tests/_injection.py 里。
   #
   # 只跑一个 OS/版本：它的价值在确定性，不在覆盖率——版本/平台差异由上面的矩阵负责。
   timing:
-    name: Timing guard (injected thread latency)
+    name: Timing guard (injected latency, CPU share)
     runs-on: ubuntu-latest
 
     steps:
@@ -88,17 +93,23 @@ jobs:
       - name: Install project with dev deps
         run: python -m pip install --upgrade pip && pip install -e ".[dev]"
 
-      # 故意不加 `-q`：报告头里的那两行横幅就是"守卫真的开着"的证据，下面 grep 它们。
+      # 故意不加 `-q`：报告头里的三行横幅就是"守卫真的开着"的证据，下面 grep 它们。
       # 一条静默失效的注入会把这条腿变成空跑，而输出仍然是绿的。
-      - name: Run tests with injected latency
+      #
+      # CPU 轴取 0.05s：默认的 GIL 切换间隔是 5ms，所以每次烧 CPU 都确定会被抢占至少
+      # 一次——"竞争"是真的发生，而不只是一段更长的计算。开销与工作线程自己的活动成
+      # 比例：套件里能被它触及的 worker sleep/wait 只有约 40 次，于是总共约 2 秒。
+      - name: Run tests with injected latency and CPU contention
         env:
           PONTE_TEST_THREAD_DELAY: "0.15"
           PONTE_TEST_THREAD_START_DELAY: "0.15"
+          PONTE_TEST_THREAD_CPU: "0.05"
         run: |
           set -o pipefail
           pytest 2>&1 | tee slow-runner.log
           grep -q "thread latency: +0.15s" slow-runner.log
           grep -q "thread start delay: +0.15s" slow-runner.log
+          grep -q "worker CPU share: +0.05s" slow-runner.log
 
   # 一条腿覆盖三个"环境"维度（不是平台维度，平台由上面的矩阵负责）。合成一条是因为
   # 它们都是同一件事的不同侧面：**这台机器不是开发者那台**。出错时 traceback 会指出

CONTRIBUTING.md

@@ -37,16 +37,22 @@ Python 3.11–3.14, a job that re-runs it under a C locale / half-hour timezone
 with deprecations as errors, and a `build` job that installs the wheel and runs
 `ponte init`. Coverage is uploaded to Codecov.
 
-One more job re-runs the suite with two independent injections (`tests/_injection.py`):
-`PONTE_TEST_THREAD_DELAY=0.15` stretches every worker-thread sleep and wait, and
+One more job re-runs the suite with three independent injections (`tests/_injection.py`):
+`PONTE_TEST_THREAD_DELAY=0.15` stretches every worker-thread sleep and wait,
 `PONTE_TEST_THREAD_START_DELAY=0.15` holds a freshly started thread before it runs
-its first line. The second one matters because it is unreachable by the first: a
-thread that has not started executing yet has no waits to stretch. Either way, a
-test that only passes because the machine is fast fails there **every time**
-instead of flaking once in a while. Run it the same way before blaming a runner:
+its first line, and `PONTE_TEST_THREAD_CPU=0.05` makes every worker sleep/wait also
+burn real CPU. Each axis is unreachable by the others, which is why there are three
+rather than one: a thread that has not started executing yet has no waits to
+stretch, and a *sleeping* thread takes nobody's CPU — so an assertion about how
+much work fits in a fixed budget walks straight past the first two and only fails
+under the third. Either way, a test that only passes because the machine is fast
+fails there **every time** instead of flaking once in a while. Run it the same way
+before blaming a runner:
 
 ```bash
-PONTE_TEST_THREAD_DELAY=0.15 PONTE_TEST_THREAD_START_DELAY=0.15 pytest
+PONTE_TEST_THREAD_DELAY=0.15 \
+  PONTE_TEST_THREAD_START_DELAY=0.15 \
+  PONTE_TEST_THREAD_CPU=0.05 pytest
 ```
 
 If you add an assertion that waits for something a background thread produces, wait
@@ -169,15 +175,20 @@ Linux 上跑 lint + 类型检查，在 Windows / Linux / macOS × Python 3.11–
 上跑同一套测试，另有一个任务在 C locale / 半时区偏移下、并把弃用告警当错误地跑一遍，
 以及 `build` 任务会安装 wheel 并执行 `ponte init`。覆盖率上报到 Codecov。
 
-还有一个任务会用两个互相独立的注入（见 `tests/_injection.py`）再跑一遍：
+还有一个任务会用三个互相独立的注入（见 `tests/_injection.py`）再跑一遍：
 `PONTE_TEST_THREAD_DELAY=0.15` 把工作线程的每次 sleep/wait 拉长；
-`PONTE_TEST_THREAD_START_DELAY=0.15` 让刚 `start()` 的线程迟迟跑不到第一行。
-后者是前者够不到的：还没开始执行的线程没有任何等待可以被拉长——而它在空闲机器上
-永远通过、在忙机器上随机失败。两者共同的效果是：“只有机器够快才通过”的测试会
-**每次都**在那里失败，而不是偶发地红一次。怀疑是 runner 抽风之前，先这样在本地跑一遍：
+`PONTE_TEST_THREAD_START_DELAY=0.15` 让刚 `start()` 的线程迟迟跑不到第一行；
+`PONTE_TEST_THREAD_CPU=0.05` 让工作线程每次 sleep/wait 额外真的烧一段 CPU。
+为什么是三条而不是一条：每条都够不到另两条——还没开始执行的线程没有任何等待可以被
+拉长；而**正在睡觉**的线程不占任何人的 CPU，所以“这段预算里该算完多少活”这类断言
+从这两条下面直接走过去，只在第三条下才失败。它们共同的效果是：“只有机器够快才通过”
+的测试会**每次都**在那里失败，而不是偶发地红一次。怀疑是 runner 抽风之前，先这样在
+本地跑一遍：
 
 ```bash
-PONTE_TEST_THREAD_DELAY=0.15 PONTE_TEST_THREAD_START_DELAY=0.15 pytest
+PONTE_TEST_THREAD_DELAY=0.15 \
+  PONTE_TEST_THREAD_START_DELAY=0.15 \
+  PONTE_TEST_THREAD_CPU=0.05 pytest
 ```
 
 如果你要写“等后台线程产出某件东西”的断言，请等截止时间（见 `tests/_waits.py`），

README.md

@@ -326,12 +326,14 @@ python _smoke_test.py                          # zero-dependency quick check
 CI runs lint + types on Linux, and the test suite across
 Windows/Linux/macOS × Python 3.11–3.14, reporting coverage to
 [Codecov](https://codecov.io/gh/modusensus/ponte). One extra job re-runs the
-suite with `PONTE_TEST_THREAD_DELAY=0.15` **and**
-`PONTE_TEST_THREAD_START_DELAY=0.15`: the first stretches every worker-thread
-sleep and wait, the second holds a freshly started thread before its first line
-runs (a case no amount of wait-stretching can reach). A test that only passes on
-a fast machine fails there *every* time instead of flaking once in a while. Set
-the same variables locally to reproduce such a machine. Another job runs the
+suite with three injections — `PONTE_TEST_THREAD_DELAY=0.15`,
+`PONTE_TEST_THREAD_START_DELAY=0.15` and `PONTE_TEST_THREAD_CPU=0.05`: the first
+stretches every worker-thread sleep and wait, the second holds a freshly started
+thread before its first line runs (a case no amount of wait-stretching can
+reach), and the third makes those same waits burn real CPU, so a test that
+measures how much work fits in a budget is starved too. A test that only passes
+on a fast machine fails there *every* time instead of flaking once in a while.
+Set the same variables locally to reproduce such a machine. Another job runs the
 suite under a C locale (non-UTF-8 stdio) in a half-hour timezone with
 deprecation warnings as errors, and a `build` job installs the built wheel and
 runs `ponte init`, so a packaging regression cannot ship again. See
@@ -634,11 +636,13 @@ python _smoke_test.py                          # 零依赖快速自检
 
 CI 在 Linux 上跑 lint + 类型检查，在 Windows/Linux/macOS × Python
 3.11–3.14 上跑测试，覆盖率上报到
-[Codecov](https://codecov.io/gh/modusensus/ponte)。另有一个任务会同时用
-`PONTE_TEST_THREAD_DELAY=0.15` 与 `PONTE_TEST_THREAD_START_DELAY=0.15` 再跑
-一遍：前者把工作线程的每次 sleep/wait 拉长，后者让刚 start() 的线程迟迟跑不到
-第一行（后半种情况没有任何等待可以被拉长）。于是"只有机器够快才通过"的测试会
-**每次都**在那里失败，而不是偶发地红一次。本地设同样两个变量即可复现这种机器。
+[Codecov](https://codecov.io/gh/modusensus/ponte)。另有一个任务会用三个注入再跑
+一遍：`PONTE_TEST_THREAD_DELAY=0.15`、`PONTE_TEST_THREAD_START_DELAY=0.15`
+与 `PONTE_TEST_THREAD_CPU=0.05`。第一条把工作线程的每次 sleep/wait 拉长；第二条让刚
+start() 的线程迟迟跑不到第一行（没有任何等待可以被拉长的那种情况）；第三条让这些
+等待额外真的烧一段 CPU，于是"这段预算里该算完多少活"的断言也会被饿着。于是"只有机器够快
+才通过"的测试会**每次都**在那里失败，而不是偶发地红一次。本地设同样三个变量即可
+复现这种机器。
 还有一个任务在 C locale（非 UTF-8 的 stdio）、半时区偏移下跑，并把弃用告警当错误；
 `build` 任务会安装打好的 wheel 并执行 `ponte init`，避免打包问题再次溜进发布。详见
 [CONTRIBUTING.md](CONTRIBUTING.md)。

tests/_injection.py

@@ -5,7 +5,7 @@
 够快"**。快机器上它永远绿，慢机器上它偶尔红，两种结果都不回答"被测代码对不对"。
 
 于是这里做的事是：在需要的时候，把"慢"变成一个**开关**，让那类测试当场失败。它属于
-测试基础设施而不是产品代码，但它也是唯一能被"证明"的部分——两条轴都有对应的自检
+测试基础设施而不是产品代码，但它也是唯一能被"证明"的部分——三条轴都有对应的自检
 （``tests/test_timing_guard.py``），因为一条静默失效的守卫比没有守卫更糟。
 
 用法：由 ``tests/conftest.py`` 在会话开始时按环境变量装上（见 :func:`active`）。
@@ -22,13 +22,17 @@
 THREAD_DELAY_ENV = "PONTE_TEST_THREAD_DELAY"
 #: Seconds a newly started thread waits *before running its body*.
 START_DELAY_ENV = "PONTE_TEST_THREAD_START_DELAY"
+#: Seconds of *busy* work (GIL held) added to every worker sleep/wait.
+THREAD_CPU_ENV = "PONTE_TEST_THREAD_CPU"
 
 # The unpatched primitives, captured at import time: without these the injection
 # would slow itself (and a thread start delay would be stretched by the wait
-# delay, so the two axes would stop being independent).
+# delay, so the axes would stop being independent). ``perf_counter`` is captured
+# for the same reason — the burn must measure its own deadline, not a patched one.
 _ORIGINAL_SLEEP = time.sleep
 _ORIGINAL_WAIT = threading.Event.wait
 _ORIGINAL_THREAD_START = threading.Thread.start
+_ORIGINAL_PERF_COUNTER = time.perf_counter
 
 
 def _number_env(name: str) -> float:
@@ -49,15 +53,31 @@ def _number_env(name: str) -> float:
     return value
 
 
+def _burn(seconds: float) -> None:
+    """Hold the GIL for *seconds* of real work — the thing a sleeping thread never does.
+
+    A busy loop rather than another ``sleep``: the point of this axis is that the
+    worker *uses* the CPU, so it competes for the GIL with whatever else is
+    running. Sleeping releases the GIL and therefore takes nobody's CPU, which is
+    exactly why the delay axis cannot emulate this.
+    """
+    if seconds <= 0:
+        return
+    deadline = _ORIGINAL_PERF_COUNTER() + seconds
+    while _ORIGINAL_PERF_COUNTER() < deadline:
+        pass
+
+
 @dataclass(frozen=True)
 class Injection:
     """The load this process is asked to emulate. Zero means "off"."""
 
     wait_delay: float = 0.0
     start_delay: float = 0.0
+    thread_cpu: float = 0.0
 
     def enabled(self) -> bool:
-        return bool(self.wait_delay or self.start_delay)
+        return bool(self.wait_delay or self.start_delay or self.thread_cpu)
 
     def banner(self) -> str:
         """One line per enabled axis, so a green run can prove the guard was on."""
@@ -66,6 +86,10 @@ def banner(self) -> str:
             lines.append(
                 f"thread latency: +{self.wait_delay:g}s injected into every worker sleep/wait"
             )
+        if self.thread_cpu:
+            lines.append(
+                f"worker CPU share: +{self.thread_cpu:g}s of busy work per worker sleep/wait"
+            )
         if self.start_delay:
             lines.append(
                 f"thread start delay: +{self.start_delay:g}s before every thread body runs"
@@ -84,7 +108,11 @@ def active() -> Injection:
     """
     global _INSTALLED
     if _INSTALLED is None:
-        _INSTALLED = Injection(_number_env(THREAD_DELAY_ENV), _number_env(START_DELAY_ENV))
+        _INSTALLED = Injection(
+            _number_env(THREAD_DELAY_ENV),
+            _number_env(START_DELAY_ENV),
+            _number_env(THREAD_CPU_ENV),
+        )
         if _INSTALLED.enabled():
             _apply(_INSTALLED)
     return _INSTALLED
@@ -98,7 +126,7 @@ def pytest_configure(config) -> None:  # noqa: ARG001 - 只为让 `-p _injection
 def _apply(injection: Injection) -> None:
     """Make the code under test slower than the test that watches it.
 
-    A slow machine fails timing-dependent tests in two distinct ways, and each
+    A slow machine fails timing-dependent tests in three distinct ways, and each
     one needs its own instrument:
 
     **1. the worker's own waits are stretched** (``wait_delay``). "Slow machine"
@@ -120,29 +148,40 @@ def _apply(injection: Injection) -> None:
     race in ``TunnelManager.connect`` (the drain thread read ``self.process``,
     which the session's ``finally`` had already cleared).
 
-    **...and deliberately no third axis for "the worker got no CPU at all".**
-    That was tried and measured — a busy Python thread competing for the GIL,
-    plus a duty-cycled variant — and it is both worse at finding bugs and far
-    more expensive. On this suite, one such thread turned pytest's *collection*
-    phase from 0.43s into 30s (two threads: 68s): GIL contention adds up to
-    ``sys.getswitchinterval()`` of latency to every reacquisition, and a suite
-    performs thousands of those. And it did not fail the very test this whole
-    mechanism exists for — a counter-scheduled worker still runs inside a 50ms
-    nap, because that nap releases the GIL. What a hog adds on top of the two
-    axes above is wall-clock cost, not coverage. The genuine "no spare CPU"
-    failure — a worker that never reaches its first bytecode in time — is axis 2.
+    **3. the worker runs, but it is not alone on the CPU** (``thread_cpu``).
+    The first two axes are both "a worker that is *late*", and both of them cost
+    nothing but wall clock — which is the point, and also the limit: a thread
+    that is merely asleep takes nobody's CPU, so a test that measures **how much
+    work fits in a fixed budget** (rather than "did this happen yet") sails
+    through both. Real contention looks different: something else is holding the
+    CPU, so whoever else wants it waits for the GIL. ``thread_cpu`` adds real
+    busy work *inside the worker's own sleep/wait*, which means two things at
+    once — the worker's iteration takes longer, and while it burns, every other
+    thread waits up to ``sys.getswitchinterval()`` to get the GIL back.
+
+    Cost is what made this axis worth designing carefully. The obvious
+    implementation — a free-running busy thread for the whole process — was tried
+    and measured, and it is unusable: one such thread turned pytest's *collection*
+    phase from 0.43s into 30s (two threads: 68s), because collection performs
+    thousands of tiny GIL reacquisitions and each one can cost up to
+    ``sys.getswitchinterval()``. Worse, it did not even fail the tests this
+    mechanism exists for, since their workers sleep (releasing the GIL). Burning
+    *proportionally to the worker's own activity* fixes both halves: the cost is
+    bounded by the number of worker sleep/waits rather than by wall time (nothing
+    burns during collection, or in the test body, or while injection is off), and
+    the contention lands exactly on the threads that are supposedly competing.
+
+    Scope, honestly: axis 3 emulates *another thread in this process* holding the
+    CPU, which is what GIL contention is. It cannot express "this container got a
+    fraction of a core" for a worker that never sleeps or waits — there is no
+    Python-level hook for "every bytecode of that thread" that would not itself
+    distort the measurement — nor OS-level effects such as a stalled filesystem
+    or a cold CPU. Don't read a green run here as "no timing assumptions left".
 
     Two primitives cover how a background thread paces itself: ``Event.wait``
     and ``time.sleep``. Clocks (``time.monotonic``) are deliberately left alone —
     moving those would corrupt deadlines instead of emulating load. A ``sleep(0)``
     is a yield rather than a wait, and load does not stretch it, so it stays.
-
-    Scope, honestly: both axes emulate *a worker that is late*, which is the
-    failure mode behind every flake this exists for. What they cannot express is a
-    main-thread budget measured in work rather than in time — "this loop should
-    have run 10_000 times in 0.2s", which is the one thing the rejected hog above
-    *would* have caught, and only by an unpredictable factor. Don't read a green
-    run here as "no timing assumptions left".
     """
     if injection.start_delay:
         # Instance-level ``run`` override: the delay happens *inside* the new
@@ -160,17 +199,19 @@ def delayed_run() -> None:
 
         threading.Thread.start = slow_start  # type: ignore[method-assign]
 
-    if injection.wait_delay:
+    if injection.wait_delay or injection.thread_cpu:
         main_thread = threading.main_thread()
 
         def slow_sleep(seconds: float) -> None:
             if seconds > 0 and threading.current_thread() is not main_thread:
                 seconds += injection.wait_delay
+                _burn(injection.thread_cpu)
             _ORIGINAL_SLEEP(seconds)
 
         def slow_wait(event: threading.Event, timeout: float | None = None) -> bool:
             if timeout is not None and threading.current_thread() is not main_thread:
                 timeout += injection.wait_delay
+                _burn(injection.thread_cpu)
             return _ORIGINAL_WAIT(event, timeout)
 
         time.sleep = slow_sleep