chongming-task (0.1.1)

Published 2026-10-11 21:20:17 +08:00 by chakcy

To install the package using pip, run the following command:

pip install --index-url http://124.71.68.6:3000/api/packages/chongming/pypi/simple/ --extra-index-url https://pypi.org/simple chongming-task

For more information on the PyPI registry, see the documentation.

精细化状态控制的 Python 管道任务控制工具

chongming-task

精细化状态控制的 Python 管道任务控制工具。

任务在这里不是一个"跑着 / 没跑"的布尔值，而是一条可审计的状态轨迹：每个中间态（排队、启动、运行、暂停中、重试等待、停止中）都能被表达、被观察、被干预，而且 结果永远如实上报。

当前状态（务必先读）

这是一个正在实现中的库。已经能用的部分与还没做的部分如下：

能力	状态
任务状态模型与合法转移表（15 个状态）	✅ 可用
单任务受控执行（`run_task` / `TaskRun`）	✅ 可用
停止 / 暂停 / 恢复	✅ 可用（协作式）
失败重试与退避	✅ 可用
三种超时（启动 / 单次尝试 / 静默）	✅ 可用
线程执行器（含 `async def` 任务）	✅ 可用
内联执行器（同步、零开销）	✅ 可用
进程执行器（硬性停止 / 绕开 GIL / 崩溃隔离）	✅ 可用
状态迁移历史（可审计）	✅ 可用
任务注册表 + 统一管理门面（`Controller`）	✅ 可用
并发额度与排队、具名资源信号量（按标签）、优先级、文本表格渲染	✅ 可用
任务分桶：稳定哈希 + 每桶独立限流 + 按桶查询（为分布式预留分配单位）	✅ 可用
节点桶归属：`owned_buckets` 划界 + 提交校验 + `reconcile` 范围	✅ 可用
持久层契约 `TaskStore`（内存 / SQLite / Postgres 实现 + 契约测试套件）	✅ 可用
跨机持久层 `PostgresTaskStore`（`SKIP LOCKED` 认领 + 连接池 + 跨进程并发用例）	✅ 可用
存储接线：写穿 / 读穿 / 跨进程查询与控制	✅ 可用
重启恢复：监督者租约 + `reconcile()` 清理僵尸任务	✅ 可用
DAG 管道：依赖、扇出/扇入、三种汇合语义、fail_fast 连坐	✅ 可用
`@task` 装饰器、`TaskContext`（进度 / 日志 / 上游数据）	✅ 可用
存储与 `Controller` 接线、跨进程查询、重启恢复	✅ 全部可用
CLI（`run` / `ps` / `status` / `stop` / `pause` / `resume` / `stats` / `buckets`）	✅ 可用

默认线程执行器，另备进程执行器（外加测试用的内联执行器）。

线程共享内存，可以直接复用连接池、缓存、已加载的模型，也没有参数必须可 pickle 的约束——闭包、lambda、@task 装饰器产物都能用。代价是强制停止做不到：优雅窗口用尽后只能放弃等待并记 lost（结果未知），绝不谎报"已停止"。

需要硬性停止保证、绕开 GIL、或崩溃隔离时用 ProcessExecutor()——任务跑在独立子进程里，优雅窗口用尽会 terminate() → kill()，段错误也只带走子进程。代价是三条结构性约束：参数与返回值必须可 pickle、目标函数必须在可导入的模块里（且调用方脚本要有 if __name__ == "__main__": 守卫）、停止与暂停有约 20ms 的轮询延迟。详见用户指南 §7。

两种执行器下都请让任务调 ctx.checkpoint()，并写成幂等或可断点续传。

完整设计（状态表、分层、里程碑、风险）见设计文档；面向使用者的完整说明见用户指南。

安装

uv sync              # 开发环境
# 或
pip install -e .

核心包零第三方运行时依赖，只用标准库。

跨机部署共用一套持久层时，再装一个可选 extra：

pip install "chongming-task[postgres]"

它只带来 psycopg（Postgres 驱动）。不装也能正常 import chongming_task—— PostgresTaskStore 是惰性取的，不装时取它会报出安装命令而不是一个光秃秃的 ModuleNotFoundError（见持久层抽象）。

快速上手

examples/01_quickstart.py：

import time
from chongming_task.run import TaskRun


