Claude CodeでBullMQのワーカーを実装する — 失敗ジョブを作って検証

BullMQのワーカーは、書いた時点では動いて見えます。壊れるのは本番で外部APIが落ちたときで、そこで初めてリトライの設定ミスに気づきます。Claude Codeに書かせるなら、失敗するジョブをわざと流して、その結果で合否を決める形にするのが確実です。

この記事では、BullMQ 6.5.0とローカルのRedisで実際に失敗ジョブを作り、attemptsとbackoffの挙動、stalledジョブ、完了ジョブの自動削除を確かめます。出力はすべて手元で実行した結果です。Claude Codeには、その検証スクリプトを回して出力を読ませます。

リトライの挙動は3つの設定で決まる

BullMQで「失敗」になるのは2通りです。ワーカーの処理関数が例外を投げたときと、ジョブがstalledになって最大回数を使い切ったときです。例外はErrorオブジェクトで投げる必要があります。文字列をthrowすると正しく動きません。

リトライを有効にするのはattemptsです。1より大きい値を指定すると、失敗したジョブが再試行されます。backoffを指定しなければ、失敗の直後に即再試行されます。

指数backoffの待ち時間は2 ^ (attempts - 1) * delayミリ秒です。delay: 1000でattempts: 3なら、再試行は1秒後と2秒後になります。delay: 3000なら7回目の試行は、前の試行から2^6 * 3000ミリ秒、つまり約3.2分後です。attemptsを大きくするほど待ちが急に伸びるので、delayと合わせて決めます。

キュー全体に同じ設定を当てたいときは、QueueのdefaultJobOptionsに書けます。ジョブ追加時に個別指定すれば、そちらが優先されます。Claude Codeに書かせるときは、どちらに置くかも規約で決めておくと、ジョブごとにばらつきません。

Claude Codeに渡す前に決めておく3点

「BullMQでメール送信のワーカーを作って」とだけ頼むと、リトライなしの素朴な実装が返りがちです。書かせる前に、プロジェクトのCLAUDE.mdへ規約を置いておきます。

## BullMQ
- キューは `src/queues/` に置く。ジョブ追加時は必ず attempts と backoff を指定する
- backoff は exponential、delay は 1000 から始める
- 処理関数は Error オブジェクトだけを throw する
- removeOnComplete と removeOnFail は必ず指定し、失敗側は完了側より多く残す
- 検証は `node scripts/verify-retry.mjs`（Redis は 127.0.0.1:6390）で行い、出力を読んで判断する

最後の1行が肝です。「実装して」で終わらせず、検証コマンドと合格の基準をClaude Codeに渡します。この設計はPySparkの記事で扱ったローカルのpytestで合否を決める手順と同じ考え方です。

検証用のRedisは、本物の開発用Redisと分けます。次のようにポートを変えて、永続化なしで立てると、失敗ジョブが後に残りません。

redis-server --port 6390 --save "" --appendonly no --daemonize yes
redis-cli -p 6390 ping

Redis自体をClaude Codeから扱う方法は、Redisプラグインの導入記事にまとめています。ここでは接続先を固定したスクリプトだけを使うので、プラグインは要りません。

attemptsと指数backoffを実測する

次のスクリプトは、最初の2回だけ失敗し、3回目に成功するジョブを1つ流します。各試行の経過時間とattemptsMadeを出力します。

import { Queue, Worker } from 'bullmq';
const connection = { host: '127.0.0.1', port: 6390, maxRetriesPerRequest: null };
const queue = new Queue('mail', { connection });
await queue.obliterate({ force: true });
const t0 = Date.now();
const worker = new Worker('mail', async (job) => {
  console.log(`attempt=${job.attemptsMade + 1} t=${Date.now() - t0}ms`);
  if (job.attemptsMade < 2) throw new Error('smtp timeout');
  return { sent: true };
}, { connection });
worker.on('failed', (job, err) =>
  console.log(`failed: attemptsMade=${job.attemptsMade} ${err.message}`));
