chongming-utils-task/docs/store-guide.md

持久层实现指南

换掉默认持久层需要实现哪些接口，以及每个接口的语义要求。

本文是规格，不是介绍。照着做即可替换内置的 SQLite 实现；有可执行的范例和可执行的 契约测试，两者都在仓库里。

---

一句话版

实现 TaskStore 协议里的 14 个方法，然后用 chongming_task.testing.store_contract.StoreContract 自证。不需要继承、不需要注册。

from chongming_task import Controller
from chongming_task.stores.protocol import TaskStore  # 结构化协议，无需继承


class MyStore:  # 只要方法齐备、语义正确
    ...


controller = Controller(store=MyStore(...))  # 直接塞进去

Controller 只做结构化检查（TaskStore 是 runtime_checkable 的 Protocol）， 所以任何满足契约的对象都能直接用。继承它、注册它都不必要。

---

1. 方法清单

分组	方法	一句话职责
任务快照（参与乐观并发）	create(snapshot)	写入新任务
	save(snapshot, *, expect_version)	按版本更新，冲突即抛错
	get(task_id)	读单个
	query(query=None)	按条件查询
	counts()	各状态计数
审计轨迹（只追加）	append(transition)	追加一条迁移，幂等
	history(task_id, *, since_version=0)	读迁移历史
高频旁路（不碰版本号）	touch(task_id, *, progress=None, heartbeat_at=None)	只更新进度与心跳
控制指令（跨进程）	submit_command(command)	写入一条待执行指令
	claim_commands(*, now, lease_seconds, limit=50)	取走一批并打租约
	complete_command(command_id, *, now, outcome=None)	标记完成
	release_command(command_id, *, now)	交还（取错了或没认领）
运维	prune(*, keep_terminal)	丢弃最旧的终态任务及其历史
	close()	释放资源

代码即真相：src/chongming_task/stores/protocol.py。本文与它的方法集合由测试 （tests/integration/test_custom_store.py）强制保持一致——协议加了方法却没写进本文， 或者本文写了一个不存在的名字，测试都会失败。

---

2. 逐方法语义

2.1 任务快照

create(snapshot)

写入一个新任务。task_id 已存在必须抛错（StoreError），不要静默覆盖。

为什么静默覆盖很危险：任务已经在跑，第二次 create 会把它的状态重置回 PENDING，而正在运行的执行载体不会因此停下——内存与存储从此永久分叉。

save(snapshot, *, expect_version)

按乐观并发更新。expect_version 是调用方写入前读到的 status_version；不一致必须抛 ConcurrencyConflict。

"校验 + 写入"必须是原子的一步。 关系库里它应当是一条 UPDATE ... WHERE task_id = ? AND status_version = ?；拆成"先查后写"必然在跨进程下 产生竞态。KV 后端用 WATCH/MULTI、Lua 脚本或 CAS。

不要自动重试冲突。 每个任务的写入者唯一（它的持有进程），冲突意味着存在第二个 写入者——那是设计缺陷，掩盖它比报出来危险得多。

⚠️ 版本号单调但不连续

expect_version 是调用方认为存储里当前的版本。它不保证等于 实际版本 - 1：

写穿失败会把迁移攒进缓冲，恢复后一次跳进（比如存储从 version 2 直接变成 4，中间的 3 只在迁移历史里出现）。所以后端不得假设：

• 版本号连续（每次 +1）；
• expect_version == 当前行版本（这是对的）但 写入的快照版本 == expect_version + 1（这 是不成立的）。

后端唯一要保证的是：expect_version 与行里的 status_version 相等时才允许写入。 把它改写成"只接受 +1 的推进"会让写穿缓冲永远补不上，等于重新引入"一次抖动永久冻结"。

get(task_id)

读单个任务，不存在返回 None（不要抛异常：查询不存在的任务是正常操作）。

query(query=None)

