التعافي من الكوارث في مسارات حل اختبارات CAPTCHA

التعافي من الكوارث في مسار حل CAPTCHA يعني هدفًا عمليًا واحدًا: حين يتعطّل عامل أو تنقطع الشبكة أو تتلف التهيئة، تستأنف المعالجة من حيث توقفت دون فقدان مهمة واحدة ودون إنتاج حلول مكررة. تتحقق هذه النتيجة عبر ثلاث ركائز واضحة — تخزين المهام بشكل دائم، وحفظ نقاط تحقّق (checkpoints) دورية، وكتاب تشغيل قابل للتنفيذ الآلي — لا عبر خطة نظرية تُحفظ في درج ولا تُختبر أبدًا.

الخطأ الأكثر شيوعًا هو معاملة التعافي من الكوارث كسيناريو "يوم القيامة" الذي يقع مرة كل بضع سنوات. في الواقع التشغيلي، لا يلزم زلزال حتى ينهار المسار: يكفي إعادة تشغيل حاوية واحدة في توقيت سيئ، أو انقطاع اتصال لبضع ثوانٍ، أو مفتاح API مكشوف، حتى يتراكم backlog من المهام، أو تُعاد معالجة اختبارات محلولة أصلًا، أو تختفي مهام لم تُخزَّن بعد بشكل دائم. لذلك يبدأ التصميم السليم من افتراض أن الأعطال ستقع، ثم يبني حولها الحواجز التي تحتوي أثرها.

حين يتحوّل عطل صغير إلى كارثة

تخيّل فريق أتمتة في منطقة الخليج يشغّل عمليات جمع بيانات على مدار الساعة خلال موسم ذروة مثل الجمعة البيضاء. في ساعة الذروة يرتفع حجم الطلبات بشكل حاد، وتتراكم آلاف المهام في الطابور خلال دقائق. لو انقطع الاتصال بمزوّد الحل لدقيقتين فقط بينما المهام محفوظة في الذاكرة وحدها، فكل مهمة "قيد التنفيذ" تضيع — ومعها الوقت والرصيد المستهلك عليها بالفعل. الأسوأ أن العودة المتسرّعة قد تعيد إرسال المهام نفسها، فتتضاعف التكلفة وتظهر حلول مكررة في قاعدة البيانات.

سرعة التعافي بعد مثل هذا العطل ترتبط مباشرة بعدد الخيوط (threads) المتاحة في خطتك، لأنها تحدّد كم مهمة يمكن تصريفها بالتوازي أثناء استئناف الطابور. فخطة مثل ADVANCE ($90 شهريًا، 50 خيطًا) تسمح باستئناف أسرع بكثير من BASIC ($15 شهريًا، 5 خيوط) عند وجود آلاف المهام المتراكمة، لأن عدد المهام المتزامنة أعلى. اختيار سعة الخيوط جزء من خطة التعافي، وليس مجرد قرار أداء يومي.

حدّد أهدافك: RPO وRTO وMTTR

قبل كتابة أي منطق تعافٍ، عرّف ثلاثة أرقام تحكم كل قرار لاحق. من دونها تصبح خطة التعافي مجرد نوايا حسنة بلا معيار للنجاح أو الفشل.

القاعدة العملية: كلما شددت هدف RPO، زادت الحاجة إلى حفظ نقاط تحقّق أكثر تكرارًا، وهو ما يرفع حِمل الكتابة على القرص. اضبط الرقم على قيمة تخدم عملك فعلًا، لا على رقم مثالي يرهق النظام دون فائدة تُذكر.

سيناريوهات الفشل الخمسة

لكل نوع عطل مسار تعافٍ مختلف، ولا يوجد إجراء واحد يعالجها جميعًا. ابدأ بتصنيف ما قد يحدث فعلًا وربط كل حالة بردّ فعل محدد مسبقًا:

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

تعطّل العامل هو الأكثر شيوعًا وأبسطها علاجًا: أعد تشغيل العمال وأعد تشغيل الطابور. أما فقدان بيانات الطابور فيكشف ما إذا كان التخزين دائمًا فعلًا. تقسيم الشبكة يختبر جاهزية المنطقة الثانوية، وكشف المفتاح يتطلب تدوير المفتاح فورًا، وتلف التهيئة يُعالَج بالعودة إلى آخر نسخة سليمة معروفة.

طبقة تخزين المهام الدائم

القاعدة الذهبية: لا تحلّ اختبار CAPTCHA انطلاقًا من طابور موجود في الذاكرة وحدها. أي مهمة غير مكتوبة على قرص أو في قاعدة بيانات دائمة هي مهمة معرّضة للضياع عند أول إعادة تشغيل. الطبقة الدائمة هي ما يمنح خطة التعافي معنى ملموسًا.

Python — قائمة انتظار دائمة تنجو من الأعطال

يستخدم المثال التالي قاعدة SQLite لتخزين كل مهمة مع حالتها وعدد محاولاتها. الأهم فيه هو الدالة recover_stale التي تُستدعى عند بدء التشغيل: فهي تعيد أي مهمة عالقة في حالة processing بعد تجاوز المهلة إلى حالة pending، فتضمن ألا تبقى مهمة معلّقة إلى الأبد بسبب انهيار العامل الذي كان يعالجها.

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")

بهذا التصميم، يصبح إعادة تشغيل الخدمة عملية آمنة: تُقرأ الحالة من القرص، وتُستأنف المهام المعلّقة تلقائيًا، ولا يعتمد شيء على ذاكرة العملية التي انهارت.

JavaScript — مدير الاسترداد ونقاط التحقّق

في بيئة Node.js، يعتمد النمط التالي على ملفات نقاط تحقّق دورية بدل قاعدة بيانات. يحفظ DisasterRecoveryManager لقطات للتقدّم على مراحل، ويحتفظ بآخر عشر نسخ فقط، بينما يقارن ResilientSolver عند الاستئناف بين المهام التي بدأت والمهام التي اكتملت ليعرف بالضبط ما تبقّى — وهو ما يمنع إعادة حل ما سبق حله.

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

لاحظ الدالة healthCheck التي تستعلم عن الرصيد عبر res.php: استخدمها كمؤشر بسيط على صحة الاتصال بمزوّد الحل قبل استئناف دفعة كبيرة، حتى لا تبدأ المعالجة بينما الخدمة ما زالت غير متاحة.

كتاب التشغيل (Runbook) القابل للتنفيذ

الخطة المكتوبة نثرًا لا تُنقذ أحدًا في الثالثة صباحًا. حوّل التعافي إلى كتاب تشغيل مرقّم يمكن لأي مهندس مناوب اتباعه خطوة بخطوة، ويفضّل أن تكون أغلب خطواته قابلة للتنفيذ الآلي:

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

المراحل الخمس — الكشف، ثم التقييم، ثم الاسترداد، ثم التحقق، ثم مراجعة ما بعد الحادث — تشكّل دورة كاملة. لا تتوقف عند إعادة الخدمة؛ فمرحلة ما بعد الحادث هي التي تمنع تكرار الكارثة نفسها مرة أخرى.

معالجة المشكلات الشائعة

الأسئلة الشائعة

ما الفرق بين RPO وRTO ولماذا يهمّان في مسار CAPTCHA؟

يقيس RPO كمية العمل التي تقبل خسارتها (كم دقيقة من المهام في الطابور)، بينما يقيس RTO المدة التي تحتاجها لإعادة الخدمة. الأول يحدد كثافة نقاط التحقّق، والثاني يحدد مدى أتمتة كتاب التشغيل. تحديد الرقمين بوضوح يحوّل التعافي من تقدير مبهم إلى معيار قابل للقياس.

كيف أتجنّب الحلول المكررة بعد استئناف المعالجة؟

اعتمد على معرّف مهمة ثابت (idempotency key) واكتب حالة كل مهمة قبل إرسالها للحل. عند الاستئناف، تحقق أولًا مما إذا كانت المهمة قد اكتملت فعلًا قبل إعادة إرسالها. هذا يمنع استهلاك الرصيد مرتين على المهمة نفسها ويحافظ على نظافة قاعدة البيانات.

كيف أختبر خطة التعافي قبل وقوع العطل الفعلي؟

نفّذ تمارين فشل مُتحكَّم بها على جدول منتظم: أوقف عاملًا عمدًا، أو اقطع الاتصال بالمنطقة الأساسية، وراقب هل تحقق أهداف RPO وRTO فعلًا. المنطقة الثانوية التي لا تُختبر إلا وقت الأزمة تفشل غالبًا وقت الأزمة. الاختبار الدوري هو الفرق بين خطة على الورق وخطة تعمل.

ما الذي يجب مراقبته لاكتشاف الكارثة مبكرًا؟

راقب ثلاثة مؤشرات على الأقل: عمق الطابور (هل ينمو بلا تصريف؟)، وحالة العمال (كم عاملًا متصلًا؟)، ومعدل الأخطاء القادمة من الـ API. تنبيه مبكر على أي منها يمنحك دقائق ثمينة قبل أن يتحول التراكم إلى فقدان بيانات.

الخطوات التالية

• ابدأ سريعًا مع CaptchaAI: حلّ أول اختبار في 5 دقائق
• حلّ reCAPTCHA v2 عبر الـ API خطوة بخطوة
• حلّ Cloudflare Turnstile عبر الـ API
• حلّ GeeTest v3 عبر الـ API

أدلة ذات صلة

• ابدأ سريعًا مع CaptchaAI: حلّ أول اختبار في 5 دقائق
• منع تكرار حلول CAPTCHA عبر قفل قاعدة البيانات