diff --git a/cloud/tests/test_coordinator.py b/cloud/tests/test_coordinator.py index d20a8e5..f378750 100644 --- a/cloud/tests/test_coordinator.py +++ b/cloud/tests/test_coordinator.py @@ -270,6 +270,60 @@ class WorkerCoordinatorTestCase(unittest.TestCase): self.assertIn("后端好了", result["stdout"]) self.assertEqual(_row(self.db_path, "coord")["status"], "completed") + def test_coordinator_completion_releases_running_slot(self): + # 回归:协调员不走 _run_attempt 的 finally,旧版完成 + # 后 _running 占位永不释放,同进程 worker 的有效容量 + # 逐个协调员递减直至再也认领不了(2026-10-28 40 人 + # 与 18 人旧轮的第 4 槽/C-6 未认领即此漏)。 + _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="前端好了") + worker = self._worker() + result = worker.run_one(worker.get_task("coord")) + self.assertEqual(result["status"], "completed") + self.assertNotIn("coord", worker._running_snapshot()) + self.assertEqual(worker._running_count(), 0) + + def test_coordinator_then_next_claim_succeeds(self): + # 容量门视角的同一回归:协调员完成后,下一个任务 + # 的 _claim 必须成功,而不是被漏掉的占位静默挡住。 + _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="前端好了") + _insert_task(self.db_path, "next1") + worker = self._worker() + worker.run_one(worker.get_task("coord")) + self.assertTrue( + worker._claim("next1", executor=worker._executor_name())) + self.assertIn("next1", worker._running_snapshot()) + worker._running_clear("next1") + + def test_capacity_refused_log_throttled(self): + # 容量门被拦以前是静默返回;修法后 30 秒内只记一次, + # 避免池线程自旋把日志打爆(这里断言时间戳节流, + # 不捕获日志输出)。 + _insert_task(self.db_path, "t1") + _insert_task(self.db_path, "t2") + _insert_task(self.db_path, "t3") + worker = self._worker() + self.assertTrue( + worker._claim("t1", executor=worker._executor_name())) + self.assertFalse( + worker._claim("t2", executor=worker._executor_name())) + first = worker._capacity_refused_logged_at + self.assertGreater(first, 0.0) + self.assertFalse( + worker._claim("t3", executor=worker._executor_name())) + self.assertEqual(worker._capacity_refused_logged_at, first) + def test_run_timeout_cancels_pending(self): _insert_task(self.db_path, "coord", coordinator_parts=self._parts()) conn = sqlite3.connect(self.db_path) diff --git a/cloud/worker.py b/cloud/worker.py index cedb4c9..5f9d66a 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -468,6 +468,10 @@ class Worker: # 都走下面的 _running_set/_running_clear/_running_snapshot。 self._running = {} self._running_lock = threading.Lock() + # 容量门被拦日志的节流时间戳:_claim 的容量门在池线程 + # 自旋时会高频触发,不节流会把日志打爆;30 秒内只记 + # 一次,够排查“worker 活着却再也认领不了”即可。 + self._capacity_refused_logged_at = 0.0 # DB 写锁:cost 记录共用一条连接(CostStore._conn), # usage 计数也是共享变量;模型回合本身在子进程/容器里 # 跑不受这把锁影响,锁只覆盖毫秒级的写库操作。 @@ -724,12 +728,36 @@ class Worker: message="code signature missing or changed," " refusing claim") return False + capacity_refused = False + capacity_log = None with self._running_lock: already_running = task_id in self._running if not already_running and \ len(self._running) >= effective_capacity(self.capacity): - return False - self._running.setdefault(task_id, {}) + # 容量门被拦以前是静默返回:worker 活着、心跳 + # 新鲜、日志无错,却再也认领不了任务(协调员 + # 槽位泄漏就是这么被掩盖的)。注意:节流的只 + # 是日志,拒绝本身恒返回 False——若把节流套在 + # 拒绝上,节流窗口内的任务会被 CAS 认领却不进 + # _running,比静默更糟。 + capacity_refused = True + now_gate = time.time() + last_gate = getattr( + self, "_capacity_refused_logged_at", 0.0) + if now_gate - last_gate >= 30.0: + self._capacity_refused_logged_at = now_gate + capacity_log = ( + len(self._running), + effective_capacity(self.capacity)) + else: + self._running.setdefault(task_id, {}) + if capacity_log is not None: + log_event(self.logger, "worker_claim_refused_capacity", + task_id=task_id, + message="worker claim refused: capacity full," + " running=%d capacity=%d" % capacity_log) + if capacity_refused: + return False now = time.time() conn = _db_connect(self.db_path) try: @@ -3436,6 +3464,8 @@ class Worker: 占一个 worker 槽位直到子任务全部终态:等齐期间每轮刷心跳 (sweep 不会误回收);子任务本身由各 worker 并行认领。 + 占位由调用方 run_one 的 finally 在终局后释放,本函数 + 不重复清(旧版直接 return 导致占位永不释放,见上)。 取消与超时都会尽力取消未终态的子任务,不留孤儿继续烧 模型费用。返回 run_one 同款结果 dict(无模型花费, cost_record_id 为 None)。 @@ -3823,8 +3853,17 @@ class Worker: # 合流汇总。已过 key 门与配额门;仓库门已对协调员任务本身 # 跑过(项目仓库解析失败会在前面直接 failed,不会建一堆 # 注定失败的子任务);子任务各自再过 key/配额/仓库门。 + # 槽位清理:普通任务在 _run_attempt 的 finally 里清 + # _running 占位,协调员不走 _run_attempt——旧版这里直接 + # return _run_coordinator,协调员完成后续占位永不释放, + # 同进程 worker 的有效容量会逐个协调员递减直至再也 + # 认领不了(2026-10-28 40 人与 18 人旧轮的第 4 槽/C-6 + # 未认领即此漏)。finally 清槽,异常路径同样不漏。 if task.get("coordinator_parts"): - return self._run_coordinator(task, started_at) + try: + return self._run_coordinator(task, started_at) + finally: + self._running_clear(task_id) events = [] fallback_count = 0