Store-and-forward MQTT : mettre en tampon les données edge pendant les coupures et les rejouer dans l'ordre

Le store-and-forward MQTT est le schéma qui préserve l'intégrité des données industrielles lorsque le réseau entre la périphérie et le cloud tombe. Au lieu de perdre les messages tant que le broker est injoignable, l'équipement edge les écrit sur un stockage local et les rejoue, dans l'ordre, dès le retour de la connexion. Cet article explique le fonctionnement de ce schéma, les choix de conception qui comptent, et propose une implémentation Python exécutable reposant sur SQLite.

Pourquoi MQTT seul ne suffit pas

MQTT offre de solides garanties de livraison : acquittements QoS 1 et 2, sessions persistantes et retained messages. Mais toutes n'entrent en jeu qu'une fois le message sorti de l'équipement. Voyez ce qui se passe lors d'une coupure WAN de deux heures dans une usine :

• La passerelle continue d'interroger les machines et de produire des valeurs.
• Le client MQTT ne peut pas se connecter : il n'y a donc aucune session dans laquelle livrer.
• La plupart des bibliothèques clientes conservent les messages non envoyés en mémoire, avec une taille limitée, et les perdent lors d'un redémarrage ou d'une coupure de courant.

Les sessions persistantes côté broker aident dans l'autre sens : elles conservent les messages destinés à un abonné hors ligne. Elles ne font rien pour un éditeur qui ne parvient pas à joindre le broker. C'est ce vide que comble le store-and-forward, et c'est pourquoi la mise en tampon figure parmi les missions essentielles de toute passerelle edge IIoT.

Fonctionnement du store-and-forward MQTT

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

1. Chaque message est d'abord écrit dans une file locale durable, avec un numéro de séquence et l'horodatage source.
2. Un relais (forwarder) lit les messages les plus anciens et les publie en QoS 1.
3. Ce n'est qu'après l'acquittement du broker (PUBACK) que le message est supprimé de la file.
4. Quand la connexion échoue, le relais s'arrête, la file grossit et les sources de données continuent sans être affectées.
5. À la reconnexion, le relais vide l'arriéré du plus ancien au plus récent à un débit maîtrisé, puis poursuit avec les données en direct.

Faire transiter toutes les données par la file, et pas seulement celles produites pendant les coupures, simplifie l'ordonnancement : il n'existe qu'un seul chemin, toujours en premier entré, premier sorti.

Les choix de conception qui comptent

Dimensionner le tampon

Estimez-le à partir de trois chiffres : le nombre de messages par seconde en fonctionnement normal, la taille moyenne d'un message surcoût compris, et la plus longue coupure que vous voulez absorber. Par exemple, 20 messages par seconde de 200 octets sur une coupure de 24 heures représentent environ 345 Mo de payload, plus le surcoût de stockage. Si c'est trop, réduire le trafic à la source avec le report by exception diminue souvent considérablement le besoin.

Choisir une politique de rejet

Quand le tampon est plein, il faut choisir : rejeter les données les plus anciennes (conserver l'état récent) ou les plus récentes (conserver un historique continu depuis le début de la coupure). Pour la surveillance, rejeter les plus anciennes est généralement le bon choix. Pour la conformité réglementaire ou les enregistrements de lots, vous préférerez peut-être cesser d'accepter les nouvelles données de faible priorité et préserver le début. Dans tous les cas, publiez un compteur de messages rejetés afin que le trou soit visible.

Le rejeu et la dernière valeur retenue

Rejouer des heures d'historique sur un topic que les tableaux de bord utilisent aussi comme « valeur courante » peut faire défiler une jauge à travers d'anciennes valeurs. Deux solutions courantes : envoyer l'arriéré sur un topic distinct comme .../history, ou inclure l'horodatage source et laisser les consommateurs ignorer les valeurs plus anciennes pour l'affichage. Avec MQTT 5, vous pouvez aussi définir un intervalle d'expiration (message expiry interval) sur les valeurs qui ne valent plus rien une fois périmées, pour que le broker les supprime au lieu de les livrer en retard.

Une implémentation store-and-forward en Python

Le script suivant utilise SQLite comme file durable et paho-mqtt 2.x comme client. Un thread producteur génère une mesure par seconde, quelle que soit la connectivité. Le relais publie les lignes les plus anciennes en QoS 1, attend chaque acquittement, supprime les lignes livrées et limite le débit de rejeu pour rester sous la limite de 10 messages par seconde du serveur MQTT public gratuit, à réserver aux tests et prototypes.

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          # limite du tampon ; les lignes les plus anciennes sont supprimées au-delà
REPLAY_RATE = 8             # messages par seconde (le broker public en autorise 10)
ACK_TIMEOUT = 10            # secondes d'attente du 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():
    """Source de données simulée : ne se bloque jamais sur le réseau."""
    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)
