diff --git a/pipeline_service/init.py b/pipeline_service/init.py index cac8dba..5457ef9 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -739,27 +739,51 @@ def load_pipeline_service(): "UPDATE pipeline_tasks SET claimed_by=NULL, updated_at=NOW() " "WHERE state='review' AND claimed_by IS NOT NULL " "AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {}) + # 候选任务:排除 paused 项目的任务(暂停门控提前到筛选,避免 paused 项目僵尸任务 + # 卡死队列头、饿死 active 项目的审核)。LIMIT 放大容纳「paused 任务排前面」的僵尸任务。 recs = await sor.sqlExe( - "SELECT id, tenant_id, role FROM pipeline_tasks " - "WHERE state='review' AND claimed_by IS NULL " - "ORDER BY created_at ASC LIMIT 5", + "SELECT t.id, t.tenant_id, t.role FROM pipeline_tasks t " + "WHERE t.state='review' AND t.claimed_by IS NULL " + "AND NOT EXISTS (SELECT 1 FROM sd_projects p WHERE p.id=t.tenant_id AND p.status='paused') " + "ORDER BY t.created_at ASC LIMIT 200", {}) + await sor.sqlExe("COMMIT", {}) + + # 项目轮询(round-robin):每个项目轮流取一个任务,保证项目间公平, + # 避免单项目占满窗口、饿死其他项目的审核(与 agent_poller 队列饥饿修复同款)。 + by_project = {} for rec in (recs or []): + by_project.setdefault(getattr(rec, 'tenant_id', ''), []).append(rec) + selected = [] + while by_project: + progressed = False + for pid in list(by_project.keys()): + tasks = by_project[pid] + if tasks: + selected.append(tasks.pop(0)) + progressed = True + if not tasks: + del by_project[pid] + if not progressed: + break + + for rec in selected: tid = getattr(rec, 'id', '') pid = getattr(rec, 'tenant_id', '') - if tid and pid and tid not in _pm_dispatched: - _pm_dispatched.add(tid) - debug(f"pm_poller dispatching review task={tid}") + if not (tid and pid) or tid in _pm_dispatched: + continue + _pm_dispatched.add(tid) + debug(f"pm_poller dispatching review task={tid}") - async def _pm_dispatch(tid, pid): - try: - await pm_review_run(pid, agent_id="pm-poller") - except Exception as e: - debug(f"pm_poller dispatch error: task={tid} err={e}") - finally: - _pm_dispatched.discard(tid) + async def _pm_dispatch(tid, pid): + try: + await pm_review_run(pid, agent_id="pm-poller") + except Exception as e: + debug(f"pm_poller dispatch error: task={tid} err={e}") + finally: + _pm_dispatched.discard(tid) - asyncio.ensure_future(_pm_dispatch(tid, pid)) + asyncio.ensure_future(_pm_dispatch(tid, pid)) async def _pm_poll_loop(): while True: