复核 R-b8(隔离时 asyncio 记一条回调异常):断点写入的守卫改成协程包装、在协程里查 fence——LangGraph 在节点的完成回调里调 aput_writes 拿协程交给后台任务,以前同步包装在回调里就抛 Fenced,每次隔离都打一条「Exception in callback … FuturesDict.on_done」加 traceback;现在异常留在写断点的那个任务里,由 LangGraph 照常收走。补测试:卡住被接手那一条里 asyncio 不记回调异常、不记没人取的任务异常(改前的写法这条不过)

This commit is contained in:
ThearchyHelios
2026-10-19 14:29:12 +08:00
parent 7fcce90f95
commit d6ecaaa7a3
2 changed files with 17 additions and 4 deletions
+7 -3
View File
@@ -1786,15 +1786,19 @@ run_manager = RunManager()
def _fenced_saver(saver: BaseCheckpointSaver) -> BaseCheckpointSaver:
"""断点库的写也要过 cluster/fence:本进程续不上租约之后,手上的执行不再写断点(接手的进程会从断点接着跑,两边都写就分叉了)。"""
"""断点库的写也要过 cluster/fence:本进程续不上租约之后,手上的执行不再写断点(接手的进程会从断点接着跑,两边都写就分叉了)。
包成协程、在协程里查(复核 R-b8):LangGraph 在节点的完成回调里调 aput_writes 拿协程交给后台任务——同步包装在回调里就抛了,
每次隔离 asyncio 都记一条「Exception in callback … FuturesDict.on_done」;在协程里抛,异常留在写断点的那个任务里,由 LangGraph
照常收走。"""
for name in ("aput", "aput_writes"):
original = getattr(saver, name, None)
if original is None:
continue
def guarded(*args: Any, __original: Any = original, **kwargs: Any) -> Any:
async def guarded(*args: Any, __original: Any = original, **kwargs: Any) -> Any:
fence.check()
return __original(*args, **kwargs)
return await __original(*args, **kwargs)
setattr(saver, name, guarded)
return saver
+10 -1
View File
@@ -111,8 +111,17 @@ async def _stall_and_take_over(slow) -> str:
return run.id
async def test_a_stalled_executor_does_not_overwrite_the_takeover(slow):
async def test_a_stalled_executor_does_not_overwrite_the_takeover(slow, caplog):
import gc
import logging
caplog.set_level(logging.ERROR, logger="asyncio")
run_id = await _stall_and_take_over(slow)
await asyncio.sleep(0.1)
gc.collect()
await asyncio.sleep(0)
noisy = [r.getMessage() for r in caplog.records if r.name == "asyncio"]
assert noisy == [], f"隔离时 asyncio 不该记回调异常、没人取的任务异常(复核 R-b8):{noisy}"
async with SessionLocal() as session:
row = await session.get(Run, run_id)
assert row.status == "interrupted" and row.owner is None, "接手之后的样子没被盖掉"