Claude Codeでインボックスパターンの冪等受信を実装し、2回流して検証する

少なくとも1回は届ける(at-least-once)ことを保証するブローカーでは、同じメッセージが複数回届きます。受信側が冪等でなければ、残高の二重引き落としのような不整合が起きます。解決策のひとつが、処理済みメッセージのIDを一意制約つきの表に記録するインボックステーブルです。記録と業務更新を同じトランザクションに入れるのが要点になります。

実装はClaude Codeに書かせられます。ただ、読んで正しそうに見えるコードと、並列に動かして壊れないコードは別物です。この記事では「同じメッセージを2回流す」試験と並列実行の試験を先に用意し、通るまで直させる流れを、手元の実行結果つきで示します。送信側(アウトボックス)はアウトボックスパターンの記事、Stripe webhookの受信はStripe webhookの冪等受信にあります。ここでは特定のサービスに依らない汎用コンシューマーを扱います。

インボックステーブルは何を記録する表か

Idempotent Consumerパターンの説明では、消費側が処理済みメッセージのIDをDBに記録し、重複を検出して捨てる形が基本とされています。記録先は2通りあります。専用のPROCESSED_MESSAGES表を持つ方法と、更新対象の業務エンティティ自身にIDを持たせる方法です。

専用表の場合の流れは次のとおりです。

主キーに受信者(subscriber)が入っている点は見落としやすい部分です。同じメッセージを複数のコンシューマーが別々の目的で受けるなら、IDだけを主キーにすると片方の処理が「処理済み」と誤判定されます。手元の確認でも、主キーを (consumer, message_id) にした表では、別のコンシューマー名で同じIDを入れると通り、同じ組を再び入れると UNIQUE constraint failed: inbox.consumer, inbox.message_id で失敗しました。

先に作る試験は「2回流す」と「並列で流す」

検証環境は、Python 3.12.2に同梱のSQLite 3.45.2です。ブローカーは使わず、メッセージのリストを受信関数へ渡して再配信を再現します。口座の残高1,000から、10ずつ50件を引き落とすので、正しい最終残高は500です。

表は2つだけです。

c.executescript("""
CREATE TABLE accounts(id INTEGER PRIMARY KEY, balance INTEGER NOT NULL);
INSERT INTO accounts VALUES (1, 1000);
CREATE TABLE inbox(
  consumer    TEXT NOT NULL,
  message_id  TEXT NOT NULL,
  received_at REAL NOT NULL,
  PRIMARY KEY (consumer, message_id)
);
""")

冪等化していない受信を、2つのワーカーで同じ50件ずつ流すと、残高は0になりました。各ワーカーが50件ぶんを引き落としたため、期待値500から500ずれています。試験が「重複を見逃す実装」を確実に落とせることの確認になります。

確認してから処理する実装が並列で破れる

ありがちな書き方は、先にSELECTで処理済みか調べ、未処理なら業務更新して記録する形です。

def check_then_act(c, m):
    seen = c.execute(
        "SELECT 1 FROM inbox WHERE consumer='debit' AND message_id=?",
        (m["id"],)).fetchone()
    if seen:
        return
    c.execute("BEGIN IMMEDIATE")
    apply_debit(c, m["amount"])
    c.execute("INSERT OR IGNORE INTO inbox VALUES ('debit', ?, ?)",
              (m["id"], time.time()))
    c.execute("COMMIT")

1本のワーカーで2回流す試験なら、これは通ります。2回目のSELECTが処理済みを見つけるからです。

並列にすると壊れます。4ワーカーで同じ50件を流したところ、期待値500に対して残高は負になりました。実行ごとに値は変わり、手元の4回では -500 / -280 / -330 / -340でした。インボックスには50行しか入っていないので、記録だけ見ると「全件1回ずつ処理済み」に見えるのが厄介です。SELECTとINSERTの間に別のワーカーが割り込み、全員が「未処理」と判断したことが原因です。

したがって「2回流す」試験だけでは足りません。重複の検出と業務更新を並列に走らせる試験を、同じ重みで置きます。

同一トランザクションと一意制約で書き直す

直し方は、確認と記録を別の操作にせず、挿入の成否を判定そのものにすることです。

def inbox(c, m):
    c.execute("BEGIN IMMEDIATE")
    try:
        c.execute("INSERT INTO inbox VALUES ('debit', ?, ?)",
                  (m["id"], time.time()))
        apply_debit(c, m["amount"])
        c.execute("COMMIT")
        return "applied"
    except sqlite3.IntegrityError:
        c.execute("ROLLBACK")
        return "duplicate"
    except Exception:
        c.execute("ROLLBACK")
        return "failed"

同じ4ワーカー×50件で流すと、残高は期待どおり500、インボックスは50行でした。重複は主キー違反として検出され、業務更新は1回しか走りません。

ここで検証しているのはSQLiteの動作です。別のDBでは、同じキーの挿入が並行したときの待ち方やエラーの出方が異なる場合があります。使うDBで同じ試験を流して確かめてください。

失敗したら再配信で処理できるか

もう一つ、見落としやすいのが「処理に失敗したメッセージ」の扱いです。ブローカーが再配信するのは、処理が失敗したり確認応答が返らなかったりしたからです。そのとき受信側が「処理済み」と記録していると、再配信されたメッセージが握りつぶされます。

