Solving CAPTCHAs in Background Job Queues: Celery, Sidekiq, BullMQ and River

This guide shows how to run CaptchaAI solves as background jobs in Celery with a Redis broker, Sidekiq, BullMQ and River without the four failures queue code usually hits: oversubscribed threads, workers parked in sleep(), retries of errors that can never succeed, and the same CAPTCHA solved twice after a redelivery. One constraint shapes every design choice. CaptchaAI bills per concurrent thread, so the number of solves in flight across every worker process, host and deploy must stay within your plan's thread count. The examples assume forms and flows you own or are authorized to automate, such as QA runs against your own staging site.

The four rules for CAPTCHA work in any job queue

1. Cap solves in flight at plan threads, across all processes. A CaptchaAI thread is one CAPTCHA in flight, busy from the moment in.php accepts the task until the answer is ready. Celery's --concurrency, a Sidekiq capsule's concurrency and River's MaxWorkers apply to one worker instance, so three instances at 5 each run 15 solves against a 5-thread plan. Only a shared counter, or BullMQ's global concurrency, holds the total.
2. Classify the error before the queue retries it. A wrong key or a missing pageurl fails identically every time, and resending bad credentials in a loop earns an IP_BANNED. Send permanent codes straight to the dead-letter path and retry only thread saturation and server errors, with exponential backoff.
3. Poll inside the job when the token belongs to a session. If a browser page or cookie jar must submit the token, the job holding that session waits for the answer. Reschedule the poll only when the token alone travels onward.
4. Make every job safe to run twice. Every queue here redelivers under some condition: a killed process, an expired lock, a shutdown timeout. Store the CaptchaAI task ID the moment in.php returns it, keyed by a business ID, resume polling on the next run, and discard tokens past their validity window.

CaptchaAI contract and error taxonomy

Every worker below implements the same two calls on one host, https://ocr.captchaai.com:

Two details catch queue code out. The polling string is CAPCHA_NOT_READY, with no T after "CAP". And json=1 does not promise JSON: the API can still answer with a plain-text error code or, rarely, an HTML 500 or 502 page, so treat anything that is neither an answer nor a code as transient.

The documented error codes fall into eight classes, and the class decides what the queue does:

The error codes reference explains each code. In every example below, the credential, request, image, proxy and unsolvable classes take the queue's no-retry path, threads and transient errors take backoff, and not-ready takes a poll.

Pattern chooser: poll inside the job or reschedule the poll

How the job waits for the answer decides how you cap concurrency:

The fourth row is the one teams miss. Rescheduling the poll saves your workers, not CaptchaAI threads: the thread stays busy until the answer is ready whether or not a worker waits. Once the poll leaves the worker, a process with --concurrency=5 can have 40 solves outstanding, so the reschedule pattern always needs an in-flight counter in shared storage.

Celery with a Redis broker

Put the CaptchaAI calls in a small module that turns every response into one of two exception types, so Celery's autoretry_for retries only what can succeed on a second try:

# captchaai_client.py: shared by the Celery tasks below
from __future__ import annotations

import os
import re

import requests

API = "https://ocr.captchaai.com"
API_KEY = os.environ["CAPTCHAAI_API_KEY"]
RETRYABLE = {"ERROR_ZERO_BALANCE", "ERROR_SERVER_ERROR", "ERROR_INTERNAL_SERVER_ERROR"}
CODE = re.compile(r"^[A-Z_]+$")


class CaptchaTransient(Exception):
    """No free thread, server error or HTML error page: retry with backoff."""


class CaptchaPermanent(Exception):
    """Key, request, image, proxy or unsolvable error: resending unchanged cannot work."""


def _call(path: str, fields: dict) -> dict:
    resp = requests.post(f"{API}/{path}", data={"key": API_KEY, "json": 1, **fields}, timeout=30)
    text = resp.text.strip()
    # json=1 is not a promise: plain-text codes and HTML 5xx pages still happen.
    return resp.json() if text.startswith("{") else {"status": 0, "request": text}


def _raise_for(code) -> None:
    code = str(code)
    if code in RETRYABLE or not CODE.match(code):
        raise CaptchaTransient(code[:80])
    raise CaptchaPermanent(code)


def submit(fields: dict) -> str:
    body = _call("in.php", fields)
    if body.get("status") != 1:
        _raise_for(body.get("request"))
    return str(body["request"])  # task IDs can arrive as JSON numbers


def get_result(captcha_id: str) -> dict | None:
    """The answer once solved, or None while the task is CAPCHA_NOT_READY."""
    body = _call("res.php", {"action": "get", "id": captcha_id})
    if body.get("status") == 1:
        # reCAPTCHA Enterprise and Cloudflare Challenge answer in "result" plus "user_agent".
        return {"token": body.get("result") or body.get("request"),
                "user_agent": body.get("user_agent", "")}
    if body.get("request") == "CAPCHA_NOT_READY":
        return None
    _raise_for(body.get("request"))
    return None

The poll-inside-the-job version is the right default when the token has to be submitted by the same HTTP session. This task loads your own form, solves its reCAPTCHA v2 and posts the token with the session's cookies:

# tasks.py: run with  celery -A tasks worker -Q captcha --concurrency=5 -l INFO
import os
import time

import redis
import requests
from celery import Celery

import captchaai_client as cai
from captchaai_client import CaptchaTransient

REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
app = Celery("tasks", broker=REDIS_URL)
app.conf.update(
    task_routes={"tasks.*": {"queue": "captcha"}},
    worker_prefetch_multiplier=1,     # reserve one message per process, not four
    task_acks_late=True,              # ack when the task ends, not when it starts
    task_reject_on_worker_lost=True,  # requeue if the process dies mid-solve
    broker_transport_options={"visibility_timeout": 3600},  # > time_limit + longest backoff
)
store = redis.Redis.from_url(REDIS_URL, decode_responses=True)
FIRST_WAIT, POLL_CAP = 15, 120


def solve(job_key: str, fields: dict) -> dict:
    """Submit once per job_key; a redelivered job resumes polling the same task."""
    key = f"captcha:{job_key}"
    state = store.hgetall(key)
    if not state or time.time() - float(state["at"]) > FIRST_WAIT + POLL_CAP:
        state = {"id": cai.submit(fields), "at": time.time()}
        store.hset(key, mapping=state)
        store.expire(key, 600)
    deadline = float(state["at"]) + FIRST_WAIT + POLL_CAP
    time.sleep(max(0.0, float(state["at"]) + FIRST_WAIT - time.time()))
    while time.time() < deadline:
        try:
            answer = cai.get_result(state["id"])
        except cai.CaptchaPermanent:
            store.delete(key)  # e.g. ERROR_CAPTCHA_UNSOLVABLE: never poll this ID again
            raise
        if answer is not None:
            store.delete(key)
            return answer
        time.sleep(5)
    store.delete(key)
    raise CaptchaTransient("no answer after 120 s of polling")


@app.task(bind=True, acks_late=True,
          autoretry_for=(CaptchaTransient, requests.RequestException),
          retry_backoff=10, retry_backoff_max=120, retry_jitter=True, max_retries=3,
          soft_time_limit=150, time_limit=180)
def submit_contact_form(self, job_key: str, page_url: str, sitekey: str, form: dict) -> int:
    """Authorized QA check of your own form: one session loads, solves and submits."""
    if store.get(f"done:{job_key}"):
        return 0  # a redelivered copy of a job that already finished
    with requests.Session() as http:
        http.get(page_url, timeout=30)  # cookies the final POST has to carry
        answer = solve(job_key, {"method": "userrecaptcha", "googlekey": sitekey, "pageurl": page_url})
        resp = http.post(page_url, data={**form, "g-recaptcha-response": answer["token"]}, timeout=30)
    store.set(f"done:{job_key}", resp.status_code, ex=86400)
    return resp.status_code

What each setting buys you:

• task_routes to a captcha queue. A dedicated worker started with -Q captcha --concurrency=5 owns the whole CAPTCHA budget; other tasks cannot starve it or push it over.
• worker_prefetch_multiplier=1 with late acks. The default multiplier is 4. With late acks and a multiplier of 1, Celery's optimization guide says a worker reserves only as many tasks as it has processes, so no process hoards messages behind a two-minute solve.
• task_acks_late and task_reject_on_worker_lost. The message is acknowledged after the task finishes and requeued if the process is killed. That is where redeliveries come from, and why solve() resumes from the task ID stored in Redis.
• The Redis visibility_timeout. It defaults to one hour, and a message not acknowledged within it goes to another worker. The Celery Redis broker caveats warn that ETA, countdown and retry tasks outlasting it run "again, and again in a loop". Keep it above time_limit plus your longest retry delay (180 + 120 s here); lowering it to speed up crash recovery turns slow solves into double solves.
• The decorator. autoretry_for lists only transient classes, so a wrong key fails once. retry_backoff=10 sets the three retry delays to 10, 20 and 40 s, and retry_backoff_max=120 bounds them if you raise max_retries. retry_jitter=True (Celery's default) treats each value as a ceiling and picks a random delay between zero and it, so workers that failed together do not retry together. soft_time_limit=150 covers the 15 s wait and 120 s of polling, with 15 s left for the page load and the form POST.

For batches, group(submit_contact_form.s(...) for ...) just enqueues many tasks that the queue's concurrency paces. A chord needs a result backend, and a task must never call .get() on another task's result.

When the token only travels onward, for example to an API that accepts it without a browser session, the reschedule pattern frees each process between polls. Processes then no longer bound the solves in flight, so a Redis sorted set does:

# Append to tasks.py: for stateless flows where only the token travels onward.
import json

INFLIGHT = "captcha:inflight"
MAX_INFLIGHT = int(os.environ.get("CAPTCHAAI_THREADS", "5"))


@app.task(bind=True, acks_late=True,
          autoretry_for=(CaptchaTransient, requests.RequestException),
          retry_backoff=10, retry_backoff_max=120, retry_jitter=True, max_retries=3)
def submit_captcha(self, job_key: str, fields: dict, busy: int = 0) -> None:
    if store.exists(f"captcha-token:{job_key}"):
        return  # already solved; this is a redelivered copy
    captcha_id = store.get(f"captcha-id:{job_key}")
    if captcha_id is None:
        store.zremrangebyscore(INFLIGHT, 0, time.time() - 180)  # slots leaked by crashes
        if store.zcard(INFLIGHT) >= MAX_INFLIGHT:
            if busy >= 24:  # two minutes without a free slot: someone should check the plan
                raise cai.CaptchaPermanent("no free thread for 120 s")
            submit_captcha.apply_async((job_key, fields, busy + 1), countdown=5)
            return
        store.zadd(INFLIGHT, {job_key: time.time()})
        try:
            captcha_id = cai.submit(fields)
        except Exception:
            store.zrem(INFLIGHT, job_key)
            raise
        store.set(f"captcha-id:{job_key}", captcha_id, ex=600)
    poll_captcha.apply_async((job_key, captcha_id), countdown=15)


@app.task(bind=True, acks_late=True, max_retries=24)  # 24 polls x 5 s = the 120 s cap
def poll_captcha(self, job_key: str, captcha_id: str) -> None:
    try:
        answer = cai.get_result(captcha_id)
    except (CaptchaTransient, requests.RequestException) as exc:
        raise self.retry(exc=exc, countdown=10)
    except cai.CaptchaPermanent:
        store.zrem(INFLIGHT, job_key)
        store.delete(f"captcha-id:{job_key}")
        raise
    if answer is None:
        if self.request.retries >= self.max_retries:
            # The 120 s cap: free the slot and forget the ID so a new submit starts fresh.
            store.zrem(INFLIGHT, job_key)
            store.delete(f"captcha-id:{job_key}")
            raise cai.CaptchaPermanent("no answer after 120 s of polling")
        raise self.retry(countdown=5)  # CAPCHA_NOT_READY
    store.zrem(INFLIGHT, job_key)
    store.delete(f"captcha-id:{job_key}")
    # The consumer reads this key; the TTL drops the token before reCAPTCHA's two-minute expiry.
    store.set(f"captcha-token:{job_key}", json.dumps(answer), ex=90)

self.retry(countdown=5) re-sends the task through the broker under the same task ID, so max_retries=24 doubles as the 120-second cap. The task checks self.request.retries itself on the last poll instead of letting Celery raise MaxRetriesExceededError, so the slot is freed at once rather than at the 180-second sweep. The busy argument does the same for thread waits: after 24 checks five seconds apart, the job fails instead of circling forever. A redelivered submit_captcha finds the stored task ID and schedules a second poll chain: duplicate polls are harmless, duplicate submits are not. The count-then-add on the sorted set is not atomic, so processes can briefly overshoot; ERROR_ZERO_BALANCE is the backstop and takes the transient path. For a plain-Python pool without Celery, see building a CAPTCHA solving queue in Python; for calling these tasks from a web app, CAPTCHA handling in Django.

Sidekiq (capsules, retry classification, perform_in re-poll)

Sidekiq threads are cheaper than Celery processes, but the same arithmetic applies: concurrency is per process. This version reschedules the poll with perform_in, and because that releases the thread, it asks CaptchaAI's threadsinfo action before each submission. That counter sees every client on the API key, across all processes and all apps.

# app/jobs/captcha_jobs.rb
require "json"
require "net/http"
require "sidekiq"

module CaptchaAI
  RETRYABLE = %w[ERROR_ZERO_BALANCE ERROR_SERVER_ERROR ERROR_INTERNAL_SERVER_ERROR].freeze
  Permanent = Class.new(StandardError)
  Transient = Class.new(StandardError)

  def self.call(path, params)
    uri = URI("https://ocr.captchaai.com#{path}")
    req = Net::HTTP::Post.new(uri)
    req.set_form_data(params.merge("key" => ENV.fetch("CAPTCHAAI_API_KEY"), "json" => "1"))
    res = Net::HTTP.start(uri.host, uri.port, use_ssl: true, open_timeout: 10, read_timeout: 30) do |http|
      http.request(req)
    end
    text = res.body.to_s.strip
    text.start_with?("{") ? JSON.parse(text) : { "status" => 0, "request" => text }
  end

  def self.raise_for(code)
    code = code.to_s
    raise Transient, code[0, 80] if RETRYABLE.include?(code) || !code.match?(/\A[A-Z_]+\z/)

    raise Permanent, code
  end

  # threadsinfo counts every client on the key, so this gate sees all processes.
  def self.thread_free?
    info = call("/res.php", "action" => "threadsinfo")
    raise_for(info["request"]) unless info.key?("threads")
    info["working_threads"].to_i < info["threads"].to_i
  end
end

class CaptchaSubmitJob
  include Sidekiq::Job
  sidekiq_options queue: "captcha", retry: 5

  sidekiq_retry_in do |count, exception, _job|
    exception.is_a?(CaptchaAI::Permanent) ? :kill : 10 * (2**count)
  end

  sidekiq_retries_exhausted do |job, ex|
    Sidekiq.logger.error("captcha #{job['args'].first} is dead: #{ex.message}")
  end

  def perform(request_id, params, waits = 0)
    key = "captcha:#{request_id}"
    task = Sidekiq.redis { |c| c.call("GET", key) }
    task = nil if task && Time.now.to_i - JSON.parse(task)["at"] > 135 # too old to resume: submit fresh
    unless task
      unless CaptchaAI.thread_free?
        raise CaptchaAI::Permanent, "no free thread for 120 s" if waits >= 12
        return self.class.perform_in(10, request_id, params, waits + 1)
      end

      body = CaptchaAI.call("/in.php", params)
      CaptchaAI.raise_for(body["request"]) unless body["status"] == 1
      task = JSON.generate("id" => body["request"].to_s, "at" => Time.now.to_i)
      Sidekiq.redis { |c| c.call("SET", key, task, "EX", "600") }
    end
    state = JSON.parse(task)
    CaptchaPollJob.perform_in(15, request_id, state["id"], state["at"])
  end
end

class CaptchaPollJob
  include Sidekiq::Job
  sidekiq_options queue: "captcha", retry: 3
  sidekiq_retry_in { |_count, exception, _job| exception.is_a?(CaptchaAI::Permanent) ? :kill : 10 }

  def perform(request_id, captcha_id, submitted_at)
    raise CaptchaAI::Permanent, "no answer after 120 s" if Time.now.to_i - submitted_at > 135

    body = CaptchaAI.call("/res.php", "action" => "get", "id" => captcha_id)
    if body["status"] == 1
      answer = JSON.generate("token" => body["result"] || body["request"], "user_agent" => body["user_agent"])
      # NX: a duplicate poll cannot overwrite; EX: the token vanishes before it expires upstream.
      Sidekiq.redis { |c| c.call("SET", "captcha-token:#{request_id}", answer, "NX", "EX", "90") }
    elsif body["request"] == "CAPCHA_NOT_READY"
      self.class.perform_in(5, request_id, captcha_id, submitted_at)
    else
      CaptchaAI.raise_for(body["request"])
    end
  end
end

# config/initializers/sidekiq.rb
Sidekiq.configure_server do |config|
  # Sidekiq 7.0+: a capsule with its own thread pool that serves only the captcha queue.
  config.capsule("captcha") do |cap|
    cap.concurrency = 5
    cap.queues = %w[captcha]
  end
end

Enqueue one job per business request, for example CaptchaSubmitJob.perform_async("signup-42", { "method" => "turnstile", "sitekey" => sitekey, "pageurl" => page_url }), with string keys so the arguments stay JSON-native.

The Sidekiq-specific parts:

• sidekiq_retry_in returning :kill. A permanent error goes straight to the Dead set instead of walking the job's retries (retry: 5 here; Sidekiq's default is 25 over about 20 days). In Sidekiq's retry code, :kill also runs the sidekiq_retries_exhausted block, so the alert fires on the first bad key; :discard would skip it.
• perform_in(5, ...) as the re-poll. The scheduler checks for due jobs about every 5 s on average (a single process sleeps a random 2.5 to 7.5 s between checks), so a 5-second re-poll lands roughly 5 to 12 s later. Fine for polling, and a reason not to go lower.
• The capsule. Capsules arrived in Sidekiq 7.0 (the current major is 8) and give the captcha queue its own thread count inside each process. Two processes with this capsule run 10 jobs, which is why the threadsinfo gate, not the capsule, protects the plan. The gate is a soft brake: jobs checking in the same second can all see a free thread, and ERROR_ZERO_BALANCE then takes the retry path. A job that finds no free thread twelve times, ten seconds apart, goes to the Dead set, so a saturated or lapsed plan pages someone instead of filling the schedule.
• Shutdown. On TERM, Sidekiq stops fetching and waits for the -t timeout (25 s by default); jobs still running are pushed back to Redis and run again at the next start. The stored task ID turns that rerun into a resume.

To poll inside the job instead, Sidekiq Enterprise's Sidekiq::Limiter.concurrent("captchaai", 5, wait_timeout: 5, lock_timeout: 150) with within_limit around the solve gives a cluster-wide cap; its docs require lock_timeout to exceed the block's run time. Plain Ruby solving outside Sidekiq is covered in solving CAPTCHAs with Ruby.

BullMQ (global concurrency, delayed jobs, UnrecoverableError)

Node is where polling inside the job is cheapest: an awaited setTimeout holds a promise, not a process. That makes BullMQ's global concurrency an exact cluster-wide cap, because an active job is a solve in flight. Per the BullMQ global concurrency docs, a worker's own concurrency acts only as a local ceiling and never lets it exceed the global value.

// captcha-worker.mjs: node captcha-worker.mjs
import { DelayedError, Queue, UnrecoverableError, Worker } from "bullmq";
import IORedis from "ioredis";

const connection = new IORedis(process.env.REDIS_URL ?? "redis://localhost:6379", {
  maxRetriesPerRequest: null,
});
const RETRYABLE = new Set(["ERROR_SERVER_ERROR", "ERROR_INTERNAL_SERVER_ERROR"]);
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

async function captchaai(path, fields) {
  const res = await fetch(`https://ocr.captchaai.com/${path}`, {
    method: "POST",
    body: new URLSearchParams({ key: process.env.CAPTCHAAI_API_KEY, json: "1", ...fields }),
    signal: AbortSignal.timeout(30_000),
  });
  const text = (await res.text()).trim();
  return text.startsWith("{") ? JSON.parse(text) : { status: 0, request: text };
}

function fail(code) {
  // Plain Error: BullMQ retries with the job's backoff. UnrecoverableError: straight to failed.
  if (RETRYABLE.has(code) || !/^[A-Z_]+$/.test(code)) throw new Error(code.slice(0, 80));
  throw new UnrecoverableError(code);
}

// One cap for every worker process on this queue: the plan's thread count.
const queue = new Queue("captcha", { connection });
await queue.setGlobalConcurrency(Number(process.env.CAPTCHAAI_THREADS ?? 5));

const worker = new Worker("captcha", async (job, token) => {
  let { captchaId, submittedAt } = job.data;
  const busyCount = job.data.busyCount ?? 0;
  if (!captchaId) {
    const sub = await captchaai("in.php", job.data.fields);
    if (sub.request === "ERROR_ZERO_BALANCE" && busyCount < 12) {
      // No free thread: wait 10 s without spending one of the job's attempts.
      await job.updateData({ ...job.data, busyCount: busyCount + 1 });
      await job.moveToDelayed(Date.now() + 10_000, token);
      throw new DelayedError();
    }
    if (sub.status !== 1) fail(String(sub.request));
    captchaId = String(sub.request);
    submittedAt = Date.now();
    await job.updateData({ ...job.data, captchaId, submittedAt }); // a stalled rerun resumes here
    await sleep(15_000);
  }
  while (Date.now() - submittedAt < 135_000) {
    const res = await captchaai("res.php", { action: "get", id: captchaId });
    if (res.status === 1) {
      return { token: res.result ?? res.request, userAgent: res.user_agent ?? null, solvedAt: Date.now() };
    }
    if (res.request !== "CAPCHA_NOT_READY") fail(String(res.request));
    await sleep(5_000);
  }
  await job.updateData({ ...job.data, captchaId: null, submittedAt: null });
  throw new Error("no answer after 120 s"); // the retry submits a fresh task
}, { connection, concurrency: 5 });

process.on("SIGTERM", async () => {
  await worker.close(); // stop taking jobs and let running solves finish
  await queue.close();
  await connection.quit();
});

// enqueue.mjs: one job per business request; a repeated jobId is ignored
import { Queue } from "bullmq";
import IORedis from "ioredis";

const connection = new IORedis(process.env.REDIS_URL ?? "redis://localhost:6379", {
  maxRetriesPerRequest: null,
});
const queue = new Queue("captcha", { connection });
await queue.add(
  "solve",
  { fields: { method: "turnstile", sitekey: process.env.SITEKEY, pageurl: process.env.PAGE_URL } },
  { jobId: "signup-42", attempts: 3, backoff: { type: "exponential", delay: 10_000 } },
);
await queue.close();
await connection.quit();

Details that matter here:

• UnrecoverableError moves the job to failed with no further attempts, whatever attempts says, so a bad key costs one request. A plain Error retries after 2 ^ (attempts - 1) * delay ms: 10 s, then 20 s here.
• moveToDelayed plus DelayedError parks the job without incrementing attemptsMade. That suits thread saturation, but attempts then no longer stops a job that keeps getting delayed, hence the busyCount limit: after two minutes without a free thread, the job fails and someone checks the plan. The worker option maxStartedAttempts is a blunter backstop, since it also counts stalls and ordinary retries.
• Why not re-poll with moveToDelayed too? A delayed job leaves the active set and hands its global-concurrency slot to the next job, which submits while the first CaptchaAI thread is still busy. For that pattern, add a Redis in-flight counter like the Celery one.
• Locks and stalls. A worker holds a lock on each active job (lockDuration defaults to 30,000 ms) and keeps renewing it while the processor runs, so a two-minute awaited poll is fine. A blocked event loop, such as synchronous image processing, is what stalls the job. The first stall sends it back to wait, where the saved captchaId lets the rerun resume; a stall beyond maxStalledCount (1 by default) fails it. Move CPU-heavy work to a sandboxed processor.
• jobId deduplication. Adding a job whose ID already exists is ignored. Custom IDs must not contain : or be digits only.

For an in-process Node queue without Redis, see building a CAPTCHA solving queue in Node.js.

River for Go (JobSnooze, JobCancel, MaxWorkers, InsertTx)

River keeps jobs in Postgres, which gives Go services two things the Redis queues lack: the job can be inserted in the same transaction as the business record, and the in-flight counter can live in the same database behind an advisory lock. MaxWorkers is per client; global concurrency limits exist only in River Pro, so the plain-River version below counts solves in a table.

// main.go: a River worker that solves CAPTCHAs and polls with JobSnooze.
package main

import (
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "io"
    "log"
    "net/http"
    "net/url"
    "os"
    "os/signal"
    "regexp"
    "strings"
    "syscall"
    "time"

    "github.com/jackc/pgx/v5"
    "github.com/jackc/pgx/v5/pgxpool"
    "github.com/riverqueue/river"
    "github.com/riverqueue/river/riverdriver/riverpgxv5"
)

const schema = `CREATE TABLE IF NOT EXISTS captcha_solves (
    request_id   text PRIMARY KEY,
    captcha_id   text,
    reserved_at  timestamptz NOT NULL DEFAULT now(),
    submitted_at timestamptz,
    token        text,
    user_agent   text,
    solved_at    timestamptz
)`

type SolveArgs struct {
    RequestID string            `json:"request_id" river:"unique"`
    Fields    map[string]string `json:"fields"`
}

func (SolveArgs) Kind() string { return "captcha_solve" }

func (SolveArgs) InsertOpts() river.InsertOpts {
    return river.InsertOpts{Queue: "captcha", MaxAttempts: 4, UniqueOpts: river.UniqueOpts{ByArgs: true}}
}

type apiResponse struct {
    Status    int             `json:"status"`
    Request   json.RawMessage `json:"request"`
    Result    string          `json:"result"`
    UserAgent string          `json:"user_agent"`
}

// text returns "request" as a string whether it arrived as a JSON string or a number.
func (r apiResponse) text() string {
    var s string
    if json.Unmarshal(r.Request, &s) == nil {
        return s
    }
    return string(r.Request)
}

var errorCode = regexp.MustCompile(`^[A-Z_]+$`)

func captchaAI(ctx context.Context, path string, form url.Values) (apiResponse, error) {
    form.Set("key", os.Getenv("CAPTCHAAI_API_KEY"))
    form.Set("json", "1")
    req, err := http.NewRequestWithContext(ctx, http.MethodPost,
        "https://ocr.captchaai.com/"+path, strings.NewReader(form.Encode()))
    if err != nil {
        return apiResponse{}, err
    }
    req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
    res, err := http.DefaultClient.Do(req)
    if err != nil {
        return apiResponse{}, err
    }
    defer res.Body.Close()
    body, err := io.ReadAll(io.LimitReader(res.Body, 1<<20))
    if err != nil {
        return apiResponse{}, err
    }
    text := strings.TrimSpace(string(body))
    var r apiResponse
    if strings.HasPrefix(text, "{") {
        err = json.Unmarshal([]byte(text), &r)
        return r, err
    }
    r.Request, _ = json.Marshal(text) // plain-text code or HTML error page
    return r, nil
}

// transient reports whether resending the same request can succeed later.
func transient(code string) bool {
    switch code {
    case "ERROR_ZERO_BALANCE", "ERROR_SERVER_ERROR", "ERROR_INTERNAL_SERVER_ERROR":
        return true
    }
    return !errorCode.MatchString(code) // HTML error page or empty body
}

// classify maps a CaptchaAI error code to a River outcome.
func classify(code string) error {
    if transient(code) {
        return fmt.Errorf("captchaai transient: %.80s", code) // retried with River's backoff
    }
    return river.JobCancel(fmt.Errorf("captchaai permanent: %s", code)) // never retried
}

type SolveWorker struct {
    river.WorkerDefaults[SolveArgs]
    pool        *pgxpool.Pool
    maxInFlight int
}

// Each Work call makes one HTTP request, far less than River's 1-minute default timeout.
func (w *SolveWorker) Timeout(*river.Job[SolveArgs]) time.Duration { return 45 * time.Second }

func (w *SolveWorker) Work(ctx context.Context, job *river.Job[SolveArgs]) error {
    var captchaID, token *string
    var submittedAt *time.Time
    err := w.pool.QueryRow(ctx,
        `SELECT captcha_id, submitted_at, token FROM captcha_solves WHERE request_id = $1`,
        job.Args.RequestID).Scan(&captchaID, &submittedAt, &token)
    if err != nil && !errors.Is(err, pgx.ErrNoRows) {
        return err
    }
    switch {
    case token != nil:
        return nil // solved by an earlier run: never solve twice
    case captchaID == nil || submittedAt == nil:
        return w.submit(ctx, job.Args)
    default:
        return w.poll(ctx, job.Args.RequestID, *captchaID, *submittedAt)
    }
}

func (w *SolveWorker) submit(ctx context.Context, args SolveArgs) error {
    ok, err := w.reserve(ctx, args.RequestID)
    if err != nil {
        return err
    }
    if !ok {
        return river.JobSnooze(5 * time.Second) // every plan thread is in flight
    }
    form := url.Values{}
    for k, v := range args.Fields {
        form.Set(k, v)
    }
    res, err := captchaAI(ctx, "in.php", form)
    if err == nil && res.Status != 1 {
        err = classify(res.text())
    }
    if err != nil {
        w.release(ctx, args.RequestID)
        return err
    }
    if _, err := w.pool.Exec(ctx,
        `UPDATE captcha_solves SET captcha_id = $2, submitted_at = now() WHERE request_id = $1`,
        args.RequestID, res.text()); err != nil {
        return err
    }
    return river.JobSnooze(15 * time.Second) // initial wait before the first poll
}

func (w *SolveWorker) poll(ctx context.Context, requestID, captchaID string, submittedAt time.Time) error {
    // Snoozes do not use up MaxAttempts, so the 120 s cap is enforced on elapsed time.
    if time.Since(submittedAt) > 135*time.Second {
        w.release(ctx, requestID)
        return errors.New("captchaai: no answer after 120 s of polling") // the retry submits afresh
    }
    res, err := captchaAI(ctx, "res.php", url.Values{"action": {"get"}, "id": {captchaID}})
    if err != nil {
        return err // network error: retried, and the stored task ID is polled again
    }
    switch {
    case res.Status == 1:
        answer := res.Result // set for reCAPTCHA Enterprise and Cloudflare Challenge
        if answer == "" {
            answer = res.text()
        }
        _, err = w.pool.Exec(ctx,
            `UPDATE captcha_solves SET token = $2, user_agent = $3, solved_at = now() WHERE request_id = $1`,
            requestID, answer, res.UserAgent)
        return err
    case res.text() == "CAPCHA_NOT_READY":
        return river.JobSnooze(5 * time.Second)
    default:
        if !transient(res.text()) {
            w.release(ctx, requestID) // permanent: free the slot
        } // transient: keep the row, so the retry polls the same task ID
        return classify(res.text())
    }
}

// reserve claims one of maxInFlight slots. The transaction-scoped advisory lock makes
// count-then-insert atomic across every process that shares the database.
func (w *SolveWorker) reserve(ctx context.Context, requestID string) (bool, error) {
    tx, err := w.pool.Begin(ctx)
    if err != nil {
        return false, err
    }
    defer tx.Rollback(ctx)
    if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(7240001)`); err != nil {
        return false, err
    }
    var inFlight int
    if err := tx.QueryRow(ctx, `SELECT count(*) FROM captcha_solves
        WHERE token IS NULL AND reserved_at > now() - interval '3 minutes'`).Scan(&inFlight); err != nil {
        return false, err
    }
    if inFlight >= w.maxInFlight {
        return false, nil
    }
    if _, err := tx.Exec(ctx, `INSERT INTO captcha_solves (request_id) VALUES ($1)
        ON CONFLICT (request_id) DO UPDATE SET reserved_at = now()`, requestID); err != nil {
        return false, err
    }
    return true, tx.Commit(ctx)
}

func (w *SolveWorker) release(ctx context.Context, requestID string) {
    _, _ = w.pool.Exec(ctx, `DELETE FROM captcha_solves WHERE request_id = $1 AND token IS NULL`, requestID)
}

func main() {
    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer stop()

    pool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
    if err != nil {
        log.Fatal(err)
    }
    defer pool.Close()
    if _, err := pool.Exec(ctx, schema); err != nil {
        log.Fatal(err)
    }

    workers := river.NewWorkers()
    river.AddWorker(workers, &SolveWorker{pool: pool, maxInFlight: 5})
    client, err := river.NewClient(riverpgxv5.New(pool), &river.Config{
        Queues:  map[string]river.QueueConfig{"captcha": {MaxWorkers: 5}}, // per process
        Workers: workers,
    })
    if err != nil {
        log.Fatal(err)
    }

    // Insert inside the transaction that creates the business record: a rolled-back
    // record never leaves a solve behind, and a committed one always has its job.
    tx, err := pool.Begin(ctx)
    if err != nil {
        log.Fatal(err)
    }
    _, err = client.InsertTx(ctx, tx, SolveArgs{RequestID: "signup-42", Fields: map[string]string{
        "method": "turnstile", "sitekey": os.Getenv("SITEKEY"), "pageurl": os.Getenv("PAGE_URL"),
    }}, nil)
    if err != nil {
        log.Fatal(err)
    }
    if err := tx.Commit(ctx); err != nil {
        log.Fatal(err)
    }

    if err := client.Start(ctx); err != nil {
        log.Fatal(err)
    }
    <-ctx.Done()
    shutdown, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    if err := client.Stop(shutdown); err != nil {
        log.Print(err)
    }
}

Run River's migrations once before starting the worker:

go install github.com/riverqueue/river/cmd/river@latest
river migrate-up --line main --database-url "$DATABASE_URL"

The River behaviours this design leans on:

• JobSnooze is not an attempt. Per the River snoozing docs, each snooze decrements the attempt count, so a job can snooze indefinitely without reaching MaxAttempts. Ideal for CAPCHA_NOT_READY, and the reason the 120-second cap is checked against submitted_at.
• JobCancel cancels the job permanently with the wrapped error stored; River never retries it.
• Timeouts. River cancels a job's context after one minute unless the client sets JobTimeout or the worker overrides Timeout(). Each Work call here makes one HTTP request, so 45 s is generous; a poll-inside-the-job worker would need about 150 s.
• Uniqueness. UniqueOpts{ByArgs: true} with the river:"unique" tag makes RequestID the key, so a second insert returns the existing job with UniqueSkippedAsDuplicate set.

Graphile Worker on Postgres applies the same table-backed approach in Node, and solving CAPTCHAs with Go covers the plain client.

Sizing total concurrency to plan threads

The rule is one sum: concurrency per process times the number of processes, across every service using the key, must not exceed the plan's threads. Where the cap lives decides whether scaling out breaks it:

Current plans are on the CaptchaAI pricing page; billing is per concurrent thread with unlimited solves. Caps kept in shared state (the Redis sorted set, BullMQ's global concurrency, the Postgres table, threadsinfo) survive scale-out and deploys unchanged. Per-process caps (--concurrency, capsule concurrency, MaxWorkers) multiply with every process, including old ones still draining during a deploy.

Check the arithmetic before the worker starts. getbalance returns the plan's thread count, and threadsinfo returns threads and working_threads for the whole account:

#!/usr/bin/env bash
# start-captcha-worker.sh: refuse to start more concurrency than the plan has threads.
set -euo pipefail
: "${CAPTCHAAI_API_KEY:?CAPTCHAAI_API_KEY is not set}"
concurrency="${CAPTCHA_CONCURRENCY:-5}"
replicas="${WORKER_REPLICAS:-1}"

api() {
  curl -fsS --max-time 30 https://ocr.captchaai.com/res.php \
    --data-urlencode "key=${CAPTCHAAI_API_KEY}" --data-urlencode "action=$1"
}

plan_threads="$(api getbalance)"   # plain-text thread count, e.g. 15
if ! [[ "$plan_threads" =~ ^[0-9]+$ ]]; then
  echo "getbalance returned: ${plan_threads}" >&2
  exit 1
fi
working="$(api threadsinfo | jq -r '.working_threads // "unknown"')"
planned=$(( concurrency * replicas ))
echo "plan threads: ${plan_threads}, busy now: ${working}, this deployment: ${planned}"
if (( planned > plan_threads )); then
  echo "refusing to start: ${planned} concurrent solves > ${plan_threads} plan threads" >&2
  exit 1
fi
exec celery -A tasks worker -Q captcha --concurrency="$concurrency" -l INFO

Saturation does not always announce itself. The official SDK's source notes that the API may queue tasks when every thread is busy, so an oversubscribed setup can show up as slower answers and more CAPCHA_NOT_READY polls rather than as ERROR_ZERO_BALANCE. That is why the client-side cap matters even when errors look quiet.

Idempotency, redelivery and token freshness

Every queue here reruns a job you thought had finished or died:

Two rules sit behind the table. Save the task ID before the first poll: the gap between in.php answering and that write is the only window where a crash still costs a duplicate solve. And resume only while the task is young; about 135 seconds after submission, drop the stored ID and submit fresh.

Tokens then have their own clock. Google's reCAPTCHA documentation says a response token is valid for two minutes and can be verified once. Cloudflare gives a Turnstile token 300 seconds, single-use, and CaptchaAI's docs say reCAPTCHA v3 tokens expire within minutes. Record when each answer arrived, expire stored tokens early (the examples use 90 s), and never hand one token to two jobs. If a consumer waits in a queue for the token, it is spending that window, which is another argument for polling inside the session-bound job.

Hosting queue workers: Render background workers and Railway

Render. A background worker is a Render service that runs continuously but receives no incoming traffic; Render's docs name Celery, Sidekiq and BullMQ among typical frameworks, often backed by a Render Key Value instance. Its zero-downtime deploy sequence applies to background workers too: once the new instance is up, the original receives SIGTERM 60 seconds later, then SIGKILL after the shutdown delay, which defaults to 30 seconds and can be raised to 300 with the maxShutdownDelaySeconds field in render.yaml. So for about a minute old and new instances overlap, and per-process caps briefly double. And 30 seconds is shorter than a solve: Celery's warm shutdown on TERM waits for running tasks, so set the delay to at least your time_limit (180 s here) and raise Sidekiq's -t to match.

Railway. When a new deployment is online, Railway sends the old one SIGTERM and by default allows 0 seconds before SIGKILL, so every deploy kills in-flight solves unless RAILWAY_DEPLOYMENT_DRAINING_SECONDS covers a full solve. RAILWAY_DEPLOYMENT_OVERLAP_SECONDS (also 0 by default) keeps the previous deployment alive after the new one becomes active; set it, and per-process caps double during the overlap as on Render. For workloads that run on a schedule, scheduled CAPTCHA solving with cron jobs is the better model; for a serverless queue, see Cloudflare Queues.

Monitoring per queue

Job state lives in each queue's dashboard: Flower for Celery (celery -A tasks flower, port 5555 by default), Sidekiq's Web UI with its Retries and Dead tabs, the open-source Bull Board for BullMQ, and River UI. Whether the CAPTCHA side is healthy takes a few counters of your own:

• Solve latency, submit to answer, p50 and p95 per CAPTCHA type. Compare against CaptchaAI's published ceilings (reCAPTCHA v2 under 60 s, Cloudflare Turnstile under 10 s, image CAPTCHAs under 0.5 s). A rising p95 with no errors usually means oversubscription.
• CAPCHA_NOT_READY polls per solve. In the Celery reschedule pattern, that is the retry count of poll_captcha.
• Permanent errors by code. A credential error should page someone; a burst of request errors usually means the target page changed its sitekey or markup.
• ERROR_ZERO_BALANCE per minute, and resumed versus fresh submissions. The first tracks cap violations, the second redelivery.
• Token age at use. Anything near the validity window will start failing downstream.

Troubleshooting

FAQ

Can Celery's rate_limit keep me within my plan's threads?

No. Celery documents rate_limit as a per-worker-instance limit, not a global one, and it limits how often a task starts, not how many run at once. Two workers at rate_limit="10/m" start 20 tasks a minute, each able to hold a thread for two minutes. Use a dedicated queue with fixed concurrency, or a shared counter.

Should I use the official Python SDK inside Celery tasks?

You can, with two adjustments. CaptchaAI(...) calls threadsinfo when it is constructed, so create one client per worker process rather than one per task. And with its default auto_retry=True, the SDK retries internally, including resubmitting for up to thread_busy_timeout (120 s) when every thread is busy, which hides sleeps inside your soft_time_limit. Construct it with auto_retry=False so Celery owns retries, map ThreadLimitError and NetworkError to your transient class, and treat InvalidKeyError, ValidationError, ProxyError, UnsolvableError and NoThreadsError as permanent. APIError is mixed: it wraps IP_BANNED, which must never be retried, but also ERROR_INTERNAL_SERVER_ERROR seen while polling, which should be, so neither retry nor fail that class wholesale. The SDK's own TimeoutError (poll deadline passed) means a fresh submission, not another poll.

Several services share one API key. How do I split the threads?

The plan's threads are account-wide, and threadsinfo reports usage across every client on the key. Either give each service a fixed share (for example 10 and 5 on a STANDARD plan) and enforce it with that service's own cap, or gate every submission on threadsinfo as the Sidekiq example does, accepting a brief overshoot that ERROR_ZERO_BALANCE backoff absorbs.

Next step

Pick the pattern that matches where your token is used, set one cap in shared state, and run the preflight check before the first deploy. Get a CaptchaAI key and run your first thread-capped queue worker.