def extract(ctx, source: str) -> list[int]:
    """长耗时循环里必须周期性地调用 ctx.checkpoint()。"""
    rows: list[int] = []
    for index in range(5):
        ctx.checkpoint()  # 响应停止/暂停，并刷新心跳
        time.sleep(0.05)
        rows.append(index)
    return rows


run = TaskRun(extract, ("s3://demo/rows",), task_id="demo-1", name="抽取")
run.start()
outcome = run.wait(timeout=30)

for record in run.transitions:
    print(f"{record.from_status.value:>10} -> {record.to_status.value:<10} by {record.trigger}")

print(outcome.status.value, outcome.value)

真实输出：

  抽取完成：s3://demo/rows -> 5 行

状态迁移历史（只追加、可审计）：
     pending -> queued     by orchestrator:dependencies_resolved
      queued -> starting   by scheduler:dispatch
    starting -> running    by executor:launched
     running -> succeeded  by executor:completed

最终状态：succeeded
返回值  ：[0, 1, 2, 3, 4]
耗时    ：0.25s

只想跑一个任务、不需要中途干预时，用一行：

from chongming_task.run import run_task

outcome = run_task(extract, "s3://demo/rows", task_id="demo-1", name="抽取")
if outcome.ok:
    print(outcome.value)

任务怎么写

唯一约定：长耗时循环里周期性调用 ctx.checkpoint()。

def my_task(ctx, arg1, *, kwarg1=1):
    for row in source:
        ctx.checkpoint()  # ← 停止、暂停、心跳都在这里生效
        process(row)
    return result

ctx 由执行器注入为第一个位置参数，包含：

成员	说明
`ctx.checkpoint()`	协作点：响应停止（抛 `TaskCancelled`）、响应暂停（挂起）、刷新心跳
`ctx.acheckpoint()`	同上，在 `async def` 任务里 `await`（执行器会把协程真的跑完）
`ctx.is_stop_requested`	只读，是否已请求停止（用于清理逻辑等不便阻塞处）
`ctx.is_pause_requested`	只读，是否已请求暂停
`ctx.task_id` / `ctx.name` / `ctx.attempt`	身份与重试次数
`ctx.heartbeat`	心跳对象，`idle` 超时依据它判定
`ctx.artifact_dir`	本次尝试的产物目录（按尝试隔离，重试不污染上一次）

不调用 checkpoint() 的任务，既无法优雅停止，也无法暂停。 这是协作式设计的硬约束——见用户指南。

控制任务

run = TaskRun(my_task, task_id="job-1")
run.start()

run.pause()  # 任务在下一个 checkpoint 挂起 → paused
run.resume()  # → running
run.stop(reason="发布窗口")  # → stopping → stopped
outcome = run.wait(timeout=60)

几个刻意设计的语义，都是"不骗人"的结果：

pause() 返回 False = 暂停没生效（任务没到达检查点），此时状态回滚为 running，绝不假装已暂停。

stop() 停不掉时不会说"已停止"。线程无法强杀，因此优雅窗口用尽后任务落 lost（结果未知），而不是 stopped：

  == 停不掉的任务：如实上报 ==
  终态          ：lost
  失败归类      ：lost
  含义          ：线程无法强杀，任务可能仍在后台产生副作用，因此不敢说'已停止'

wait(timeout=...) 超时只抛 TimeoutError，不改变任务状态。把"我没等到" 当成"任务失败"是最常见的误用。
pause() 不释放并发槽位。我们无法把任意的 Python 调用栈换出内存再换回来，所以暂停是就地挂起；暂停大量任务可能耗尽并发额度。

管理一批任务

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

from chongming_task.controller import Controller, render_table

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

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

controller.submit("transform", 100, task_id="t0", label="转换-1")
controller.submit("transform", 200, task_id="t1", label="转换-2")
controller.submit("transform", 300, task_id="t2", label="转换-3")
controller.submit("transform", 400, task_id="t3", label="转换-4")

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

真实输出（examples/04_controller.py，节选）：

--- 查询：运行中的视图 ---
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      -        -         -
    未结束 4（其中排队 2，占用执行额度 2）

三个必须知道的设计点：

