Sagaパターン完全ガイド｜Orchestration vs Choreography と補償トランザクション（Node.js）

Saga 分散トランザクション マイクロサービス Node.js 設計 イベント駆動 補償トランザクション
結論
  • マイクロサービスでは 2PCを避け、Sagaで最終的整合性を設計する。
  • 失敗時は 補償トランザクション（Compensating Transaction） で逆順に巻き戻す。
  • 複雑な分岐・タイムアウト制御は Orchestration（中央オーケストレーター）、疎結合なイベント連鎖は Choreography が向く。
  • いずれも各ステップは べき等性（Idempotency） を前提にし、Choreographyでは WebHook・イベント受信側の重複排除 もセットで設計する。

分散トランザクションの課題：なぜ2PCでは足りないか

モノリス時代は、1つのデータベース上で BEGIN → 複数テーブル更新 → COMMIT というACIDトランザクションで整合性を担保できました。マイクロサービス構成では、注文・在庫・決済・配送がそれぞれ独立したDBとサービスを持ちます。ユーザーから見る「1回の購入」は、裏側では複数サービスへのHTTP呼び出しやメッセージ配信に分解されます。

二相コミット（2PC） は、コーディネータが全参加者に「準備完了（Prepare）」を問い合わせ、全員がOKなら一斉にコミットする方式です。教科書的には美しいですが、本番の分散システムでは次の問題が顕在化します。

  • 可用性: 参加者の1台が応答しないと、他サービスのトランザクションも長時間ブロックされる
  • ロック時間: Prepare〜Commitの間、在庫行や決済レコードがロックされ、スループットが落ちる
  • 運用複雑性: ネットワーク分断時に「コーディネータだけコミットしたが参加者が不明」という不確定状態が残る
  • スケール: サービス追加のたびに2PC参加者が増え、障害ドメインが連鎖する

Sagaパターンは、各ステップを独立したローカルトランザクションとして順次実行し、途中で失敗したらすでに成功したステップを補償（Compensate）で打ち消すアプローチです。即座のグローバル一貫性（Strong Consistency）の代わりに最終的整合性（Eventual Consistency）を受け入れ、可用性と独立デプロイのしやすさを得ます。

Sagaパターンの2つのスタイル

Sagaには大きく Orchestration（オーケストレーション） と Choreography（コレオグラフィ） の2スタイルがあります。どちらも「順次ステップ + 失敗時の補償」という本質は同じですが、誰がフローを制御するかが異なります。

観点 Orchestration vs Choreography
制御主体 Orchestration: 中央のオーケストレーターが次ステップを明示的に呼ぶ。Choreography: 各サービスがイベントを購読し、次のイベントを自律的に発行する
フローの可視性 Orchestration: 1ファイル（または状態機械）に全体フローが集約され、デバッグしやすい。Choreography: フローがイベントの連鎖に分散し、全体像の把握が難しい
結合度 Orchestration: オーケストレーターが各サービスのAPIを知る（やや密結合）。Choreography: サービス間はイベントスキーマだけを共有（疎結合）
分岐・条件 Orchestration: if/else、タイムアウト、リトライを1箇所で記述可能。Choreography: 条件分岐が各ハンドラに分散し、複雑化しやすい
単一障害点 Orchestration: オーケストレーター自体がSPOFになりうる（冗長化が必要）。Choreography: 中央制御がない分、個々のサービス障害に局所化しやすい
向いている例 Orchestration: EC購入（在庫→決済→配送→通知）、KYCフロー。Choreography: 在庫更新→検索インデックス更新→分析パイプライン起動

実務では「購入SagaはOrchestration、副次効果の連鎖はChoreography」というハイブリッドも多いです。本記事では両方のNode.js実装例を示し、選定基準を具体化します。

補償トランザクション（Compensating Transaction）の設計原則

Sagaのステップ3で決済が失敗した場合、ステップ1（在庫引当）とステップ2（クーポン消費）の副作用を打ち消す必要があります。この逆操作が補償トランザクションです。通常のSQL ROLLBACK とは根本的に異なります。

通常のロールバックSagaの補償
同一DBトランザクション内で自動的に巻き戻る各サービスのDB/APIを個別に逆操作する
外部API（Stripe返金、配送キャンセル）には届かない外部副作用も明示的に打ち消す必要がある
原子性がDBが保証補償も失敗しうるため、再試行と人手介入が必要

補償設計のチェックリストは次のとおりです。

  1. 意味的な逆操作: 「在庫+1」「返金」「クーポン復元」——ビジネス上正しい逆効果か
  2. べき等性: 同じ補償が2回走っても結果が壊れない（Idempotency-Key設計 を各ステップに適用）
  3. 順序: 成功ステップの逆順で補償（後から成功したものから先に打ち消す）
  4. 部分成功の許容: 補償中に一部だけ成功した状態を compensating_partial 等で記録
  5. 可観測性: Saga IDをログ・トレース・DBに統一して載せる
