Claude CodeでRedis Streamsのコンシューマーグループを実装し、停止ワーカーの未ACKを回収する

Redis Streamsのコンシューマーグループは、1本のストリームを複数のワーカーで分け合う仕組みです。書くこと自体は難しくありません。落とし穴は、ワーカーがメッセージを受け取ったあと、ACKを返す前に止まった場合の扱いです。

手元のRedisで試すと、ワーカーAが3件を受け取ったまま止まり、3件は誰にも処理されないままグループの待機リストに残りました。別のワーカーが > で読み続けても、その3件は二度と届きません。

この記事では、その状態を意図的に作り、XCLAIMとXAUTOCLAIMで別のワーカーへ引き継ぐ流れを確かめます。Claude Codeには、実装より先に「止めて確かめる」テストを書かせます。

コンシューマーグループで何が起きているのか

XREADGROUPで読んだメッセージは、グループの中の待機リスト(PEL: Pending Entries List)に入ります。公式のXREADGROUPのページは、PELを「配信済みで、まだ確認応答されていないメッセージIDの一覧」と説明しています。リストから外れるのは、クライアントがXACKを呼んだときです。

読み方のIDで、返ってくるメッセージが変わります。

もう一つの前提が「1つのメッセージの持ち主は1人」という保証です。コンシューマーは自分に配信された履歴しか見られません。止まったワーカーの分を誰かが引き取るには、別途メッセージの「所有権の移転」(claim)が必要です。

止まったワーカーを作って観察する

仕組みを読むより、壊して見たほうが早く分かります。次の手順は、Redisを別ポートで起動し、ワーカーAが3件を読んだまま止まった状況を再現します。

redis-cli -p 6392 XGROUP CREATE orders g1 '$' MKSTREAM
for i in 1 2 3 4 5; do
  redis-cli -p 6392 XADD orders '*' order $i
done
# ワーカーAが3件を受け取り、ACKせずに止まる
redis-cli -p 6392 XREADGROUP GROUP g1 workerA COUNT 3 \
  STREAMS orders '>'
# ワーカーBは残りの2件を読んで処理し、ACKする
redis-cli -p 6392 XREADGROUP GROUP g1 workerB COUNT 10 \
  STREAMS orders '>'
# <ID4> <ID5> は、ワーカーBが受け取った2件のID
redis-cli -p 6392 XACK orders g1 <ID4> <ID5>

XGROUP CREATEの $ は「最後のエントリのID」を指し、グループ作成後に追加されたメッセージから読み始めます。MKSTREAM を付けると、ストリームがまだ無くても作られます。ストリームがまだ存在しないと、グループ作成はエラーになります。

ワーカーBは新しい2件を受け取り、XACKで2件を消しました。ここで待機リストを見ます。

redis-cli -p 6392 XPENDING orders g1
redis-cli -p 6392 XPENDING orders g1 - + 10

要約形のXPENDINGは、待機中が3件で、最小IDと最大IDがワーカーAのメッセージ、持ち主がワーカーAの3件という結果でした。範囲指定の形では、各メッセージの「持ち主・アイドル時間(ミリ秒)・配信回数」が出ます。この時点では3件とも配信回数は1です。

要するに、ワーカーAが戻らない限り、この3件を処理する者はいません。ワーカーBが > でいくら読んでも、新しいメッセージしか来ないためです。

XCLAIMで引き継ぐ

Redis 6.0.9の手元環境では、まずXCLAIMで所有権を移せます。第4引数の min-idle-time は、そのミリ秒を超えてアイドルなメッセージだけを移す条件です。

sleep 2
redis-cli -p 6392 XCLAIM orders g1 workerB 1000 \
  <ID1> <ID2> <ID3>
redis-cli -p 6392 XPENDING orders g1 - + 10

2秒待ってから実行すると、3件ともワーカーBへ移りました。XPENDINGの範囲指定で見ると、持ち主がworkerBになり、配信回数は2に増えています。

次の2点が、実装で効いてきます。