只接受名字，绝不接受代码。 submit("extract", ...) 里的 extract 必须已在启动时 register 过，否则抛 ConfigurationError。一个能启动任意可调用对象的入口等于远程代码执行——这条在接 CLI 之后同样不能破。
超限是排队，不是拒绝，也不是无限起线程。 排队的任务状态为 queued （已被受理、等待执行槽），与 pending（尚未受理）不同。排队中的任务 ELAPSED 显示 -，因为它还没执行过。
stop_all() 返回的是"已发出请求"的数量，不是"已停止"。 请求发出后任务先进入 stopping，落终态是异步的；要等就接着 wait_all()。

跨进程限制：Controller 管理的是当前进程内的任务。另一个进程里的 chongming-task ps 看不到它们——内存不共享。跨进程查询与持久化需要共享媒介：同机用 sqlite3（它在标准库里，不引入任何第三方依赖），跨机用 PostgresTaskStore（可选 extra）。

任务分桶

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

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")

真实输出（examples/05_buckets.py，节选）——A 桶积压，B 桶照常并行：

分桶策略：64 个桶
并发限制：全局 10，但每个桶最多 1 个

业务键 -> 桶号（同一个键在任何进程、任何机器上都会落到同一个桶）：
  tenant-A   -> bucket 7
  tenant-B   -> bucket 28

--- 查询：A 桶积压，但不影响 B 桶 ---
TASK_ID  NAME    BUCKET  STATUS     ATTEMPT  ELAPSED  PROGRESS  FAILURE
-------  ------  ------  ---------  -------  -------  --------  -------
a0       A-同步  7       succeeded  1/1      0.4s     -         -
a1       A-同步  7       running    1/1      0.1s     -         -
a2       A-同步  7       queued     0/1      -        -         -
a3       A-同步  7       queued     0/1      -        -         -
b0       B-同步  28      succeeded  1/1      0.4s     -         -
b1       B-同步  28      running    1/1      0.1s     -         -
    占用额度 2，排队 2，A 桶在用 1

两条硬约束

一、桶数一旦开始用就不能随便改。 bucket = hash(key) % N，改 N 会让几乎所有键改换桶位。所以任务会同时记下 bucket_key——将来真要改桶数，只有拿得到原始键才能重新分桶。

二、绝不能用内置 hash()。 CPython 对 str 的 hash() 带进程级随机盐 （PYTHONHASHSEED）：同一个键在不同进程里会落到不同的桶，于是"同桶任务在同一台机器上"会静默失效——没有任何报错，只是本地性与缓存复用全都不成立。 BucketPolicy 用 blake2b，并有真实子进程 + 不同 PYTHONHASHSEED 的用例守着。

多节点部署：自治节点 + 统一平面

每个节点有自己的任务定义与触发方式；节点之间不互相执行对方的任务；但共用同一套存储与控制层，靠桶区分"哪一份是我的"。

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

维度	自治（每节点自有）	统一（全局共享）
任务定义、触发方式、负责的桶	✅	—
存储（任务状态、审计轨迹、指令）、控制层	—	✅ 一套

owned_buckets 的作用是划界：

提交时校验——任务落进别人的桶会在提交那一刻报错。否则它会永远停在 queued，而且没有任何报错；
reconcile() 只看自己的桶——不去动别的节点的任务。

# 运维：只看"我这一份"
chongming-task --db tasks.db --num-buckets 64 --owned-buckets 0-15 ps --all --mine

# 每个桶多少任务；星号标出本节点负责的桶
chongming-task --db tasks.db --num-buckets 64 --owned-buckets 0-15 buckets

统一平面给的是看得见与管得着：一个看板看全部、跨节点 stop/pause（走已有的指令通道）、统一审计轨迹、以及发现"某个桶没人管、任务在静默堆积"——这是分散存储做不到的。

跨机器用 Postgres：SQLite 文件不能跨机共享，把 store 换成 PostgresTaskStore 即可，其余代码（owner、owned_buckets、reconcile()、跨节点 stop）一行都不用改。

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

controller = Controller(
    store=PostgresTaskStore("postgresql://user:pw@db.internal:5432/chongming"),
    owner="node-a",
    owned_buckets=range(0, 16),
)

CLI 同样认连接串，运维不必在每个节点上找那个 .db 文件：

chongming-task --db 'postgresql://user:pw@db.internal:5432/chongming' --num-buckets 64 ps --all

