mirror of
https://github.com/shy3130/tick-stock-panel.git
synced 2026-11-02 10:26:13 +00:00
fix(enriched): commit() 发布后复位脏标记, 多分区空提交不再误抛 ownership (#417)
EnrichedPublication.commit() 发布完成后 _publishing 置 False、标记写回
ready payload (不含 publication_id), 但 _changed 未复位。多日期分区写入
复用同一发布对象 (repository._write_daily_partition), 有变化分区提交后
再遇到无变化分区时, _write_partition_locked 的"无害空提交"带着过期脏
标记命中 ownership 校验, 必然抛 "ownership was lost", 整批写入中断。
e3baba6 让 ETF 增量日K用完整本地历史重算前复权 (单次 245+ 日期分区),
必然出现无变化分区, 该旧缺陷自此恒定复现 (盘后管道 index/etf 阶段失败)。
修复: commit() 发布完成后复位 _changed — 待发布内容已全部落盘, 之后任何
写入都会经 _claim_or_verify 重新认领并再次置位, 语义不变; 空提交经由
既有的是否脏守卫自然幂等。
回归测试: 单对象重复 commit 幂等 + 仓储级「有变化分区在前、无变化分区
在后」两用例, 修复前均失败、修复后通过。
This commit is contained in:
@@ -331,6 +331,11 @@ class EnrichedPublication:
|
||||
generation = uuid.uuid4().hex
|
||||
_write_marker(path, _ready_payload(generation))
|
||||
self._publishing = False
|
||||
# 待发布内容已全部落盘, 复位脏标记: 多日期分区写入复用同一发布对象,
|
||||
# 无变化分区会再调 commit() (空提交) — 不复位会以过期脏标记命中
|
||||
# ownership 校验而误抛 "ownership was lost" (Issue #417)。
|
||||
# 之后的写入会经 _claim_or_verify 重新认领并再次置位。
|
||||
self._changed = False
|
||||
return generation
|
||||
|
||||
def _claim_or_verify(self) -> None:
|
||||
|
||||
@@ -41,6 +41,42 @@ def test_repository_enriched_noop_does_not_bump_generation(tmp_path) -> None:
|
||||
assert repo.get_matrix_data_generation("stock") == first
|
||||
|
||||
|
||||
def test_commit_twice_is_idempotent_issue417(tmp_path) -> None:
|
||||
"""Issue #417 回归: 有变化分区提交后再遇到无变化分区的空提交,
|
||||
必须幂等返回而非抛 ownership was lost (修复前第二次 commit 抛错)。"""
|
||||
publication = EnrichedPublication(tmp_path, recover=True)
|
||||
first = tmp_path / "kline_daily_enriched" / "date=2026-08-13" / "part.parquet"
|
||||
second = tmp_path / "kline_daily_enriched" / "date=2026-08-14" / "part.parquet"
|
||||
publication.write_parquet(_frame(10.0), first)
|
||||
assert publication.commit() is not None # 分区1: 有变化, 正常发布
|
||||
assert publication.commit() is None # 分区2: 无变化, 空提交幂等
|
||||
# 已提交过的发布对象继续写入仍可用 (write 重新认领)
|
||||
publication.write_parquet(_frame(11.0), second)
|
||||
assert publication.commit() is not None
|
||||
|
||||
|
||||
def test_repository_changed_partition_before_unchanged_issue417(tmp_path) -> None:
|
||||
"""Issue #417 回归: 一次写多个日期分区且「有变化在前、无变化在后」,
|
||||
修复前第二个 (无变化) 分区的空提交必然抛 ownership was lost, 整批失败。"""
|
||||
repo = KlineRepository(DataStore(tmp_path))
|
||||
d1, d2 = date(2026, 8, 13), date(2026, 8, 14)
|
||||
base = {"symbol": ["000001.SZ"] * 2, "date": [d1, d2]}
|
||||
repo.append_enriched(pl.DataFrame({
|
||||
**base,
|
||||
"open": [10.0, 10.0], "high": [10.0, 10.0], "low": [10.0, 10.0],
|
||||
"close": [10.0, 10.0], "volume": [1_000.0, 1_000.0],
|
||||
}))
|
||||
first = repo.get_matrix_data_generation("stock")
|
||||
# 第二批: d1 有变化, d2 与库中内容一致 (无变化分区排在其后)
|
||||
repo.append_enriched(pl.DataFrame({
|
||||
**base,
|
||||
"open": [10.5, 10.0], "high": [10.5, 10.0], "low": [10.5, 10.0],
|
||||
"close": [10.5, 10.0], "volume": [1_000.0, 1_000.0],
|
||||
}))
|
||||
assert repo.get_matrix_data_generation("stock") != first
|
||||
assert repo.get_matrix_data_generation("stock") != first # 幂等不抛
|
||||
|
||||
|
||||
def test_failed_multi_partition_publication_remains_fail_closed(
|
||||
tmp_path,
|
||||
monkeypatch,
|
||||
|
||||
Reference in New Issue
Block a user