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

用户指南

面向使用者的完整说明。设计动机与取舍见 设计文档，可直接 运行的代码在 examples/。

适用范围：单任务执行、DAG 管道、并发额度、持久化与重启恢复、CLI 均已可用 （状态表见 README）。真正的边界是语义上的，不是功能 缺失：不调 checkpoint() 停不下来、暂停不释放槽位、任务不跨节点迁移——见 §11。

---

目录

1. 核心概念
2. 写一个任务
3. 跑一个任务
4. 控制任务
5. 重试
6. 超时
7. 执行器
8. 管理一批任务（Controller）
9. 状态模型参考
10. 错误与异常
11. 限制与 FAQ
12. 测试自己的任务

---

1. 核心概念

1.1 任务是一条轨迹，不是一个布尔值

大多数"任务控制"只是把任务记成"跑着 / 成功 / 失败"。这里不是。每个任务是一条 只追加的状态轨迹，每个中间态都可表达、可观察、可干预：

pending → queued → starting → running → succeeded
                     ↓           ↓
                 retry_wait   pausing → paused → running
                     ↓           ↓
                  starting    stopping → stopped

每次状态变更都会产生一条 Transition 记录，带触发源、操作者、原因和 版本号：

for record in run.transitions:
    print(record.from_status, "->", record.to_status, "by", record.trigger)

1.2 状态只表达"生命周期阶段"

status 是一个枚举，只回答"它现在处于哪个阶段"。其余信息属于正交维度，绝不 塞进同一个枚举：

维度	字段	为什么必须分开
生命周期	status	单一枚举，便于穷举校验
重试计数	attempt / max_attempts	"第 3 次尝试"与"正在运行"是两件事
进度	progress	运行中的任务可能 0% 也可能 99%
健康	last_heartbeat_at	用于发现"没报错但也不动"
结果	failure / outcome	失败原因是技术细节，不是生命周期

最典型的反例是把超时做成状态。那样会立刻逼出"已超时但正在重试"这种单一枚举 表达不了的情况。因此超时是 failure.kind = timeout_attempt。

1.3 状态机是唯一写者

只有 StateMachine 能改 status。业务代码、执行器、控制面都不能直接赋值。这条 纪律换来三件事：非法转移会被立刻拒绝、状态历史必然完整、并发写入可检测 （status_version 乐观锁）。

from chongming_task.core.state_machine import StateMachine
from chongming_task.states import TaskStatus

machine = StateMachine.create("t1", "抽取", max_attempts=3)
machine.mark_queued()
machine.mark_starting(executor="thread")
machine.mark_running()
machine.mark_succeeded(outcome={"rows": 42})

# 非法转移会抛异常，而不是静默忽略
machine.mark_running()
# InvalidTransition: 非法状态转移：succeeded -> running；动作 'executor:launched'；当前允许：无（终态）

非法转移抛异常而不返回 False：静默失败会让"停止指令没生效"这类问题极难排查。 需要预检时用 machine.can(target) / can_stop / can_pause / can_resume / can_retry。

1.4 终态与复活

终态没有出边，这是转移表里显式声明的空集。复活只能走两个显式 API：

• retry()：从 failed / stopped / cancelled / lost 复活，保留尝试历史。
• reset()：回到 pending，丢弃全部尝试历史（危险，需二次确认）。

---

2. 写一个任务

2.1 签名

任务的第一个位置参数由执行器注入为执行期上下文：

def my_task(ctx, source: str, *, batch_size: int = 100) -> list[Row]: ...

调用时不要自己传 ctx：

run_task(my_task, "s3://bucket/x", batch_size=500)

为什么 ctx 必须注入而不是闭包捕获？ 因为上下文属于某一次尝试：取消令牌 在启动那一刻才由执行器造出来，任务不可能提前攥着一个。若靠闭包捕获（lambda: f(ctx, arg)）， 任务定义里握住的只可能是上一次尝试留下的令牌——停止指令会发给一个已经结束的执行， 而且完全静默。注入式签名让同一份任务定义既能被线程执行器驱动，也能被内联执行器 同步跑完，语义只有一份。

2.2 为什么必须调用 checkpoint()

def my_task(ctx):
    for row in rows:
        ctx.checkpoint()  # ← 就这一行
        process(row)

ctx.checkpoint() 同时做三件事：

1. 若已请求停止 → 抛 TaskCancelled；
2. 若已请求暂停 → 在此挂起，直到恢复或停止；
3. 刷新心跳（idle 超时的依据）。

不调用它的任务既停不掉也暂停不了。 Python 不能安全地终止线程，也没有跨平台的 抢占式暂停手段，所以停止与暂停只能是协作式的。这是硬约束，不是可以绕过的实现细节。

长循环里每轮都调用即可，开销可以忽略（无请求时它只是两次布尔判断）。

2.3 清理逻辑里的 is_stop_requested

在 except / finally 块里不要再调用 checkpoint()——它会阻塞，导致清理永远 完不成。改用只读属性：

def my_task(ctx):
    conn = connect()
    try:
        for row in rows:
            ctx.checkpoint()
            write(conn, row)
    finally:
        conn.commit()  # 无论如何都要收尾
        if ctx.is_stop_requested:
            print("任务被中断，已提交部分结果")  # log/progress 属于 P4，此处暂用 print
        conn.close()

2.4 产物目录

def my_task(ctx):
    target = ctx.artifact_dir / "part-0.parquet"
    ...

ctx.artifact_dir 按 attempt 隔离，重试不会污染上一次的产物。传 artifact_dir 给 TaskRun / run_task 即可启用。

---

3. 跑一个任务

3.1 一行版：run_task

from chongming_task.run import run_task

outcome = run_task(my_task, "s3://bucket/x", task_id="job-1", name="抽取")

if outcome.ok:
    print(outcome.value)
else:
    print(outcome.status.value, outcome.failure.kind.value)

run_task 会阻塞到任务落终态。不要在需要中途干预时用它。

3.2 完整版：TaskRun

from chongming_task.run import TaskRun

run = TaskRun(my_task, ("s3://bucket/x",), task_id="job-1", name="抽取")
run.start()  # 立即返回
...  # 这里可以做任何事：暂停、停止、轮询状态
outcome = run.wait(timeout=60)

