⬡ API & SDK/2026-05-07上級

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

Claude Agent SDK で発生する「DBには書かれたのにメール/決済/通知が飛ばなかった」事故を防ぐため、Transactional Outbox パターンを Postgres と Cloudflare Queues を例に本番品質で実装する手順をまとめます。

claude-agent-sdk⁶ transactional-outbox production⁸⁷ reliability¹⁴ postgres queues

✦ プレミアム記事

「ユーザー登録は成功したのにウェルカムメールが飛んでいない」「注文DBには記録されたのに決済が走っていない」— Claude Agent SDK で書いたエージェントが本番に出ると、こういう「副作用の取りこぼし」が必ず一度は起こります。

私自身、4サイト分のメンバーシップを Claude Agent SDK と Stripe で動かしていて、最初の3ヶ月で似た事故を3回出しました。原因は毎回同じで、ツール内で db.insert(...) と sendEmail(...) を続けて呼んでいて、前者が成功した直後にプロセスが落ちると、後者がそのまま消えるのです。冪等性キーを正しくつけても、そもそも呼ばれなかった処理は冪等性では救えません。

この問題を構造的に解くのが Transactional Outbox パターンです。マイクロサービス文献では古典的なパターンですが、Claude Agent SDK のようにツール実行とDB更新が同居する世界での具体的な作り方は、まとまった日本語資料が少ないのが現状です。本記事は、私が本番運用で実際に組んだ構成を、Postgres と Cloudflare Queues を例に、そのままコピーして動く形でまとめたものです。

なぜ Agent SDK で「副作用の取りこぼし」が起きるのか

エージェントが落ちる原因はいくつかありますが、Outbox を語るうえで知っておきたいのは次の3つです。

ツール内 await の途中でループが死ぬ: db.insert の直後、sendMail の前で SDK タイムアウトや OOM が起きると、外部副作用は走らないままDBだけが進みます
モデル応答が壊れて再開がスキップされる: ツール結果のJSONパースエラーで Agent ループが新しいセッションとして再起動すると、モデルからは「もうそのツールは呼んだ」と見えるためリトライされません
手動で停止 → 翌日再開: Claude Agent SDK のセッション再開で会話履歴は復元されますが、未送信のメールはどこにも残っていないので、人間が気づくまで永久に飛びません

冪等性（前回記事 Claude Agent SDK で冪等性を設計するで詳しく扱いました）は「同じ副作用が二度走らない」を保証する技術ですが、ここで必要なのは 「副作用が必ず一度は走る」 を保証する技術です。Outbox はまさにそれを担います。

Transactional Outbox パターンとは

ひとことで言えば、「副作用を直接実行する代わりに、副作用を実行する依頼を業務データと同じトランザクションで outbox テーブルに書く」 だけのパターンです。

[Agent ツール]
    │
    ├─ BEGIN
    ├─ INSERT INTO orders ...
    ├─ INSERT INTO outbox (topic, payload) VALUES ('mail.send', ...)
    └─ COMMIT
              │
              ▼
[Outbox Dispatcher]
    │
    ├─ SELECT 未配信レコード
    ├─ Cloudflare Queues / SNS / Kafka に送信
    └─ UPDATE outbox SET dispatched_at = now()
              │
              ▼
[Consumer]
    └─ 実際にメール送信・決済・Slack通知 等

業務データと「副作用の依頼書」が同じトランザクションでコミットされるので、コミットされたなら依頼書は必ずある が成り立ちます。あとは依頼書を読んで非同期に副作用を起こすだけのジョブ（Dispatcher）が、 at-least-once でそれを送り出します。Consumer 側を冪等に作っておけば、結果として exactly-once 相当の振る舞いが手に入ります。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦Agent ループのクラッシュやタイムアウトで「DBには記録されたのに副作用が一度も実行されなかった」というタイプの事故を、原理的に防げる設計が手に入ります

✦Postgres + Cloudflare Queues を使った Outbox テーブル・ディスパッチャ・コンシューマの実装を、コピーすれば動く形で持ち帰れます

✦順序保証・重複排除・観測メトリクス・段階的な導入手順まで揃うので、既存の Claude Agent SDK アプリに今日から組み込めます

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

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 が赤くなります。必ず必須テストにしておきましょう。

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

正直に書くと「小さければほぼゼロ、大きくなっても安い」が答えです。私の4サイト合計で月50,000行ほど outbox に書かれていますが、Postgres は表200MB以下で平然としていますし、Cloudflare Queues はこの規模で月1ドルを切っています。

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