fix(enriched): commit() 幂等, 多日期分区写入不再误报 ownership was lost

EnrichedPublication.commit() 提交后 _changed 未复位, 且标记被写成 ready
payload (不含 publication_id)、_publishing=False。_write_daily_partition
对整张表只创建一个发布对象并在 partition_by("date") 循环里复用, 于是
"有变化的分区" 之后若跟着一个 "内容无变化的分区", 后者在
_write_partition_locked 里直接 commit() 必然命中 publication_id 不匹配,
抛 enriched publication ownership was lost 并打断整批写入。

e3baba6 起 ETF enriched 按完整本地历史重算前复权, 一次写 245+ 个 date=
分区, 必然出现无变化分区 —— 盘后管道 index/etf 阶段因此恒定失败, 且失败
位置固定 (第一批日K响应后 ~0.3s), 与并发无关: recover=True、清理标记、
停掉实时轮询都无法绕过。

修复: commit() 在未处于 publishing 状态 (即本对象已提交过) 时幂等返回。
新增回归测试覆盖单对象重复 commit 与多分区写入中的无变化分区两种情况,
修复前均失败。
This commit is contained in:
Ken
2026-09-24 17:27:40 +08:00
parent ecc418bd94
commit a9e573f1ed
2 changed files with 74 additions and 0 deletions
+10
View File
@@ -321,6 +321,16 @@ class EnrichedPublication:
def commit(self) -> str | None:
if not self._changed:
return None
if not self._publishing:
# 本次是同一发布对象的重复提交: 上一次 commit 已落盘并复位
# _publishing, 标记也回到了 ready (ready payload 不含
# publication_id)。写入方会把一个发布对象复用到多个分区
# (如 ETF enriched 按完整本地历史一次写 245+ 个交易日分区),
# 其中内容无变化的分区会直接走到这里 —— 此时没有任何待提交
# 内容, 必须幂等返回, 否则会误判为 "ownership lost" 并打断
# 整批写入。
self._changed = False
return None
path = _marker_path(self.data_dir, self.asset_type)
with _exclusive_generation_lock(self.data_dir, self.asset_type):
current = _read_marker(path)
@@ -0,0 +1,64 @@
"""多日期分区写入回归: 同一发布对象被复用时, 无变化分区的空提交不能打断整批写入。
`234d31a` 起 ETF enriched 按完整本地历史重算前复权, 一次 `append_etf_enriched`
会写 245+ 个 `date=` 分区; 而 `_write_daily_partition` 对整张表只创建一个
`EnrichedPublication`, 在 `partition_by("date")` 循环里复用。`_write_partition_locked`
对"内容无变化"的分区直接调 `publication.commit()`, 而 commit 提交后 `_changed`
未复位、标记已回到 ready (ready payload 不含 `publication_id`)、`_publishing=False`,
于是第二次 commit 必然命中 `publication_id` 不匹配 → 抛
`enriched publication ownership was lost`。
结果: 只要"有变化的分区"后面跟着一个"无变化的分区", 整批写入就被打断。
盘后管道表现为 `index/etf sync: enriched publication ownership was lost` 恒定失败,
且失败位置固定 (第一批日K响应后 ~0.3s), 与并发无关 —— `recover=True`、清理标记、
停掉实时轮询都不能绕过。
"""
from __future__ import annotations
from datetime import date
import polars as pl
import pytest
from app.enriched_generation import EnrichedPublication
from app.tickflow.repository import DataStore, KlineRepository
def _frame(day: str, close: float) -> pl.DataFrame:
return pl.DataFrame({
"symbol": ["000001.SZ"],
"date": [date.fromisoformat(day)],
"open": [close],
"high": [close],
"low": [close],
"close": [close],
"volume": [1_000.0],
})
def _two_days(close_13: float, close_14: float) -> pl.DataFrame:
return pl.concat([_frame("2026-08-13", close_13), _frame("2026-08-14", close_14)])
def test_repeated_commit_is_idempotent(tmp_path) -> None:
"""分区1 写入并 commit 后, 分区2 无变化 → 空提交必须幂等返回。"""
publication = EnrichedPublication(tmp_path, recover=True)
out = tmp_path / "kline_daily_enriched" / "date=2026-08-13" / "part.parquet"
publication.write_parquet(_frame("2026-08-13", 10.0), out)
assert publication.commit() is not None
assert publication.commit() is None
def test_multi_partition_write_tolerates_unchanged_partition(tmp_path) -> None:
"""一次写两个分区, 第二个完全无变化 —— 修复前整批写入会被打断。"""
repo = KlineRepository(DataStore(tmp_path))
repo.append_enriched(_two_days(10.0, 11.0))
# 第一个分区价格变化, 第二个与磁盘完全一致
repo.append_enriched(_two_days(12.0, 11.0))
part_13 = tmp_path / "kline_daily_enriched" / "date=2026-08-13" / "part.parquet"
part_14 = tmp_path / "kline_daily_enriched" / "date=2026-08-14" / "part.parquet"
assert pl.read_parquet(part_13)["close"].item() == pytest.approx(12.0)
assert pl.read_parquet(part_14)["close"].item() == pytest.approx(11.0)