fix: poller 加 watchdog(wait_for 60s 超时)，防连接池/MDL 锁卡死导致停摆
agent_poller 和 pm_poller 的 poll_loop 改为 _poll_once + asyncio.wait_for(timeout=60) 包裹，每轮 poll 卡住超过 60s 就跳过本轮继续，不再因连接池耗尽/MDL 锁等待 (lock_wait_timeout 默认 24h)而永久停摆(2026-08 生产事故：poller 停摆 15.7h)。
This commit is contained in:
ymq
parent
0b75f10731
commit
b819a665f6
@ -581,54 +581,60 @@ def load_pipeline_service():
|
||||
poll_db = DBPools()
|
||||
_dispatched = set() # 防止重复分发
|
||||
|
||||
async def _poll_once(poll_db):
|
||||
"""执行一轮 agent poll。可能因连接获取/SQL 锁等待卡住，由外层 wait_for 超时保护。"""
|
||||
async with poll_db.sqlorContext("pipeline") as sor:
|
||||
# 回收僵尸 running 任务：进程崩溃/协程挂起遗留（心跳超时未更新）。
|
||||
# 阈值 20 分钟 > LLM 单轮最坏时长（3 次 × 300s 重试 ≈ 15 分钟），避免误杀慢任务。
|
||||
await sor.sqlExe(
|
||||
"UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL, updated_at=NOW() "
|
||||
"WHERE state='running' AND pipeline_id='role_task' "
|
||||
"AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {})
|
||||
|
||||
# 全局并发 agent 数限制（max_concurrent_agents，appbase params 表，默认 3）。
|
||||
# 以 DB state='running' 计数为准（权威），超出上限时本轮回合不再派发。
|
||||
from .workspace import get_max_concurrent_agents
|
||||
max_n = await get_max_concurrent_agents(sor)
|
||||
running_rows = await sor.sqlExe(
|
||||
"SELECT COUNT(*) as c FROM pipeline_tasks "
|
||||
"WHERE state='running' AND pipeline_id='role_task'", {})
|
||||
cur = getattr(running_rows[0], 'c', 0) if running_rows else 0
|
||||
avail = max(0, max_n - cur)
|
||||
if avail <= 0:
|
||||
# 达到并发上限，本轮不派发（退出 context 后末尾 sleep 再 poll）
|
||||
return
|
||||
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT id, tenant_id, role FROM pipeline_tasks "
|
||||
"WHERE state='submitted' AND pipeline_id='role_task' "
|
||||
"AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 20",
|
||||
{})
|
||||
for rec in (recs or [])[:avail]:
|
||||
tid = getattr(rec, 'id', '')
|
||||
pid = getattr(rec, 'tenant_id', '')
|
||||
r = getattr(rec, 'role', '')
|
||||
if tid and pid and r and tid not in _dispatched:
|
||||
_dispatched.add(tid)
|
||||
debug(f"agent_poller dispatching task={tid} role={r}")
|
||||
|
||||
async def _dispatch(tid, pid, r):
|
||||
try:
|
||||
await role_agent_loop(pid, r, agent_id=f"poller-{r}")
|
||||
except Exception as e:
|
||||
debug(f"agent_poller dispatch error: task={tid} err={e}")
|
||||
finally:
|
||||
_dispatched.discard(tid)
|
||||
|
||||
asyncio.ensure_future(_dispatch(tid, pid, r))
|
||||
|
||||
async def _poll_loop():
|
||||
while True:
|
||||
try:
|
||||
async with poll_db.sqlorContext("pipeline") as sor:
|
||||
# 回收僵尸 running 任务：进程崩溃/协程挂起遗留（心跳超时未更新）。
|
||||
# 阈值 20 分钟 > LLM 单轮最坏时长（3 次 × 300s 重试 ≈ 15 分钟），避免误杀慢任务。
|
||||
await sor.sqlExe(
|
||||
"UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL, updated_at=NOW() "
|
||||
"WHERE state='running' AND pipeline_id='role_task' "
|
||||
"AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {})
|
||||
|
||||
# 全局并发 agent 数限制（max_concurrent_agents，appbase params 表，默认 3）。
|
||||
# 以 DB state='running' 计数为准（权威），超出上限时本轮回合不再派发。
|
||||
from .workspace import get_max_concurrent_agents
|
||||
max_n = await get_max_concurrent_agents(sor)
|
||||
running_rows = await sor.sqlExe(
|
||||
"SELECT COUNT(*) as c FROM pipeline_tasks "
|
||||
"WHERE state='running' AND pipeline_id='role_task'", {})
|
||||
cur = getattr(running_rows[0], 'c', 0) if running_rows else 0
|
||||
avail = max(0, max_n - cur)
|
||||
if avail <= 0:
|
||||
# 达到并发上限，本轮不派发（退出 context 后末尾 sleep 再 poll）
|
||||
continue
|
||||
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT id, tenant_id, role FROM pipeline_tasks "
|
||||
"WHERE state='submitted' AND pipeline_id='role_task' "
|
||||
"AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 20",
|
||||
{})
|
||||
for rec in (recs or [])[:avail]:
|
||||
tid = getattr(rec, 'id', '')
|
||||
pid = getattr(rec, 'tenant_id', '')
|
||||
r = getattr(rec, 'role', '')
|
||||
if tid and pid and r and tid not in _dispatched:
|
||||
_dispatched.add(tid)
|
||||
debug(f"agent_poller dispatching task={tid} role={r}")
|
||||
|
||||
async def _dispatch(tid, pid, r):
|
||||
try:
|
||||
await role_agent_loop(pid, r, agent_id=f"poller-{r}")
|
||||
except Exception as e:
|
||||
debug(f"agent_poller dispatch error: task={tid} err={e}")
|
||||
finally:
|
||||
_dispatched.discard(tid)
|
||||
|
||||
asyncio.ensure_future(_dispatch(tid, pid, r))
|
||||
# watchdog：每轮 poll 最多 60 秒，超时则跳过本轮，防止连接池/MDL 锁
|
||||
# 卡死导致 poller 永久停摆（2026-08 生产事故：poller 停摆 15.7h）
|
||||
await asyncio.wait_for(_poll_once(poll_db), timeout=60)
|
||||
except Exception as e:
|
||||
debug(f"agent_poller error: {e}")
|
||||
debug(f"agent_poller error/timeout: {e}")
|
||||
await asyncio.sleep(10)
|
||||
|
||||
asyncio.create_task(_poll_loop())
|
||||
@ -643,38 +649,43 @@ def load_pipeline_service():
|
||||
pm_db = _DBP()
|
||||
_pm_dispatched = set()
|
||||
|
||||
async def _pm_poll_once(pm_db):
|
||||
"""执行一轮 PM poll。可能因连接获取/SQL 锁等待卡住，由外层 wait_for 超时保护。"""
|
||||
async with pm_db.sqlorContext("pipeline") as sor:
|
||||
# 回收僵尸 review 任务：PM 审核进程崩溃后 claimed_by 残留，重新放回审核队列。
|
||||
await sor.sqlExe(
|
||||
"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)", {})
|
||||
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",
|
||||
{})
|
||||
for rec in (recs or []):
|
||||
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}")
|
||||
|
||||
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))
|
||||
|
||||
async def _pm_poll_loop():
|
||||
while True:
|
||||
try:
|
||||
async with pm_db.sqlorContext("pipeline") as sor:
|
||||
# 回收僵尸 review 任务：PM 审核进程崩溃后 claimed_by 残留，重新放回审核队列。
|
||||
await sor.sqlExe(
|
||||
"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)", {})
|
||||
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",
|
||||
{})
|
||||
for rec in (recs or []):
|
||||
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}")
|
||||
|
||||
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))
|
||||
# watchdog：每轮 PM poll 最多 60 秒，超时则跳过本轮，防止 poller 永久停摆
|
||||
await asyncio.wait_for(_pm_poll_once(pm_db), timeout=60)
|
||||
except Exception as e:
|
||||
debug(f"pm_poller error: {e}")
|
||||
debug(f"pm_poller error/timeout: {e}")
|
||||
await asyncio.sleep(15)
|
||||
|
||||
asyncio.create_task(_pm_poll_loop())
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user