Posted inデータベース

Transactional Outbox Pattern：マイクロサービスにおける信頼性の高いイベント発行

Database tutorial - IT technology blog
Database tutorial - IT technology blog

クイックスタート（5分）：二重書き込み（デュアルライト）の悪夢を断ち切る

典型的な注文決済フローを思い浮かべてみてください。顧客がECサイトで注文を行い、サービスがその注文データをPostgreSQLにコミットした直後に、order_createdイベントをRabbitMQやApache Kafkaへ送信します。

# 素朴なアプローチ：二重書き込み（デュアルライト）のアンチパターン
def create_order(order_data):
    db.orders.insert(order_data)  # ステップ1: Postgresに保存
    message_broker.publish("order_created", order_data)  # ステップ2: これが失敗した場合は？

もしデータベースへのコミット直後にKafkaとの接続が切断されたらどうなるでしょうか？データベースには注文情報が保存されますが、請求、倉庫、通知などの後続サービスには一切伝達されません。処理の順序を逆にしても同様に破綻します。先にメッセージブローカーへ発行し、その後のデータベーストランザクションがロールバックされた場合、ダウンストリームのワーカーは存在しない「幻の注文」を処理することになってしまいます。

エンジニアはこの課題を二重書き込み問題（デュアルライト問題）と呼びます。リレーショナルデータベースとメッセージブローカーは単一の分散トランザクションを共有できないため、両システムにまたがる書き込みの整合性を保つのは極めて困難です。2相コミット（2PC）はマイクロサービスにおいてはあまりに遅く脆弱すぎます。そこで活用されるのがTransactional Outbox Pattern（トランザクショナルアウトボックスパターン）です。

5分でわかるアウトボックスの解決策

リクエスト処理のパスで2つの異なるシステムと通信するのはやめましょう。ビジネスエンティティと送信予定メッセージの両方を、単一のアトミックなACIDトランザクション内で同一のデータベースに書き込むのです。

-- データベース内にアウトボックステーブルを作成
CREATE TABLE outbox_messages (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type VARCHAR(50) NOT NULL,
    aggregate_id VARCHAR(50) NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    processed BOOLEAN DEFAULT FALSE
);

これで、両方のレコードをまとめてコミットできます：

def place_order(order_data):
    with db.transaction():
        order_id = db.orders.insert(order_data)
        db.outbox.insert({
            "aggregate_type": "order",
            "aggregate_id": str(order_id),
            "event_type": "OrderPlaced",
            "payload": order_data,
            "processed": False
        })

これらは1つのローカルトランザクションを共有しています。両方のレコードがディスクにコミットされるか、あるいはどちらもコミットされないかのどちらかです。その後、専用のバックグラウンドワーカーがoutbox_messagesから行を読み取り、安全にブローカーへ送信します。

詳細解説：Transactional Outbox Patternの仕組み

このパターンは、ビジネスドメインロジックを非同期のメッセージ転送処理から切り離します。リアルタイムな外部ネットワーク呼び出しを、データベース駆動の信頼性の高いパイプラインへと変換します。

主要コンポーネント

  • コマンドハンドラー（Command Handler）： 受信したAPIコールを処理し、ローカルトランザクション内でordersなどのドメインテーブルを更新すると同時に、送信イベントをoutbox_messagesに書き込みます。
  • アウトボックステーブル（Outbox Table）： プライマリデータストア（PostgreSQL、MySQL、SQLiteなど）の内部で、永続化されたクエリ可能なステージングキューとして機能します。
  • メッセージリレー（Message Relay）： アウトボックステーブルから未処理のメッセージを取り出し、ブローカー（Kafka、RabbitMQ、AWS SNS/SQSなど）へ転送する非同期バックグラウンドワーカーです。

メッセージリレーの2大戦略

1. ポーリングパブリッシャー（Polling Publisher）

バックグラウンドワーカーが定期的な間隔（例: 500ミリ秒ごと）でデータベースにクエリを実行します：

SELECT * FROM outbox_messages
WHERE processed = FALSE
ORDER BY created_at ASC
LIMIT 100
FOR UPDATE SKIP LOCKED;

ここで極めて重要なのがFOR UPDATE SKIP LOCKED句です。これにより、他のワーカースレッドが現在ロックしている行をスキップするようPostgresに指示できます。分散ロックや競合状態（レースコンディション）を気にすることなく、4〜8個のリレーインスタンスを並行稼働させることが可能です。バッチ処理がブローカーに正常に届いたら、ワーカーはprocessed = TRUEに更新するか、行を完全に削除します。

2. トランザクションログの追跡（Change Data Capture / CDC）

プライマリテーブルに対してポーリングクエリを実行する代わりに、データベースエンジンのトランザクションストリームを直接読み取る方法もあります。Debeziumなどのツールは、PostgreSQLのWrite-Ahead Log（WAL）やMySQLのbinlogを読み取り、10ミリ秒未満のレイテンシで変更をKafkaへストリーミングします。これによりポーリングクエリが完全に不要となり、プライマリデータベースのクエリ負荷を軽減できます。

