mirror of
https://github.com/shy3130/tick-stock-panel.git
synced 2026-11-02 10:26:13 +00:00
shy3130
eefd9521bb
用户反馈(升级后数据页分钟K同步无限停在「等待其他计算任务完成…」): 被取消/
已终止的残留线程挂死在单次 SDK 网络读里(SDK timeout=30s×3 只保护其覆盖的
阶段), 协作式取消要到分块边界才生效 → 它持有的独占槽(capacity=2, exclusive
权重=2)永不释放; 排队任务等槽无超时且 status=pending 不在 reap_stale 视野
(旧实现只回收 running), 无限空转只能重启进程。机制为 0.3.1 统一重任务并发
控制(b16957a8)引入, 非 0.3.2 回归。
- pipeline_jobs.run_with_capacity: 槽位等待加 timeout(取 job 自身 timeout_s,
与停滞阈值同源、用户可配), 超时以可读错误失败并按已取消上报 — fn 未执行,
零数据副作用, 四个调用方(kline/pipeline/daily/integrity)的异常分支免改
- pipeline_jobs.reap_stale: pending 排队停滞纳入回收(pending 无 started_at,
总时长硬上限跳过); create() 锚定 last_progress_at, executor 排队中从未
上报进度的任务也有兜底
- kline_sync: SDK 批量调用(日K区间/日K条数/复权因子/分钟K区间/分钟K条数共
5 处)统一 _sdk_call_with_deadline 守护(600s, daemon 线程执行 + join 超时
放弃); 超时走既有分块失败路径, 且补发 on_chunk_done 心跳保留进度上报与
协作取消检查点 — 持有侧线程可退出并释放槽位, 被放弃线程为 daemon 不阻
塞进程退出
测试: 新增排队超时可读失败/槽空闲行为不变/pending 回收与不误杀/硬上限跳过/
SDK deadline 单元与分块循环心跳(分钟+日K); test_queued_pipeline_is_
cancellable_and_not_reaped 按新契约改写为 cancellable_and_reaped_when_stalled
(其旧断言「排队不被 reap」即本 bug 的排队侧契约)。本地与 TZ=UTC 双跑全量
2470 passed 6 skipped
98 lines
3.6 KiB
Python
98 lines
3.6 KiB
Python
"""回归测试: SDK 批量调用守护 deadline(「分钟K同步卡死空转」持有侧修复)。
|
||
|
||
TickFlow SDK 自带 timeout=30s×3 重试, 但只保护其覆盖的请求阶段; 真挂死在
|
||
不可中断的网络读里时调用永不返回, 持有独占槽的重任务线程被连坐卡死, 后续
|
||
重任务全部无限排队。deadline 包装在超时后放弃等待, 调用方按分块失败路径
|
||
继续, 心跳与协作取消在分块边界照常生效。纯逻辑, 不触网。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import functools
|
||
import time
|
||
from datetime import datetime
|
||
|
||
import pytest
|
||
|
||
from app.services import kline_sync
|
||
from app.services.kline_sync import SDKCallTimeoutError, _sdk_call_with_deadline
|
||
|
||
|
||
# ── 包装器单元行为 ──────────────────────────────────────────────────────
|
||
|
||
def test_result_passthrough():
|
||
assert _sdk_call_with_deadline(lambda: 42) == 42
|
||
|
||
|
||
def test_exception_propagates_unchanged():
|
||
def _boom():
|
||
raise ValueError("sdk error")
|
||
|
||
with pytest.raises(ValueError, match="sdk error"):
|
||
_sdk_call_with_deadline(_boom)
|
||
|
||
|
||
def test_deadline_exceeded_abandons_wait():
|
||
started = time.monotonic()
|
||
with pytest.raises(SDKCallTimeoutError):
|
||
_sdk_call_with_deadline(lambda: time.sleep(3), timeout_s=0.1)
|
||
assert time.monotonic() - started < 2 # 不陪着挂死的调用一起等
|
||
|
||
|
||
# ── 分块循环集成: 超时走失败路径且心跳不断 ─────────────────────────────
|
||
|
||
class _HangingKlines:
|
||
@staticmethod
|
||
def batch(*args, **kwargs):
|
||
time.sleep(3) # 模拟挂死; worker 是 daemon 线程, 不拖慢进程退出
|
||
|
||
|
||
class _HangingClient:
|
||
klines = _HangingKlines
|
||
|
||
|
||
def _fast_deadline(monkeypatch):
|
||
"""循环内 deadline 默认 600s, 换成 0.1s 版本方便测试。"""
|
||
real = kline_sync._sdk_call_with_deadline
|
||
monkeypatch.setattr(
|
||
kline_sync, "_sdk_call_with_deadline",
|
||
functools.partial(real, timeout_s=0.1),
|
||
)
|
||
|
||
|
||
def test_minute_loop_timeout_keeps_heartbeat(monkeypatch):
|
||
"""分钟分块超时 → 本批放弃并继续, on_chunk_done 心跳仍被调用
|
||
(保留进度上报与协作取消的检查点, 防僵尸线程滞留独占槽)。"""
|
||
_fast_deadline(monkeypatch)
|
||
monkeypatch.setattr(kline_sync, "_try_custom_minute", lambda *a, **k: (None, True))
|
||
monkeypatch.setattr(kline_sync, "get_client", lambda: _HangingClient())
|
||
|
||
beats: list[tuple[int, int, str]] = []
|
||
df = kline_sync.sync_minute_batch(
|
||
["600000.SH", "000001.SZ"],
|
||
start_time=datetime(2026, 8, 3), end_time=datetime(2026, 8, 7),
|
||
batch_size=1, rpm=None,
|
||
on_chunk_done=lambda cur, total, seg: beats.append((cur, total, seg)),
|
||
)
|
||
|
||
assert df.is_empty()
|
||
assert [b[0] for b in beats] == [1, 2] # 每个超时分块都推进心跳
|
||
assert beats[0][1] == 2
|
||
|
||
|
||
def test_daily_loop_timeout_marks_failed(monkeypatch):
|
||
"""日K分块超时 → failed_out 收编本批(上层可判「部分失败」), 心跳推进。"""
|
||
_fast_deadline(monkeypatch)
|
||
monkeypatch.setattr(kline_sync, "get_client", lambda: _HangingClient())
|
||
|
||
beats: list[tuple[int, int]] = []
|
||
failed: list[str] = []
|
||
df = kline_sync.sync_daily_batch(
|
||
["600000.SH", "000001.SZ"], count=250, batch_size=1, rpm=None,
|
||
on_chunk_done=lambda cur, total: beats.append((cur, total)),
|
||
failed_out=failed,
|
||
)
|
||
|
||
assert df.is_empty()
|
||
assert sorted(failed) == ["000001.SZ", "600000.SH"]
|
||
assert [b[0] for b in beats] == [1, 2]
|