成员	说明
start()	启动受监督执行，立即返回；重复调用报错
wait(timeout=None)	等待终态；超时抛 TimeoutError 且不改变任务状态
stop(reason=...)	请求停止，不阻塞
pause(timeout=...)	请求暂停并等待确认，返回 bool
resume()	撤销暂停
snapshot	当前 TaskSnapshot，运行期可安全读取
status	当前 TaskStatus
transitions	完整迁移历史

TaskRun 是线程安全的：上述方法可从不同线程调用。所有状态机写入都在同一把锁下 进行，因此"唯一写者"这条不变量仍然成立。

3.3 结果对象

outcome.ok  # bool：是否 succeeded
outcome.status  # TaskStatus
outcome.value  # 任务返回值（未成功时为 None）
outcome.failure  # Failure | None
outcome.snapshot  # 终态快照：attempt、时间戳、progress、failure 等

---

4. 控制任务

4.1 停止的三段式

请求停止 → 优雅窗口（graceful_stop 秒）→ 兜底

• 任务在窗口内响应 checkpoint() → 正常收尾，落 stopped。
• 窗口用尽 → 线程无法强杀，放弃等待并落 lost。

run.stop(reason="发布窗口")  # 不阻塞
outcome = run.wait(timeout=30)

停不掉时不会说"已停止"：

	任务配合	任务不配合（线程）
outcome.status	stopped	lost
outcome.failure.kind	cancelled	lost
failure.retryable	False	True
后台副作用	已收尾	可能仍在产生

把两者都简化成"已停止"不只是不准确，而是会让运维误判资源已释放。

4.2 暂停与恢复

run.pause()  # → True 表示已确认挂起，状态变为 paused
run.resume()  # → running

pause() 返回 False 不是错误，是"暂停没生效"：任务在窗口内没到达检查点。 此时状态回滚为 running，并留下一条 runtime:pause_aborted 记录，绝不假装 已暂停。

两条必须知道的性质：

1. 暂停不释放并发槽位。 就地挂起——任务的调用栈还在内存里。我们没有办法把 任意的 Python 调用栈换出再换回，所以暂停大量任务可能耗尽并发额度。
2. 暂停期间不判超时。 被暂停的任务本来就不会推进心跳，若照常判 idle 超时， 暂停会被误判成卡死。

4.3 恢复的目标状态取决于暂停层级

这是实现期修正过的一处语义，容易踩：

暂停发生在	恢复后回到	理由
运行中（经 pausing）	running	就地暂停，槽位从未释放，调用栈还在
排队/重试等待中（未启动）	queued	从未启动过，重新排队

TaskRun.resume() 会自动判定，无需你关心。

4.4 停止请求发出后任务恰好跑完

这是合法结果：状态走 stopping → succeeded，不判为失败。停止是请求，不是判决。

4.5 无限循环（常驻）任务：怎么启、怎么停

一个"永远跑下去"的任务，任务体每轮必须调 ctx.checkpoint()：

@task
def worker(ctx, source: str) -> int:
    processed = 0
    while not ctx.is_stop_requested:  # ← 优雅收尾的写法
        ctx.checkpoint()  # ← 不调用它 = 停不掉
        processed += 1
        ctx.progress(processed, unit="batch")
        time.sleep(0.02)
    ctx.log(f"收到停止请求，本次共处理 {processed} 批")
    return processed  # 正常返回 → 终态 succeeded，返回值保留

停止请求的三种收尾方式，终态不同

任务体的写法	终态	返回值	含义
while True: ctx.checkpoint()（让它抛 TaskCancelled）	stopped	无	立刻中止
检查 ctx.is_stop_requested 后正常 return	succeeded	✅ 有	请收工——把已处理的结果留下来
完全不调 checkpoint()	lost	无	载体被放弃，线程仍在后台跑

第二行不是笔误：状态机允许 stopping → succeeded，因为"停止"是请求而不是判决。对"跑到 下一个安全点再退出"这类需求，它比 stopped 更有用——已经做好的结果不会丢。

日常"先停一下"应该用 pause / resume，不是 stop

stop 是终态且不可逆：当前没有 retry，停止之后想再跑必须换一个 task_id 重新提交 （store.create 会拒绝重复 id）。所以：

controller.pause("orders-worker")  # → True：已确认挂起；False：未确认（状态回滚 running）
controller.resume("orders-worker")  # 继续跑
controller.stop("orders-worker", reason="发布窗口")  # 彻底结束

pause 是就地挂起：任务阻塞在 checkpoint() 里，继续占用执行槽。暂停大量常驻 任务可能耗尽并发额度。

启停的两种方式

# 方式一：同进程（需要这个进程一直活着）
controller = Controller(store=store, owner="node-a")
run = controller.submit("worker", "kafka://orders", task_id="orders-worker")
controller.stop("orders-worker", reason="发布窗口")
outcome = controller.wait("orders-worker", timeout=10)  # → succeeded，value=已处理批数

# 方式二：命令行（前台执行，交给 systemd / nohup；停止从另一个终端发起）
# --bucket-key 在限定了 --owned-buckets 时是**必需**的：没有归属桶的任务不会有任何
# 节点认领它，所以 CLI 会在提交时报错，而不是留下一个永远排队的任务。
chongming-task --db tasks.db --num-buckets 64 --owned-buckets 0-15 \
    run --tasks myapp.tasks worker --arg source=kafka://orders \
        --task-id orders-worker --bucket-key orders

chongming-task --db tasks.db --owned-buckets 0-15 status orders-worker
chongming-task --db tasks.db --owned-buckets 0-15 stop orders-worker --reason 发布窗口

run 是前台进程：它退出时任务就没人管了（监督线程是守护线程）。要长期跑请交给 systemd / nohup。Ctrl-C 是安全的——CLI 会走 finally 里的 controller.close()， 把在跑的任务停干净。

如果任务不调 checkpoint()

它不是"停得慢"，而是停不掉：Python 无法安全终止线程，所以系统只能如实报 lost （结果未知），而任务体仍在后台继续执行——真正终止它只能靠结束进程。这正是"诚实优先于 便利"的那条设计原则：与其假装已停止，不如报"结果未知"。

完整可运行演示见 examples/09_long_running.py。

---

5. 重试

from chongming_task.policies import ExponentialBackoff, RetryPolicy

RetryPolicy(
    max_attempts=5,  # 总尝试次数（含首次）
    backoff=ExponentialBackoff(base=1, factor=2, max_delay=300),
    retry_on=(Exception,),  # 哪些异常值得重试
    give_up_on=(ValueError, KeyError),  # 哪些立刻放弃
)

