A+B 複合 — 再代入防止 × 副作用排除 × 関数の影響範囲（在庫引当処理 Bad→Good 第4章）× inventory-reservation-service Memorystore Redis 分散ロックによる二重引当防止（fencing token × 冪等性キー × 指数バックオフ × Standard tier HA）

📋 コピー# tests/test_inventory_reservation.py
import pytest
from inventory_reservation import OrderItem, reserve_items, calculate_total_fee


class TestReserveItems:
    def test_does_not_mutate_input_stock(self):
        stock = {"SKU-A": 5}
        items = [OrderItem(sku="SKU-A", qty=2)]
        reserve_items(items, stock)
        assert stock == {"SKU-A": 5}  # 修正2: 元のdictは無変更

    def test_shortage_sku_is_reported(self):
        stock = {"SKU-A": 1}
        items = [OrderItem(sku="SKU-A", qty=10)]
        result = reserve_items(items, stock)
        assert result.shortage_skus == ("SKU-A",)

    def test_reserved_count_sums_only_successful_items(self):
        stock = {"SKU-A": 5, "SKU-B": 20}
        items = [OrderItem(sku="SKU-A", qty=10), OrderItem(sku="SKU-B", qty=3)]
        result = reserve_items(items, stock)
        assert result.reserved_count == 3


class TestCalculateTotalFee:
    def test_priority_fee_is_not_overwritten(self):
        items = [OrderItem(sku="X", qty=1, backorder_fee=100)]
        assert calculate_total_fee(items, priority_fee=500) == 600

📋 コピー"""inventory_lock.py — fencing token(修正2) × 冪等性チェック(修正3) × 指数バックオフ(修正4)"""
from __future__ import annotations

import random
import time
from dataclasses import dataclass
from typing import Final
from uuid import uuid4

import redis
from opentelemetry import trace  # 修正5: トレーシング

tracer = trace.get_tracer(__name__)

_LOCK_TTL_MS: Final[int] = 5_000
_MAX_RETRY_ATTEMPTS: Final[int] = 5
_BASE_BACKOFF_SECONDS: Final[float] = 0.05
_JITTER_SECONDS: Final[float] = 0.03
_IDEMPOTENCY_TTL_SECONDS: Final[int] = 86_400

# 修正2: 自分のトークンと一致する場合のみ削除する原子的なLuaスクリプト
_RELEASE_LOCK_SCRIPT = """
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end
"""


@dataclass(frozen=True, slots=True)
class LockHandle:
    """取得したロックのfencing token(不変)。"""
    sku: str
    token: str


def acquire_lock(client: redis.Redis, sku: str) -> LockHandle | None:
    """SKU単位のロックをfencing token付きで取得する(修正2)。

    Args:
        client: Memorystore Redisクライアント。
        sku: ロック対象のSKU。

    Returns:
        取得に成功した場合はLockHandle、取得できなかった場合はNone。
    """
    with tracer.start_as_current_span("inventory_lock.acquire") as span:
        for attempt in range(_MAX_RETRY_ATTEMPTS):
            token = str(uuid4())  # 修正2: ロックごとに一意なトークンを発行
            acquired = client.set(f"lock:{sku}", token, nx=True, px=_LOCK_TTL_MS)
            if acquired:
                span.set_attribute("lock.attempts", attempt + 1)
                return LockHandle(sku=sku, token=token)

            # 修正4: 指数バックオフ + ジッターでthundering herdを回避
            delay = _BASE_BACKOFF_SECONDS * (2 ** attempt) + random.uniform(0, _JITTER_SECONDS)
            time.sleep(delay)

        span.set_attribute("lock.acquire_failed", True)
        return None


def release_lock(client: redis.Redis, handle: LockHandle) -> bool:
    """自分が取得したロックであることを確認した上で解放する(修正2)。

    Args:
        client: Memorystore Redisクライアント。
        handle: acquire_lockで取得したLockHandle。

    Returns:
        解放に成功した場合True。既にTTL切れ等で他者のロックに置き換わっていた場合False。
    """
    released = client.eval(_RELEASE_LOCK_SCRIPT, 1, f"lock:{handle.sku}", handle.token)
    return bool(released)


def check_and_mark_processed(client: redis.Redis, order_id: str, version: int) -> bool:
    """Cloud Tasksのリトライによる二重引当を防ぐ冪等性チェック(修正3)。

    Args:
        client: Memorystore Redisクライアント。
        order_id: 注文ID。
        version: 引当バージョン(注文更新のたびにインクリメント)。

    Returns:
        初回実行の場合True(処理を続行してよい)。既に処理済みの場合False(スキップすべき)。
    """
    key = f"idempotency:{order_id}:{version}"
    is_first_time = client.set(key, "1", nx=True, ex=_IDEMPOTENCY_TTL_SECONDS)
    return bool(is_first_time)

Bad vs Good 設計比較

確認コマンド

📋 コピー# 1. Memorystore がStandard tier(HA構成)であることを確認
gcloud redis instances describe inventory-lock-cache --region=asia-northeast1 \
  --format="value(tier)"
# Expected: STANDARD_HA

# 2. fencing tokenが正しく機能し、他者のロックを誤って解放しないことを確認
kubectl exec -n mops deploy/inventory-reservation-service -c app -- \
  python -c "
from inventory_lock import acquire_lock, release_lock
import redis
c = redis.Redis(host='127.0.0.1', port=6379)
h1 = acquire_lock(c, 'SKU-TEST')
c.set('lock:SKU-TEST', 'other-token', xx=True)  # TTL切れ後に別Podが取得したと仮定
print(release_lock(c, h1))  # Expected: False (自分のトークンと不一致のため解放されない)
"

# 3. 冪等性チェックにより同一注文の二重処理がスキップされることを確認
kubectl exec -n mops deploy/inventory-reservation-service -c app -- \
  python -c "
from inventory_lock import check_and_mark_processed
import redis
c = redis.Redis(host='127.0.0.1', port=6379)
print(check_and_mark_processed(c, 'ORDER-1', 1))  # Expected: True (初回)
print(check_and_mark_processed(c, 'ORDER-1', 1))  # Expected: False (2回目はスキップ)
"

# 4. DataDogでロック取得待ち時間p99と二重引当検知件数を確認
datadog-ci metrics query "avg:inventory_lock.acquire_latency_ms{service:inventory-reservation-service}.rollup(p99)"
# Expected: p99が閾値500ms以内で安定