Colas de Tareas: APIs Rápidas con Redis y Workers

TL;DR:

• La Trampa: Ejecutar operaciones pesadas de forma síncrona dentro del ciclo de vida HTTP (generar facturas en PDF, procesar analítica, disparar webhooks) secuestra los workers del servidor por 10 a 30 segundos, causando errores HTTP 504 Gateway Timeout y caídas en cascada ante picos de tráfico.
• La Arquitectura: Un pipeline de procesamiento de eventos de 4 etapas: respuestas HTTP 202 Accepted ultrarrápidas en menos de 10ms con un job_id, Redis Streams con Consumer Groups para entrega persistente al menos una vez, deduplicación con claves de idempotencia y colas de mensajes fallidos (Dead-Letter Queues).
• El Impacto: Mantiene la latencia p99 de la API por debajo de 35ms sin importar el volumen de trabajo de fondo, evita cobros duplicados por reintentos de red y escala el procesamiento horizontalmente sobre infraestructura en la nube de menos de $50 USD/mes.

La Paradoja Síncrona: Por Qué los Procesos Pesados Tumban tu Backend

Obligar a un usuario a esperar ocho segundos frente a una pantalla de carga mientras tu backend genera un reporte en PDF, consulta analíticas externas y dispara tres webhooks síncronos es la forma más rápida de destruir la retención de clientes y la credibilidad de tu producto.

Cuando los equipos de ingeniería ejecutan procesamiento computacional o tareas intensivas de entrada/salida (I/O) dentro del ciclo de vida ordinario de una petición HTTP, el hilo del servidor web permanece completamente bloqueado.

Bajo picos moderados de tráfico—como un cierre de ventas, un lanzamiento de producto o la facturación de fin de mes—, los pools de conexiones entrantes se agotan en cuestión de segundos. El resultado es una avalancha de errores catastróficos 504 Gateway Timeout que arrastra a otros microservicios y APIs de la empresa.

Para que un sistema soporte miles de usuarios concurrentes con latencias predecibles, los controladores HTTP de la API pública jamás deben realizar trabajo computacional pesado.

La respuesta arquitectónica es desacoplar la intención del usuario de la ejecución real mediante Colas de Tareas Asíncronas.

Análisis Arquitectónico: Sobreviviendo a la Realidad de las Colas Distribuidas

1. La Falacia de la Entrega “Exactly-Once” y la Idempotencia

Muchos desarrolladores asumen erróneamente que cuando un sistema encola una tarea, el worker la ejecutará exactamente una sola vez. En sistemas distribuidos, esto es un mito peligroso.

Las fallas de red, los reinicios inesperados de workers y los timeouts de confirmación (XACK) en Redis significan que los brokers de mensajes proporcionan entrega al menos una vez (at-least-once delivery):

• Si el Worker A procesa un cobro pero se reinicia 2 milisegundos antes de enviar su acuse de recibo a Redis, el broker reasigna el mensaje no confirmado al Worker B.
• Si el Worker B ejecuta la tarea a ciegas sin un Guardián de Idempotencia, el cliente sufrirá un cobro duplicado.

La Solución: Cada mensaje encolado debe incluir una clave determinista única idempotency_key (por ejemplo, id_orden + accion). Antes de ejecutar efectos secundarios externos, el worker establece un candado atómico en Redis (SET lock:{idempotency_key} NX EX 300) y valida en PostgreSQL que la transición de estado no haya sido aplicada previamente.

2. Tareas “Poison Pill” y Colas Dead-Letter (DLQ)

Una “píldora venenosa” (poison pill) es un mensaje con datos corruptos o malformados que provoca que el worker lance una excepción fatal o un error de memoria no recuperable (por ejemplo, división por cero o un binario dañado).

Sin una política de reintentos acotada y una Dead-Letter Queue:

1. El worker falla al procesar el mensaje corrupto.
2. El orquestador (Docker o Kubernetes) reinicia el worker.
3. El worker vuelve a leer el mismo mensaje corrupto de la cola.
4. El worker vuelve a colapsar en un bucle infinito (CrashLoopBackOff), congelando el procesamiento de todas las tareas saludables pendientes.

La Solución: Los workers registran el número de intentos en la metadata del mensaje. Si un trabajo supera su límite de reintentos (por ejemplo, 3 intentos), se retira del flujo activo y se deriva a una Dead-Letter Queue (jobs:dlq) junto con la traza completa del error, alertando al equipo de ingeniería sin detener el procesamiento de las tareas sanas.

3. Partición por Prioridad para Evitar Inanición de Recursos

Si tu plataforma procesa los correos de compra de usuarios y los reportes analíticos masivos mensuales en la misma cola genérica, un cliente corporativo que solicite exportar 100,000 registros saturará toda la flota de workers. Los compradores esperando su recibo sufrirán demoras de 40 minutos.

Las arquitecturas de producción aplican Partición por Prioridades:

• queue:critical: Eventos de checkout, autenticación y recuperación de contraseñas (procesados de inmediato por workers dedicados).
• queue:default: Webhooks operacionales estándar y sincronizaciones cotidianas.
• queue:batch: Generación de PDFs masivos, procesamiento analítico y tareas nocturnas de mantenimiento.

Comparativa de Código: Endpoint Síncrono Bloqueante vs Motor Asíncrono Resiliente

Naive: Controlador Síncrono Bloqueante en el Ciclo HTTP

# Endpoint Naive: Ejecuta computo pesado bloqueando el servidor HTTP
import time
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class InvoiceRequest(BaseModel):
    tenant_id: str
    user_id: str
    month: str

@app.post("/api/v1/invoices/generate")
def generate_invoice_naive(req: InvoiceRequest):
    try:
        # CUELLO DE BOTELLA 1: Simula 12 segundos renderizando PDF y analitica
        # Bloquea completamente el hilo del servidor, impidiendo atender a otros clientes
        time.sleep(12)
        
        # CUELLO DE BOTELLA 2: Llamada HTTP sincrona externa. Si Stripe demora, se cuelga
        # CUELLO DE BOTELLA 3: Si el usuario cierra el navegador, el trabajo se pierde sin reintentos
        pdf_url = f"https://s3.amazonaws.com/invoices/{req.tenant_id}_{req.month}.pdf"
        
        return {
            "status": "success",
            "pdf_url": pdf_url
        }
    except Exception as exc:
        raise HTTPException(status_code=500, detail="Fallo en la generacion")

Producción: Motor Asíncrono de Eventos con Redis Streams y DLQ

# Motor de Produccion: HTTP 202 en <10ms, Redis Streams, Consumer Groups, DLQ e Idempotencia
import asyncio
import json
import logging
import uuid
import time
from typing import Optional
from fastapi import FastAPI, Request, Response, status, HTTPException
from pydantic import BaseModel, Field
import redis.asyncio as aioredis

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
logger = logging.getLogger("async_task_engine")

app = FastAPI()
redis_client = aioredis.from_url("redis://localhost:6379", decode_responses=True)

STREAM_NAME = "jobs:stream:invoices"
GROUP_NAME = "workers:group:invoicing"
DLQ_NAME = "jobs:dlq:invoices"

class InvoiceJobPayload(BaseModel):
    tenant_id: str = Field(..., min_length=3, max_length=64)
    invoice_id: str = Field(..., min_length=3, max_length=64)
    idempotency_key: str = Field(..., min_length=8, max_length=128)
    month: str = Field(..., regex=r"^\d{4}-\d{2}$")

# 1. Productor de Alta Velocidad (Respuesta HTTP 202 en menos de 10ms)
@app.post("/api/v1/invoices/generate", status_code=status.HTTP_202_ACCEPTED)
async def enqueue_invoice_job(payload: InvoiceJobPayload, response: Response):
    # GUARDIA DE IDEMPOTENCIA: Prevenir envios duplicados por doble clic o reintentos
    idempotency_lock = f"idempotency:lock:{payload.tenant_id}:{payload.idempotency_key}"
    is_new = await redis_client.set(idempotency_lock, "PROCESSING", nx=True, ex=3600)
    
    if not is_new:
        logger.info(f"Peticion duplicada detectada para idempotency_key={payload.idempotency_key}")
        return {
            "status": "Accepted",
            "message": "Trabajo previamente registrado y actualmente en procesamiento",
            "idempotency_key": payload.idempotency_key
        }

    job_id = f"job_{uuid.uuid4().hex[:12]}"
    
    # Encolar la intencion de trabajo en Redis Streams persistente
    job_envelope = {
        "job_id": job_id,
        "tenant_id": payload.tenant_id,
        "invoice_id": payload.invoice_id,
        "idempotency_key": payload.idempotency_key,
        "month": payload.month,
        "attempts": "0",
        "created_at": str(time.time())
    }
    
    await redis_client.xadd(STREAM_NAME, job_envelope)
    
    # Retornar 202 Accepted con cabecera estandar de seguimiento
    response.headers["Location"] = f"/api/v1/jobs/{job_id}"
    return {
        "status": "Accepted",
        "job_id": job_id,
        "poll_url": f"/api/v1/jobs/{job_id}",
        "message": "Generacion de factura encolada exitosamente"
    }

# 2. Worker Resiliente con Grupos de Consumidores y Manejo de DLQ
class InvoicingWorker:
    def __init__(self, worker_id: str):
        self.worker_id = worker_id
        self.running = True

    async def setup_stream(self):
        """Crea el grupo de consumidores si no existe previamente."""
        try:
            await redis_client.xgroup_create(STREAM_NAME, GROUP_NAME, id="0", mkstream=True)
            logger.info(f"Grupo de consumidores {GROUP_NAME} inicializado.")
        except Exception as e:
            if "BUSYGROUP" not in str(e):
                logger.error(f"Error al crear grupo de consumidores: {e}")

    async def run_loop(self):
        await self.setup_stream()
        logger.info(f"Worker {self.worker_id} escuchando en {STREAM_NAME}...")
        
        while self.running:
            try:
                # Leer 1 mensaje pendiente con espera no bloqueante de 2 segundos
                entries = await redis_client.xreadgroup(
                    GROUP_NAME, self.worker_id, {STREAM_NAME: ">"}, count=1, block=2000
                )
                
                if not entries:
                    continue

                for stream_key, messages in entries:
                    for msg_id, data in messages:
                        await self.process_message(msg_id, data)

            except Exception as loop_err:
                logger.error(f"Excepcion en el bucle del worker {self.worker_id}: {loop_err}")
                await asyncio.sleep(1)

    async def process_message(self, msg_id: str, data: dict):
        job_id = data.get("job_id")
        attempts = int(data.get("attempts", "0")) + 1
        
        try:
            logger.info(f"Worker {self.worker_id} procesando job_id={job_id} (Intento {attempts})")
            
            # --- Trabajo Simulado: Computo pesado de PDF y subida a almacenamiento ---
            await asyncio.sleep(2.0) # Procesamiento asincrono no bloqueante
            
            # Registro de estado atómico y confirmacion
            await redis_client.set(
                f"job:result:{job_id}", 
                json.dumps({"status": "COMPLETED", "completed_at": time.time()}), 
                ex=86400
            )
            # Confirmar mensaje para retirarlo de la lista de pendientes (PEL)
            await redis_client.xack(STREAM_NAME, GROUP_NAME, msg_id)
            logger.info(f"Trabajo {job_id} confirmado exitosamente (XACK).")

        except Exception as task_err:
            logger.error(f"Falla en la ejecucion del trabajo {job_id}: {task_err}")
            
            if attempts >= 3:
                # DERIVACION A DEAD-LETTER QUEUE: Presupuesto de reintentos agotado
                logger.critical(f"Trabajo {job_id} supero maximo de intentos. Derivando a DLQ: {DLQ_NAME}")
                await redis_client.xadd(DLQ_NAME, {
                    **data,
                    "final_error": str(task_err),
                    "failed_at": str(time.time())
                })
                # Confirmar en el stream principal para que no bloquee otros trabajos
                await redis_client.xack(STREAM_NAME, GROUP_NAME, msg_id)
            else:
                # Re-encolar con contador incrementado y backoff exponencial
                backoff_delay = 2 ** attempts
                logger.warning(f"Re-encolando trabajo {job_id} tras espera de {backoff_delay}s...")
                await asyncio.sleep(backoff_delay)
                data["attempts"] = str(attempts)
                await redis_client.xadd(STREAM_NAME, data)
                await redis_client.xack(STREAM_NAME, GROUP_NAME, msg_id)

Casos de Éxito Reales: Rendimiento Orientado a Eventos

Strategio (Ganador Startup Perú 2da Generación): Analítica Masiva sin Fricción

Durante la arquitectura y lanzamiento de Strategio, la plataforma de Inteligencia de Negocios creada en 12 semanas:

• El Problema: La plataforma necesitaba procesar miles de transacciones de ventas diarias, calcular márgenes multivariables y consolidar reportes analíticos ejecutivos para cientos de comercios cada noche. Los prototipos síncronos iniciales arrojaban errores 504 Timeout en inventarios superiores a 10,000 productos.
• La Solución: Desacoplamos todo el procesamiento analítico mediante un pipeline asíncrono con Redis Streams y workers en Python ejecutándose en contenedores Docker sobre servidores Hetzner.
• El Resultado: La plataforma procesó millones de eventos analíticos sin interrupción, manteniendo la latencia p99 de la API pública por debajo de 35ms, con un costo total de infraestructura cloud inferior a $50 USD/mes.

PróximaBolilla (proximabolilla.com): Simulaciones Estocásticas Masivas

PróximaBolilla ejecuta simulaciones probabilísticas evaluando más de 1,000,000 de combinaciones numéricas en tiempo real.

• El Problema: Ejecutar simulaciones combinatorias intensivas de forma síncrona congelaba las sesiones web de los usuarios y provocaba cancelaciones de red durante los picos de visitas en noches de sorteos.
• La Solución: Implementamos un pipeline de tareas asíncronas. La solicitud emite de inmediato una respuesta 202 Accepted con un identificador de seguimiento, mientras workers independientes procesan los cálculos estocásticos en paralelo a través de múltiples núcleos de CPU.
• El Resultado: Eliminación absoluta de errores de timeout HTTP, escalabilidad lineal en múltiples núcleos de cómputo y una interfaz que responde en menos de 10ms.

Lista de Verificación Previa para Colas de Tareas Asíncronas

Antes de desplegar una arquitectura de colas de tareas a producción, valida estos cinco puntos de control:

• Apretón de Manos HTTP 202 Inmediato: ¿Tu endpoint valida los datos, encola el mensaje y responde en menos de 15ms sin ejecutar lógica pesada en línea?
• Protección de Idempotencia: ¿Tu worker valida una idempotency_key o candado distribuido para evitar cobros dobles o efectos secundarios repetidos?
• Persistencia en el Broker: ¿Tu broker está configurado con persistencia en disco (Redis AOF/RDB o colas durables en RabbitMQ) para sobrevivir a reinicios del servidor?
• Aislamiento en Dead-Letter Queue: ¿Las tareas que fallan reiteradamente se desvían a una DLQ con el registro del error para inspección del equipo de ingeniería?
• Concurrencia en Grupos de Consumidores: ¿Los workers escalan horizontalmente mediante grupos de consumidores con confirmaciones XACK únicamente tras completar la transacción?

Asesoría de Arquitectura

Desacoplar APIs de alto tráfico de cargas computacionales pesadas requiere experiencia en sistemas distribuidos, control de concurrencia y garantías de idempotencia.

En Coins5, Marlon ayuda a startups y empresas en crecimiento a eliminar cuellos de botella de latencia, diseñar pipelines de eventos resilientes y optimizar el rendimiento de sus workers para reducir costos de infraestructura.

Si tu plataforma experimenta errores HTTP 504 o tu backend se satura con procesos lentos, conversemos en una llamada de descubrimiento técnico de 15 minutos.