B システム設計/インフラ — Memorystore for Redis キャッシュ層（STANDARD_HA 自動フェイルオーバー × PRIVATE_SERVICE_ACCESS × AUTH/TLS × maxmemory-policy=allkeys-lru × コネクションプール × キャッシュスタンピード対策(分散ロック) × TTLジッター + Circuit Breaker フォールバック）（MOps フラッシュセール在庫参照API Bad→Good）

📋 コピー"""stock_cache.py — フラッシュセール在庫数キャッシュ（Redis cache-aside）"""
from __future__ import annotations

import random
import time
from dataclasses import dataclass, replace

import redis
from redis.exceptions import RedisError

CACHE_KEY_PREFIX = "stock"
BASE_TTL_SECONDS = 60
TTL_JITTER_SECONDS = 10           # 修正⑦: 同時失効を防ぐジッター幅
LOCK_TTL_SECONDS = 5
LOCK_WAIT_TIMEOUT_SECONDS = 3.0
LOCK_POLL_INTERVAL_SECONDS = 0.05
FAILURE_THRESHOLD = 5
OPEN_DURATION_SECONDS = 10.0

# 修正⑤: プロセス内で共有するコネクションプール
_pool = redis.ConnectionPool(
    host="REDIS_HOST_PLACEHOLDER",
    port=6379,
    password="REDIS_AUTH_PLACEHOLDER",
    ssl=True,
    max_connections=50,
    socket_timeout=1.0,
    socket_connect_timeout=1.0,
)


@dataclass(frozen=True, slots=True)
class CircuitBreakerState:
    """Redis障害検知の簡易サーキットブレーカー状態（値オブジェクト）。"""

    failure_count: int = 0
    opened_at: float | None = None


_breaker = CircuitBreakerState()


def _redis_client() -> redis.Redis:
    return redis.Redis(connection_pool=_pool)


def _is_breaker_open() -> bool:
    if _breaker.opened_at is None:
        return False
    return (time.monotonic() - _breaker.opened_at) < OPEN_DURATION_SECONDS


def _record_failure() -> None:
    global _breaker
    new_count = _breaker.failure_count + 1
    opened_at = time.monotonic() if new_count >= FAILURE_THRESHOLD else _breaker.opened_at
    _breaker = replace(_breaker, failure_count=new_count, opened_at=opened_at)


def _record_success() -> None:
    global _breaker
    _breaker = CircuitBreakerState()


def get_stock(product_id: str, db_lookup=query_stock_from_db) -> int:
    """在庫数を取得する（Redis優先、障害時はCloud SQLへフォールバック）。

    Args:
        product_id: 商品ID。
        db_lookup: DB問い合わせ関数（テスト時に差し替え可能）。

    Returns:
        在庫数。
    """
    cache_key = f"{CACHE_KEY_PREFIX}:{product_id}"

    if _is_breaker_open():                       # 修正⑦
        return db_lookup(product_id)

    try:
        client = _redis_client()                 # 修正⑤
        cached = client.get(cache_key)
        if cached is not None:
            _record_success()
            return int(cached)

        stock = _fetch_with_stampede_protection(  # 修正⑥
            client, product_id, cache_key, db_lookup
        )
        _record_success()
        return stock

    except RedisError:
        _record_failure()                         # 修正⑦
        return db_lookup(product_id)


def _fetch_with_stampede_protection(
    client: redis.Redis, product_id: str, cache_key: str, db_lookup
) -> int:
    """分散ロックでDB問い合わせを1リクエストに絞り込む。"""
    lock_key = f"lock:{cache_key}"
    deadline = time.monotonic() + LOCK_WAIT_TIMEOUT_SECONDS

    while time.monotonic() < deadline:
        acquired = client.set(lock_key, "1", nx=True, ex=LOCK_TTL_SECONDS)
        if acquired:
            try:
                stock = db_lookup(product_id)
                ttl = BASE_TTL_SECONDS + random.randint(0, TTL_JITTER_SECONDS)
                client.set(cache_key, stock, ex=ttl)
                return stock
            finally:
                client.delete(lock_key)

        time.sleep(LOCK_POLL_INTERVAL_SECONDS)
        cached = client.get(cache_key)
        if cached is not None:
            return int(cached)

    return db_lookup(product_id)


def query_stock_from_db(product_id: str) -> int:
    """Cloud SQL から在庫数を取得する（実装は省略）。"""
    ...