クラウド上で日次・時間単位のバッチジョブを運用しているエンジニアや、外部APIを大量に呼び出すパイプラインを設計しているSRE担当者に向けた内容です。再実行(レジューム)機能を最初から実装したのに、なぜか本番で「終わったはずの処理」が重複したり、逆に永遠に終わらなくなったりする現象を扱います。設計は正しく見えるのに動かない、という状況の切り分けに悩んでいるなら参考になれば幸いです。
ある事例では、コンテンツ生成パイプラインのバッチランナー(大量のジョブを順次または並列に処理する実行基盤)が、初回コミットの時点で再開フラグ・状態ログ・エラー分類・クォータ超過時の一時停止・ロックファイルまで揃っていました。設計としては教科書的です。それでも実際に正しく再開できるまでに3ヶ月かかりました。
何が起きるか:再開したはずが「二重課金」される
影響は主にコスト面とデータ整合性の両方に出ます。
バッチが中断され、再実行時に「未完了のタスクだけ」を拾い直すはずが、実際には完了済みのタスクまで再処理してしまうケースがあります。外部の生成AI APIやクラウドの従量課金サービスを呼んでいる場合、これはそのまま二重の課金につながります。
逆のパターンもあります。クォータ超過(API利用上限に達した状態)で一時停止したジョブが、状態が壊れたまま「新規扱い」で処理され、本来は待つべきだったタスクが失敗として記録されてしまう現象です。どちらも監視ダッシュボード上は「正常終了」に見えてしまうため、発覚が遅れがちです。
なぜ起きるか:4つの原因を分解する
原因は単一のバグではなく、独立した4つの穴が同じ日に見つかったという点が重要です。それぞれ根本原因が違うので、1つ直しても残り3つは残ります。
1つ目は、エラーが正常値として扱われる問題です。並列実行しているタスクの集約処理で、クォータエラーが例外として送出されず、空の結果として返ってきました。集約側はそれを「成功した空データ」として扱い、実際には未生成のコンテンツを「生成済み」として記録してしまいました。
2つ目は、エラー判定ロジックの重複と不一致です。レート制限エラー(API呼び出し頻度が上限を超えた状態)を検知する処理が、スタックの3箇所に別々の文字列マッチングで実装されていました。ある層で検知できるエラーが、別の層では見逃される状態です。これは複数のマイクロサービスやレイヤーをまたぐシステムでよく見る構造的な問題で、ルールが実装ごとに少しでもズレる余地があると、時間とともに必ずズレていきます。
3つ目は、破損した状態ファイルの誤読です。状態を記録するファイルが壊れていた場合、それを「エラー」ではなく「進捗ゼロの新規開始」として解釈してしまい、完了済みの処理をすべて最初からやり直す挙動になっていました。
4つ目は、ロック機構の競合状態(レースコンディション)です。「誰かがロックを持っているか確認する」処理と「自分がロックを取得する」処理の間に隙間があり、2つのランナーが同時にその隙間を通り抜けられる設計になっていました。分散システムでの排他制御の基本的な落とし穴です。
さらに補足すると、状態管理には2つの独立した軸がありました。1つはタスクの状態(未着手・実行中・完了・失敗・一時停止の5種類)、もう1つは失敗の分類(タイムアウト・クォータ超過・検証エラー・処理エラー・計画エラー・不明の6種類)です。この2軸を分離しておくことで、「一時停止(待てば直る)」と「失敗(待っても直らない)」を区別できます。クォータ超過は待てば復旧する一時停止として扱い、検証エラー(決定的な失敗)は人間の介入が必要な失敗として扱う、という設計判断です。実行中に強制終了されたランナーが残すタスクは「実行中」のまま固まってしまうため、再開時にそれらを「未着手」へ戻す掃除処理も必要になります。ただしこの掃除はタスクの状態にしか及ばず、ロックファイル自体は別の仕組みで管理しないと消し忘れが残ります。
自分のプロジェクトが該当するか確認する方法
以下のポイントを実際のコードと設定で確認してみてください。
- 並列処理の集約部分:Promise.allやasyncio.gather相当の処理で、例外が握りつぶされて空の結果に変換されていないか。例外ハンドラのcatch節でログだけ出して正常値を返していないか確認します
- エラー判定ロジックの分散:レート制限やクォータ超過を判定する文字列マッチングやステータスコード判定が、コードベース内に何箇所あるかgrepコマンドで洗い出します
- 状態ファイルの読み込み処理:ファイルが存在しない場合と、存在するが壊れている場合(JSONパースエラーなど)を同じ分岐で処理していないか確認します
- ロック取得処理:「確認」と「取得」が同一のアトミック操作になっているか。クラウド環境ならDynamoDBの条件付き書き込みや、Redisのsetnxのような単一命令で実装されているか確認します
バージョン確認という観点では、使っているジョブスケジューラやワークフローエンジン(Airflow、Step Functions、Cloud Composerなど)の再試行・冪等性(同じ処理を何度実行しても結果が変わらない性質)に関する仕様を、公式ドキュメントの「Idempotency」や「Retry Behavior」の章で確認するのが確実です。
対策の手順
以下の順番で見直すと、4つの穴を1つずつ潰せます。
# 1. 並列集約部分でエラーが握りつぶされていないか静的に確認
grep -rn "except.*:.*pass\|except.*:.*return {}" --include="*.py" ./pipeline
# 2. レート制限判定の重複箇所を洗い出す
grep -rn "429\|rate.limit\|quota" --include="*.py" ./pipeline | wc -l
# 3. 状態ファイルの整合性チェック(壊れていたら明示的にfailedへ)
python -c "import json; json.load(open('state.json'))" || echo "state file corrupted"- エラー判定ロジックが複数箇所にあると分かったら、共通の判定関数に一本化し、呼び出し元をすべてそこに寄せます
- 状態ファイル読み込みは「ファイルなし=新規」「パース失敗=要人間確認」を明確に分岐させ、後者は自動で新規扱いにしないようにします
- ロック取得は言語やインフラのアトミック操作(DBの条件付き更新、分散ロックサービスなど)に置き換え、確認と取得を1操作にまとめます
- 監視面では、タスクの状態(pending/running/done/failed/paused)と失敗分類をダッシュボードで別軸として可視化し、一時停止が失敗として誤集計されていないかを定点観測します
まとめ
再開可能なバッチ処理は、フラグやロックファイルを用意しただけでは完成しません。
エラーの握りつぶし、判定ロジックの重複、状態ファイルの誤読、ロックの競合状態という4つは、それぞれ別のバグとして独立に潰す必要があります。
まず着手すべきは、並列処理の集約部分で例外が正常値に化けていないかのコード確認です。次に、レート制限やクォータ判定のロジックがコードベースに何箇所散らばっているかをgrepで数え、1箇所に統合することから始めてみてください。