Chaos Engineering for CAPTCHA Solving Pipelines

You built circuit breakers, retries, and bulkheads for your CAPTCHA solving pipeline. But do they actually work? Chaos engineering injects controlled failures — network timeouts, API errors, slow responses — into your pipeline to verify that resilience patterns behave correctly before production discovers the gaps.

What to Test

Failure	What breaks	Expected behaviour
API timeout	Submit hangs forever	Retry after timeout, then circuit breaker opens
429 rate limit	Too many requests	Backoff, reduce concurrency
`ERROR_ZERO_BALANCE`	No budget	Stop solving, alert, don't retry
Slow responses	30s+ solve times	Timeout, retry with fresh task
Malformed response	Invalid JSON	Error handling, retry
Network disconnect	Connection refused	Retry with backoff, failover

Python: Chaos Middleware

Intercept API calls and inject failures based on configurable rules.

import requests
import random
import time
import json
from dataclasses import dataclass, field
from unittest.mock import patch

API_KEY = "YOUR_API_KEY"
SUBMIT_URL = "https://ocr.captchaai.com/in.php"
RESULT_URL = "https://ocr.captchaai.com/res.php"


@dataclass
class ChaosRule:
    """Defines a chaos injection rule."""
    name: str
    probability: float  # 0.0 to 1.0
    enabled: bool = True


@dataclass
class ChaosConfig:
    """Configuration for chaos experiments."""
    timeout_probability: float = 0.0
    error_probability: float = 0.0
    slow_response_probability: float = 0.0
    slow_response_delay: float = 15.0
    malformed_probability: float = 0.0
    error_responses: list[str] = field(default_factory=lambda: [
        "ERROR_NO_SLOT_AVAILABLE",
        "ERROR_CAPTCHA_UNSOLVABLE",
        "ERROR_TOO_MUCH_REQUESTS",
    ])


class ChaosProxy:
    """Wraps HTTP requests to inject chaos."""

    def __init__(self, config: ChaosConfig):
        self.config = config
        self.injections: list[dict] = []

    def _maybe_inject(self, probability: float) -> bool:
        return random.random() < probability

    def _log(self, injection_type: str, details: str = ""):
        entry = {"type": injection_type, "time": time.monotonic(), "details": details}
        self.injections.append(entry)
        print(f"[CHAOS] Injected: {injection_type} {details}")

    def intercept_post(self, original_post, url, **kwargs):
        """Intercept POST requests with chaos."""
        if self.config.timeout_probability and self._maybe_inject(self.config.timeout_probability):
            self._log("timeout", f"POST {url}")
            raise requests.Timeout("Chaos: connection timed out")

        if self.config.slow_response_probability and self._maybe_inject(self.config.slow_response_probability):
            delay = self.config.slow_response_delay
            self._log("slow_response", f"{delay}s delay")
            time.sleep(delay)

        if self.config.error_probability and self._maybe_inject(self.config.error_probability):
            error = random.choice(self.config.error_responses)
            self._log("error_response", error)
            # Return a mock response with error
            mock_resp = requests.models.Response()
            mock_resp.status_code = 200
            mock_resp._content = json.dumps({"status": 0, "request": error}).encode()
            return mock_resp

        if self.config.malformed_probability and self._maybe_inject(self.config.malformed_probability):
            self._log("malformed_response")
            mock_resp = requests.models.Response()
            mock_resp.status_code = 200
            mock_resp._content = b"not json at all"
            return mock_resp

        return original_post(url, **kwargs)

    def intercept_get(self, original_get, url, **kwargs):
        """Intercept GET requests with chaos."""
        if self.config.timeout_probability and self._maybe_inject(self.config.timeout_probability):
            self._log("timeout", f"GET {url}")
            raise requests.Timeout("Chaos: connection timed out")

        return original_get(url, **kwargs)

    def report(self) -> dict:
        """Summary of injected failures."""
        types = {}
        for entry in self.injections:
            types[entry["type"]] = types.get(entry["type"], 0) + 1
        return {"total_injections": len(self.injections), "by_type": types}


# --- Run a chaos experiment ---

def run_experiment(solver_fn, chaos_config: ChaosConfig, iterations: int = 20):
    """Run a solver function under chaos conditions."""
    proxy = ChaosProxy(chaos_config)
    original_post = requests.post
    original_get = requests.get

    results = {"success": 0, "failure": 0, "errors": []}

    with patch.object(requests, "post",
                      side_effect=lambda url, **kw: proxy.intercept_post(original_post, url, **kw)):
        with patch.object(requests, "get",
                          side_effect=lambda url, **kw: proxy.intercept_get(original_get, url, **kw)):
            for i in range(iterations):
                try:
                    token = solver_fn()
                    results["success"] += 1
                except Exception as e:
                    results["failure"] += 1
                    results["errors"].append(str(e))

    results["chaos_report"] = proxy.report()
    return results


# --- Example: Test your solver under chaos ---

