diff --git a/backend/app/api/data.py b/backend/app/api/data.py index 3526d7f..24e12a6 100644 --- a/backend/app/api/data.py +++ b/backend/app/api/data.py @@ -622,9 +622,12 @@ def clear_data(request: Request): repo = request.app.state.repo data_dir = repo.store.data_dir deleted = 0 + # recover=True: 清空语义就是接管一切 — 外部进程(崩掉的脚本/中断的管道) + # 残留的 publishing 标记(owner pid 已死)不应永久阻塞清空; 活进程的发布 + # 仍会被拦(another publication is active), 写盘竞态保护不受影响。 publications = { - "kline_daily_enriched": EnrichedPublication(data_dir, "stock"), - "kline_etf_enriched": EnrichedPublication(data_dir, "etf"), + "kline_daily_enriched": EnrichedPublication(data_dir, "stock", recover=True), + "kline_etf_enriched": EnrichedPublication(data_dir, "etf", recover=True), } for sub in ( diff --git a/backend/app/tickflow/repository.py b/backend/app/tickflow/repository.py index c97c4f0..64f89b7 100644 --- a/backend/app/tickflow/repository.py +++ b/backend/app/tickflow/repository.py @@ -2107,8 +2107,10 @@ class KlineRepository: "kline_daily_enriched": "stock", "kline_etf_enriched": "etf", }.get(table) + # recover=True: 外部进程残留的僵死 publishing 标记不应阻塞实时/管道 + # enriched 落盘, 首次写入即接管自愈; 活进程的发布仍会抛错保护竞态。 publication = ( - EnrichedPublication(self.store.data_dir, generation_asset) + EnrichedPublication(self.store.data_dir, generation_asset, recover=True) if generation_asset is not None else None ) @@ -2210,7 +2212,7 @@ class KlineRepository: out = base / f"date={ds}" / "part.parquet" out.parent.mkdir(parents=True, exist_ok=True) publication = ( - EnrichedPublication(self.store.data_dir, asset_type) + EnrichedPublication(self.store.data_dir, asset_type, recover=True) if asset_type in {"stock", "etf"} else None ) @@ -2295,7 +2297,7 @@ class KlineRepository: out = base / f"date={ds}" / "part.parquet" out.parent.mkdir(parents=True, exist_ok=True) publication = ( - EnrichedPublication(self.store.data_dir, asset_type) + EnrichedPublication(self.store.data_dir, asset_type, recover=True) if asset_type in {"stock", "etf"} else None ) diff --git a/backend/tests/test_data_clear_generation.py b/backend/tests/test_data_clear_generation.py index 6d0096d..ec0394e 100644 --- a/backend/tests/test_data_clear_generation.py +++ b/backend/tests/test_data_clear_generation.py @@ -152,3 +152,36 @@ def test_clear_data_keeps_generation_publishing_after_partial_delete( assert marker["state"] == "publishing" with pytest.raises(EnrichedGenerationUnavailableError, match="being published"): get_enriched_generation(tmp_path, "stock") + + +def test_clear_data_recovers_stale_publishing_marker_from_dead_process( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """外部进程(崩掉的脚本/中断的管道)残留 publishing 标记且 owner pid 已死时, + 清空不应 500 — clear 语义就是接管一切 (用户实测场景: POST /api/data/clear + 报 another enriched publication is incomplete)。""" + _stub_clear_side_effects(monkeypatch) + repo = _RepoStub(tmp_path) + target = tmp_path / "kline_daily_enriched" / "date=2026-08-14" / "part.parquet" + _write_parquet_placeholder(target) + (tmp_path / ".matrix_generation_stock.json").write_text( + json.dumps({ + "state": "publishing", + "generation": "stale-generation", + "publication_id": "stale-publication", + "owner_pid": 999999999, # 不存在的 pid → 探测判定已死 + "updated_at_ns": 0, + }), + encoding="utf-8", + ) + + result = data_api.clear_data(_request(repo)) + + assert result == {"deleted_files": 1} + assert not target.exists() + marker = json.loads( + (tmp_path / ".matrix_generation_stock.json").read_text(encoding="utf-8") + ) + assert marker["state"] == "ready" + assert get_enriched_generation(tmp_path, "stock") == marker["generation"] diff --git a/backend/tests/test_enriched_generation.py b/backend/tests/test_enriched_generation.py index f9b9130..d5d6f0c 100644 --- a/backend/tests/test_enriched_generation.py +++ b/backend/tests/test_enriched_generation.py @@ -198,3 +198,30 @@ def test_matrix_reader_retries_when_generation_changes_during_build( assert result is market assert calls == ["generation-a", "generation-b"] + + +def test_live_flush_write_recovers_stale_marker_from_dead_process( + tmp_path, monkeypatch +) -> None: + """实时 enriched 落盘(repository 路径)遇到僵死 publishing 标记应接管自愈, + 而非持续抛错直到下一次盘后管道。""" + from app.tickflow.repository import DataStore, KlineRepository + + (tmp_path / ".matrix_generation_stock.json").write_text( + json.dumps({ + "state": "publishing", + "generation": "stale-generation", + "publication_id": "stale-publication", + "owner_pid": 999999999, + "updated_at_ns": 0, + }), + encoding="utf-8", + ) + + repo = KlineRepository(DataStore(tmp_path)) + repo.append_enriched(_frame(10.0)) + + marker = json.loads( + (tmp_path / ".matrix_generation_stock.json").read_text(encoding="utf-8") + ) + assert marker["state"] == "ready"