弱点補強: 信頼性設計 — Circuit Breaker（CLOSED/OPEN/HALF_OPEN）× SLO 99.9% エラーバジェット × マルチウィンドウ・マルチバーンレートアラート（MOps ベンダー通知API Bad→Good）

📋 コピーfrom __future__ import annotations

import asyncio
import random
import time
from dataclasses import dataclass
from enum import StrEnum
from typing import Callable, Final

import httpx

# ── 定数 ──────────────────────────────────────────
VENDOR_API_URL: Final[str] = "https://vendor-sms.example.com/v1/send"
REQUEST_TIMEOUT_SEC: Final[float] = 3.0
MAX_RETRY_ATTEMPTS: Final[int] = 3
BASE_BACKOFF_SEC: Final[float] = 0.5
MAX_BACKOFF_SEC: Final[float] = 4.0


class CircuitState(StrEnum):
    """Circuit Breaker の状態（3値のみを許容する値オブジェクト）."""
    CLOSED = "closed"        # 正常時: リクエストを全て通す
    OPEN = "open"              # 遮断時: リクエストを即座に拒否
    HALF_OPEN = "half_open"    # 復旧確認中: 限定的にプローブのみ通す


@dataclass(frozen=True, slots=True)
class CircuitBreakerConfig:
    """Circuit Breaker の設定値（完全コンストラクタ・不変）."""
    failure_threshold: int = 5
    recovery_timeout_sec: float = 30.0
    half_open_max_calls: int = 3
    half_open_success_threshold: int = 2


class VendorUnavailableError(RuntimeError):
    """Circuit が OPEN 中にリクエストが拒否されたことを示す例外."""


class CircuitBreaker:
    """ベンダー API 呼び出しを保護する Circuit Breaker（asyncio 対応）."""

    def __init__(self, config: CircuitBreakerConfig | None = None) -> None:
        self._config = config or CircuitBreakerConfig()
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._half_open_calls = 0
        self._half_open_successes = 0
        self._opened_at: float = 0.0
        self._lock = asyncio.Lock()

    async def call(self, func: Callable, *args: object, **kwargs: object) -> object:
        """保護対象の非同期関数を Circuit Breaker 経由で呼び出す.

        Args:
            func: 呼び出す非同期関数。
            *args, **kwargs: func に渡す引数。

        Returns:
            func の戻り値。

        Raises:
            VendorUnavailableError: OPEN で拒否した場合。
        """
        async with self._lock:
            self._transition_if_needed()
            if self._state is CircuitState.OPEN:
                raise VendorUnavailableError("circuit is OPEN: vendor API を保護中")
            if self._state is CircuitState.HALF_OPEN:
                if self._half_open_calls >= self._config.half_open_max_calls:
                    raise VendorUnavailableError("circuit is HALF_OPEN: プローブ上限")
                self._half_open_calls += 1

        try:
            # タイムアウトを一元管理（Bad実装は無限待機）
            result = await asyncio.wait_for(
                func(*args, **kwargs), timeout=REQUEST_TIMEOUT_SEC
            )
        except Exception:
            await self._on_failure()
            raise
        else:
            await self._on_success()
            return result

    def _transition_if_needed(self) -> None:
        """OPEN → HALF_OPEN への時間経過遷移（lock内で呼ぶ前提）."""
        match self._state:
            case CircuitState.OPEN:
                elapsed = time.monotonic() - self._opened_at
                if elapsed >= self._config.recovery_timeout_sec:
                    self._state = CircuitState.HALF_OPEN
                    self._half_open_calls = 0
                    self._half_open_successes = 0
            case _:
                pass

    async def _on_success(self) -> None:
        async with self._lock:
            match self._state:
                case CircuitState.HALF_OPEN:
                    self._half_open_successes += 1
                    if self._half_open_successes >= self._config.half_open_success_threshold:
                        self._state = CircuitState.CLOSED
                        self._failure_count = 0
                case CircuitState.CLOSED:
                    self._failure_count = 0
                case _:
                    pass

    async def _on_failure(self) -> None:
        async with self._lock:
            match self._state:
                case CircuitState.HALF_OPEN:
                    # プローブ失敗 → 即座に OPEN に戻す
                    self._state = CircuitState.OPEN
                    self._opened_at = time.monotonic()
                case CircuitState.CLOSED:
                    self._failure_count += 1
                    if self._failure_count >= self._config.failure_threshold:
                        self._state = CircuitState.OPEN
                        self._opened_at = time.monotonic()
                case _:
                    pass


async def _post_vendor_sms(phone: str, message: str) -> bool:
    """ベンダー API への実際の POST（タイムアウトは呼び出し元で一元管理）."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(VENDOR_API_URL, json={"to": phone, "body": message})
        resp.raise_for_status()
        return True


_breaker = CircuitBreaker()


async def send_sms(phone: str, message: str) -> bool:
    """SMS を送信する（Circuit Breaker + 指数バックオフ付き）.

    Args:
        phone: 送信先電話番号。
        message: 送信本文。

    Returns:
        送信成功なら True。

    Raises:
        VendorUnavailableError: OPEN 中で送信不可（リトライしない）。
    """
    last_exc: Exception | None = None
    for attempt in range(MAX_RETRY_ATTEMPTS):
        try:
            return await _breaker.call(_post_vendor_sms, phone, message)
        except VendorUnavailableError:
            raise  # OPEN中はリトライせず即座に失敗（フェイルファスト）
        except Exception as exc:  # noqa: BLE001
            last_exc = exc
            backoff = min(BASE_BACKOFF_SEC * (2**attempt), MAX_BACKOFF_SEC)
            jitter = random.uniform(0, backoff * 0.1)
            await asyncio.sleep(backoff + jitter)
    raise last_exc  # type: ignore[misc]