fix: detect stale pipeline jobs by progress

backend/app/api/pipeline.py

@@ -24,7 +24,7 @@ async def run_now(request: Request) -> dict:
     """异步触发盘后管道,立即返回 job_id。客户端轮询 /jobs/{id} 拿进度。
 
     若已有任务在跑,**返回该任务 id 而不是开新任务**(防止并发拉数据撞限流)。
-    但如果该任务已运行超过 10 分钟 (可能因 reload 卡死), 强制标记为失败后重新创建。
+    但如果该任务连续超过无进度阈值 (可能因 reload 卡死), 强制标记为失败。
     """
     repo = request.app.state.repo
     capset = request.app.state.capabilities
@@ -77,7 +77,7 @@ def _run() -> dict:
 
 @router.get("/jobs/{job_id}")
 def get_job(job_id: str) -> dict:
-    # 每次轮询都检查卡死 job — 前端每秒轮询,STALE_JOB_TIMEOUT_S(10min)后必定自愈,
+    # 每次轮询都检查卡死 job — 连续超过任务自身的无进度阈值后自动回收,
     # 无需用户再次手动点「同步」。
     job_store.reap_stale()
     j = job_store.get(job_id)

backend/app/services/pipeline_jobs.py

@@ -15,19 +15,19 @@
 import os
 import threading
 import uuid
-from datetime import datetime
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any, Literal
 
 logger = logging.getLogger(__name__)
 
 JobStatus = Literal["pending", "running", "succeeded", "failed"]
 
-# 运行超过此秒数视为卡死(reload 后孤儿 task / 网络读无限阻塞等)。
+# 连续超过此秒数没有进度视为卡死(reload 后孤儿 task / 网络读无限阻塞等)。
 # 由 reap_stale() 在 /run 和 /jobs/{id} 轮询端点检查 — 保证卡死后能自愈,
 # 无需用户再次点击「同步」。
 #
-# 超时阈值按任务类型区分:
+# 无进度超时阈值按任务类型区分:
 #   - 普通任务(日K管道/扩展/修正/重算): 1200s (20 分钟)
 #   - 长任务(分钟K全市场同步,数据量是日K的 ~240 倍): 1800s (30 分钟)
 # 分钟K即使流式落盘后仍可能跑十几到数十分钟(限速 sleep 是主因),
@@ -38,6 +38,10 @@
 STALE_JOB_TIMEOUT_S = DEFAULT_JOB_TIMEOUT_S
 
 
+def _utc_now_iso() -> str:
+    return datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
 def _default_store_dir() -> Path:
     from app.config import settings
     return settings.data_dir / "job_store"
@@ -115,7 +119,7 @@ def create(self, timeout_s: int = DEFAULT_JOB_TIMEOUT_S) -> tuple[str, bool]:
 
         is_new=False 表示复用了已有活跃任务,调用方**不得**再调度新的后台任务。
 
