diff --git a/backend/app/chat/job.py b/backend/app/chat/job.py index b29b3a8c..438f6fba 100644 --- a/backend/app/chat/job.py +++ b/backend/app/chat/job.py @@ -1234,7 +1234,9 @@ async def startup(*, resume: bool | None = None, live: Any = None) -> dict[str, await session.commit() for turn_id in gone: turn_log.notify(turn_id) - for row in legacy: + # 老轮次(没有任务行)只在启动时整理:当班进程隔一会儿来的那一次(live.trust_mine)不碰——老页面开的轮次(不占锁的写入) + # 正在建图时也是这个样子 + for row in [] if live.trust_mine else legacy: if turn_run_id(row): continue # 有运行的老轮次:页面打开时请服务端接着收尾 async with SessionLocal() as session: diff --git a/backend/app/process/actions.py b/backend/app/process/actions.py index 2b6a6393..f8c8e2b2 100644 --- a/backend/app/process/actions.py +++ b/backend/app/process/actions.py @@ -729,7 +729,8 @@ async def startup(*, live: Any = None) -> int: """启动时:上次进程没了的时候已经发出(sent)、还没拿到结果的动作收成结果未知,安排核实;结果未知却还没安排核实的 (挂起之后、安排之前进程没了)补上。返回处理的条数。 - 多进程:所属的运行还在别的活着的进程里执行的不碰(那边正在发、正在等结果)。""" + 多进程:所属的运行还在别的活着的进程里执行的不碰(那边正在发、正在等结果);当班进程隔一会儿来的那一次只收拾停了两个租约 + 时长以上的。""" from app.cluster import state from app.db.models import Run from app.engine.runner import run_manager @@ -741,6 +742,16 @@ async def startup(*, live: Any = None) -> int: ActionRecord.state.in_(("sent", "unknown")), ActionRecord.resolution.is_(None)))).all() ids = [action_id for action_id, run_id, status, owner in rows if not (status in ("running", "queued") and live.alive(owner, local=run_manager.is_active(run_id)))] + if live.trust_mine and ids: + # 当班进程隔一会儿来的那一次:刚挂起、执行它的进程正在安排核实的(收尾和安排核实之间)不碰,只收拾停了两个租约时长以上的 + from datetime import timedelta + + from app.cluster.workers import lease_s + + floor = utcnow() - timedelta(seconds=2 * lease_s()) + async with SessionLocal() as session: + ids = list((await session.execute(select(ActionRecord.id).where( + ActionRecord.id.in_(ids), ActionRecord.updated_at < floor))).scalars()) for action_id in ids: try: await on_unknown(action_id) diff --git a/backend/app/process/tasks.py b/backend/app/process/tasks.py index e1690f40..a8bbbb4c 100644 --- a/backend/app/process/tasks.py +++ b/backend/app/process/tasks.py @@ -1027,14 +1027,15 @@ async def startup(*, live: Any = None) -> int: """启动时:存量的待审批补建 Task(派给任何人,幂等);上次停在 resolving 的(判定之后、恢复之前进程没了)按 Approval 的状态收拾——还在等的退回 pending、撤回那一票,已经交给引擎的记成已处理。返回补建的条数。 - 多进程:别的进程活着时,resolving 可能正是它在判定、恢复(几秒的事),只收拾停在 resolving 超过两个租约时长的。""" + 多进程:别的进程活着时,resolving 可能正是它在判定、恢复(几秒的事),只收拾停在 resolving 超过两个租约时长的;当班进程隔一会儿 + 来的那一次(live.trust_mine)不论有没有别的进程都这样——本进程自己也可能正在判定。只有启动时、又没有别的进程的那一次全收拾。""" from datetime import timedelta from app.cluster import state from app.cluster.workers import lease_s live = live if live is not None else await state.current(fresh=True) - stale_before = utcnow() - timedelta(seconds=2 * lease_s()) if live.others else None + stale_before = utcnow() - timedelta(seconds=2 * lease_s()) if (live.others or live.trust_mine) else None async with SessionLocal() as session: orphans = list((await session.execute( select(Approval).outerjoin(Task, Task.approval_id == Approval.id) diff --git a/backend/tests/test_task_resume_errors.py b/backend/tests/test_task_resume_errors.py index db9b885e..721a7fd2 100644 --- a/backend/tests/test_task_resume_errors.py +++ b/backend/tests/test_task_resume_errors.py @@ -84,3 +84,32 @@ async def test_any_resume_error_puts_the_task_back(monkeypatch): out = await tasks.respond(task.id, ANN, decision="approve") assert out["resolved"] is True assert (await _wait(run.id, ("succeeded", "failed"))).status == "succeeded" + + +async def test_the_periodic_takeover_leaves_a_task_that_is_being_resolved_alone(monkeypatch): + """复核 R-b2:当班进程隔一会儿的那一次接手(live.trust_mine)撞上一次正在判定、恢复的待办(停在 resolving 的那几秒), + 单进程时也不能把它退回 pending、撤回那一票——只收拾停了两个租约时长以上的。启动时、又没有别的进程的那一次照旧全收拾。""" + from app.cluster.workers import Liveness + + run = await run_manager.start(graph=FLOW, input_payload={"q": "x"}) + await _wait(run.id, ("interrupted",)) + task = await _pending(run.id) + real = run_manager.resume + reaped = asyncio.Event() + + async def slow(*a, **k): + await asyncio.sleep(0.3) + await tasks.startup(live=Liveness().trusting()) # 判定到一半,当班进程来了一次 + reaped.set() + return await real(*a, **k) + + monkeypatch.setattr(run_manager, "resume", slow) + out = await tasks.respond(task.id, ANN, decision="approve") + assert reaped.is_set() and out["resolved"] is True + assert (await _wait(run.id, ("succeeded", "failed"))).status == "succeeded" + async with SessionLocal() as session: + row = await session.get(Task, task.id) + mine = (await session.execute(select(TaskAssignment).where(TaskAssignment.task_id == task.id))).scalars().all() + assert row.status == "resolved", row.status + assert mine and all(a.responded_by == "ann" and a.status not in ("pending", "withdrawn") for a in mine), \ + [(a.status, a.responded_by) for a in mine]