From afbf432eae21e964f9f871ff23b0bfbfaa98f204 Mon Sep 17 00:00:00 2001
From: shy3130 <415333856@qq.com>
Date: Wed, 26 Aug 2026 22:07:40 +0800
Subject: [PATCH] =?UTF-8?q?fix(backtest):=20=E5=AD=90=E8=BF=9B=E7=A8=8B?=
 =?UTF-8?q?=E9=80=80=E5=87=BA=E6=94=B6=E5=B0=BE=E8=B6=85=E6=97=B6=E4=B8=8D?=
 =?UTF-8?q?=E5=86=8D=E4=B8=A2=E5=BC=83=E5=B7=B2=E9=80=81=E8=BE=BE=E7=9A=84?=
 =?UTF-8?q?=E5=9B=9E=E6=B5=8B=E7=BB=93=E6=9E=9C?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

终态消息入队后显式冲刷队列并以 os._exit 立即退出, 跳过大数据量下
可达数十秒的解释器 teardown (GC/DuckDB 线程 join/DLL 卸载); 父进程
在子进程超时未退出时改为强杀并采纳已送达结果, 记录 worker_exit_forcibly
指标, 错误场景优先抛出 worker 真实异常。
---
 backend/app/backtest/worker.py                | 22 +++++++-
 backend/tests/backtest/test_worker_process.py | 56 +++++++++++++++++++
 2 files changed, 77 insertions(+), 1 deletion(-)

diff --git a/backend/app/backtest/worker.py b/backend/app/backtest/worker.py
index f4fb102..be1763f 100644
--- a/backend/app/backtest/worker.py
+++ b/backend/app/backtest/worker.py
@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import multiprocessing as mp
 import os
 import queue
@@ -17,6 +18,8 @@ from typing import Any
 
 import psutil
 
+logger = logging.getLogger(__name__)
+
 
 class BacktestWorkerError(RuntimeError):
     """Raised when a spawned worker fails before returning a task result."""
@@ -254,6 +257,13 @@ def _worker_entry(task: dict[str, Any], event_queue, cancel_event) -> None:
         if store is not None:
             with suppress(Exception):
                 store.db.close()
+        # 终态消息已入队: 显式冲刷队列后立即退出。大数据量任务跳过解释器
+        # teardown (GC、DuckDB 线程 join、DLL 卸载), 否则收尾可达数十秒,
+        # 会撞上父进程 10s 退出预算。close+join_thread 保证消息完整落管。
+        with suppress(Exception):
+            event_queue.close()
+            event_queue.join_thread()
+        os._exit(0)
 
 
 def run_worker_task(
@@ -311,10 +321,19 @@ def run_worker_task(
                 failure = message
 
         process.join(timeout=10.0)
+        worker_exit_forcibly = False
         if process.is_alive():
+            # 终态消息 (result/error) 已完整送达, 子进程只是退出收尾慢:
+            # 强制结束并继续走结果/错误处理, 不把已送达的成功结果当失败丢弃。
             process.terminate()
             process.join(timeout=5.0)
-            raise BacktestWorkerError("backtest worker returned but did not exit within 10 seconds")
+            worker_exit_forcibly = True
+            logger.warning(
+                "%s worker delivered its terminal message but did not exit within "
+                "10s; terminated forcibly (exitcode=%s)",
+                task["kind"],
+                process.exitcode,
+            )
         if failure is not None:
             raise BacktestWorkerError(
                 f"{failure.get('message', 'worker failed')}\n{failure.get('traceback', '')}".rstrip()
@@ -329,6 +348,7 @@ def run_worker_task(
             "parent_rss_before_bytes": parent_rss_before,
             "parent_rss_after_worker_exit_bytes": _rss_bytes(),
             "worker_exitcode": process.exitcode,
+            "worker_exit_forcibly": worker_exit_forcibly,
         }
         kind = task["kind"]
         if kind == "backtest":
diff --git a/backend/tests/backtest/test_worker_process.py b/backend/tests/backtest/test_worker_process.py
index d45e32b..e76193f 100644
--- a/backend/tests/backtest/test_worker_process.py
+++ b/backend/tests/backtest/test_worker_process.py
@@ -448,6 +448,62 @@ def test_worker_terminates_child_after_cancel_grace(monkeypatch, tmp_path):
     assert process.exitcode == -15
 
 
+def test_worker_accepts_delivered_result_when_child_exit_is_slow(monkeypatch, tmp_path):
+    """终态消息已送达但子进程退出收尾超时: 应强杀后采纳结果, 而非丢弃报错。"""
+
+    class FakeQueue:
+        def __init__(self):
+            self._messages = [{"type": "result", "payload": {"status": "ok"}}]
+
+        def get(self, timeout):
+            if self._messages:
+                return self._messages.pop(0)
+            raise queue.Empty
+
+        def close(self):
+            pass
+
+        def join_thread(self):
+            pass
+
+    class FakeEvent:
+        def set(self):
+            pass
+
+    class FakeProcess:
+        def __init__(self):
+            self.alive = True
+            self.exitcode = None
+
+        def start(self):
+            pass
+
+        def is_alive(self):
+            return self.alive
+
+        def join(self, timeout=None):
+            pass
+
+        def terminate(self):
+            self.alive = False
+            self.exitcode = -15
+
+    process = FakeProcess()
+    context = SimpleNamespace(
+        Queue=FakeQueue,
+        Event=FakeEvent,
+        Process=lambda **_kwargs: process,
+    )
+    monkeypatch.setattr(worker_module.mp, "get_context", lambda _method: context)
+
+    result = run_worker_task({"kind": "mining", "data_dir": str(tmp_path), "config": {}})
+
+    assert result["status"] == "ok"
+    assert result["worker"]["worker_exit_forcibly"] is True
+    assert result["worker"]["worker_exitcode"] == -15
+    assert process.exitcode == -15
+
+
 def test_spawn_walkforward_skips_folds_before_available_matrix_data(tmp_path):
     configured_start = date(2024, 1, 1)
     market_start = configured_start + timedelta(days=4)