字段	说明
max_attempts	总尝试次数，因此 1 表示不重试
backoff	FixedBackoff(delay) 或 ExponentialBackoff(base, factor, max_delay, jitter)
retry_on	值得重试的异常类型
give_up_on	命中即放弃，优先级高于 retry_on

退避默认带抖动（Jitter.FULL，在 [0, delay) 内均匀取值）。这不是花哨：没有 抖动时，一整批同时失败的任务会在同一时刻集体重试，形成风暴。

5.1 取消永远不会被重试

TaskCancelled 继承自 BaseException，而 retry_on 默认是 (Exception,)，因此 取消信号天然不匹配任何重试条件。同时 Failure.retryable 对取消置为 False。 重试一个已被取消的任务，等于无视停止指令。

5.2 失败分类

每次失败都会带一个 FailureKind：

kind	含义
exception	任务抛出了异常
nonzero_exit	外部命令返回非零退出码
timeout_start / timeout_attempt / timeout_idle	三档超时
cancelled	被取消
lost	结果未知
dependency_failed	上游失败
condition_false	条件不满足（配合 skipped）

---

6. 超时

from chongming_task.policies import Timeout

Timeout(start=30, per_attempt=600, idle=60)

三档相互独立，取最早到期者：

作用域	含义	生效条件
start	从提交到真正开始执行	一旦开始执行即失效
per_attempt	单次尝试的总时长	本次尝试已开始
idle	距最近一次 checkpoint() 的静默时长	本次尝试已开始

超时到期后系统会请求停止该任务，并把结果记成对应的 timeout_* 失败，而不是 一个"超时状态"。超时默认可重试（多半是暂时性的）。

6.1 idle 的语义与陷阱

idle 衡量的是 checkpoint() 间隔，不是"CPU 是否在干活"。这是唯一能发现 "卡住但没报错"（死锁、连接挂起）的机制，但它有个前提：

如果你有一段几分钟不调用 checkpoint() 的重计算，它会被判定为疑似卡死。

所以 idle 阈值必须与任务里 checkpoint() 的频率相匹配。如果某段计算确实无法 中途打断，要么把它调大，要么在进入该段之前手动 ctx.heartbeat.beat()。

另外两条边界：idle 只在本次尝试已经开始后才参与判定，因此排队等待的时长不会 被误判成静默超时；暂停期间同样不判 idle，因为被暂停的任务本来就不会推进心跳。

---

7. 执行器

7.1 线程执行器（默认）

from chongming_task.runtime.executors.thread import ThreadExecutor

run = TaskRun(my_task, executor=ThreadExecutor())

适合阻塞式 I/O（数据库、HTTP、文件）与普通业务函数。特点：

• 创建开销几十微秒，共享内存，可复用连接池与缓存；
• 无参数序列化要求，闭包、lambda、局部函数都能用；
• async def 任务同样能跑：执行器在本线程内把协程跑完，await ctx.acheckpoint() 照常响应停止与暂停；
• 无法被强制停止（Python 不能安全终止线程）。

7.2 内联执行器

from chongming_task.runtime.executors.inline import InlineExecutor

run_task(trivial_step, executor=InlineExecutor())

在调用线程内同步跑完，零调度开销，适合微秒级的纯计算步骤与确定性测试。

它刻意违反"launch 不得阻塞"的约定（这就是它存在的意义），因此异步调度器 不得使用它。它也无法被暂停——pause() 如实返回 False，不假装成功。

7.3 为什么不用 ThreadPoolExecutor

ThreadPoolExecutor 的工作线程会在解释器退出时被 join。一个不响应取消的 任务会让整个进程无法退出——对一个专门监督"可能不配合"的任务的工具来说这是致命 缺陷；此外池子也无法放弃卡住的线程。

因此 ThreadExecutor 每个任务起一个守护线程，代价是被放弃的线程会泄漏（所以 结果记 lost）。并发上限由调度层统一负责，执行器不再叠一层池。

7.4 进程执行器：硬性停止 / 绕开 GIL / 崩溃隔离

from chongming_task import ProcessExecutor, run_task

outcome = run_task(fetch_all, executor=ProcessExecutor())  # 单个任务
controller = Controller(executor=ProcessExecutor())  # 或整批任务

任务跑在独立子进程里。它换来三件线程给不了的事：

能力	线程执行器	进程执行器
优雅窗口用尽后	只能放弃等待，记 lost（结果未知）	terminate() → kill()，真的停掉
CPU 密集	受 GIL 限制	真并行
原生扩展段错误	带走整个监督进程	只带走子进程，任务如实记一次失败

代码里的体现是：超时或停止不下来的任务会被真的终止，最终状态是 failed （超时/崩溃）或 stopped（停止请求），而不是 lost。快照里的 pid 就是那个子进程的 进程号，ps 能直接对上号。

代价是三条结构性的约束，不是实现偷懒：

1. 参数、返回值、上游数据必须可 pickle：lambda、闭包、局部函数一概不行——它们 在子进程里找不到。违反时提交那一刻就报 ConfigurationError 并告诉你改用什么。 返回值的约束也比线程执行器严（那边只要求可 JSON 序列化）。
2. 目标函数必须在可导入的模块里，且调用方脚本要有 if __name__ == "__main__": 守卫（multiprocessing 的 spawn 会重新 import 主模块）。用 @task 装饰过的 模块级函数没问题；写在函数体里的临时函数不行。
3. 停止与暂停有约 20ms 的轮询延迟，信号要过进程边界；线程执行器是即时的。 另外子进程是守护进程，不能再创建自己的子进程——任务内部要并行请开线程。

@task
def resize(ctx, path: str) -> str:  # ✅ 模块级、参数可 pickle
    ...


async def _inline():  # ❌ 闭包 / 局部函数
    ...

什么时候该用它：需要"停得住"的保证（例：任务里跑着别人写的、可能不响应取消的 代码）、CPU 密集、或跑第三方原生扩展（段错误隔离）。什么时候不该：绝大多数 阻塞式 I/O 任务——线程更轻、约束更少，而且能直接复用连接池与缓存。

---

8. 管理一批任务（Controller）

Controller 是统一的管理入口：它持有任务定义的白名单、运行中的任务 与并发额度。将来的 CLI 与任何看板都只是这一层的薄适配。

8.1 基本用法

from chongming_task.controller import Controller, render_table