它实现的是同一个 TaskStore 契约，通过 StoreContract 全绿，并额外带着自己的并发证据（SKIP LOCKED 认领、并发 save 恰好一个赢家、持有者被强杀后租约照常到期）。细节与建表语句见持久层抽象与持久层实现指南 §6。

这个架构的代价是没有东西替你检查配置：两个节点配了同一个桶（可能重复执行）或某个桶没人配（任务静默堆积）都不会自动报错。缓解手段与其它风险见分布式设计说明 §11。

早先设计过一个真正的集群方案（任意节点可跑任意任务：配方落库、跨节点原子认领、成员表、 rendezvous 分配、再均衡、fencing）。它被否决——定时任务是有归属的，不是无差别算力池。被否决的理由与仍然成立的结论见该文 §17。

嵌入到你自己的应用

它是库优先的：Controller 就是集成点，CLI 只是可选外壳。不需要单独进程、不需要网络层、不需要先起服务。

from chongming_task import Controller, ProcessExecutor, ThreadExecutor

controller = Controller(max_parallel=4, executor=ThreadExecutor())  # 默认：线程
# 需要硬性停止 / 绕开 GIL / 崩溃隔离时换成子进程：
# controller = Controller(max_parallel=4, executor=ProcessExecutor())

controller.register("extract", extract)  # 启动时登记一次
run = controller.submit("extract", "s3://bucket/x", task_id="job-1")

# 你的 Web 框架 / 定时任务 / 消息消费者里都可以直接用：
controller.list_tasks()
controller.get("job-1")
controller.stop("job-1", reason="用户取消")
controller.stats()

完整的一条集成主线（含多节点桶划分、启动恢复、跨节点控制、优雅退出）见 examples/11_best_practice.py —— 照抄它比照抄本节的片段更省事。

公开 API 一行导入，__all__ 里就是全部受支持的表面：

from chongming_task import (  # noqa: F401
    Controller,
    TaskRegistry,
    TaskRun,
    run_task,
    render_table,
    TaskStatus,
    TaskSnapshot,
    TaskStore,
    InMemoryTaskStore,
    SqliteTaskStore,
    RetryPolicy,
    ExponentialBackoff,
    Timeout,
    ThreadExecutor,
)

持久层抽象

TaskStore 是持久层的唯一契约，第三方后端（Redis、对象存储……）只需满足它。内置三个实现：

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

from chongming_task import InMemoryTaskStore, 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))  # 高频旁路

跨机器时换成 DSN 就完了——同一个契约，同一套语义：

from chongming_task import PostgresTaskStore

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

它建表所需的三张表（tasks / transitions / commands）在首次连接时自动创建；若当前角色没有 DDL 权限，它检测到表已存在就直接使用——只读运维账号因此也能连。 close() 幂等，关闭后再用会抛 StoreError 而不是未定义行为。

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

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

自证契约

写一个新后端时别靠读文档猜语义——跑契约套件。完整的接口规格见持久层实现指南（14 个方法的逐条语义、并发要求、陷阱清单，以及一个可执行的第三方后端范例）：

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，没有就干净跳过）。

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

Controller(store=...) 一旦给出，本实例就成为可跨进程的一环：

方向	行为
写	每次状态迁移写穿到存储（乐观并发，带迁移历史）
读	`get` / `list_tasks` / `stats` 走存储，因此能看到别的进程的任务
控制	对不属于本进程的任务，`stop` / `pause` / `resume` 写成一条指令，由持有者轮询取走

真实输出（examples/06_cross_process.py，节选）：

--- ops-console 看到的状态（它并不持有这些任务）---
TASK_ID  NAME    BUCKET  STATUS   ATTEMPT  ELAPSED  PROGRESS  FAILURE
-------  ------  ------  -------  -------  -------  --------  -------
job-1    A-同步  7       running  1/1      0.5s     -         -
job-2    B-同步  28      running  1/1      0.5s     -         -
    它本地的任务句柄数：0

>>> ops-console 跨进程停止 job-1
    下发结果：True
    worker-1 已取走指令并停止该任务

--- job-1 的迁移历史（从存储读，可审计）---
     1     pending -> queued     by orchestrator:dependencies_resolved
     2      queued -> starting   by scheduler:dispatch
     3    starting -> running    by executor:launched
     4     running -> stopping   by user:stop
     5    stopping -> stopped    by runtime:stop_confirmed

