From 58b161b16db8bd4a2572457020d697e8ecbeaad8 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Mon, 7 Sep 2026 15:25:27 +0800 Subject: [PATCH] =?UTF-8?q?fix(enriched):=20=E8=87=AA=E6=84=88=E5=AD=A4?= =?UTF-8?q?=E5=84=BF=20publishing=20=E6=A0=87=E8=AE=B0,=20=E6=AD=A5?= =?UTF-8?q?=E8=BF=9B=E4=BC=98=E5=8C=96=E4=B8=8D=E5=86=8D=E8=AF=AF=E6=8A=A5?= =?UTF-8?q?=20publishing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 清库半删除等异常会留下 owner 已死但状态仍为 publishing 的 generation 标记, 后续读取直接抛 EnrichedGenerationUnavailableError, 步进优化连锁失败。 - 读取侧仅在「owner 是其他进程且该进程已死」时判孤儿并在独占锁内 自愈 (换新 uuid 重建 ready 标记); 同 pid 无活跃对象时保守不判, 避免暴露清库半删除的数据 (fail-closed) - 回测引擎改为 data_generation_await 轮询等待 (300s 上限), 遇短暂 publishing 自动重试而非直接失败; worker 错误文案同步为可重试提示 --- backend/app/backtest/engine.py | 38 +++++-- backend/app/backtest/worker.py | 11 +- backend/app/enriched_generation.py | 80 ++++++++++++--- backend/tests/test_enriched_generation.py | 119 ++++++++++++++++++++++ 4 files changed, 224 insertions(+), 24 deletions(-) diff --git a/backend/app/backtest/engine.py b/backend/app/backtest/engine.py index 6501fff..5834e5e 100644 --- a/backend/app/backtest/engine.py +++ b/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( diff --git a/backend/app/backtest/worker.py b/backend/app/backtest/worker.py index e9a2225..4931038 100644 --- a/backend/app/backtest/worker.py +++ b/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: diff --git a/backend/app/enriched_generation.py b/backend/app/enriched_generation.py index 66b6b11..9933a44 100644 --- a/backend/app/enriched_generation.py +++ b/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" ) diff --git a/backend/tests/test_enriched_generation.py b/backend/tests/test_enriched_generation.py index d5d6f0c..b5594ef 100644 --- a/backend/tests/test_enriched_generation.py +++ b/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"