controller = Controller(max_parallel=2)  # 最多同时跑 2 个

controller.register("extract", extract, tags={"io"})  # 先登记，之后只传名字
controller.register("transform", transform)

run = controller.submit("transform", 100, task_id="t0", label="转换-1")
controller.submit("transform", 200, task_id="t1", label="转换-2")

print(render_table(controller.list_tasks()))  # 查询
controller.stop("t0", reason="手工干预")  # 控制
controller.wait_all(timeout=60)  # 等全部落终态

8.2 查询接口

方法	作用
get(task_id)	单个快照；不存在返回 None
list_tasks(status=, name_prefix=, tag=, limit=)	按条件查询
stats()	各状态计数
runs()	全部 TaskRun 句柄（可按 id 继续控制）
active_count / queued_count	真的在占用执行额度 / 正在排队
render_table(snapshots)	渲染等宽文本表，CLI 与看板共用同一份数据

方法名是 list_tasks 而不是 list：后者会在类作用域里遮蔽内置的 list， 使类内所有 list[...] 类型注解失效（mypy 会报 "Function is not valid as a type"）。

8.3 控制接口

controller.stop(task_id, reason=...)  # → bool（是否找到该任务）
controller.pause(task_id, timeout=...)  # → bool（是否**确认**已挂起）
controller.resume(task_id)  # → bool
controller.stop_all(reason=...)  # → int（本轮**发出**的请求数）
controller.close(reason=..., timeout=...)  # 停止一切并收尾

两个容易误读的返回值：

• stop_all() 返回的是已发出请求的数量，不是"已停止"。请求发出后任务先进入 stopping，落终态是异步的；要等就接着 wait_all()。
• pause() 返回 False 表示暂停没生效（状态已回滚为 running）。

close() 会停止一切，且排队中的任务不会真的执行任务体——它们被"启动后立刻 停止"，监督循环第一轮就落 stopped。

8.4 并发额度

max_parallel 限制同时执行的任务数；超出部分排队，状态为 queued：

TASK_ID  NAME    STATUS   ATTEMPT  ELAPSED  PROGRESS  FAILURE
-------  ------  -------  -------  -------  --------  -------
t0       转换-1  running  1/1      0.4s     -         -
t1       转换-2  running  1/1      0.4s     -         -
t2       转换-3  queued   0/1      -        -         -
t3       转换-4  queued   0/1      -        -         -

queued 是"已被受理、等待执行槽"，与 pending（尚未受理）不同。排队中的任务 ELAPSED 显示 -——它还没执行过，显示一个时长会被误读成"已经跑了这么久"。

注意 stats().unfinished 是非终态数量，含排队与暂停；要知道"真的在占用执行 资源"的有几个，用 controller.active_count。

具名资源信号量

保护外部稀缺资源（连接数、配额、GPU）用 ConcurrencyPolicy.limits——按标签 记账，与全局额度叠加：

from chongming_task import ConcurrencyPolicy, Controller

controller = Controller(concurrency=ConcurrencyPolicy(max_parallel=20, limits={"db": 4, "gpu": 1}))
controller.register("sync", sync_tenant)
controller.register("train", train_model)

controller.submit("sync", "tenant-A", tags=["db"])  # 占一个 db 名额
controller.submit("sync", "tenant-B", tags=["db"])  # 再占一个
controller.submit("train", "v3", tags=["gpu"])  # 占那个唯一的 gpu 名额

规则只有三条，都好记：

• 没写进 limits 的标签不受限，打标签不会自动限流；
• 一个任务带多个受限期标签时，每个资源都要有余量才放行；
• 名额算的是活跃任务（已占执行槽的），含被暂停的——暂停不释放执行槽， 也就不该释放资源名额。

额度是本进程的账，不做跨进程协调：多节点各自配自己那一份（例如每个节点到库里 的连接数）。

优先级

priority 决定排队顺序：数值越大越先被执行，同优先级仍按提交顺序（先进先出）。

controller.submit("sync", "tenant-A", priority=10)  # 用户等着的那次同步
controller.submit("sync", "tenant-B")  # 默认优先级 0：批量回填

默认值来自 ConcurrencyPolicy.priority，submit(priority=...) 可逐条覆盖。它只在 本进程的就绪队列里比较——优先让"同一批排队任务里的急件"先走，不做跨节点抢占。

8.5 安全底线

只接受名字，绝不接受代码。 submit("extract", ...) 里的 extract 必须在启动时 register 过，否则抛 ConfigurationError。一个能启动任意可调用对象的入口等于 远程代码执行——将来接 CLI 之后这条同样不能破。

重名注册也会抛错，不静默覆盖：静默覆盖会让"我明明改了实现，跑的还是旧的"这类问题 极难发现。

8.6 跨进程限制（重要）

Controller 管理的是当前进程内的任务。另一个进程里执行 chongming-task ps 看不到它们——内存不共享。

跨进程查询与持久化需要共享媒介：

方案	引入网络层	说明
共享存储（sqlite3）	否	标准库自带，无 IPC，天然跨平台，顺带就是持久化
本地 IPC（Unix socket / 命名管道）	否	仍需写 IPC 与心跳，且平台分叉
不跨进程	否	CLI 只能管自己启动的任务，不成其为"后台"

已选定第一条。prune(keep_terminal=N) 会丢弃最旧的终态记录以防内存无限增长。

8.7 持久层抽象

持久层契约是 TaskStore，内置三个实现：

实现	适用	依赖
InMemoryTaskStore	单进程嵌入、单元测试、作为其他实现的参照物	标准库
SqliteTaskStore	跨进程可见 + 持久化（默认）	标准库 sqlite3
PostgresTaskStore	跨机器共用一套存储与指令通道	可选 chongming-task[postgres]

from chongming_task import Progress, SqliteTaskStore, TaskQuery, TaskStatus

store = SqliteTaskStore("tasks.db")
store.create(snapshot)
store.save(next_snapshot, expect_version=snapshot.status_version)  # 乐观并发
store.query(TaskQuery(statuses=frozenset({TaskStatus.RUNNING}), tag="io"))
store.touch("job-1", progress=Progress(done=120, total=500))

同机跨进程用 SQLite 就够；跨机器换成连接串即可，语义与 API 完全一致：

from chongming_task import PostgresTaskStore  # pip install "chongming-task[postgres]"

store = PostgresTaskStore("postgresql://user:pw@db.internal:5432/chongming")

