Claude Agent SDK の Transactional Outbox 実装 — 「副作用の取りこぼし」を本番でゼロにする設計

Outbox テーブルのスキーマ設計

最低限必要なカラムは多くありませんが、運用で後悔しないために最初から入れておきたいものを示します。

-- Postgres の例
CREATE TABLE outbox (
  id            BIGSERIAL PRIMARY KEY,
  aggregate_id  TEXT        NOT NULL,            -- 注文ID等。partitioning key としても使う
  topic         TEXT        NOT NULL,            -- 'mail.send' / 'stripe.charge' 等
  payload       JSONB       NOT NULL,
  headers       JSONB       NOT NULL DEFAULT '{}'::jsonb,
  created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
  available_at  TIMESTAMPTZ NOT NULL DEFAULT now(), -- 遅延配信用
  dispatched_at TIMESTAMPTZ,
  attempts      INT         NOT NULL DEFAULT 0,
  last_error    TEXT
);
 
-- 未配信レコードを高速に拾うインデックス
CREATE INDEX outbox_pending_idx
  ON outbox (available_at)
  WHERE dispatched_at IS NULL;
 
-- aggregate_id 単位で順序を守りたいときに役立つ
CREATE INDEX outbox_aggregate_idx
  ON outbox (aggregate_id, id)
  WHERE dispatched_at IS NULL;

available_at を持たせておくと、リトライのバックオフや「3分後にウェルカムメールを送る」のような遅延配信を同じテーブルで扱えます。headers には trace_id・schema_version・tenant_id を入れておくと、後から監視と移行で大きく助かります。

私が運用で痛感したのは、dispatched_at IS NULL の Partial Index を最初から張らないと、テーブルが100万行を超えたあたりで Dispatcher のクエリが急激に遅くなることです。未配信だけを引くクエリは必ず Partial Index で、と覚えておくのが良いです。

ツール内で原子的に Outbox に書く

Claude Agent SDK のツール本体は、副作用そのものを呼ばず「Outbox に依頼書を入れる」だけにします。

// tools/createOrder.ts
import { tool } from "@anthropic-ai/agent-sdk";
import { z } from "zod";
import { db } from "../db";
 
export const createOrder = tool({
  name: "create_order",
  description: "注文を作成し、関連する副作用を outbox に登録する",
  input_schema: z.object({
    user_id: z.string(),
    amount: z.number().int().positive(),
    currency: z.enum(["JPY", "USD"]),
    idempotency_key: z.string(), // モデルではなく上位層で生成して渡す
  }),
  async run(input, ctx) {
    return await db.transaction(async (tx) => {
      // 1) 業務データ
      const order = await tx
        .insertInto("orders")
        .values({
          user_id: input.user_id,
          amount: input.amount,
          currency: input.currency,
          idempotency_key: input.idempotency_key,
          status: "pending",
        })
        .onConflict((oc) => oc.column("idempotency_key").doNothing())
        .returningAll()
        .executeTakeFirst();
 
      if (!order) {
        // 既に同じキーで作成済み → 二重コール
        return { ok: true, deduped: true };
      }
 
      // 2) 副作用依頼書（同じトランザクション内）
      await tx.insertInto("outbox").values([
        {
          aggregate_id: order.id,
          topic: "stripe.charge",
          payload: {
            order_id: order.id,
            amount: order.amount,
            currency: order.currency,
          },
          headers: { trace_id: ctx.trace_id, schema_version: 1 },
        },
        {
          aggregate_id: order.id,
          topic: "mail.send",
          payload: {
            template: "order_received",
            user_id: order.user_id,
            order_id: order.id,
          },
          headers: { trace_id: ctx.trace_id, schema_version: 1 },
        },
      ]).execute();
 
      return { ok: true, order_id: order.id };
    });
  },
});

このツールが返した時点で、業務データと「あとで Stripe を叩く」「あとでメールを送る」依頼が原子的にコミットされています。ここから先は Dispatcher の責務であり、Agent のループが落ちようが Cloudflare Workers の CPU 制限に引っかかろうが、依頼書は残ります。

ポイントは、Agent SDK のツールから直接 Stripe や SES を呼ばないことです。これが守れていないと、いくら Outbox を導入しても DB と外部副作用の整合性は壊れます。レビュー時のチェックポイントとして「ツール本体の中で fetch が呼ばれていたら赤信号」と覚えておくと、コードベース全体で習慣化できます。CI で fetch 呼び出しを検出する lint ルールを足したい場合は、Claude Code で安全に ESLint を回す手順 のセットアップをそのまま流用できます。

Outbox Dispatcher を実装する

最初は素朴なポーリング型で十分です。Cloudflare Workers の Cron Trigger を使えば、別サービスを増やさず動かせます。

