mcpField
D
distributed-job-safety

distributed-job-safety

github.com/terrylica云与 DevOps1119 次安装今天更新
代码仓库

关于 distributed-job-safety

面向 pueue 并发任务工作流的安全实践技能，汇总分布式数据管道中的规则与反模式。它覆盖文件隔离、原子写入、资源调度、监控和远程部署。

主要功能

  1. 01九项并发安全不变量
  2. 02避免文件名冲突与状态竞争
  3. 03原子写入及幂等文件操作
  4. 04按资源调整并监控并行度
  5. 05安全审计和远程部署流程

适用场景

  1. 01运行带参数的并发数据任务
  2. 02保护检查点和生成产物免受交叉污染
  3. 03部署修复前检查运行中的任务
  4. 04按主机资源调节 pueue 并发度

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/terrylica/cc-skills --skill distributed-job-safety

安装后在 Claude Code 中输入/distributed-job-safety确认已加载。

包含的文件

除 SKILL.md 外，安装时一起下载的参考资料和脚本。

references/
anti-patterns.md
autoscaler.md
concurrency-invariants.md
deployment-checklist.md
environment-gotchas.md
evolution-log.md
stack-architecture.md

详细介绍

分布式任务安全

在分布式数据管道编排中从生产故障中总结出的并发任务管理模式与反模式，涉及 pueue + moon + systemd-run。

范围：适用于任何包含并发参数化任务的 pueue 工作流的通用原则。示例使用说明性名称，但这些原则适用于任何领域。

前置技能：devops-tools:pueue-job-orchestration、itp:bootstrap-monorepo（moon 任务 + proto 工具链）

持续演进的技能：本技能会在使用中不断完善。如果说明有误、参数发生偏移，或需要采用变通方案，请立即修复此文件，不要拖延。仅针对真实且可复现的问题进行更新。

九项不变量

并发任务安全不可妥协的规则。违反其中任何一项都会导致静默数据损坏或任务失败。

完整的形式化规范：references/concurrency-invariants.md

1. 文件名必须包含所有任务参数，确保唯一

并发任务之间共享的每个文件路径都必须包含区分这些任务的所有参数。

WRONG:  {symbol}_{start}_{end}.json                    # Two thresholds collide
RIGHT:  {symbol}_{threshold}_{start}_{end}.json         # Each job gets its own file

测试：如果两个 pueue 任务可能同时运行且参数值不同，那么这些值必须出现在每个共享文件名、临时目录和锁文件中。

2. 修改前先验证（禁止盲目入队）

将任务加入队列前，检查当前有哪些任务正在运行。删除状态前，检查其归属。

# WRONG: Blind queue
for item in "${ITEMS[@]}"; do
    pueue add --group mygroup -- run_job "$item" "$param"
done

# RIGHT: Check first
running=$(pueue status --json | jq '[.tasks[] | select(.status | keys[0] == "Running") | .label] | join(",")')
if echo "$running" | grep -q "${item}@${param}"; then
    echo "SKIP: ${item}@${param} already running"
    continue
fi
3. 文件操作必须幂等（missing_ok=True）

并发上下文中的所有文件删除操作都必须能容忍文件已不存在的情况。

# WRONG: TOCTOU race
if path.exists():
    path.unlink()        # Crashes if another job deleted between check and unlink

# RIGHT: Idempotent
path.unlink(missing_ok=True)
4. 共享状态必须原子写入

检查点文件绝不能只写入一部分。使用 tempfile-fsync-rename 模式。

fd, temp_path = tempfile.mkstemp(dir=path.parent, prefix=".ckpt_", suffix=".tmp")
with os.fdopen(fd, "w") as f:
    f.write(json.dumps(data))
    f.flush()
    os.fsync(f.fileno())
os.replace(temp_path, path)  # POSIX atomic rename

Bash 等效方式（用于 NDJSON 遥测追加）：

# Atomic multi-line append via flock + temp file
TMPOUT=$(mktemp)
# ... write lines to $TMPOUT ...
flock "${LOG_FILE}.lock" bash -c "cat '${TMPOUT}' >> '${LOG_FILE}'"
rm -f "$TMPOUT"
5. 配置文件是 SSoT

