Pelacakan Error Budget untuk Keandalan Solve CAPTCHA

Berapa banyak kegagalan solve CAPTCHA yang masih boleh terjadi sebelum sistem Anda dianggap bermasalah? Error budget menjawab pertanyaan itu dengan angka, bukan perasaan. Alih-alih menebak-nebak begitu dashboard menunjukkan warna merah, Anda menetapkan SLO, menghitung sisa jatah kegagalan, dan tahu persis kapan harus menahan deployment — bukan setelah tim support kebanjiran tiket.

Bayangkan tim automation di Jakarta yang menjalankan seribuan task solve Turnstile per jam dari worker di AWS ap-southeast-1: target SLO 95%, tapi realisasi minggu ini 94,2%. Tanpa error budget, angka itu cuma terasa "agak turun". Dengan error budget, Anda tahu persis apakah ini masih dalam toleransi atau sudah waktunya menahan perubahan berisiko.

Apa Itu Error Budget dalam Solve CAPTCHA?

• SLO — target tingkat keberhasilan solve yang disepakati. Contoh: 95% solve harus berhasil.
• Error budget — jatah kegagalan yang masih ditoleransi sebelum SLO dianggap dilanggar. Contoh: 5% dari total solve boleh gagal.
• Burn rate — kecepatan jatah kegagalan itu terpakai. Contoh: 2× artinya budget habis di separuh window.
• Window — rentang waktu pengukuran. Contoh: rolling 24 jam atau 7 hari.

Contoh konkret: SLO 95% pada window 24 jam dengan 10.000 solve berarti jatah kegagalan Anda 500 kali. Begitu counter kegagalan menyentuh angka 500, deployment baru dan perubahan konfigurasi berisiko harus dihentikan dulu.

Aturan praktis: begitu counter kegagalan menyentuh batas error budget, deployment baru dan perubahan berisiko berhenti dulu — bukan keputusan yang diambil panik di tengah insiden, tapi aturan main yang sudah disepakati sejak awal.

Ambang Alert Berdasarkan Burn Rate

Burn rate memberi tahu Anda seberapa cepat jatah kegagalan itu terpakai dibanding laju normal. Gunakan ambang berikut untuk menentukan respons yang tepat sebelum budget benar-benar habis:

Membangun Error Budget Tracker CAPTCHA dengan Python

Tracker berikut menghitung jatah kegagalan secara real-time dari deque event bertanda waktu, lalu memicu callback begitu status berubah — dari healthy ke warning, critical, sampai exhausted. Cocok dijalankan berdampingan dengan fungsi solve_with_budget yang memanggil in.php dan res.php CaptchaAI:

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

Versi JavaScript: Error Budget Tracker CAPTCHA untuk Node.js

Tim yang menjalankan worker solve di Node.js bisa pakai versi berikut — logikanya identik dengan versi Python, hanya memakai private class field dan callback per status:

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

Troubleshooting Error Budget yang Meleset

Masalah yang paling sering muncul di lapangan — termasuk saat traffic melonjak mendadak, misalnya saat kampanye flash sale seperti Harbolnas atau 11.11 ketika volume solve naik drastis dalam hitungan menit:

Tanya Jawab Error Budget CAPTCHA

Berapa lama window pengukuran yang tepat untuk error budget CAPTCHA?

Untuk kebanyakan tim automation, window rolling 24 jam sudah cukup stabil — cukup panjang untuk meredam fluktuasi jam sibuk vs jam sepi, tapi tetap responsif untuk menangkap masalah dalam satu hari kerja. Tim dengan volume solve sangat tinggi (puluhan ribu per hari) bisa mempersempit ke window 6–12 jam; tim dengan volume rendah sebaiknya memakai window 7 hari supaya jumlah event yang terhitung cukup banyak untuk bermakna secara statistik.

Berapa target SLO yang realistis untuk solve CAPTCHA?

Tergantung tipe CAPTCHA yang Anda proses. reCAPTCHA v2 biasanya mencapai tingkat keberhasilan 90–95%, Turnstile cenderung lebih tinggi, sementara image CAPTCHA bervariasi tergantung kualitas gambar. Ukur dulu baseline tingkat keberhasilan Anda saat ini selama beberapa minggu, lalu tetapkan SLO 2–3% di bawah baseline itu — supaya error budget-nya benar-benar bermakna, bukan angka yang selalu longgar atau selalu habis.

Apa yang harus dilakukan saat error budget CAPTCHA habis?

Ada beberapa opsi, dari paling ringan sampai paling agresif: kirim alert ke tim, throttle request baru, pause solve yang tidak kritis, atau sementara alihkan ke penanganan CAPTCHA manual. Karena CaptchaAI menagih per thread dan bukan per solve, menjeda task yang tidak kritis saat budget kritis tidak membuang kuota yang sudah dibayar — thread yang menganggur tinggal menunggu task berikutnya. Yang tidak boleh dilakukan: mendiamkan budget yang sudah habis dan membiarkan pipeline berjalan seperti biasa.

Apakah error budget tracking butuh tool monitoring terpisah seperti Prometheus atau Grafana?

Tidak wajib untuk mulai. Tracker Python atau JavaScript pada panduan ini sudah cukup untuk pipeline skala kecil-menengah — cukup simpan hasil get_report() atau report() ke log atau database Anda. Begitu volume solve naik dan Anda butuh dashboard historis serta alerting multi-channel, barulah menyambungkan metrik ini ke Prometheus dan Grafana jadi masuk akal.

Bagaimana menyesuaikan error budget saat volume solve masih kecil?

Pada volume rendah (di bawah beberapa ratus solve per window), burn rate gampang menyesatkan karena satu-dua kegagalan bisa melonjakkan angkanya drastis. Solusinya: perpanjang window pengukuran, dan tetapkan jumlah event minimum (misalnya 100 solve) sebelum sistem mulai menghitung dan melaporkan burn rate.

Artikel Terkait

• Melacak Solve CAPTCHA Serverless dengan DynamoDB

Langkah Selanjutnya

Lacak keandalan solve CAPTCHA Anda dengan angka, bukan tebakan — dapatkan API key CaptchaAI Anda dan pasang error budget tracker ini di pipeline Anda.

Panduan terkait:

• Circuit Breaker untuk Panggilan API CAPTCHA
• Health Check Endpoint untuk Worker CAPTCHA
• Monitoring Solve Rate CAPTCHA dengan Prometheus dan Grafana