CAPTCHA 错误预算跟踪：用 SLO 量化解决可靠性

凌晨两点，监控群弹出一条消息：验证码识别成功率跌到了 91%。要不要叫醒值班工程师？

光看百分比很难判断——错误预算把目标成功率换算成不断消耗的失败配额：配额充足就正常观察，快耗尽才需要限流、报警或暂停发布。

本文给出可直接复用的错误预算跟踪实现（Python 与 JavaScript 各一份），并讲清楚燃烧率怎么算、报警阈值怎么定。

经验法则：先跑满一个完整统计窗口再定报警阈值，凭感觉设的数字大概率不是太松就是太紧。

什么是错误预算：从“感觉还行”到可以量化的数字

错误预算源自 SRE（可靠性工程）的 SLO 体系：先定目标成功率，再把允许失败次数换算成具体数字，核心是四个概念：

1. SLO——目标成功率（如 ≥ 95%）。
2. 错误预算——允许的失败次数（如 5% 可失败）。
3. 燃烧率——预算消耗速度，2× 即窗口过半就耗尽。
4. 窗口——统计周期，常见为滚动 24 小时或 7 天。

举例：24 小时窗口识别 10,000 次、SLO 95%，预算即 500 次失败；累计达到后应暂停高风险改动排查。

Python 实现：验证码错误预算跟踪器

跟踪器用滑动窗口保存最近的识别结果，实时算出剩余比例和燃烧率，状态变化时触发回调：

import time
import threading
from dataclasses import dataclass, field
from collections import deque
from enum import Enum

API_KEY = "YOUR_API_KEY"


class BudgetStatus(Enum):
    HEALTHY = "healthy"          # Budget > 50% remaining
    WARNING = "warning"          # Budget 10-50% remaining
    CRITICAL = "critical"        # Budget < 10% remaining
    EXHAUSTED = "exhausted"      # Budget depleted


@dataclass
class SLOConfig:
    """Service Level Objective configuration."""
    target_success_rate: float = 0.95  # 95%
    window_seconds: int = 86400        # 24 hours
    warning_threshold: float = 0.50    # Alert at 50% budget
    critical_threshold: float = 0.10   # Alert at 10% budget


@dataclass
class ErrorBudgetEvent:
    timestamp: float
    success: bool


class ErrorBudgetTracker:
    """Tracks error budget consumption for CAPTCHA solving."""

    def __init__(self, config: SLOConfig = SLOConfig()):
        self.config = config
        self._events: deque[ErrorBudgetEvent] = deque()
        self._lock = threading.Lock()
        self._callbacks: dict[BudgetStatus, list[callable]] = {
            status: [] for status in BudgetStatus
        }
        self._last_status = BudgetStatus.HEALTHY

    def on_status_change(self, status: BudgetStatus, callback: callable):
        """Register a callback for status transitions."""
        self._callbacks[status].append(callback)

    def record(self, success: bool):
        """Record a solve attempt."""
        now = time.monotonic()
        event = ErrorBudgetEvent(timestamp=now, success=success)

        with self._lock:
            self._events.append(event)
            self._prune(now)
            new_status = self._compute_status()

            if new_status != self._last_status:
                self._last_status = new_status
                for cb in self._callbacks.get(new_status, []):
                    try:
                        cb(self.get_report())
                    except Exception as e:
                        print(f"[BUDGET] Callback error: {e}")

    def _prune(self, now: float):
        """Remove events outside the window."""
        cutoff = now - self.config.window_seconds
        while self._events and self._events[0].timestamp < cutoff:
            self._events.popleft()

    def _compute_status(self) -> BudgetStatus:
        remaining = self.remaining_fraction
        if remaining <= 0:
            return BudgetStatus.EXHAUSTED
        if remaining < self.config.critical_threshold:
            return BudgetStatus.CRITICAL
        if remaining < self.config.warning_threshold:
            return BudgetStatus.WARNING
        return BudgetStatus.HEALTHY

    @property
    def total_events(self) -> int:
        with self._lock:
            return len(self._events)

    @property
    def success_count(self) -> int:
        with self._lock:
            return sum(1 for e in self._events if e.success)

    @property
    def failure_count(self) -> int:
        with self._lock:
            return sum(1 for e in self._events if not e.success)

    @property
    def current_success_rate(self) -> float:
        total = self.total_events
        return self.success_count / total if total > 0 else 1.0

    @property
    def error_budget_total(self) -> float:
        """Total allowed failures in the window."""
        total = self.total_events
        if total == 0:
            return 0
        return total * (1 - self.config.target_success_rate)

    @property
    def error_budget_remaining(self) -> float:
        """Remaining failure allowance."""
        return max(0, self.error_budget_total - self.failure_count)

    @property
    def remaining_fraction(self) -> float:
        """Fraction of error budget remaining (0.0 to 1.0)."""
        budget = self.error_budget_total
        if budget <= 0:
            return 1.0 if self.failure_count == 0 else 0.0
        return max(0, self.error_budget_remaining / budget)

    @property
    def burn_rate(self) -> float:
        """How fast the budget is being consumed (1.0 = normal, 2.0 = 2× faster)."""
        total = self.total_events
        if total == 0:
            return 0.0
        expected_failures = total * (1 - self.config.target_success_rate)
        if expected_failures == 0:
            return 0.0
        return self.failure_count / expected_failures

    def get_report(self) -> dict:
        return {
            "status": self._last_status.value,
            "slo_target": self.config.target_success_rate,
            "current_rate": round(self.current_success_rate, 4),
            "total_events": self.total_events,
            "successes": self.success_count,
            "failures": self.failure_count,
            "budget_total": round(self.error_budget_total, 1),
            "budget_remaining": round(self.error_budget_remaining, 1),
            "budget_remaining_pct": round(self.remaining_fraction * 100, 1),
            "burn_rate": round(self.burn_rate, 2),
        }