三个刻意的设计点

纯查询客户端不轮询指令。 轮询线程只在提交过任务之后才启动。否则一个只读的看板会把别人的停止请求抢走——即便它把指令交还，持续轮询也会让真正的持有者反复错过。
取到不属于自己的指令是"交还"，不是"完成"。 存储契约里有 release_command：标记完成会让真正的持有者永远收不到这次请求（一个静默的丢指令）；什么都不做、只等租约过期又会被持续续租饿死。所以需要第三种结局。
存储故障不让任务失败。 任务本身没问题，坏的只是可观测性。但也绝不静默—— 写穿失败记进 controller.persist_failures，让"看板上的状态是旧的"有迹可循。

两个使用注意

跨进程的 pause() 返回 False。 那不是失败，是未确认——跨进程没有回执通道。要观测结果请轮询 get()。
用完一定要关存储。 controller.close(close_store=True) 或自己 store.close()。没关的 SQLite 连接会泄漏文件句柄，在 Windows 上还会让数据库文件无法删除。

命令行控制台

CLI 是另一个进程，所以它只能通过共享存储看和管——这正是持久层存在的理由。

# 一个终端里跑（前台执行到结束）
chongming-task --db tasks.db run --tasks myapp.tasks sync --arg tenant=A

# 另一个终端里看和管
chongming-task --db tasks.db ps
chongming-task --db tasks.db status demo-1
chongming-task --db tasks.db stop demo-1 --reason "发布窗口"

真实输出（examples/tasks_demo.py + PYTHONPATH=examples，节选）：

$ chongming-task --db tasks.db ps
TASK_ID  NAME    BUCKET  STATUS   ATTEMPT  ELAPSED  PROGRESS  FAILURE
-------  ------  ------  -------  -------  -------  --------  -------
demo-1   A-同步  7       running  1/1      7.4s     -         -

$ chongming-task --db tasks.db history demo-1
VER  FROM      TO        AT                         TRIGGER                             ACTOR   REASON
---  --------  --------  -------------------------  ----------------------------------  ------  ------
1    pending   queued    2026-10-02T14:01:23+00:00  orchestrator:dependencies_resolved  system  -
2    queued    starting  2026-10-02T14:01:23+00:00  scheduler:dispatch                  system  -
3    starting  running   2026-10-02T14:01:23+00:00  executor:launched                   system  -
4    running   pausing   2026-10-02T14:01:40+00:00  user:pause                          user    -
5    pausing   paused    2026-10-02T14:01:40+00:00  runtime:pause_reached               system  -
6    paused    running   2026-10-02T14:01:42+00:00  user:resume                         user    -
7    running   stopping  2026-10-02T14:01:44+00:00  user:stop                           user    发布窗口
8    stopping  stopped   2026-10-02T14:01:44+00:00  runtime:stop_confirmed              system  发布窗口

第 4–6 行是一次完整的暂停往返，第 7–8 行是停止——谁按的、为什么按，都在轨迹里。

退出码分三档

码	含义
`0`	成功
`1`	命令本身失败（参数错、任务不存在、存储打不开……）
`3`	命令成功，但它查询/执行的任务没有成功

3 是给脚本判分支用的——把"运维脚本写错了"和"任务本身失败了"分开，脚本才可能做对判断。

任务定义来自模块，CLI 只传名字

chongming-task tasks --tasks myapp.tasks

模块必须定义 register(controller)。一个能启动任意可调用对象的命令行入口等于任意代码执行，所以这条约定必须从第一天立好。

子命令、错误路径与完整走查见 命令行控制台。

DAG 管道

把多个任务按依赖关系串起来，由一个调度器推进：节点先停在 blocked 等上游，上游满足后才放行；注定跑不了的下游会被 skipped。

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

pipeline = Pipeline("nightly-etl")

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)  # succeeded / failed / cancelled

真实输出（examples/07_pipeline.py）：

--- 刚提交时：下游都停在 blocked ---
NODE    TASK       STATUS   DEPENDS_ON            JOIN
------  ---------  -------  --------------------  -------------
fetch   fetch      running  -                     all_succeeded
shard0  transform  blocked  fetch                 all_succeeded
shard1  transform  blocked  fetch                 all_succeeded
shard2  transform  blocked  fetch                 all_succeeded
load    load       blocked  shard0,shard1,shard2  all_succeeded
audit   audit      blocked  load                  all_settled

依赖由状态机表达，不是调度器自造的

blocked 是一个真正的任务状态（早已在转移表里），不是管道额外发明的标记。所以：

PENDING → BLOCKED（等待上游）→ QUEUED（上游满足）——每一步都走状态机、都写穿到存储、都留下审计轨迹；
等待中的节点不占并发额度（它不在待执行队列里），一条长管道不会把额度锁死。

两个不同的轴，别混

概念	回答的问题
`JoinMode`（逐节点）	这一个节点何时该跑：`ALL_SUCCEEDED` / `ALL_SETTLED` / `ANY_SUCCEEDED`
`AggregationPolicy`（整条管道）	管道何时算结束、要不要连坐：`FAIL_FAST` / `ALL_SETTLED`

三种结局必须分得清

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

把它们混成一个状态，事后就再也分不出"上游挂了"、"被连坐了"和"人停的"。

CANCELLED 刻意不作为 fail_fast 的触发条件——它是连坐的结果，不是原因。否则用户主动取消整条管道会被记成"失败"，被连坐的节点还会再触发一轮连坐。

诚实的边界

管道结构在内存里，不落库。 任务的函数与参数从未持久化，所以进程重启后一条跑到一半的管道无法续跑——它的节点会被 reconcile() 如实标记，而不是被假装恢复。
不支持 quorum 结算。 quorum 描述的是"同一件事的 N 份副本成功 k 份"，它属于扇出分组而不是一条节点各异的 DAG。硬套上去只能得到一个没人能预测含义的数字，所以 submit_pipeline 会明确拒绝并说明原因。

写一个任务：`@task` 与 `TaskContext`

from chongming_task import RetryPolicy, task


@task(tags={"io"}, retry=RetryPolicy(max_attempts=3))
def extract(ctx, day: str) -> list[str]:
    """按天抽取分区列表。"""
    partitions = [f"{day}/part{i}" for i in range(6)]
    for index, part in enumerate(partitions, start=1):
        ctx.checkpoint()  # 响应停止/暂停 + 刷新心跳
        ctx.progress(index, total=len(partitions), message=part)
        ctx.log(f"处理 {part}")
    return partitions

一次登记整个模块：

controller.register_module(myapp.tasks)

ctx 是任务与运行时之间唯一的协作入口：

成员	作用
`ctx.checkpoint()`	协作点：响应停止与暂停，并刷新心跳。不调用它就不响应停止
`ctx.progress(done, total=, unit=, message=)`	上报进度（高频旁路，不碰版本号）
`ctx.log(msg)` / `.warning` / `.error`	走标准库 `logging`，logger 名 `chongming_task.task.<task_id>`
`ctx.input(节点id, 默认值)` / `ctx.upstream`	取上游节点的返回值（管道数据传递）
`ctx.artifact_dir`	本次尝试的产物目录（按 attempt 隔离）

返回值必须可 JSON 序列化（它会写入存储）。违反时任务以 result_invalid 失败—— 不会静默地变成"存储里停在 running、重启后被判 LOST"。大数据请传指针。

进度会进快照、进存储，并出现在 ps 的 PROGRESS 列：

TASK_ID                 NAME                 STATUS     ATTEMPT  ELAPSED  PROGRESS  FAILURE
----------------------  -------------------  ---------  -------  -------  --------  -------
2b4e5e326776:extract    daily-etl/extract    succeeded  1/2      0.3s     6/6       -
2b4e5e326776:transform  daily-etl/transform  succeeded  1/1      0.3s     6/6       -
2b4e5e326776:load       daily-etl/load       succeeded  1/1      0.2s     4/4       -

三个刻意的设计点

@task 不产生全局注册表。 它只把规格挂在函数对象上（__chongming_task__），谁注册、注册到哪个控制器由你决定。全局注册表意味着 import 即注册，而 import 可能发生在任何地方（包括测试收集阶段），那种隐式行为极难排查。
progress() 不刷新心跳。 心跳是"我还在推进"的整体信号，进度是"推进到哪了"。把心跳藏进进度里，会让一个只报进度、不做实际工作的任务看起来永远健康——那正是 idle 超时要抓的东西。
失败的上游不传 None。 只有成功的上游才出现在 ctx.upstream 里。塞一个 None 会让下游以为"上游返回了 None"，而不是"上游没有值"。

