修复任务卡死与安能启动失败
- runtime.py: launch_anneng 拉起前清除 NODE_OPTIONS,避免 Electron 因 --use-system-ca 启动即退出(rc=9) - server.py: POST /tasks 在 worker 未就绪时返回 409 拦截误操作;worker 就绪后调用 fail_stale_tasks 实现重启自愈 - state_store.py: 新增 fail_stale_tasks(),将遗留 pending/running 任务标记为 failed
This commit is contained in:
@@ -107,9 +107,14 @@ def _wait_cdp_up(port, timeout=60.0):
|
|||||||
|
|
||||||
def launch_anneng(app_path):
|
def launch_anneng(app_path):
|
||||||
"""以调试模式启动安能 Electron 应用(自动选取空闲端口),返回子进程对象。"""
|
"""以调试模式启动安能 Electron 应用(自动选取空闲端口),返回子进程对象。"""
|
||||||
|
# 【环境兼容】WorkBuddy 等 shell 会注入 NODE_OPTIONS(含 --use-system-ca),
|
||||||
|
# Electron 内置 Node 拒绝该 flag 导致安能启动即退出(rc=9)。
|
||||||
|
# 拉起前从子进程环境里摘掉 NODE_OPTIONS。
|
||||||
|
anneng_env = os.environ.copy()
|
||||||
|
anneng_env.pop("NODE_OPTIONS", None)
|
||||||
port = _find_free_port()
|
port = _find_free_port()
|
||||||
print(f">> 以调试模式启动【安能】应用(端口 {port}):{app_path}")
|
print(f">> 以调试模式启动【安能】应用(端口 {port}):{app_path}")
|
||||||
proc = subprocess.Popen([app_path, f"--remote-debugging-port={port}"])
|
proc = subprocess.Popen([app_path, f"--remote-debugging-port={port}"], env=anneng_env)
|
||||||
site_anneng.set_cdp_port(port)
|
site_anneng.set_cdp_port(port)
|
||||||
if not _wait_cdp_up(port):
|
if not _wait_cdp_up(port):
|
||||||
raise RuntimeError(
|
raise RuntimeError(
|
||||||
|
|||||||
@@ -61,6 +61,10 @@ def _worker_loop():
|
|||||||
ctx = launch_and_prepare()
|
ctx = launch_and_prepare()
|
||||||
worker_state["ctx"] = ctx
|
worker_state["ctx"] = ctx
|
||||||
worker_state["ready"] = True
|
worker_state["ready"] = True
|
||||||
|
# 【P1-2 重启自愈】worker 就绪后清理上轮遗留的 pending/running 僵尸任务
|
||||||
|
cleaned = state_store.fail_stale_tasks()
|
||||||
|
if cleaned:
|
||||||
|
print(f">> [worker] 自愈:清理 {cleaned} 条遗留任务(pending/running → failed)")
|
||||||
print(">> [worker] 各站就绪,开始接收任务 ...")
|
print(">> [worker] 各站就绪,开始接收任务 ...")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
worker_state["error"] = str(e)
|
worker_state["error"] = str(e)
|
||||||
@@ -155,6 +159,9 @@ class TaskRequest(BaseModel):
|
|||||||
@app.post("/tasks")
|
@app.post("/tasks")
|
||||||
def create_task(req: TaskRequest):
|
def create_task(req: TaskRequest):
|
||||||
"""提交任务 {site, kind} → 入队,返回 task_id。"""
|
"""提交任务 {site, kind} → 入队,返回 task_id。"""
|
||||||
|
# 【P0】后端未就绪时直接拒绝,避免任务在 worker 启动前入队卡死
|
||||||
|
if not worker_state["ready"]:
|
||||||
|
raise HTTPException(status_code=409, detail="后端尚未就绪,请等待各站点登录完成后再操作")
|
||||||
if (req.site, req.kind) not in TASK_HANDLERS:
|
if (req.site, req.kind) not in TASK_HANDLERS:
|
||||||
raise HTTPException(status_code=400, detail=f"无效任务: {req.site}/{req.kind}")
|
raise HTTPException(status_code=400, detail=f"无效任务: {req.site}/{req.kind}")
|
||||||
task_id = state_store.create_task(req.site, req.kind)
|
task_id = state_store.create_task(req.site, req.kind)
|
||||||
|
|||||||
@@ -426,3 +426,16 @@ def list_tasks(limit=20):
|
|||||||
}
|
}
|
||||||
for r in rows
|
for r in rows
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def fail_stale_tasks(reason: str = "服务重启,上轮未完成任务,请手动重跑") -> int:
|
||||||
|
"""worker 启动时调用:把遗留的 pending/running 任务标记为 failed,实现重启自愈。
|
||||||
|
返回被清理的任务数量。"""
|
||||||
|
with sqlite3.connect(STATE_DB_PATH) as conn:
|
||||||
|
cur = conn.execute(
|
||||||
|
"UPDATE task_history SET status=?, finished_at=?, error=? "
|
||||||
|
"WHERE status IN (?, ?)",
|
||||||
|
(TASK_FAILED, _now(), reason, TASK_PENDING, TASK_RUNNING),
|
||||||
|
)
|
||||||
|
conn.commit()
|
||||||
|
return cur.rowcount
|
||||||
|
|||||||
Reference in New Issue
Block a user