fix(backtest): 子进程退出收尾超时不再丢弃已送达的回测结果

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

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":

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)