核心包不因此多出任何依赖：PostgresTaskStore 是惰性取用的，没装驱动时 import chongming_task 照常工作（只有真正取它才会给出安装命令）。建表在首次连接时 自动完成；角色没有 DDL 权限时，它检测到表已存在就直接用，所以只读运维账号也能连。

四类数据的写入频次与一致性要求不同，因此刻意分开：

类别	方法	特点
任务快照	create / save / get / query	参与乐观并发（status_version）
审计轨迹	append / history	只追加，重复写入幂等
高频旁路	touch	进度与心跳，不碰版本号（否则会与状态写入持续冲突）
控制指令	submit_command / claim_commands / complete_command	跨进程控制的媒介

claim_commands 的租约解决的是"取走指令的进程崩了怎么办"：租约到期后指令可被 重新取走，因此既不会重复下发，也不会因为一次崩溃而永久卡住。

写自己的后端

别靠读文档猜语义——跑契约套件：

from chongming_task.testing.store_contract import StoreContract


class TestMyStore(StoreContract):
    def make_store(self):
        return MyStore(...)

套件随包发布且不依赖 pytest；内置实现跑的正是同一套用例（内存 / SQLite 文件 / SQLite 内存 / Postgres——最后一份需要 CHONGMING_TEST_POSTGRES 给出 DSN，没有就跳过）。

要换掉默认持久层，请看 持久层实现指南——那里有 14 个方法的逐条 语义、并发与原子性要求、序列化保真要求，以及一个可执行的第三方后端范例和陷阱清单。 PostgresTaskStore 本身就是照这份规格写出来的第二个真实后端：它的模块文档记了 "与 SQLite 的三点差异"（连接池、SKIP LOCKED、原生类型）与跨进程并发用例， 可以当成比文件后端更接近生产的参照。

已经接线：Controller(store=...) 可用——每次状态迁移写穿到存储，查询走存储 （因此能看到别的进程的任务），对不属于本进程的任务会把控制动作写成指令。详见 §8.9。

8.8 任务分桶

桶既是本地并发隔离的单位，也是将来分布式分配的单位（一台机器拥有一组桶）。

controller = Controller(max_parallel=10, max_parallel_per_bucket=1)
controller.submit("sync", "租户A", task_id="a0", bucket_key="tenant-A")
controller.submit("sync", "租户B", task_id="b0", bucket_key="tenant-B")

bucket_key 是业务键（租户 id、数据源、日期分区……）。同一个键在任何进程、任何 机器上都落到同一个桶。

成员	作用
submit(..., bucket_key=...)	提交时给出业务键
snapshot.bucket / .bucket_key	桶号与原始键都记在任务上
max_parallel_per_bucket=	每桶独立限流；一个桶积压不会饿死其他桶
list_tasks(bucket=...) / (buckets=[...])	按桶查询（buckets 是为分布式预留的：只取自己拥有的那组）
bucket_counts()	每桶任务数，用于决定把桶分给谁
active_in_bucket(bucket)	某桶当前占用的执行额度
controller.bucket_policy	分桶策略（num_buckets 等）

render_table 也带上了 BUCKET 列。

两条硬约束（详见 设计文档 §15）：

1. 桶数一旦开始用就不能随便改。 bucket = hash(key) % N，改 N 会让几乎所有键 改换桶位。所以任务会同时记下 bucket_key——将来真要改桶数，只有拿得到原始键 才能重新分桶。
2. 绝不能用内置 hash()。 CPython 对 str 的 hash() 带进程级随机盐 （PYTHONHASHSEED）：同一个键在不同进程里会落到不同的桶，于是"同桶任务在同一台 机器上"会静默失效。BucketPolicy 用 blake2b，跨进程一致。

调度器取下一个可跑任务时是扫描而不是取队首：某个桶满时队首可能正好被它挡住， 而后面属于其他桶的任务本可以先跑。

8.9 跨进程：一个进程跑，另一个进程看和管

from chongming_task import Controller, SqliteTaskStore

worker = Controller(store=SqliteTaskStore("tasks.db"), max_parallel=2, owner="worker-1")
worker.register("sync", sync_tenant)
worker.submit("sync", "租户A", task_id="job-1", bucket_key="tenant-A")

# 另一个进程（同一个 SQLite 文件，各自一条连接）
console = Controller(store=SqliteTaskStore("tasks.db"), owner="ops-console")
console.list_tasks()  # 看得到 job-1，尽管它并不持有
console.stop("job-1", reason="运维停止")  # 写成一条指令

方向	行为
写	每次状态迁移写穿到存储（乐观并发，附迁移历史）
读	get / list_tasks / stats / bucket_counts 走存储
控制	不属于本进程的任务 → 写一条指令，由持有者轮询取走

三个必须知道的行为

1. 跨进程的 pause() 返回 False。 那不是失败，是未确认——跨进程没有回执 通道。要观测结果请轮询 get()。stop() / resume() 返回的是"请求已受理"。
2. 纯查询客户端不会抢指令。 轮询线程只在提交过任务之后才启动；即便取到了不属于 自己的指令，也会通过 release_command 立即交还，而不是标记完成（那会让真正的 持有者永远收不到）。
3. 存储故障不让任务失败，但会记进 controller.persist_failures（最多 20 条）。 任务本身没问题，坏的只是可观测性——但绝不静默。

两个使用注意

• 用完一定要关存储：controller.close(close_store=True)，或自己 store.close()。 没关的 SQLite 连接会泄漏文件句柄，在 Windows 上还会让数据库文件无法删除。
• stop_all() 只作用于本进程持有的任务。"全部停止"跨进程需要一份全局任务清单， 那是分布式调度该决定的事。

controller.owner 是写进指令里的发起者标识（默认 主机名:pid），便于事后追溯是谁 按的停止。

8.10 命令行控制台

chongming-task 是 Controller 之上的一层薄适配：解析参数、调用库、渲染输出。 它不引入任何库没有的概念，也不绕过任何不变量。

chongming-task --db tasks.db ps
chongming-task --db tasks.db status demo-1
chongming-task --db tasks.db stop demo-1 --reason "发布窗口"

命令	作用
tasks --tasks MODULE	列出模块里注册的任务定义
run TASK --tasks MODULE	提交并前台执行到落终态
ps / status / history / stats / buckets	查询
stop / pause / resume	控制

