Tutorials

CAPTCHA Çözme Güvenilirliği için Hata Bütçesi Takibi

Hata bütçesi, güvenilirlik hedefinizden ne kadar sapabileceğinizi somut bir sayıya çeviren araçtır: belirlediğiniz SLO'nun altına düşmeden kaç çözümün başarısız olabileceğini ve bu pay tükendiğinde ne yapacağınızı önceden söyler. Böylece "geçen hafta başarı oranı %95'ti, bu hafta %94,2'ye indi — sorun mu?" sorusuna hisle değil rakamla yanıt verirsiniz.

Bu rehberde şunları kuruyoruz:

  • Hata bütçesini Python ve JavaScript ile kayan bir pencere üzerinden takip etme
  • Yanma hızı (burn rate) eşiklerine göre otomatik uyarı verme
  • Bütçe tükendiğinde çözüm hattını kademeli olarak yavaşlatma

Hata bütçesi nedir? Dört temel kavram

Kavram Tanımı Örnek
SLO Hedeflediğiniz başarı oranı Çözümlerin %95'i başarılı
Hata bütçesi Tolere edilen başarısızlık payı Çözümlerin %5'i başarısız olabilir
Yanma hızı Bütçenin ne kadar hızlı tükendiği 2× hız, pencerenin yarısında bitmesi demek
Pencere Ölçüm periyodu Kayan 24 saat veya 7 gün

SLO'nuz kayan 24 saatlik pencerede 10.000 çözüm üzerinden %95 ise hata bütçeniz 500 başarısızlıktır. Bu 500'e ulaştığınızda yeni dağıtımları ve riskli değişiklikleri durdurmak, kalan payı korumak anlamına gelir.

Bütçeyi CAPTCHA türüne göre boyutlandırın

Tek bir global SLO farklı CAPTCHA türlerini aynı kefeye koyar ve sizi yanıltır. Türler hız ve zorluk açısından ayrışır; her tür için ayrı bir bütçe izlediğiniz sürece bir türdeki bozulma diğerini maskeleyemez. Pratik bir başlangıç noktası:

  • Hızlı türler (örn. Cloudflare Turnstile): tipik olarak 10 saniyenin altında temizlenir; daha sıkı bir SLO tutabilirsiniz.
  • Yavaş, yüksek eşzamanlılıklı türler (örn. reCAPTCHA v2): 60 saniyeye kadar çıkabilir; payı daha geniş bırakın.
  • Görüntü/OCR türleri: hacme ve içeriğe göre değişir; taban oranınızı ölçtükten sonra hedef koyun.

İstanbul'da bir e-ticaret firmasına otomasyon geliştiren bir ekip düşünün: ödeme adımındaki Turnstile doğrulamaları için %97 gibi sıkı bir SLO, kampanya kayıtlarındaki görüntü tabanlı CAPTCHA'lar için ise daha temkinli bir hedef tutuyorlar. İki tür ayrı bütçelerle izlendiği için birindeki dalgalanma diğerinin sağlık göstergesini gizlemiyor. Kazınan verinin kişisel veri içerdiği senaryolarda bu ölçümleri yetkili QA ve veri toplama akışları içinde tutmak, KVKK açısından da doğru yaklaşımdır.

Doğru başlangıç noktası tahmin değil ölçümdür: mevcut başarı oranınızı bir süre kaydedin, ardından SLO'nuzu bu taban çizginin 2–3 puan altına koyarak anlamlı bir hata bütçesi bırakın.

Python ile hata bütçesi takibi

Aşağıdaki ErrorBudgetTracker sınıfı, kayan pencere içindeki her çözüm denemesini kaydeder, kalan bütçeyi hesaplar ve durum HEALTHY → WARNING → CRITICAL → EXHAUSTED arasında değiştikçe kayıtlı callback'leri tetikler. solve_with_budget fonksiyonu ise bütçe tükendiğinde yeni istekleri durdurur.

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 ile hata bütçesi takibi

Aynı mantığın Node.js karşılığı; solver'ınızdan gelen her sonucu record(true/false) ile besler, eşik aşıldığında on(...) ile kayıtlı uyarıları çalıştırır.

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

Yanma hızını okuma (burn rate)

