From b8fa08f027ef05d285043df410cd16496b6d824f Mon Sep 17 00:00:00 2001
From: shy3130 <415333856@qq.com>
Date: Sun, 30 Aug 2026 19:05:19 +0800
Subject: [PATCH] =?UTF-8?q?fix(worker):=20=E6=B6=88=E9=99=A4=E5=9B=9E?=
 =?UTF-8?q?=E6=B5=8B=E5=AD=90=E8=BF=9B=E7=A8=8B=E7=BB=93=E6=9E=9C=E6=B6=88?=
 =?UTF-8?q?=E6=81=AF=E4=B8=A2=E5=A4=B1=E7=AB=9E=E6=80=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

现象: 整年区间回测偶发 'backtest worker exited without result (exitcode=0)'。
根因: 子进程 event_queue.put 只是把消息交给后台 feeder 线程, 主线程随即退出,
feeder 随进程销毁, 大结果/高负载下消息尾部未刷入管道; 父进程 0.1s 轮询
恰在 Empty+进程已死时跳出循环, 未读到的 result 被丢弃。

修复 (双侧):
- 子进程 finally 中 close+join_thread 队列, 保证退出前消息完整刷入管道
- 父进程在判定无结果前做一次兜底排空 (join 后 1s×2 轮 get)
---
 backend/app/backtest/worker.py | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/backend/app/backtest/worker.py b/backend/app/backtest/worker.py
index f4fb102..a84c295 100644
--- a/backend/app/backtest/worker.py
+++ b/backend/app/backtest/worker.py
@@ -254,6 +254,12 @@ def _worker_entry(task: dict[str, Any], event_queue, cancel_event) -> None:
         if store is not None:
             with suppress(Exception):
                 store.db.close()
+        # 保证结果消息在进程退出前完整刷入管道: put 只是入队,
+        # 实际写管道的是后台 feeder 线程; 不 join 的话主线程先退出,
+        # feeder 随进程销毁, 消息尾部丢失 → 父进程误判 "exited without result"。
+        with suppress(Exception):
+            event_queue.close()
+            event_queue.join_thread()
 
 
 def run_worker_task(
@@ -310,6 +316,21 @@ def run_worker_task(
             elif message_type == "error":
                 failure = message
 
+        # 子进程退出后, 队列读线程可能尚未把管道尾部的 result/error 搬进本地缓冲
+        # (0.1s 轮询在系统高负载下会先看到 Empty+进程已死)。join 后做一次兜底排空,
+        # 只要消息完整刷入过管道就一定能取到。
+        if result is None and failure is None:
+            for _ in range(2):
+                try:
+                    message = events.get(timeout=1.0)
+                except queue.Empty:
+                    break
+                message_type = message.get("type")
+                if message_type == "result":
+                    result = message["payload"]
+                elif message_type == "error":
+                    failure = message
+
         process.join(timeout=10.0)
         if process.is_alive():
             process.terminate()