复核 R-b2(单进程下周期整理退回判定中的待办):当班进程隔一会儿的那一次接手(trust_mine)不论有没有别的进程,停在 resolving 的待办只收拾停了两个租约时长以上的,启动时又没有别的进程的那一次照旧全收拾;顺带把其余周期调用的整理过一遍——结果未知的写入只补停了两个租约时长以上的(收尾和安排核实之间的不碰),问数据没有任务行的老轮次只在启动时整理(老页面开的轮次建图时也是这个样子)。补测试:判定到一半当班进程来一次,待办照样记已处理、那一票不撤回(改前的写法这条不过)

This commit is contained in:
ThearchyHelios
2026-10-19 13:18:23 +08:00
parent 35ab097b45
commit 96bb372ef3
4 changed files with 47 additions and 4 deletions
+3 -1
View File
@@ -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:
+12 -1
View File
@@ -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)
+3 -2
View File
@@ -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)
+29
View File
@@ -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]