项目的 .env 文件（由 python-dotenv 在运行时加载，见 AP-16）是环境默认值的唯一事实来源。按任务设置的 env 覆盖项会绕过 SSoT，允许使用未经审查的任意值。

# WRONG: Per-job override bypasses the .env SSoT
pueue add -- env MY_APP_MIN_THRESHOLD=50 uv run python script.py

# RIGHT: Set the correct value in .env, no per-job override needed
pueue add -- uv run python script.py

受控例外：对于已暂存或已排队的任务，可接受使用 pueue env set <id> KEY VALUE 设置一次性覆盖项（例如超参数扫描）。关键区别在于：.env 是适用于所有运行的默认值的 SSoT；pueue env set 用于为特定任务设置一次性参数，而无需修改配置文件。参见 devops-tools:pueue-job-orchestration 中的“按任务设置环境覆盖项”部分。

6. 在安全范围内最大化并行度

务必探测主机资源，并调整并行度以利用可用容量。保守的默认值会浪费数小时的空闲算力。

# Probe host resources
ssh host 'nproc && free -h && uptime'

# Sizing formula (leave 20% margin for OS + DB + overhead)
# max_jobs = min(
#     (available_memory_gb * 0.8) / per_job_memory_gb,
#     (total_cores * 0.8) / per_job_cpu_cores
# )

对于 ClickHouse 工作负载：瓶颈往往是 ClickHouse 的 concurrent_threads_soft_limit（默认值：2 x nproc），而不是 pueue 的并行度。每个查询都会申请 max_threads 个线程（默认值：nproc）。应为每个查询合理设置 --max_threads，使其与有效线程数（soft_limit / pueue_slots）相匹配，然后增加 pueue 槽位数。无需重启正在运行的任务，即可实时调整 Pueue 并行度。

提升并行度后的监控（任何并行度变更后，都必须持续监控 5 分钟）：

  • uptime —— 平均负载应低于 0.9 x nproc
  • vmstat 1 5 —— si/so 列必须保持为 0（没有正在进行的交换）
  • ClickHouse 错误：SELECT count() FROM system.query_log WHERE event_time > now() - INTERVAL 5 MINUTE AND type = 'ExceptionWhileProcessing' —— 必须为 0

交叉参考：完整的决策矩阵见 devops-tools:pueue-job-orchestration 中的 ClickHouse 并行度调优部分。

7. 使用 systemd-run 为每个任务设置内存上限

在使用 cgroups v2 的 Linux 系统上，使用 systemd-run 包装每个任务，以强制实施硬性内存限制。

systemd-run --user --scope -p MemoryMax=8G -p MemorySwapMax=0 \
    uv run python scripts/process.py --symbol BTCUSDT --threshold 250

重要：必须设置 MemorySwapMax=0。否则，进程会转而使用交换空间，内存限制实际上就失去了意义。

8. 按稳定标识符监控，而非临时 ID（INV-8）

Pueue 作业 ID 是临时的——作业被移除、重新入队或拆分时，ID 会改变。监控时请使用组名称和标签模式。

# WRONG: Hardcoded job IDs
if pueue status --json | jq -e ".tasks.\"14\"" >/dev/null; then ...

# RIGHT: Query by group/label
pueue status --json | jq -r '.tasks | to_entries[] | select(.value.group == "mygroup") | .value.id'

完整规范：references/concurrency-invariants.md

9. 派生产物的文件名必须包含所有类别维度（INV-9）

当并发或顺序执行的流水线阶段生成共享同一目录的派生产物（Parquet 分块、JSONL 摘要、临时文件）时，每个文件名都必须包含所有用于区分的维度——不仅要包含作业级参数（INV-1），还要包含方向、策略或生成等流水线级类别。

WRONG:  _chunk_{formation}_{symbol}_{threshold}.parquet     # No direction -- LONG glob eats SHORT files
RIGHT:  _chunk_{direction}_{formation}_{symbol}_{threshold}.parquet  # Direction-scoped

Glob 范围规则：清理和合并用的 glob 必须与文件名模式完全匹配：

# WRONG: Unscoped glob -- consumes artifacts from other categories
chunk_files = folds_dir.glob("_chunk_*.parquet")

# RIGHT: Category-scoped glob -- only touches this category's artifacts
chunk_files = folds_dir.glob(f"_chunk_{direction}_*.parquet")