2通りの実装で、1件目の処理を途中で失敗させ、そのあと同じメッセージを再配信しました。

同一トランザクションなら、失敗時は記録ごとロールバックされるので、再配信で正しく1回だけ適用されます(残高990)。記録を先に確定させると、業務更新が失敗してもIDだけ残り、再配信は重複扱いで捨てられます(残高1,000のまま、10の引き落としが永久に失われます)。

「重複を1回で止める」ことと「失敗を再試行で回復する」ことは、同じ一つのトランザクション境界でまとめて満たされます。この試験項目が無いと、重複排除が成功しているように見えて取りこぼしが出ます。

Claude Codeに渡す指示とCLAUDE.mdの規約

Claude Codeの典型的なワークフロー集では、テストを追加するときに「どの振る舞いを検証したいかを具体的に書く」ことが勧められています。冪等性はまさにそれで、「重複しても結果が変わらない」を試験の言葉に直して渡します。プロンプトは次のような形にできます。

orders-consumer の受信関数に、インボックステーブルでの冪等化を追加して。
- 主キーは (consumer, message_id)。メッセージIDはブローカーが付けた値を使う
- 記録の挿入と業務更新は同じトランザクションに入れる
- 先にSELECTで確認する実装にしない。挿入の一意制約違反を重複の判定にする
- 業務処理が例外のときは記録もロールバックされること
先にテストを書いて: (1) 同じメッセージを2回流す (2) 4並列で同じ50件を流す
(3) 1回目を失敗させてから再配信する。テストが落ちることを確認してから実装して。

チームで同じ規約を毎回書かなくて済むよう、CLAUDE.mdにも置けます。

## メッセージ受信の規約
- コンシューマーは at-least-once を前提に、必ず冪等にする
- 処理済みIDの記録は一意制約つきの表で行い、業務更新と同じトランザクションに入れる
- 「SELECTで確認してから処理」は禁止。挿入の失敗で重複を判定する
- 冪等性の変更には「2回流す」「並列で流す」「失敗後に再配信」の3つの試験を付ける

試験を回して直させる

実装後の手順は単純です。Claude Codeに試験を実行させ、失敗した理由を読ませて直させます。ワークフロー集にも、新しい試験を走らせて失敗を直すよう頼む例が載っています。

python3 -m pytest tests/test_inbox.py -q

最初の実装が check_then_act の形で出てきたら、並列試験が落ちます。そのとき、落ちた出力(残高が負になる行)を見せて「確認と記録を分けているのが原因」と指摘すれば、一意制約を使う形へ直ります。ここでの狙いは、実装の良し悪しをレビューで言い当てることではなく、落ちる試験で判定を機械に渡すことです。

並列試験は実行ごとに結果が揺れます。1回通っただけでは安心できないため、並列試験は回数を増やして流す、ワーカー数を上げる、といった強め方を指示に含めておくと取りこぼしが減ります。

インボックスで守れない範囲

インボックスが守るのは、同じDBに閉じた更新です。次の場面は別の手当てが要ります。

• 外部への副作用: メールの送信や外部APIの呼び出しは、ロールバックで取り消せません。DB更新はロールバックされても、先に出た外部呼び出しは残ります。外部呼び出しには相手側の冪等キーを使うか、送信自体をアウトボックスに積みます
• 記録の保持期間: 処理済みIDを無期限に残すと表が膨らみます。消すなら、ブローカーが再配信しうる最長期間より長く残す必要があります。短く消すと、遅れて来た重複が通ります
• 中身が同じで別IDのメッセージ: IDが違えば別メッセージとして扱われます。送信側が再送のたびに新しいIDを振る設計では、インボックスは効きません。IDは送信側で一度だけ決めて、再送でも変えないことが前提です
• ブローカー側の重複排除: Amazon SQSのFIFOキューでは、SendMessageを再試行しても、5分の重複排除の間隔内なら重複が入らないとされています。これは送信の再試行に対する保護で、受信側の再処理を防ぐものではありません

よくあるつまずき

• メッセージIDを受信側で生成してしまう: 受信のたびに新しいUUIDを作ると、重複を検出できません。ブローカーが付けるID、または送信側が決めた業務IDを使います
• 記録だけ先にコミットする: 前の節の表のとおり、失敗した処理が二度と実行されなくなります
• 並列試験を省く: 2回流すだけでは、確認してから処理する実装が通ってしまいます
• 重複を例外としてログに流す: 重複は正常系です。エラー扱いにすると、アラートが再配信のたびに鳴ります

まとめ

受信側の冪等化は、一意制約つきのインボックス表への挿入と業務更新を同じトランザクションに入れれば足ります。試験は「2回流す」だけでなく、「並列で流す」と「失敗後に再配信」の3つを揃えて初めて、確認してから処理する実装と、記録だけ先に確定する実装の両方を落とせます。Claude Codeには、この3つの試験を先に書かせ、落ちる様子を見てから実装させる流れが向いています。

関連する実装として、ジョブ基盤側の冪等性はSidekiqのジョブの冪等性、再試行の挙動はBullMQのワーカーの再試行にあります。