Store and forward MQTT: búfer de datos edge durante las caídas y reenvío en orden

Store and forward MQTT es el patrón que mantiene intactos los datos industriales cuando cae la red entre el edge y la nube. En lugar de descartar mensajes mientras el broker no está accesible, el dispositivo edge los escribe en un almacenamiento local y los reenvía, en orden, cuando vuelve la conexión. Este artículo explica cómo funciona el patrón y qué decisiones de diseño importan, y te ofrece una implementación en Python lista para ejecutar y respaldada por SQLite.

Por qué MQTT por sí solo no basta

MQTT tiene sólidas funciones de entrega: confirmaciones de QoS 1 y 2, sesiones persistentes y retained messages. Pero todas empiezan a funcionar solo cuando un mensaje ya ha salido del dispositivo. Piensa en lo que ocurre durante una caída de la WAN de dos horas en una planta:

• El gateway sigue sondeando las máquinas y generando valores.
• El cliente MQTT no puede conectarse, así que no hay ninguna sesión en la que entregar.
• La mayoría de las librerías cliente guardan los mensajes no enviados en memoria, con un límite de tamaño, y los pierden con un reinicio o un corte de alimentación.

Las sesiones persistentes en el broker ayudan en la otra dirección: retienen mensajes para un suscriptor desconectado. No hacen nada por un publicador que no puede llegar al broker. Ese hueco es el que cubre store and forward, y por eso el búfer figura como una tarea básica de todo gateway edge IIoT.

Cómo funciona store and forward MQTT

  Data source --> [ write to local queue (disk) ] --> forwarder --> MQTT broker
                         ^                               |
                         |      delete only after        |
                         +------- PUBACK received -------+

1. Cada mensaje se escribe primero en una cola local duradera, con un número de secuencia y la marca de tiempo de origen.
2. Un reenviador lee los mensajes más antiguos y los publica con QoS 1.
3. Solo cuando el broker confirma (PUBACK) se elimina el mensaje de la cola.
4. Cuando la conexión falla, el reenviador se detiene, la cola crece y las fuentes de datos siguen funcionando sin verse afectadas.
5. Al reconectar, el reenviador vacía el atraso empezando por lo más antiguo a un ritmo controlado y después continúa con los datos en vivo.

Hacer pasar todos los datos por la cola, y no solo los generados durante las caídas, simplifica el orden: solo hay un camino y siempre es FIFO (primero en entrar, primero en salir).

Decisiones de diseño que importan

Dimensionar el búfer

Estímalo a partir de tres números: mensajes por segundo en funcionamiento normal, tamaño medio del mensaje incluida la sobrecarga y la caída más larga que quieras soportar. Por ejemplo, 20 mensajes por segundo de 200 bytes durante una caída de 24 horas suponen unos 345 MB de payload, más la sobrecarga del almacenamiento. Si es demasiado, reducir el tráfico en origen con report by exception suele recortar drásticamente el requisito.

Elegir una política de descarte

Cuando el búfer está lleno, tienes que elegir: descartar los datos más antiguos (conservar el estado reciente) o descartar los más nuevos (conservar un histórico continuo desde el inicio de la caída). Para monitorización, descartar los más antiguos suele ser lo correcto. Para registros de cumplimiento o de lotes, quizá prefieras dejar de aceptar datos nuevos de baja prioridad y conservar el principio. En cualquier caso, publica un contador de mensajes descartados para que el hueco sea visible.

El reenvío y el último valor retenido

Reenviar horas de histórico a un topic que los paneles también usan como "valor actual" puede hacer que un indicador salte por valores antiguos. Dos soluciones habituales son enviar el atraso a un topic separado como .../history o incluir la marca de tiempo de origen y dejar que los consumidores ignoren los valores más antiguos al mostrarlos. Con MQTT 5 también puedes fijar un intervalo de expiración del mensaje en los valores que no sirven de nada una vez obsoletos, para que el broker los descarte en lugar de entregarlos tarde.

Una implementación de store and forward en Python

