Reconexión de Paho MQTT: backoff robusto, resuscripción y búfer offline en Python

El comportamiento de reconexión de Paho MQTT marca la diferencia entre un cliente que sobrevive a una red inestable y uno que deja de funcionar en silencio a las 3 de la madrugada. Esta guía explica qué hace paho-mqtt 2.x automáticamente, los tres huecos que deja abiertos, y te da un cliente Python probado con backoff exponencial, resuscripción al conectar, un búfer offline y sesiones persistentes de MQTT 5.

Síntomas de una lógica de reconexión rota

Los errores de reconexión rara vez hacen caer un programa. Se manifiestan como fallos más silenciosos y fáciles de diagnosticar mal:

• El cliente se reconecta tras una caída del Wi-Fi, registra "connected", pero no vuelve a recibir ningún mensaje.
• Un script arrancado mientras la red estaba caída termina de inmediato con un error de socket en lugar de esperar.
• Las lecturas tomadas durante un corte breve desaparecen y dejan huecos en las gráficas.
• Cientos de dispositivos se reconectan exactamente en el mismo momento tras reiniciarse el broker y saturan la red.
• Dos procesos se desconectan mutuamente cada pocos segundos.

Cada uno tiene una causa y una solución concretas.

Lo que Paho hace automáticamente

Cuando ejecutas el bucle de red con loop_start() o loop_forever(), paho-mqtt se reconecta por sí solo cuando cae una conexión ya establecida. La espera entre intentos empieza en min_delay y se duplica tras cada fallo hasta alcanzar max_delay, configurados con reconnect_delay_set(). Los valores por defecto son 1 y 120 segundos. Una conexión con éxito reinicia la espera.

Paho también conserva los mensajes QoS 1 y QoS 2 en vuelo que el broker aún no ha confirmado y los retransmite tras reconectarse, siempre que se reanude la sesión.

Eso cubre mucho, pero no todo. Los huecos restantes son responsabilidad tuya.

Hueco 1: falla la primera conexión

client.connect() realiza la primera conexión de forma síncrona y lanza una excepción, normalmente ConnectionRefusedError, socket.gaierror o TimeoutError, si el broker no es accesible. La reconexión automática nunca arranca porque nunca hubo conexión. Por eso los scripts lanzados al arrancar el equipo, antes de que la red esté lista, suelen morir.

La solución es connect_async() seguido de loop_start(). El intento de conexión pasa al bucle de red, que reintenta con las mismas reglas de backoff.

Hueco 2: las suscripciones desaparecen

Con una sesión limpia, el broker descarta tus suscripciones cuando termina la conexión. Paho no las registra ni se vuelve a suscribir por ti. Si llamas a subscribe() una sola vez después de connect(), el cliente se reconectará tan contento y luego no recibirá nada.

La solución es sencilla e innegociable: suscríbete dentro de on_connect. Ese callback se ejecuta tras cada conexión con éxito, reconexiones incluidas, así que las suscripciones siempre se restauran.

Hueco 3: mensajes publicados sin conexión

Las lecturas generadas mientras el enlace está caído necesitan algún sitio adonde ir. Los mensajes QoS 0 son de tipo "dispara y olvida", así que publicarlos sin conexión los descarta. Confiar en la cola interna de paho para niveles de QoS superiores tampoco es buena idea, porque no tiene límite y es invisible para tu aplicación. Un búfer offline explícito y acotado que vacías tras reconectar te da control sobre la memoria y sobre qué descartar cuando un corte se alarga demasiado.

Un cliente robusto de reconexión Paho MQTT

La clase de abajo cierra los tres huecos y añade jitter al backoff para que una flota de dispositivos no se reconecte al unísono. Usa MQTT 5 con session expiry para que el broker conserve las suscripciones y los mensajes QoS 1 encolados durante cortes breves.

pip install "paho-mqtt>=2.0"

import json
import logging
import random
import threading
import time
import uuid
from collections import deque

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("mqtt")


