冪等性と整合性

冪等性と整合性

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

冪等性の重要性

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

問題のあるコード

# ❌ 問題のあるコード: 非冪等な処理
def create_order(order_data):
    # 問題: 再実行時に注文が二重作成される
    return Order.objects.create(**order_data)

なぜ問題か:

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

冪等性の担保

Idempotency Keyの使用

# ✅ 良い例: Idempotency Keyによる冪等性の担保
def create_order(order_data, idempotency_key):
    # Idempotency Keyで既存の注文を確認
    existing_order = Order.objects.filter(
        idempotency_key=idempotency_key
    ).first()

    if existing_order:
        # 既に存在する場合は、既存の注文を返す
        return existing_order

    # 新規作成
    return Order.objects.create(
        **order_data,
        idempotency_key=idempotency_key
    )

# マイグレーションでIdempotency Keyを追加
# python manage.py makemigrations
# python manage.py migrate

なぜ重要か:

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

トランザクション境界

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

問題のあるコード

# ❌ 問題のあるコード: トランザクション内で外部APIを呼ぶ
from django.db import transaction

@transaction.atomic
def create_order(order_data):
    # 1. 注文を作成（DBトランザクション内）
    order = Order.objects.create(**order_data)

    # 2. トランザクション内で外部APIを呼ぶ（問題）
    response = requests.post(
        'https://payment-api.example.com/charge',
        json={'order_id': order.id, 'amount': order_data['amount']},
    )

    if response.status_code != 200:
        raise PaymentError("Payment failed")

    # 3. 決済結果を保存
    order.payment_status = "COMPLETED"
    order.save()

    return order

なぜ問題か:

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

Outboxパターンの実装

# ✅ 良い例: Outboxパターンによる解決
from django.db import transaction

def create_order(order_data, idempotency_key):
    with transaction.atomic():
        # 1. Idempotency Keyで既存の注文を確認
        existing_order = Order.objects.filter(
            idempotency_key=idempotency_key
        ).first()

        if existing_order:
            return existing_order

        # 2. トランザクション内で注文を作成
        order = Order.objects.create(
            **order_data,
            idempotency_key=idempotency_key
        )

        # 3. Outboxテーブルに外部API呼び出しを記録（トランザクション内）
        import json
        OutboxEvent.objects.create(
            event_type='PAYMENT_CHARGE',
            aggregate_id=str(order.id),
            payload=json.dumps({
                'order_id': order.id,
                'amount': order_data['amount']
            }),
            idempotency_key=idempotency_key,
            status='PENDING'
        )

        # 4. トランザクションをコミット（外部APIは呼ばない）
        return order

# 別プロセスでOutboxを処理
from celery import shared_task

@shared_task
def process_outbox():
    pending_events = OutboxEvent.objects.filter(
        status='PENDING'
    )[:10]

    for event in pending_events:
        try:
            # 外部APIを呼ぶ（トランザクション外）
            import json
            payload = json.loads(event.payload)

            response = requests.post(
                'https://payment-api.example.com/charge',
                headers={'Idempotency-Key': event.idempotency_key},
                json=payload,
                timeout=3
            )

            if response.status_code == 200:
                event.status = 'COMPLETED'
                event.save()
            else:
                event.status = 'FAILED'
                event.retry_count += 1
                event.save()
        except Exception as e:
            event.status = 'FAILED'
            event.retry_count += 1
            event.save()

# 定期的にOutboxを処理
# celery beatで5秒ごとに実行

なぜ重要か:

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

再送安全なフロー

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

整合性レベルの分類

# 厳密整合性が必要なデータ（ACIDトランザクション）
class Order(models.Model):
    id = models.AutoField(primary_key=True)
    amount = models.DecimalField(max_digits=10, decimal_places=2)
    status = models.CharField(max_length=20)  # CREATED, PAID, CANCELLED

# 結果整合性で良いデータ（イベント駆動）
class OrderAnalytics(models.Model):
    id = models.AutoField(primary_key=True)
    order_id = models.IntegerField()
    total_amount = models.DecimalField(max_digits=10, decimal_places=2)  # 集計値（最終的に整合性が取れれば良い）
    last_updated = models.DateTimeField(auto_now=True)

使い分け:

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

まとめ

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

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

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