退出码分三档：0 成功、1 命令失败、3 命令成功但任务没成功——最后一档是给 脚本判分支用的。

三点与库行为一致、但在命令行里更容易踩到的地方：

• run 是前台进程，退出时任务就没人管了；长期运行请交给 nohup / systemd。
• pause 跨进程只报"未确认"（退出码仍是 0），要看结果得 status。
• 任务定义必须来自模块的 register(controller)，CLI 只传名字——一个能启动任意 可调用对象的入口等于任意代码执行。

完整走查（含真实输出）见 命令行控制台。

8.11 重启恢复（reconcile）

进程意外退出后，存储里会留下一批永远停在 running 的任务——它们看着像在跑，其实 载体早就随进程消失了。进程启动时先跑一次 reconcile() 把它们清理掉：

controller = Controller(store=SqliteTaskStore("tasks.db"), owner="worker-2")
report = controller.reconcile()
if report.lost:
    log.warning("上次有 %d 个任务失去监督者", len(report.lost))

判据是"还有人在管它吗"，不是"它还在推进吗"

每个监督进程都会给自己负责的任务续租（lease_owner / lease_until，由 touch 写入）。于是：

情况	处置
租约未过期	有活着的进程在管，一动不动
租约过期 + 任务占着执行资源（starting/running/pausing/stopping）	载体已消失，判 LOST
租约过期 + 任务不占资源（queued/paused/retry_wait）	只是没人接手，不改状态，报告为 adoptable
原本的监督者又活过来抢先写入	检测到版本冲突，放手（报 skipped）

绝不能用任务心跳当判据。 last_heartbeat_at 由任务自己的 checkpoint() 驱动， 一个长时间不调 checkpoint 的健康任务（纯 CPU 计算）心跳同样停滞——拿它判断进程存活， 会把一个正常任务判成 LOST 并杀掉。这正是租约与心跳必须分成两个字段的原因。

report = controller.reconcile(dry_run=True)  # 先看一眼，不写任何状态
report.lost  # 会被（或已被）判为丢失的
report.adoptable  # 需要重新提交的
report.skipped  # 有人正在写、被跳过的

它不会重新执行任务

这是能力边界，不是遗漏：存储里记的是状态，不是配方。任务的函数与参数从未落库 （序列化任意参数会带来不可反序列化、版本漂移等一堆问题），所以恢复一个失去载体的 任务在原理上不可能。

reconcile() 能做的、也是它应当做的，是如实标记——让看板不再永远显示一个假的 "运行中"。真正"重跑"要靠调用方自己重新 submit，而那需要调用方知道参数。

lease_ttl（默认 30 秒，可调）决定崩溃多久之后会被发现。

8.12 DAG 管道

把多个任务按依赖关系串起来：

from chongming_task import Pipeline
from chongming_task.policies import AggregationPolicy, JoinMode

pipeline = Pipeline("nightly-etl", aggregation=AggregationPolicy.all_settled())

fetch = pipeline.node("fetch", "2026-10-02", node_id="fetch")
shards = [
    pipeline.node("transform", index, node_id=f"shard{index}", depends_on=[fetch])
    for index in range(3)
]
load = pipeline.node("load", node_id="load", depends_on=shards)
pipeline.node("audit", node_id="audit", depends_on=[load], join=JoinMode.ALL_SETTLED)

run = controller.submit_pipeline(pipeline)
outcome = run.wait(timeout=60)
print(outcome.status, outcome.nodes)

第一个位置参数是已注册的任务名，其后才是传给任务的位置参数。节点 id 由 node_id= 指定；省略时默认取任务名，同名多次则自动加 #2 / #3 后缀。 p.node("mark", "after") 里的 "after" 是传给 mark 的参数，不是节点名—— 这正是那个参数叫 node_id 而不是 name 的原因。

三种汇合语义（逐节点）

JoinMode	何时放行
ALL_SUCCEEDED（默认）	全部上游成功——最常见的屏障
ALL_SETTLED	全部上游落终态即可，不管成败——用于清理、通知、汇总
ANY_SUCCEEDED	任一上游成功即可——多源冗余取数的先到先得

两种结算方式（整条管道）

AggregationPolicy	行为
fail_fast()（默认）	首个真失败即取消其余已经放行的节点
all_settled()	等所有节点落终态，再统一判定

quorum 会被明确拒绝：它描述的是"同一件事的 N 份副本成功 k 份"，属于扇出分组而不是 一条节点各异的 DAG。与其发明一个没人能预测的语义，不如说清楚。

等待上游的节点不占并发额度

依赖由一个真正的任务状态 blocked 表达，而且等待中的节点不进待执行队列：

assert controller.active_count == 1  # 只有根节点占额度
assert controller.queued_count == 0  # 下游还在 blocked，不是 queued

一条 20 个节点的串行管道不会因为"等待"而把额度全锁死。

三种"没成功"必须分得清

状态	含义
skipped	上游没成功，它本来就不该跑
cancelled	本来要跑，被 fail_fast 连坐了
stopped	值班同学手动停的

CANCELLED 刻意不作为 fail_fast 的触发条件，否则用户主动取消整条管道会被记成"失败"， 被连坐的节点也会再触发一轮连坐，因果就搅成一团了。

限制

管道结构在内存里，不落库（任务的函数与参数从未持久化）。进程重启后一条跑到一半的 管道无法续跑——节点会被 reconcile() 如实标记为丢失。

8.13 写一个任务：@task 与 TaskContext

@task 装饰器

from chongming_task import RetryPolicy, Timeout, task


@task(tags={"io"}, retry=RetryPolicy(max_attempts=3), timeout=Timeout(per_attempt=60))
def extract(ctx, source: str) -> list[Row]:
    """从 source 读取数据。"""
    ...

描述默认取 docstring 第一行，名字默认取函数名。裸 @task（不带括号）也支持。

一次登记整个模块：

controller.register_module(myapp.tasks)  # 模块里所有 @task 都被登记
controller.register_all(extract, load)  # 或者逐个登记

它不产生全局注册表。 规格只挂在函数对象上（__chongming_task__），谁注册、注册到 哪个控制器由调用方决定。全局注册表意味着 import 即注册，而 import 可能发生在任何 地方（包括测试收集阶段），那种隐式行为极难排查。

TaskContext

任务作者面对的就是这一个对象（公开名字 TaskContext，与内部的 RuntimeContext 是 同一个类）：