# Production : 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                      # permet aux consommateurs d'éliminer les doublons
        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                              # garder la ligne, réessayer plus tard dans l'ordre
        delete(seq)
        time.sleep(interval)
    with db_lock:
        backlog = db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
    if backlog > 1:
        print("Backlog:", backlog)

Pour le tester, lancez le script, observez le topic dans un autre terminal, puis coupez votre réseau pendant une minute ou deux :

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

Hors ligne, le script affiche la déconnexion et continue de mettre en file. Au retour du réseau, le client se reconnecte automatiquement et l'arriéré se vide dans l'ordre des séquences. Le champ ts indique quand chaque valeur a été mesurée, et seq augmente sans trou. Vous pouvez même arrêter complètement le script pendant la coupure : au redémarrage, la file SQLite est toujours là.

Remarques sur l'implémentation

• Un seul message en vol à la fois rend l'ordonnancement trivial et suffit en débit pour la plupart des passerelles. Pour un débit plus élevé, publiez une petite fenêtre et supprimez les lignes à mesure que leurs acquittements arrivent.
• Au moins une fois, pas exactement une fois. Si le PUBACK est perdu après que le broker a accepté le message, la ligne est renvoyée. Le champ seq permet à un historien ou à un abonné d'écarter les doublons. Le QoS 2 de MQTT éviterait les doublons sur le réseau, mais pas à travers un redémarrage de la passerelle : des identifiants au niveau applicatif restent donc utiles. Consultez les niveaux de QoS MQTT expliqués pour les compromis.
• Le mode WAL de SQLite permet aux lectures et aux écritures de se dérouler en parallèle avec une bonne durabilité.
• Le rejeu à débit limité protège le broker et les consommateurs en aval. Les offres mutualisées imposent des limites par client, et l'offre Serverless gratuite de ConnectMQ prend en charge le QoS 1, exactement ce dont ce schéma a besoin.

Check-list store-and-forward MQTT

1. Écrivez chaque message sur disque avant de tenter de l'envoyer.
2. Incluez un horodatage source et un numéro de séquence dans chaque payload.
3. Ne supprimez du tampon qu'après l'acquittement du broker.
4. Rejouez du plus ancien au plus récent, avec un débit inférieur aux limites du broker et des consommateurs.
5. Définissez une taille maximale de tampon et une politique de rejet, et signalez les rejets.
6. Décidez comment l'historique rejoué interagit avec les topics de « valeur courante ».
7. Testez avec de vraies coupures : débranchez le câble, redémarrez la passerelle en pleine coupure, remplissez le tampon.

Store-and-forward sur réseau cellulaire et sites isolés

Les équipements isolés sur liaison cellulaire perdent la connectivité plus souvent et paient à l'octet. Là, le store-and-forward se combine naturellement avec le regroupement (plusieurs mesures par message) et la compression. Le guide MQTT sur réseau cellulaire traite du réglage du keep alive et des budgets de données pour ces liaisons, et le guide de l'IoT industriel avec MQTT montre la place de la mise en tampon dans une architecture d'usine complète.

Un broker fiable à l'autre bout

Le store-and-forward protège le côté edge de la connexion. Côté cloud, il vous faut un broker joignable, sécurisé par TLS et dimensionné pour les rafales de rejeu. Un broker MQTT managé vous décharge de l'exploitation du broker, et le guide du broker MQTT dans le cloud explique comment comparer les options.

Démarrer avec ConnectMQ

Pointez vos passerelles à tampon vers ConnectMQ : TLS sur le port 8883, identifiants émis, expiration des messages MQTT 5 et un explorateur de topics en direct pour regarder un arriéré se vider après une coupure. Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire.

Questions fréquentes

MQTT intègre-t-il le store-and-forward ?

Pas pour les éditeurs qui ne parviennent pas à joindre le broker. Le QoS et les sessions persistantes fonctionnent une fois le message arrivé au broker ; la mise en tampon en amont relève de l'application edge.

Faut-il mettre en tampon en mémoire ou sur disque ?

Sur disque. Les tampons en mémoire sont perdus lors d'un redémarrage ou d'une coupure de courant, ce qui survient souvent lors des mêmes événements qui font tomber le réseau.

Comment les consommateurs gèrent-ils les messages en double après un rejeu ?

Incluez un numéro de séquence ou un identifiant unique dans chaque payload et laissez les consommateurs ignorer les identifiants déjà stockés. Les historiens dédupliquent souvent aussi sur l'horodatage et la variable.

Un rejeu important va-t-il surcharger le broker ?

Il peut atteindre les limites de débit ou submerger les abonnés s'il est envoyé d'un coup. Limitez le rejeu à un débit fixe, comme dans l'exemple, et dimensionnez ce débit selon votre offre.