mcpField
D
distributed-job-safety

distributed-job-safety

github.com/terrylicaクラウド・DevOps1119 回インストール今日更新
リポジトリ

distributed-job-safety について

分散データパイプラインの障害事例をもとに、pueue の並行ワークフロー向け安全パターンをまとめたスキルです。ファイル分離、アトミック書き込み、リソースに応じた実行制御、監視、リモートデプロイを扱います。

主な機能

  1. 019 項目の並行処理安全不変条件
  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 ツールチェーン）

自己進化するスキル：このスキルは利用を通じて改善されます。手順に誤りがある、パラメーターにずれが生じた、または回避策が必要になった場合は、先送りせず、すぐにこのファイルを修正してください。実際に再現できる問題がある場合にのみ更新してください。

9 つの不変条件

並行ジョブの安全性を守るための、譲れないルールです。1 つでも違反すると、気付かないうちにデータが破損するか、ジョブが失敗します。

完全な形式仕様：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

テスト：2 つの 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. 共有状態のアトミックな書き込み

チェックポイントファイルが不完全な状態で書き込まれることがあってはなりません。一時ファイル、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 のワークロードの場合：ボトルネックは pueue の並列度ではなく、ClickHouse の concurrent_threads_soft_limit（デフォルト：2 x nproc）であることがよくあります。各クエリは 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 は一時的なもので、ジョブの削除、再キューイング、分割によって変わります。監視にはグループ名とラベルパターンを使用してください。

# 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+requeue 後のチェックポイント衝突INV-2
AP-2チェックポイントのファイル名にパラメーターがないチェックポイントの削除時に FileNotFoundError が発生INV-1
AP-3pueue restart のログを信頼する再起動後に古いエラーが表示される--
AP-4PyPI への反映が即時だと思い込む公開後に「バージョンが見つからない」と表示される--
AP-5編集可能なソースとインストール済みの wheel の違いを見落とすpip のアップグレード後も uv run が古いコードを使う--
AP-6フェーズは逐次実行されると思い込む同時にキューに追加したことでフェーズ間の競合が発生--
AP-7手動の後処理手順に頼る「完了後に optimize を実行する」が実行されない--
AP-8監視でジョブ ID をハードコードするジョブの再キューイング後に監視がクラッシュするINV-8
AP-9エポック数を活用すれば並列化できるのに逐次実行する25 コア以上あるのに1,700時間をシングルスレッドで実行INV-6
AP-10状態ファイルの肥大化ジョブの送信が気付かないうちに60倍遅くなる--
AP-11リモートジョブの作業ディレクトリが誤っている[Errno 2] No such file or directory が発生--
AP-12一括送信でファイルごとに SSH 接続する300K 件のジョブに数日かかる（SSH のオーバーヘッド）--
AP-13set -euo pipefail のもとで SIGPIPE が発生問題のないパイプ処理で終了コード141が返る--
AP-14可変長 NDJSON によるデータ損失の誤認wc -l で行数が3〜6%少なく表示される--
AP-15完了時にカーソルファイルを削除する差分再開ではなく全件を再実行する--
AP-16デーモンのシークレットにシェルで有効化した環境を使うデーモンジョブ内の環境変数が空になるINV-5
AP-17パイプラインフェーズをまたぐスコープ未指定の GlobフェーズAがフェーズBのアーティファクトを消費するINV-9

moon + Pueue + systemd-run スタック

アーキテクチャ図全体と責任範囲：references/stack-architecture.md

レイヤー責任
moon/protoタスクの検出（moon）、ツールのバージョン（proto .prototools）
.envpython-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」

2 層の呼び出しパターン：このスキルが呼び出されたら、プロジェクトの .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 によるアクティビティの再試行
INV-6（安全に並列性を最大化）ワーカーごとの max_concurrent_activities
INV-8（安定した識別子）ワークフロー ID はユーザーが定義し、永続的に保持される

Temporal を検討するタイミング：パイプラインに明確に定義されたアクティビティがある（場当たり的なシェルコマンドではない）、重複排除／冪等性の保証が必要、または pueue のガードレール（オートスケーラーエージェント、手動での再試行分類）の運用コストが Temporal サーバーの実行コストを上回る場合です。

インストール：pip install temporalio（Python SDK）、brew install temporal（CLI ＋ 開発サーバー）。

2026-03-04 のインシデントから得た教訓：5 つの自律型 Claude Code エージェントが 60 件の pueue ジョブを監視した結果、約 12,800 件の暴走タスクが発生しました。原因は、pueue の restart がタスクをその場で再起動せず新規作成すること、エージェントに変更予算がなかったこと、永続的な失敗を無条件に再試行していたことです。Temporal は、これら 3 つの障害モードをすべて標準機能で防ぎます。

参考資料

実行後の振り返り

このスキルの実行が完了したら、終了前に次の点を確認してください。

  1. コマンドは成功しましたか？ — 失敗した場合は、原因となった手順またはエラーテーブルを修正してください。
  2. パラメーターや出力に変更はありましたか？ — 基盤となるツールのインターフェースに変更があった場合は、それに合わせて使用例とパラメーターテーブルを更新してください。
  3. 回避策が必要でしたか？ — 別のフラグや追加の手順など、臨機応変な対応が必要だった場合は、次回同じ回避策が不要になるよう、この SKILL.md を更新してください。

問題が実在し、再現可能な場合にのみ更新してください。推測に基づく問題は対象外です。

コメント

Markdown 対応