fix(enriched): 自愈孤儿 publishing 标记, 步进优化不再误报 publishing

清库半删除等异常会留下 owner 已死但状态仍为 publishing 的 generation
标记, 后续读取直接抛 EnrichedGenerationUnavailableError, 步进优化连锁失败。

- 读取侧仅在「owner 是其他进程且该进程已死」时判孤儿并在独占锁内
  自愈 (换新 uuid 重建 ready 标记); 同 pid 无活跃对象时保守不判,
  避免暴露清库半删除的数据 (fail-closed)
- 回测引擎改为 data_generation_await 轮询等待 (300s 上限), 遇短暂
  publishing 自动重试而非直接失败; worker 错误文案同步为可重试提示

backend/app/backtest/engine.py

@@ -300,6 +300,14 @@ class PanelCache:
         return f"{asset_type}:{generation or 'unmanaged'}:{h}:{start}:{end}:{cols}"
 
 
+# 等待进行中 enriched 发布的上限与轮询间隔。孤儿标记由 get_enriched_generation
+# 在读取时直接自愈, 因此这里等到的 EnrichedGenerationUnavailableError 意味着
+# 发布方确实存活 —— 对回测/优化这类长任务, 有界等待优于立即失败。仅用于
+# worker 任务路径 (矩阵加载), 实时热路径不得调用 data_generation_await。
+_GENERATION_WAIT_TIMEOUT_S = 300.0
+_GENERATION_POLL_S = 1.0
+
+
 # ================================================================
 # BacktestEngine
 # ================================================================
@@ -317,6 +325,25 @@ class BacktestEngine:
         loader = getattr(self.repo, "get_matrix_data_generation", None)
         return loader(asset_type) if callable(loader) else None
 
+    def data_generation_await(
+        self,
+        asset_type: str = "stock",
+        *,
+        cancel_event: threading.Event | None = None,
+        timeout_s: float = _GENERATION_WAIT_TIMEOUT_S,
+    ) -> str | None:
+        """获取 generation; 发布进行中时在超时窗口内轮询, 可被取消事件打断。"""
+        deadline = time.monotonic() + timeout_s
+        while True:
+            try:
+                return self.data_generation(asset_type)
+            except EnrichedGenerationUnavailableError:
+                if cancel_event is not None and cancel_event.is_set():
+                    raise
+                if time.monotonic() >= deadline:
+                    raise
+                time.sleep(_GENERATION_POLL_S)
+
     def assert_data_generation(
         self,
         asset_type: str,
@@ -506,15 +533,10 @@ class BacktestEngine:
             if cache_profile is not None
             else settings.backtest_matrix_cache_max_mb * 1024 * 1024
         )
-        generation_loader = getattr(self.repo, "get_matrix_data_generation", None)
         source_generation = (
             expected_generation
             if expected_generation is not None
-            else (
-                generation_loader(asset_type)
-                if callable(generation_loader)
-                else None
-            )
+            else self.data_generation_await(asset_type, cancel_event=cancel_event)
         )
         attempts = 1 if expected_generation is not None else 2
         for attempt in range(attempts):
@@ -555,7 +577,9 @@ class BacktestEngine:
             except EnrichedGenerationUnavailableError:
                 if attempt + 1 >= attempts:
                     raise