按条件查询，按 created_at 升序（newest_first=True 则降序）。None 表示不限制。

过滤要下推到后端，不要取回全部再筛——buckets 那条尤其重要，它是分布式下 "取我拥有的那组桶"的高频查询。SQLite 侧为它建了索引。

需要支持的过滤条件：statuses / name / name_prefix / tag / buckets / pipeline_run_id / limit。其中 pipeline_run_id 是"把这条管道的节点整条列出来" 的入口——管道恰恰是最需要按整体查看的东西，没有它就只能全表拉回来自己筛。 tag 必须精确匹配（io 不能命中 ios）。

counts()

返回 {TaskStatus: 数量}。后端能用 GROUP BY 就别把全部记录读出来数。

2.2 审计轨迹

append(transition)

追加一条迁移记录。必须按 task_id + version 幂等：崩溃恢复会重放，重复写入 不应产生重复行。关系库用 INSERT OR IGNORE / ON CONFLICT DO NOTHING。

history(task_id, *, since_version=0)

返回 version > since_version 的记录，按版本升序。since_version 是增量拉取的 入口（看板轮询靠它）。

2.3 高频旁路

touch(task_id, *, progress=None, heartbeat_at=None, lease_owner=None, lease_until=None)

只更新这几个字段，不校验也不递增 status_version。

这是整个契约里最容易做错的一条。进度可能每秒几十次；它若参与乐观并发，会与状态 写入持续冲突，把版本号变成噪声，最终表现是"状态老是写不进去"，但没有任何报错指向 真正的原因。

任务不存在时静默忽略（心跳晚到一步不该报错）。参数全为 None 时可以什么都不做。 每个参数独立可选，不要要求调用方凑齐。

两个信号必须分开，不能互相推导：

参数	谁在写	回答的问题
heartbeat_at	任务自己的 checkpoint()	任务在推进吗（停滞 = 卡死）
lease_owner / lease_until	监督进程定期续租	还有人在管它吗（过期 = 监督者没了）

拿任务心跳去判断"进程是否还活着"是错的：一个长时间不调 checkpoint() 的健康任务 （纯 CPU 计算）心跳同样会停滞，于是重启恢复会把它误判为丢失并杀掉。反过来拿租约 判断卡死也不对。实现必须原样存这两个信号，不要在中间做任何换算。

⚠️ save 不得写这两个租约列

这条单列出来，因为它是本契约里第二个"同一行两条写入路径"的坑（第一个是心跳）。

touch 写入的租约值不在调用方的内存快照里。若 save 把整份快照写回去，就会把 租约覆盖成快照里的旧值（通常是 None），于是：

• 一个活得好好的任务看起来"没有监督者"；
• 下一次 reconcile() 把它判成 LOST——把一个正常任务杀掉；
• 而且只在重启恢复时才暴露，平时完全看不出来。

所以 save 必须显式跳过租约列：

# SQLite 实现：save 的列集合 = 全部列 - 租约列
_SAVE_COLUMNS = tuple(c for c in _TASK_COLUMNS if not c.startswith("lease_"))

# 内存实现：整体替换时把租约从旧值带过来
replace(snapshot, lease_owner=current.lease_owner, lease_until=current.lease_until)

StoreContract 里有一条用例（test_save_does_not_clobber_the_lease）专门盯这件事， 三个内置实现都跑它。它抓出过真实缺陷：内存实现最初就是整体替换，契约测试当场 失败。

2.4 控制指令

submit_command(command)

写入一条指令。存储层只传递，不解释、不执行——执行者是持有该任务的进程。

claim_commands(*, now, lease_seconds, limit=50)

取走一批可执行指令并打上租约。可被取走的条件：未完成，且无租约或租约已过期。