// dispatcher/poll.ts
import { db } from "../db";
import { sendToQueue } from "../queue";
 
export async function dispatchBatch(now = new Date()) {
  // 1) ロックを取りながら未配信レコードを取り出す
  const rows = await db.transaction(async (tx) => {
    const pending = await tx
      .selectFrom("outbox")
      .selectAll()
      .where("dispatched_at", "is", null)
      .where("available_at", "<=", now)
      .orderBy("id", "asc")
      .limit(200)
      .forUpdate()        // 別ワーカーと競合しないように行ロック
      .skipLocked()       // 取れない行は飛ばす(=並列化可能)
      .execute();
 
    return pending;
  });
 
  // 2) Queue に流す（成功時のみ dispatched_at を立てる)
  for (const row of rows) {
    try {
      await sendToQueue(row.topic, {
        id: String(row.id),
        aggregate_id: row.aggregate_id,
        payload: row.payload,
        headers: row.headers,
      });
      await db
        .updateTable("outbox")
        .set({ dispatched_at: new Date() })
        .where("id", "=", row.id)
        .execute();
    } catch (e) {
      const next = backoff(row.attempts);
      await db
        .updateTable("outbox")
        .set({
          attempts: row.attempts + 1,
          last_error: String(e),
          available_at: new Date(now.getTime() + next),
        })
        .where("id", "=", row.id)
        .execute();
    }
  }
}
 
function backoff(attempts: number) {
  // 30秒, 1分, 2分, 4分, ... ジッタ込みで上限15分
  const base = Math.min(30_000 * 2 ** attempts, 15 * 60_000);
  return base + Math.floor(Math.random() * 5_000);
}

FOR UPDATE SKIP LOCKED が肝で、複数の Dispatcher を同時に走らせても 同じ outbox 行を二重に処理しない ことが Postgres レベルで保証されます。これを知らずに LIMIT 200 だけで分けようとすると、ピーク時に重複配信が発生して、せっかくの Outbox の信頼性が崩れます。

期待される動作の確認には、SELECT count(*) FROM outbox WHERE dispatched_at IS NULL をメトリクスとして毎分エクスポートし、Grafana に出すのがおすすめです。これが平常時に常時0付近、ピーク時でも数百件で落ち着いていれば健康な Dispatcher の証拠になります。

順序保証と Partitioning Key

「同じ注文に対するイベントは順番通りに処理してほしい」という要求は本番で必ず出てきます。Outbox 自体は ORDER BY id で取り出すので入力順は保たれますが、Queue より先（コンシューマ側）で順序が崩れる のが要注意です。

Cloudflare Queues や SQS FIFO では partition key（または MessageGroupId）に aggregate_id を渡すと、同一注文IDのメッセージは順序が保たれます。逆に、partition key を入れず単に投入してしまうと、コンシューマの並列度が上がるほど順序が崩れます。

await env.MY_QUEUE.send(
  { topic, payload, headers },
  { contentType: "json", partitionKey: aggregate_id }, // ← 必須
);

順序を厳密に保ちたい topic と、そうでない topic を分けて Queue を用意するのも有効な戦略です。たとえばメール送信は順序非依存ですが、決済と返金は厳密順序が必要、という具合に分割します。

CDC（Change Data Capture）型への進化

ポーリング型 Dispatcher は最大数百ms 〜 数秒のラグが避けられません。これが許容できなくなったら、Postgres の論理レプリケーション (wal2json / pgoutput) で outbox テーブルの INSERT を Debezium などにストリームし、直接 Queue に流す CDC 型に切り替えると、ラグが数十msまで縮みます。

ただし最初から CDC を導入するのはおすすめしません。私は最初の半年はポーリング型で運用し、p95 ラグが 5 秒を超えるようになってから CDC へ移行しました。CDC は運用コストが跳ね上がる（レプリケーションスロットの監視・スキーマ変更時の調整など）ため、ポーリングで困ってから移行するのが正しい順序です。

観測可能性: 必須メトリクスとアラート

最低でも次の3つを Prometheus などで取り、Grafana に常時出すのが本番運用の最低ラインです。

• outbox_pending_total — 未配信レコード数（gauge、毎分）
• outbox_dispatch_latency_seconds — created_at から dispatched_at までの差（histogram）
• outbox_attempts_histogram — リトライ回数の分布

アラートは outbox_pending_total > 1000 が5分続いた場合、p95(dispatch_latency) > 30s が10分続いた場合、outbox_attempts > 5 の行が出現した場合の3本でほぼ事故は捕まえられます。

私が一番助けられたのは last_error カラムです。dispatcher が落ち続けているとき、ここに最後のエラー文字列が残っていると、調査が10倍速くなります。ログだけ見ていると、どの行で何が起きたのかすぐに失われがちです。

