Merge pull request #118 from openqj/feat-coord-running-leak

worker: release running slot after coordinator completes

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)

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