# 持久层实现指南

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

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

---

## 一句话版

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

```python
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` 必须**显式跳过**租约列：

```python
# 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`，**可以直接复用**：

```python
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. 接线

```python
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. 自证：契约测试

```python
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，跑完即删）：

```bash
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） |