class ResilientClient:
    def __init__(self, host, port, client_id, subscriptions,
                 username=None, password=None, use_tls=False,
                 session_expiry_s=1800, buffer_size=1000):
        self.host = host
        self.port = port
        self.subscriptions = subscriptions  # lista de (topic, qos)
        self.session_expiry_s = session_expiry_s
        self.offline = deque(maxlen=buffer_size)  # se descarta el más antiguo al llenarse
        self.lock = threading.Lock()
        self.on_payload = None

        c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                        client_id=client_id, protocol=mqtt.MQTTv5)
        if username:
            c.username_pw_set(username, password)
        if use_tls:
            c.tls_set()
        # Backoff exponencial; un mínimo aleatorio dispersa las flotas que se reconectan
        c.reconnect_delay_set(min_delay=random.randint(1, 5), max_delay=60)
        c.on_connect = self._on_connect
        c.on_disconnect = self._on_disconnect
        c.on_connect_fail = self._on_connect_fail
        c.on_message = self._on_message
        self.client = c

    def start(self):
        props = Properties(PacketTypes.CONNECT)
        props.SessionExpiryInterval = self.session_expiry_s
        self.client.connect_async(self.host, self.port, keepalive=30,
                                  clean_start=False, properties=props)
        self.client.loop_start()

    def stop(self):
        self.client.disconnect()
        self.client.loop_stop()

    # --- callbacks -------------------------------------------------------
    def _on_connect(self, client, userdata, flags, reason_code, properties):
        if reason_code.is_failure:
            log.warning("connect refused: %s", reason_code)
            return
        log.info("connected, session_present=%s", flags.session_present)
        if self.subscriptions:
            client.subscribe(self.subscriptions)  # resuscribirse siempre
        self._flush()

    def _on_disconnect(self, client, userdata, flags, reason_code, properties):
        if reason_code == 0:
            log.info("clean disconnect")
        else:
            log.warning("connection lost: %s, reconnecting", reason_code)

    def _on_connect_fail(self, client, userdata):
        log.warning("connection attempt failed, retrying with backoff")

    def _on_message(self, client, userdata, msg):
        if self.on_payload:
            self.on_payload(msg.topic, msg.payload)

    # --- publicación -----------------------------------------------------
    def publish(self, topic, payload, qos=1, retain=False):
        if self.client.is_connected():
            info = self.client.publish(topic, payload, qos=qos, retain=retain)
            if info.rc == mqtt.MQTT_ERR_SUCCESS:
                return True
        with self.lock:
            self.offline.append((topic, payload, qos, retain))
        return False

    def _flush(self):
        with self.lock:
            pending = list(self.offline)
            self.offline.clear()
        if pending:
            log.info("flushing %d buffered messages", len(pending))
        for topic, payload, qos, retain in pending:
            self.client.publish(topic, payload, qos=qos, retain=retain)


if __name__ == "__main__":
    run_id = uuid.uuid4().hex[:8]
    prefix = f"connectmq/demo/{run_id}"
    rc = ResilientClient(
        host="broker.connectmq.com", port=1883,
        client_id=f"resilient-{run_id}",
        subscriptions=[(f"{prefix}/cmd/#", 1)],
    )
    rc.on_payload = lambda t, p: log.info("command on %s: %s", t, p)
    rc.start()
    print(f"publishing to {prefix}/telemetry, send commands to {prefix}/cmd/...")
    try:
        n = 0
        while True:
            n += 1
            payload = json.dumps({"seq": n, "ts": int(time.time() * 1000)})
            sent = rc.publish(f"{prefix}/telemetry", payload, qos=1)
            log.info("seq %d %s (buffered: %d)", n, "sent" if sent else "buffered", len(rc.offline))
            time.sleep(2)
    except KeyboardInterrupt:
        rc.stop()

Por qué está cada pieza

El número de secuencia de cada payload facilita las pruebas: tras un corte, los huecos o duplicados en seq muestran exactamente qué pasó. Como QoS 1 es entrega al menos una vez, los consumidores deben tolerar algún duplicado ocasional, por ejemplo ignorando los números de secuencia que ya han visto. Nuestra explicación de los niveles de QoS de MQTT cubre por qué pueden darse duplicados.

Cómo probar la lógica de reconexión

No te fíes de un código de reconexión que no hayas roto a propósito. Ejecuta el script y un suscriptor en paralelo:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-<run-id> -t "connectmq/demo/<run-id>/#" -v

Después prueba cada modo de fallo por turnos:

1. Caída de red. Desactiva el Wi-Fi o desenchufa el cable durante 30 segundos. El registro muestra "connection lost", luego intentos fallidos con esperas crecientes y después "connected" y "flushing N buffered messages". El suscriptor ve llegar de golpe los números de secuencia almacenados.
2. Puerto bloqueado en Linux. Simula un broker inaccesible mientras el resto de la red funciona:

sudo iptables -A OUTPUT -p tcp --dport 1883 -j DROP
# espera 60 segundos y luego elimina la regla
sudo iptables -D OUTPUT -p tcp --dport 1883 -j DROP

Los paquetes descartados son una prueba más dura que una conexión rechazada, porque el cliente solo se da cuenta por el timeout del keep-alive, aproximadamente una vez y media el intervalo de keep-alive. 3. Arrancar sin conexión. Desconecta la red antes de arrancar el script. Con connect_async(), espera y se conecta cuando vuelve la red en lugar de terminar. 4. Toma de control de la sesión. Arranca un segundo cliente con el mismo client ID, por ejemplo mosquitto_sub -i resilient-<run-id> -t x. El broker desconecta al cliente original, que se reconecta y a su vez desconecta al intruso. Es el síntoma de "dos procesos peleándose"; verlo a propósito facilita reconocerlo más adelante. 5. Comandos durante un corte. Mientras el script está desconectado, publica un comando QoS 1 en su topic cmd. Como se conservó la sesión, el comando se entrega cuando el cliente vuelve.

Reason codes que conviene gestionar

MQTT 5 te dice por qué terminó una conexión. Registra el reason code y reacciona ante aquellos que reintentar no va a solucionar.

Reintentar sin fin ante un rechazo 0x86 solo inunda los registros del broker, y muchos brokers, ConnectMQ incluido, bloquean una dirección IP tras varios inicios de sesión fallidos. En _on_connect, plantéate llamar a client.disconnect() y avisar a una persona cuando el código indique un problema permanente.

Ajustes para producción

En despliegues reales, conéctate por TLS en el puerto 8883 con credenciales emitidas por tu broker. Contra un broker MQTT gestionado de ConnectMQ, construye el cliente así:

rc = ResilientClient(
    host="broker.connectmq.com", port=8883, use_tls=True,
    client_id="YOUR_CLIENT_ID", username="YOUR_USERNAME", password="YOUR_PASSWORD",
    subscriptions=[("connectmq/factory1/production/line1/+/cmd", 1)],
)

Ten en cuenta algunos límites. En el servidor MQTT público gratuito, las sesiones expiran como máximo a los 30 minutos, así que la expiración de 1.800 segundos usada arriba es el valor útil máximo allí. El plan Serverless gratuito admite QoS hasta 1, que es lo que usa este cliente. Los planes de pago añaden QoS 2 si necesitas entrega exactamente una vez. El broker también publica el estado de conexión en $SYS/broker/connection/{clientId}/state, algo práctico para confirmar que un dispositivo volvió tras un corte.

Para un contexto más amplio, nuestro tutorial de paho-mqtt en Python cubre los fundamentos, la guía de la API de callbacks de paho-mqtt 2.0 explica las firmas de callbacks de la versión 2 usadas aquí, y la comparativa de bibliotecas cliente MQTT te ayuda si estás eligiendo entre paho y bibliotecas asíncronas. La documentación oficial de paho-mqtt enumera todas las opciones relacionadas con la reconexión.

Empieza con ConnectMQ

Prueba hoy tu lógica de reconexión contra el servidor MQTT público gratuito y pasa a credenciales privadas cuando despliegues. ConnectMQ ofrece un broker MQTT gestionado compatible con MQTT 5, con TLS, session expiry y reason codes claros, para que tus clientes reciban información útil cuando algo falla. Crea un broker Serverless gratuito en app.connectmq.com sin tarjeta de crédito.

Preguntas frecuentes

¿Se reconecta paho-mqtt automáticamente?

Sí, una vez establecida una conexión y con el bucle de red en marcha mediante loop_start() o loop_forever(). No reintenta una primera llamada a connect() fallida ni se vuelve a suscribir por ti.

¿Por qué mi cliente paho deja de recibir mensajes tras reconectarse?

Las suscripciones se perdieron con la sesión antigua. Mueve tus llamadas a subscribe() a on_connect para que se ejecuten tras cada conexión con éxito.

¿Qué espera de reconexión debo usar?

Empieza entre 1 y 5 segundos y limítala a entre 60 y 120 segundos. Aleatorizar la espera inicial por dispositivo evita que toda una flota se reconecte en el mismo instante.

¿Debo usar clean_start=False?

Úsalo junto con un session expiry interval cuando quieras que el broker conserve las suscripciones y encole los mensajes QoS 1 durante cortes breves. Mantén estable el client ID, porque la sesión está ligada a él.