Trabalhadores de resolução de CAPTCHA com escalonamento automático

Quantos workers de resolução de CAPTCHA sua aplicação realmente precisa às 3h da manhã? Provavelmente uma fração do que precisa no horário de pico. Um pool fixo dimensionado para o pico paga workers ociosos boa parte do dia; dimensionado para o vale, cria fila e estoura o SLA de latência assim que o tráfego sobe.

Neste guia você vai configurar:

• Um autoescalador por threads para chamadas de API puras (o caso mais comum com a CaptchaAI).
• Um autoescalador por processos para quando há pré-processamento de imagem pesado no meio do caminho.
• Uma verificação de saldo que interrompe o scale-up automaticamente antes que a conta fique negativa.

Qual estratégia de escalonamento usar

Antes de escrever qualquer código, vale decidir a estratégia certa para o seu workload — trocar de abordagem depois custa mais retrabalho do que escolher bem na largada.

Como o trabalho de resolução em si é I/O-bound (a CPU fica ociosa esperando a resposta da API), a maioria dos times começa com threads e só migra para processos ou Kubernetes quando o pipeline cresce além de uma única máquina.

Sinais que disparam o escalonamento

Se seus workers rodam fora do Brasil, meça a latência de rede até https://ocr.captchaai.com antes de calibrar o limite de P95. Workers em regiões como sa-east-1 (São Paulo) costumam ter RTT mais baixo para tráfego de usuários brasileiros — o que muda o ponto em que o P95 realmente indica saturação, em vez de simples latência de rede.

Autoescalador baseado em threads

Para cargas dominadas por chamadas de API — que é o caso da CaptchaAI — threads dentro de um único processo Python são a opção mais simples de operar. O pool abaixo faz três coisas:

1. Consome tarefas de uma fila Redis e resolve o CAPTCHA correspondente pela API da CaptchaAI.
2. Mede profundidade da fila e utilização dos workers a cada 10 segundos.
3. Adiciona ou remove threads automaticamente com base nesses dois números.

import os
import time
import threading
import requests
import json
import redis


class AutoScalingPool:
    """Dynamically scale CaptchaAI worker threads."""

    def __init__(self, api_key, redis_url="redis://localhost:6379"):
        self.api_key = api_key
        self.redis = redis.from_url(redis_url)
        self.base = "https://ocr.captchaai.com"
        self.queue_key = "captcha:tasks"
        self.results_key = "captcha:results"

        self.min_workers = 2
        self.max_workers = 20
        self.workers = []
        self.active_count = 0
        self.lock = threading.Lock()
        self.running = True

    def start(self):
        """Start the pool with minimum workers."""
        for _ in range(self.min_workers):
            self._add_worker()

        # Start scaler in background
        scaler = threading.Thread(target=self._scaling_loop, daemon=True)
        scaler.start()
        print(f"Pool started with {self.min_workers} workers")

    def _add_worker(self):
        """Add a worker thread."""
        if len(self.workers) >= self.max_workers:
            return
        t = threading.Thread(target=self._worker_loop, daemon=True)
        t.start()
        self.workers.append(t)

    def _remove_worker(self):
        """Signal one worker to stop (lazy removal)."""
        if len(self.workers) <= self.min_workers:
            return
        self.workers.pop()  # Thread will exit on next idle cycle

    def _worker_loop(self):
        """Worker loop: fetch and process tasks."""
        while self.running and threading.current_thread() in self.workers:
            result = self.redis.blpop(self.queue_key, timeout=10)
            if result is None:
                continue

            _, raw = result
            task = json.loads(raw)
            task_id = task["id"]

            with self.lock:
                self.active_count += 1

            try:
                token = self._solve(task["method"], task["params"])
                self.redis.hset(self.results_key, task_id, json.dumps({
                    "status": "success", "token": token,
                }))
            except Exception as e:
                self.redis.hset(self.results_key, task_id, json.dumps({
                    "status": "error", "error": str(e),
                }))
            finally:
                with self.lock:
                    self.active_count -= 1

    def _scaling_loop(self):
        """Periodically adjust worker count."""
        while self.running:
            time.sleep(10)

            queue_depth = self.redis.llen(self.queue_key)
            current = len(self.workers)
            utilization = (
                self.active_count / current * 100 if current > 0 else 0
            )

            # Scale up: queue growing and workers busy
            if queue_depth > 20 and utilization > 70:
                new_count = min(current + 2, self.max_workers)
                while len(self.workers) < new_count:
                    self._add_worker()
                print(f"Scaled up: {current} → {len(self.workers)} workers")

            # Scale down: queue empty and workers idle
            elif queue_depth < 5 and utilization < 20:
                target = max(current - 1, self.min_workers)
                while len(self.workers) > target:
                    self._remove_worker()
                if len(self.workers) < current:
                    print(f"Scaled down: {current} → {len(self.workers)} workers")

    def _solve(self, method, params, timeout=120):
        data = {"key": self.api_key, "method": method, "json": 1}
        data.update(params)

        resp = requests.post(
            f"{self.base}/in.php", data=data, timeout=30,
        )
        result = resp.json()

        if result.get("status") != 1:
            raise RuntimeError(result.get("request"))

        captcha_id = result["request"]
        start = time.time()

        while time.time() - start < timeout:
            time.sleep(5)
            resp = requests.get(f"{self.base}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": captcha_id,
                "json": 1,
            }, timeout=15)
            data = resp.json()
            if data["request"] != "CAPCHA_NOT_READY":
                if data.get("status") == 1:
                    return data["request"]
                raise RuntimeError(data["request"])

        raise TimeoutError("Solve timeout")

    def stats(self):
        return {
            "workers": len(self.workers),
            "active": self.active_count,
            "queue": self.redis.llen(self.queue_key),
        }


