Cómo probar las garantías de entrega de QoS 1 y QoS 2

Para probar las garantías de entrega de MQTT, publica mensajes con número de secuencia, corta la conexión mientras hay mensajes en vuelo y cuenta lo que llega. QoS 1 aprueba cuando no falta ningún número de secuencia (los duplicados están bien). QoS 2 aprueba cuando no falta nada y no hay duplicados. Todo lo demás —sesiones persistentes, paquetes de confirmación, límites de cola del broker— sirve para que ese experimento sea honesto.

Esta guía muestra cómo observar el flujo de confirmaciones, un arnés en Python que cuenta huecos y duplicados, los fallos que merece la pena inyectar y cómo interpretar los resultados.

Qué promete cada nivel

Dos detalles deciden la mayoría de resultados. Primero, QoS es por salto: el QoS del publicador cubre publicador → broker, el QoS de la suscripción cubre broker → suscriptor, y el QoS efectivo de entrega es el menor de los dos. Un publish en QoS 2 hacia una suscripción QoS 1 es una entrega QoS 1. Segundo, las garantías solo sobreviven a una desconexión si sobrevive la sesión; si no, el broker tira el estado en vuelo. La guía de niveles de QoS cubre la teoría; aquí la ponemos a prueba.

Paso 1: observa las confirmaciones

Antes de automatizar nada, mira los paquetes. Los clientes de Mosquitto imprimen cada paquete de control con -d:

# Terminal 1: subscribe at QoS 2 with debug output
mosquitto_sub -h broker.connectmq.com -p 1883 -i testmqtt-qos-sub -t 'testmqtt/qos/demo' -q 2 -d

# Terminal 2: publish one QoS 2 message with debug output
mosquitto_pub -h broker.connectmq.com -p 1883 -i testmqtt-qos-pub -t 'testmqtt/qos/demo' -m 'hello' -q 2 -d

El publicador imprime un flujo parecido a este:

Client testmqtt-qos-pub sending CONNECT
Client testmqtt-qos-pub received CONNACK (0)
Client testmqtt-qos-pub sending PUBLISH (d0, q2, r0, m1, 'testmqtt/qos/demo', ... (5 bytes))
Client testmqtt-qos-pub received PUBREC (Mid: 1)
Client testmqtt-qos-pub sending PUBREL (m1)
Client testmqtt-qos-pub received PUBCOMP (Mid: 1)
Client testmqtt-qos-pub sending DISCONNECT

d0 es el flag DUP, q2 el QoS y m1 el packet identifier. Repite con -q 1 y deberías ver un único PUBACK en lugar del handshake de tres pasos. En el lado del suscriptor, revisa el valor q del PUBLISH recibido: si te suscribiste en QoS 1, mostrará q1 aunque el publicador usara QoS 2.

Para inspeccionar más a fondo, captura el tráfico y filtra en Wireshark con mqtt.msgtype >= 3 && mqtt.msgtype <= 7 (de PUBLISH a PUBCOMP) o con mqtt.dupflag == 1 para ver solo las retransmisiones. En MQTT 5 las confirmaciones también llevan un reason code: un PUBACK con 0x10 (No matching subscribers) es un éxito, no un error; consulta los códigos en la referencia de reason codes.

Paso 2: un contador de huecos y duplicados

El arnés tiene tres partes: un suscriptor con sesión persistente que registra cada número de secuencia recibido, un publicador que envía un rango conocido y cuenta las confirmaciones, y un pequeño script de análisis.

import sys
import paho.mqtt.client as mqtt

QOS = int(sys.argv[1]) if len(sys.argv) > 1 else 1
TOPIC = "testmqtt/qos-test/seq"
log = open(f"received-q{QOS}.log", "a", buffering=1)  # line-buffered: survives kill -9

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"connected: {reason_code}, session present: {flags.session_present}")
    client.subscribe(TOPIC, qos=QOS)

def on_message(client, userdata, msg):
    log.write(f"{msg.payload.decode()} {int(msg.dup)}\n")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"testmqtt-qos-sub-q{QOS}", clean_session=False)
# client.username_pw_set("username", "password")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=10)
client.loop_forever()   # reconnects automatically after a network error

import sys, time
import paho.mqtt.client as mqtt

QOS = int(sys.argv[1]) if len(sys.argv) > 1 else 1
N = int(sys.argv[2]) if len(sys.argv) > 2 else 2000
TOPIC = "testmqtt/qos-test/seq"
acked = 0

def on_publish(client, userdata, mid, reason_code, properties):
    global acked
    acked += 1          # QoS 1: on PUBACK, QoS 2: on PUBCOMP

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"testmqtt-qos-pub-q{QOS}", clean_session=False)
client.on_publish = on_publish
client.connect("broker.connectmq.com", 1883, keepalive=10)
client.loop_start()

for seq in range(N):
    client.publish(TOPIC, str(seq), qos=QOS)   # queued by paho while disconnected
    time.sleep(0.005)

deadline = time.time() + 60
while acked < N and time.time() < deadline:
    time.sleep(0.2)
print(f"published {N}, acknowledged {acked}")
client.disconnect()
client.loop_stop()

#!/usr/bin/env bash
LOG=${1:-received-q1.log}; N=${2:-2000}
echo "received lines : $(wc -l < "$LOG")"
echo "unique         : $(cut -d' ' -f1 "$LOG" | sort -u | wc -l)"
echo "duplicates     : $(cut -d' ' -f1 "$LOG" | sort | uniq -d | wc -l)"
echo "with DUP flag  : $(awk '$2 == 1' "$LOG" | wc -l)"
echo "missing        : $(comm -23 <(seq 0 $((N - 1)) | sort) <(cut -d' ' -f1 "$LOG" | sort -u) | wc -l)"

Ejecuta python sub.py 1, después python pub.py 1 y luego ./analyze.sh received-q1.log. Empieza con una ejecución sin fallos: debe mostrar 2.000 únicos y 0 perdidos. Si la línea base ya falla, arregla eso primero (QoS incorrecto en la suscripción, una ACL, un límite de tamaño de mensaje) antes de inyectar fallos.

Paso 3: rompe cosas en pleno vuelo

Ejecuta cada escenario mientras pub.py está enviando y luego analiza. Usa un archivo de log nuevo en cada ejecución.

BROKER_IP=$(dig +short broker.connectmq.com | head -1)
sudo iptables -I OUTPUT -p tcp -d "$BROKER_IP" --dport 1883 -j DROP
sleep 20
sudo iptables -D OUTPUT -p tcp -d "$BROKER_IP" --dport 1883 -j DROP

El agujero negro es el fallo más realista: no llega ningún RST, así que el cliente solo se entera por el keep-alive (unas 1,5 × los 10 segundos de keep-alive en el lado del broker) y entonces se reconecta. Al reconectar, el suscriptor debería imprimir session present: True y el broker debería reenviar los mensajes sin confirmar con DUP=1, seguidos de la cola pendiente. Tienes más técnicas de corte en probar QoS con un corte de red.

La caída del proceso es el caso interesante para QoS 2. paho-mqtt guarda su estado de sesión en memoria, así que kill -9 borra la mitad del handshake QoS 2 que corresponde al cliente. Según si una librería entrega el mensaje a tu código al recibir PUBLISH o al recibir PUBREL, una caída en el momento equivocado puede producir un duplicado o un mensaje perdido. Exactly-once es una propiedad del protocolo entre dos extremos que conservan su estado: si tu aplicación no debe procesar dos veces tras una caída, deduplica por un ID de mensaje en tu propio almacenamiento.

Interpretar los resultados

• Mensajes perdidos en QoS 1/2 tras un corte de red: la sesión no persistió. Revisa session present en el log, el client ID (debe ser idéntico al reconectar) y el session expiry de MQTT 5.
• Mensajes perdidos solo tras cortes largos: la cola por cliente del broker se llenó y descartó mensajes por política (por ejemplo, max_queued_messages en Mosquitto). Es un límite de capacidad, no un bug de QoS; consulta cola de mensajes llena.
• Duplicados en QoS 2: comprueba el QoS efectivo en el suscriptor (q2 en mosquitto_sub -d) y busca caídas o un segundo cliente que comparta el client ID. Más causas en mensajes duplicados.
• Duplicados sin flag DUP en QoS 1: normalmente el publicador reenvió tras reconectar, lo que para el broker es un mensaje nuevo. Es normal en at-least-once.
• acknowledged menor que published: el publicador se rindió antes de vaciar la cola pendiente; amplía el plazo o revisa si hay un límite de inflight en el broker.

Una comprobación manual rápida

Para una comprobación rápida sin scripts, suscríbete en QoS 1 en el cliente MQTT online, publica desde una terminal con mosquitto_pub -q 1 -d y compara el QoS que aparece en ambos lados. Después desconecta el cliente del navegador, publica algunos mensajes más y reconecta con el mismo client ID y clean session desactivado: los mensajes en cola deberían llegar.

Observa entregas QoS en tu navegador

QoS 1 se demuestra con cero huecos y QoS 2 con cero huecos y cero duplicados, medidos con números de secuencia a través de desconexiones reales, no supuestos a partir de la especificación. Guarda el arnés en tu repositorio y vuelve a ejecutarlo cada vez que cambies el broker, la librería cliente o la configuración de sesión.

Preguntas frecuentes