Simuler un réseau dégradé pour MQTT avec Toxiproxy

Toxiproxy est un petit proxy TCP de Shopify qui s'intercale entre votre client MQTT et le broker et permet de casser la connexion volontairement : ajouter de la latence et du jitter, brider la bande passante, geler le trafic ou réinitialiser le socket. Le tout se pilote par une API HTTP, si bien que vos tests peuvent activer et désactiver les pannes dans le code.

Ce guide place Toxiproxy devant un broker MQTT, passe en revue les toxics utiles pour MQTT et les transforme en tests automatisés du keep-alive, de la reconnexion et du comportement QoS.

Pourquoi un proxy plutôt que débrancher le câble

Couper le Wi-Fi ou tuer le broker sont des outils grossiers. Difficiles à synchroniser, difficiles à rejouer en CI, ils ne simulent qu'une panne : « tout a disparu ». Les vrais réseaux échouent de façon plus subtile : un lien mobile qui ajoute deux secondes de latence, un NAT qui abandonne silencieusement un flux inactif, un uplink saturé qui ne transmet que 5 Ko par seconde. Ce sont ces conditions qui révèlent les bugs dans les boucles de reconnexion, les fenêtres inflight et les timeouts.

Toxiproxy fournit chacune de ces pannes sous forme de « toxic » nommé et réversible, sur une seule connexion TCP. Le broker et vos autres clients ne sont pas affectés, et le test peut vérifier précisément ce qu'a fait votre code pendant que le toxic était actif.

Lancer Toxiproxy devant le broker

Le plus rapide est le conteneur officiel. Le port 8474 est l'API de contrôle ; chaque proxy créé reçoit son propre port d'écoute.

docker run -d --name toxiproxy \
  -p 8474:8474 -p 21883:21883 \
  ghcr.io/shopify/toxiproxy

Créez un proxy qui écoute sur 21883 et relaie vers le broker :

curl -s -X POST localhost:8474/proxies -d '{
  "name": "mqtt",
  "listen": "0.0.0.0:21883",
  "upstream": "broker.connectmq.com:1883"
}'

Pointez n'importe quel client MQTT vers localhost:21883 au lieu du broker et vérifiez que le trafic passe toujours :

mosquitto_sub -h localhost -p 21883 -t 'testmqtt/toxiproxy/#' -v &
mosquitto_pub -h localhost -p 21883 -t 'testmqtt/toxiproxy/hello' -m 'through the proxy'

Vous pouvez suivre le même topic dans le client MQTT en ligne connecté directement au broker : les messages publiés via le proxy y apparaissent aussi, ce qui permet de voir facilement quand un toxic les retarde ou les bloque.

Pour une stack de test, déclarez le proxy dans un fichier de configuration afin qu'il existe dès le démarrage du conteneur :

services:
  toxiproxy:
    image: ghcr.io/shopify/toxiproxy
    command: ["-host=0.0.0.0", "-config=/config/toxiproxy.json"]
    volumes:
      - ./toxiproxy.json:/config/toxiproxy.json:ro
    ports:
      - "8474:8474"
      - "21883:21883"

[
  { "name": "mqtt", "listen": "0.0.0.0:21883", "upstream": "broker.connectmq.com:1883", "enabled": true }
]

Si le broker tourne dans le même fichier Compose (par exemple un conteneur Mosquitto avec Testcontainers ou un service nommé mosquitto), utilisez mosquitto:1883 comme upstream.

Les toxics utiles pour MQTT

Les toxics s'ajoutent par proxy et par sens. downstream correspond à broker → client (CONNACK, PUBACK, messages livrés), upstream à client → broker (CONNECT, PUBLISH, PINGREQ). La valeur par défaut est downstream.

# 800 ms ± 200 ms on everything the broker sends back
curl -s -X POST localhost:8474/proxies/mqtt/toxics -d '{
  "name": "slow", "type": "latency", "stream": "downstream",
  "attributes": { "latency": 800, "jitter": 200 }
}'

# Throttle the client's uplink to 4 KB/s
curl -s -X POST localhost:8474/proxies/mqtt/toxics -d '{
  "name": "thin", "type": "bandwidth", "stream": "upstream",
  "attributes": { "rate": 4 }
}'

curl -s localhost:8474/proxies/mqtt/toxics       # list active toxics
curl -s -X DELETE localhost:8474/proxies/mqtt/toxics/slow
curl -s -X POST localhost:8474/reset              # remove all toxics, re-enable proxies

Quelques remarques propres à MQTT :

• latency retarde chaque bloc de données. En downstream, il allonge chaque aller-retour : CONNACK, SUBACK et PUBACK arrivent en retard. Utilisez-le pour débusquer les timeouts trop serrés — un connect timeout d'1 s échoue dès que la latence le dépasse.
• bandwidth (rate en Ko/s) montre ce qui se passe quand vous publiez plus vite que le lien ne peut transporter. En QoS 1, les messages en attente s'accumulent dans la fenêtre inflight du client ; voir le contrôle de flux MQTT.
• timeout avec timeout: 0 bloque toutes les données mais laisse le socket ouvert. Aucun côté ne reçoit d'erreur TCP : c'est la connexion semi-ouverte classique, que seul le keep alive MQTT remarquera.
• reset_peer ferme la connexion par un TCP RST après timeout ms, ce qui donne ECONNRESET côté client.
• slicer découpe les données en petits segments espacés d'un délai. Un client correct réassemble les paquets MQTT sur plusieurs lectures ; un parser embarqué naïf, pas forcément.
• limit_data ferme la connexion après bytes octets — pratique pour couper un gros PUBLISH en deux.

Pour couper complètement la connexion, désactivez le proxy : curl -X POST localhost:8474/proxies/mqtt -d '{"enabled": false}'. Les connexions existantes sont fermées et les nouvelles refusées jusqu'à sa réactivation.

Automatiser : un test de keep-alive en Python

L'API de contrôle est du HTTP simple, aucune bibliothèque cliente n'est nécessaire. Ce test (paho-mqtt 2.x) se connecte via le proxy avec un keep-alive de 5 secondes, gèle le lien et vérifie que le client s'en aperçoit dans la fenêtre de keep-alive.

import json, threading, time, urllib.request
import paho.mqtt.client as mqtt

API = "http://localhost:8474"

def toxiproxy(method, path, body=None):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(API + path, data=data, method=method)
    with urllib.request.urlopen(req) as resp:
        return resp.read()

def test_client_detects_half_open_connection():
    toxiproxy("POST", "/reset")
    connected, disconnected = threading.Event(), threading.Event()

    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="toxi-keepalive-test")
    client.on_connect = lambda c, u, f, rc, p: connected.set()
    client.on_disconnect = lambda c, u, f, rc, p: disconnected.set()
    client.connect("localhost", 21883, keepalive=5)
    client.loop_start()
    try:
        assert connected.wait(5), "no CONNACK through the proxy"

        # Geler la connexion dans les deux sens sans la fermer
        for stream in ("upstream", "downstream"):
            toxiproxy("POST", "/proxies/mqtt/toxics", {
                "name": f"freeze_{stream}", "type": "timeout",
                "stream": stream, "attributes": {"timeout": 0},
            })
        started = time.monotonic()

        assert disconnected.wait(15), "client never noticed the dead link"
        print(f"dead link detected after {time.monotonic() - started:.1f} s")
    finally:
        toxiproxy("POST", "/reset")
        client.loop_stop()
        client.disconnect()

Lancez-le avec pytest -s test_keepalive.py. La détection devrait prendre environ un à deux intervalles de keep-alive, selon le moment du cycle de ping où survient le gel. Si cela prend beaucoup plus longtemps, ou n'arrive jamais, votre client n'applique pas le keep-alive : un bug qui laisserait de vrais appareils « connectés » à rien pendant des heures. Pour une organisation des fixtures plus propre, voir les fixtures pytest pour MQTT.

Le même schéma couvre d'autres scénarios :

1. Reconnexion — désactivez le proxy, attendez, réactivez-le et vérifiez que le client se reconnecte et se réabonne (détails dans les tests de reconnexion MQTT).
2. Retransmission QoS 1 — ajoutez limit_data en upstream, publiez un gros message QoS 1, retirez le toxic et vérifiez que le subscriber le reçoit exactement une fois ou au moins une fois, selon ce que prévoit votre conception.
3. Budget de timeouts — ajoutez 3 s de latence en downstream et vérifiez que votre timeout de connexion ou de request-response échoue proprement avec une erreur claire au lieu de rester bloqué.

MQTT sur TLS et WebSockets à travers Toxiproxy

Toxiproxy relaie des octets bruts, donc TLS (8883) et WebSockets (8083, 8084 avec le chemin /mqtt) fonctionnent sans changement : créez un proxy dont l'upstream est broker.connectmq.com:8883 et le handshake passe tel quel.

Seule la vérification du nom d'hôte demande un réglage. Le client se connecte à localhost, mais le certificat désigne le broker. Options :

• Gardez le nom d'hôte du broker dans le client et faites-le pointer vers le proxy via /etc/hosts ou extra_hosts de Docker sur le runner de test.
• Fixez explicitement le server name TLS — avec MQTT.js, l'option servername ; en Python, en enveloppant le socket avec server_hostname égal au nom du broker.
• Utilisez TCP en clair pour les tests de pannes réseau et gardez TLS pour un test de connexion TLS séparé.

Ne désactivez jamais la vérification des certificats pour faire fonctionner le proxy dans un test qui tourne aussi avec la configuration de production : cette habitude finit souvent dans le vrai code.

Erreurs fréquentes

• Ajouter de la latence uniquement en upstream. Beaucoup de timeouts intéressants attendent la réponse du broker. Testez les deux sens, ou downstream d'abord.
• Confondre timeout: 0 et désactivation du proxy. Désactiver ferme les sockets (le client voit une déconnexion immédiate) ; timeout: 0 les gèle (seul le keep-alive vous sauve).
• Tester uniquement en QoS 0. QoS 0 masque ce qui arrive aux messages non acquittés. Rejouez les scénarios critiques en QoS 1.
• Mesurer via un broker public. Les sandboxes partagés ajoutent leur propre variabilité. Pour tout ce que vous voulez vérifier dans un assert, utilisez un broker que vous contrôlez.

Suivre les messages du proxy dans le navigateur

Toxiproxy transforme « ça se déconnecte parfois sur le terrain » en un test exécutable à chaque commit. Commencez par un proxy, un test de keep-alive avec lien gelé et un test de reconnexion ; une fois qu'ils passent, ajoutez des profils de latence et de bande passante proches des réseaux réellement utilisés par vos appareils, comme les profils cellulaires de tester MQTT sur des réseaux cellulaires instables.

Questions fréquentes