diff --git a/backend/app/api/pipeline.py b/backend/app/api/pipeline.py index 11f522b..f39611d 100644 --- a/backend/app/api/pipeline.py +++ b/backend/app/api/pipeline.py @@ -29,22 +29,9 @@ async def run_now(request: Request) -> dict: repo = request.app.state.repo capset = request.app.state.capabilities - # 检测卡死的 running job (如 reload 后孤儿 task) - existing_id = job_store.active_id() - if existing_id: - existing = job_store.get(existing_id) - if existing and existing["status"] == "running": - from datetime import datetime, timezone - started = existing.get("started_at") - if started: - try: - start_dt = datetime.fromisoformat(started.replace("Z", "+00:00")) - elapsed = (datetime.now(timezone.utc) - start_dt).total_seconds() - if elapsed > 600: # 超过 10 分钟视为卡死 - logger.warning("强制取消卡死 job %s (已运行 %.0fs)", existing_id, elapsed) - job_store.fail(existing_id, "超时自动取消 (疑似 reload 后孤儿 task)") - except Exception: - pass + # 检测卡死的 running job (如 reload 后孤儿 task / 网络读无限阻塞)。 + # reap_stale 会在 /run 和 /jobs/{id} 轮询端点都调用,保证卡死后能自愈。 + job_store.reap_stale() job_id = job_store.create() @@ -81,6 +68,9 @@ async def run_now(request: Request) -> dict: @router.get("/jobs/{job_id}") def get_job(job_id: str) -> dict: + # 每次轮询都检查卡死 job — 前端每秒轮询,STALE_JOB_TIMEOUT_S(10min)后必定自愈, + # 无需用户再次手动点「同步」。 + job_store.reap_stale() j = job_store.get(job_id) if not j: raise HTTPException(status_code=404, detail="job not found") diff --git a/backend/app/services/index_sync.py b/backend/app/services/index_sync.py index 42f3b1b..1f474e3 100644 --- a/backend/app/services/index_sync.py +++ b/backend/app/services/index_sync.py @@ -233,7 +233,7 @@ def sync_and_persist_index_daily( interval = (60.0 / rpm) if rpm else 0 chunks = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)] for i, chunk in enumerate(chunks): - if i > 0 and interval > 0 and len(chunks) > rpm: + if i > 0 and interval > 0: import time time.sleep(interval) raw = kline_sync.sync_daily_batch( @@ -332,7 +332,7 @@ def sync_and_persist_etf_daily( chunks = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)] factors = _load_etf_factors(repo) for i, chunk in enumerate(chunks): - if i > 0 and interval > 0 and len(chunks) > rpm: + if i > 0 and interval > 0: import time time.sleep(interval) raw = kline_sync.sync_daily_batch( diff --git a/backend/app/services/kline_sync.py b/backend/app/services/kline_sync.py index 7a4d5f5..ba9a751 100644 --- a/backend/app/services/kline_sync.py +++ b/backend/app/services/kline_sync.py @@ -92,7 +92,7 @@ def sync_daily_batch(symbols: list[str], chunks = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)] for i, chunk in enumerate(chunks): - if i > 0 and interval > 0 and len(chunks) > rpm: + if i > 0 and interval > 0: time.sleep(interval) try: if start_time and end_time: @@ -295,7 +295,7 @@ def sync_adj_factor(symbols: list[str], repo: KlineRepository, all_dfs: list[pl.DataFrame] = [] for i, chunk in enumerate(chunks): - if i > 0 and interval > 0 and len(chunks) > rpm: + if i > 0 and interval > 0: time.sleep(interval) try: raw = tf.klines.ex_factors(chunk, **sdk_kwargs) @@ -427,7 +427,7 @@ def sync_minute_batch( chunks = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)] for i, chunk in enumerate(chunks): - if i > 0 and interval > 0 and len(chunks) > rpm: + if i > 0 and interval > 0: time.sleep(interval) try: if start_time and end_time: diff --git a/backend/app/services/pipeline_jobs.py b/backend/app/services/pipeline_jobs.py index 979eddb..81ec0c6 100644 --- a/backend/app/services/pipeline_jobs.py +++ b/backend/app/services/pipeline_jobs.py @@ -23,6 +23,11 @@ logger = logging.getLogger(__name__) JobStatus = Literal["pending", "running", "succeeded", "failed"] +# 运行超过此秒数视为卡死(reload 后孤儿 task / 网络读无限阻塞等)。 +# 由 reap_stale() 在 /run 和 /jobs/{id} 轮询端点检查 — 保证卡死后能自愈, +# 无需用户再次点击「同步」。 +STALE_JOB_TIMEOUT_S = 600 + def _default_store_dir() -> Path: from app.config import settings @@ -208,6 +213,36 @@ class JobStore: def active_id(self) -> str | None: return self._active_id + def reap_stale(self, timeout_s: int = STALE_JOB_TIMEOUT_S) -> None: + """回收运行超过 timeout_s 的卡死 running job(标记为 failed)。 + + 在 /run 和 /jobs/{id} 轮询端点都会调用 — 保证卡死后任意轮询都能自愈, + 无需用户再次手动触发同步。reload 后的孤儿 task(内存里已无 job 记录) + 不在此处理:它们没有 active_id,只能靠 executor 线程自然结束或进程重启。 + """ + with self._lock: + jid = self._active_id + if not jid: + return + j = self._active_jobs.get(jid) + if not j or j.get("status") != "running": + return + started = j.get("started_at") + if not started: + return + # 时间计算放到锁外(避免 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 > timeout_s: + logger.warning("reap_stale: 强制取消卡死 job %s (已运行 %.0fs)", + jid, elapsed) + self.fail(jid, f"超时自动取消 (运行 {int(elapsed)}s, 疑似卡死)") + def clear(self) -> None: """清空所有任务（内存 + 磁盘文件）。""" with self._lock: diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index beaaa7e..260acaf 100644 --- a/backend/app/tickflow/repository.py +++ b/backend/app/tickflow/repository.py @@ -180,7 +180,10 @@ class DataStore: for sql in statements: try: self.db.execute(sql) - except duckdb.IOException: + except Exception as e: # noqa: BLE001 + # 空数据目录(首次启动)或权限问题时 DuckDB 会抛 IOException; + # 跨版本/平台也可能抛 CatalogException 等。空目录缺视图不影响启动 + # (后续同步写入数据后会重新刷新视图),这里一律降级为 debug 日志。 logger.debug("view registration skipped (no parquet yet): %s", sql[:60]) self._register_unified_views()