Budget d'erreurs et fiabilité de la résolution CAPTCHA

Un budget d'erreurs transforme une intuition — « la résolution semble moins fiable cette semaine » — en une décision chiffrée : combien d'échecs tolérer avant de passer sous votre objectif de fiabilité ? Voici comment poser un SLO, mesurer le budget restant et automatiser alertes et throttling, en Python et JavaScript.

Budget d'erreurs : SLO, fenêtre et taux de consommation

Concept	Définition	Exemple
SLO	Taux de réussite cible	95 % de résolutions réussies
Budget d'erreurs	Taux d'échec autorisé	5 % des résolutions
Taux de consommation (burn rate)	Vitesse de consommation	2× = budget épuisé à mi-fenêtre
Fenêtre	Période de mesure	Glissante sur 24 h ou 7 jours

Le calcul est direct : avec un SLO de 95 % sur une fenêtre glissante de 24 heures et 10 000 résolutions, votre budget vaut 500 échecs. Passé ce seuil, gelez tout déploiement risqué jusqu'au retour à la normale.

Prenons une équipe QA e-commerce dont les workers tournent sur Scaleway à Paris, visant 96 % sur 24 heures. Tant que le budget reste positif, elle déploie ; dès qu'il approche de zéro, elle gèle tout changement. Le budget devient une règle partagée.

Suivre le budget d'erreurs en Python

La classe ci-dessous garde les événements dans une fenêtre glissante, calcule le budget restant et déclenche un callback à chaque changement d'état. Un verrou la rend utilisable depuis plusieurs threads de résolution.

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())

L'essentiel : solve_with_budget enregistre chaque tentative — in.php pour la soumission, res.php pour l'interrogation — et refuse de démarrer si le budget est épuisé. La coupure du trafic découle alors du budget.

Le même suivi côté client en JavaScript

Pour un worker Node.js ou un tableau de bord navigateur, voici l'équivalent, avec champs privés et fenêtre en millisecondes.

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());

Alertes sur le taux de consommation

Le taux de consommation (burn rate) mesure ce que le budget ignore : à quelle vitesse fond-il ? À 1,0, vous l'épuisez pile en fin de fenêtre ; au-delà, vous dérivez.

Taux de consommation	Signification	Action
< 1,0	Plus lent que prévu	Aucune action
1,0	Épuisement en fin de fenêtre	Surveiller de près
2,0	Budget épuisé à mi-fenêtre	Enquêter et ralentir
5,0+	Consommation très rapide	Suspendre le non-critique

Câblez ces seuils sur vos callbacks :

Alerte informative dès 1,0.
Page d'astreinte à 2,0 : épuisement avant la fin de la fenêtre.
Throttling automatique dès 5,0, couplé au besoin à un disjoncteur (circuit breaker).

Dépannage

Les défauts courants viennent d'un SLO mal calibré ou d'une fenêtre inadaptée.

Problème	Cause	Correctif
Budget épuisé trop vite	SLO trop serré	Calez le SLO sur l'historique
Budget jamais consommé	SLO trop généreux	Resserrez le SLO
État qui oscille	Fenêtre trop courte	Allongez la fenêtre (24 h)
Taux de consommation trompeur	Trop peu d'événements	Exigez un minimum d'événements
Mémoire du tracker qui gonfle	Événements non élagués	`_prune` doit tourner à chaque `record()`

FAQ

Quelle différence entre un SLO et un budget d'erreurs ?

Le SLO est l'objectif — par exemple 95 % de réussite. Le budget d'erreurs en découle : la quantité d'échecs tolérée sur la fenêtre. Le SLO fixe la cible ; le budget, la marge restante.

Sur quelle fenêtre mesurer le budget d'erreurs ?

Une fenêtre glissante de 24 heures convient à la plupart, 7 jours pour lisser le bruit. Trop courte, l'état oscille ; trop longue, le budget réagit trop tard. Commencez à 24 heures.

Faut-il suivre un budget distinct par type de CAPTCHA ?

Oui, dès que vos taux de réussite diffèrent d'un type à l'autre. Un SLO de 93 % côté reCAPTCHA v2 et de 96 % côté Cloudflare Turnstile mérite deux budgets distincts. Instanciez un tracker par method.

Le budget d'erreurs remplace-t-il un disjoncteur ?

Non, ils sont complémentaires. Le budget pilote vos décisions de déploiement sur la fenêtre ; le disjoncteur coupe le trafic dès que les échecs s'enchaînent. On le branche sur un seuil de taux de consommation.