worker.on('completed', (job) => console.log(`completed id=${job.id}`));
await queue.add('welcome', { to: 'a@example.com' }, {
  attempts: 3,
  backoff: { type: 'exponential', delay: 1000 },
  removeOnComplete: true,
});
await new Promise((r) => setTimeout(r, 5000));
console.log('counts', await queue.getJobCounts('completed', 'failed', 'delayed', 'waiting'));
await worker.close();
await queue.close();

実行結果は次のとおりです。

attempt=1 t=9ms
failed: attemptsMade=1 smtp timeout
attempt=2 t=1086ms
failed: attemptsMade=2 smtp timeout
attempt=3 t=3120ms
completed id=1
counts { completed: 0, failed: 0, delayed: 0, waiting: 0 }

読み取れることが3つあります。

• 試行の間隔は約1秒、約2秒で、delay: 1000の指数backoffと一致します
• failedイベントは再試行のたびに発火します。最終失敗のときだけ呼ばれるわけではありません
• removeOnComplete: trueのため、完了後のカウントは0です

2つ目は見落としやすい点です。failedイベントにSlack通知を付けると、3回試行するジョブで3回通知が飛びます。最終失敗だけを通知したい場合は、job.attemptsMadeがjob.opts.attemptsに達したかを条件に入れます。

worker.on('failed', (job, err) => {
  if (job && job.attemptsMade >= (job.opts.attempts ?? 1)) {
    notify(`最終失敗: ${job.name} ${err.message}`); // notify は通知用の自前関数
  }
});

最終失敗したジョブはfailedに残る

全試行が失敗したジョブは、failedのセットに残ります。次のスクリプトは、常に失敗するジョブをattempts: 2、fixedの200ミリ秒で流します。

failed: attemptsMade=1 always fails
failed: attemptsMade=2 always fails
counts { completed: 0, failed: 1, delayed: 0, waiting: 0 }
failedReason= always fails attemptsMade= 2

removeOnFail: { count: 100 }を付けたので、最新100件まで残ります。原因調査のために、失敗ジョブは完了ジョブより長く残す設計にします。CLAUDE.mdの規約で「失敗側は完了側より多く」と書いたのは、このためです。

完了ジョブの自動削除は「次のジョブが終わるまで」動かない

removeOnCompleteとremoveOnFailには、3つの書き方があります。

• trueで、終わったジョブを即座に消す。ジョブ名に関係なく消える
• 数値で、残す件数を指定する。例はremoveOnComplete: 1000, removeOnFail: 5000
• { age, count }で、残す期間（秒）と件数の上限を指定する

どれを選んでも、削除は遅延実行です。新しいジョブが完了または失敗したときに、まとめて削除されます。最後のジョブが終わった直後にRedisを覗いて古いジョブが残っていても、不具合ではありません。

もう1点、固有のジョブIDで冪等性を保つ設計とは相性に注意が要ります。同じIDのジョブを追加すると、キューに残っている間は新しいジョブが無視されます。自動削除でジョブが消えると、同じIDで再び追加できてしまいます。

stalledジョブはワーカーをSIGKILLして再現する

stalledは、アクティブなジョブを持つワーカーがロックを更新できなくなった状態です。ワーカーの異常終了や、イベントループを塞ぐ重い処理で起きます。ジョブは待機に戻され、キューにstalledイベントが出ます。stalledという状態があるわけではなく、イベントだけが出る点に注意してください。

maxStalledCount（既定1）を超えると、ジョブは「job stalled more than allowable limit」で恒久的に失敗します。検査の間隔は既定で30秒です。

テストではロック時間と検査間隔を短縮し、子プロセスのワーカーをSIGKILLで落とします。

// 子プロセス側（node stall.mjs child）
new Worker('mail', async () => {
  console.log('child start');
  await new Promise(() => {}); // 終わらない処理
}, { connection, lockDuration: 1000, stalledInterval: 500, maxStalledCount: 1 });

親はQueueEventsでstalledとfailedを購読し、子を起動して1.5秒後にSIGKILLする操作を2回繰り返します。実行結果は次のとおりです。