"挑选 + 打租约"必须是原子的。 否则两个进程会同时取到同一条指令，于是同一次停止 被下发两次（对停止还算无害，对将来的"重试"就是灾难）。SQLite 用 BEGIN IMMEDIATE + COMMIT；Postgres 下标准做法是 SELECT ... FOR UPDATE SKIP LOCKED 再 UPDATE（内置 PostgresTaskStore 就是这么做的， 见 §6）。

按 created_at 升序取，并遵守 limit。

complete_command(command_id, *, now, outcome=None)

标记完成，此后不再被取走。outcome 是执行结果的摘要（如 "stopped" / "already-terminal"），用于事后追溯。

release_command(command_id, *, now)

交还一条已取走但无人认领的指令——清除租约，但不标记完成。

这个方法看似多余，实则不可省。多进程下任何能取指令的进程都可能取到不属于自己的 任务（它手上没有那个任务）。此时只有三种选择，而前两种都是错的：

做法	后果
标记完成	真正的持有者永远收不到这次请求——一个静默的丢指令
什么都不做，等租约过期	只要该进程持续轮询就会不断续租，真正的持有者被无限期饿死
release_command	立即交还，真正的持有者马上能取到 ✓

2.5 运维

prune(*, keep_terminal)

丢弃最旧的终态任务及其历史，返回丢弃数量。绝不能碰非终态任务——删掉一个正在跑的 任务等于把它的状态机扔了。按 created_at 升序决定丢谁。

close()

释放资源。内存实现是空操作；文件/连接型实现必须真的关——没关的 SQLite 连接会 泄漏文件句柄，在 Windows 上还会让数据库文件无法删除。

必须幂等：调用两次不报错。关闭之后再调用其它方法应当抛 StoreError（而不是 AttributeError 或段错误）——把它当成一个明确的状态，不是未定义行为。

---

3. 并发要求汇总

要求	为什么
save 的"校验 + 写入"原子	否则跨进程下版本检查形同虚设
版本号单调但不连续：只要求 expect_version == 行里的版本	写穿缓冲恢复后会一次跳进；若后端要求"只接受 +1"就等于重新引入"一次抖动永久冻结"
不要自动重试冲突	冲突意味着存在第二个写入者，是设计缺陷
claim_commands 的"挑选 + 打租约"原子	否则同一指令被下发两次
租约到期可重取	取走指令的进程崩溃后，指令不能永久卡住
append 幂等	崩溃恢复会重放
touch 不参与乐观并发	高频写入会与状态写入持续冲突
save 写了租约列	租约被覆盖回旧值，重启恢复把活着的任务判成 LOST

契约测试不测跨进程。 它只保证单连接语义正确；跨进程原子性必须由后端自己保证， 并在你的环境里单独压测。本文的范例实现（JSON 文件）就只在单进程内正确——它的 模块文档里写明了这一点，不要照抄成多进程方案。

---

4. 序列化：别自己重写

TaskSnapshot 里有 datetime、frozenset、枚举和嵌套的 Progress / Failure。 这些细节收敛在 chongming_task.stores.codec，可以直接复用：

from chongming_task.stores.codec import (
    snapshot_to_row,
    row_to_snapshot,  # 快照 <-> 扁平原语映射
    transition_to_row,
    row_to_transition,  # 迁移记录
    command_to_row,
    row_to_command,  # 控制指令
    encode_dt,
    decode_dt,
    encode_tags,
    decode_tags,
)

映射的形状是扁平的一层字典（列名 → 原语）——关系库直接就是一行，文档库存成子 对象，KV 直接写键值对。

三条硬要求：

1. 时间一律先转 UTC 再存。 把本地时间原样写进去，跨时区部署时排序与比较都会错， 而且不报错。
2. 往返必须保真：row_to_snapshot(snapshot_to_row(s)) == s。契约测试用一条 "每个字段都填满"的快照验证——只测几个字段会漏掉时区、frozenset、嵌套结构这类 静默丢失。
3. 枚举写 .value，不要指望 json.dumps 对枚举的行为。

---

5. 接线

