12 · M4：分布式执行层（M4a 批处理调度）
状态：M4a 核心完成（2026-11）。M4b（cell ownership / remote wire / 迁移 / 故障恢复）仍在后续。
12.1 完成了什么
新增 crates/ain-sched：
| 模块 | 内容 |
|---|---|
protocol.rs |
Request::{Ping,Run,Stats} / Response::{Pong,Result,Error,Stats}；NDJSON/TCP |
wire.rs |
连接、超时、一行 JSON 写读 |
worker.rs |
TCP worker：鉴权、执行 FlowSpec、jobs/failures/uptime |
coordinator.rs |
静态节点列表、轮询放置、失败重试、批量并发、ping/stats |
| CLI | ain node serve/ping/stats、ain flow distribute |
12.2 架构
┌─────────────────────────────┐
│ Coordinator │
│ - nodes: [n1, n2, ...] │
│ - round-robin placement │
│ - retries on failure │
│ - batch concurrency │
└──────────┬──────────────────┘
│ TCP NDJSON
┌────────────────┼────────────────┐
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ worker │ │ worker │ │ worker │
│ n1 │ │ n2 │ │ n3 │
│ runtime │ │ runtime │ │ runtime │
└─────────┘ └─────────┘ └─────────┘
- 每个
run请求携带完整FlowSpec；worker 本地demo::build_runtime(spec)执行。 - 结果包含完整
FlowResult（trace/ledger/outputs/state hash）。 - worker 无状态；失败重试从头跑（M4a）。
12.3 使用
# 终端 1 / 2
ain node serve --bind 127.0.0.1:7101 --id n1
ain node serve --bind 127.0.0.1:7102 --id n2
# 健康检查
ain node ping --nodes 127.0.0.1:7101,127.0.0.1:7102
# 分布式跑 20 次
ain flow distribute examples/research.flow \
--nodes 127.0.0.1:7101,127.0.0.1:7102 \
--repeat 20 --parallel 4 --retries 2
# 统计
ain node stats --nodes 127.0.0.1:7101,127.0.0.1:7102
实测：
ok 127.0.0.1:7101 -> n1
ok 127.0.0.1:7102 -> n2
distributed flow=Research repeat=20 nodes=2 parallel=4 elapsed_ms=28 ok=20 errors=0 unique_hashes=1
worker n1: 10
worker n2: 10
127.0.0.1:7101 jobs=10 failures=0 uptime_s=0
127.0.0.1:7102 jobs=10 failures=0 uptime_s=0
12.4 失败与安全
- 重试：连接失败/超时/remote error → 下一个节点；
--retries 2。 - 死节点演练：
--nodes 127.0.0.1:7109,127.0.0.1:7103（第一个不可用）仍 4/4 成功，全部落到 n3。 - 鉴权：worker
--token s3；coordinator 必须带同 token；错误 token →unauthorized。 - 隔离：M4a 只传
FlowSpec，worker 不接收任意代码；效果在 worker 本地执行（mock/HTTP/shell allowlist）。 - 限制：共享 secret，无 TLS/mTLS；生产需补。
12.5 测试
crates/ain-sched/tests/distributed.rs（5 个）：
- 单 worker 结果与本地执行同 state hash。
- 2 worker × 10 flow：顺序保持、两边都分到任务、hash 全等。
- 死节点 + 重试：成功且 attempts ≥ 2。
- 错误 token：
unauthorized。 - stats：jobs/failures 计数。
总计：59 个测试（51 + 8 sched/runtime chunk）。
12.6 与完整 M4 的差距
| 规格（docs/06） | M4a 实际 | M4b |
|---|---|---|
| 静态成员 + 心跳 | 静态列表 + ping/stats | 动态成员/心跳/故障检测 |
| cell ownership / 路由 | 整 flow 调度 | cell 级归属/路由 |
| remote wire / seq/ack | 无 | 需要 |
| 迁移 | 无 | 需要 |
| 检查点跨节点恢复 | 无（重试从头） | 需要 |
| TLS/mTLS/审计/租户 | 共享 token | 需要 |
| 100 agents / 杀节点 5s 恢复 | 2 节点 batch demo | 需要 |
12.7 M4b 追加（checkpointed chunks + 任务迁移）
已落地：
Runtime::run_chunk(max_steps)→ChunkResult { result, checkpoint, done }。- 协议
run_chunk/chunk：worker 可接收Checkpoint，恢复后只跑一段。 Coordinator::run_checkpointed(spec, chunk_steps)：- chunk 0 在 n1，chunk 1 在 n2，chunk 2 回 n1……（round-robin）。
- 每个 chunk 结束把 checkpoint 交给下一个 worker。
- 失败时同一 checkpoint 换节点重试。
- 聚合所有 chunk 的 log，最终 trace 完整。
Coordinator::run_batch_checkpointed+ CLI--chunk-steps N。Coordinator::healthy_nodes()+ CLI--health-check（先把失联节点摘掉）。
实测：
ain flow distribute examples/research.flow \
--nodes 127.0.0.1:7105,127.0.0.1:7106 \
--repeat 3 --parallel 2 --chunk-steps 2 --health-check
# distributed ... chunk_steps=2 chunks=12 workers_used=2 ok=3 errors=0 unique_hashes=1
3 节点 × 100 flow：
ain flow distribute examples/research.flow \
--nodes 127.0.0.1:7111,127.0.0.1:7112,127.0.0.1:7113 \
--repeat 100 --parallel 8 --health-check
# distributed ... chunks=100 workers_used=3 elapsed_ms=75 ok=100 errors=0 unique_hashes=1
# n7111: 34 / n7112: 33 / n7113: 33
测试新增：
chunked_run_matches_full_run（runtime）。chunked_execution_migrates_across_workers（sched，2 worker，chunk=2，workers 同时包含 n1/n2）。healthy_nodes_filters_dead_workers。
12.8 下一步（M4b 剩余）
- remote wire：单 flow 内跨节点 cell 通道；seq/ack/去重/背压。
- cell ownership + lease/fencing：cell 级路由与迁移。
- mTLS / capability token / 审计 / 租户配额。
- 杀节点演练：worker 在 chunk 中途消失 → 从 checkpoint 续跑（当前重试同节点列表）。
- 100 agents 单 flow（当前是 100 个 flow）。