合并后验证：合并产物后，断言类别列中的值符合预期：

merged_df = pl.concat([pl.read_parquet(p) for p in chunk_files])
assert set(merged_df["strategy"].unique()) == {"standard"}, "Direction contamination!"

与 INV-1 的关系：INV-1 通过作业参数确保检查点文件的唯一性（运行时隔离）。INV-9 将此要求扩展到跨流水线阶段保留的派生产物（产物隔离）。两者都能防止同一类问题——文件名冲突导致的静默交叉污染。

完整规范：references/concurrency-invariants.md

反模式（从生产环境中总结）

从生产故障中记录了 17 个反模式。完整详情和代码示例：references/anti-patterns.md

AP名称主要症状相关不变量
AP-1未检查运行中的作业就重新部署kill 并重新入队后检查点发生冲突INV-2
AP-2检查点文件名缺少参数删除检查点时出现 FileNotFoundErrorINV-1
AP-3轻信 pueue restart 日志重启后出现旧错误--
AP-4误以为 PyPI 会即时完成传播发布后出现“找不到版本”--
AP-5可编辑源代码与已安装的 wheel 不一致pip 升级后，uv run 仍使用旧代码--
AP-6假设阶段按顺序执行同时入队导致阶段争用--
AP-7手动执行后处理步骤“等它们完成后运行 optimize”从未发生--
AP-8监控器中硬编码作业 ID作业重新入队后监控器崩溃INV-8
AP-9在 epoch 支持并行时仍按顺序执行在 25 个以上核心上单线程运行 1,700 小时INV-6
AP-10状态文件膨胀作业提交速度无声地降低 60 倍--
AP-11远程作业的工作目录错误出现 [Errno 2] No such file or directory--
AP-12批量提交时逐个文件执行 SSH300K 个作业需要数天（SSH 开销）--
AP-13在 set -euo pipefail 下发生 SIGPIPE无害的管道操作导致退出代码为 141--
AP-14将可变 NDJSON 误判为数据丢失wc -l 显示行数少了 3-6%--
AP-15完成时删除游标文件全部重新运行，而非增量恢复--
AP-16用 shell 激活的环境传递守护进程密钥守护进程作业中的环境变量为空INV-5
AP-17未限定范围的 glob 跨流水线阶段匹配阶段 A 误用阶段 B 的产物INV-9

moon + Pueue + systemd-run 技术栈

完整的架构图和职责边界：references/stack-architecture.md

层职责
moon/proto任务发现（moon）、工具版本（proto .prototools）
.env环境默认值和密钥，由 python-dotenv 加载
pueue守护进程持久化、并行度限制、重启、--after
systemd-run每个任务的 cgroup 内存上限（仅限 Linux，在 macOS 上不执行任何操作）
autoscaler根据主机资源动态调整并行度
Python/app领域逻辑、检查点管理、数据完整性

远程部署协议

在运行中的主机上部署修复时：

1. AUDIT:   ssh host 'pueue status --json' -> count running/queued/failed
2. DECIDE:  Wait for running jobs? Kill? Let them finish with old code?
3. PULL:    ssh host 'cd ~/project && git fetch origin main && git reset --hard origin/main'
4. VERIFY:  ssh host 'cd ~/project && python -c "import pkg; print(pkg.__version__)"'
5. UPGRADE: ssh host 'cd ~/project && uv pip install --python .venv/bin/python --refresh pkg==X.Y.Z'
6. RESTART: ssh host 'pueue restart <failed_id>' OR add fresh jobs
7. MONITOR: ssh host 'pueue status --group mygroup'

重要：步骤 1（AUDIT）为必需步骤。跳过它是导致级联故障的根本原因。

完整协议请参见：references/deployment-checklist.md。

并发安全决策树

Adding a new parameter to a resumable job function?
|-- Is it job-differentiating (two jobs can have different values)?
|   |-- YES -> Add to checkpoint filename
|   |          Add to pueue job label
|   |          Add to remote checkpoint key
|   |-- NO  -> Skip (e.g., verbose, notify are per-run, not per-job)
|
|-- Does the function delete files?
|   |-- YES -> Use missing_ok=True
|   |          Use atomic write for creates
|   |-- NO  -> Standard operation
|
|-- Does the function write to shared storage?
    |-- YES -> Force deduplication after write
    |          Use UPSERT semantics where possible
    |-- NO  -> Standard operation

