yumoqing/pipeline-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: SELECT-only 事务显式 COMMIT，修复 MDL 元数据锁泄漏阻塞 DDL

Browse Source 
failed poller / handle_failed_task 的 SELECT-only 事务未提交，连接回池后仍持有元数据锁，
导致 ALTER TABLE ADD COLUMN 卡死并阻塞整表。改为 SELECT 后显式 COMMIT 释放锁。
This commit is contained in:
ymq 2026-08-15 01:09:05 +08:00
parent aa0dcc9c04
commit 14a42aac7b
2 changed files with 8 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
pipeline_service/agent_loop.py
View File

@ -941,6 +941,8 @@ async def handle_failed_task(task_id: str, project_id: str) -> dict:
"WHERE id=${tid}$ AND state='failed'", "WHERE id=${tid}$ AND state='failed'",
{"tid": task_id}) {"tid": task_id})
if not recs: if not recs:
# 释放 SELECT 的元数据锁，防止连接回池后阻塞 DDL
await sor.sqlExe("COMMIT", {})
return {"status": "idle", "task_id": task_id} return {"status": "idle", "task_id": task_id}
t = recs[0] t = recs[0]
title = getattr(t, "title", "") or "" title = getattr(t, "title", "") or ""
@ -951,6 +953,9 @@ async def handle_failed_task(task_id: str, project_id: str) -> dict:
retry_count = 0 retry_count = 0
last_error = getattr(t, "last_error", "") or "" last_error = getattr(t, "last_error", "") or ""
# 释放 SELECT 的元数据锁，避免决策/建问题期间长时间持有 MDL
await sor.sqlExe("COMMIT", {})
# 已重试 3 次仍未成功 → 报故障 # 已重试 3 次仍未成功 → 报故障
if retry_count >= 3: if retry_count >= 3:
decision = "fault" decision = "fault"

3
pipeline_service/init.py
View File

@ -663,6 +663,9 @@ def load_pipeline_service():
_fd_dispatched.discard(tid) _fd_dispatched.discard(tid)
asyncio.ensure_future(_fd_dispatch(tid, pid)) asyncio.ensure_future(_fd_dispatch(tid, pid))
# SELECT-only 事务必须显式 COMMIT，否则连接回到池后仍持有元数据锁(MDL)，
# 会阻塞后续 ALTER TABLE / DDL（sqlor 只在有写入时才自动提交）。
await sor.sqlExe("COMMIT", {})
except Exception as e: except Exception as e:
debug(f"failed_poller error: {e}") debug(f"failed_poller error: {e}")
await asyncio.sleep(60) await asyncio.sleep(60)