Tester l'expiration de session et les files hors ligne MQTT

Une session MQTT persistante permet à un appareil de se déconnecter tout en recevant ensuite chaque message QoS 1 manqué. Que cela fonctionne dépend de quatre réglages faciles à rater : le client ID, Clean Start, le Session Expiry Interval et les limites de file du broker. La méthode fiable pour le tester : abonnez-vous avec une session persistante, déconnectez-vous, publiez, reconnectez-vous, puis vérifiez Session Present et les messages exacts reçus. Recommencez ensuite une fois que la session aurait dû expirer.

Ce guide vous donne un test manuel avec mosquitto_sub, un script Python qui vérifie les trois comportements et une liste des schémas de panne que vous rencontrerez.

Ce qui doit être vrai pour qu'une session survive

La valeur par défaut du Session Expiry Interval en MQTT 5 est 0. Un client qui met Clean Start à false mais oublie l'intervalle perd quand même sa session dès qu'il se déconnecte. La différence entre les deux versions du protocole est expliquée dans Clean Session vs Clean Start, et le concept général dans les sessions persistantes MQTT.

Le broker peut aussi remplacer votre valeur : s'il autorise moins que demandé, un broker MQTT 5 renvoie son propre Session Expiry Interval dans le CONNACK. Journalisez-le toujours.

Ce que le broker stocke dans une session

• Les abonnements du client (avec leur QoS et leurs options).
• Les messages QoS 1 et QoS 2 arrivés pendant que le client était hors ligne.
• Les messages QoS 1 et QoS 2 en cours qui n'ont pas été entièrement acquittés.
• Éventuellement les messages QoS 0, si le broker est configuré pour les mettre en file.

Tout ce qui sort de cette liste, comme les messages retenus, vit indépendamment de la session. Un message doté de sa propre expiration de message peut disparaître de la file avant la session.

Test manuel avec mosquitto_sub

Choisissez un client ID et un topic uniques, puis créez la session :

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

Avec -d, vous voyez chaque paquet échangé. Toutes les versions de Mosquitto n'affichent pas le flag Session Present : la preuve, ici, c'est le retard reçu, msg 1 à msg 5, dans l'ordre, juste après le CONNACK. Le test Python ci-dessous vérifie le flag explicitement. Arrêtez maintenant le client, publiez de nouveau, attendez plus de 120 secondes et relancez l'étape 3 : la session doit avoir disparu et aucun des nouveaux messages ne doit arriver.

Deux pièges reviennent souvent. D'abord, le topic de l'étape 3 n'a aucune importance pour le retard : c'est l'abonnement stocké qui le livre. Ensuite, lancer l'étape 3 pendant qu'un autre client utilise le même ID en éjecte un des deux. C'est une prise de contrôle de session (session takeover), pas un test de session.

Un test de session automatisé en Python

Ce script paho-mqtt 2.x exécute trois tests d'un coup : la session survit et met les messages en file dans l'ordre, la session expire vraiment, et ce qui se passe quand la file déborde.

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}")

Lancez-le avec python session_test.py ou passez une rafale plus grande, par exemple python session_test.py 5000. Sur un broker qui conserve les sessions, une exécution saine ressemble à ceci :

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 dernière ligne est informative, pas un pass/fail. Elle indique la limite effective de la file et la politique de suppression : first=0 last=999 signifie que le broker a gardé les plus anciens messages et supprimé les plus récents ; first=1000 last=1999 signifierait l'inverse. Si expiry affiche un nombre, le broker a remplacé l'intervalle demandé.

Le script termine chaque session avec Session Expiry Interval = 0 dans le DISCONNECT, pour que les tests ne laissent pas de sessions orphelines. On ne peut ramener l'intervalle à 0 de cette façon que s'il était non nul dans le CONNECT. L'augmenter depuis 0 dans le DISCONNECT est une erreur de protocole.

Tester les limites de file sur votre propre broker

Chaque broker limite ce qu'il stocke par client hors ligne, car mille appareils morts avec des files illimitées épuiseraient sa mémoire. Dans Mosquitto, les réglages concernés sont :

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

Testez la limite que vous avez configurée, pas celle que vous supposez : publiez une rafale de max_queued_messages + 500, reconnectez-vous et comparez. Décidez ensuite de ce dont vos appareils ont besoin. Un capteur qui ne s'intéresse qu'à la dernière commande devrait utiliser l'expiration de message ou un état retenu, pas une file énorme. Quand des clients voient des erreurs de file pleine, file de messages MQTT pleine présente les correctifs. Le générateur de configuration Mosquitto peut produire ces réglages pour vous.

Testez aussi un redémarrage du broker pendant que le client est hors ligne. Sans persistance activée, Mosquitto ne garde les sessions qu'en mémoire, et un redémarrage les perd toutes.

Schémas de panne et leur signification

Pour un diagnostic plus complet, voir pourquoi une session MQTT ne persiste pas.

Tester dans le navigateur

Pour une vérification visuelle rapide, ouvrez le client MQTT en ligne, connectez-vous avec un client ID fixe, Clean Start désactivé et une expiration de session, abonnez-vous en QoS 1 puis déconnectez-vous. Publiez quelques messages depuis un terminal, reconnectez-vous dans le navigateur et regardez le retard arriver.

Essayer une session persistante dans le navigateur

Un test de session prend une minute et évite des heures de débogage sur des tickets « mon appareil a raté la commande ». Gardez le script dans votre dépôt, lancez-le sur chaque broker que vous déployez et considérez la ligne de limite de file comme une partie de votre plan de capacité.

Questions fréquentes