Gérer les messages MQTT dupliqués avec la QoS 1

Les messages dupliqués MQTT ne sont pas un bug de votre broker ; ils sont une conséquence documentée de la QoS 1, qui promet une livraison « au moins une fois ». Lorsqu'un incident réseau avale un accusé de réception, l'émetteur retransmet, et votre consommateur voit deux fois le même message. Dans ce guide de dépannage, vous apprendrez précisément quand surviennent les doublons, ce que le flag DUP vous apprend et ce qu'il ne vous apprend pas, et comment écrire en Python un consommateur idempotent qui les gère correctement.

Pourquoi la QoS 1 produit des doublons

La QoS 1 utilise une poignée de main en deux étapes sur chaque saut : l'émetteur transmet un PUBLISH, le destinataire répond par un PUBACK. Tant que le PUBACK n'est pas arrivé, l'émetteur conserve le message et le renverra. L'article les niveaux de QoS MQTT expliqués couvre les trois niveaux ; ici, nous nous concentrons sur ce qui peut mal tourner avec le niveau 1.

Un doublon est créé chaque fois que le destinataire a reçu le message mais que l'émetteur n'a pas reçu l'accusé de réception :

1. L'appareil publie la mesure n° 42 vers le broker.
2. Le broker la reçoit, la transmet aux abonnés et envoie un PUBACK.
3. La connexion tombe avant que le PUBACK n'atteigne l'appareil.
4. L'appareil se reconnecte avec sa session et renvoie la n° 42 avec le flag DUP activé.
5. Le broker la traite comme un nouveau message et la transmet à nouveau.

La même chose se produit sur le saut broker-abonné. Si l'abonné a traité un message mais que son PUBACK a été perdu, le broker le renvoie lorsque la session de l'abonné reprend.

Notez qu'un doublon peut survenir indépendamment sur chacun des deux sauts : un message publié peut donc occasionnellement atteindre un consommateur plus de deux fois.

Ce que vous apprend le flag DUP

Chaque paquet PUBLISH comporte un bit DUP. Un émetteur l'active lorsqu'il retransmet un paquet QoS 1 ou QoS 2. Il est tentant d'écarter tout message portant le flag DUP, mais c'est une erreur, pour plusieurs raisons :

Le flag DUP est utile pour la journalisation et le diagnostic. Ce n'est pas une clé de déduplication. Une véritable déduplication nécessite un identifiant que vous placez vous-même dans le message.

Reproduire des doublons

Une vraie relivraison dépend de paquets perdus et se déclenche difficilement à la demande. La façon fiable de tester votre consommateur consiste à injecter vous-même des doublons. Sur le serveur MQTT public gratuit broker.connectmq.com, réservé aux tests et aux prototypes, publiez deux fois le même identifiant de message :

for n in 1 2; do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i dup-pub-t9v4 -q 1 \
    -t connectmq/demo/t9v4/meter-3/energy \
    -m '{"msgId":"meter-3-000042","kwh":0.25}'
done

Un consommateur qui additionne kwh sans déduplication affiche maintenant 0,5 au lieu de 0,25. C'est exactement la catégorie de bug que provoque la relivraison QoS 1 en production, simplement plus rare et plus difficile à repérer.

Créer un consommateur idempotent

Un consommateur idempotent produit le même résultat qu'il traite un message une fois ou cinq fois. Il existe deux techniques courantes :

1. Idempotence naturelle. Concevez l'effet pour que la répétition soit sans conséquence : « régler la température à 21,5 » plutôt que « l'augmenter de 0,5 » ; faire un upsert d'une ligne indexée par appareil et horodatage plutôt qu'une insertion.
2. Déduplication explicite. Donnez à chaque message un identifiant unique et mémorisez les identifiants récemment traités.

Voici un consommateur paho-mqtt 2.x complet qui utilise une déduplication explicite avec un cache borné et limité dans le temps :

import json
import time
import uuid
from collections import OrderedDict
import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
TOPIC = "connectmq/demo/t9v4/+/energy"
DEDUP_TTL = 600          # secondes pendant lesquelles un ID est mémorisé
DEDUP_MAX = 50_000       # plafond de mémoire

seen = OrderedDict()     # msgId -> heure de première réception
totals = {}

def already_seen(msg_id):
    now = time.time()
    while seen:
        oldest_id, ts = next(iter(seen.items()))
        if now - ts > DEDUP_TTL or len(seen) > DEDUP_MAX:
            seen.popitem(last=False)
        else:
            break
    if msg_id in seen:
        return True
    seen[msg_id] = now
    return False

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TOPIC, qos=1)

def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload)
        msg_id = data["msgId"]
    except (ValueError, KeyError):
        print("rejected message without msgId")
        return
    if already_seen(msg_id):
        print(f"duplicate {msg_id} (dup flag={msg.dup}) ignored")
        return
    meter = msg.topic.split("/")[-2]
    totals[meter] = totals.get(meter, 0) + data["kwh"]
    print(f"{meter} total={totals[meter]:.2f} kWh")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"dedup-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=30)
client.loop_forever()