-        timeout_s: reap_stale 判定卡死的阈值。普通任务默认 1200s;
+        timeout_s: reap_stale 判定无进度卡死的阈值。普通任务默认 1200s;
             分钟K全市场同步等长任务传 LONG_JOB_TIMEOUT_S (1800s)。
         """
         with self._lock:
@@ -133,6 +137,7 @@ def create(self, timeout_s: int = DEFAULT_JOB_TIMEOUT_S) -> tuple[str, bool]:
                 "stage_pct": 0,
                 "log": [],
                 "started_at": None,
+                "last_progress_at": None,
                 "finished_at": None,
                 "duration_s": None,
                 "result": None,
@@ -147,16 +152,18 @@ def start(self, job_id: str) -> None:
             j = self._active_jobs.get(job_id)
             if not j:
                 return
+            now = _utc_now_iso()
             j["status"] = "running"
-            j["started_at"] = datetime.utcnow().isoformat(timespec="seconds") + "Z"
+            j["started_at"] = now
+            j["last_progress_at"] = now
 
     def succeed(self, job_id: str, result: Any) -> None:
         with self._lock:
             j = self._active_jobs.pop(job_id, None)
             if not j:
                 return
             j["status"] = "succeeded"
-            j["finished_at"] = datetime.utcnow().isoformat(timespec="seconds") + "Z"
+            j["finished_at"] = _utc_now_iso()
             j["progress"] = 100
             j["result"] = result
             j["duration_s"] = _duration_s(j)
@@ -167,17 +174,20 @@ def succeed(self, job_id: str, result: Any) -> None:
 
     def fail(self, job_id: str, error: str) -> None:
         with self._lock:
-            j = self._active_jobs.pop(job_id, None)
-            if not j:
-                return
-            j["status"] = "failed"
-            j["finished_at"] = datetime.utcnow().isoformat(timespec="seconds") + "Z"
-            j["error"] = error
-            j["duration_s"] = _duration_s(j)
-            if self._active_id == job_id:
-                self._active_id = None
-            self._delete_oldest()
-            self._write_file(j)
+            self._fail_locked(job_id, error)
+
+    def _fail_locked(self, job_id: str, error: str) -> None:
+        j = self._active_jobs.pop(job_id, None)
+        if not j:
+            return
+        j["status"] = "failed"
+        j["finished_at"] = _utc_now_iso()
+        j["error"] = error
+        j["duration_s"] = _duration_s(j)
+        if self._active_id == job_id:
+            self._active_id = None
+        self._delete_oldest()
+        self._write_file(j)
 
     # ===== progress =====
 
@@ -189,12 +199,13 @@ def progress(self, job_id: str, stage: str, pct: int, msg: str,
                 return
             j["stage"] = stage
             j["progress"] = max(0, min(100, int(pct)))
+            j["last_progress_at"] = _utc_now_iso()
             if stage_pct is not None:
                 j["stage_pct"] = max(0, min(100, int(stage_pct)))
             elif j["stage"] != stage:
                 j["stage_pct"] = 0
             entry = {
-                "ts": datetime.utcnow().isoformat(timespec="seconds") + "Z",
+                "ts": _utc_now_iso(),
                 "stage": stage,
                 "msg": msg,
             }
@@ -238,7 +249,7 @@ def active_id(self) -> str | None:
         return self._active_id
 
     def reap_stale(self, timeout_s: int | None = None) -> None:
-        """回收运行超过阈值(卡死)的 running job(标记为 failed)。
+        """回收连续无进度超过阈值的 running job(标记为 failed)。
 
         在 /run 和 /jobs/{id} 轮询端点都会调用 — 保证卡死后任意轮询都能自愈,
         无需用户再次手动触发同步。reload 后的孤儿 task(内存里已无 job 记录)
@@ -255,29 +266,26 @@ def reap_stale(self, timeout_s: int | None = None) -> None:
             j = self._active_jobs.get(jid)
             if not j or j.get("status") != "running":
                 return
-            started = j.get("started_at")
-            if not started:
+            # 兼容升级前创建、没有 last_progress_at 字段的活跃任务。
+            last_progress_at = j.get("last_progress_at") or j.get("started_at")
+            if not last_progress_at:
                 return
             # 优先用显式传入, 其次 job 自身阈值, 最后默认值
             effective_timeout = timeout_s if timeout_s is not None else j.get("timeout_s", DEFAULT_JOB_TIMEOUT_S)
-        # 时间计算放到锁外(避免 datetime 解析持锁)。
-        # started_at 形如 "2026-07-04T12:00:00Z"(start() 用 datetime.utcnow 存)。
-        # 两端都用 timezone-aware UTC 比较,避免 naive/aware 混用导致 TypeError。
-        try:
-            start_dt = datetime.fromisoformat(started.replace("Z", "+00:00"))
-            elapsed = (datetime.now(start_dt.tzinfo) - start_dt).total_seconds()
-        except Exception:  # noqa: BLE001
-            return
-        if elapsed > effective_timeout:
-            logger.warning("reap_stale: 强制取消卡死 job %s (已运行 %.0fs, 阈值 %ss)",
-                           jid, elapsed, effective_timeout)
-            self.fail(jid, f"超时自动取消 (运行 {int(elapsed)}s, 疑似卡死)")
-            # 强制释放重任务锁: 卡死的线程无法被中断, 锁永远不会自然释放。
-            # job 已标记 failed, 即使僵尸线程后续写入 parquet, 下次拉取会覆盖, 安全。
+            # 时间形如 "2026-07-04T12:00:00Z"。解析和终态切换保持在同一把锁内,
+            # 避免工作线程恰好上报新进度时仍被轮询线程误判为卡死。
             try:
-                _heavy_run_lock.release()
-            except RuntimeError:
-                pass
+                progress_dt = datetime.fromisoformat(last_progress_at.replace("Z", "+00:00"))
+                idle_s = (datetime.now(progress_dt.tzinfo) - progress_dt).total_seconds()
+            except Exception:
+                return
+            if idle_s <= effective_timeout:
+                return
+            logger.warning("reap_stale: 强制取消卡死 job %s (无进度 %.0fs, 阈值 %ss)",
+                           jid, idle_s, effective_timeout)
+            self._fail_locked(jid, f"超时自动取消 (连续 {int(idle_s)}s 无进度, 疑似卡死)")
+        # 不释放重任务锁。执行线程无法被强制终止,必须由其 finally 自然释放;
+        # 若线程永久卡死,用户需重启进程,避免新旧任务并发写同一批数据。
 
     def clear(self) -> None:
         """清空所有任务（内存 + 磁盘文件）。"""

backend/tests/test_pipeline_and_monitor_fixes.py

@@ -4,6 +4,8 @@
 """
 from __future__ import annotations
 