from chongming_task import Controller

controller = Controller(
    store=MyStore(...),
    owner="worker-1",  # 写进它下发的指令里，便于追溯
    command_poll_interval=0.5,  # 指令轮询间隔
    command_lease=30.0,  # 取走指令后的独占时间
)

接上之后：

方向	行为
写	每次状态迁移写穿到存储（挂在状态机的 _apply 上，因此不会漏）
读	get / list_tasks / stats / bucket_counts 走存储
控制	不属于本进程的任务 → 写一条指令，由持有者轮询取走
高频	进度与心跳由轮询线程定期经 touch 回写

两个必须知道的边界：

• 纯查询客户端不轮询指令。 轮询线程只在提交过任务之后才启动——否则一个只读看板 会把别人的停止请求抢走。这是靠"提交过才启动"实现的，后端不需要配合。
• BucketPolicy.num_buckets 是部署范围内的约定。 多节点各配不同的桶数会让同一个 键落到不同的桶。后端不感知这件事，但你的部署脚本应当断言各处一致。

---

6. 自证：契约测试

from chongming_task.testing.store_contract import StoreContract


class TestMyStore(StoreContract):
    def make_store(self):
        """每次调用返回一个全新的、隔离的存储。"""
        return MyStore(...)

套件随包发布，且不依赖 pytest（只用普通断言），因为它是给后端作者的规格而 不是本仓库的内部测试。pytest 仍会自动收集子类里的用例。

它覆盖什么：

• 协议的结构性要件（isinstance(store, TaskStore)）；
• 每个字段的往返保真；
• 乐观并发的接受与拒绝，以及拒绝后状态未被改动；
• 查询的每一种过滤组合、排序、limit、空存储；
• 标签的精确匹配（io 不能命中 ios）；
• since_version 增量、幂等重放；
• touch 不改版本号、任务不存在时静默；
• 指令的取走 / 完成 / 租约过期重取 / 交还；
• prune 只丢终态、历史一并清理。

内置的实现跑的是同一套用例——InMemoryTaskStore、SqliteTaskStore（文件库）、 SqliteTaskStore(":memory:")（内存库），以及 PostgresTaskStore （import chongming_task 后惰性取用，需要 pip install "chongming-task[postgres]"） ——所以"通过"这件事是有基准的。

契约测试不测跨进程。 它只保证单连接语义正确；跨进程原子性必须由后端自己保证， 并在你的环境里单独压测。内置 Postgres 实现把这份证据留在了仓库里 （tests/integration/test_postgres_store.py，需要一台可连的库）：

用例	钉住的承诺
8 个线程并发 save 同一版本	恰好一个成功，其余全部 ConcurrencyConflict
8 个消费者并发 claim_commands	每条指令恰好被取走一次（SKIP LOCKED 真的在起作用）
持有者被 pg_terminate_backend 强杀	租约不提前释放也不永锁，到期可接管，连接池自愈
权限受限的角色	只用已有 schema，不做 DDL——只读运维控制台可用

要跑它们，给一个 DSN 即可（每个用例自建独立 schema，跑完即删）：

CHONGMING_TEST_POSTGRES='postgresql://user@host:5432/db' uv run pytest tests/integration

不给 DSN 时整个模块干净跳过——跨进程证据缺席会显式表现为 skip，而不是默默变绿。 自己实现后端时，照这个模式补你自己的并发用例：契约测试全绿只说明单连接语义对， 不说明多进程下对。

---

7. 可执行范例

tests/integration/test_custom_store.py 里有一个完整的第三方后端 （JsonFileTaskStore，把全部状态放在一个 JSON 文件里）。它：

• 只实现契约，不继承任何东西；
• 复用 stores.codec 做序列化；
• 跑通完整的 StoreContract；
• 另有三条用例证明它真的能接进 Controller、并支持跨进程控制。

