fix(orchestrator): worker被OOM击杀后ProcessPoolExecutor永久破损——BrokenProcessPool让任务中心瘫痪到容器重启(factor批次连撞两次500实锤);submit_work捕获后重建池重试一次 [nas]

sanguo_orchestrator/pool.py

@@ -4,6 +4,7 @@ Provides task storage and status tracking (not actual multiprocessing)
 """
 import logging
 from concurrent.futures import ProcessPoolExecutor, Future
+from concurrent.futures.process import BrokenProcessPool
 from multiprocessing import get_context
 from .task import Task, TaskState
 
@@ -28,9 +29,21 @@ class TaskPool:
         return task
 
     def submit_work(self, task_id: str, func, *args) -> Future:
-        """Submit work to the process pool executor"""
+        """Submit work to the process pool executor.
+
+        worker 被 OOM 击杀后 ProcessPoolExecutor 永久破损(BrokenProcessPool),
+        重建池重试一次,避免一次异常让任务中心瘫痪到容器重启。
+        """
         logger.debug("submit_work task_id=%s", task_id)
-        return self.executor.submit(func, *args)
+        try:
+            return self.executor.submit(func, *args)
+        except BrokenProcessPool:
+            logger.warning("pool broken (worker died?), rebuilding for %s", task_id)
+            self.executor.shutdown(wait=False)
+            self.executor = ProcessPoolExecutor(
+                max_workers=self.max_workers, mp_context=get_context("spawn")
+            )
+            return self.executor.submit(func, *args)
 
     def update_stage(self, task_id: str, stage: str):
         """Update the stage of a task"""