def my_solver():
    """Your existing solver function."""
    params = {
        "key": API_KEY, "json": 1,
        "method": "turnstile",
        "sitekey": "0x4XXXXXXXXXXXXXXXXX",
        "pageurl": "https://example.com",
    }
    resp = requests.post(SUBMIT_URL, data=params, timeout=30).json()
    if resp.get("status") != 1:
        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(RESULT_URL, 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:
            return poll["request"]
        raise RuntimeError(f"Solve: {poll.get('request')}")
    raise RuntimeError("Timeout")


# Experiment 1: 20% timeout rate
print("=== Experiment: Timeout Injection ===")
results = run_experiment(
    my_solver,
    ChaosConfig(timeout_probability=0.2),
    iterations=10,
)
print(f"Success: {results['success']}, Failure: {results['failure']}")
print(f"Injections: {results['chaos_report']}")

# Experiment 2: Mixed failures
print("\n=== Experiment: Mixed Chaos ===")
results = run_experiment(
    my_solver,
    ChaosConfig(
        timeout_probability=0.1,
        error_probability=0.15,
        slow_response_probability=0.1,
        slow_response_delay=10.0,
    ),
    iterations=10,
)
print(f"Success: {results['success']}, Failure: {results['failure']}")

JavaScript: Chaos Wrapper

const API_KEY = "YOUR_API_KEY";
const SUBMIT_URL = "https://ocr.captchaai.com/in.php";
const RESULT_URL = "https://ocr.captchaai.com/res.php";

class ChaosInjector {
  constructor(config = {}) {
    this.config = {
      timeoutRate: config.timeoutRate || 0,
      errorRate: config.errorRate || 0,
      slowRate: config.slowRate || 0,
      slowDelayMs: config.slowDelayMs || 15000,
      errors: config.errors || ["ERROR_NO_SLOT_AVAILABLE", "ERROR_CAPTCHA_UNSOLVABLE"],
    };
    this.injections = [];
  }

  async wrapFetch(originalFetch, url, options = {}) {
    // Timeout injection
    if (Math.random() < this.config.timeoutRate) {
      this.injections.push({ type: "timeout", url });
      throw new Error("Chaos: timeout");
    }

    // Slow response injection
    if (Math.random() < this.config.slowRate) {
      this.injections.push({ type: "slow", url });
      await new Promise((r) => setTimeout(r, this.config.slowDelayMs));
    }

    // Error response injection
    if (Math.random() < this.config.errorRate) {
      const error = this.config.errors[Math.floor(Math.random() * this.config.errors.length)];
      this.injections.push({ type: "error", error });
      return new Response(JSON.stringify({ status: 0, request: error }), {
        status: 200,
        headers: { "Content-Type": "application/json" },
      });
    }

    return originalFetch(url, options);
  }

  report() {
    const byType = {};
    for (const i of this.injections) {
      byType[i.type] = (byType[i.type] || 0) + 1;
    }
    return { total: this.injections.length, byType };
  }
}

async function runExperiment(solverFn, chaosConfig, iterations = 10) {
  const injector = new ChaosInjector(chaosConfig);
  const originalFetch = globalThis.fetch;
  globalThis.fetch = (url, opts) => injector.wrapFetch(originalFetch, url, opts);

  const results = { success: 0, failure: 0, errors: [] };

  for (let i = 0; i < iterations; i++) {
    try {
      await solverFn();
      results.success++;
    } catch (e) {
      results.failure++;
      results.errors.push(e.message);
    }
  }

  globalThis.fetch = originalFetch;
  results.chaosReport = injector.report();
  return results;
}

// Run experiment
const results = await runExperiment(
  () => solveCaptcha({ method: "turnstile", sitekey: "SITEKEY", pageurl: "https://example.com" }),
  { timeoutRate: 0.2, errorRate: 0.1 },
  10
);
console.log(`Success: ${results.success}, Failure: ${results.failure}`);
console.log("Injections:", results.chaosReport);

Chaos Experiment Checklist

Experiment	Inject	Verify
Timeout at submit	30% timeout on POST	Retries succeed within budget
Timeout at poll	30% timeout on GET	Fresh task submitted on timeout
Rate limit storm	50% return 429	Backoff slows retries, no storm
Balance exhausted	Return `ERROR_ZERO_BALANCE`	Stops immediately, alerts fired
Slow API	15s delay on responses	Timeout kicks in, retries work
Total outage	100% connection refused	Circuit breaker opens, graceful degradation

Troubleshooting

Issue	Cause	Fix
Chaos affects production	No environment check	Gate chaos behind `CHAOS_ENABLED=true` env var
Monkey-patching leaks across tests	Global state not restored	Use context managers or try/finally to restore originals
False positives in results	Chaos rate too high	Start at 5–10% injection rate, increase gradually
Experiment not reproducible	Random seed not set	Set `random.seed()` for reproducible experiments
Real API calls during chaos	Not all paths intercepted	Verify all HTTP methods are wrapped

FAQ

Should I run chaos experiments against the live CaptchaAI API?

Run chaos at the HTTP layer so failures are injected before reaching the API. This tests your error handling without consuming API credits. Only test against the live API when you specifically need to verify end-to-end behaviour under normal conditions.

How often should I run chaos experiments?

Run them after every change to retry logic, circuit breakers, or error handling code. Add them to your CI pipeline as integration tests. For ongoing monitoring, run weekly experiments against staging environments.

What's the right failure injection rate?

Start at 5–10% for exploratory tests. Increase to 20–30% to stress-test specific patterns. 50%+ is useful for verifying circuit breaker thresholds. Never run 100% injection in production — that's not chaos engineering, that's an outage.