child start
event stalled 1
child start
before healthy worker { active: 1, waiting: 0, failed: 0 }
event stalled 1
event failed: job stalled more than allowable limit
after healthy worker { active: 0, waiting: 0, failed: 1, completed: 0 }

1回目のSIGKILLではstalledが出てジョブが待機に戻り、2つ目の子が拾いました。2回目のSIGKILLの後は、ジョブがactiveのまま残ります。生きているワーカーがいないと、stalledの検査そのものが走らないためです。健全なワーカーを1つ起動した時点でstalledが再度検出され、maxStalledCount超過でfailedに移りました。

stalledを減らす手段として、処理を別のNode.jsプロセスで動かすサンドボックス化プロセッサがあります。処理を別ファイル（painter.ts）に切り出してnew Worker('Paint', painter)に渡す例が載っていて、CPUを使う重い処理がメインのイベントループを塞ぐのを避けられます。Claude Codeに重い変換処理を書かせたときは、この形にするかどうかを確認します。

運用では次のことが言えます。ワーカーを全台止めたまま放置すると、クラッシュしたジョブはactiveに居座ります。ジョブ数の監視はfailedだけでなくactiveの滞留も対象にします。

検証をClaude Codeに回させる

スクリプトがそろったら、Claude Codeの非対話モード（-p）で検証を依頼できます。権限は--allowedToolsで必要最小限に絞ります。

claude -p "scripts/verify-retry.mjs を実行し、attempts=3・exponential 1000ms の \
ジョブが 1 秒・2 秒間隔で再試行されているか出力から判定して。 \
合わなければ src/queues/ を直して再実行" \
  --allowedTools "Bash(node scripts/*),Read,Edit"

Bash(node scripts/*)は、scripts/配下のスクリプト実行だけを許可する書き方です。裸のBashを許可すると任意のシェルコマンドが通ります。許可されていない呼び出しは、-pの実行では承認を求められずに拒否されます。承認する人がいない実行では、--permission-prompts noneを付けると、拒否された内容を再試行しないようClaudeに伝えて処理を続けます。権限の絞り方はallowedTools/disallowedToolsの記事で詳しく扱っています。

終了コードで分岐したい場合、claude -pは成功なら0、失敗なら0以外で終了します。ただしそれは「Claude Codeの実行が成功したか」であり、リトライの挙動が正しいかではありません。合否は、検証スクリプト側がprocess.exit(1)を返す形で決めると、CIにも載せられます。

Claude Codeの実装を見るときの確認点

返ってきたコードは、次の4点で見ます。

• attemptsとbackoffが、ジョブ追加側かキューのdefaultJobOptionsのどちらかに入っているか
• 処理関数がError以外をthrowしていないか
• removeOnCompleteだけ指定して、removeOnFailが既定のまま（無制限）になっていないか
• 子プロセスやワーカーの終了処理でworker.close()を呼んでいるか

よくあるつまずき

ほかにも、backoffを指定しないとリトライが即座に連続するため、外部APIが落ちている間にattemptsを一瞬で使い切ります。外部障害を想定するジョブには、delayを数秒以上にした指数backoffを付けます。

複数のワーカーが同時に再試行する負荷を散らすなら、jitterオプションが使えます。0から1の値で、0はランダム性なし、1は0から最大値までの間で遅延が決まります。

カスタムのbackoffは、ワーカーのsettings.backoffStrategyに書きます。0を返すとジョブは待機リストの末尾に戻り、-1を返すと再試行されずにそのままfailedへ移ります。

まとめ

BullMQのリトライは、attemptsとbackoffをジョブ側に書き、removeOnCompleteとremoveOnFailで残す量を決めれば成立します。Claude Codeに任せるなら、実装の依頼と同時に、失敗ジョブを流す検証スクリプトと合格基準を渡すのが近道です。

stalledだけは、ワーカーが生きていないと検出されません。ワーカーの死活監視と、activeの滞留アラートをセットで置くかどうかが、リトライ設定の次の判断になります。