# --- Integration with solver ---

budget = ErrorBudgetTracker(SLOConfig(
    target_success_rate=0.95,
    window_seconds=3600,  # 1-hour window for demo
))

# Register alerts
budget.on_status_change(BudgetStatus.WARNING, lambda r:
    print(f"[ALERT] Budget warning: {r['budget_remaining_pct']}% remaining"))

budget.on_status_change(BudgetStatus.CRITICAL, lambda r:
    print(f"[ALERT] Budget critical: {r['budget_remaining_pct']}% remaining"))

budget.on_status_change(BudgetStatus.EXHAUSTED, lambda r:
    print(f"[ALERT] Budget EXHAUSTED — throttle new requests"))


def solve_with_budget(params: dict) -> str:
    """Solve CAPTCHA while tracking error budget."""
    import requests

    if budget._last_status == BudgetStatus.EXHAUSTED:
        raise RuntimeError("Error budget exhausted — solving paused")

    try:
        submit_params = {**params, "key": API_KEY, "json": 1}
        resp = requests.post(
            "https://ocr.captchaai.com/in.php", data=submit_params, timeout=30
        ).json()
        if resp.get("status") != 1:
            budget.record(False)
            raise RuntimeError(f"Submit: {resp.get('request')}")

        task_id = resp["request"]
        start = time.monotonic()
        while time.monotonic() - start < 180:
            time.sleep(5)
            poll = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": API_KEY, "action": "get", "id": task_id, "json": 1,
            }, timeout=15).json()

            if poll.get("request") == "CAPCHA_NOT_READY":
                continue
            if poll.get("status") == 1:
                budget.record(True)
                return poll["request"]

            budget.record(False)
            raise RuntimeError(f"Solve: {poll.get('request')}")

        budget.record(False)
        raise RuntimeError("Timeout")

    except Exception:
        budget.record(False)
        raise


# Usage
for i in range(100):
    try:
        token = solve_with_budget({
            "method": "turnstile",
            "sitekey": "0x4XXXXXXXXXXXXXXXXX",
            "pageurl": "https://example.com",
        })
    except RuntimeError as e:
        if "exhausted" in str(e):
            print(f"Stopped at iteration {i}")
            break

print(budget.get_report())

JavaScript 实现：验证码错误预算跟踪器

逻辑与 Python 版一致，方便在 Node.js 或监控面板里复用：

class ErrorBudgetTracker {
  #events = [];
  #config;
  #callbacks = {};