-                source_generation = self.data_generation(asset_type)
+                source_generation = self.data_generation_await(
+                    asset_type, cancel_event=cancel_event
+                )
             except pa.ArrowException as exc:
                 raise ValueError(f"direct market matrix parquet scan failed: {exc}") from exc
         raise EnrichedGenerationUnavailableError(

backend/app/backtest/worker.py

@@ -166,6 +166,15 @@ def _attach_worker_metrics(
         result["worker"] = metrics
 
 
+def _error_message(exc: BaseException) -> str:
+    """任务级错误文案: enriched 发布类失败对用户是"稍后再试", 不透出原始异常。"""
+    from app.enriched_generation import EnrichedGenerationUnavailableError
+
+    if isinstance(exc, EnrichedGenerationUnavailableError):
+        return "指标数据正在发布更新，请稍后重试"
+    return str(exc)
+
+
 def _worker_entry(task: dict[str, Any], event_queue, cancel_event) -> None:
     sampler = _PeakRssSampler()
     sampler.start()
@@ -256,7 +265,7 @@ def _worker_entry(task: dict[str, Any], event_queue, cancel_event) -> None:
             sampler.stop()
         event_queue.put({
             "type": "error",
-            "message": str(exc),
+            "message": _error_message(exc),
             "traceback": traceback.format_exc(),
         })
     finally:

backend/app/enriched_generation.py

@@ -154,6 +154,45 @@ def _ready_payload(generation: str) -> dict[str, Any]:
     }
 
 
+def _is_ready_payload(payload: dict[str, Any]) -> bool:
+    generation = payload.get("generation")
+    return (
+        payload.get("state", "ready") == "ready"
+        and isinstance(generation, str)
+        and bool(generation)
+    )
+
+
+def _publication_claim_is_running(payload: dict[str, Any]) -> bool:
+    """标记指向的发布是否仍在推进: 进程内活跃对象存在, 或属主进程仍存活。
+
+    owner_pid 等于当前进程但无活跃对象视为可接管 (同进程上一次尝试的遗留),
+    与写入方 recover 接管的判定一致。
+    """
+    if _ACTIVE_PUBLICATIONS.get(str(payload.get("publication_id"))) is not None:
+        return True
+    owner_pid = payload.get("owner_pid")
+    return owner_pid != os.getpid() and _process_is_alive(owner_pid)
+
+
+def _orphaned_publishing_claim(payload: dict[str, Any]) -> bool:
+    """标记是否指向确定已死的发布: 属主是其他进程且已退出。
+
+    owner_pid 等于当前进程但无活跃对象时保守不判孤儿 —— 同进程异常遗留的
+    publishing 标记意味着磁盘可能处于部分修改状态 (如清库删了一半), 读取方
+    恢复 ready 会放行读取半修改数据; 必须由下一个写入方接管重发布。
+    """
+    if _ACTIVE_PUBLICATIONS.get(str(payload.get("publication_id"))) is not None:
+        return False
+    owner_pid = payload.get("owner_pid")
+    return (
+        isinstance(owner_pid, int)
+        and owner_pid > 0
+        and owner_pid != os.getpid()
+        and not _process_is_alive(owner_pid)
+    )
+
+
 def get_enriched_generation(
     data_dir: Path,
     asset_type: str = "stock",
@@ -167,19 +206,33 @@ def get_enriched_generation(
             raise EnrichedGenerationUnavailableError(
                 "enriched data generation marker is unavailable"
             )
-        with _exclusive_generation_lock(data_dir, asset_type):
-            payload = _read_marker(path)
-            if payload is None:
-                generation = uuid.uuid4().hex
-                _write_marker(path, _ready_payload(generation))
-                return generation
-    state = payload.get("state", "ready")
-    generation = payload.get("generation")
-    if state != "ready" or not isinstance(generation, str) or not generation:
+    elif _is_ready_payload(payload):
+        return payload["generation"]
+    elif not _orphaned_publishing_claim(payload):
+        # 发布仍在推进, 或为同进程异常遗留 (无法证明属主已死): 读取保持 fail-closed。
         raise EnrichedGenerationUnavailableError(
             "enriched data is being published; retry after the update finishes"
         )
-    return generation
+    # 指向已死发布的僵死标记: 在独占锁内二次确认后恢复 ready。
+    with _exclusive_generation_lock(data_dir, asset_type):
+        payload = _read_marker(path)
+        if payload is None:
+            generation = uuid.uuid4().hex
+            _write_marker(path, _ready_payload(generation))
+            return generation
+        if _is_ready_payload(payload):
+            return payload["generation"]
+        if not _orphaned_publishing_claim(payload):
+            raise EnrichedGenerationUnavailableError(
+                "enriched data is being published; retry after the update finishes"
+            )
+        # 属主已死的 publishing 标记永远不会 commit, 读取方持续失败直到某个
+        # 写入方碰巧接管 (dev 热重载杀掉发布进程即产生这种孤儿)。恢复为 ready
+        # 并换新 generation: 磁盘可能残留部分替换的文件, 新 generation 让按代
+        # 缓存全部失效, 避免把混合状态混入旧快照 —— 与写入方 recover 接管同语义。
+        generation = uuid.uuid4().hex
+        _write_marker(path, _ready_payload(generation))
+        return generation
 
 
 def enriched_publication_incomplete(
@@ -296,12 +349,7 @@ class EnrichedPublication:
             return
         _ACTIVE_PUBLICATIONS[self._publication_id] = self
         if current is not None and current.get("state", "ready") != "ready":
-            current_id = current.get("publication_id")
-            current_owner = _ACTIVE_PUBLICATIONS.get(str(current_id))
-            owner_pid = current.get("owner_pid")
-            if current_owner is not None or (
-                owner_pid != os.getpid() and _process_is_alive(owner_pid)
-            ):
+            if _publication_claim_is_running(current):
                 raise EnrichedGenerationUnavailableError(
                     "another enriched publication is active"
                 )

backend/tests/test_enriched_generation.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import json
+import os
+import threading
 from datetime import date
 from types import SimpleNamespace
 
@@ -225,3 +227,120 @@ def test_live_flush_write_recovers_stale_marker_from_dead_process(
         (tmp_path / ".matrix_generation_stock.json").read_text(encoding="utf-8")
     )
     assert marker["state"] == "ready"
+
+
+def _stale_publishing_marker(tmp_path, owner_pid: int = 999999999) -> None:
+    (tmp_path / ".matrix_generation_stock.json").write_text(
+        json.dumps({
+            "state": "publishing",
+            "generation": "stale-generation",
+            "publication_id": "stale-publication",
+            "owner_pid": owner_pid,
+            "updated_at_ns": 0,
+        }),
+        encoding="utf-8",
+    )
+
+
+def test_reader_self_heals_stale_publishing_marker_from_dead_owner(tmp_path) -> None:
+    """读取方遇到属主已死的 publishing 标记应就地恢复 ready, 而非持续失败
+    直到某个写入方碰巧接管 (dev 热重载杀掉发布进程即产生这种孤儿)。"""
+    _stale_publishing_marker(tmp_path)
+
+    generation = get_enriched_generation(tmp_path, "stock")
+
+    marker = json.loads(
+        (tmp_path / ".matrix_generation_stock.json").read_text(encoding="utf-8")
+    )
+    assert marker["state"] == "ready"
+    # 恢复时换新 generation: 磁盘可能残留部分替换的文件, 按代缓存需要失效。
+    assert generation not in ("", "stale-generation")
+    assert generation == marker["generation"]
+    assert get_enriched_generation(tmp_path, "stock") == generation
+
+
+def test_reader_still_fails_closed_while_owner_is_alive(
+    tmp_path, monkeypatch
+) -> None:
+    monkeypatch.setattr("app.enriched_generation._process_is_alive", lambda pid: True)
+    _stale_publishing_marker(tmp_path, owner_pid=os.getpid() + 1)
+
+    with pytest.raises(EnrichedGenerationUnavailableError, match="being published"):
+        get_enriched_generation(tmp_path, "stock")
+    marker = json.loads(
+        (tmp_path / ".matrix_generation_stock.json").read_text(encoding="utf-8")
+    )
+    assert marker["state"] == "publishing"  # 标记未被读取方改动
+
+
+def test_reader_respects_active_in_process_publication(tmp_path) -> None:
+    publication = EnrichedPublication(tmp_path, recover=True)
+    publication.begin()
+    try:
+        with pytest.raises(EnrichedGenerationUnavailableError, match="being published"):
+            get_enriched_generation(tmp_path, "stock")
+    finally:
+        publication.abandon()
+
+    assert _is_ready_marker(tmp_path)
+
+
+def _is_ready_marker(tmp_path) -> bool:
+    marker = json.loads(
+        (tmp_path / ".matrix_generation_stock.json").read_text(encoding="utf-8")
+    )
+    return marker["state"] == "ready"
+
+
+def test_data_generation_await_retries_during_publication(tmp_path, monkeypatch) -> None:
+    engine = BacktestEngine(KlineRepository(DataStore(tmp_path)))
+    calls = {"n": 0}
+
+    def flaky(asset_type: str = "stock"):
+        calls["n"] += 1
+        if calls["n"] < 3:
+            raise EnrichedGenerationUnavailableError(
+                "enriched data is being published; retry after the update finishes"
+            )
+        return "gen-final"
+
+    monkeypatch.setattr(engine, "data_generation", flaky)
+    monkeypatch.setattr("app.backtest.engine._GENERATION_POLL_S", 0.001)
+
+    assert engine.data_generation_await("stock") == "gen-final"
+    assert calls["n"] == 3
+
+
+def test_data_generation_await_times_out_and_respects_cancel(
+    tmp_path, monkeypatch
+) -> None:
+    engine = BacktestEngine(KlineRepository(DataStore(tmp_path)))
+    monkeypatch.setattr("app.backtest.engine._GENERATION_POLL_S", 0.001)
+
+    def always_publishing(asset_type: str = "stock"):
+        raise EnrichedGenerationUnavailableError(
+            "enriched data is being published; retry after the update finishes"
+        )
+
+    monkeypatch.setattr(engine, "data_generation", always_publishing)
+
+    with pytest.raises(EnrichedGenerationUnavailableError):
+        engine.data_generation_await("stock", timeout_s=0.01)
+
+    cancel = threading.Event()
+    cancel.set()
+    with pytest.raises(EnrichedGenerationUnavailableError):
+        engine.data_generation_await("stock", cancel_event=cancel, timeout_s=30.0)
+
+
+def test_worker_error_message_translates_publishing_error() -> None:
+    from app.backtest.worker import _error_message
+    from app.enriched_generation import EnrichedGenerationUnavailableError
+
+    translated = _error_message(
+        EnrichedGenerationUnavailableError(
+            "enriched data is being published; retry after the update finishes"
+        )
+    )
+    assert translated == "指标数据正在发布更新，请稍后重试"
+    assert _error_message(ValueError("boom")) == "boom"