補償は「削除」ではなく「打ち消し」

物理削除（DELETE FROM orders WHERE id = ?）は監査ログや外部連携と相性が悪いです。注文ステータスを cancelled に更新し、在庫は release APIで戻す、決済は Refund APIを呼ぶ——状態遷移と逆APIで設計してください。Stripe返金も Idempotency-Key を Saga ID + step名で固定し、再試行で二重返金しないようにします。

典型フロー：EC購入Saga

次の4ステップを例に、OrchestrationとChoreographyの両方を実装します。

1

Step 1: 在庫サービスで在庫を引当（reserveInventory）——失敗なら補償不要

2

Step 2: 決済サービスでPaymentIntentを確定（capturePayment）——失敗なら Step 1 を releaseInventory で補償

3

Step 3: 注文サービスで注文レコードを確定（confirmOrder）——失敗なら Step 2 を refundPayment、Step 1 を releaseInventory

4

Step 4: 通知サービスで確認メール送信（sendConfirmation）——失敗なら Step 3〜1 を逆順に補償

5

いずれかの補償が失敗: Saga状態を failed にし、DLQ + 運用アラートへ

ドメインモデルとSaga状態の定義

まず、TypeScript風の型とSaga状態を定義します。Node.js本番では saga_instances テーブルに状態を永続化し、プロセス再起動後も再開できるようにします。

// saga-types.js
'use strict';

/**
 * @typedef {'pending'|'running'|'completed'|'compensating'|'failed'} SagaStatus
 */

/**
 * @typedef {Object} SagaContext
 * @property {string} sagaId
 * @property {string} orderId
 * @property {string} userId
 * @property {string} productId
 * @property {number} quantity
 * @property {number} amountJpy
 * @property {Record<string, unknown>} stepResults
 */

/**
 * @typedef {Object} SagaStep
 * @property {string} name
 * @property {(ctx: SagaContext) => Promise<void>} execute
 * @property {(ctx: SagaContext) => Promise<void>} compensate
 */

/** @type {SagaStatus[]} */
const TERMINAL_STATUSES = ['completed', 'failed'];

function createInitialContext(payload) {
  return {
    sagaId: payload.sagaId,
    orderId: payload.orderId,
    userId: payload.userId,
    productId: payload.productId,
    quantity: payload.quantity,
    amountJpy: payload.amountJpy,
    stepResults: {},
  };
}

module.exports = {
  TERMINAL_STATUSES,
  createInitialContext,
};

Orchestration：中央オーケストレーターの実装

Orchestrationでは SagaOrchestrator クラスがステップを順次実行し、失敗時に逆順で補償を呼び出します。各ステップの実行前後でDB上のSaga状態を更新し、クラッシュ後の再開点を記録します。

// saga-orchestrator.js
'use strict';

const { randomUUID } = require('crypto');
const { createInitialContext, TERMINAL_STATUSES } = require('./saga-types');
const { SagaRepository } = require('./saga-repository');
const { inventoryStep, paymentStep, orderStep, notificationStep } = require('./saga-steps');

class SagaOrchestrator {
  /**
   * @param {import('./saga-repository').SagaRepository} repository
   * @param {import('./saga-types').SagaStep[]} steps
   */
  constructor(repository, steps) {
    this.repository = repository;
    this.steps = steps;
  }

  /**
   * @param {object} payload
   * @returns {Promise<{ sagaId: string, status: string }>}
   */
  async start(payload) {
    const sagaId = payload.sagaId || randomUUID();
    const ctx = createInitialContext({ ...payload, sagaId });

    await this.repository.create({
      sagaId,
      status: 'pending',
      currentStepIndex: 0,
      payload: ctx,
      completedSteps: [],
    });

    return this.run(sagaId);
  }

  /**
   * @param {string} sagaId
   */
  async run(sagaId) {
    const record = await this.repository.findById(sagaId);
    if (!record) {
      throw new Error(`Saga not found: ${sagaId}`);
    }
    if (TERMINAL_STATUSES.includes(record.status)) {
      return { sagaId, status: record.status };
    }

    const ctx = record.payload;
    let startIndex = record.currentStepIndex;

    await this.repository.updateStatus(sagaId, 'running', { currentStepIndex: startIndex });

    const completedSteps = [...record.completedSteps];

    for (let i = startIndex; i < this.steps.length; i += 1) {
      const step = this.steps[i];
      try {
        await step.execute(ctx);
        completedSteps.push(step.name);
        await this.repository.updateStatus(sagaId, 'running', {
          currentStepIndex: i + 1,
          completedSteps,
          payload: ctx,
        });
      } catch (error) {
        await this.repository.updateStatus(sagaId, 'compensating', {
          failedStep: step.name,
          errorMessage: error.message,
        });
        await this.compensate(sagaId, ctx, completedSteps);
        return { sagaId, status: 'failed' };
      }
    }

    await this.repository.updateStatus(sagaId, 'completed', { payload: ctx });
    return { sagaId, status: 'completed' };
  }

  /**
   * @param {string} sagaId
   * @param {import('./saga-types').SagaContext} ctx
   * @param {string[]} completedSteps
   */
  async compensate(sagaId, ctx, completedSteps) {
    const stepsByName = new Map(this.steps.map((s) => [s.name, s]));
    const reversed = [...completedSteps].reverse();

    for (const stepName of reversed) {
      const step = stepsByName.get(stepName);
      if (!step) {
        continue;
      }
      try {
        await step.compensate(ctx);
        await this.repository.appendCompensationLog(sagaId, stepName, 'success');
      } catch (compError) {
        await this.repository.appendCompensationLog(sagaId, stepName, 'failed', compError.message);
        await this.repository.updateStatus(sagaId, 'failed', {
          compensationFailedStep: stepName,
          compensationError: compError.message,
        });
        throw compError;
      }
    }

    await this.repository.updateStatus(sagaId, 'failed', { payload: ctx });
  }
}

const defaultOrchestrator = new SagaOrchestrator(
  new SagaRepository(),
  [inventoryStep, paymentStep, orderStep, notificationStep]
);

module.exports = { SagaOrchestrator, defaultOrchestrator };

オーケストレーターの利点は、フロー全体が1箇所に集約されることです。新ステップ追加、条件分岐、タイムアウトは saga-steps.js とオーケストレーターだけを変更すれば済み、各ドメインサービスのコードは「1ステップの execute/compensate」に専念できます。

各ステップの実装：べき等な execute / compensate

各ステップは HTTP クライアント経由でマイクロサービスを呼び出します。リクエストヘッダーに Idempotency-Key: ${sagaId}:${stepName} を付与し、べき等性設計 と同じ原則で二重実行を防ぎます。

// saga-steps.js
'use strict';

const { requestJson } = require('./http-client');

const INVENTORY_BASE = process.env.INVENTORY_SERVICE_URL || 'http://localhost:4001';
const PAYMENT_BASE = process.env.PAYMENT_SERVICE_URL || 'http://localhost:4002';
const ORDER_BASE = process.env.ORDER_SERVICE_URL || 'http://localhost:4003';
const NOTIFY_BASE = process.env.NOTIFY_SERVICE_URL || 'http://localhost:4004';

function idempotencyKey(sagaId, stepName) {
  return `${sagaId}:${stepName}`;
}

/** @type {import('./saga-types').SagaStep} */
const inventoryStep = {
  name: 'reserveInventory',
  async execute(ctx) {
    const body = {
      productId: ctx.productId,
      quantity: ctx.quantity,
      orderId: ctx.orderId,
    };
    const result = await requestJson(`${INVENTORY_BASE}/v1/inventory/reserve`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Idempotency-Key': idempotencyKey(ctx.sagaId, 'reserveInventory'),
      },
      body: JSON.stringify(body),
    });
    ctx.stepResults.reservationId = result.reservationId;
  },
  async compensate(ctx) {
    if (!ctx.stepResults.reservationId) {
      return;
    }
    await requestJson(`${INVENTORY_BASE}/v1/inventory/release`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Idempotency-Key': idempotencyKey(ctx.sagaId, 'releaseInventory'),
      },
      body: JSON.stringify({
        reservationId: ctx.stepResults.reservationId,
        orderId: ctx.orderId,
      }),
    });
  },
};

/** @type {import('./saga-types').SagaStep} */
const paymentStep = {
  name: 'capturePayment',
  async execute(ctx) {
    const result = await requestJson(`${PAYMENT_BASE}/v1/payments/capture`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Idempotency-Key': idempotencyKey(ctx.sagaId, 'capturePayment'),
      },
      body: JSON.stringify({
        userId: ctx.userId,
        orderId: ctx.orderId,
        amountJpy: ctx.amountJpy,
      }),
    });
    ctx.stepResults.paymentIntentId = result.paymentIntentId;
    ctx.stepResults.chargeId = result.chargeId;
  },
  async compensate(ctx) {
    if (!ctx.stepResults.paymentIntentId) {
      return;
    }
    await requestJson(`${PAYMENT_BASE}/v1/payments/refund`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Idempotency-Key': idempotencyKey(ctx.sagaId, 'refundPayment'),
      },
      body: JSON.stringify({
        paymentIntentId: ctx.stepResults.paymentIntentId,
        chargeId: ctx.stepResults.chargeId,
        orderId: ctx.orderId,
      }),
    });
  },
};

/** @type {import('./saga-types').SagaStep} */
const orderStep = {
  name: 'confirmOrder',
  async execute(ctx) {
    const result = await requestJson(`${ORDER_BASE}/v1/orders/confirm`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Idempotency-Key': idempotencyKey(ctx.sagaId, 'confirmOrder'),
      },
      body: JSON.stringify({
        orderId: ctx.orderId,
        userId: ctx.userId,
        reservationId: ctx.stepResults.reservationId,
        paymentIntentId: ctx.stepResults.paymentIntentId,
      }),
    });
    ctx.stepResults.confirmedAt = result.confirmedAt;
  },
  async compensate(ctx) {
    await requestJson(`${ORDER_BASE}/v1/orders/cancel`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Idempotency-Key': idempotencyKey(ctx.sagaId, 'cancelOrder'),
      },
      body: JSON.stringify({
        orderId: ctx.orderId,
        reason: 'saga_compensation',
      }),
    });
  },
};

/** @type {import('./saga-types').SagaStep} */
const notificationStep = {
  name: 'sendConfirmation',
  async execute(ctx) {
    await requestJson(`${NOTIFY_BASE}/v1/notifications/order-confirmed`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Idempotency-Key': idempotencyKey(ctx.sagaId, 'sendConfirmation'),
      },
      body: JSON.stringify({
        userId: ctx.userId,
        orderId: ctx.orderId,
        emailTemplate: 'order_confirmed_v1',
      }),
    });
    ctx.stepResults.notificationSent = true;
  },
  async compensate(ctx) {
    if (!ctx.stepResults.notificationSent) {
      return;
    }
    await requestJson(`${NOTIFY_BASE}/v1/notifications/order-cancelled`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Idempotency-Key': idempotencyKey(ctx.sagaId, 'sendCancellationNotice'),
      },
      body: JSON.stringify({
        userId: ctx.userId,
        orderId: ctx.orderId,
        emailTemplate: 'order_cancelled_v1',
      }),
    });
  },
};

module.exports = {
  inventoryStep,
  paymentStep,
  orderStep,
  notificationStep,
};

Saga状態のPostgreSQL永続化

オーケストレータープロセスが落ちても Saga を再開できるよう、状態をDBに保存します。saga_id を主キーにし、status と completed_steps を JSONB で保持する構成が一般的です。

// saga-repository.js
'use strict';

const { Pool } = require('pg');

const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: 10,
});

class SagaRepository {
  async create(record) {
    const query = `
      INSERT INTO saga_instances (
        saga_id, status, current_step_index, payload, completed_steps, created_at, updated_at
      ) VALUES ($1, $2, $3, $4::jsonb, $5::jsonb, NOW(), NOW())
    `;
    await pool.query(query, [
      record.sagaId,
      record.status,
      record.currentStepIndex,
      JSON.stringify(record.payload),
      JSON.stringify(record.completedSteps || []),
    ]);
  }

  async findById(sagaId) {
    const result = await pool.query(
      `SELECT saga_id, status, current_step_index, payload, completed_steps
       FROM saga_instances WHERE saga_id = $1`,
      [sagaId]
    );
    if (result.rowCount === 0) {
      return null;
    }
    const row = result.rows[0];
    return {
      sagaId: row.saga_id,
      status: row.status,
      currentStepIndex: row.current_step_index,
      payload: row.payload,
      completedSteps: row.completed_steps,
    };
  }

  async updateStatus(sagaId, status, extra = {}) {
    const fields = ['status = $2', 'updated_at = NOW()'];
    const values = [sagaId, status];
    let paramIndex = 3;

    if (extra.currentStepIndex !== undefined) {
      fields.push(`current_step_index = $${paramIndex}`);
      values.push(extra.currentStepIndex);
      paramIndex += 1;
    }
    if (extra.completedSteps !== undefined) {
      fields.push(`completed_steps = $${paramIndex}::jsonb`);
      values.push(JSON.stringify(extra.completedSteps));
      paramIndex += 1;
    }
    if (extra.payload !== undefined) {
      fields.push(`payload = $${paramIndex}::jsonb`);
      values.push(JSON.stringify(extra.payload));
      paramIndex += 1;
    }
    if (extra.failedStep !== undefined) {
      fields.push(`failed_step = $${paramIndex}`);
      values.push(extra.failedStep);
      paramIndex += 1;
    }
    if (extra.errorMessage !== undefined) {
      fields.push(`error_message = $${paramIndex}`);
      values.push(extra.errorMessage);
      paramIndex += 1;
    }

    const query = `UPDATE saga_instances SET ${fields.join(', ')} WHERE saga_id = $1`;
    await pool.query(query, values);
  }

  async appendCompensationLog(sagaId, stepName, result, errorMessage = null) {
    await pool.query(
      `INSERT INTO saga_compensation_logs (saga_id, step_name, result, error_message, created_at)
       VALUES ($1, $2, $3, $4, NOW())`,
      [sagaId, stepName, result, errorMessage]
    );
  }
}

module.exports = { SagaRepository };

対応するDDLは次のとおりです。

-- migrations/001_saga_instances.sql
CREATE TABLE saga_instances (
  saga_id UUID PRIMARY KEY,
  status VARCHAR(32) NOT NULL,
  current_step_index INT NOT NULL DEFAULT 0,
  payload JSONB NOT NULL,
  completed_steps JSONB NOT NULL DEFAULT '[]'::jsonb,
  failed_step VARCHAR(64),
  error_message TEXT,
  compensation_failed_step VARCHAR(64),
  compensation_error TEXT,
  created_at TIMESTAMPTZ NOT NULL,
  updated_at TIMESTAMPTZ NOT NULL
);

CREATE INDEX idx_saga_instances_status ON saga_instances (status);
CREATE INDEX idx_saga_instances_updated_at ON saga_instances (updated_at);

CREATE TABLE saga_compensation_logs (
  id BIGSERIAL PRIMARY KEY,
  saga_id UUID NOT NULL REFERENCES saga_instances(saga_id),
  step_name VARCHAR(64) NOT NULL,
  result VARCHAR(16) NOT NULL,
  error_message TEXT,
  created_at TIMESTAMPTZ NOT NULL
);

Express API：Saga開始エンドポイント

購入APIは Saga を非同期で起動し、クライアントには 202 Accepted と sagaId を返します。ポーリングまたは WebSocket / Server-Sent Events で完了を通知する構成が一般的です。

// saga-api.js
'use strict';

const express = require('express');
const { randomUUID } = require('crypto');
const { defaultOrchestrator } = require('./saga-orchestrator');
const { SagaRepository } = require('./saga-repository');

const app = express();
app.use(express.json());

const repository = new SagaRepository();

app.post('/v1/checkout', async (req, res) => {
  const idempotencyKey = req.headers['idempotency-key'];
  if (!idempotencyKey) {
    return res.status(400).json({ error: 'Idempotency-Key header is required' });
  }

  const existing = await repository.findByClientKey(idempotencyKey);
  if (existing) {
    return res.status(existing.httpStatus).json(existing.responseBody);
  }

  const { userId, productId, quantity, amountJpy } = req.body;
  if (!userId || !productId || !quantity || !amountJpy) {
    return res.status(400).json({ error: 'Missing required fields' });
  }

  const orderId = randomUUID();
  const sagaId = randomUUID();

  const acceptedBody = {
    sagaId,
    orderId,
    status: 'pending',
    pollUrl: `/v1/sagas/${sagaId}`,
  };

  await repository.saveClientKeyMapping(idempotencyKey, 202, acceptedBody);

  res.status(202).json(acceptedBody);

  setImmediate(async () => {
    try {
      await defaultOrchestrator.start({
        sagaId,
        orderId,
        userId,
        productId,
        quantity,
        amountJpy,
      });
    } catch (error) {
      console.error(JSON.stringify({
        level: 'error',
        msg: 'Saga execution failed',
        sagaId,
        error: error.message,
      }));
    }
  });
});

app.get('/v1/sagas/:sagaId', async (req, res) => {
  const record = await repository.findById(req.params.sagaId);
  if (!record) {
    return res.status(404).json({ error: 'Saga not found' });
  }
  return res.json({
    sagaId: record.sagaId,
    status: record.status,
    currentStepIndex: record.currentStepIndex,
    completedSteps: record.completedSteps,
  });
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
  console.log(`Saga API listening on port ${PORT}`);
});

Choreography：イベント駆動Sagaの実装

Choreographyでは中央オーケストレーターは存在せず、各サービスがドメインイベントを発行・購読して次のアクションを起こします。Node.jsでは RabbitMQ、Amazon SQS、Redis Streams、または社内イベントバスを使います。ここではシンプルに EventEmitter を模したインメモリバスで概念を示します（本番は永続キュー必須）。

// saga-choreography.js
'use strict';

const EventEmitter = require('events');
const { randomUUID } = require('crypto');

const bus = new EventEmitter();
bus.setMaxListeners(50);

const EVENTS = {
  ORDER_CREATED: 'OrderCreated',
  INVENTORY_RESERVED: 'InventoryReserved',
  INVENTORY_RESERVE_FAILED: 'InventoryReserveFailed',
  PAYMENT_CAPTURED: 'PaymentCaptured',
  PAYMENT_FAILED: 'PaymentFailed',
  ORDER_CONFIRMED: 'OrderConfirmed',
  ORDER_FAILED: 'OrderFailed',
};

function publish(eventName, payload) {
  const envelope = {
    eventId: randomUUID(),
    eventName,
    occurredAt: new Date().toISOString(),
    payload,
  };
  bus.emit(eventName, envelope);
  return envelope;
}

// --- 在庫サービス ---
bus.on(EVENTS.ORDER_CREATED, async (envelope) => {
  const { sagaId, orderId, productId, quantity } = envelope.payload;
  try {
    const reservationId = await reserveInventoryLocally(sagaId, orderId, productId, quantity);
    publish(EVENTS.INVENTORY_RESERVED, {
      sagaId,
      orderId,
      reservationId,
      productId,
      quantity,
    });
  } catch (error) {
    publish(EVENTS.INVENTORY_RESERVE_FAILED, {
      sagaId,
      orderId,
      reason: error.message,
    });
  }
});

// --- 決済サービス ---
bus.on(EVENTS.INVENTORY_RESERVED, async (envelope) => {
  const { sagaId, orderId, reservationId, amountJpy, userId } = envelope.payload;
  try {
    const paymentIntentId = await capturePaymentLocally(sagaId, orderId, userId, amountJpy);
    publish(EVENTS.PAYMENT_CAPTURED, {
      sagaId,
      orderId,
      reservationId,
      paymentIntentId,
      userId,
    });
  } catch (error) {
    publish(EVENTS.PAYMENT_FAILED, {
      sagaId,
      orderId,
      reservationId,
      reason: error.message,
    });
  }
});

// 決済失敗 → 在庫解放（補償）
bus.on(EVENTS.PAYMENT_FAILED, async (envelope) => {
  const { sagaId, orderId, reservationId } = envelope.payload;
  await releaseInventoryLocally(sagaId, orderId, reservationId);
  publish(EVENTS.ORDER_FAILED, { sagaId, orderId, reason: 'payment_failed' });
});

// --- 注文サービス ---
bus.on(EVENTS.PAYMENT_CAPTURED, async (envelope) => {
  const { sagaId, orderId, reservationId, paymentIntentId, userId } = envelope.payload;
  try {
    await confirmOrderLocally(sagaId, orderId, userId, reservationId, paymentIntentId);
    publish(EVENTS.ORDER_CONFIRMED, { sagaId, orderId, userId });
  } catch (error) {
    publish(EVENTS.ORDER_FAILED, {
      sagaId,
      orderId,
      reservationId,
      paymentIntentId,
      reason: error.message,
    });
  }
});

// 注文確定失敗 → 返金 + 在庫解放
bus.on(EVENTS.ORDER_FAILED, async (envelope) => {
  const { sagaId, orderId, reservationId, paymentIntentId, reason } = envelope.payload;
  if (paymentIntentId) {
    await refundPaymentLocally(sagaId, orderId, paymentIntentId);
  }
  if (reservationId) {
    await releaseInventoryLocally(sagaId, orderId, reservationId);
  }
  console.error(JSON.stringify({
    level: 'error',
    msg: 'Choreography saga failed',
    sagaId,
    orderId,
    reason,
  }));
});

async function reserveInventoryLocally(sagaId, orderId, productId, quantity) {
  return `res_${sagaId.slice(0, 8)}`;
}

async function releaseInventoryLocally(sagaId, orderId, reservationId) {
  return { released: true, reservationId, sagaId, orderId };
}

async function capturePaymentLocally(sagaId, orderId, userId, amountJpy) {
  return `pi_${sagaId.slice(0, 8)}`;
}

async function refundPaymentLocally(sagaId, orderId, paymentIntentId) {
  return { refunded: true, paymentIntentId, sagaId, orderId };
}

async function confirmOrderLocally(sagaId, orderId, userId, reservationId, paymentIntentId) {
  return { confirmed: true, orderId, sagaId };
}

function startCheckoutChoreography(input) {
  const sagaId = randomUUID();
  const orderId = randomUUID();
  publish(EVENTS.ORDER_CREATED, {
    sagaId,
    orderId,
    userId: input.userId,
    productId: input.productId,
    quantity: input.quantity,
    amountJpy: input.amountJpy,
  });
  return { sagaId, orderId };
}

module.exports = {
  EVENTS,
  publish,
  startCheckoutChoreography,
  bus,
};

Choreographyの落とし穴は、補償ロジックが複数のイベントハンドラに分散することです。PAYMENT_FAILED と ORDER_FAILED の両方で在庫解放を書くと、二重解放のリスクがあります。イベントハンドラ側も WebHook設計と同様 に eventId のべき等テーブルで重複処理を防ぎ、補償API自体も Saga ID ベースの Idempotency-Key を使います。

決済サービス側：補償APIのべき等実装

補償の核心は決済返金です。Stripe等の外部APIを呼ぶ場合、オーケストレーターからの再試行で二重返金しないよう、サービス側で Idempotency-Key を検証します。

// payment-service-refund.js
'use strict';

const express = require('express');
const Stripe = require('stripe');
const { Pool } = require('pg');

const stripe = new Stripe(process.env.STRIPE_SECRET_KEY);
const pool = new Pool({ connectionString: process.env.PAYMENT_DATABASE_URL });

const app = express();
app.use(express.json());

async function getIdempotentRecord(key) {
  const result = await pool.query(
    `SELECT response_status, response_body FROM idempotency_keys WHERE key = $1`,
    [key]
  );
  return result.rows[0] || null;
}

async function saveIdempotentRecord(key, status, body) {
  await pool.query(
    `INSERT INTO idempotency_keys (key, response_status, response_body, created_at)
     VALUES ($1, $2, $3::jsonb, NOW())
     ON CONFLICT (key) DO NOTHING`,
    [key, status, JSON.stringify(body)]
  );
}

app.post('/v1/payments/refund', async (req, res) => {
  const idempotencyKey = req.headers['idempotency-key'];
  if (!idempotencyKey) {
    return res.status(400).json({ error: 'Idempotency-Key required' });
  }

  const cached = await getIdempotentRecord(idempotencyKey);
  if (cached) {
    return res.status(cached.response_status).json(cached.response_body);
  }

  const { paymentIntentId, chargeId, orderId } = req.body;
  if (!paymentIntentId || !orderId) {
    return res.status(400).json({ error: 'paymentIntentId and orderId required' });
  }

  try {
    const refund = await stripe.refunds.create(
      {
        payment_intent: paymentIntentId,
        metadata: { orderId, reason: 'saga_compensation' },
      },
      { idempotencyKey }
    );

    const body = {
      refundId: refund.id,
      status: refund.status,
      paymentIntentId,
      orderId,
    };

    await saveIdempotentRecord(idempotencyKey, 200, body);
    return res.status(200).json(body);
  } catch (error) {
    const body = { error: error.message, paymentIntentId, orderId };
    await saveIdempotentRecord(idempotencyKey, 502, body);
    return res.status(502).json(body);
  }
});

app.listen(4002, () => {
  console.log('Payment service listening on 4002');
});

Orchestration vs Choreography：選定フローチャート

判断に迷ったときは、次の順序で検討します。

  1. ステップ数が5以上、または条件分岐がある → Orchestration
  2. 監査・コンプライアンスで「誰がいつ何を実行したか」の一元ログが必要 → Orchestration
  3. チームが独立してサービスをリリースし、フロー変更の調整コストを避けたい → Choreography
  4. 副次効果（検索インデックス、分析、キャッシュ無効化）だけが連鎖 → Choreography
  5. 両方の要件がある → コア購入フローは Orchestration、それ以外は Choreography
Temporal / Step Functions との関係

AWS Step Functions、Temporal、Netflix Conductor などのワークフローエンジンは、Orchestration Saga の実装基盤として広く使われます。自前オーケストレーターは学習と小規模PoCに適しますが、本番ではリトライ・タイマー・可視化UIが組み込まれたマネージド/フレームワークへの移行を検討してください。Node.js から Temporal を使う場合も、各 Activity がべき等であること、Compensation Activity が逆順で登録されること——本記事の原則はそのまま適用されます。

可観測性：Saga IDをトレースに載せる

Sagaは1リクエストが複数サービスを横断するため、Saga ID = 分散トレーシングの correlation ID として扱います。各HTTP呼び出しに X-Saga-Id ヘッダーを付与し、構造化ログの saga_id フィールドに同じ値を入れます。障害時は「どのステップで止まったか」「補償はどこまで成功したか」を1クエリで追えるようにします。

// saga-logging.js
'use strict';

function logSagaEvent(level, sagaId, stepName, message, extra = {}) {
  const entry = {
    level,
    timestamp: new Date().toISOString(),
    saga_id: sagaId,
    step_name: stepName,
    message,
    ...extra,
  };
  console.log(JSON.stringify(entry));
}

function sagaHttpHeaders(sagaId, stepName) {
  return {
    'X-Saga-Id': sagaId,
    'X-Saga-Step': stepName,
    'Idempotency-Key': `${sagaId}:${stepName}`,
  };
}

module.exports = { logSagaEvent, sagaHttpHeaders };

運用ダッシュボードでは status = 'compensating' または compensation_failed_step IS NOT NULL の Saga を stuck 一覧として表示し、15分以上更新がなければアラートを上げます。

よくあるアンチパターン

アンチパターン問題対策
補償なしでステップを追加失敗時に在庫・決済が不整合のまま残る新ステップ追加時は必ず compensate をペアで実装
物理削除による補償監査不能、外部連携が追えないステータス遷移・逆APIで打ち消す
Choreographyで補償が分散二重補償・補償漏れ補償は1イベント1責務、eventIdべき等
Saga状態の非永続化プロセス再起動でフロー消失PostgreSQL等に状態保存、再開API
同期HTTPで全ステップ実行タイムアウト・カスケード障害202 + 非同期実行、またはメッセージキュー

テスト戦略

Sagaのテストは「ハッピーパス」だけでは不十分です。最低限次のシナリオをカバーします。

  1. 全ステップ成功 → status: completed
  2. Step 2（決済）失敗 → Step 1 のみ補償、status: failed
  3. Step 4（通知）失敗 → Step 3, 2, 1 の逆順補償
  4. 補償中の再試行 → 同じ Idempotency-Key で副作用が増えない
  5. オーケストレーター再起動 → current_step_index から再開

テストでは各マイクロサービスを WireMock や MSW でスタブし、特定ステップだけ 500 を返すようにします。Choreography ではイベント順序の乱れ（PaymentCaptured が InventoryReserved より先に届く）を防ぐため、イベントに causationId / sagaId を載せ、ハンドラ側で前提イベントの完了を確認するガードを入れます。

関連記事

この記事は API・バックエンド テーマの一環です。あわせて読むと理解が深まる関連記事をまとめました。

トピック記事
APIキャッシュ設計ガイドAPIキャッシュ設計ガイド｜ETag・Cache-Control・304・CDNキャッシュキーの実務
Web APIのエラーハンドリング設計Web APIのエラーハンドリング設計｜RFC 9457・error_code・400/422/409の使い分け
API設計におけるページネーションの使い分け：オフセット型とカーソル型の比較API設計におけるページネーションの使い分け：オフセット型とカーソル型の比較
APIのレート制限（Rate Limiting）実装ガイドAPIのレート制限（Rate Limiting）実装ガイド｜アルゴリズム選定と429エラーの適切なハンドリング
APIレート制限と429対応APIレート制限と429対応｜X-RateLimit・Retry-Afterを使ったクライアント実装
ZodでAPI境界のランタイム検証を設計するZodでAPI境界のランタイム検証を設計する｜z.infer・Expressミドルウェア・zod-to-openapi（2026）
OpenAPI 3.1仕様のCIバリデーション完全ガイドOpenAPI 3.1仕様のCIバリデーション完全ガイド｜Spectral Lint・Breaking Change検出・GitHub Actions
GraphQL N+1 問題の解決GraphQL N+1 問題の解決｜DataLoader バッチング実装と Apollo Server 完全ガイド
Circuit Breaker + Retry パターン実装ガイドCircuit Breaker + Retry パターン実装ガイド｜Node.js opossum/cockatiel・指数バックオフ・Jitter
Event SourcingとCQRSの基礎Event SourcingとCQRSの基礎｜Node.js + PostgreSQL 実装ガイド
Redisキャッシュスタンピード防止ガイドRedisキャッシュスタンピード防止ガイド｜Singleflight・Mutex・SET NX・確率的早期失効
PostgreSQL接続プール枯渇の対処法PostgreSQL接続プール枯渇の対処法｜too many connections・PgBouncer・Node.js pg

まとめ

Sagaパターンは、マイクロサービス環境で2PCの代わりに最終的整合性を実現する定番パターンです。Orchestration はフロー制御・可観測性・複雑な分岐に強く、Choreography はサービス自律性と疎結合に優れます。どちらを選んでも、成功の鍵は 補償トランザクションの設計 と 各ステップのべき等性 です。

  • 各 execute / compensate に SagaId:StepName 形式の Idempotency-Key を付ける（べき等性設計）
  • Choreography では eventId で重複イベントを排除する（WebHook設計）
  • Saga 状態を DB に永続化し、補償失敗は DLQ + アラートで人手にエスカレーション
  • コア購入フローは Orchestration、副次連鎖は Choreography——ハイブリッドも有効

次のステップとして、Temporal や AWS Step Functions への移行、OpenTelemetry との Saga ID 連携、Chaos Engineering による補償パスの定期検証を検討すると、本番運用の信頼性がさらに高まります。