重启恢复

进程意外退出后，存储里会留下一批永远停在 running 的任务。新进程启动时先跑一次 reconcile()：

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

chongming-task --db crash.db reconcile --dry-run     # 先看一眼，不写任何状态
chongming-task --db crash.db reconcile --reason "worker-old 崩溃"

$ chongming-task --db crash.db reconcile --dry-run
OUTCOME      TASK_ID      WAS      DETAIL
-----------  -----------  -------  -----------------
adoptable    job-queued   queued   没有执行载体，等待重新提交
marked-lost  job-crashed  running  监督者已失联（进程重启后未再续租）

判据是租约，不是心跳

每个监督进程都会给自己负责的任务续租。租约过期才说明"没人在管了"：

情况	处置
租约未过期	有活着的进程在管，一动不动
租约过期 + 任务占着执行资源	载体已消失，判 `LOST`
租约过期 + 任务不占资源（排队/暂停/退避）	只是没人接手，不改状态
原本的监督者又活过来抢先写入	检测到版本冲突，放手

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

它不会重新执行任务

这是能力边界，不是遗漏：存储里记的是状态，不是配方——任务的函数与参数从未落库。所以 reconcile() 做的是如实标记，让看板不再显示一个假的"运行中"；真正重跑要靠调用方自己重新 submit。

重试与超时

from chongming_task.policies import ExponentialBackoff, RetryPolicy, Timeout

outcome = run_task(
    flaky_task,
    retry=RetryPolicy(max_attempts=5, backoff=ExponentialBackoff(base=0.05, factor=2)),
    timeout=Timeout(per_attempt=30, idle=10),
)

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

作用域	含义
`start`	从提交到真正开始执行的等待上限（一旦开始即失效）
`per_attempt`	单次尝试的总时长
`idle`	距最近一次 `checkpoint()` 的静默时长——唯一能发现"卡住但没报错"的机制

真实输出（examples/03_retry_and_timeout.py）：

== 重试：失败两次后成功 ==
  终态      ：succeeded
  返回值    ：第 3 次尝试成功
  尝试次数  ：3/5

== 重试：预算耗尽 ==
  终态      ：failed
  尝试次数  ：2/2
  失败归类  ：exception

== 超时：单次尝试不超过 0.5 秒 ==
  终态      ：failed
  失败归类  ：timeout_attempt
  可重试    ：True

状态模型速览

生命周期状态有 15 个，分为非终态与终态：

非终态  pending  blocked  queued  starting  running  pausing  paused  retry_wait  stopping
终态    succeeded  failed  stopped  cancelled  skipped  lost

关键区分：

stopped（人工主动停止）与 cancelled（编排层级联取消）责任方不同，分开记录。
lost 表示结果未知（线程被放弃、载体失联），与"已停止"是两回事。
超时不是状态，是失败原因（failure.kind = timeout_attempt 等）。否则会逼出 "已超时但正在重试"这类单枚举表达不了的情况。
状态只由状态机写入，每次变更都递增 status_version 并留下一条带触发源的 Transition 记录。

完整转移表与理由见设计文档 §2。

已知限制

不调用 ctx.checkpoint() 就无法优雅停止或暂停，只能等超时：线程执行器下那个线程停不掉，进程执行器下则要等强杀（任务会被真的停掉，但已经产生的那部分副作用无法撤销）。
线程无法被强制终止。优雅窗口用尽后任务落 lost，且可能仍在后台产生副作用。需要"停得住"的保证时换 ProcessExecutor()，见用户指南 §7.4。
暂停不释放执行槽位（就地挂起）。
idle 超时衡量的是 checkpoint() 间隔，不是"CPU 是否在干活"。长时间不 checkpoint 的重计算会被判定为疑似卡死——这要求任务作者让 checkpoint 频率与 idle 阈值相匹配。
跨机之后仍没有自动告警：两个节点配了同一个桶（会重复执行）或某个桶没人管（任务静默堆积）都不会报错——缓解手段见分布式设计说明 §11。
重试总是从头开始，任务不跨节点迁移：没有用户态检查点 / 断点续传。请把任务写成幂等或可断点续传。

