Files
chakcy f4de43c743 初始提交:chongming-task 任务控制库
单任务受控执行(15 态状态机 + 审计轨迹)、DAG 管道、Controller 门面。
执行器三个:线程(默认,含 async def 任务)、内联(同步 / 测试)、进程
(spawn 子进程:硬性停止、绕开 GIL、崩溃隔离)。
调度:并发额度 + 每桶限流 + 具名资源信号量(按标签)+ 优先级排队。
持久层:TaskStore 契约 + 内存 / SQLite / Postgres 实现,契约测试随包发布。
跨进程与多节点:写穿 / 读穿、指令通道与租约、owned_buckets 划界、reconcile。
CLI、12 个可直接运行的示例、docs/ 下 5 份文档。

门禁:ruff / ruff format / mypy(strict) 干净;
      pytest 563 passed(44 skipped 是需要真实 Postgres 的那批);
      带一台可连的 Postgres 时 607 passed。
2026-10-08 23:47:50 +08:00

507 lines
20 KiB
Markdown
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 命令行控制台
`chongming-task` 是**另一个进程**:它看不到任务进程的内存,只能通过共享存储查询与
控制。这正是持久层存在的理由——没有它,`ps` 什么也看不到。
```bash
# 一个终端里跑任务(前台执行到结束)
chongming-task --db /var/lib/tasks.db run --tasks myapp.tasks sync --arg tenant=A
# 另一个终端里看和管
chongming-task --db /var/lib/tasks.db ps
chongming-task --db /var/lib/tasks.db status demo-1
chongming-task --db /var/lib/tasks.db stop demo-1 --reason "发布窗口"
```
---
## 1. 命令总览
| 命令 | 作用 |
|---|---|
| `tasks --tasks MODULE` | 列出某个任务模块里注册的定义(名字、标签、描述) |
| `run TASK --tasks MODULE [--executor thread\|process]` | 提交并**前台执行**一个任务,直到落终态 |
| `ps [--all] [--status S] [--bucket N] [--pipeline RUN_ID] [--mine] [--limit N]` | 列出任务。默认只显示**未落终态**的 |
| `status TASK_ID` | 单个任务的详情 + 迁移历史 |
| `history TASK_ID` | 只列迁移历史(可审计轨迹) |
| `stats` | 各状态计数 |
| `buckets [--num-buckets N]` | 每个桶的任务数 |
| `reconcile [--dry-run] [--stale-after S] [--reason R]` | **重启恢复**:清理上次进程留下的僵尸任务 |
| `stop TASK_ID [--reason R]` | 请求停止 |
| `pause TASK_ID [--timeout S]` | 请求暂停 |
| `resume TASK_ID` | 恢复被暂停的任务 |
全局选项:
| 选项 | 说明 |
|---|---|
| `--db PATH_OR_DSN` | 任务存储:SQLite 文件路径,或 `postgresql://` 连接串(跨机部署;需要 `pip install "chongming-task[postgres]"`)。未给出时读环境变量 `CHONGMING_TASK_DB` |
| `--num-buckets N` | 集群的桶数。**必须与部署一致**——它决定 `--owned-buckets` 的合法性与 `buckets` 视图的行数;不给出按默认值 64 |
| `--owned-buckets SPEC` | 本节点负责的桶,如 `0-15,20,30`。给出后 `reconcile` 只清理这些桶、`ps --mine` 只看这些桶 |
| `--json` | 以 JSON 输出,便于脚本消费 |
| `--owner NAME` | 本客户端的标识,会写进它下发的指令里,便于事后追溯是谁按的停止 |
### 退出码
| 码 | 含义 |
|---|---|
| `0` | 命令成功 |
| `1` | 命令本身失败(参数错、任务不存在、存储打不开、目录不存在……) |
| `3` | 命令成功,但它查询/执行的**任务没有成功** |
`3` 是给脚本判分支用的:
```bash
if ! chongming-task --db tasks.db run --tasks myapp.tasks load --arg day=2026-10-02; then
echo "任务没成功,别继续往下走了" >&2
exit 1
fi
```
> 注意 `1` 与 `3` 的区别:`1` 是"我连命令都没跑成",`3` 是"命令跑成了,但任务本身
> 没成功"。混在一起会让脚本无法区分"运维脚本写错了"和"数据质量有问题"。
---
## 2. 任务定义从哪里来
CLI **只传任务名,绝不接受代码**——一个能启动任意可调用对象的命令行入口等于任意
代码执行。所以必须先有一个模块把定义登记进来:
```python
# myapp/tasks.py
def sync_tenant(ctx, tenant: str, batches: int = 120) -> str:
"""同步一个租户的数据。"""
for index in range(batches):
ctx.checkpoint()
...
return f"{tenant} 同步完成"
def register(controller) -> None:
"""约定:CLI 导入模块后会调用它。"""
controller.register("sync", sync_tenant, tags={"io"})
```
然后 `--tasks myapp.tasks`。模块必须在 `PYTHONPATH` 上(或是一个已安装的包)。
仓库里的 `examples/tasks_demo.py` 就是一个可直接用的例子:
```bash
PYTHONPATH=examples chongming-task tasks --tasks tasks_demo
```
```
NAME TAGS MAX_ATTEMPTS DESCRIPTION
------------------------------------------------------------
add cpu 1 两数相加。
flaky - 1 前几次故意失败,用来演示重试。
sync io 1 同步一个租户的数据。
```
模块没有 `register` 时会给一句明确的话,而不是 `AttributeError`:
```
错误:任务模块 'json' 必须定义 register(controller) 函数。
```
---
## 3. 完整走查
以下全部是**真实运行输出**(`examples/tasks_demo.py`,先在一个终端启动,再从另一个
终端查询与控制)。
### 3.1 启动一个长任务(前台)
```bash
chongming-task --db .cli-demo/tasks.db run --tasks tasks_demo sync \
--arg tenant=租户A --arg batches=20000 --bucket-key tenant-A \
--task-id demo-1 --label A-同步
```
`run` 是**前台**的:它自己就是 worker,执行到任务落终态才退出。要后台跑就交给
`nohup`、`systemd` 或你已有的进程管理器——CLI 不假装自己是守护进程管理器。
### 3.2 从另一个进程查询
```bash
chongming-task --db .cli-demo/tasks.db ps
```
```
TASK_ID NAME BUCKET STATUS ATTEMPT ELAPSED PROGRESS FAILURE
------- ------ ------ ------- ------- ------- -------- -------
demo-1 A-同步 7 running 1/1 7.4s - -
```
```bash
chongming-task --db .cli-demo/tasks.db status demo-1
```
```
task_id : demo-1
name : A-同步
status : running
bucket : 7
tags : io
bucket_key : tenant-A
attempt : 1/1
status_version : 3
executor : thread
created_at : 2026-10-02T14:01:23+00:00
started_at : 2026-10-02T14:01:23+00:00
ended_at : -
last_heartbeat_at : 2026-10-02T14:01:31+00:00
elapsed : 7.8s
progress : -
failure : -
retryable : -
overridden : False
迁移历史:
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 -
```
**`last_heartbeat_at` 只有在这里才有意义。** 它由任务自己的 `ctx.checkpoint()` 驱动,
所以一个"卡住但没报错"的任务会显示心跳停滞——这是唯一能发现死锁的信号。
`ps` 的 `PROGRESS` 列来自任务里的 `ctx.progress(...)`。它走**高频旁路**(不参与乐观
并发),由持有任务的进程定期回写,因此可能有最多一个轮询周期的延迟:
```
TASK_ID NAME STATUS PROGRESS
---------------------- ------------------- --------- --------
2b4e5e326776:extract daily-etl/extract succeeded 6/6
2b4e5e326776:load daily-etl/load succeeded 4/4
```
`stats` 与 `buckets`:
```
$ chongming-task --db .cli-demo/tasks.db stats
running 1
--------------------
total 1
unfinished 1
$ chongming-task --db .cli-demo/tasks.db buckets --num-buckets 10
BUCKET TASKS
--------------
0 0
...
7 1
...
```
### 3.3 暂停与恢复
```bash
chongming-task --db .cli-demo/tasks.db --owner ops@console pause demo-1
```
```
已请求暂停 demo-1(用 status 观察是否真的挂起)
暂停**未确认**。若任务由别的进程持有,这是正常的——跨进程没有回执通道,请用 status 观察结果。
```
第一行在 stdout,第二行在 stderr,**退出码仍是 0**。跨进程没有回执通道,所以
`pause` 只能如实说"未确认"——这不是失败,而是它做不到的事就别假装做到。
紧接着 `ps` 能看到它确实停了:
```
TASK_ID NAME BUCKET STATUS ATTEMPT ELAPSED PROGRESS FAILURE
------- ------ ------ ------ ------- ------- -------- -------
demo-1 A-同步 7 paused 1/1 18.4s - -
```
`resume` 之后回到 `running`。
### 3.4 停止,并留下原因
```bash
chongming-task --db .cli-demo/tasks.db --owner ops@console stop demo-1 --reason "发布窗口"
chongming-task --db .cli-demo/tasks.db ps --all
```
```
TASK_ID NAME BUCKET STATUS ATTEMPT ELAPSED PROGRESS FAILURE
------- ------ ------ ------- ------- ------- -------- ---------
demo-1 A-同步 7 stopped 1/1 20.9s - cancelled
```
原因会完整落进审计轨迹,并且**区分了是谁按的**:
```
$ chongming-task --db .cli-demo/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 行是停止。**`REASON` 只在人为动作上出现**——机器自己的
迁移没有原因要记。
### 3.5 `--json`:给脚本用
```bash
chongming-task --db .cli-demo/tasks.db --json ps --all
```
```json
[{"attempt": 1, "bucket": 7, "bucket_key": "tenant-A", "created_at": "2026-10-02T14:01:23.570947+00:00",
"ended_at": "2026-10-02T14:01:44.485082+00:00", "executor": "thread", "failure": "{...}",
"last_heartbeat_at": "2026-10-02T14:01:44.430100+00:00", "max_attempts": 1, "name": "A-同步",
"outcome": null, "overridden": 0, "status": "stopped", "status_version": 8,
"tags": "[\"io\"]", "task_id": "demo-1"}]
```
JSON 与**存储行同构**(全部是原语),因此脚本消费它不需要了解 `datetime` / 枚举的
序列化规则。三点约定:
* 时间是 ISO 字符串(UTC);
* 枚举是 `.value` 字符串;
* 嵌套结构(`failure` / `outcome` / `tags`)是 JSON **字符串**而不是对象——保持与
存储行一致,代价是脚本里要多一次 `json.loads`。
错误走 stderr,stdout 保持干净,所以 `--json` 的输出可以直接管道给 `jq`。
### 3.6 按管道查看
管道里的每个任务都带 `pipeline_run_id`(形如 `ead248368f4a:load` 的 task_id),因此
可以整条一起看:
```bash
chongming-task --db tasks.db ps --all --pipeline ead248368f4a
```
过滤**下推到存储层**(SQLite 侧有 `idx_tasks_pipeline_run` 索引),不是取回全部再筛。
`--pipeline` 也可以与 `--status` 组合:
```bash
# 这条管道里有哪些节点没成功
chongming-task --db tasks.db ps --status failed --pipeline ead248368f4a
```
---
## 4. 多节点:`--owned-buckets` 与 `--num-buckets`
多个节点各跑各的任务时,靠桶区分"哪一份是我的"。两个选项都必须与部署一致:
```bash
# 只看"我这一份"
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
```
```
BUCKET TASKS
----------------
* 0 3
* 1 0
2 0
```
`reconcile` 会**限定在自己负责的桶内**:
```bash
chongming-task --db tasks.db --owned-buckets 0-15 reconcile --dry-run
```
**`run` 同样受归属约束**:限定了 `--owned-buckets` 时,`--bucket-key` 是必需的——没有归属
桶的任务不会有任何节点认领它,所以 CLI 在提交时就报错,而不是留下一个永远排队的任务:
```
$ chongming-task --db tasks.db --num-buckets 64 --owned-buckets 0-15 run \
--tasks myapp.tasks worker --task-id orders-worker
错误:本节点只负责桶 [0, 1, ..., 15],因此提交任务 'orders-worker' 时必须给出 bucket_key
——否则它没有归属桶,不会有任何节点认领它。
```
这很重要:统一存储里有别的节点的任务,去动它既越界(那是它的责任)也没有意义(它的租约
由它自己续)。
**桶归属的校验在做任何事之前完成**,包括 `ps` / `stats` / `buckets` 这些不构造
`Controller`、直接查存储的命令。否则越界的桶号会被静默接受——筛选条件匹配不到任何东西,
看起来就像"这个桶没有任务":
```
$ chongming-task --db tasks.db --owned-buckets 0-999 ps
错误:桶号越界:64;桶数是 64(合法范围 0..63)。若部署的桶数不是 64,请用 --num-buckets 指明。
```
### 4.1 `run --executor process`:能被真的杀掉的执行方式
默认的线程执行器在"任务不响应停止"时只能放弃等待(记 `lost`,结果未知)。需要**停得住**
的保证时,让任务跑在独立子进程里:
```bash
# 前台跑一个可能不配合的任务
chongming-task --db tasks.db run --tasks tasks_demo sync --arg tenant=acme --executor process --graceful-stop 5
# 另一个终端:优雅窗口用尽后会 terminate() → kill(),任务被真的停掉
chongming-task --db tasks.db stop <task_id> --reason "跑太久"
```
代价写在这里,省得踩了才知道(与库的行为完全一致,见[用户指南 §7.4](user-guide.md#74-进程执行器硬性停止--绕开-gil--崩溃隔离)):
- **参数与返回值必须可 pickle**——`lambda`、闭包、局部函数都不行;违反时提交那一刻就报错;
- **任务函数必须在可导入的模块里**(`--tasks` 指定的那个模块正好满足),且该模块的顶层
不能有副作用;
- 停止与暂停有约 20ms 的轮询延迟(信号要过进程边界)。
`run` 输出里的 `pid` 就是那个子进程的进程号,`ps` 能直接对上号。
### 4.2 跨机部署:`--db` 换成连接串
SQLite 文件不能跨机共享。多台机器共用一套存储时,给 `--db` 一个 Postgres 连接串即可:
```bash
pip install "chongming-task[postgres]"
chongming-task --db 'postgresql://user:pw@db.internal:5432/chongming' \
--num-buckets 64 --owned-buckets 0-15 ps --all --mine
```
**除连接串本身,其余一切都一样**:`--json`、退出码、`--owned-buckets` 校验、`reconcile`
的语义都不变。三张表(`tasks` / `transitions` / `commands`)在首次连接时自动创建,不需要
手工迁移;当前角色没有建表权限时,它检测到表已存在就直接使用——**只读运维账号因此也能连**。
没装可选依赖时给的是安装命令,而不是一个 `ModuleNotFoundError`:
```
$ chongming-task --db postgresql://… ps
错误:--db 给的是 Postgres 连接串,但可选依赖未安装:PostgresTaskStore 需要可选依赖 psycopg 与 psycopg-pool。安装:
pip install "chongming-task[postgres]"
或:pip install "psycopg[binary]" psycopg-pool
```
连不上时同样只说该说的——DSN 里的**密码不会出现在任何错误信息里**:
```
$ chongming-task --db 'postgresql://someuser:s3cret@nope.invalid:5432/x' ps
错误:无法连接 Postgres(someuser@nope.invalid:5432/x):failed to resolve host 'nope.invalid': [Errno 11001] getaddrinfo failed
```
---
## 5. 重启恢复:`reconcile`
进程意外退出后,存储里会留下一批永远停在 `running` 的任务。**新进程启动时先跑一次
`reconcile`**,否则看板会永远显示一个假的"运行中"。
真实输出。先看崩溃后的现场:
```
$ chongming-task --db crash.db ps
TASK_ID NAME BUCKET STATUS ATTEMPT ELAPSED PROGRESS FAILURE
----------- ------------ ------ ------- ------- ------- -------- -------
job-queued 等待中的任务 - queued 0/1 - - -
job-crashed 夜间抽取 - running 1/1 600.4s - -
```
`job-crashed` 已经"运行"了 600 秒——它的进程其实十分钟前就没了。
**先预演**(这类操作会改状态,看一眼是值得的):
```bash
chongming-task --db crash.db reconcile --dry-run
```
```
OUTCOME TASK_ID WAS DETAIL
----------- ----------- ------- -----------------
adoptable job-queued queued 没有执行载体,等待重新提交
marked-lost job-crashed running 监督者已失联(进程重启后未再续租)
预演:有 1 个任务**会被**标记为 lost,1 个等待重新提交。未写入任何状态。
注意:本命令不会、也无法重新执行任务——存储里记的是状态,不是参数。
```
确认无误后真正执行:
```bash
chongming-task --db crash.db reconcile --reason "worker-old 崩溃"
```
```
已把 1 个僵尸任务标记为 lost,1 个任务无人接手(状态未改动,需要重新提交)。
```
之后 `ps --all` 里 `job-crashed` 是 `lost`,而 `job-queued` **仍然是 `queued`**——
它没占执行资源,只是没人接手,把它的排队事实改掉才是错的。原因也进了审计轨迹:
```
$ chongming-task --db crash.db history job-crashed
VER FROM TO AT TRIGGER ACTOR REASON
--- ------- ---- ------------------------- ---------------------- ------ -------------
4 running lost 2026-10-02T14:10:47+00:00 runtime:heartbeat_lost system worker-old 崩溃
```
### 两件必须清楚的事
1. **它不会、也无法重新执行任务。** 存储里记的是状态,不是配方:任务的函数与参数从未
落库。所以"重跑"只能靠你重新 `run`,而那需要你自己知道参数。
2. **判据是租约,不是心跳。** 一个长时间不调 `checkpoint()` 的健康任务心跳也会停滞;
若拿心跳判断进程存活,它会被误判丢失并**杀掉**。所以心跳与租约是两个独立字段。
`--stale-after` 只影响"从未续过租的任务"的宽限期(默认取租约时长的两倍),用来避开
"刚提交、工作线程还没轮到第一次续租"这个窗口。
---
## 6. 错误路径
```
$ chongming-task ps
错误:未指定任务存储。请用 --db PATH,或设置环境变量 CHONGMING_TASK_DB。
CLI 是另一个进程,没有共享存储就看不到任何任务。
$ chongming-task --db .cli-demo/tasks.db stop typo-task-id
错误:没有找到任务 typo-task-id;无法停止。
$ chongming-task --db /nope/tasks.db ps
错误:数据库所在目录不存在:/nope。请先创建它,或改用别的路径。
```
最后一条是刻意加的:SQLite 原本只会报 `unable to open database file`,完全看不出是
目录不存在——这是最容易浪费时间的错误之一。
---
## 7. 桌面上的注意事项
* **`run` 是前台进程。** 它退出时任务就没人管了。要长期跑请交给进程管理器。
* **`pause` 跨进程返回"未确认"。** 想看结果请 `status`;`stop` 则是"请求已受理"。
* **`stop` 一个不存在的任务会失败并留下零条指令。** 拼错的 id 不会制造孤儿指令。
* **`ps` 默认不含已结束的任务。** 想看历史加 `--all`,或按状态过滤 `--status failed`。
* **`--db` 必须显式给出或设环境变量。** 不会静默在当前目录建一个库——那类"我的任务
去哪了"的问题很难查。
---
## 8. 与库的关系
CLI 是 `Controller` 之上的一层**薄适配**:它做的全部事情就是解析参数、调
`Controller` / `TaskStore`、渲染输出。所以:
* 库能做的,CLI 不一定都暴露(比如 `wait_all`、`close_store`、`prune`)——直接在代码里用
[Controller](user-guide.md#8-管理一批任务controller) 更灵活;
* CLI 不引入任何库没有的概念,也不绕过任何不变量(比如它不会直接改状态)。
替换持久层的步骤见 [持久层实现指南](store-guide.md)——CLI 会自动用上你的实现,因为它
只依赖 `TaskStore` 契约。