test(data_platform): CI 三连挂根因修——CLOSE_WAIT 模拟线程泄漏×全局 time.sleep 补丁相遇(#1276/#1290/#1291 三案同根)——①泄漏源 test_with_timeout_breaks_close_wait_silent_hang: 模拟体 while True:sleep(0.1) 无退出条件,_with_timeout 超时弃线程是生产文档化设计(daemon+relogin 兜底)但测试把永生线程留在进程里,改 threading.Event stop 可终止+收尾断言 active_count 归零(泄漏回归钉死) ②地雷 test_corpus_download 两处 setattr(cd.time,'sleep')=补 stdlib time 本体(全局): 泄漏线程 sleep 被换零延迟 append→单核忙循环百万次/秒塞列表,pytest 渲染巨列表分钟级=CI 楔死;改补 cd 模块属性(scoped SimpleNamespace,sleep/monotonic/time 面透真) 定证:本地 8 连发第 5 发复现(320s rc=1,faulthandler dump 双线程现行+assert [0.1,0.1,0....] 巨表);flaky=断言窗口与泄漏线程 0.1s 醒转赛跑(轻载绿/重载红,17:22 #1289 绿→17:27 负载升三连挂);修复后定向 88 绿+同进程活永生线程污染对抗 3 绿+全套 6 连发全绿 26s 级;同型地雷存量 8 处(bs_eod_resilience 5+ak_sw_industry 3)触发前提已灭记 backlog 集中加固 [nas] [no-doc]

tests/data_platform/test_bs_eod_resilience.py

@@ -109,17 +109,38 @@ def test_with_timeout_breaks_close_wait_silent_hang():
       - socket.setdefaulttimeout(30) 救不了 (客户端不死在 socket 层)
       - 被动 relogin (只在异常时触发) 永远不触发
     唯一能打破静默 hang 的就是外层 _with_timeout. 此处显式模拟"无异常纯阻塞"调用.
+
+    2026-09-19 CI 三连挂根因修: 旧版 `while True: time.sleep(0.1)` 模拟体无退出
+    条件——_with_timeout 超时弃线程是生产设计(daemon+relogin 兜底), 但测试把这条
+    永生线程留在进程里; 下游 test_corpus_download 若全局补丁 time.sleep, 该线程
+    的 sleep 被换成零延迟 append → 单核忙循环+巨列表, pytest 渲染卡分钟级=CI 楔死
+    (#1290/#1291/重跑三连 16% 段挂, 本地 8 连发第 5 发复现 320s)。改 stop 事件可
+    终止+收尾断言线程归零, 超时语义不变(外层仍在 1s 打破)。
     """
+    import threading
+
+    stop = threading.Event()
+
     def close_wait_silent_block():
         # 模拟 baostock 客户端在 CLOSE_WAIT 上死循环: 永不返回, 也不抛异常
-        while True:
-            time.sleep(0.1)
+        while not stop.is_set():
+            time.sleep(0.01)
 
-    t0 = time.time()
-    with pytest.raises(TimeoutError):
-        bs_eod._with_timeout(close_wait_silent_block, timeout=1)
-    elapsed = time.time() - t0
-    assert elapsed < 3, f"CLOSE_WAIT 静默 hang 未被外层超时打破: elapsed={elapsed:.1f}s"
+    n_before = threading.active_count()
+    try:
+        t0 = time.time()
+        with pytest.raises(TimeoutError):
+            bs_eod._with_timeout(close_wait_silent_block, timeout=1)
+        elapsed = time.time() - t0
+        assert elapsed < 3, f"CLOSE_WAIT 静默 hang 未被外层超时打破: elapsed={elapsed:.1f}s"
+    finally:
+        stop.set()
+        for _ in range(200):  # 最多等 2s, 泄漏线程应感知 stop 退出
+            if threading.active_count() <= n_before:
+                break
+            time.sleep(0.01)
+        assert threading.active_count() <= n_before, \
+            "模拟 CLOSE_WAIT 线程未随 stop 事件退出(测试线程泄漏=CI 楔死回归)"
 
 
 # ---------- fetch_k_with_timeout / fetch_all_stocks_with_timeout ----------

tests/data_platform/test_corpus_download.py

@@ -8,6 +8,8 @@ import datetime as dt
 import fcntl
 import json
 import sys
+import time
+import types
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -15,6 +17,23 @@ import pytest
 from scripts.data_platform import corpus_download as cd
 
 
+def _patch_cd_sleep_record(monkeypatch, slept):
+    """cd 域内 time.sleep 替换为记录器(不动全局 time 模块).
+
+    2026-09-19 CI 三连挂根因修: 旧写法 `monkeypatch.setattr(cd.time, "sleep", ...)`
+    补丁的是 stdlib time 模块本体(cd.time 就是全局 time)——与 test_bs_eod_resilience
+    泄漏的 CLOSE_WAIT 模拟线程相遇时, 其 sleep(0.1) 被换成零延迟 append → 单核忙
+    循环+巨列表, pytest 渲染分钟级=CI 楔死(#1290/#1291/rerun 三连 16% 段)。改补
+    cd 模块属性: 只影响 corpus_download 自身, 外部线程的 sleep 不受波及。
+    cd 用到的 time 面=sleep/monotonic/time, 其余透传真实现。
+    """
+    monkeypatch.setattr(cd, "time", types.SimpleNamespace(
+        sleep=lambda s: slept.append(s),
+        monotonic=time.monotonic,
+        time=time.time,
+    ))
+
+
 # ---------- Fixtures ----------
 
 @pytest.fixture(autouse=True)
@@ -572,7 +591,7 @@ def test_ann_tail_retry_skipped_when_no_pending(root, monkeypatch):
     """主 pass 全 done → 零重试开销(不等待不重走)。"""
     _wire_daily(monkeypatch, root)
     slept = []
-    monkeypatch.setattr(cd.time, "sleep", lambda s: slept.append(s))
+    _patch_cd_sleep_record(monkeypatch, slept)
     assert cd.run_lane("daily") == 0
     assert slept == []
 
@@ -588,7 +607,7 @@ def test_ann_tail_retry_skipped_when_past_deadline(root, monkeypatch):
 
     monkeypatch.setattr(cd, "fetch_cninfo_announcements", always_504)
     slept = []
-    monkeypatch.setattr(cd.time, "sleep", lambda s: slept.append(s))
+    _patch_cd_sleep_record(monkeypatch, slept)
     rc = cd.run_lane("daily", until="00:00")   # deadline=今天 00:00 已过
     assert rc == 1
     assert slept == []                          # 不等待