Autoscaler

根据主机 CPU 和内存动态调整 pueue 组的并行度。详情请参见：references/autoscaler.md

CPU < 40% AND MEM < 60%  ->  SCALE UP (+1 per group)
CPU > 80% OR  MEM > 80%  ->  SCALE DOWN (-1 per group)
Otherwise                 ->  HOLD

关键原则：逐步提高并行度（不要直接调到最大值）。任务内存占用会随时间增长——直接调到最大并行度可能导致所有任务同时达到峰值时发生 OOM。

项目专属扩展

此技能提供适用于任何分布式任务流水线的通用模式。项目应在 .claude/skills/ 目录中创建本地扩展技能（例如 myproject-job-safety），提供以下内容：

本地扩展提供的内容示例
具体函数名称run_resumable_job() -> myapp_populate_cache()
应用专属环境变量MY_APP_MIN_THRESHOLD、MY_APP_CH_HOSTS
按任务类型划分的内存配置“250 dbps 峰值为 5 GB，使用 MemoryMax=8G”
数据库专属审计查询SELECT ... FROM mydb.mytable ... countIf(x < 0)
问题来源追踪“检查点竞争：GH-84”
主机专属配置“：32 核、61 GB、组 p1/p2/p3/p4”

双层调用模式：触发此技能时，还应检查并调用项目 .claude/skills/ 目录中的任何本地 *-job-safety 技能，以获取项目专属配置。

devops-tools:distributed-job-safety    (universal patterns - this skill)
  + .claude/skills/myproject-job-safety  (project-specific config)
  = Complete operational knowledge

SOTA 替代方案：用于持久化工作流的 Temporal

对于结构清晰、可重复执行的作业流水线，Temporal 内置了对本技能中多项不变量的强制保障：

本技能的不变量Temporal 对应机制
INV-2（变更前验证）工作流 ID 唯一性——拒绝重复启动
INV-3（幂等操作）使用 non_retryable_error_types 重试 Activity
INV-6（安全地最大化并行度）每个 Worker 的 max_concurrent_activities
INV-8（稳定标识符）工作流 ID 由用户定义且永久不变

考虑使用 Temporal 的时机：当流水线包含定义明确的 Activity（而非临时的 shell 命令）、需要去重／幂等保证，或 pueue 防护机制（自动扩缩容 Agent、手动重试分类）的开销超过运行 Temporal 服务的开销时。

安装：pip install temporalio（Python SDK），brew install temporal（CLI + 开发服务器）。

2026-03-04 事故的教训：5 个自主运行的 Claude Code Agent 监控 60 个 pueue 作业，结果产生了约 12,800 个失控任务，因为 pueue 的 restart 会创建新任务（而非原地重启），Agent 没有变更预算，并且持续失败的任务被盲目重试。Temporal 可原生防止这三种故障模式。

参考资料

  • 反模式 -- 17 种生产环境故障模式（AP-1 至 AP-17）
  • 并发不变量 -- 不变量的正式规范（INV-1 至 INV-9）
  • 部署检查清单 -- 远程部署的逐步操作流程
  • 环境注意事项 -- 特定主机的常见问题（G-1 至 G-17）
  • 技术栈架构 -- moon + Pueue + systemd-run 层示意图
  • 自动扩缩容 -- 动态并行度调优模式
  • 交叉参考：devops-tools:pueue-job-orchestration -- Pueue 基础知识、依赖链式编排、安装
  • SOTA 替代方案：Temporal -- 提供内置去重和重试机制的持久化工作流编排

执行后反思

此技能执行完毕后，请在结束前检查：

  1. 命令是否成功？ — 如果没有，请修正导致失败的指令或错误表。
  2. 参数或输出是否有变化？ — 如果底层工具的接口发生变化，请更新 Usage 示例和 Parameters 表，使其保持一致。
  3. 是否需要变通处理？ — 如果你不得不临时调整（使用不同标志、增加额外步骤），请更新此 SKILL.md，以便下次调用时不必再采用相同的变通方法。

仅当问题确实存在且可复现时才更新——不要基于猜测更新。

评论

支持 Markdown