  constructor(config = {}) {
    this.#config = {
      targetRate: config.targetRate || 0.95,
      windowMs: config.windowMs || 3600_000,
      warningThreshold: config.warningThreshold || 0.5,
      criticalThreshold: config.criticalThreshold || 0.1,
    };
    this.lastStatus = "healthy";
  }

  on(status, callback) {
    this.#callbacks[status] = this.#callbacks[status] || [];
    this.#callbacks[status].push(callback);
  }

  record(success) {
    const now = Date.now();
    this.#events.push({ time: now, success });
    this.#prune(now);

    const newStatus = this.#computeStatus();
    if (newStatus !== this.lastStatus) {
      this.lastStatus = newStatus;
      for (const cb of this.#callbacks[newStatus] || []) {
        cb(this.report());
      }
    }
  }

  #prune(now) {
    const cutoff = now - this.#config.windowMs;
    while (this.#events.length && this.#events[0].time < cutoff) {
      this.#events.shift();
    }
  }

  #computeStatus() {
    const frac = this.remainingFraction;
    if (frac <= 0) return "exhausted";
    if (frac < this.#config.criticalThreshold) return "critical";
    if (frac < this.#config.warningThreshold) return "warning";
    return "healthy";
  }

  get total() { return this.#events.length; }
  get successes() { return this.#events.filter((e) => e.success).length; }
  get failures() { return this.#events.filter((e) => !e.success).length; }
  get currentRate() { return this.total ? this.successes / this.total : 1; }

  get budgetTotal() {
    return this.total * (1 - this.#config.targetRate);
  }

  get budgetRemaining() {
    return Math.max(0, this.budgetTotal - this.failures);
  }

  get remainingFraction() {
    const bt = this.budgetTotal;
    if (bt <= 0) return this.failures === 0 ? 1 : 0;
    return Math.max(0, this.budgetRemaining / bt);
  }

  get burnRate() {
    const expected = this.total * (1 - this.#config.targetRate);
    return expected > 0 ? this.failures / expected : 0;
  }

  report() {
    return {
      status: this.lastStatus,
      currentRate: Math.round(this.currentRate * 10000) / 10000,
      total: this.total,
      failures: this.failures,
      budgetRemainingPct: Math.round(this.remainingFraction * 1000) / 10,
      burnRate: Math.round(this.burnRate * 100) / 100,
    };
  }
}

// Usage
const budget = new ErrorBudgetTracker({ targetRate: 0.95, windowMs: 3600_000 });

budget.on("warning", (r) => console.log(`[WARN] ${r.budgetRemainingPct}% budget left`));
budget.on("exhausted", (r) => console.log("[ALERT] Budget exhausted!"));

// Record results from your solver
budget.record(true);   // success
budget.record(false);  // failure
console.log(budget.report());

燃烧率怎么看：预算消耗速度的报警线

燃烧率告诉你“按这个速度预算还能撑多久”，比“现在还剩多少”更能提前预警：

• 燃烧率 < 1.0：慢于预期，无需处理。
• 燃烧率 = 1.0：窗口结束时刚好耗尽，需观察。
• 燃烧率 = 2.0：窗口过半即耗尽，需降速排查。
• 燃烧率 ≥ 5.0：急速消耗，暂停非关键任务。

燃烧率 ≥ 2 比“跌破 SLO”更值得关注。

国内团队怎么落地：报警接入 + 按类型拆分预算

落地顺序建议如下：

1. 把报警接到企业微信或钉钉机器人上，进入 critical 状态时自动 @ 值班人。
2. 按验证码类型分开算预算——国内多用 GeeTest（极验），海外多为 reCAPTCHA/Turnstile，基线不同，合并统计会掩盖问题。
3. 给每种类型单独设 SLO：CaptchaAI 支持 GeeTest v3、reCAPTCHA v2/v3、Turnstile，可以分别跟踪。

国内依赖安装建议加上镜像源（如清华 TUNA），减少无关噪音。

常见故障排查

常见问题

需要多久检查一次错误预算？

日常按小时巡检即可，报警是实时的——每次 record() 都会重算一次剩余比例。

报警阈值要跟 SLA 罚则对齐吗？

对外 SLA 需更早触发 critical；内部预算可以更激进，但 SLA 始终是下限。

reCAPTCHA、Turnstile、GeeTest 混用时要共用一个预算吗？

不建议，按类型分开：reCAPTCHA v2 设 93%，Turnstile 设 97%，避免异常被平均。

相关文章

• 用 DynamoDB 做无服务器验证码识别记录

下一步

与其等成功率跌破阈值才手忙脚乱，不如现在接入——获取 CaptchaAI API Key。

相关指南：

• 验证码 API 调用的断路器模式
• 验证码 Worker 的健康检查接口
• 用 Prometheus 和 Grafana 监控验证码识别成功率