よくある落とし穴とアンチパターン

実装段階で必ず一度ハマる箇所を、先に潰しておくためにまとめておきます。

第一に、ツール内で副作用を直接呼ぶコードと、Outbox 経由を混在させる のは最悪の状態です。「急いでいるから今回だけ直接呼ぶ」が積み重なると、何が Outbox を通っていて何が通っていないかが分からなくなります。社内ルールとして「ツールから外部 API を直接叩くPRは即リジェクト」と決めるのが現実的です。

第二に、Consumer 側を冪等にしないことです。Outbox は at-least-once です。同じメッセージが必ず2回以上は来る前提で、Consumer 側に冪等性キー（前回記事参照）を入れる必要があります。これを怠ると、リトライ時にメールが2通飛ぶ・課金が2回走るといった事故が必ず再発します。

第三に、Outbox テーブルを永久に肥大化させる ケース。dispatched_at IS NOT NULL AND created_at < now() - interval '30 days' を毎日 DELETE（または別テーブルにアーカイブ）する Cron を必ず仕込みましょう。Partial Index は健全な状態を保ちますが、全件 VACUUM のコストはじわじわ効いてきます。

第四に、Dispatcher のスケジュール間隔を1秒以下に詰めること。Cloudflare Workers の Cron は最短1分ですし、もっと短いスケジューラを別建てすると、Postgres 側の FOR UPDATE 競合が増えてスループットがむしろ落ちます。本当にラグを縮めたいなら CDC 型への移行を検討するのが正解です。

コンシューマ実装のひな形

Dispatcher の責務は「Queue に渡したらおしまい」です。実際に副作用を起こすのは Consumer 側で、ここを冪等にしておかないと outbox の at-least-once が裏目に出ます。Cloudflare Queues コンシューマの最小実装を載せます。

// consumers/mailConsumer.ts
import type { MessageBatch } from "@cloudflare/workers-types";
import { db } from "../db";
import { mailer } from "../mailer";
 
interface OutboxMessage {
  id: string;             // outbox.id をそのまま冪等性キーに使う
  aggregate_id: string;
  payload: { template: string; user_id: string; order_id: string };
  headers: { trace_id?: string; schema_version: number };
}
 
export default {
  async queue(batch: MessageBatch<OutboxMessage>, env: Env) {
    for (const msg of batch.messages) {
      const body = msg.body;
      try {
        const seen = await db
          .selectFrom("processed_messages")
          .select("id")
          .where("id", "=", body.id)
          .executeTakeFirst();
        if (seen) { msg.ack(); continue; }
 
        await mailer.send({
          template: body.payload.template,
          user_id: body.payload.user_id,
          metadata: { order_id: body.payload.order_id, trace_id: body.headers.trace_id },
        });
 
        await db.insertInto("processed_messages")
          .values({ id: body.id, processed_at: new Date() })
          .onConflict((oc) => oc.column("id").doNothing())
          .execute();
 
        msg.ack();
      } catch (e) {
        msg.retry({ delaySeconds: 60 });  // 例外を握りつぶさず Queue に再配送させる
      }
    }
  },
};

processed_messages は Consumer 側の冪等性台帳です。14日でプルーニングすれば肥大化しません。Resend や Postmark のように API 側で idempotency key を受け付けるサービスを使う場合は、body.id をそのまま渡せば台帳すら不要です。

ここで「失敗時に ack しない」を強調しているのは、握りつぶされた例外こそが、Outbox が防ぎたい「メッセージの静かな消失」を再現するからです。Consumer の失敗は必ず大きな声で失敗させてください。

既存システムへ後付け導入する3つの戦略

新規プロジェクトに Outbox を入れるのは簡単ですが、すでに1年動いているサービスに入れるときは段取りが命です。私が試して効果があった3つの戦略を、効くシーン順に並べます。

最も簡単なのは 「ある日付以降だけ Outbox に乗せる」 方式です。カットオフ日以降に作られた注文だけが Outbox 経由になり、それ以前の注文は旧コードのまま。多くのプロダクトはこれで十分です。1ヶ月前のメールが今さら届いても誰も嬉しくないからです。

次が 「ダブルライト移行」。新旧両方のコードパスを同時に走らせ、Outbox 側はペイロードに shadow_mode: true を付けて Dispatcher で短絡させます。1週間メトリクスを比べて挙動が一致したら旧コードを撤去します。

最後が 「既存行から intent を逆算して outbox を埋める」 やり方です。SELECT * FROM orders WHERE welcome_email_sent_at IS NULL AND created_at > now() - interval '7 days' のような条件で過去ぶんを outbox に追記します。慎重にやれば取りこぼしを救えますが、雑にやると数週間ぶんのウェルカムメールを一斉送信してしまいます。必ず --dry-run から始めるのが鉄則です。

