Schlechte Netzwerke für MQTT mit Toxiproxy simulieren

Toxiproxy ist ein kleiner TCP-Proxy von Shopify, der zwischen deinem MQTT-Client und dem Broker sitzt und die Verbindung gezielt kaputt macht: Latenz und Jitter hinzufügen, Bandbreite drosseln, den Verkehr einfrieren oder den Socket zurücksetzen. Gesteuert wird alles über eine HTTP-API, sodass deine Tests Fehler im Code ein- und ausschalten können.

Diese Anleitung stellt Toxiproxy vor einen MQTT-Broker, geht die für MQTT relevanten Toxics durch und macht daraus automatisierte Tests für Keep-Alive, Reconnect und QoS-Verhalten.

Warum ein Proxy statt Kabel ziehen

WLAN trennen oder den Broker killen sind grobe Werkzeuge. Sie lassen sich schlecht timen, kaum in CI wiederholen und simulieren nur einen Fehler: „alles weg“. Echte Netze versagen subtiler – ein Mobilfunklink mit zwei Sekunden zusätzlicher Latenz, eine NAT-Box, die einen inaktiven Flow still verwirft, ein überlasteter Uplink mit 5 KB pro Sekunde. Genau diese Bedingungen decken Bugs in Reconnect-Schleifen, Inflight-Fenstern und Timeouts auf.

Toxiproxy liefert jeden dieser Fehler als benanntes, umkehrbares „Toxic“ auf einer einzelnen TCP-Verbindung. Broker und andere Clients bleiben unberührt, und der Test kann genau prüfen, was dein Code getan hat, während das Toxic aktiv war.

Toxiproxy vor dem Broker starten

Am schnellsten geht es mit dem offiziellen Container. Port 8474 ist die Steuer-API; jeder Proxy, den du anlegst, bekommt einen eigenen Listen-Port.

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

Lege einen Proxy an, der auf 21883 lauscht und an den Broker weiterleitet:

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

Richte einen beliebigen MQTT-Client auf localhost:21883 statt auf den Broker und prüfe, dass weiterhin Daten fließen:

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

Denselben Topic kannst du im Online-MQTT-Client beobachten, der direkt mit dem Broker verbunden ist: Über den Proxy veröffentlichte Nachrichten erscheinen auch dort, sodass du leicht siehst, wann ein Toxic sie verzögert oder blockiert.

Für einen Test-Stack deklarierst du den Proxy in einer Konfigurationsdatei, damit er existiert, sobald der Container startet:

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 }
]

Läuft der Broker in derselben Compose-Datei (etwa ein Mosquitto-Container mit Testcontainers oder ein Service namens mosquitto), setze den Upstream auf mosquitto:1883.

Die für MQTT relevanten Toxics

Toxics werden pro Proxy und pro Richtung hinzugefügt. downstream ist Broker → Client (CONNACK, PUBACK, zugestellte Nachrichten), upstream ist Client → Broker (CONNECT, PUBLISH, PINGREQ). Standard ist 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

Einige MQTT-spezifische Hinweise:

• latency verzögert jeden Datenblock. Downstream angewendet verlängert es jeden Roundtrip: CONNACK, SUBACK und PUBACK kommen verspätet an. Damit findest du zu knappe Timeouts – ein Connect-Timeout von 1 s schlägt fehl, sobald die Latenz darüber liegt.
• bandwidth (rate in KB/s) zeigt, was passiert, wenn du schneller publishst, als die Leitung transportieren kann. Bei QoS 1 stauen sich offene Nachrichten im Inflight-Fenster des Clients; siehe MQTT Flow Control.
• timeout mit timeout: 0 stoppt alle Daten, lässt den Socket aber offen. Keine Seite bekommt einen TCP-Fehler – die klassische halboffene Verbindung, die nur MQTT Keep Alive bemerkt.
• reset_peer schließt die Verbindung nach timeout ms mit einem TCP-RST, der Client sieht ECONNRESET.
• slicer zerlegt die Daten in kleine Segmente mit Pausen dazwischen. Ein korrekter Client setzt MQTT-Pakete über mehrere Reads zusammen; ein naiver Embedded-Parser vielleicht nicht.
• limit_data schließt die Verbindung nach bytes Bytes – praktisch, um ein großes PUBLISH in der Mitte abzuschneiden.