• XCLAIMが成功すると、アイドル時間はリセットされる。直後に別のワーカーが1000ミリ秒の条件で同じIDをXCLAIMしても、何も返らなかった
• 配信回数が増える。このカウンターが増えるのは、XCLAIMが成功したときと、XREADGROUPで自分の待機履歴を読んだときの2つ

1つ目は二重処理を防ぐ設計です。複数のワーカーが同時に回収を試みても、最初に取った1人だけがメッセージを手にします。2つ目は、後で説明する「処理できないメッセージの検出」に使います。

XAUTOCLAIMで回収ループを作る

XCLAIMは、先にXPENDINGでIDを調べてから呼ぶ2段階の操作です。XAUTOCLAIMは、その2つを1回のコマンドにまとめたものです。同じ手元環境で呼ぶと、次の結果でした。

redis-cli -p 6392 XAUTOCLAIM orders g1 workerB 1000 0-0
# ERR unknown command `XAUTOCLAIM`, with args beginning with: ...

XAUTOCLAIMはRedis 6.2.0で追加されたコマンドです。6.0.9の環境が未知のコマンドとして拒否するのはそのためです。自分のRedisのバージョンは redis-cli INFO server の redis_version で確認できます。回収ループを書かせる前に確かめておくと、「コマンドが存在しない」という失敗を実装の不具合と取り違えずに済みます。

6.2以降で使える構文は次のとおりです。

XAUTOCLAIM key group consumer min-idle-time start \
  [COUNT count] [JUSTID]

XAUTOCLAIMをループにするときの仕様は次のとおりです。

カーソルが 0-0 になっても、呼び出しを止める必要はありません。時間が経てば古い待機エントリが新たに条件を満たすので、0-0 から呼び直すとよいと書いています。常駐ワーカーなら、一定間隔で 0-0 から回収を始める形になります。

削除済みIDの配列は、Redis 7.0で加わった挙動です。待機リストにあるのに、ストリームからは消えている(トリムかXDELで)メッセージを、XAUTOCLAIMは回収せず、待機リストから取り除きます。取り除いたIDは3番目の戻り値に入ります。トリムを使うストリームでは、ここを無視すると「回収したはずの件数」と「待機リストが減った件数」が合いません。

Claude Codeに書かせる順番

ここからは、実装をClaude Codeに任せる部分です。公式のcommon workflowsには、テストを書かせる流れとして「テストの足場を作る → 意味のあるケースを足す → 実行して失敗を直させる」の順が載っています。さらに、確かめたい振る舞いを具体的に書くようにとも書かれています。この記事のテーマでは、「確かめたい振る舞い」が止まったワーカーの回収です。

実装より先に、CLAUDE.mdへ制約を置きます。

## Redis Streams ワーカーの規約
- 消費は XREADGROUP のコンシューマーグループで行い、処理が成功してから XACK する
- ACK前にプロセスが止まる前提で書く。回収は XAUTOCLAIM の min-idle-time を
  処理時間の最大値より長くする
- 同じメッセージが2回届いても結果が変わらないようにする(冪等)
- 回収のテストは、ACKしないワーカーを実際に止めて確かめる

そのうえで、依頼は次のようにします。

orders ストリームを消費するワーカーを実装して。
先にテストを書いて。テストでは、ワーカーAに3件読ませて
ACKさせずに止め、min-idle-time 経過後にワーカーBが
3件とも処理してACKし、XPENDINGの待機件数が0になることを確かめて。
実行して失敗したら直して。

この依頼で意図しているのは次の3点です。

1. テストが先にあるので、Claude Codeは「ACKしない状態」を作る方法を自分で考える必要がある
2. 「止める」を実際のプロセス停止にすると、コードの中でACKを省略するだけの偽の再現を避けられる
3. 待機件数が0という検証は、コマンドの出力で判定できる