応用編：並行性、メッセージ順序、障害への対応

基本的な仕組みを動かすだけなら半日もあれば十分です。しかし、本番環境での障害に対応するには綿密な設計が求められます。

少なくとも1回の配信（At-Least-Once Delivery）とべき等性

ネットワーク障害は避けられません。リレーワーカーがKafkaへのメッセージ送信に成功した後、PostgreSQLにprocessed = TRUEをコミットする直前にクラッシュする可能性があります。そのワーカーが再起動すると、まったく同じイベントを再送することになります。アウトボックスパターンが保証するのは少なくとも1回の配信（at-least-once delivery）であり、厳密に1回（exactly-once）ではありません。

そのため、ダウンストリームのコンシューマーはべき等（idempotent）でなければなりません。処理済みメッセージIDは、必ず専用の重複排除テーブルに記録するようにしてください：

def handle_order_placed(event):
    event_id = event["id"]
    with db.transaction():
        if db.processed_events.exists(event_id):
            return  # 重複メッセージを受信。安全に無視する
        
        apply_billing(event["payload"])
        db.processed_events.insert({"event_id": event_id})

パーティショニングとイベントの順序保証

例えば、ある注文が2秒の間に「作成」「配送先変更」「キャンセル」と3回更新されたとします。これらが出荷サービスに順不同で到着した場合、キャンセルされた注文を誤って発送してしまうリスクがあります。アウトボックスの行をKafkaに発行する際は、メッセージキーにaggregate_id（order_idなど）を設定してください。Kafkaはこのキーをハッシュ化し、その注文に関するすべての更新を完全に同一のパーティションにルーティングするため、厳密なFIFO順序が保たれます。

大量データ下でのアウトボックスのクリーンアップ

1日あたり5万件の注文を処理するECサービスでは、毎月150万行のアウトボックスデータが生成されます。放置するとテーブルの肥大化（テーブルブロート）によってスキャン速度が低下し、ディスク容量を圧迫します。実践的な対策として以下の3つの戦略があります：

  1. 送信時の即時物理削除（Hard Delete on Publish）： ブローカーから受領通知（ACK）を受け取った瞬間にDELETE FROM outbox_messages WHERE id = :idを実行します。これによりアウトボックステーブルは常に軽量に保たれ、数十行を超えることはほとんどありません。
  2. 宣言的テーブルパーティショニング（Declarative Table Partitioning）： outbox_messagesを日単位でパーティショニングします。該当日のパーティションの処理が完了したらDROP TABLEを実行します。行ごとの重いVACUUM処理を発生させることなく、ディスク容量を即座に解放できます。
  3. スケジュールされた定期バッチ削除（Scheduled Batch Pruning）： トラフィックの少ない時間帯に夜間クリーンアップジョブを実行し、不要になった行を削除します：DELETE FROM outbox_messages WHERE processed = TRUE AND created_at < NOW() - INTERVAL '3 days' LIMIT 10000;。

実践的なヒント：実装チェックリストと現場からのアドバイス

本番環境にアウトボックステーブルを導入する際は、実績に裏打ちされた以下のルールを意識してください：

  • ペイロードを自己完結させる： すべての状態変更をイベントペイロード内に直接含めます。不足している顧客詳細を取得するためだけに、コンシューマーが元のサービスのREST APIをわざわざ呼び出さなければならないような設計は避けましょう。
  • ネイティブな構造化データとして保存する： インデックスのない文字列ではなく、PostgreSQLのJSONB型を使用します。処理が滞ったペイロードをデバッグする際、イベントプロパティに対して柔軟な調査クエリを実行できます。
  • ポーリングクエリに最適化したインデックスを作成する： 部分複合インデックスを追加します：CREATE INDEX idx_unprocessed_outbox ON outbox_messages (created_at ASC) WHERE processed = FALSE;。これにより、過去の履歴レコードが蓄積してもインデックス検索を常に高速に保てます。
  • リレー遅延のアラートを設定する： processed = FALSEの最も古い行の経過時間を監視します。30秒以上未処理のメッセージが存在する場合は、PagerDuty等を通じてオンコールエンジニアに通知を送ります。

ステージング環境へのデータ投入やテスト用ペイロードのモック作成時、手動でのデータ変換は手間がかかります。アウトボックスのテスト実行用にCSVフィクスチャデータをJSONに変換する必要がある場合、toolcraft.app/ja/tools/data/csv-to-jsonを使えばブラウザ上で直接変換できるため、内部スキーマがローカル環境の外に出る心配もありません。エッジケースを検証する際に使い捨てのインポートスクリプトを書く手間を省けます。

イベント生成をローカルデータベースのトランザクションに直接紐付けることで、データ不整合の根本原因を断ち切り、高負荷下でも耐えうる堅牢なマイクロサービス間通信の基盤を構築できます。

Share:
Tags: