Planificación de disaster recovery para pipelines de resolución de CAPTCHA

El disaster recovery de un pipeline de resolución de CAPTCHA se mide por una sola pregunta: cuando algo se cae —un worker, la red, tu proveedor— ¿cuántas tareas en curso pierdes? Con una cola persistente y checkpoints frecuentes, la respuesta debería ser cero o casi cero. Este artículo traduce esa meta en objetivos medibles (RPO, RTO, MTTR), una capa de persistencia con SQLite o Redis, y un runbook de failover que tu guardia pueda seguir sin improvisar.

Fija tus objetivos: RPO, RTO y MTTR

Antes de escribir una línea de recuperación, ponle números a lo que consideras aceptable. Estas tres métricas convierten "que no se pierda nada" en objetivos que puedes verificar:

El RPO manda sobre tu persistencia; el RTO, sobre tu infraestructura de failover.

De los objetivos a decisiones de diseño

Cada objetivo exige una decisión técnica y una comprobación que confirme que funciona:

Cataloga tus escenarios de falla

No puedes recuperarte de lo que no anticipaste. En un pipeline de CAPTCHA, cinco escenarios cubren la mayoría de los incidentes reales:

Scenario 1: Worker crash         → Restart workers, replay queue
Scenario 2: Queue data loss      → Restore from persistent backup
Scenario 3: Network partition    → Failover to secondary region
Scenario 4: API key compromised  → Rotate key, update workers
Scenario 5: Config corruption    → Rollback to last known good

Cada fila empareja un fallo con su respuesta. La clave: ninguna respuesta debe depender de que un humano recuerde qué hacer bajo presión, así que automatízalas o documéntalas paso a paso en el runbook.

Persiste cada tarea antes de resolverla

La regla de oro: nunca resuelvas CAPTCHAs desde una cola que vive solo en memoria. Si el proceso muere, esas tareas desaparecen y con ellas el RPO que prometiste. Persiste cada tarea en almacenamiento durable antes de enviarla a la API, marca su estado (pending, processing, completed, failed) y registra los intentos para reintentar sin duplicar.

La implementación mínima cabe en SQLite: una tabla de tareas y métodos para encolar, tomar la siguiente y rescatar las que quedaron a medias. Ese último —recover_stale()— es el corazón del disaster recovery: al arrancar, devuelve a la cola cualquier tarea atascada demasiado tiempo en processing.

Python: cola de tareas persistente con SQLite

import os
import json
import time
import sqlite3
import threading
import requests
from datetime import datetime

API_KEY = os.environ["CAPTCHAAI_API_KEY"]


class PersistentTaskQueue:
    """SQLite-backed task queue that survives crashes."""

    def __init__(self, db_path="captcha_tasks.db"):
        self.db_path = db_path
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.lock = threading.Lock()
        self._init_db()

    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS tasks (
                id TEXT PRIMARY KEY,
                payload TEXT NOT NULL,
                status TEXT DEFAULT 'pending',
                created_at TEXT DEFAULT CURRENT_TIMESTAMP,
                started_at TEXT,
                completed_at TEXT,
                result TEXT,
                attempts INTEGER DEFAULT 0
            )
        """)
        self.conn.commit()

    def enqueue(self, task_id, payload):
        with self.lock:
            self.conn.execute(
                "INSERT INTO tasks (id, payload) VALUES (?, ?)",
                (task_id, json.dumps(payload))
            )
            self.conn.commit()

    def dequeue(self):
        with self.lock:
            cursor = self.conn.execute(
                "SELECT id, payload FROM tasks "
                "WHERE status = 'pending' ORDER BY created_at LIMIT 1"
            )
            row = cursor.fetchone()
            if not row:
                return None

            task_id, payload = row
            self.conn.execute(
                "UPDATE tasks SET status = 'processing', "
                "started_at = ?, attempts = attempts + 1 WHERE id = ?",
                (datetime.utcnow().isoformat(), task_id)
            )
            self.conn.commit()
            return {"id": task_id, "payload": json.loads(payload)}

    def complete(self, task_id, result):
        with self.lock:
            self.conn.execute(
                "UPDATE tasks SET status = 'completed', "
                "completed_at = ?, result = ? WHERE id = ?",
                (datetime.utcnow().isoformat(), json.dumps(result), task_id)
            )
            self.conn.commit()

    def fail(self, task_id, error):
        with self.lock:
            # Requeue if under retry limit
            cursor = self.conn.execute(
                "SELECT attempts FROM tasks WHERE id = ?", (task_id,)
            )
            row = cursor.fetchone()
            if row and row[0] < 3:
                self.conn.execute(
                    "UPDATE tasks SET status = 'pending' WHERE id = ?",
                    (task_id,)
                )
            else:
                self.conn.execute(
                    "UPDATE tasks SET status = 'failed', "
                    "result = ? WHERE id = ?",
                    (json.dumps({"error": error}), task_id)
                )
            self.conn.commit()

    def recover_stale(self, timeout_seconds=600):
        """Reset tasks stuck in 'processing' after a crash."""
        with self.lock:
            cutoff = datetime.utcnow().timestamp() - timeout_seconds
            self.conn.execute(
                "UPDATE tasks SET status = 'pending' "
                "WHERE status = 'processing' "
                "AND started_at < datetime(?, 'unixepoch')",
                (cutoff,)
            )
            count = self.conn.total_changes
            self.conn.commit()
            return count

    @property
    def stats(self):
        cursor = self.conn.execute(
            "SELECT status, COUNT(*) FROM tasks GROUP BY status"
        )
        return dict(cursor.fetchall())


# On startup: recover tasks that were processing during a crash
queue = PersistentTaskQueue()
recovered = queue.recover_stale(timeout_seconds=600)
print(f"Recovered {recovered} stale tasks after restart")

Checkpoints y verificación de salud en Node.js

Cuando el trabajo llega en lotes, los checkpoints periódicos complementan a la cola: guardas el progreso cada pocas tareas y, tras una interrupción, reanudas desde el último punto en vez de desde cero. El siguiente gestor en Node.js versiona checkpoints en disco, conserva solo los más recientes y expone un health check que consulta el saldo con getbalance para saber si la API de CaptchaAI responde antes de reanudar.

JavaScript: checkpoints y recuperación de lotes

const axios = require("axios");
const fs = require("fs");

const API_KEY = process.env.CAPTCHAAI_API_KEY;

class DisasterRecoveryManager {
  constructor(checkpointDir = "./dr-checkpoints") {
    this.checkpointDir = checkpointDir;
    if (!fs.existsSync(checkpointDir)) {
      fs.mkdirSync(checkpointDir, { recursive: true });
    }
  }

  checkpoint(label, data) {
    const filename = `${this.checkpointDir}/${label}-${Date.now()}.json`;
    fs.writeFileSync(filename, JSON.stringify(data, null, 2));
    this.pruneOldCheckpoints(label, 10); // Keep last 10
    return filename;
  }

  restore(label) {
    const files = fs.readdirSync(this.checkpointDir)
      .filter((f) => f.startsWith(label) && f.endsWith(".json"))
      .sort()
      .reverse();

    if (files.length === 0) return null;
    const latest = fs.readFileSync(
      `${this.checkpointDir}/${files[0]}`, "utf8"
    );
    return JSON.parse(latest);
  }

  pruneOldCheckpoints(label, keep) {
    const files = fs.readdirSync(this.checkpointDir)
      .filter((f) => f.startsWith(label) && f.endsWith(".json"))
      .sort();

    while (files.length > keep) {
      const old = files.shift();
      fs.unlinkSync(`${this.checkpointDir}/${old}`);
    }
  }

  async healthCheck() {
    try {
      const resp = await axios.get("https://ocr.captchaai.com/res.php", {
        params: { key: API_KEY, action: "getbalance", json: 1 },
        timeout: 10000,
      });
      return {
        healthy: resp.data.status === 1,
        balance: parseFloat(resp.data.request || 0),
      };
    } catch (err) {
      return { healthy: false, error: err.message };
    }
  }
}

class ResilientSolver {
  constructor() {
    this.dr = new DisasterRecoveryManager();
    this.pendingTasks = [];
  }

  async solveBatch(tasks) {
    // Checkpoint before starting
    this.dr.checkpoint("batch-pending", {
      tasks,
      startedAt: new Date().toISOString(),
    });

    const results = [];
    for (const task of tasks) {
      try {
        const result = await this.solveSingle(task);
        results.push({ taskId: task.id, ...result });
      } catch (err) {
        results.push({ taskId: task.id, error: err.message });
      }

      // Checkpoint progress periodically
      if (results.length % 10 === 0) {
        this.dr.checkpoint("batch-progress", { results, remaining: tasks.length - results.length });
      }
    }

    // Final checkpoint
    this.dr.checkpoint("batch-complete", { results });
    return results;
  }

  async recover() {
    // Check for incomplete batch
    const progress = this.dr.restore("batch-progress");
    const pending = this.dr.restore("batch-pending");

    if (progress) {
      const completedIds = new Set(progress.results.map((r) => r.taskId));
      const remaining = pending?.tasks.filter((t) => !completedIds.has(t.id));
      console.log(
        `Recovering: ${progress.results.length} done, ${remaining?.length || 0} remaining`
      );
      return remaining || [];
    }

    if (pending) {
      console.log(`Recovering full batch: ${pending.tasks.length} tasks`);
      return pending.tasks;
    }

    return [];
  }

  async solveSingle(task) {
    const resp = await axios.post("https://ocr.captchaai.com/in.php", null, {
      params: {
        key: API_KEY,
        method: "userrecaptcha",
        googlekey: task.sitekey,
        pageurl: task.pageurl,
        json: 1,
      },
    });

    if (resp.data.status !== 1) throw new Error(resp.data.request);

    const captchaId = resp.data.request;
    for (let i = 0; i < 60; i++) {
      await new Promise((r) => setTimeout(r, 5000));
      const poll = await axios.get("https://ocr.captchaai.com/res.php", {
        params: { key: API_KEY, action: "get", id: captchaId, json: 1 },
      });
      if (poll.data.status === 1) return { solution: poll.data.request };
      if (poll.data.request !== "CAPCHA_NOT_READY")
        throw new Error(poll.data.request);
    }
    throw new Error("TIMEOUT");
  }
}

// Start with recovery check
const solver = new ResilientSolver();
solver.recover().then((remaining) => {
  if (remaining.length > 0) {
    console.log(`Resuming ${remaining.length} tasks from checkpoint`);
    solver.solveBatch(remaining);
  }
});

Un runbook que la guardia pueda seguir de madrugada

El código evita perder tareas; el runbook evita perder tiempo. Un buen runbook no explica teoría: da los comandos exactos para detectar, evaluar, recuperar y verificar. Escríbelo pensando en la persona con menos contexto de tu equipo, no en quien diseñó el sistema.

RUNBOOK: CAPTCHA Pipeline Recovery
====================================

1. DETECT
   - Alert fires: [PagerDuty / Slack / Email]
   - Symptom: [Queue growing / Workers offline / Error spike]

2. ASSESS
   - Check worker health: curl http://workers/health
   - Check API status: GET /res.php?action=getbalance
   - Check queue depth: SELECT COUNT(*) FROM tasks WHERE status='pending'

3. RECOVER
   If: Workers crashed
     → Restart worker containers: docker-compose up -d workers
     → Run stale task recovery: recovery.py --recover-stale

   If: Network partition
     → Failover to secondary region
     → Update DNS or load balancer routing

   If: API key compromised
     → Generate new key at captchaai.com
     → Update secret store
     → Rolling restart workers

4. VERIFY
   - Confirm solve rate > 90%
   - Confirm queue draining
   - Confirm no duplicate solves

5. POST-MORTEM
   - Document root cause
   - Update runbook if needed

Tras cada incidente, cierra el ciclo con un post-mortem breve: causa raíz, qué falló en la detección y qué línea del runbook corregir. Un runbook desactualizado envejece más rápido que tu infraestructura.

Un escenario real: una agencia que monitorea portales públicos

Imagina una agencia en Ciudad de México que monitorea la disponibilidad de citas en portales públicos —trámites del SAT, citas de visado en centros BLS, cita previa en sedes electrónicas europeas— para sus clientes, respetando los términos de servicio de cada portal y la normativa de protección de datos aplicable. Muchos de esos formularios están protegidos por reCAPTCHA v2 o Cloudflare Turnstile, así que la agencia resuelve el CAPTCHA con la API de CaptchaAI antes de comprobar cada portal.

Una tarde, su proveedor de nube tuvo una caída de zona y los workers se apagaron a mitad de un lote de varios cientos de tareas. Sin persistencia, habrían perdido todo el trabajo de esa franja horaria. Con la cola en SQLite y los checkpoints descritos arriba, al reiniciar los workers recover_stale() devolvió a pending las tareas que habían quedado en processing, y el pipeline retomó justo donde se detuvo, sin volver a resolver lo ya completado. Su costo mensual en USD por thread —predecible frente al pago por resolución— tampoco se movió.

Errores comunes de recuperación y cómo evitarlos

Preguntas frecuentes

¿En qué se diferencian el RPO y el RTO cuando algo falla?

El RPO mide cuánto trabajo puedes perder (datos); el RTO, cuánto tiempo puedes estar caído (minutos). Un pipeline puede tener un RTO holgado pero un RPO estricto si cada resolución cuesta cara: ahí priorizas la persistencia sobre la velocidad de arranque.

¿Cómo evito resolver dos veces el mismo CAPTCHA tras un failover?

Asigna a cada tarea una clave de idempotencia estable y guárdala junto a su estado. Antes de reenviar una tarea recuperada, comprueba en tu cola si ya tiene un resultado; si lo tiene, la saltas. Así el replay tras una caída no duplica solicitudes ni gasto.

¿Puedo ensayar el plan de recuperación sin tocar producción?

Sí, y deberías. Monta un staging con una copia de la cola, provoca un fallo controlado (mata los workers, corta la red) y cronometra tu RTO real. Estos "game days" revelan los pasos manuales que faltan en el runbook.

¿Necesito una segunda región para tener failover?

No siempre. Para muchos pipelines basta con workers redundantes en la misma región más una cola persistente respaldada. La segunda región cobra sentido cuando tu RTO no tolera la caída de una zona completa; ahí, mantén bajo el TTL de DNS y ensaya el cambio.

Guías relacionadas

• Alta disponibilidad y failover para resolución de CAPTCHA
• Arquitectura multirregional para pipelines de CAPTCHA
• Cómo implementar lógica de reintento con la API de CaptchaAI