mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
ThearchyHelios
ed83aa2fa4
两条并行支路一条失败，LangGraph 会取消另一条；那一条若正在写库（节点收尾写 工件最常见），整个运行就可能永远停在 running，占着的并发名额也不还。修审批时 加的并行驳回用例约五次卡一次，才把它翻出来。 根因在两层，都是"取消打进进行中的数据库清理"： 1. aiosqlite 0.22 的 stop() 每调一次都往工作线程队列里放一个"关闭并退出" 请求，线程处理完第一个就退出，之后的请求没人处理，等它的一方永远挂着。 close() 的 finally 会调 stop()，SQLAlchemy 优雅关闭被取消后退回的强制 terminate 也会直接调——同一连接上两者一碰就卡死。并发 close 两次，10 次卡 10 次。现在同一条连接的 stop() 只真正执行一次，之后的调用拿同一个 future。 2. 卡死解开后露出第二层：SQLAlchemy 的 _close_connection 记完日志会把 CancelledError 原样抛出，跳过了把连接记录清空的那一行，已经关掉的连接留在 池里，下一个取到它的人拿到 "no active connection"。主引擎开 pool_pre_ping： 探活失败按断开处理、换新连接（aiosqlite 方言把这句判为断开，已核对源码）。 验证：test_aiosqlite_close.py 4 例是确定性的（并发 close、close 撞上强制 stop、被取消后的 close、重复 close），去掉补丁 3 例超时失败。复现脚本同一场景 连跑两批各 40 轮：修前 12 轮内必卡，修后 80 轮 0 卡、0 次坏连接抛给调用方； SQLAlchemy 在被取消时仍会记一条"Exception terminating connection"日志，那是 它自愈过程的记录，不影响结果。334 → 338 passed。
49 lines
2.0 KiB
Python
49 lines
2.0 KiB
Python
"""aiosqlite 连接的关闭不能把调用方永远挂住。
|
||
|
||
aiosqlite 0.22 的 stop() 每调一次都往工作线程队列里放一个"关闭并退出线程"
|
||
的请求：线程处理完第一个就退出，之后的请求没人处理，等它的一方永远挂着。
|
||
close() 的 finally 会调 stop()，SQLAlchemy 的强制 terminate 也会直接调——
|
||
不打补丁时同一连接并发 close 两次，10 次卡 10 次。
|
||
|
||
它在这个项目里的真实后果是运行卡死：并行支路一条失败，LangGraph 取消另一条，
|
||
那一条正在写工件，SQLAlchemy 就在两条路径上各 terminate 一次同一条连接——
|
||
节点任务取消不完，运行永远停在 running（test_approval 里并行驳回那例约五次
|
||
卡一次，就是这么被发现的）。补丁在 app.db.base。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
|
||
import aiosqlite
|
||
|
||
import app.db.base # noqa: F401 补丁在这里打上
|
||
|
||
|
||
async def test_concurrent_close_of_one_connection_finishes(tmp_path):
|
||
conn = await aiosqlite.connect(tmp_path / "x.db")
|
||
await conn.execute("CREATE TABLE t (x INTEGER)")
|
||
await asyncio.wait_for(asyncio.gather(conn.close(), conn.close(), conn.close()), timeout=5)
|
||
|
||
|
||
async def test_close_racing_a_forced_stop_finishes(tmp_path):
|
||
"""SQLAlchemy 的优雅关闭被取消时会退回强制 terminate，直接调 stop()。"""
|
||
conn = await aiosqlite.connect(tmp_path / "w.db")
|
||
closing = asyncio.ensure_future(conn.close())
|
||
await asyncio.sleep(0)
|
||
conn.stop()
|
||
await asyncio.wait_for(closing, timeout=5)
|
||
|
||
|
||
async def test_close_after_a_cancelled_close_finishes(tmp_path):
|
||
conn = await aiosqlite.connect(tmp_path / "y.db")
|
||
first = asyncio.ensure_future(conn.close())
|
||
await asyncio.sleep(0)
|
||
first.cancel()
|
||
await asyncio.wait_for(conn.close(), timeout=5)
|
||
|
||
|
||
async def test_closing_an_already_closed_connection_is_a_no_op(tmp_path):
|
||
conn = await aiosqlite.connect(tmp_path / "z.db")
|
||
await conn.close()
|
||
await asyncio.wait_for(conn.close(), timeout=5)
|