Lancez-le, puis exécutez la boucle de double publication de la section précédente. Le premier message met à jour le total ; le second est signalé comme doublon et ignoré.

Choisir un bon identifiant de message

L'identifiant doit être créé par l'éditeur d'origine, une seule fois, et réutilisé à chaque renvoi. Les options :

• Identifiant d'appareil plus numéro de séquence, comme meter-3-000042. Compact et également utile pour les vérifications d'ordre des messages MQTT. Persistez le compteur ou ajoutez un identifiant de démarrage pour qu'il ne se répète pas après un redémarrage.
• UUID généré pour chaque message. Simple et unique au niveau global, au prix de 36 octets.
• Identifiant d'appareil plus horodatage de la mesure, si l'appareil ne produit jamais deux mesures avec le même horodatage.

Ne générez pas l'identifiant dans le consommateur ou dans une couche intermédiaire : à ce stade, un doublon ressemble à un nouveau message.

Faire passer la déduplication à l'échelle

Le cache en mémoire ci-dessus fonctionne pour un seul processus consommateur. Pour des systèmes plus importants :

• Contraintes de base de données. Un index unique sur msg_id transforme les doublons en conflits inoffensifs. Avec un upsert (INSERT ... ON CONFLICT DO NOTHING dans PostgreSQL), c'est souvent la seule déduplication nécessaire.
• Cache partagé. Si plusieurs consommateurs d'un abonnement partagé peuvent recevoir des renvois du même message, l'ensemble des identifiants vus doit être partagé, par exemple dans Redis avec un TTL.
• Dimensionnement du TTL. La fenêtre doit seulement couvrir des délais de relivraison réalistes : votre session expiry plus une marge. Les doublons arrivent rarement à des heures d'intervalle, sauf si les sessions durent aussi longtemps.

Faut-il simplement utiliser la QoS 2 ?

La QoS 2 ajoute une poignée de main en quatre étapes qui garantit une livraison exactement une fois sur chaque saut. Elle coûte davantage d'allers-retours et d'état côté broker, et elle ne vous protège pas contre un appareil qui expire et republie la même mesure comme un nouveau message au niveau applicatif. Sur les plans limités, elle peut même être indisponible ; par exemple, le plan Serverless gratuit de ConnectMQ prend en charge la QoS jusqu'à 1, tandis que le plan Serverless payant ajoute la QoS 2. Consultez la QoS maximale MQTT pour savoir comment les brokers annoncent et dégradent la QoS. Dans la plupart des systèmes, la QoS 1 associée à des consommateurs idempotents est la conception la plus robuste et la moins coûteuse.

Check-list de dépannage

1. Les doublons sont-ils corrélés aux reconnexions ? Vérifiez dans les journaux des clients les déconnexions et les timeouts de keepalive.
2. Votre éditeur effectue-t-il des réessais au niveau applicatif en plus de la QoS 1 ? Cela double les chemins de renvoi.
3. Deux clients partagent-ils un client ID et s'éjectent-ils mutuellement, provoquant des reprises de session constantes ? Consultez le client MQTT qui se déconnecte sans cesse.
4. Le consommateur tarde-t-il à accuser réception ? Un long traitement dans le callback retarde le PUBACK et élargit la fenêtre de relivraison.
5. Chaque message porte-t-il un identifiant créé à la source ?

Les doublons et le choix de votre broker

ConnectMQ implémente la sémantique standard des QoS 1 et QoS 2 : les conseils ci-dessus s'appliquent donc directement. Comme les client IDs sont liés aux identifiants émis, la boucle de reconnexion due à des client IDs en double, souvent à l'origine de relivraisons massives, est plus facile à éviter. Un broker MQTT managé gère pour vous le stockage des sessions et les renvois ; l'idempotence de vos consommateurs reste votre responsabilité, quel que soit le broker. Le tutoriel MQTT est une bonne étape suivante pour avoir une vue d'ensemble.

Démarrer avec ConnectMQ

Créez un broker gratuit sur app.connectmq.com avec le plan Serverless, sans carte bancaire. Faites pointer le consommateur avec déduplication vers lui avec votre nom d'utilisateur, votre mot de passe et votre client ID ConnectMQ, et vos totaux restent justes même quand le réseau ne l'est pas.

Questions fréquentes

Pourquoi est-ce que je reçois deux fois le même message MQTT ?

En QoS 1, un accusé de réception perdu pousse l'émetteur à renvoyer le message. Le destinataire le reçoit à nouveau, ce que permet la garantie « au moins une fois ».

Puis-je me fier au flag DUP pour écarter les doublons ?

Non. DUP marque une retransmission sur un saut, mais la première copie n'est peut-être jamais arrivée, et le broker ne propage pas le flag jusqu'aux abonnés.

La QoS 0 produit-elle des doublons ?

Non, la QoS 0 n'est jamais retransmise par le protocole. Elle peut en revanche perdre des messages.

Combien de temps faut-il mémoriser les identifiants de message ?

Assez longtemps pour couvrir les délais de reconnexion et de reprise de session, généralement quelques minutes. Dimensionnez la durée à partir de votre session expiry interval plus une marge de sécurité.