它是可以照抄的骨架。但它演示的是接口形状，不是跨进程方案。 无锁的文件型后端在 Windows 上无法支撑"两个实例共用一个文件"：Path.replace 的原子替换与并发打开会 互相冲突，实测抛 PermissionError；直接 write_text 更糟，读方会拿到空文件或半截 JSON（JSONDecodeError: Expecting value: line 1 column 1）。

范例真实踩到的两个坑，都已固化进代码注释与用例：

1. 不能缓存。 两个实例各持一份内存副本时彼此看不到对方的写入，跨进程控制会 静默失效（控制端把指令写进自己的副本，worker 压根不知道）。本实现因此每次 操作都重新读盘。
2. 写回必须原子替换。 见上。但即便如此，Windows 上仍不足以支撑共享。

要跨进程就用内置的 SqliteTaskStore——文件锁、WAL、BEGIN IMMEDIATE 都由 SQLite 负责，这正是它作为默认实现的理由。范例里有一条用例（ test_the_naive_file_store_is_not_cross_process_safe）把这个局限钉在测试里， 而不是写在注释里自欺欺人。

---

8. 检查清单

替换默认持久层之前，逐条确认：

• 14 个方法齐备，签名与 protocol.py 一致
• create 对已存在的 task_id 抛错，不覆盖
• save 的"校验 + 写入"是原子的，冲突抛 ConcurrencyConflict，且不自动重试
• save 允许版本跳号（不要求连续），只按 expect_version 相等判定
• append 幂等（task_id + version）
• history 按版本升序，支持 since_version
• touch 不改 status_version，任务不存在时静默
• touch 支持 lease_owner / lease_until，且每个参数独立可选
• save 不写租约列（否则重启恢复会误杀活着的任务）
• claim_commands 的"挑选 + 打租约"原子（关系库：SELECT ... FOR UPDATE SKIP LOCKED），租约到期可重取
• 实现了 release_command，且它不标记完成
• prune 只丢终态，且一并清理历史
• 时间以 UTC 存储，往返保真
• 过滤（尤其 buckets）下推到后端，不是取回全部再筛
• close() 真的释放连接/文件句柄，幂等，关闭后再用抛 StoreError
• StoreContract 全绿
• 若要在多实例间共享：有真正的跨进程锁（无锁文件方案在 Windows 上不可行）
• 跨进程原子性在你的环境里单独验证过（契约测试不覆盖它）

---

9. 常见陷阱

陷阱	症状	正确做法
save 拆成"先查后写"	跨进程下版本检查形同虚设，偶发丢更新	一条带 WHERE version = ? 的原子写
claim 不加租约	取走指令的进程崩溃后，指令永久卡住	租约 + 到期可重取
claim 不原子	同一次停止被下发两次	BEGIN IMMEDIATE / Lua / CAS
把 release 实现成 complete	静默丢指令，持有者永远收不到	只清租约，不设 finished_at
touch 递增版本号	状态写入频繁冲突，但报错指向别处	高频旁路绝不碰版本号
append 用普通 INSERT	崩溃恢复产生重复历史行	幂等写
时间存本地时区	跨时区排序错，且不报错	先转 UTC
标签用子串匹配	io 命中 ios，筛选结果悄悄错	精确匹配（JSON 数组 + 带引号匹配）
prune 删掉非终态任务	正在跑的任务状态机被扔掉	只丢终态
忘记 close()	文件句柄泄漏；Windows 下数据库文件删不掉	真的关
create 静默覆盖	任务重跑但旧执行载体还在跑，内存与存储分叉	已存在就抛错
在内存里缓存整份数据	多实例各持副本，静默看不到对方写入	每次操作重新读，或配文件锁 + 失效判断
write_text 直接覆盖	读方拿到空文件/半截内容	Path.replace 原子替换（含 pid 临时名）
无锁文件当共享存储	Windows 上并发读写直接 PermissionError	需要共享就用带锁的存储（如 SQLite）