Repartir trabajo y recuperar lo que quedó pendiente

Pub/Sub envía mensajes a suscriptores conectados. Si nadie escucha o un suscriptor pierde la conexión, no tiene un historial duradero al que volver. Sirve para avisos cuya pérdida es tolerable, por ejemplo pedir que una pantalla vuelva a consultar su estado. No lo uses como único registro de una tarea que prometiste realizar.

Una lista con LPUSH y BRPOP permite repartir elementos, pero BRPOP los retira al entregarlos. Si el consumidor cae después de extraerlos y antes de trabajar, ya no están en la lista. Puedes diseñar una lista de procesamiento con movimiento atómico y recuperación, pero estás construyendo un protocolo. Streams ofrece un registro y grupos de consumidores que hacen visible esa situación.

Mecanismo	Conserva un historial	Registra entregados sin confirmar	Uso razonable
Pub/Sub	No	No	Avisos efímeros
List + extracción	Mientras no se extraen	No automáticamente	Buffer simple con pérdida aceptada o protocolo adicional
Stream + grupo	Sí, hasta borrado/recorte	Sí, PEL	Trabajo recuperable con confirmaciones explícitas

La persistencia del servidor y la política de memoria siguen importando en los tres casos. Que un Stream exista en RAM no garantiza que sobreviva a una pérdida del disco.

Cada entrada tiene un identificador ordenado y pares campo-valor. Con XADD * Redis genera el identificador, normalmente representado como milisegundos y una secuencia. Ese ID describe una entrada en el Stream; tu job_id describe la intención de negocio. Publicar dos veces el mismo trabajo produce dos entradas distintas salvo que diseñes deduplicación.

Ejecuta este bloque Python con la instancia de práctica. Borra únicamente la clave de esta demostración antes de crear el grupo:

import redis

r = redis.Redis(host="127.0.0.1", port=6380, decode_responses=True, protocol=2)
stream = "curso:trabajos-demo"
r.delete(stream)
r.xgroup_create(stream, "operadores", id="0-0", mkstream=True)
message_id = r.xadd(stream, {"job_id": "informe-1", "lote": "A"})
print(message_id)
print(r.xrange(stream, count=10))

mkstream=True crea un Stream vacío si no existe. id="0-0" hace que el grupo pueda consumir desde el principio; $ empezaría después de las entradas que ya existieran al crear el grupo. El grupo no crea un proceso consumidor: conserva estado para los procesos que conectes después.

lotes = r.xreadgroup("operadores", "ana", {stream: ">"}, count=1, block=1000)
print(lotes)
print(r.xpending(stream, "operadores"))

> pide entradas todavía no entregadas a ningún consumidor de ese grupo. count=1 limita el lote. block=1000 espera como máximo un segundo si no hay mensajes nuevos; requiere que el timeout de socket sea mayor si lo has configurado. La respuesta contiene el nombre del Stream y sus entradas. XPENDING muestra ahora una pendiente.

El consumidor Ana no enviará XACK. Para reproducir inmediatamente una reclamación, usamos min_idle_time=0 solo en la demostración:

cursor, mensajes, borrados = r.xautoclaim(
    stream, "operadores", "bruno", min_idle_time=0,
    start_id="0-0", count=10,
)
print(cursor, mensajes, borrados)
print(r.xpending(stream, "operadores"))
r.xack(stream, "operadores", message_id)
print(r.xpending(stream, "operadores")["pending"])  # 0
print(r.xlen(stream))  # 1: ACK no elimina el historial.

En Redis 7 o posterior, la respuesta de XAUTOCLAIM incluye cursor, entradas reclamadas e IDs cuyas entradas ya fueron borradas del Stream. Si has recortado datos todavía pendientes, Redis no puede reconstruir su cuerpo. Es una pérdida que debes detectar, no una tarea completada.

El cursor permite continuar el recorrido. Sigue hasta 0-0 aunque una tanda venga vacía: COUNT limita el trabajo y puede haber pendientes que todavía no sean reclamables. Tras terminar, otra pasada futura puede encontrar elegibles que antes no lo eran. No reinicies siempre en 0-0 y asumas que una llamada ha visto toda una cola grande.

En operación, el plazo mínimo debe superar la duración esperada del procesamiento y tener margen. Aun así, Ana puede sufrir una pausa larga y seguir trabajando cuando Bruno ya ha reclamado. Redis cambia el dueño registrado; no mata a Ana ni anula sus escrituras en otro sistema.

Por eso el efecto usa un identificador estable y un registro de resultados. Si job_id=informe-1 ya produjo un informe con el mismo contenido, la repetición devuelve el resultado anterior. Si intenta reutilizar ese ID con otro contenido, se rechaza. El recibo de idempotencia y el efecto deben confirmarse en la misma transacción del receptor. El mini proyecto lo implementa con SQLite.

Esto ofrece procesamiento con reentregas y un efecto deduplicado dentro de un alcance concreto. No una garantía universal de “exactamente una vez”: un email enviado, una impresión física o una API externa necesitan su propio contrato de idempotencia o reconciliación.

Un mensaje inválido no mejora reintentándolo cien veces. Guarda un rechazo durable con su identidad y motivo, y confirma solo después de conservar esa evidencia. Un fallo temporal de conexión a la base de datos, en cambio, debe quedar pendiente para recuperación. Distingue estos casos por tipos de error, no por un except Exception que trate todo igual.

No uses MAXLEN 1000 como solución automática al crecimiento: puede borrar trabajo no procesado o necesario para otro grupo. Define cuánto historial necesitas, cuántos grupos lo consumen, cuál es el más retrasado y qué harás si supera la retención. Si un único grupo termina y ya conservas resultados duraderos, puedes diseñar limpieza de entradas confirmadas; añadir un segundo grupo cambia esa decisión.

Observa pendientes, antigüedad, reentregas, rechazos y lag del grupo. Una cola con diez mensajes puede estar averiada si el más antiguo lleva una hora; diez mil entradas históricas confirmadas pueden ser normales. El manual de Streams y la referencia de XAUTOCLAIM documentan el protocolo. El mini proyecto añade lo que el protocolo no sabe: la intención y el efecto real.