壁紙アプリの新作分を一括でタグ付けしようと、画像メタデータ 800 件を Claude にまとめて投げたことがあります。300 件ほど処理が進んだところでパイプラインが固まり、ログを追うと、たった 1 件の壊れた JSON がコンシューマを例外で落とし、その後ろの全件がリトライの渋滞に巻き込まれていました。原因の 1 件は、過去のエクスポート不具合で description フィールドに生のバイナリが混ざっていたものでした。
この「1 件の壊れた入力が、健全な残り全部を巻き添えにする」現象は、メッセージキューの世界で 毒メッセージ（poison message） と呼ばれています。同期的に for ループで API を叩いているうちは表面化しませんが、非同期キューに載せて本番運用を始めた瞬間、必ず一度はぶつかる落とし穴です。私が壁紙アプリのバックエンド処理と 4 サイトのコンテンツ生成パイプラインで実際に踏んだこの問題を、ここでは デッドレターキュー（DLQ） で隔離する設計として整理してみます。
毒メッセージがパイプライン全体を殺す仕組み
非同期パイプラインの基本は「キューにジョブを積み、コンシューマが 1 件ずつ取り出して Claude API を呼ぶ」という形です。健全に動いているうちは美しいのですが、コンシューマが処理に失敗したときの挙動が問題になります。
多くのキューは「失敗したメッセージは ack せず、一定時間後に再配信する」という at-least-once 配信を採ります。これは一時的なネットワーク断や 429（レート制限）には有効です。しかし、メッセージそのものが壊れていて 何度処理しても必ず失敗する 場合、同じメッセージが延々と再配信され続けます。
私の最初の実装では、この再配信が次の 3 つの形で残りの処理を巻き添えにしました。まず、壊れた 1 件がバッチの先頭に居座り、retryAll() を呼ぶ実装になっていたため、同じバッチに入っていた健全な 9 件も一緒に再送されました。次に、再送のたびに Claude API のトークンを消費するため、1 件の毒メッセージが静かに課金を膨らませました。最後に、リトライ回数の上限を設けていなかったため、毒メッセージがキューに永久に居座り、可視性タイムアウトの間スループットを食い続けました。
毒メッセージ対策の本質は、「直る失敗」と「直らない失敗」を切り分け、後者を健全なフローから物理的に追い出す ことです。
まず「直る失敗」と「直らない失敗」を切り分ける
すべての失敗を一律にリトライする実装は、毒メッセージを増幅させます。Claude API を呼ぶパイプラインでは、エラーを大きく 3 種類に分けて考えると整理できます。
一時的でリトライすべきもの（transient）は、429 レート制限、529 過負荷、500/503、ネットワークタイムアウトです。これらは時間を置けば直るので、指数バックオフでリトライする価値があります。
入力起因で何度やっても直らないもの（毒メッセージ）は、400 invalid_request（リクエスト本体が壊れている）、メッセージのデシリアライズ失敗、スキーマ検証で弾かれる構造化出力、max_tokens に対して入力が大きすぎてどうしても収まらないケースです。これらは即座に DLQ へ送るべきです。
判断が分かれるグレーゾーンもあります。stop_reason が max_tokens で途中で切れた応答、refusal で拒否された応答などです。これらは「リトライしても結果が変わりにくいが、入力の前処理を変えれば直る可能性がある」ため、回数を絞ったリトライののち DLQ へ送り、人間または別ロジックで再処理する設計が合っています。
この切り分けを 1 つの関数に閉じ込めておくと、コンシューマ本体がすっきりします。
// classify.ts — 失敗を「リトライ可 / 毒メッセージ」に分類する
import Anthropic from "@anthropic-ai/sdk";
export type Verdict = "retry" | "poison";
export function classifyError(err: unknown): Verdict {
// SDK が投げる APIError には status が乗る
if (err instanceof Anthropic.APIError) {
const s = err.status;
if (s === 429 || s === 529 || s === 500 || s === 503) return "retry";
if (s === 400 || s === 422) return "poison"; // 入力が壊れている
if (s === 401 || s === 403) return "retry"; // 鍵やIP制限。設定直しで回復
}
// ネットワーク系（abort / ETIMEDOUT など）は一時障害扱い
const name = (err as Error)?.name ?? "";
if (name === "AbortError" || /timeout|ECONNRESET|ETIMEDOUT/i.test(String(err))) {
return "retry";
}
// JSON.parse 失敗やスキーマ違反は毒メッセージ
if (err instanceof SyntaxError) return "poison";
// 分類不能は安全側（リトライ）に倒す。回数上限で最終的に DLQ へ落ちる
return "retry";
}
ポイントは、分類不能なものを「とりあえず毒メッセージ」にしない ことです。未知のエラーはリトライ側に倒し、後述する max_retries の上限で自然に DLQ へ落とします。こうすると、こちらが想定していなかった一時障害を、誤って毒メッセージとして捨ててしまう事故を防げます。
Cloudflare Queues で DLQ を構成する
私は 4 サイトを Cloudflare Workers 上で動かしている関係で、非同期処理にも Cloudflare Queues を使っています。Queues の良いところは、DLQ がインフラ側の標準機能として用意されていることです。アプリ側で「リトライ回数を数えるカウンタ」を自前で持つ必要がありません。
設定はコンシューマ定義の 2 つのフィールドだけです。max_retries がメッセージあたりの最大リトライ回数（デフォルト 3）、dead_letter_queue が上限到達後に送り先となるキュー名です。送り先のキューが存在しなければ自動で作成されます。
# wrangler.toml
name = "wallpaper-tagger"
main = "src/index.ts"
[[queues.producers]]
queue = "tag-jobs"
binding = "TAG_QUEUE"
[[queues.consumers]]
queue = "tag-jobs"
max_retries = 3 # 3回失敗したら DLQ へ
max_batch_size = 10
dead_letter_queue = "tag-jobs-dlq"
# DLQ 自体にもコンシューマを付け、観測・再処理できるようにする
[[queues.consumers]]
queue = "tag-jobs-dlq"
max_batch_size = 5
dead_letter_queue を 定義しないと、上限到達後のメッセージは黙って破棄されます。これは初見だと気づきにくい挙動で、私は最初「リトライ上限は付けたのに、失敗したジョブの記録がどこにも残らない」と数時間悩みました。失敗を捨てたくないなら DLQ の指定は必須です。
コンシューマ実装 — Before / After
最初に私が書いていたコンシューマは、失敗時に一括で retryAll() を呼ぶ、典型的な毒メッセージ製造機でした。
// Before — 1件の失敗でバッチ全体を巻き込む
export default {
async queue(batch: MessageBatch, env: Env) {
try {
for (const msg of batch.messages) {
await tagOneImage(msg.body, env); // どれか1件が throw すると…
}
batch.ackAll();
} catch (e) {
batch.retryAll(); // …健全な9件も道連れで再送される
}
},
};
この実装の罪は 2 つあります。1 件の失敗で健全なメッセージまで再送する点と、失敗が一時障害なのか毒メッセージなのかを区別していない点です。
修正版では、メッセージを 1 件ずつ独立して ack / retry し、毒メッセージは即座に ack して DLQ へ送ります。Cloudflare Queues では、個々の message.retry() を呼ばずに ack すれば「処理済み」とみなされ、retry() を呼べば再配信され、max_retries 到達で自動的に DLQ へ流れます。
// After — 1件ずつ独立して処理し、毒メッセージだけを隔離する
import { classifyError } from "./classify";
export default {
async queue(batch: MessageBatch, env: Env) {
for (const msg of batch.messages) {
try {
await tagOneImage(msg.body, env);
msg.ack(); // この1件だけ成功扱い
} catch (e) {
const verdict = classifyError(e);
if (verdict === "poison") {
// 毒メッセージ: 即座に ack して通常フローから外し、
// 失敗理由を添えて DLQ 相当のログに記録する
await env.DLQ_LOG.put(
`poison:${msg.id}`,
JSON.stringify({ body: msg.body, reason: String(e), at: Date.now() }),
{ expirationTtl: 60 * 60 * 24 * 30 }
);
msg.ack();
} else {
// 一時障害: リトライに回す。max_retries 到達で自動的に DLQ へ
msg.retry({ delaySeconds: backoff(msg.attempts) });
}
}
}
},
};
// 指数バックオフ（attempts は 1 始まり）+ ジッタ
function backoff(attempts: number): number {
const base = Math.min(2 ** attempts, 60);
return base + Math.floor(Math.random() * base * 0.3);
}
msg.retry() に delaySeconds を渡せるのが地味に効きます。これがないと、レート制限を食らった直後に即再送して、また 429 を食らう、という無駄なリトライループになります。バックオフにジッタ（揺らぎ）を足しているのは、同じバッチの複数メッセージが完全に同時刻で再送されて再び輻輳するのを避けるためです。
DLQ のメッセージを観測し、安全に再処理する
DLQ は「失敗を捨てる場所」ではなく「失敗をいったん預けて、原因を見てから戻す場所」です。私は DLQ コンシューマで、入ってきたメッセージを原因別に分類してから次の判断をしています。
// dlq-consumer.ts — DLQ に流れた毒メッセージを分類して棚卸しする
export async function handleDlq(batch: MessageBatch, env: Env) {
for (const msg of batch.messages) {
const job = msg.body as TagJob;
const reason = await diagnose(job); // 原因を再判定
await env.DLQ_INDEX.put(`${reason}:${job.id}`, JSON.stringify(job));
msg.ack(); // DLQ から取り出したら ack（さらに溜め込まない）
}
}
// 再処理（リドライブ）: 修正可能なものだけを本キューに戻す
export async function redrive(env: Env, reason: string) {
const list = await env.DLQ_INDEX.list({ prefix: `${reason}:` });
for (const key of list.keys) {
const job = JSON.parse((await env.DLQ_INDEX.get(key.name))!);
const repaired = repairJob(job); // 例: 不正バイト除去・description 切り詰め
if (!repaired) continue; // 直せないものは戻さない
await env.TAG_QUEUE.send({ ...repaired, idemKey: `redrive:${job.id}` });
await env.DLQ_INDEX.delete(key.name);
}
}
リドライブ（再投入）で最も大事なのは 冪等性 です。DLQ に落ちたジョブの一部は「Claude の呼び出しには成功していたが、結果の保存で失敗した」ものかもしれません。これを無条件に再処理すると、同じ画像に対して 2 回 Claude を呼んで二重に課金します。私は各ジョブに idemKey を持たせ、結果保存時に「このキーで既に保存済みなら API 呼び出しをスキップする」というガードを入れています。冪等キーがあるだけで、リドライブを怖がらずに何度でも実行できるようになります。
Claude API 特有の「毒メッセージを生むパターン」
汎用的なキュー設計に加えて、Claude API を呼ぶパイプラインならではの毒メッセージの源泉があります。実際に DLQ に溜まったメッセージを棚卸しすると、多くが次のいずれかでした。
入力が max_tokens の上限を超えて、どう刻んでも 1 リクエストに収まらないケース。私の壁紙メタデータでは、ごく一部の画像に巨大な EXIF コメントが混入していて、これが該当しました。対処は、キューに積む前に入力長を測り、上限超過は最初から別レーン（要約してから投げる）に振り分けることです。
tool_use を使った構造化出力で、Claude が返した JSON がこちらのスキーマに合わないケース。リトライしても同じ入力なら同じ構造で返ってくることが多く、毒メッセージとして扱うのが妥当です。対処として、スキーマ違反は DLQ に送りつつ、プロンプト側の指示を見直すフラグを立てています。
stop_reason が refusal で、Claude がコンテンツポリシー上の理由で生成を拒否したケース。これはリトライで解決しないので、即 DLQ 行きにして人間が確認します。私の用途（壁紙のタグ付け）では、稀にアート系の画像説明が引っかかることがありました。
これらを踏まえると、stop_reason を必ず検査し、end_turn 以外は成功扱いにしない という規律が効きます。多くの事故は「200 が返ってきたから成功」と決めつけて、max_tokens で途中切れした不完全な応答を保存してしまうことから始まります。
// レスポンスは status だけでなく stop_reason で成否を判定する
const res = await client.messages.create({ /* ... */ });
if (res.stop_reason !== "end_turn" && res.stop_reason !== "tool_use") {
// max_tokens / refusal などは「成功した200」ではない
throw new IncompleteResponseError(res.stop_reason);
}
運用してみて効いた判断
半年ほどこの構成で 4 サイトのコンテンツ生成と壁紙アプリのバックエンド処理を回してきて、いくつか自分なりの結論が出ました。
max_retries は欲張らず 3 が実用的です。レート制限と過負荷の大半は 3 回のバックオフ内で回復しますし、それ以上回しても毒メッセージを長く居座らせるだけでした。私は最初 5 に設定していましたが、3 に下げてから DLQ への滞留がおよそ 70% 減り、スループットの谷も明らかに浅くなりました。
DLQ は「監視対象」として扱うのが正解です。私は DLQ のメッセージ数が一定を超えたら通知を飛ばすようにしています。DLQ が静かに増えているときは、たいてい入力データ側に新しい壊れ方が生まれた合図でした。隔離しただけで放置すると、せっかく追い出した失敗を見落とします。
ちなみにこのタグ付けパイプラインの出力は、最終的に AdMob で収益化している壁紙アプリ群のメタデータに直結しています。1 件のタグ欠落がそのまま検索性とユーザー体験に響くため、「壊れた入力を黙って通さない」ことの重みは普段の運用で常に意識しています。
そして、毒メッセージ対策は「壊れた入力をどう扱うか」の設計そのものだと感じています。私の両祖父はどちらも宮大工で、木材は使う前に必ず一本ずつ検めてから組んでいたそうです。腐りや節のある材を黙って構造体に混ぜないのと同じで、壊れた 1 件を健全なフローに混ぜないことが、結局はパイプライン全体の寿命を延ばします。2014 年から個人開発を続け、累計 5,000 万ダウンロードのアプリ群を一人で回す中で、「壊れたものを早く、静かに、確実に脇へ寄せる」仕組みの価値を何度も実感してきました。
次に試すなら、まずは手元の同期ループに classifyError 相当の切り分けを 1 つ入れるところから始めるのをお勧めします。失敗を「直る／直らない」で見るだけで、パイプラインの安定度はかなり変わります。お読みいただきありがとうございました。