Yanma hızı, bütçeyi beklenene kıyasla ne kadar hızlı tükettiğinizi gösterir. 1,0 "tam beklenen tempo" demektir; üzerine çıktıkça pencere sonundan önce tükenirsiniz.

Yanma hızı Anlamı Yapılacak
< 1,0 Beklenenden yavaş tükeniyor Aksiyon gerekmez
1,0 Pencere sonunda tam tükenme temposunda Yakından izleyin
2,0 Bütçe pencerenin yarısında bitiyor İnceleyin ve yavaşlayın
5,0+ Çok hızlı tükeniyor Kritik olmayan çözümleri duraklatın

Bütçe tükendiğinde ne yapmalı?

Tükenmiş bir bütçeyi sessizce görmezden gelmek en kötü seçenektir. Müdahaleyi en hafiften en serte doğru kademelendirin ve her adımı otomatikleştirin:

  1. Ekibi ve nöbetçiyi uyarın (ilk sinyal, henüz eylem değil).
  2. Yeni istekleri kısarak (rate limiting) hacmi düşürün.
  3. Kritik olmayan çözümleri geçici olarak duraklatın.
  4. Gerekirse manuel CAPTCHA işlemeye ya da yedek iş akışına geçin.

Yukarıdaki Python örneğindeki EXHAUSTED durumu, tam olarak bu kademelendirmeyi tetiklemek için vardır: callback'i uyarıya, solve_with_budget içindeki kontrolü ise yeni isteklerin durdurulmasına bağlayın.

Sorun giderme

Sorun Neden Çözüm
Bütçe çok çabuk tükeniyor SLO gerçek koşullara göre fazla sıkı Geçmiş verilere dayalı gerçekçi bir SLO belirleyin
Bütçe hiç tükenmiyor SLO fazla cömert Güvenilirliği zorlamak için SLO'yu bir miktar sıkın
Durum sürekli durumlar arasında gidip geliyor Pencere çok kısa Daha uzun ölçüm penceresi kullanın (1 saat yerine 24 saat)
Düşük hacimde yanma hızı yanıltıcı Birkaç olay hesabı çarpıtıyor Hesaplamadan önce minimum bir olay sayısı şartı koyun
Takip belleği sürekli büyüyor Olaylar budanmıyor _prune'ın her record() çağrısında çalıştığını doğrulayın

Sık sorulan sorular

Hata bütçesi ile yanma hızı arasındaki fark nedir?

Hata bütçesi bir stoktur: pencere içinde geriye kalan başarısızlık payıdır. Yanma hızı ise bir akıştır; bu payı ne kadar hızlı tükettiğinizi gösterir. Bütçe hâlâ yarı dolu görünse bile yanma hızı 5,0 ise pay dakikalar içinde bitebilir, bu yüzden ikisini birlikte izleyin.

Ölçüm penceresi için 24 saat mi yoksa 7 gün mü seçmeliyim?

Trafiğiniz gün içinde dalgalanıyorsa daha uzun pencere (7 gün) gürültüyü yumuşatır ve durumun sürekli değişmesini önler. Ani bozulmalara hızlı tepki vermeniz gerekiyorsa 24 saat daha uygundur. Çoğu ekip kısa pencereyle uyarır, uzun pencereyle karar verir.

Az sayıda çözüm varken yanma hızına güvenebilir miyim?

Hayır. Birkaç yüz olayın altında tek bir başarısızlık bile oranı olağanüstü şişirir. Hesaplamayı çalıştırmadan önce minimum bir olay sayısı (örneğin birkaç yüz) şartı koyun; aksi halde uyarılar sürekli yanlış alarm üretir.

SLO hedefimi hangi başarı oranına göre belirlemeliyim?

Tek bir "doğru" oran yoktur; türe ve iş yükünüze bağlıdır. Önce mevcut başarı oranınızı ölçün, ardından SLO'yu bu taban çizginin biraz altına koyun. reCAPTCHA v2 gibi daha yavaş türlerde payı geniş, Turnstile gibi hızlı türlerde daha dar tutabilirsiniz.

İlgili makaleler

Sonraki adımlar

CAPTCHA çözüm güvenilirliğinizi tahminle değil rakamla yönetin — CaptchaAI API anahtarınızı alın ve hata bütçesi takibini iş akışınıza ekleyin.

İlgili rehberler:

Bu makale için yorumlar devre dışı bırakılmıştır.