El siguiente script usa SQLite como cola duradera y paho-mqtt 2.x como cliente. Un hilo productor genera una lectura cada segundo, haya o no conectividad. El reenviador publica las filas más antiguas con QoS 1, espera cada confirmación, elimina las filas entregadas y limita el ritmo de reenvío para quedarse por debajo del límite de 10 mensajes por segundo del servidor MQTT público gratuito, que es solo para pruebas y no para datos de producción.

import json
import sqlite3
import threading
import time
import random
from datetime import datetime, timezone

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
CLIENT_ID = "saf-gateway-2b7c"
TOPIC = "connectmq/demo/saf-2b7c/plant1/compressor1/telemetry"
DB_PATH = "saf_queue.db"
MAX_ROWS = 500_000          # límite del búfer; por encima se descartan las filas más antiguas
REPLAY_RATE = 8             # mensajes por segundo (el broker público permite 10)
ACK_TIMEOUT = 10            # segundos de espera del PUBACK

db_lock = threading.Lock()
db = sqlite3.connect(DB_PATH, check_same_thread=False)
db.execute("PRAGMA journal_mode=WAL")
db.execute("""CREATE TABLE IF NOT EXISTS queue (
                seq INTEGER PRIMARY KEY AUTOINCREMENT,
                topic TEXT NOT NULL,
                payload TEXT NOT NULL)""")
db.commit()


def enqueue(topic, data):
    with db_lock:
        db.execute("INSERT INTO queue (topic, payload) VALUES (?, ?)",
                   (topic, json.dumps(data)))
        count = db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
        if count > MAX_ROWS:
            db.execute("DELETE FROM queue WHERE seq IN "
                       "(SELECT seq FROM queue ORDER BY seq LIMIT ?)", (count - MAX_ROWS,))
        db.commit()


def oldest(n=50):
    with db_lock:
        return db.execute("SELECT seq, topic, payload FROM queue ORDER BY seq LIMIT ?",
                          (n,)).fetchall()


def delete(seq):
    with db_lock:
        db.execute("DELETE FROM queue WHERE seq = ?", (seq,))
        db.commit()


def producer():
    """Fuente de datos simulada: nunca se bloquea por la red."""
    while True:
        enqueue(TOPIC, {
            "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
            "pressure_bar": round(7 + random.gauss(0, 0.1), 3),
            "motor_temp_c": round(65 + random.gauss(0, 0.5), 2),
        })
        time.sleep(1)


def on_connect(client, userdata, flags, reason_code, properties):
    print("Connected:", reason_code)


def on_disconnect(client, userdata, flags, reason_code, properties):
    print("Disconnected:", reason_code, "- buffering locally")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=60)
# Producción: client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"); client.tls_set(); PORT = 8883
client.connect_async(BROKER, PORT, keepalive=30)
client.loop_start()

threading.Thread(target=producer, daemon=True).start()

interval = 1.0 / REPLAY_RATE
while True:
    if not client.is_connected():
        time.sleep(1)
        continue
    rows = oldest()
    if not rows:
        time.sleep(0.5)
        continue
    for seq, topic, payload in rows:
        data = json.loads(payload)
        data["seq"] = seq                      # permite a los consumidores eliminar duplicados
        try:
            info = client.publish(topic, json.dumps(data), qos=1)
            info.wait_for_publish(timeout=ACK_TIMEOUT)
        except (RuntimeError, ValueError) as exc:
            print("Publish failed:", exc)
            break
        if not info.is_published():
            print("No PUBACK, will retry from seq", seq)
            break                              # conserva la fila y reintenta después, en orden
        delete(seq)
        time.sleep(interval)
    with db_lock:
        backlog = db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
    if backlog > 1:
        print("Backlog:", backlog)

Para probarlo, ejecuta el script, observa el topic en otro terminal y desconecta tu red durante uno o dos minutos:

mosquitto_sub -h broker.connectmq.com -p 1883 -i saf-2b7c-viewer -t "connectmq/demo/saf-2b7c/#" -v

Mientras está sin conexión, el script muestra la desconexión y sigue encolando. Cuando vuelve la red, el cliente se reconecta automáticamente y el atraso se vacía en orden de secuencia. El campo ts indica cuándo se midió cada valor y seq aumenta sin huecos. También puedes detener el script por completo durante la caída; al reiniciarlo, la cola SQLite sigue ahí.

Notas sobre la implementación

• Un solo mensaje en vuelo a la vez mantiene el orden trivial y es lo bastante rápido para la mayoría de los gateways. Para un mayor rendimiento, publica una pequeña ventana y elimina las filas a medida que llegan sus confirmaciones.
• Al menos una vez, no exactamente una vez. Si el PUBACK se pierde después de que el broker aceptara el mensaje, la fila se reenvía. El campo seq permite a un historiador o suscriptor descartar duplicados. QoS 2 de MQTT evitaría duplicados en la red, pero no tras un reinicio del gateway, así que sigue mereciendo la pena tener IDs a nivel de aplicación. Consulta los niveles de QoS de MQTT explicados para conocer las ventajas e inconvenientes.
• El modo WAL de SQLite permite que lecturas y escrituras avancen en paralelo con buena durabilidad.
• El reenvío limitado protege al broker y a los consumidores posteriores. Los planes compartidos tienen límites por cliente, y el plan Serverless gratuito de ConnectMQ admite QoS 1, que es exactamente lo que necesita este patrón.

Checklist de store and forward MQTT

1. Escribe cada mensaje en disco antes de intentar enviarlo.
2. Incluye una marca de tiempo de origen y un número de secuencia en cada payload.
3. Elimina del búfer solo después de que el broker confirme.
4. Reenvía primero lo más antiguo, con un ritmo limitado por debajo de los límites del broker y de los consumidores.
5. Define un tamaño máximo de búfer y una política de descarte, e informa de los descartes.
6. Decide cómo interactúa el histórico reenviado con los topics de "valor actual".
7. Prueba con caídas reales: desconecta el cable, reinicia el gateway en mitad de la caída, llena el búfer.

Store and forward en redes celulares y emplazamientos remotos

Los activos remotos con enlaces celulares pierden la conectividad con más frecuencia y pagan por byte. Allí, store and forward se combina de forma natural con el agrupamiento (varias lecturas por mensaje) y la compresión. La guía de MQTT sobre red celular trata el ajuste del keep alive y los presupuestos de datos para esos enlaces, y la guía de IoT industrial con MQTT muestra dónde encaja el búfer en una arquitectura de planta completa.

Un broker fiable al otro lado

Store and forward protege el lado edge de la conexión. En el lado de la nube necesitas un broker accesible, protegido con TLS y dimensionado para las ráfagas de reenvío. Un broker MQTT gestionado te quita de encima la operación del broker, y la guía del broker MQTT en la nube explica cómo comparar opciones.

Empieza con ConnectMQ

Apunta tus gateways con búfer a ConnectMQ: TLS en el puerto 8883, credenciales emitidas, expiración de mensajes de MQTT 5 y un explorador de topics en vivo para ver cómo se vacía un atraso después de una caída. Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito.

Preguntas frecuentes

¿Tiene MQTT store and forward integrado?

No para los publicadores que no pueden llegar al broker. QoS y las sesiones persistentes funcionan una vez que el mensaje llega al broker; el búfer previo es tarea de la aplicación edge.

¿Debo usar un búfer en memoria o en disco?

En disco. Los búferes en memoria se pierden con un reinicio o un corte de alimentación, algo que a menudo ocurre en los mismos eventos que tumban la red.

¿Cómo gestionan los consumidores los mensajes duplicados tras el reenvío?

Incluye un número de secuencia o un ID único en cada payload y deja que los consumidores ignoren los IDs que ya han almacenado. Los historiadores también suelen eliminar duplicados por marca de tiempo y tag.

¿Sobrecargará el broker un reenvío grande?

Puede chocar con los límites de frecuencia o saturar a los suscriptores si se envía todo de golpe. Limita el reenvío a un ritmo fijo, como en el ejemplo, y ajústalo a tu plan.