Posted inDevelopment

Node.jsによるマイクロサービスでのSagaパターンの実装ガイド：分散トランザクションの管理

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

Quick start: Dựng Saga Orchestrator chạy ngay trong 5 phút

モノリスからマイクロサービスへの分割には、常にデータの分散という難題が伴います。例えば、ユーザーが注文を行い、Payment Serviceで500,000ドンの決済が完了したものの、Inventory Serviceで在庫切れが発覚したとします。このとき、単一データベースが提供するACID特性はもはや通用せず、開発者が自力で返金処理を行わなければなりません。

この解決メカニズムを理解するために、ピュアなNode.jsでOrchestration型のSagaフローを構築してみましょう。saga-demo.jsファイルを作成します：

// saga-demo.js
class OrderSagaOrchestrator {
  constructor() {
    this.steps = [];
  }

  addStep({ name, execute, compensate }) {
    this.steps.push({ name, execute, compensate });
    return this;
  }

  async run(context) {
    const executedSteps = [];
    console.log(`[START] 注文 #${context.orderId} のSaga処理を開始`);

    for (const step of this.steps) {
      try {
        console.log(`-> 実行中: ${step.name}`);
        await step.execute(context);
        executedSteps.push(step);
      } catch (error) {
        console.error(`<!> [${step.name}] でエラーが発生: ${error.message}`);
        await this.rollback(executedSteps, context);
        throw new Error(`Sagaが中断されました: ${error.message}`);
      }
    }
    console.log(`[DONE] 分散トランザクションが正常に完了しました！`);
  }

  async rollback(executedSteps, context) {
    console.log(`[ROLLBACK] 補償トランザクション（Compensating Transactions）を開始...`);
    // 以前に正常実行されたステップを逆順に並び替え
    const stepsToRevert = [...executedSteps].reverse();
    for (const step of stepsToRevert) {
      try {
        console.log(`<- ロールバック中: ${step.name}`);
        await step.compensate(context);
      } catch (compensateError) {
        // 補償処理が失敗した場合は重大エラーとしてログを記録
        console.error(`[CRITICAL] [${step.name}] のロールバックに失敗: ${compensateError.message}`);
      }
    }
  }
}

// APIコールを想定した独立したMicroservicesのモック
const PaymentService = {
  charge: async (ctx) => { console.log(`   [Payment] ${ctx.amount}$ の決済が完了しました。`); },
  refund: async (ctx) => { console.log(`   [Payment] ${ctx.amount}$ をユーザーのウォレットへ返金しました。`); }
};

const InventoryService = {
  reserve: async (ctx) => {
    if (ctx.item === 'out-of-stock-item') {
      throw new Error('商品SKU-9901は在庫切れです！');
    }
    console.log(`   [Inventory] アイテム ${ctx.item} の引当に成功しました。`);
  },
  release: async (ctx) => { console.log(`   [Inventory] アイテム ${ctx.item} の在庫を引き当て解除しました。`); }
};

// シナリオ：在庫切れ商品を購入する場合
async function main() {
  const saga = new OrderSagaOrchestrator();

  saga
    .addStep({
      name: '決済処理',
      execute: (ctx) => PaymentService.charge(ctx),
      compensate: (ctx) => PaymentService.refund(ctx)
    })
    .addStep({
      name: '在庫の引き当て',
      execute: (ctx) => InventoryService.reserve(ctx),
      compensate: (ctx) => InventoryService.release(ctx)
    });

  try {
    await saga.run({
      orderId: 'ORD-9921',
      amount: 150,
      item: 'out-of-stock-item'
    });
  } catch (err) {
    console.log(`結果: 注文をキャンセルし、安全に返金処理を行いました。`);
  }
}

main();

ターミナルでスクリプトを実行してみます：

node saga-demo.js

実行ログには、決済完了 → 在庫エラー発生 → 返金処理（Compensate）実行という一連の流れが明確に現れます。システムの状態は最初の整合性のある状態へと戻り、重量級な2相コミット（2PC）のロック機構を一切使うことなく処理が完了しています。

Sagaパターンの技術的本質

Database-per-serviceの壁

各サービスは独立したデータベースを管理します。例えばユーザーがMacBook Proを購入した際、データフローはOrderDB（PostgreSQL）、PaymentDB（MySQL）、InventoryDB（MongoDB）という3つの独立したデータベースにまたがってアクセスします。

従来は複数のデータベースにまたがってリソースをロックするために、2相コミット（2PC）が広く利用されていました。しかし分散クラウド環境において、2PCには致命的な弱点があります。ネットワーク遅延によってレスポンスタイムが20msから2000msに跳ね上がる可能性があり、テーブルが長時間ロックされることでシステム全体のコネクションプール枯渇を引き起こします。

Sagaの動作メカニズム

Sagaはこの課題を、一連の独立したLocal Transactionsにプロセスを分解することで解決します。各サービスは自身の内部データベースのみを操作し、直ちにコミットを完了させてから次のステップへシグナルを送信します。

Sagaの核心はCompensating Transaction（補償トランザクション）にあります。もしN番目のステップで障害が発生した場合、SQLのROLLBACKコマンドで過去へ巻き戻すことはできません。ステップ1からN-1はすでにコミットが完了しているためです。その代わりに、OrchestratorがN-1から1へと逆順にロールバック処理（例：注文のキャンセル、決済ゲートウェイへの返金リクエスト）を発行します。

Choreographyか、Orchestrationか？

  • Choreography（イベント駆動型）： 各サービスがメッセージブローカー（Kafka、RabbitMQなど）を介して自律的に連携します。サービスAが処理を終えるとOrderCreatedイベントを発行し、サービスBがそれを受信して決済を実行後にPaymentCompletedを発行します。中央集権的な制御コンポーネントが不要で、2〜3ステップ程度の短いフローでは非常に軽快に動作します。しかし、システムが8〜10個のマイクロサービスに拡大すると、イベントフローの追跡が極めて複雑になります。
  • Orchestration（中央調整型）： 専用のサービスが「指揮者」として機能します。各ワーカーへ直接コマンドを送信し、その応答を待ってから次のステップの実行または補償処理の発行を判断します。最大の利点は、デバッグの容易さ、ログの集約、そしてリアルタイムでの注文ステータス把握のしやすさにあります。

ネットワーク障害への対処、べき等性（Idempotency）、状態永続化

先ほどのサンプルスクリプトはローカル環境では問題なく動作しますが、実際のKubernetesクラスタにデプロイすると、一時的なネットワーク切断などの障害が不可避となります。

べき等なレシーバーの設計（二重引き落としの防止）

次のような状況を想像してください。OrchestratorがPayment Serviceを呼び出して決済を実行します。Payment Serviceのデータベース側では決済が成功したものの、5000ms後にネットワーク遅延でHTTP 200のレスポンスがタイムアウトしました。Orchestratorはリクエストが失敗したと判断して再試行（リトライ）を行います。このときIdempotencyの仕組みがなければ、ユーザーのアカウントから二重に引き落とされてしまいます。

// Payment ServiceでのIdempotency Key検証ミドルウェア
async function processPayment(req, res) {
  const { idempotencyKey, orderId, amount } = req.body;

  // 1. トランザクションが過去に処理済みか確認
  const existingTx = await db.paymentTransactions.findOne({ idempotencyKey });
  if (existingTx) {
    return res.status(200).json(existingTx.responsePayload);
  }

  try {
    // 2. 決済処理とIdempotency Keyの記録を同一の内部DBトランザクションで実行
    const result = await db.transaction(async (trx) => {
      const payment = await trx.payments.insert({ orderId, amount, status: 'SUCCESS' });
      await trx.idempotencyRecords.insert({
        idempotencyKey,
        responsePayload: payment
      });
      return payment;
    });
    return res.status(201).json(result);
  } catch (err) {
    // 同一ミリ秒に重複リクエストが届いた場合のユニーク制約エラーをキャッチ
    if (err.code === 'ER_DUP_ENTRY' || err.code === '23505') {
      const duplicateTx = await db.paymentTransactions.findOne({ idempotencyKey });
      return res.status(200).json(duplicateTx.responsePayload);
    }
    return res.status(500).json({ error: 'トランザクション処理エラー' });
  }
}

Sagaの状態をデータベースへ永続化

Orchestratorのサーバーがクラッシュしたり、KubernetesによってPodが途中で強制終了されるリスクは常に存在します。新しく起動したPodが処理を続行すべきかロールバックすべきかを正確に判断できるよう、各ステップの状態をデータベースに記録する必要があります：

CREATE TABLE saga_instances (
  id VARCHAR(64) PRIMARY KEY,
  saga_type VARCHAR(50) NOT NULL,
  current_step VARCHAR(50) NOT NULL,
  status VARCHAR(20) NOT NULL, -- PENDING, COMPENSATING, COMPLETED, FAILED
  payload JSONB NOT NULL,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Orchestratorは次のサービスへリクエストを送信する前に、レコードのステップを更新します。バックグラウンドワーカーが60秒以上PENDING状態のまま停滞しているレコードを検知し、リトライの実行または補償トランザクションの開始を判断します。

本番環境におけるSaga運用の実践ノウハウ

Sagaを本番環境で運用するには、厳格なロギング規律と境界エラーの処理が不可欠です。主な重要ポイントは以下の通りです：

  • Payload形式の標準化： HTTPリクエストやKafkaトピック間でやり取りされるデータは肥大化しやすく、デバッグ時の可読性が低下しがちです。余計な拡張機能をインストールせずにJSON構造の確認、スキーマ検証、正規表現テストを迅速に行いたい場合は、toolcraft.appのオンラインフォーマッター（例：toolcraft.app/ja/tools/developer/json-formatter）などのツールを活用するのが便利です。
  • Trace Contextによる分散トレーシング： API Gatewayの段階で、一意の識別子（UUID）をx-correlation-idヘッダーやW3C標準のtraceparentとして付与しておきましょう。補償トランザクションで問題が発生した場合でも、このIDをGrafana LokiやDatadogに入力するだけで、実行された全ステップの足跡を即座に追跡できます。
  • 補償関数（Compensate）は絶対に失敗させてはならない： 返金や在庫の戻し処理のロジックには、例外をスローするようなビジネスバリデーションを絶対に組み込んではいけません。外部システム（銀行の決済ゲートウェイなど）がメンテナンス中の場合は、定期的なリトライ機構を備えたDead Letter Queue（DLQ）にメッセージを退避させ、同時にSlack等へ即時アラートを送信してオンコール担当者が対応できるようにします。
  • 専用フレームワークの活用： ステートマシンを独自にゼロから実装するのは、デモや小規模なユースケースにのみ適しています。システムが大規模化すると、タイムアウトやリトライ機構の独自メンテナンスには膨大なリソースを要します。RedisベースのBullMQや、成熟したワークフロー基盤であるTemporal.io、Cadenceなどの採用を検討することをお勧めします。
Share:
Tags: