冪等性と整合性

冪等性と整合性

分散システムでは「1回しか実行されない」は幻想。「何度実行されても最終状態が正しい」ことを保証する設計を詳しく解説します。

冪等性の重要性

分散システムでは、ネットワークエラー、タイムアウト、再起動などにより、同じ処理が複数回実行される可能性があります。

問題のあるコード

// ❌ 問題のあるコード: 非冪等な処理
func createOrder(orderData OrderData) (*Order, error) {
    // 問題: 再実行時に注文が二重作成される
    order := &Order{
        UserID: orderData.UserID,
        Amount: orderData.Amount,
    }
    return orderRepo.Save(order)
}

なぜ問題か:

• 再実行時の二重作成: ネットワークエラーでクライアントが再送すると、注文が2つ作成される
• データの不整合: 同じ注文が複数存在し、在庫や決済に影響する

冪等性の担保

Idempotency Keyの使用

// ✅ 良い例: Idempotency Keyによる冪等性の担保
func createOrder(orderData OrderData, idempotencyKey string) (*Order, error) {
    // Idempotency Keyで既存の注文を確認
    existingOrder, err := orderRepo.FindByIdempotencyKey(idempotencyKey)
    if err == nil && existingOrder != nil {
        // 既に存在する場合は、既存の注文を返す
        return existingOrder, nil
    }

    // 新規作成
    order := &Order{
        UserID:         orderData.UserID,
        Amount:         orderData.Amount,
        IdempotencyKey: idempotencyKey,
    }
    return orderRepo.Save(order)
}

// リポジトリの実装
func (r *OrderRepository) FindByIdempotencyKey(key string) (*Order, error) {
    var order Order
    err := r.db.Where("idempotency_key = ?", key).First(&order).Error
    if err != nil {
        return nil, err
    }
    return &order, nil
}

// マイグレーションでIdempotency Keyを追加
// CREATE UNIQUE INDEX idx_orders_idempotency_key ON orders(idempotency_key);

なぜ重要か:

• 重複防止: 同じIdempotency Keyで再実行しても、同じ結果が返される
• データの整合性: 注文の重複作成を防止

トランザクション境界

DB取引中に外部APIを呼ばない（失敗時復旧不可）。

問題のあるコード

// ❌ 問題のあるコード: トランザクション内で外部APIを呼ぶ
func createOrder(orderData OrderData) (*Order, error) {
    tx := db.Begin()
    defer func() {
        if r := recover(); r != nil {
            tx.Rollback()
        }
    }()

    // 1. 注文を作成（DBトランザクション内）
    order := &Order{UserID: orderData.UserID, Amount: orderData.Amount}
    if err := tx.Create(order).Error; err != nil {
        tx.Rollback()
        return nil, err
    }

    // 2. トランザクション内で外部APIを呼ぶ（問題）
    resp, err := http.Post(
        "https://payment-api.example.com/charge",
        "application/json",
        bytes.NewBuffer(orderJSON),
    )
    if err != nil {
        tx.Rollback()
        return nil, err
    }

    // 3. 決済結果を保存
    order.PaymentStatus = "COMPLETED"
    if err := tx.Save(order).Error; err != nil {
        tx.Rollback()
        return nil, err
    }

    tx.Commit()
    return order, nil
}

なぜ問題か:

• ロールバック不可: 外部APIが成功した後にトランザクションが失敗した場合、外部APIのロールバックが困難
• データの不整合: 外部APIは成功しているが、DBには注文が存在しない状態になる可能性

Outboxパターンの実装

// ✅ 良い例: Outboxパターンによる解決
func createOrder(orderData OrderData, idempotencyKey string) (*Order, error) {
    tx := db.Begin()
    defer func() {
        if r := recover(); r != nil {
            tx.Rollback()
        }
    }()

    // 1. Idempotency Keyで既存の注文を確認
    var existingOrder Order
    err := tx.Where("idempotency_key = ?", idempotencyKey).First(&existingOrder).Error
    if err == nil {
        tx.Commit()
        return &existingOrder, nil
    }

    // 2. トランザクション内で注文を作成
    order := &Order{
        UserID:         orderData.UserID,
        Amount:         orderData.Amount,
        IdempotencyKey: idempotencyKey,
    }
    if err := tx.Create(order).Error; err != nil {
        tx.Rollback()
        return nil, err
    }

    // 3. Outboxテーブルに外部API呼び出しを記録（トランザクション内）
    payload, _ := json.Marshal(map[string]interface{}{
        "orderId": order.ID,
        "amount":  orderData.Amount,
    })

    outboxEvent := &OutboxEvent{
        EventType:      "PAYMENT_CHARGE",
        AggregateID:    fmt.Sprintf("%d", order.ID),
        Payload:        string(payload),
        IdempotencyKey: idempotencyKey,
        Status:         "PENDING",
    }
    if err := tx.Create(outboxEvent).Error; err != nil {
        tx.Rollback()
        return nil, err
    }

    // 4. トランザクションをコミット（外部APIは呼ばない）
    tx.Commit()
    return order, nil
}

// 別プロセスでOutboxを処理
func processOutbox() {
    var pendingEvents []OutboxEvent
    db.Where("status = ?", "PENDING").Limit(10).Find(&pendingEvents)

    for _, event := range pendingEvents {
        var payload map[string]interface{}
        json.Unmarshal([]byte(event.Payload), &payload)

        // 外部APIを呼ぶ（トランザクション外）
        client := &http.Client{Timeout: 3 * time.Second}
        req, _ := http.NewRequest("POST", "https://payment-api.example.com/charge", bytes.NewBuffer(orderJSON))
        req.Header.Set("Idempotency-Key", event.IdempotencyKey)

        resp, err := client.Do(req)
        if err == nil && resp.StatusCode == 200 {
            db.Model(&event).Update("status", "COMPLETED")
        } else {
            db.Model(&event).Updates(map[string]interface{}{
                "status":      "FAILED",
                "retry_count": event.RetryCount + 1,
            })
        }
    }
}

// 定期的にOutboxを処理
func init() {
    go func() {
        ticker := time.NewTicker(5 * time.Second)
        for range ticker.C {
            processOutbox()
        }
    }()
}

なぜ重要か:

• トランザクションの短縮: DBのロック時間が短縮される
• 外部障害の分離: 外部APIの障害がトランザクションに影響しない
• 再実行の容易さ: Outboxテーブルから再実行可能
• 冪等性の保証: Idempotency Keyにより重複実行を防止

再送安全なフロー

「結果整合性で良いデータ」と「厳密整合性が必要なデータ」を明示的に分類する。

整合性レベルの分類

// 厳密整合性が必要なデータ（ACIDトランザクション）
type Order struct {
    ID     int64
    Amount decimal.Decimal
    Status string // CREATED, PAID, CANCELLED
}

// 結果整合性で良いデータ（イベント駆動）
type OrderAnalytics struct {
    ID          int64
    OrderID     int64
    TotalAmount decimal.Decimal // 集計値（最終的に整合性が取れれば良い）
    LastUpdated time.Time
}

使い分け:

• 厳密整合性: 注文、決済、在庫など、ビジネス的に重要なデータ
• 結果整合性: 分析データ、ログ、通知など、最終的に整合性が取れれば良いデータ

まとめ

冪等性と整合性のポイント：

• 冪等性の担保: Idempotency Keyで再実行を安全化
• トランザクション境界: DB取引中に外部APIを呼ばない（Outboxパターンを使用）
• 再送安全なフロー: 厳密整合性と結果整合性を明示的に分類

これらの原則により、「何度実行されても最終状態が正しい」堅牢なシステムを構築できます。