成员	作用
ctx.checkpoint()	协作点：响应停止与暂停 + 刷新心跳。不调用它就不响应停止，也会被 idle 超时判为疑似卡死
ctx.acheckpoint()	异步版本，async def 任务里 await 它（执行器会把协程真的跑完）
ctx.progress(done, *, total=, unit=, message=)	上报进度
ctx.log(msg, *, level=) / .warning / .error	记日志
ctx.logs	本任务最近 200 条日志（内存中）
ctx.input(节点id, 默认值) / ctx.upstream	取上游节点的返回值
ctx.artifact_dir	本次尝试的产物目录（按 attempt 隔离）
ctx.is_stop_requested / ctx.is_pause_requested	只想探一下、不想抛异常时用

进度：进快照、进存储、不碰版本号

for index, part in enumerate(partitions, start=1):
    ctx.checkpoint()
    ctx.progress(index, total=len(partitions), unit="partitions", message=part)

进度走高频旁路：不参与状态机、不递增 status_version，由 Controller 经 touch 定期回写到存储。所以它出现在 ps 的 PROGRESS 列，而不会把版本号变成噪声。

progress() 刻意不刷新心跳。 心跳是"我还在推进"的整体信号，进度是"推进到哪了"。 把心跳藏进进度里，会让一个只报进度、不做实际工作的任务看起来永远健康——那正是 idle 超时要抓的东西。

日志：走标准库

ctx.log("开始处理")  # INFO
ctx.warning("分区为空，跳过")
ctx.error("写入失败")

logger 名是 chongming_task.task.<task_id>，并且 task_id / task_name / attempt 会作为 extra 附在 LogRecord 上，因此可以按任务维度配 handler、过滤或加字段：

logging.getLogger("chongming_task.task").setLevel(logging.DEBUG)

ctx.logs 保留最近 200 条（内存中，进程退出即消失）。它不是日志存储方案——那属于 应用自己的事，挂个 handler 就能落盘或入库。

上游数据

管道的下游节点可以直接取上游的返回值：

@task
def transform(ctx, day: str) -> int:
    partitions = ctx.input("extract", [])  # extract 节点的返回值
    ...

只有成功的上游才会出现在 ctx.upstream 里。失败或跳过的上游没有值可传——塞一个 None 会让下游以为"上游返回了 None"，而不是"上游没有值"。

返回值必须可 JSON 序列化

返回值会写入存储（outcome 列），所以它必须是 JSON 可序列化的：

@task
def good(ctx) -> dict[str, Any]:
    return {"rows": 3, "path": "s3://bucket/x.parquet"}  # ✅


@task
def bad(ctx) -> set[int]:
    return {1, 2, 3}  # ❌ set 不可序列化

违反时任务会失败，失败类型是 result_invalid，消息写明原因。大数据请传指针 （对象存储 key、表名 + 主键、文件路径）而不是内容。

为什么这么严：返回值不落库的话，进程内会显示 succeeded 而存储里停在 running ——随后重启恢复会把一个真正跑完的任务判成 LOST。把一个成功过的任务报成"结果 未知"会让人放心地重跑，而副作用已经发生过一次。所以宁可让它在返回时就响亮地失败。

标签同理：tags 必须是字符串（类型契约是 frozenset[str]），否则在提交时就会被拒绝。

参数暂时不强制可序列化——参数今天不落库，所以 Path、连接对象这类本地专用参数 仍然合法。（节点之间不互相执行对方的任务，所以也不存在"参数要跨机器传输"的问题。）

8.14 节点桶归属：多节点部署

多个节点各跑各的任务时，靠桶区分"哪一份是我的"：

controller = Controller(
    store=SqliteTaskStore("tasks.db"),
    owner="node-a",
    owned_buckets=range(0, 16),  # 本节点负责 0..15 号桶（按业务划分，可多于一个）
)

成员	说明
owned_buckets	本节点负责的桶；None（默认）表示不限定
owns_all_buckets	是否不构成划分：不限定，或显式覆盖了全部桶
owns_bucket(bucket)	某个桶是否归本节点

限定时有两条行为：

1. 提交时校验——任务必须落在自己的桶里：

   controller.submit("sync", bucket_key="tenant-B")
   # ConfigurationError: 桶 42（来自 bucket_key='tenant-B'）不属于本节点；
   # 本节点负责 [0, 1, ..., 15]。任务若写进别人的桶，不会有任何节点执行它。
   

   这是硬错误而不是警告：落在别人桶里的任务不会有任何节点去跑它，它会永远停在 queued，而且没有任何报错。与其让它静默堆积，不如在提交那一刻就说清楚。

2. reconcile() 只清理自己桶里的僵尸任务，不去动别的节点的任务。

覆盖全部桶不算划分。 如果显式列举了全部桶（例如生成出来的"负责 0-63"配置），此时没有 任何"别的节点"，无桶任务同样会且只会由本节点执行——因此不再强制 bucket_key。

启动时会校验桶号范围，越界立刻报错：放过去的症状同样是"任务永远排队且没有报错"。

声明了归属的节点会在构造那一刻就启动指令轮询，而不是等第一次提交。否则别人 stop 一个还排在队列里的任务时，指令要等到本节点第一次提交才被取走——对定时任务节点来说可能 是几小时之后。没有声明归属的纯查询客户端仍然不轮询，避免抢走别人的指令。

多节点部署的完整说明（能做什么、不能做什么、新增的风险、跨机部署）见 分布式设计说明。跨机只需换存储后端：把 SqliteTaskStore("tasks.db") 换成 PostgresTaskStore("postgresql://...")， owner / owned_buckets / reconcile() / 跨节点控制全部照旧（见 §8.7）。

一份可以直接照抄的完整模板（部署参数与桶划分 buckets_for_node()、任务定义、 build_node() 接线、启动恢复、按桶提交、跨节点停止、收尾校验，全流程带断言）见 examples/11_best_practice.py。

---

9. 状态模型参考

9.1 状态一览

非终态（9 个）：

状态	含义
pending	已创建，依赖尚未解析
blocked	依赖未满足，等待上游
queued	就绪，等待执行槽；未启动的暂停也在恢复时回到这里
starting	执行器启动中
running	正在执行
pausing	已请求暂停，等待到达静默点
paused	已就地挂起
retry_wait	本次尝试失败，正在退避等待
stopping	已请求停止，处于优雅窗口内

终态（6 个，无出边）：

状态	含义
succeeded	正常完成
failed	出错且重试耗尽，或遇到不可重试的错误
stopped	人工主动停止（不是错误）
cancelled	编排层级联取消（如兄弟任务失败）
skipped	条件为假或依策略跳过
lost	载体失联或被放弃，结果未知

stopped 与 cancelled 分开，是因为两者责任方不同。排障时"这次跑批失败是人工 干预还是真故障"是个必须能回答的问题。

9.2 完整转移表

PENDING:    {BLOCKED, QUEUED, SKIPPED, CANCELLED, STOPPED}
BLOCKED:    {QUEUED, SKIPPED, CANCELLED, STOPPED}
QUEUED:     {STARTING, PAUSED, SKIPPED, STOPPED, CANCELLED}
STARTING:   {RUNNING, RETRY_WAIT, PAUSING, STOPPING, FAILED, CANCELLED, LOST}
RUNNING:    {SUCCEEDED, FAILED, RETRY_WAIT, PAUSING, STOPPING, CANCELLED, LOST}
PAUSING:    {PAUSED, RUNNING, STOPPING, FAILED, CANCELLED, LOST}
PAUSED:     {QUEUED, RUNNING, STOPPED, CANCELLED}
RETRY_WAIT: {STARTING, QUEUED, PAUSED, STOPPED, FAILED, CANCELLED}
STOPPING:   {STOPPED, SUCCEEDED, FAILED, LOST}
SUCCEEDED / FAILED / STOPPED / CANCELLED / SKIPPED / LOST:  {}

一条贯穿全表的规则：控制动作是否需要中间态，取决于任务是否已占用执行资源。 未占用（pending/blocked/queued/paused/retry_wait）→ 立即生效；已占用 （starting/running/pausing）→ 经中间态缓冲。让未启动的任务显示"停止中/暂停中" 是虚假信息，会让运维误以为需要等待。

9.3 转移表之外的守卫

有些规则无法用"状态 → 状态"表达，它们是守卫：

动作	守卫	理由
mark_running()	必须已有 started_at	paused → running 对就地恢复合法，对"排队期间被暂停、从未启动"却毫无意义
abort_pause()	必须处于 pausing	paused → running 是给恢复用的；把"回滚"也放进来会产出没有开始时间的 running 快照
mark_starting()	attempt + 1 <= max_attempts	否则抛 AttemptsExhausted（预算问题，不是状态表问题）
force_transition()	必须给出非空 actor 与 reason	这是唯一能任意跳转的入口，没有审计信息就无法事后追溯
mark_lost()	建议带 failure	lost 的全部意义是"不知道结果"，不记录原因就无从判断该重跑还是人工核对

---

10. 错误与异常

TaskError
├─ InvalidTransition      非法状态转移（含守卫不通过）
├─ DependencyFailed       上游未成功
├─ CircularDependency     依赖成环
├─ TaskTimeout            超时
├─ TaskExecutionError     用户代码异常的统一包装（保留 __cause__）
├─ AttemptsExhausted      重试预算耗尽
├─ ConcurrencyConflict    乐观并发写冲突
├─ StoreError             存储层故障
└─ ConfigurationError     配置错误

TaskCancelled             ← 独立：继承 BaseException，不是 TaskError

9.1 TaskCancelled 为什么继承 BaseException

因为它可能出现在用户代码的任何一行，而下面这种写法极其常见：

try:
    while True:
        ctx.checkpoint()
        work()
except Exception as exc:  # 想捕获业务异常
    log(exc)  # 但会把停止信号一起吃掉

若 TaskCancelled 继承 Exception，任务将永远无法被停止。因此它和 asyncio.CancelledError 一样继承 BaseException：

try:
    ...
except TaskCancelled:
    raise  # 取消要放行
except Exception as exc:
    ...  # 只处理真正的业务异常

测试里有一条用例专门用 except Exception 包裹循环来验证取消不会被吞掉。

---

11. 限制与 FAQ

Q：任务停不下来？ 检查它是否调用了 ctx.checkpoint()。不调用的任务只能等超时，然后落 lost。

Q：为什么我的任务被判"疑似卡死"（timeout_idle）？ idle 衡量的是 checkpoint() 间隔。长段无 checkpoint() 的计算会被误判——调大 idle 阈值，或提高 checkpoint() 频率。

Q：pause() 返回 False 是什么情况？ 任务在窗口内没到达检查点，暂停没生效，状态已回滚为 running。这是如实告知， 不是 bug。

Q：暂停会不会释放并发额度？ 不会。暂停是就地挂起，槽位仍被占用。

Q：任务被判 lost 之后怎么办？ lost 表示结果未知。系统不会替你猜——你可以 state_machine.retry() 重跑，也可以 人工核对副作用后 force_transition() 标记终态。

Q：wait(timeout=...) 超时了，任务会被杀掉吗？ 不会。它只抛 TimeoutError，任务状态完全不变。要停止请显式 stop()。

Q：能让任务跑到一半重启而不重跑吗？ 不能。没有用户态检查点机制（那属于 Dask/Ray 式的图重算，是明确的非目标）。重试 总是从头开始，请把任务设计成幂等或可断点续传。

---

12. 测试自己的任务

确定性时钟随包发布，你的测试不必 sleep：

from chongming_task.testing import FakeClock
from chongming_task.run import run_task
from chongming_task.policies import Timeout


def test_idle_timeout_fires_without_waiting():
    clock = FakeClock()
    # 通过 clock 注入控制时间，配合 handle.heartbeat 即可在毫秒内验证超时逻辑

常用做法：

• FakeClock 注入 TaskRun(..., clock=clock)，用 clock.advance(n) 推进时间；
• InlineExecutor 让执行同步化，测试不必等线程；
• RetryPolicy(backoff=FixedBackoff(0)) 去掉重试等待；
• 断言 run.transitions，而不是只断言最终状态——中间态才是本项目的价值所在。

---

下一步

已实现的能力以 README 状态表为准：单任务与 DAG 管道执行、 并发额度与排队、SQLite / Postgres 持久化与重启恢复、@task 装饰器与 TaskContext （log / progress / upstream）、CLI 控制台、节点桶归属。

仍在计划中的部分：跨机的运维告警（权威桶声明、桶重复/空洞、孤儿任务，见 分布式设计说明 §11.2–11.3）与 transitions 的归档策略 （规模上 prune() 不够用，见 持久层实现指南）。

设计细节与理由见 设计文档。