Cloudflare 環境の落とし穴

スタックが Cloudflare Workers + D1 / Workers + 外部 Postgres の場合、特有の制約をいくつか先に知っておくと事故が減ります。

D1 は現時点で FOR UPDATE SKIP LOCKED をサポートしていません。Dispatcher を複数並列で走らせたい場合は、別建ての outbox_lease テーブルでロックを取るか、id % N でシャードするしかありません。低トラフィックの個人開発であれば Cron Trigger 上の単一 Dispatcher で十分まわります。

Workers Cron は最短 1 分間隔です。秒単位のラグが必要になったら、長寿命の Durable Object でポーリングするか、CDC に切り替える方が筋が良いです。複数 Cron を時差発火で疑似的に詰めるのは、想定外のレース条件を呼び込むのでやめましょう。

Cloudflare Queues はバッチサイズと処理ウィンドウ（30秒）に制限があります。下流の API 呼び出しが遅い場合、大きなバッチより小さなバッチを多く流す方が安定します。Consumer 内のタイムアウトには防衛的に。

エッジでの trace_id 伝播は標準では行われません。cf-trace-id のような独自ヘッダーを Workers のエッジで付与し、outbox.headers 経由で Consumer まで運ぶ仕組みを最初から仕込んでおくと、深夜の障害対応で過去の自分に感謝することになります。

CI で守るべき2つのテスト

Outbox は契約テストとの相性が非常に良いパターンです。私が書いて最も効いたテストは次の2つです。

「業務書き込み失敗時に outbox 行が残らない」テスト。ツール呼び出しを wrap し、二つ目の insert の直後に強制例外を投げて、SELECT count(*) FROM outbox が増えていないことを確認します。これは「outbox.insert を transaction の外に出してしまう」典型的なリファクタリング事故を即座に捕まえます。

「Dispatcher 中断時に再配送される」テスト。Queue 側はモック成功・DB の dispatched_at 更新だけ失敗するように仕掛け、もう一度 Dispatcher を走らせると同じメッセージが2回送られることを確認します。これは at-least-once が「お題目」ではなく実装として効いていることを保証します。

どちらも50行以内で書けて、誰かが「このトランザクション、要らないよね」と最適化を試みた瞬間に CI が赤くなります。必ず必須テストにしておきましょう。

コストはどれくらいかかるのか

正直に書くと「小さければほぼゼロ、大きくなっても安い」が答えです。月数万行規模であれば Postgres の表は数百MB以内に収まり、Cloudflare Queues もこの規模なら無料枠か月1ドル未満で済みます。

コストが目に見えるのは、もともと同期だった処理を非同期化した場面です。サインアップ直後のウェルカムメールは送信完了を待たなくなる（=サインアップが速くなる）一方で、メール自体の到着は1〜2秒遅れます。99%のプロダクトでこれは正味プラスですが、ユーザーが結果を待っているインタラクティブなフローは Outbox 化しないのが正解です。なお Consumer 側で Anthropic API を呼び出す処理がある場合は、Outbox の上に Claude API のバジェットサーキットブレーカー を組み合わせると、暴走したリトライによる課金事故を二重に防げます。

Outbox を選ぶべきでない場面

公平のために、Outbox が向かないケースも書いておきます。

副作用が 同じDBへのリードクエリだけ であれば Outbox は不要です。後で取り直せばいい話だからです。Outbox は「外部に出ていく イベント」に対するパターンです。

副作用が 同期的に完了している必要がある（たとえばユーザーが結果画面を待っている決済確認）場合、Queue を挟むと体感のレイテンシが増えます。この場合は Saga パターンと明示的な補償処理の組み合わせか、同期呼び出し＋別途リコンサイルジョブの方が適しています。

業務書き込みが 強いトランザクションを持たないストア（DynamoDB や弱整合の Spanner 構成など）であれば、「業務行と outbox 行を原子的にコミット」が無料では成立しません。条件付き書き込みやストア固有のトランザクションで補強する必要があります。

それ以外のほぼすべてのケースで、Outbox は導入コストに見合うどころか、最初の事故を1件防いだ瞬間に元が取れます。

次に試してほしい一歩

Outbox は実装よりも、最初の Topic を1つ選んで実走させるところに最大の壁があります。手元にあるエージェントから、副作用を1つだけ Outbox 化してみてください。私の経験では、それが回り始めた瞬間に「なぜもっと早く入れなかったのか」と感じるはずです。

最後までお読みいただきありがとうございました。同じ事故で深夜に飛び起きる人が一人でも減れば、これ以上のことはありません。