コネクタが遅くなり始めた夜、処理を自動で退避させる — 個人運用のヘルス連動サーキットブレーカー

import time
from collections import deque
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
 
 
class State(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"
 
 
@dataclass
class Sample:
    ok: bool
    latency_ms: float
    ts: float
 
 
@dataclass
class ConnectorBreaker:
    name: str
    window_sec: float = 900.0        # 直近15分を健康の観測窓にする
    min_samples: int = 8             # これ未満なら判定を保留する
    error_rate_open: float = 0.5     # エラー率がこれを超えたら開放
    p95_open_ms: float = 8000.0      # p95レイテンシがこれを超えたら開放
    cooldown_sec: float = 300.0      # 開放後、探索まで待つ時間
    half_open_success: int = 3       # 復旧確定に必要な連続成功数
 
    state: State = State.CLOSED
    opened_at: float = 0.0
    probe_success: int = 0
    _samples: deque = field(default_factory=deque)
 
    def _prune(self, now: float) -> None:
        limit = now - self.window_sec
        while self._samples and self._samples[0].ts < limit:
            self._samples.popleft()
 
    def _error_rate(self) -> float:
        if not self._samples:
            return 0.0
        failed = sum(1 for s in self._samples if not s.ok)
        return failed / len(self._samples)
 
    def _p95_latency(self) -> float:
        if not self._samples:
            return 0.0
        values = sorted(s.latency_ms for s in self._samples)
        idx = max(0, int(len(values) * 0.95) - 1)
        return values[idx]

    def allow(self, now: Optional[float] = None) -> bool:
        """呼び出し前に必ず問い合わせる。False なら退避側へ回す。"""
        now = time.time() if now is None else now
        self._prune(now)
        if self.state is State.OPEN:
            if now - self.opened_at >= self.cooldown_sec:
                self.state = State.HALF_OPEN
                self.probe_success = 0
                return True   # 探索を一回だけ通す
            return False
        return True
 
    def record(self, ok: bool, latency_ms: float,
               now: Optional[float] = None) -> None:
        now = time.time() if now is None else now
        self._samples.append(Sample(ok=ok, latency_ms=latency_ms, ts=now))
        self._prune(now)
 
        if self.state is State.HALF_OPEN:
            if ok:
                self.probe_success += 1
                if self.probe_success >= self.half_open_success:
                    self.state = State.CLOSED
            else:
                self.state = State.OPEN
                self.opened_at = now
            return
 
        if len(self._samples) < self.min_samples:
            return
        if (self._error_rate() >= self.error_rate_open
                or self._p95_latency() >= self.p95_open_ms):
            self.state = State.OPEN
            self.opened_at = now

import time
 
 
def call_with_breaker(breaker, fn, fallback):
    """コネクタ呼び出しをブレーカーで包む。開放中は fallback を返す。"""
    if not breaker.allow():
        return fallback()
 
    start = time.perf_counter()
    try:
        result = fn()
    except Exception:
        elapsed_ms = (time.perf_counter() - start) * 1000
        breaker.record(ok=False, latency_ms=elapsed_ms)
        return fallback()
 
    elapsed_ms = (time.perf_counter() - start) * 1000
    breaker.record(ok=True, latency_ms=elapsed_ms)
    return result

import json
from pathlib import Path
 
 
def enqueue_for_retry(task, path="retry_queue.jsonl"):
    """開放中に落ちた仕事を退避キューへ書き出す。次サイクルで拾い直す。"""
    line = json.dumps(task, ensure_ascii=False)
    with Path(path).open("a", encoding="utf-8") as f:
        f.write(line + "\n")

import json
from pathlib import Path
 
 
def save_samples(breaker, path):
    rows = [{"ok": s.ok, "latency_ms": s.latency_ms, "ts": s.ts}
            for s in breaker._samples]
    Path(path).write_text(json.dumps(rows), encoding="utf-8")
 
 
def load_samples(breaker, path, now):
    p = Path(path)
    if not p.exists():
        return
    for row in json.loads(p.read_text(encoding="utf-8")):
        breaker._samples.append(
            Sample(ok=row["ok"], latency_ms=row["latency_ms"], ts=row["ts"])
        )
    breaker._prune(now)