# Usage
pool = AutoScalingPool(os.environ["CAPTCHAAI_KEY"])
pool.start()

# Monitor
while True:
    print(pool.stats())
    time.sleep(30)

O scaler roda em uma thread separada (_scaling_loop) e nunca deixa o pool abaixo de min_workers nem acima de max_workers. Ajuste os dois valores conforme o teto de threads do seu plano CaptchaAI: no plano BASIC (US$ 15/mês, 5 threads), configurar max_workers acima de 5 não aumenta o throughput — as chamadas extras esperam na fila do lado do provedor mesmo assim.

Autoescalador baseado em processos

Quando o worker também faz algo vinculado à CPU antes de chamar a API — recorte de imagem, pré-processamento com OpenCV, normalização de payload — threads não ajudam por causa do GIL do Python. Nesse caso, escale processos:

import multiprocessing
import time
import redis
import os


class ProcessScaler:
    """Scale worker processes based on queue depth."""

    def __init__(self, worker_fn, redis_url="redis://localhost:6379"):
        self.worker_fn = worker_fn
        self.redis = redis.from_url(redis_url)
        self.processes = []
        self.min_workers = 2
        self.max_workers = 16

    def run(self, check_interval=15):
        """Run the scaler loop."""
        # Start minimum workers
        for _ in range(self.min_workers):
            self._spawn()

        while True:
            time.sleep(check_interval)
            self._cleanup_dead()

            queue_depth = self.redis.llen("captcha:tasks")
            current = len(self.processes)

            # Scale up
            if queue_depth > current * 5 and current < self.max_workers:
                to_add = min(
                    max(1, queue_depth // 10),
                    self.max_workers - current,
                )
                for _ in range(to_add):
                    self._spawn()
                print(f"Scaled up to {len(self.processes)} workers")

            # Scale down
            elif queue_depth < 3 and current > self.min_workers:
                to_remove = min(2, current - self.min_workers)
                for _ in range(to_remove):
                    p = self.processes.pop()
                    p.terminate()
                print(f"Scaled down to {len(self.processes)} workers")

    def _spawn(self):
        p = multiprocessing.Process(target=self.worker_fn)
        p.start()
        self.processes.append(p)

    def _cleanup_dead(self):
        self.processes = [p for p in self.processes if p.is_alive()]
        # Ensure minimum
        while len(self.processes) < self.min_workers:
            self._spawn()

Note a diferença de abordagem: em vez de comparar contra thresholds fixos como no exemplo por threads, este scaler dimensiona proporcionalmente à fila (queue_depth // 10), o que reage mais rápido a picos grandes de volume.

Escalonamento com verificação de saldo

Escalar sem olhar o saldo é a forma mais comum de estourar o orçamento durante um pico de tráfego inesperado. Antes de adicionar workers, verifique se a conta ainda tem fôlego:

def check_balance(api_key, min_balance=2.0):
    """Check if balance is sufficient for scaling."""
    resp = requests.get("https://ocr.captchaai.com/res.php", params={
        "key": api_key,
        "action": "getbalance",
        "json": 1,
    }, timeout=15)
    balance = float(resp.json()["request"])

    if balance < min_balance:
        print(f"Balance ${balance:.2f} below ${min_balance} — halting scale-up")
        return False
    return True

Integre a verificação ao loop de escalonamento para pausar o scale-up automaticamente quando o saldo cair abaixo do limite:

# In _scaling_loop:
if queue_depth > 20 and utilization > 70:
    if check_balance(self.api_key, min_balance=2.0):
        # Scale up
        ...
    else:
        print("Scaling paused — low balance")

Checklist antes de colocar em produção

• Defina min_workers e max_workers respeitando o teto de threads do seu plano CaptchaAI.
• Configure a verificação de saldo com uma margem (min_balance) que dê tempo de recarregar antes de zerar.
• Suba rápido, desça devagar — use intervalos de checagem diferentes para scale-up e scale-down.
• Monitore _cleanup_dead() (ou equivalente) para não acumular processos zumbis.
• Registre métricas de fila, utilização e saldo em um painel — não confie só nos print() dos exemplos acima.

Solução de problemas

Perguntas frequentes

O autoscaler funciona com qualquer plano CaptchaAI?

Sim, mas o teto de threads do plano é o limite real de paralelismo. Configurar max_workers acima do número de threads contratado — por exemplo, 50 no plano ADVANCE (US$ 90/mês) — não aumenta o throughput, porque as chamadas extras ficam esperando do lado do provedor. Ajuste max_workers ao plano, não o contrário.

Preciso do Kubernetes para fazer escalonamento automático?

Não. Threads e processos resolvem a maioria dos casos até algumas dezenas de workers. HPA e KEDA fazem sentido quando você já roda em Kubernetes por outros motivos e quer unificar o escalonamento de todos os serviços — não são pré-requisito para escalar workers de CAPTCHA.

Threads ou processos: qual devo escolher primeiro?

Comece com threads — a chamada à API da CaptchaAI é I/O-bound, então threads bastam e são mais simples de operar. Migre para processos só se adicionar pré-processamento de imagem pesado (OpenCV, recorte, normalização) antes de enviar o CAPTCHA para resolução.

Como evito que o autoscaler fique oscilando entre subir e reduzir workers?

Suba rápido e desça devagar: verifique a cada 10 a 15 segundos para decidir um scale-up, mas exija 30 a 60 segundos de carga baixa sustentada antes de reduzir. Isso evita remover um worker que vai ser necessário de novo poucos segundos depois.

Vale a pena registrar o histórico de escalonamento para auditoria (LGPD)?

Sim, mas com cuidado: se os logs de escalonamento guardam metadados de solicitações de usuários (IP, user-agent, timestamps), trate isso como dado pessoal sob a LGPD — defina um tempo de retenção, restrinja o acesso e documente a finalidade. Guardar apenas contadores agregados (fila, workers ativos, saldo) evita esse problema por completo.

Guias relacionados

• Filas de tarefas no Kubernetes para CAPTCHA em escala
• Monitoramento de taxas de resolução com Prometheus e Grafana

Escale com inteligência — crie sua chave CaptchaAI hoje e ajuste a capacidade à demanda real.