文档

文档	内容
最佳实践	照什么做生产上不会踩坑：选型、任务写法、并发与资源、排障、反模式清单、上线检查清单
用户指南	概念、任务编写、控制语义、超时、执行器、FAQ
命令行控制台	子命令、退出码、任务模块约定、完整走查（真实输出）
持久层实现指南	换掉默认持久层要实现哪些接口、逐方法语义、并发要求、陷阱清单
分布式设计说明	多节点部署：边界表、自治节点 + 统一平面架构、风险、Postgres 落地状态
设计文档	状态模型、转移表、分层架构、里程碑、风险
示例	十二个可直接运行的脚本 + 两个 CLI 任务模块（多节点集成模板见 11 号）

uv run python examples/11_best_practice.py   # ← 想接入自己的应用，从这个开始
uv run python examples/01_quickstart.py
uv run python examples/02_control.py
uv run python examples/03_retry_and_timeout.py
uv run python examples/04_controller.py
uv run python examples/05_buckets.py
uv run python examples/06_cross_process.py
uv run python examples/07_pipeline.py
uv run python examples/08_task_context.py
uv run python examples/09_long_running.py
uv run python examples/10_worker_walkthrough.py
uv run python examples/12_process_executor.py   # 硬性停止 / 绕开 GIL / 崩溃隔离

其中 11 号是"最佳实践"主模板：一个可以直接照抄的多节点任务服务 —— 部署参数与桶划分（buckets_for_node 连续分区）→ 任务定义 → build_node() 接线 → 启动 / 提交 / 观测 / 跨节点控制 / 收尾，全流程自带断言校验。 12 号演示进程执行器：不响应停止的任务被真的杀掉、崩溃隔离、以及不可 pickle 的目标在提交那一刻被拦下。其余示例是单点机制演示，按需查阅即可。

examples/tasks_demo.py 不是可运行脚本，而是给 CLI 用的任务模块（演示 register(controller) 约定）：

PYTHONPATH=examples uv run chongming-task tasks --tasks tasks_demo

开发

uv run pytest -q        # 测试
uv run ruff check .     # 静态检查
uv run ruff format .    # 格式化
uv run mypy             # 类型检查（strict）

质量门禁：测试全绿、ruff 无告警、mypy strict 在全部源文件上零问题。

Requires Python: >=3.11

Details

PyPI

2026-10-11 21:20:17 +08:00

608 KiB

Assets (1)

chongming_task-0.1.1.tar.gz 608 KiB

Versions (2) View all

0.1.1

2026-10-11

0.1.0

2026-10-10

chongming-task (0.1.1)

Installation

About this package

chongming-task

当前状态（务必先读）

安装

快速上手

任务怎么写

控制任务

管理一批任务

任务分桶

两条硬约束

多节点部署：自治节点 + 统一平面

嵌入到你自己的应用

持久层抽象

自证契约

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

三个刻意的设计点

两个使用注意

命令行控制台

退出码分三档

任务定义来自模块，CLI 只传名字

DAG 管道

依赖由状态机表达，不是调度器自造的

两个不同的轴，别混

三种结局必须分得清

诚实的边界

写一个任务：`@task` 与 `TaskContext`

三个刻意的设计点

重启恢复

判据是租约，不是心跳

它不会重新执行任务

重试与超时

状态模型速览

已知限制

文档

开发

Requirements

chongming-task (0.1.1)

Installation

About this package

chongming-task

当前状态（务必先读）

安装

快速上手

任务怎么写

控制任务

管理一批任务

任务分桶

两条硬约束

多节点部署：自治节点 + 统一平面

嵌入到你自己的应用

持久层抽象

自证契约

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

三个刻意的设计点

两个使用注意

命令行控制台

退出码分三档

任务定义来自模块，CLI 只传名字

DAG 管道

依赖由状态机表达，不是调度器自造的

两个不同的轴，别混

三种结局必须分得清

诚实的边界

写一个任务：@task 与 TaskContext

三个刻意的设计点

重启恢复

判据是租约，不是心跳

它不会重新执行任务

重试与超时

状态模型速览

已知限制

文档

开发

Requirements

写一个任务：`@task` 与 `TaskContext`