+from datetime import UTC, datetime, timedelta
+
 import polars as pl
 import pytest
 
@@ -47,6 +49,65 @@ def test_create_new_after_terminal(tmp_path):
     assert new2 is True
 
 
+def _utc_iso(delta: timedelta) -> str:
+    return (datetime.now(UTC) + delta).isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
+def test_reap_stale_uses_last_progress_instead_of_total_runtime(tmp_path):
+    """任务总时长很长但仍在推进时,不得误判为卡死。"""
+    store = JobStore(store_dir=tmp_path / "jobs")
+    jid, _ = store.create(timeout_s=1200)
+    store.start(jid)
+    store._active_jobs[jid]["started_at"] = _utc_iso(-timedelta(hours=2))
+
+    store.progress(jid, "sync_adj", 53, "除权因子批次 50/139")
+    store.reap_stale()
+
+    assert store.get(jid)["status"] == "running"
+
+
+def test_reap_stale_fails_job_after_progress_timeout(tmp_path):
+    """只有连续无进度超过阈值时才回收任务。"""
+    store = JobStore(store_dir=tmp_path / "jobs")
+    jid, _ = store.create(timeout_s=1200)
+    store.start(jid)
+    store._active_jobs[jid]["last_progress_at"] = _utc_iso(-timedelta(seconds=1201))
+
+    store.reap_stale()
+
+    job = store.get(jid)
+    assert job["status"] == "failed"
+    assert "无进度, 疑似卡死" in job["error"]
+
+
+def test_reap_stale_falls_back_to_started_at_for_legacy_job(tmp_path):
+    """升级前没有 last_progress_at 的活跃任务仍可被回收。"""
+    store = JobStore(store_dir=tmp_path / "jobs")
+    jid, _ = store.create(timeout_s=1200)
+    store.start(jid)
+    del store._active_jobs[jid]["last_progress_at"]
+    store._active_jobs[jid]["started_at"] = _utc_iso(-timedelta(seconds=1201))
+
+    store.reap_stale()
+
+    assert store.get(jid)["status"] == "failed"
+
+
+def test_reap_stale_keeps_run_slot_until_worker_exits(tmp_path):
+    """标记超时不能放行新任务与仍存活的旧线程并发写数据。"""
+    store = JobStore(store_dir=tmp_path / "jobs")
+    jid, _ = store.create(timeout_s=1200)
+    store.start(jid)
+    store._active_jobs[jid]["last_progress_at"] = _utc_iso(-timedelta(seconds=1201))
+
+    assert pipeline_jobs.try_acquire_run_slot() is True
+    try:
+        store.reap_stale()
+        assert pipeline_jobs.try_acquire_run_slot() is False
+    finally:
+        pipeline_jobs.release_run_slot()
+
+
 def test_run_slot_is_exclusive():
     """重任务执行槽同一时刻只允许一个持有者(防僵尸并发)。"""
     assert pipeline_jobs.try_acquire_run_slot() is True