协调员行消失按中断处理:本体行消失即放槽,子任务行消失按 interrupted 合流

- _run_coordinator 等待循环里 get_task 为 None 时按取消跳出,
  不再等到 deadline 占槽;终局 CAS 写不回去,最终为 interrupted,
  不写回、不抛错、不当完成。
- _coord_child_rows 对已删子任务合成 interrupted 行并注明行已消失,
  等齐判断不再拖到超时,合流报告标 interrupted 且整体不为 completed,
  其它子任务不受影响;docstring 旧说法同步改成新语义。
- 新增 cloud/tests/test_coordinator_row_vanished.py,3 个单测:
  本体行消失、子任务行消失、正常等齐;cloud 全量 1020 测通过。
This commit is contained in:
host committed 2026-10-28 08:57:05 +00:00
1 parent 5026e75b2c
commit c38ee8fbe9
2 files changed
+223 -3

No files matched your search

+19 -3
View File
@@ -2873,9 +2873,11 @@ class Worker:
def _coord_child_rows(self, child_ids):
"""子任务当前行:{id: {status, result_text, error_text}}。
已删或查不到的子任务不在结果里,调用方按缺失处理(等齐
判断里缺失不算终态,会一直等到超时——子任务行不应被删,
真删了就是超时失败,不静默当完成)。
已删或查不到的子任务按 interrupted 终态合成一行(error_text
注明行已消失):等齐判断里缺失不再等到超时,合流报告里
该子任务标 interrupted 且整体不为 completed;孤儿标记不
会对不存在的行再 UPDATE(rowcount 本来就是 0)。行消失
不是完成,绝不能静默当 completed 计。
"""
rows = {}
conn = _db_connect(self.db_path)
@@ -2885,6 +2887,10 @@ class Worker:
"SELECT id, status, result_text, error_text FROM tasks"
" WHERE id = ?", (cid,)).fetchone()
if row is None:
rows[cid] = {
"status": "interrupted",
"result_text": None,
"error_text": "子任务行已消失(已删除)"}
continue
try:
rows[row["id"]] = {
@@ -3027,9 +3033,19 @@ class Worker:
self.heartbeat(task_id)
except Exception:
pass
# 行消失防线:协调员本体行被删时 _was_cancelled 查不到
# 行会返回 False(get_task 为 None),不能因此等到
# deadline——行没了即按取消处理,立即跳出放槽;终局
# CAS 写不回去,最终状态按 interrupted 计(见循环后
# final_written 分支),不写回、不抛错、不当完成。
if self.get_task(task_id) is None:
cancelled = True
break
if self._was_cancelled(task_id):
cancelled = True
break
# 等齐判断:_coord_child_rows 已把行消失的子任务合成
# interrupted 终态,缺失不再拖住循环等到超时。
rows = self._coord_child_rows(child_ids)
if all((rows.get(cid) or {}).get("status") in TERMINAL_STATUSES
for cid in child_ids):