并行支路一条失败时,运行不再永久卡在 running

两条并行支路一条失败,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。
This commit is contained in:
ThearchyHelios
2026-09-25 15:19:54 +08:00
parent 26ff8e1e06
commit ed83aa2fa4
2 changed files with 87 additions and 1 deletions
+39 -1
View File
@@ -40,7 +40,45 @@ class TimestampMixin:
)
engine = create_async_engine(settings.db_url, echo=False, future=True)
def _stop_aiosqlite_once() -> None:
"""aiosqlite 连接的工作线程只停一次;之后再要求停的,拿到的是同一个结果。
aiosqlite 0.22 的 stop() 每调一次都往工作线程队列里放一个"关闭并退出线程"
的请求,而线程处理完第一个就退出了——之后放进去的请求没人处理,等它的一方
永远挂着。close() 的 finally 里会调 stop(),SQLAlchemy 的强制 terminate 也会
直接调;两个 close 并发、或者 close 撞上强制 terminate,就会卡死。实测同一
连接并发 close 两次,10 次卡 10 次。
在这里触发它的路径:并行的两条支路一条失败,LangGraph 取消另一条,而那一条
正在写工件。取消打进进行中的数据库操作,SQLAlchemy 在两条路径上各 terminate
一次同一条连接——节点任务取消不完,整个运行卡在 running,占着的并发名额也不
还。大约五次里有一次。连接池和 checkpointer 都走 aiosqlite,补在这一层都管得到。
"""
import weakref
import aiosqlite
original = aiosqlite.Connection.stop
if getattr(original, "_agentlab_once", False):
return
stopping: weakref.WeakKeyDictionary = weakref.WeakKeyDictionary()
def stop(self): # type: ignore[no-untyped-def]
if self not in stopping:
stopping[self] = original(self)
return stopping[self]
stop._agentlab_once = True # type: ignore[attr-defined]
aiosqlite.Connection.stop = stop # type: ignore[method-assign]
_stop_aiosqlite_once()
# pool_pre_ping:取连接前先探一下活。取消异常打进 SQLAlchemy 的清理流程时
# (_close_connection 记完日志会把 CancelledError 原样抛出,跳过了把连接记录
# 清空的那一行),已经关掉的连接会留在池里,下一个取到它的人拿到的是
# "no active connection"。探活失败会被当成断开,换一条新的——SQLite 本地一次
# SELECT 1 的代价,换池子不被污染。
engine = create_async_engine(settings.db_url, echo=False, future=True, pool_pre_ping=True)
@event.listens_for(engine.sync_engine, "connect")
+48
View File
@@ -0,0 +1,48 @@
"""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)