Mensajes duplicados MQTT: causas y soluciones con QoS 1

Los mensajes duplicados en MQTT no son un fallo de tu broker; son una consecuencia documentada de QoS 1, que promete la entrega "al menos una vez". Cuando un corte de red se come un acuse de recibo, el emisor retransmite y tu consumidor ve el mismo mensaje dos veces. En esta guía de resolución de problemas aprenderás exactamente cuándo se producen los duplicados, qué te dice (y qué no) el flag DUP y cómo escribir un consumidor idempotente en Python que los gestione correctamente.

Por qué QoS 1 produce duplicados

QoS 1 utiliza un intercambio en dos pasos en cada salto: el emisor transmite PUBLISH y el receptor responde con PUBACK. Hasta que llega el PUBACK, el emisor conserva el mensaje y lo volverá a enviar. El artículo niveles de QoS de MQTT explicados cubre los tres niveles; aquí nos centramos en lo que falla con el nivel 1.

Se crea un duplicado siempre que el receptor recibió el mensaje pero el emisor no recibió el acuse de recibo:

El dispositivo publica la lectura n.º 42 en el broker.
El broker la recibe, la reenvía a los suscriptores y envía PUBACK.
La conexión se cae antes de que el PUBACK llegue al dispositivo.
El dispositivo se reconecta con su sesión y reenvía la n.º 42 con el flag DUP activado.
El broker la trata como un mensaje nuevo y la reenvía de nuevo.

Lo mismo ocurre en el salto del broker al suscriptor. Si el suscriptor procesó un mensaje pero su PUBACK se perdió, el broker lo reenvía cuando se reanuda la sesión del suscriptor.

Ten en cuenta que un duplicado puede llegar en cualquiera de los dos saltos de forma independiente, así que un mismo mensaje publicado puede, ocasionalmente, llegar a un consumidor más de dos veces.

Qué te dice el flag DUP

Cada paquete PUBLISH tiene un bit DUP. El emisor lo activa cuando retransmite un paquete QoS 1 o QoS 2. Resulta tentador descartar todos los mensajes con DUP activado, pero es un error por varias razones:

Suposición	Realidad
DUP significa que ya tengo este mensaje	No necesariamente. La primera copia pudo perderse, y el reintento es la única copia que recibes
Sin DUP significa que es nuevo	No. El broker reenvía el reintento de un publicador como una entrega nueva, sin DUP en el salto de salida
Los identificadores de paquete permiten detectar duplicados	Solo dentro de una sesión y de una ventana de mensajes en vuelo. Los ID se reutilizan constantemente
QoS 2 lo resuelve todo	QoS 2 evita los duplicados por salto, pero los reintentos a nivel de aplicación aún pueden crearlos

El flag DUP es útil para registros y diagnóstico. No es una clave de deduplicación. Una deduplicación real necesita un identificador que tú mismo incluyas en el mensaje.

Reproducir duplicados

La reentrega real depende de paquetes perdidos y es difícil de provocar a voluntad. La forma fiable de probar tu consumidor es inyectar duplicados tú mismo. Contra el servidor MQTT público gratuito en broker.connectmq.com, pensado para pruebas y no para datos de producción, publica el mismo ID de mensaje dos veces:

for n in 1 2; do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i dup-pub-t9v4 -q 1 \
    -t connectmq/demo/t9v4/meter-3/energy \
    -m '{"msgId":"meter-3-000042","kwh":0.25}'
done

Un consumidor que suma kwh sin deduplicación informa ahora de 0,5 en lugar de 0,25. Ese es exactamente el tipo de fallo que provoca la reentrega de QoS 1 en producción, solo que más raro y más difícil de detectar.

Crear un consumidor idempotente

Un consumidor idempotente produce el mismo resultado tanto si procesa un mensaje una vez como cinco. Hay dos técnicas habituales:

Idempotencia natural. Diseña el efecto para que repetirlo sea inofensivo: "fijar la temperatura en 21,5" en lugar de "subirla 0,5"; hacer un upsert de una fila con clave por dispositivo y marca de tiempo en lugar de insertarla.
Deduplicación explícita. Asigna a cada mensaje un ID único y recuerda los ID procesados recientemente.

Este es un consumidor completo con paho-mqtt 2.x que usa deduplicación explícita con una caché limitada en tamaño y en tiempo:

import json
import time
import uuid
from collections import OrderedDict
import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
TOPIC = "connectmq/demo/t9v4/+/energy"
DEDUP_TTL = 600          # segundos que se recuerda un ID
DEDUP_MAX = 50_000       # límite de uso de memoria

seen = OrderedDict()     # msgId -> momento en que se vio por primera vez
totals = {}

def already_seen(msg_id):
    now = time.time()
    while seen:
        oldest_id, ts = next(iter(seen.items()))
        if now - ts > DEDUP_TTL or len(seen) > DEDUP_MAX:
            seen.popitem(last=False)
        else:
            break
    if msg_id in seen:
        return True
    seen[msg_id] = now
    return False

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TOPIC, qos=1)

def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload)
        msg_id = data["msgId"]
    except (ValueError, KeyError):
        print("rejected message without msgId")
        return
    if already_seen(msg_id):
        print(f"duplicate {msg_id} (dup flag={msg.dup}) ignored")
        return
    meter = msg.topic.split("/")[-2]
    totals[meter] = totals.get(meter, 0) + data["kwh"]
    print(f"{meter} total={totals[meter]:.2f} kWh")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"dedup-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=30)
client.loop_forever()

Ejecútalo y luego lanza el bucle de doble publicación de la sección anterior. El primer mensaje actualiza el total; el segundo se notifica como duplicado y se ignora.

Elegir un buen ID de mensaje

El ID debe crearlo el publicador original, una sola vez, y reutilizarse en cada reintento. Opciones:

ID de dispositivo más número de secuencia, como meter-3-000042. Compacto y útil también para comprobar el orden de los mensajes MQTT. Persiste el contador o añade un ID de arranque para que no se repita tras un reinicio.
UUID generado por mensaje. Sencillo y único a nivel global, a costa de 36 bytes.
ID de dispositivo más marca de tiempo de la medición, si el dispositivo nunca produce dos lecturas con la misma marca de tiempo.

No generes el ID en el consumidor ni en una capa intermedia; para entonces, un duplicado parece un mensaje nuevo.

Escalar la deduplicación

La caché en memoria anterior funciona para un único proceso consumidor. Para sistemas más grandes:

Restricciones de base de datos. Un índice único sobre msg_id convierte los duplicados en conflictos inofensivos. Con un upsert (INSERT ... ON CONFLICT DO NOTHING en PostgreSQL), a menudo es la única deduplicación que necesitas.
Caché compartida. Si varios consumidores de una suscripción compartida pueden recibir reintentos del mismo mensaje, el conjunto de ID vistos debe ser compartido, por ejemplo en Redis con un TTL.
Dimensionado del TTL. La ventana solo necesita cubrir retrasos de reentrega realistas: la expiración de tu sesión más un margen. Los duplicados rara vez llegan con horas de diferencia, salvo que las sesiones duren tanto.

¿Deberías usar simplemente QoS 2?

QoS 2 añade un intercambio en cuatro pasos que garantiza la entrega exactamente una vez en cada salto. Cuesta más idas y vueltas y más estado en el broker, y no te protege frente a un dispositivo que agota su tiempo de espera y vuelve a publicar la misma lectura como un mensaje nuevo a nivel de aplicación. En planes limitados puede que ni siquiera esté disponible; por ejemplo, el plan Serverless gratuito de ConnectMQ admite QoS hasta 1, mientras que el plan Serverless de pago añade QoS 2. Consulta QoS máximo en MQTT para ver cómo los brokers anuncian y rebajan el QoS. En la mayoría de los sistemas, QoS 1 más consumidores idempotentes es el diseño más robusto y económico.

Lista de comprobación para la resolución de problemas

¿Los duplicados coinciden con reconexiones? Revisa los registros del cliente en busca de desconexiones y timeouts de keepalive.
¿Tu publicador reintenta a nivel de aplicación además de QoS 1? Eso duplica las vías de reintento.
¿Dos clientes comparten un client ID y se expulsan mutuamente, provocando reanudaciones de sesión constantes? Consulta el cliente MQTT se desconecta constantemente.
¿El consumidor tarda en confirmar? Un procesamiento largo dentro del callback retrasa el PUBACK y amplía la ventana de reentrega.
¿Cada mensaje lleva un ID creado en el origen?

Los duplicados y la elección de tu broker

ConnectMQ implementa la semántica estándar de QoS 1 y QoS 2, así que los consejos anteriores se aplican directamente. Como los client ID están vinculados a las credenciales emitidas, es más fácil evitar el bucle de reconexión por client ID duplicado que a menudo provoca reentregas masivas. Un broker MQTT gestionado se encarga por ti del almacenamiento de sesiones y de los reintentos; la idempotencia de tus consumidores sigue siendo responsabilidad tuya, en cualquier broker. El tutorial de MQTT es un buen siguiente paso para tener una visión general.

Empieza con ConnectMQ

Crea un broker gratuito en app.connectmq.com con el plan Serverless, sin tarjeta de crédito. Apunta el consumidor con deduplicación hacia él con tu usuario, contraseña y client ID de ConnectMQ, y tus totales seguirán siendo correctos aunque la red no lo sea.

Preguntas frecuentes

¿Por qué recibo el mismo mensaje MQTT dos veces?

Con QoS 1, un acuse de recibo perdido hace que el emisor reenvíe. El receptor vuelve a recibir el mensaje, algo permitido por la garantía de al menos una vez.

¿Puedo confiar en el flag DUP para descartar duplicados?

No. DUP marca una retransmisión en un salto, pero puede que la primera copia nunca llegara, y el broker no traslada el flag a los suscriptores.

¿QoS 0 produce duplicados?

No, el protocolo nunca retransmite QoS 0. En cambio, puede perder mensajes.

¿Durante cuánto tiempo debo recordar los ID de mensaje?

El tiempo suficiente para cubrir los retrasos de reconexión y reanudación de sesión, normalmente minutos. Dimensiónalo a partir del intervalo de expiración de tu sesión más un margen de seguridad.

Cómo gestionar mensajes MQTT duplicados con QoS 1