Um die Verbindung komplett zu trennen, deaktivierst du den Proxy: curl -X POST localhost:8474/proxies/mqtt -d '{"enabled": false}'. Bestehende Verbindungen werden geschlossen, neue abgelehnt, bis du ihn wieder aktivierst.

Automatisieren: ein Keep-Alive-Test in Python

Die Steuer-API ist reines HTTP, eine Client-Bibliothek brauchst du nicht. Dieser Test (paho-mqtt 2.x) verbindet sich mit 5 Sekunden Keep-Alive über den Proxy, friert die Leitung ein und prüft, dass der Client das innerhalb des Keep-Alive-Fensters bemerkt.

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"

        # Verbindung in beide Richtungen einfrieren, ohne sie zu schließen
        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()

Starte ihn mit pytest -s test_keepalive.py. Die Erkennung sollte etwa ein bis zwei Keep-Alive-Intervalle dauern, je nachdem, wo im Ping-Zyklus das Einfrieren passiert. Dauert es deutlich länger oder passiert es nie, setzt dein Client Keep-Alive nicht durch – ein Bug, der echte Geräte stundenlang mit nichts „verbunden“ lässt. Eine sauberere Fixture-Struktur zeigt pytest-Fixtures für MQTT.

Dasselbe Muster deckt weitere Szenarien ab:

1. Reconnect – Proxy deaktivieren, warten, wieder aktivieren und prüfen, dass der Client sich neu verbindet und neu abonniert (Details in MQTT-Reconnect-Tests).
2. QoS-1-Wiederholung – limit_data auf upstream setzen, eine große QoS-1-Nachricht publishen, das Toxic entfernen und prüfen, dass der Subscriber sie genau einmal bzw. mindestens einmal erhält, je nachdem, was dein Design erwartet.
3. Timeout-Budget – 3 s Downstream-Latenz hinzufügen und prüfen, dass dein Connect- oder Request-Response-Timeout sauber mit einer klaren Fehlermeldung scheitert, statt zu hängen.

MQTT über TLS und WebSockets durch Toxiproxy

Toxiproxy leitet rohe Bytes weiter, daher funktionieren TLS (8883) und WebSockets (8083, 8084 mit Pfad /mqtt) ohne Änderungen: Lege einen Proxy mit Upstream broker.connectmq.com:8883 an, und der Handshake läuft direkt durch.

Nur die Hostname-Prüfung musst du lösen. Der Client verbindet sich mit localhost, das Zertifikat nennt aber den Broker. Möglichkeiten:

• Behalte den Broker-Hostnamen im Client und leite ihn per /etc/hosts oder Docker-extra_hosts auf dem Test-Runner auf den Proxy um.
• Setze den TLS-Servernamen explizit – in MQTT.js über die Option servername, in Python, indem du den Socket mit server_hostname gleich dem Broker-Namen wrappst.
• Nutze für Netzwerkfehler-Tests reines TCP und halte TLS für einen separaten TLS-Verbindungstest bereit.

Deaktiviere nie die Zertifikatsprüfung, nur damit der Proxy in einem Test funktioniert, der auch mit Produktionseinstellungen läuft – diese Gewohnheit landet gern im echten Code.

Häufige Fehler

• Latenz nur upstream hinzufügen. Viele interessante Timeouts warten auf die Antwort des Brokers. Teste beide Richtungen, oder zuerst downstream.
• timeout: 0 mit dem Deaktivieren des Proxys verwechseln. Deaktivieren schließt die Sockets (der Client sieht sofort einen Disconnect); timeout: 0 friert sie ein (nur Keep-Alive rettet dich).
• Nur mit QoS 0 testen. QoS 0 verbirgt, was mit unbestätigten Nachrichten passiert. Wiederhole kritische Szenarien mit QoS 1.
• Über einen öffentlichen Broker messen. Geteilte Sandboxes bringen eigene Schwankungen mit. Für alles, worauf du asserten willst, nutze einen Broker unter deiner Kontrolle.

Nachrichten über den Proxy im Browser beobachten

Toxiproxy macht aus „im Feld bricht es manchmal ab“ einen Test, der bei jedem Commit läuft. Starte mit einem Proxy, einem Keep-Alive-Test mit eingefrorener Leitung und einem Reconnect-Test; sobald die grün sind, ergänze Latenz- und Bandbreitenprofile, die den Netzen deiner Geräte entsprechen, etwa die Mobilfunkprofile aus MQTT in langsamen, instabilen Mobilfunknetzen testen.

Häufig gestellte Fragen