Quy trình giải CAPTCHA của bạn hướng tới tỷ lệ thành công 95%. Tuần trước là 94,2%. Đó có phải là một vấn đề? Nếu không có ngân sách sai sót, bạn không thể trả lời câu hỏi đó một cách định lượng. Ngân sách lỗi cho bạn biết chính xác số lượng lỗi bạn có thể chịu đựng trước khi độ tin cậy giảm xuống dưới mức SLO của bạn - và phải làm gì khi hết ngân sách.
Lỗi cơ bản về ngân sách
| Khái niệm | Định nghĩa | Ví dụ |
|---|---|---|
| CHƯA | Tỷ lệ thành công mục tiêu | Giải quyết thành công 95% |
| Lỗi ngân sách | Tỷ lệ thất bại cho phép | 5% tổng số lần giải quyết có thể thất bại |
| Tốc độ ghi | Ngân sách được tiêu thụ nhanh như thế nào | 2× có nghĩa là ngân sách đã cạn trong một nửa thời lượng |
| Cửa sổ | Thời gian đo | Lăn 24 giờ hoặc 7 ngày |
Nếu SLO của bạn là 95% trong khoảng thời gian 24 giờ với 10.000 lần giải quyết, thì ngân sách lỗi của bạn là 500 lần thất bại. Khi bạn gặp phải 500 lần thất bại, việc triển khai mới hoặc các thay đổi rủi ro sẽ dừng lại.
Python: Trình theo dõi ngân sách lỗi
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: Trình theo dõi ngân sách lỗi
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());
Cảnh báo tốc độ ghi
| Tốc độ ghi | Ý nghĩa | hành động |
|---|---|---|
| < 1,0 | Tiêu thụ chậm hơn dự kiến | Không cần hành động |
| 1.0 | Đang trên đà xả hơi ở cuối cửa sổ | Giám sát chặt chẽ |
| 2.0 | Ngân sách đã cạn kiệt trong một nửa thời hạn | Hãy điều tra và làm chậm lại |
| 5.0+ | Tiêu thụ ngân sách nhanh chóng | Tạm dừng các giải pháp không quan trọng |
Khắc phục sự cố
| Vấn đề | Nguyên nhân | Cách xử lý |
|---|---|---|
| Ngân sách cạn kiệt quá nhanh | SLO quá chặt so với điều kiện thực tế | Đặt SLO thực tế dựa trên dữ liệu lịch sử |
| Ngân sách không bao giờ tiêu dùng | SLO quá hào phóng | Thắt chặt SLO để thúc đẩy cải thiện độ tin cậy |
| Chuyển đổi trạng thái giữa các trạng thái | Cửa sổ quá ngắn | Sử dụng khoảng thời gian đo dài hơn (24h so với 1h) |
| Tốc độ ghi sai lệch ở âm lượng thấp | Một vài sự kiện làm lệch phép tính | Yêu cầu số lượng sự kiện tối thiểu trước khi tính tỷ lệ đốt cháy |
| Bộ nhớ theo dõi ngân sách tăng lên | Sự kiện không được cắt tỉa | Xác minh _prune chạy trên mọi lệnh gọi record() |
Câu hỏi thường gặp
SLO thực tế để giải CAPTCHA là gì?
Phụ thuộc vào loại CAPTCHA. reCAPTCHA v2 thường đạt tỷ lệ giải quyết 90–95%. Cloudflare Turnstile có thể cao hơn. CAPTCHA hình ảnh khác nhau. Bắt đầu bằng cách đo lường tỷ lệ thành công hiện tại của bạn, sau đó đặt SLO 2–3% dưới mức cơ sở đó để tạo ra mức sai sót có ý nghĩa.
Điều gì sẽ xảy ra khi hết số lượng lỗi?
Các tùy chọn từ ít nhất đến tích cực nhất: cảnh báo nhóm, điều tiết các yêu cầu mới, tạm dừng các giải pháp không cần thiết, chuyển sang xử lý CAPTCHA thủ công. Đừng bao giờ âm thầm bỏ qua một khoản ngân sách đã cạn kiệt.
Làm cách nào để xử lý số lượng lỗi trên nhiều loại CAPTCHA?
Theo dõi ngân sách riêng cho từng loại. reCAPTCHA có thể có SLO 93% trong khi Turnstile có 97%. Việc tổng hợp chúng vào một ngân sách sẽ che giấu các vấn đề cụ thể về loại hình.
bài viết liên quan
Các bước tiếp theo
Theo dõi độ tin cậy giải CAPTCHA của bạn một cách định lượng —lấy khóa API CaptchaAI của bạnvà thực hiện theo dõi ngân sách lỗi.
Hướng dẫn liên quan:
- Mẫu ngắt mạch cho lệnh gọi API CAPTCHA
- Điểm cuối kiểm tra sức khỏe cho nhân viên CAPTCHA
- Theo dõi tỷ lệ giải CAPTCHA bằng Prometheus và Grafana