Posted inDevelopment

Transactional Outbox Pattern: PostgreSQLとNode.jsでデータ損失を防ぐ究極の手法

Development tutorial - IT technology blog
Development tutorial - IT technology blog

悪夢：データベースとメッセージブローカーの「ズレ」

想像してみてください。顧客が200万ドンの注文を決済し、システムは正常に課金処理を行い、データベースへの保存も完了しました。しかし、その瞬間にネットワークが不安定になり、在庫管理サービスが発送準備のメッセージを受け取れませんでした。結果として、顧客はお金を払ったのに注文は放置され、あなたは夜中の2時までログを突き合わせ、手作業でデータを修正する羽目になります。

私は以前、大規模な決済システムでイベントメッセージの約5%が消失するというトラブルを直接経験しました。当時、チームは単純な方法を採用していました。PostgreSQLに注文を保存した後、publishMessage() を呼び出してRabbitMQに送信するというものです。トラフィックが増加した際、データベースのコミットは完了したものの、メッセージ送信コマンドがタイムアウトしてしまいました。注文データはDB内に残されたまま、後続のインフラサービスはそれを全く検知できない状態に陥ったのです。

そのトラブルのせいで、データ復旧用のスクリプトを書くのに丸3日を費やしました。ここでの教訓は、「厳格な保護メカニズムがなければ、2つの独立したシステムが常に同期して動作すると信じてはならない」ということです。

なぜ一般的な手法は失敗しやすいのか？

1. Dual Write（二重書き込み）

これは最も一般的な間違いです。コード内で、DBへの書き込みとMessage Queueへの書き込みという2つの独立したアクションを実行します。

async function createOrder(orderData) {
  const order = await db.orders.create(orderData); // (1) 成功
  await rabbitMQ.publish('order_created', order);  // (2) ここでネットワークエラーが発生すると、データが不整合になります！
}

問題は、ステップ(1)と(2)が単一のトランザクション内にないことです。ステップ(2)が失敗しても、ステップ(1)は自動的にロールバックされず、データの不整合を招きます。

2. 分散トランザクション (2PC)

この方法は、DBとブローカーを一つの共通トランザクションにまとめようとします。しかし、2PCは極めて複雑で、システムのパフォーマンスを大幅に低下させます。現代のマイクロサービス環境において、2PCは保守性と拡張性の面で「悪夢」となることが多いです。

Transactional Outbox Pattern：ACID特性を活かした解決策

メッセージをすぐに送信しようとする代わりに、PostgreSQL自体のACID特性を活用します。核心となるアイデアは非常に実用的です：

  • データベース内に outbox という補助テーブルを作成します。
  • 注文を保存する際、同じトランザクション内で outbox テーブルにもレコードを挿入します。
  • 別のプロセス（Relay）が outbox テーブルをスキャンし、メッセージを送信して成功のマークを付けます。

注文の保存とoutboxへのメッセージ保存が同じトランザクションを使用するため、両方が成功するか、あるいは何も変更されないかのどちらかになります。これにより、データは常に絶対的な整合性を保つことができます。

PostgreSQLとNode.jsによる実践的な実装ガイド

ステップ1：Outboxテーブルの設計

様々な種類のイベントを保持できるように、柔軟なテーブル構造が必要です。

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    customer_name TEXT NOT NULL,
    total_amount DECIMAL NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE outbox (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type TEXT NOT NULL, -- 例: 'Order'
    aggregate_id TEXT NOT NULL,   -- 注文ID
    type TEXT NOT NULL,            -- イベントタイプ: 'OrderCreated'
    payload JSONB NOT NULL,        -- 実際のデータ
    status TEXT DEFAULT 'PENDING', -- PENDING（保留中）, PROCESSED（処理済み）, FAILED（失敗）
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

ステップ2：トランザクションに準拠したビジネスロジックの記述

pg ライブラリを使用する場合、原子性（Atomicity）を保証するために共通の client を使用する必要があります。一つのコマンドが失敗すれば、すべての変更が破棄されます。

async function createOrder(customerName, amount) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    // 1. メインテーブルに注文を保存
    const orderRes = await client.query(
      'INSERT INTO orders (customer_name, total_amount) VALUES ($1, $2) RETURNING id',
      [customerName, amount]
    );
    const orderId = orderRes.rows[0].id;

    // 2. outboxテーブルにメッセージを保存（同一トランザクション）
    const outboxPayload = { orderId, customerName, amount };
    await client.query(
      'INSERT INTO outbox (aggregate_type, aggregate_id, type, payload) VALUES ($1, $2, $3, $4)',
      ['Order', orderId.toString(), 'OrderCreated', JSON.stringify(outboxPayload)]
    );

    await client.query('COMMIT');
    console.log('トランザクション成功！');
  } catch (error) {
    await client.query('ROLLBACK');
    throw error;
  } finally {
    client.release();
  }
}

ステップ3：Message Relay（転送ワーカー）の構築

メッセージをDBからブローカーに渡すには、ポーリング（Polling）が中小規模のシステムにとって最もシンプルで効果的なアプローチです。

async function relayMessages() {
  const client = await pool.connect();
  try {
    // 未処理のメッセージを10件取得し、他のワーカーによる重複読み込みを防ぐために行をロックする
    const res = await client.query(
      "SELECT * FROM outbox WHERE status = 'PENDING' FOR UPDATE SKIP LOCKED LIMIT 10"
    );

    for (const row of res.rows) {
      try {
        await messageBroker.publish(row.type, row.payload);
        await client.query(
          "UPDATE outbox SET status = 'PROCESSED' WHERE id = $1",
          [row.id]
        );
      } catch (err) {
        console.error(`メッセージ送信エラー ${row.id}:`, err);
      }
    }
  } finally {
    client.release();
  }
}
setInterval(relayMessages, 5000);

ここで FOR UPDATE SKIP LOCKED という技術が鍵となります。これにより、複数のワーカーインスタンスを同時に実行しても、同じメッセージを重複して送信する心配がなくなります。

実運用における最適化のポイント

このパターンを数百万件のトランザクションが発生するプロジェクトに適用する場合、以下の3点に注意してください。

  • DB容量の管理: outboxテーブルは急速に肥大化します。1日100万件の注文があれば、1ヶ月で3,000万行に達し、クエリが遅くなります。3〜7日後に PROCESSED のレコードを削除（クリーンアップ）するスケジュールを設定しましょう。
  • 重複処理の対応 (Idempotency): このパターンは、メッセージが「少なくとも1回は確実に送信される」ことを保証します。しかし、送信直後かつDB更新前にワーカーがクラッシュした場合、メッセージは再送されます。受信側（コンシューマー）での重複チェック（冪等性の確保）は必須です。
  • CDCの検討: システムが毎秒1,000トランザクションを超えるような場合は、継続的なポーリングがDBの負荷になります。その際は、PostgreSQLのWALファイルを読み取る Debezium などの使用を検討してください。これにより、直接クエリを投げずにデータを転送できます。

終わりに

Transactional Outboxは単なるコーディング手法ではなく、堅牢なシステム設計の考え方です。ネットワーク接続の運に頼るのではなく、データベースの一貫性を信頼しましょう。皆さんの実装が成功し、高負荷時でもぐっすり眠れるようになることを願っています！

Share:
Tags: