Cómo probar la expiración de sesión y las colas offline en MQTT

Una sesión MQTT persistente permite que un dispositivo se desconecte y aun así reciba todos los mensajes QoS 1 que se perdió. Que funcione depende de cuatro ajustes fáciles de equivocar: el client ID, Clean Start, el Session Expiry Interval y los límites de cola del broker. La forma fiable de probarlo: suscríbete con una sesión persistente, desconecta, publica, reconecta y comprueba Session Present y los mensajes exactos que recibes. Después, repite cuando la sesión ya debería haber expirado.

Esta guía te da una prueba manual con mosquitto_sub, un script de Python que comprueba los tres comportamientos y una lista de los patrones de fallo que te vas a encontrar.

Qué tiene que cumplirse para que una sesión sobreviva

El valor por defecto de Session Expiry Interval en MQTT 5 es 0. Un cliente que pone Clean Start a false pero olvida el intervalo pierde igualmente su sesión en cuanto se desconecta. La diferencia entre ambas versiones del protocolo se explica en Clean Session vs Clean Start, y el concepto general en las sesiones persistentes de MQTT.

El broker también puede sobrescribir tu valor: si permite menos de lo que pediste, un broker MQTT 5 devuelve su propio Session Expiry Interval en el CONNACK. Regístralo siempre.

Qué guarda el broker en una sesión

• Las suscripciones del cliente (con su QoS y sus opciones).
• Los mensajes QoS 1 y QoS 2 que llegaron mientras el cliente estaba offline.
• Los mensajes QoS 1 y QoS 2 en vuelo que no se confirmaron por completo.
• Opcionalmente, mensajes QoS 0, si el broker está configurado para encolarlos.

Todo lo que está fuera de esta lista, como los mensajes retenidos, vive con independencia de la sesión. Un mensaje con su propia expiración de mensaje puede desaparecer de la cola antes que la sesión.

Prueba manual con mosquitto_sub

Elige un client ID y un topic únicos y crea la sesión:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -i sess-demo-41c2 -c -x 120 -q 1 \
  -t 'testmqtt/session/41c2/#' -v
# press Ctrl+C after "subscribed"

for i in 1 2 3 4 5; do
  mosquitto_pub -h broker.connectmq.com -V mqttv5 -q 1 \
    -t testmqtt/session/41c2/cmd -m "msg $i"
done

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 \
  -i sess-demo-41c2 -c -x 120 -q 1 \
  -t 'testmqtt/session/41c2/#' -v -d

Con -d ves todos los paquetes intercambiados. No todas las versiones de Mosquitto imprimen el flag Session Present, así que aquí la prueba es la cola pendiente: msg 1 a msg 5, en orden, justo después del CONNACK. El test de Python de más abajo comprueba el flag de forma explícita. Ahora detén el cliente, vuelve a publicar, espera más de 120 segundos y ejecuta el paso 3 otra vez: la sesión tiene que haber desaparecido y no debe llegar ninguno de los mensajes nuevos.

Hay dos cosas que confunden aquí. Primero, el topic del paso 3 no importa para la cola pendiente: la entrega la hace la suscripción guardada. Segundo, si ejecutas el paso 3 mientras otro cliente usa el mismo ID, uno de los dos queda fuera. Eso es una toma de control de sesión (session takeover), no una prueba de sesión.

Una prueba de sesión automatizada en Python

Este script de paho-mqtt 2.x ejecuta tres pruebas de una vez: la sesión sobrevive y encola mensajes en orden, la sesión expira de verdad, y qué ocurre cuando la cola se desborda.

import sys, time, uuid
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

HOST, PORT = "broker.connectmq.com", 1883
RUN = uuid.uuid4().hex[:8]
BURST = int(sys.argv[1]) if len(sys.argv) > 1 else 2000   # for the queue-limit test

def connect(cid, clean_start, session_expiry, inbox=None):
    """Connect with MQTT 5, wait for CONNACK, return (client, connack state)."""
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=cid, protocol=mqtt.MQTTv5)
    state = {}
    c.on_connect = lambda cl, u, flags, rc, props: state.update(
        present=flags.session_present, rc=str(rc),
        expiry=getattr(props, "SessionExpiryInterval", None))   # set only if the broker overrides it
    if inbox is not None:
        c.on_message = lambda cl, u, msg: inbox.append(int(msg.payload))
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = session_expiry
    c.connect(HOST, PORT, clean_start=clean_start, properties=props)
    c.loop_start()
    for _ in range(100):
        if state: break
        time.sleep(0.05)
    return c, state

def close(c, end_session=False):
    props = None
    if end_session:
        props = Properties(PacketTypes.DISCONNECT)
        props.SessionExpiryInterval = 0
    c.disconnect(properties=props)
    c.loop_stop()

def go_offline(cid, topic, session_expiry):
    c, state = connect(cid, True, session_expiry)
    c.subscribe(topic, qos=1)
    time.sleep(1)
    close(c)
    return state

def publish(topic, count):
    pub, _ = connect(f"sess-pub-{RUN}", True, 0)
    info = None
    for i in range(count):
        info = pub.publish(topic, str(i), qos=1)
    info.wait_for_publish(timeout=60)
    close(pub)

def drain(inbox, quiet=2.0):
    """Wait until no new message has arrived for `quiet` seconds."""
    last, since = -1, time.time()
    while time.time() - since < quiet:
        if len(inbox) != last:
            last, since = len(inbox), time.time()
        time.sleep(0.1)

def report(name, ok, detail=""):
    print(f"{'PASS' if ok else 'FAIL'} {name} {detail}")

# Test 1: the session survives a disconnect and queues QoS 1 messages
cid, topic = f"sess-a-{RUN}", f"testmqtt/session/{RUN}/a"
go_offline(cid, topic, 300)
publish(topic, 50)
inbox = []
c, s = connect(cid, False, 300, inbox)
drain(inbox)
close(c, end_session=True)
report("session present after reconnect", s.get("present") is True, s)
report("all 50 queued messages delivered in order", inbox == list(range(50)), f"got {len(inbox)}")
if not s.get("present"):
    raise SystemExit("ABORT: the broker does not keep sessions, the other tests are meaningless")

# Test 2: the session really expires
cid, topic = f"sess-b-{RUN}", f"testmqtt/session/{RUN}/b"
go_offline(cid, topic, 5)
time.sleep(10)
c, s = connect(cid, False, 5)
close(c, end_session=True)
report("session gone after 5 s expiry", s.get("present") is False, s)

# Test 3: what happens when the offline queue overflows
cid, topic = f"sess-c-{RUN}", f"testmqtt/session/{RUN}/c"
go_offline(cid, topic, 300)
publish(topic, BURST)
inbox = []
c, s = connect(cid, False, 300, inbox)
drain(inbox, quiet=3)
close(c, end_session=True)
kept = f"first={inbox[0]} last={inbox[-1]}" if inbox else ""
print(f"INFO queue limit: published {BURST}, delivered {len(inbox)} {kept}")

Ejecútalo con python session_test.py o pasa una ráfaga mayor, por ejemplo python session_test.py 5000. En un broker que conserva sesiones, una ejecución sana se ve así:

PASS session present after reconnect {'present': True, 'rc': 'Success', 'expiry': None}
PASS all 50 queued messages delivered in order got 50
PASS session gone after 5 s expiry {'present': False, 'rc': 'Success', 'expiry': None}
INFO queue limit: published 2000, delivered 1000 first=0 last=999

La última línea es informativa, no pass/fail. Te dice el límite real de la cola y la política de descarte: first=0 last=999 significa que el broker conservó los mensajes más antiguos y descartó los más nuevos; first=1000 last=1999 significaría lo contrario. Si expiry muestra un número, el broker sobrescribió el intervalo que pediste.

El script termina cada sesión con Session Expiry Interval = 0 en el DISCONNECT, para que las pruebas no dejen sesiones huérfanas. Solo puedes bajar el intervalo a 0 de esta forma si era distinto de cero en el CONNECT. Subirlo desde 0 en el DISCONNECT es un error de protocolo.

Probar los límites de cola en tu propio broker

Todos los brokers limitan cuánto guardan por cliente offline, porque mil dispositivos muertos con colas ilimitadas agotarían su memoria. En Mosquitto, los ajustes relevantes son:

persistence true
persistence_location /mosquitto/data/

# remove persistent sessions of clients offline for more than 2 days
persistent_client_expiration 2d

# per-client offline queue (0 bytes = no byte limit)
max_queued_messages 1000
max_queued_bytes 0

# QoS 0 messages are not queued for offline clients unless enabled
queue_qos0_messages false

Prueba el límite que configuraste, no el que supones: publica una ráfaga de max_queued_messages + 500, reconecta y compara. Luego decide qué necesitan tus dispositivos. Un sensor al que solo le importa el último comando debería usar expiración de mensajes o estado retenido, no una cola enorme. Cuando los clientes ven errores de cola llena, cola de mensajes MQTT llena explica cómo solucionarlo. El generador de configuración de Mosquitto puede crear estos ajustes por ti.

Prueba también un reinicio del broker mientras el cliente está offline. Sin persistencia activada, Mosquitto guarda las sesiones solo en memoria, y un reinicio las pierde todas.

Patrones de fallo y qué significan

Para una guía de depuración más amplia, consulta por qué una sesión MQTT no persiste.

Probar en el navegador

Para una comprobación visual rápida, abre el cliente MQTT online, conecta con un client ID fijo, Clean Start desactivado y una expiración de sesión, suscríbete en QoS 1 y desconecta. Publica unos cuantos mensajes desde una terminal, reconecta en el navegador y observa cómo llega lo pendiente.

Prueba una sesión persistente en el navegador

Una prueba de sesión tarda un minuto en ejecutarse y ahorra horas de depurar reportes de "mi dispositivo no recibió el comando". Guarda el script en tu repositorio, ejecútalo contra cada broker que despliegues y trata la línea del límite de cola como parte de tu plan de capacidad.

Preguntas frecuentes