テスト結果は、Redisの出力で判定できる形にしておくと反復が速くなります。たとえば XPENDING orders g1 の先頭の数字が 0 なら成功、というシェルの終了コードにまで落とせば、Claude Codeは「実行して失敗を直す」を人の判断なしで回せます。公式のcommon workflowsには、CIやバッチ向けに claude -p で非対話実行する方法も載っており、ワーカーのテストをそこへ繋ぐことも可能です。

なお、Redisへの接続や操作そのものをClaude Codeに任せたいときは、RedisのMCPサーバーの使い方が出発点になります。本記事は、接続した先で動かす消費側のコードが対象です。

min-idle-timeが短すぎると二重処理になる

回収の条件で一番の失敗要因は、min-idle-time を短く設定することです。処理に5秒かかるワーカーに対して、1秒で回収する設定を入れると、まだ処理中のメッセージを別のワーカーが奪います。XCLAIMはメッセージの持ち主を変えるだけで、元のワーカーの処理を止めません。両方が最後まで走り、同じメッセージが2回処理されます。

観察で確かめた「アイドル時間のリセット」は、この衝突を完全には防ぎません。リセットが防げるのは、複数の回収者が同じ瞬間に奪い合う場合だけです。処理中の元ワーカーまでは止められません。

防ぎ方は2つあります。

• min-idle-time を、処理時間の最大値より長く取る
• 処理側を冪等にし、2回届いても結果が同じになるようにする

一般には、完全な1回きりの処理は得られません。二重配信が起こりうる以上、冪等性は選択肢というより前提です。処理済みのIDを記録する方法は、アウトボックスパターンのクラッシュ試験で扱った受信側の冪等化と同じ考え方で書けます。

何度も失敗するメッセージを見分ける

Claude Codeが書いた回収ループは、放っておくと同じメッセージを永遠に回収し続けます。処理のたびにワーカーが落ちるメッセージ(たとえば、読み込むとクラッシュする不正なデータ)があるからです。

公式のXAUTOCLAIMのページは、システム的にコンシューマーがクラッシュする類のメッセージは、配信回数が高くなるので、監視で検出できると書いています。この配信回数は、先に見た範囲指定のXPENDINGで取得できます。

実装に入れる判断は、たとえば次のとおりです。

「N回で隔離する」というしきい値の具体値は、Redis側が定めているものではありません。ワーカーの処理時間や許容する再試行の回数から、サービスごとに決める値です。Claude Codeに書かせるときは、しきい値を定数として1か所に置かせ、テストでは小さい値で隔離まで通るようにします。BullMQでの失敗ジョブの扱いと見比べたい場合は、BullMQのワーカーを失敗ジョブで検証する記事が参考になります。

完了の確認

実装が済んだら、最後に次の4点を手元で確かめます。

• ACKせずに止めたワーカーの分が、min-idle-time 経過後に別のワーカーへ移る
• 移った後に、XPENDINGの待機件数が0になる
• 回収直後に、別のワーカーが同じIDを奪おうとしても何も返らない
• 同じメッセージが2回届いても、業務データが二重にならない

一番上の項目が通らないとき、まず疑うのはRedisのバージョンです。XAUTOCLAIMが6.2以降の機能で、CLAIM付きのXREADGROUPは8.4以降の機能である点は、Claude Codeに依頼する前にプロンプトへ書いておけます。

まとめ

コンシューマーグループの実装は、XREADGROUPとXACKの組み合わせで完結しているように見えて、本番の品質は「止まったワーカーの分をどう取り戻すか」で決まります。XCLAIMは6.2より前から、XAUTOCLAIMは6.2から使えるので、まず自分のRedisがどちらを許すかを確かめてください。

Claude Codeには、回収ループの実装より先に、ワーカーを実際に止めるテストを書かせます。テストが通ることと、min-idle-time を処理時間より長く取ること、冪等に作ることの3つが揃えば、停止したワーカーの未ACKが静かに溜まる事態は避けられます。キャッシュ側の負荷対策に興味があれば、キャッシュスタンピードの防止やレート制限の原子化も同じRedisの設計の話です。