diff --git a/cloud/tests/test_coordinator_row_vanished.py b/cloud/tests/test_coordinator_row_vanished.py new file mode 100644 index 0000000..06a165c --- /dev/null +++ b/cloud/tests/test_coordinator_row_vanished.py @@ -0,0 +1,204 @@ +"""协调员行消失单测：本体行消失与子任务行消失都不等到超时。 + +不跑真实模型回合、不联网：worker 侧用临时 SQLite + 假 runner， +等待中消失用“线程里跑协调员、主线程删行、join 带超时断言” +的办法——不用“删行后直接 run_one”（run_one 先认领，行已没了 +根本认领不上），也不靠 sleep 时长赌竞态：删行后下一轮轮询 +（poll_interval=0.05 秒）必然看到消失，join 超时即为失败信号。 +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import time +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.server import init_db + from cloud.worker import Worker +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.server import init_db + from cloud.worker import Worker + + +def _completed_runner(cmd): + return SimpleNamespace(returncode=0, stdout="{}", stderr="") + + +def _insert_task(db_path, task_id, prompt="do work", status="queued", + project_id="p1", coordinator_parts=None, + parent_task_id=None, coordinator_part=None, + result_text=None, error_text=None): + now = time.time() + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat, coordinator_parts," + " parent_task_id, coordinator_part, result_text, error_text)" + " VALUES(?, ?, ?, ?, NULL, '[]', 0, 'tester', ?, ?, NULL," + " ?, ?, ?, ?, ?)", + (task_id, project_id, prompt, status, now, now, + json.dumps(coordinator_parts) if coordinator_parts else None, + parent_task_id, coordinator_part, result_text, error_text), + ) + conn.commit() + finally: + conn.close() + + +def _row(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + row = conn.execute( + "SELECT id, status, result_text, error_text FROM tasks" + " WHERE id = ?", (task_id,)).fetchone() + finally: + conn.close() + if row is None: + return None + return {"id": row[0], "status": row[1], "result_text": row[2], + "error_text": row[3]} + + +def _delete_row(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,)) + conn.commit() + finally: + conn.close() + + +def _set_running(db_path, task_id): + conn = sqlite3.connect(db_path) + try: + conn.execute( + "UPDATE tasks SET status = 'running' WHERE id = ?", + (task_id,)) + conn.commit() + finally: + conn.close() + + +class RowVanishedTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.env = {"XCM_API_KEY": "test-placeholder"} + + def tearDown(self): + self.tmp.cleanup() + + def _worker(self): + return Worker(self.db_path, "/fake/host.py", + runner=_completed_runner, env=self.env) + + def _parts(self): + return [{"name": "后端", "prompt": "做后端"}, + {"name": "前端", "prompt": "做前端"}] + + def _run_in_thread(self, worker, task, timeout): + """线程里跑 _run_coordinator，返回 (结果, 耗时)；join 超时即失败。""" + box = {} + + def _target(): + started = time.time() + box["result"] = worker._run_coordinator( + task, time.time(), poll_interval=0.05, timeout=timeout) + box["elapsed"] = time.time() - started + + thread = threading.Thread(target=_target) + thread.start() + return thread, box + + def test_parent_row_vanished_exits_as_interrupted(self): + """(a) 协调员本体行在等待中消失 → 远小于 timeout 退出， + 返回 interrupted，不是 coordinator timeout 的 failed。""" + _insert_task(self.db_path, "coord", + coordinator_parts=self._parts()) + _insert_task(self.db_path, "c1", status="completed", + parent_task_id="coord", coordinator_part="后端", + result_text="后端好了") + _insert_task(self.db_path, "c2", status="queued", + parent_task_id="coord", coordinator_part="前端") + _set_running(self.db_path, "coord") + worker = self._worker() + task = worker.get_task("coord") + thread, box = self._run_in_thread(worker, task, timeout=60) + time.sleep(0.3) + _delete_row(self.db_path, "coord") + thread.join(timeout=10) + self.assertFalse(thread.is_alive(), + "本体行消失后协调员循环应在 10 秒内退出") + self.assertLess(box["elapsed"], 10) + self.assertEqual(box["result"]["status"], "interrupted") + self.assertNotIn("timeout", box["result"]["error"] or "") + # 行已没了，终局写回写不回去，也不抛错、不当完成。 + self.assertIsNone(_row(self.db_path, "coord")) + + def test_child_row_vanished_counts_as_interrupted(self): + """(b) 子任务行在等待中消失 → 合流报告里该子任务为 + interrupted 且注明行已消失，整体不为 completed，其它 + 子任务终态不受影响，且不拖到 timeout。""" + _insert_task(self.db_path, "coord", + coordinator_parts=self._parts()) + _insert_task(self.db_path, "c1", status="completed", + parent_task_id="coord", coordinator_part="后端", + result_text="后端好了") + _insert_task(self.db_path, "c2", status="queued", + parent_task_id="coord", coordinator_part="前端") + _set_running(self.db_path, "coord") + worker = self._worker() + task = worker.get_task("coord") + thread, box = self._run_in_thread(worker, task, timeout=60) + time.sleep(0.3) + _delete_row(self.db_path, "c2") + thread.join(timeout=10) + self.assertFalse(thread.is_alive(), + "子任务行消失后协调员应在 10 秒内合流退出") + self.assertLess(box["elapsed"], 10) + result = box["result"] + self.assertNotEqual(result["status"], "completed") + self.assertNotIn("timeout", result["error"] or "") + self.assertIn("前端", result["error"] or "") + self.assertIn("(interrupted)", result["stdout"]) + self.assertIn("行已消失", result["stdout"]) + self.assertIn("后端好了", result["stdout"]) + # 其它子任务不受影响：c1 仍是 completed，结果 intact。 + c1 = _row(self.db_path, "c1") + self.assertEqual(c1["status"], "completed") + self.assertEqual(c1["result_text"], "后端好了") + # 协调员本体行仍在，且因有子任务未完成而整体 failed。 + self.assertEqual(_row(self.db_path, "coord")["status"], "failed") + + def test_all_completed_still_passes(self): + """(c) 正常等齐仍通过：行消失防线不改变正常合流语义。""" + _insert_task(self.db_path, "coord", + coordinator_parts=self._parts()) + _insert_task(self.db_path, "c1", status="completed", + parent_task_id="coord", coordinator_part="后端", + result_text="后端好了") + _insert_task(self.db_path, "c2", status="completed", + parent_task_id="coord", coordinator_part="前端", + result_text="前端好了") + _set_running(self.db_path, "coord") + worker = self._worker() + result = worker._run_coordinator( + worker.get_task("coord"), time.time(), + poll_interval=0.01, timeout=60) + self.assertEqual(result["status"], "completed") + self.assertIn("2/2", result["stdout"]) + self.assertEqual(_row(self.db_path, "coord")["status"], "completed") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 0226e42..6cb3b85 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -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):