mirror of
https://ghfast.top/https://github.com/aeroxw/tick-stock-panel.git
synced 2026-09-12 16:44:15 +08:00
wshyandshy3130
57f417e6eb
三处根因:
1. 限流逻辑失效 (kline_sync.py ×3, index_sync.py ×2)
条件 `len(chunks) > rpm` 在 free 档恒为 False (56 chunks < 60 rpm),
导致节流 sleep 永不执行, 请求密集打出触发服务端限流 → 卡死。
改为始终按 interval (60/rpm) 节流。
2. 卡死看门狗无法自愈 (pipeline_jobs.py, api/pipeline.py)
原超时检查只在再次点「同步」时触发, 等待中永不回收。
抽出 JobStore.reap_stale() (STALE_JOB_TIMEOUT_S=600), 在
/run 和 /jobs/{id} 轮询端点都调用 — 前端每秒轮询, 10分钟后
必定自动回收卡死 job, UI 不再永久停在 26%。
3. 启动期视图注册异常捕获不足 (repository.py)
_register_views 只捕获 duckdb.IOException, 跨版本/平台空目录
可能抛 CatalogException 等炸掉 lifespan。放宽到 Exception。
Co-authored-by: shy3130 <shy3130@users.noreply.github.com>
98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
"""盘后管道 API — 异步触发 + 进度跟踪。"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import concurrent.futures as _cf
|
|
import logging
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
|
|
from app.jobs import daily_pipeline
|
|
from app.services.pipeline_jobs import job_store
|
|
from app.api.data import invalidate_storage_cache
|
|
|
|
# 长时间任务专用线程池（隔离于 FastAPI 默认线程池，防止阻塞请求处理）
|
|
_long_task_executor = _cf.ThreadPoolExecutor(max_workers=2, thread_name_prefix="long-task")
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/pipeline", tags=["pipeline"])
|
|
|
|
|
|
@router.post("/run")
|
|
async def run_now(request: Request) -> dict:
|
|
"""异步触发盘后管道,立即返回 job_id。客户端轮询 /jobs/{id} 拿进度。
|
|
|
|
若已有任务在跑,**返回该任务 id 而不是开新任务**(防止并发拉数据撞限流)。
|
|
但如果该任务已运行超过 10 分钟 (可能因 reload 卡死), 强制标记为失败后重新创建。
|
|
"""
|
|
repo = request.app.state.repo
|
|
capset = request.app.state.capabilities
|
|
|
|
# 检测卡死的 running job (如 reload 后孤儿 task / 网络读无限阻塞)。
|
|
# reap_stale 会在 /run 和 /jobs/{id} 轮询端点都调用,保证卡死后能自愈。
|
|
job_store.reap_stale()
|
|
|
|
job_id = job_store.create()
|
|
|
|
# 如果是复用的 active job,直接返回(不重启)
|
|
existing = job_store.get(job_id)
|
|
if existing and existing["status"] == "running":
|
|
return {"job_id": job_id, "reused": True}
|
|
|
|
# 在 executor 里跑同步任务(pipeline 内部都是阻塞 IO + CPU)
|
|
async def task() -> None:
|
|
job_store.start(job_id)
|
|
loop = asyncio.get_event_loop()
|
|
|
|
def progress(stage: str, pct: int, msg: str, stage_pct: int | None = None,
|
|
skip_log: bool = False) -> None:
|
|
job_store.progress(job_id, stage, pct, msg, stage_pct=stage_pct, skip_log=skip_log)
|
|
|
|
try:
|
|
result = await loop.run_in_executor(
|
|
_long_task_executor,
|
|
lambda: daily_pipeline.run_now(repo, capset, on_progress=progress),
|
|
)
|
|
job_store.succeed(job_id, result)
|
|
invalidate_storage_cache()
|
|
repo.refresh_cache() # 刷新 Polars 缓存
|
|
except Exception as e: # noqa: BLE001
|
|
logger.exception("pipeline failed")
|
|
job_store.fail(job_id, str(e))
|
|
invalidate_storage_cache()
|
|
|
|
asyncio.create_task(task())
|
|
return {"job_id": job_id, "reused": False}
|
|
|
|
|
|
@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")
|
|
return j
|
|
|
|
|
|
@router.post("/jobs/{job_id}/cancel")
|
|
def cancel_job(job_id: str) -> dict:
|
|
"""手动取消一个 running 的 job。"""
|
|
j = job_store.get(job_id)
|
|
if not j:
|
|
raise HTTPException(status_code=404, detail="job not found")
|
|
if j["status"] not in ("running", "pending"):
|
|
raise HTTPException(status_code=400, detail=f"job status is {j['status']}, cannot cancel")
|
|
job_store.fail(job_id, "用户手动取消")
|
|
return {"cancelled": job_id}
|
|
|
|
|
|
@router.get("/jobs")
|
|
def list_jobs(limit: int = 20) -> dict:
|
|
return {
|
|
"active_id": job_store.active_id(),
|
|
"jobs": job_store.list_recent(limit=limit),
|
|
}
|