Zustellgarantien von QoS 1 und QoS 2 testen

Um die Zustellgarantien von MQTT zu testen, veröffentlichst du durchnummerierte Nachrichten, unterbrichst die Verbindung, während Nachrichten unterwegs sind, und zählst, was ankommt. QoS 1 besteht, wenn keine Sequenznummer fehlt (Duplikate sind in Ordnung). QoS 2 besteht, wenn nichts fehlt und nichts doppelt ankommt. Alles andere – Persistent Sessions, Bestätigungspakete, Queue-Limits des Brokers – dient dazu, dieses Experiment ehrlich zu machen.

Dieser Guide zeigt, wie du den Bestätigungsfluss beobachtest, ein Python-Harness, das Lücken und Duplikate zählt, die Fehlerszenarien, die sich lohnen, und wie du die Ergebnisse liest.

Was jede Stufe verspricht

Zwei Details entscheiden die meisten Testergebnisse. Erstens gilt QoS pro Hop: Das QoS des Publishers deckt Publisher → Broker ab, das QoS der Subscription Broker → Subscriber, und das effektive Zustell-QoS ist das niedrigere von beiden. Ein QoS-2-Publish an eine QoS-1-Subscription ist eine QoS-1-Zustellung. Zweitens überstehen die Garantien einen Verbindungsabbruch nur, wenn die Session ihn übersteht – sonst wirft der Broker den In-Flight-State weg. Der Guide zu den QoS-Stufen erklärt die Theorie; hier testen wir sie.

Schritt 1: die Bestätigungen beobachten

Bevor du irgendetwas automatisierst, schau dir die Pakete an. Die Mosquitto-Clients geben mit -d jedes Control Packet aus:

# Terminal 1: subscribe at QoS 2 with debug output
mosquitto_sub -h broker.connectmq.com -p 1883 -i testmqtt-qos-sub -t 'testmqtt/qos/demo' -q 2 -d

# Terminal 2: publish one QoS 2 message with debug output
mosquitto_pub -h broker.connectmq.com -p 1883 -i testmqtt-qos-pub -t 'testmqtt/qos/demo' -m 'hello' -q 2 -d

Der Publisher gibt einen Ablauf ähnlich wie diesen aus:

Client testmqtt-qos-pub sending CONNECT
Client testmqtt-qos-pub received CONNACK (0)
Client testmqtt-qos-pub sending PUBLISH (d0, q2, r0, m1, 'testmqtt/qos/demo', ... (5 bytes))
Client testmqtt-qos-pub received PUBREC (Mid: 1)
Client testmqtt-qos-pub sending PUBREL (m1)
Client testmqtt-qos-pub received PUBCOMP (Mid: 1)
Client testmqtt-qos-pub sending DISCONNECT

d0 ist das DUP-Flag, q2 das QoS, m1 der Packet Identifier. Wiederhole das mit -q 1, dann siehst du statt des dreistufigen Handshakes ein einzelnes PUBACK. Prüfe auf Subscriber-Seite den q-Wert im empfangenen PUBLISH – hast du mit QoS 1 abonniert, steht dort q1, auch wenn der Publisher QoS 2 verwendet hat.

Für eine genauere Analyse zeichnest du den Traffic auf und filterst in Wireshark mit mqtt.msgtype >= 3 && mqtt.msgtype <= 7 (PUBLISH bis PUBCOMP) oder mit mqtt.dupflag == 1, um nur Neuübertragungen zu sehen. In MQTT 5 tragen Bestätigungen zusätzlich einen Reason Code: Ein PUBACK mit 0x10 (No matching subscribers) ist ein Erfolg, kein Fehler – nachschlagen kannst du die Codes in der Reason-Code-Referenz.

Schritt 2: ein Zähler für Lücken und Duplikate

Das Harness besteht aus drei Teilen: einem Subscriber mit Persistent Session, der jede empfangene Sequenznummer protokolliert, einem Publisher, der einen bekannten Bereich sendet und Bestätigungen zählt, und einem kurzen Analyse-Skript.

import sys
import paho.mqtt.client as mqtt

QOS = int(sys.argv[1]) if len(sys.argv) > 1 else 1
TOPIC = "testmqtt/qos-test/seq"
log = open(f"received-q{QOS}.log", "a", buffering=1)  # line-buffered: survives kill -9

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"connected: {reason_code}, session present: {flags.session_present}")
    client.subscribe(TOPIC, qos=QOS)

def on_message(client, userdata, msg):
    log.write(f"{msg.payload.decode()} {int(msg.dup)}\n")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"testmqtt-qos-sub-q{QOS}", clean_session=False)
# client.username_pw_set("username", "password")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=10)
client.loop_forever()   # reconnects automatically after a network error

import sys, time
import paho.mqtt.client as mqtt

QOS = int(sys.argv[1]) if len(sys.argv) > 1 else 1
N = int(sys.argv[2]) if len(sys.argv) > 2 else 2000
TOPIC = "testmqtt/qos-test/seq"
acked = 0

def on_publish(client, userdata, mid, reason_code, properties):
    global acked
    acked += 1          # QoS 1: on PUBACK, QoS 2: on PUBCOMP

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"testmqtt-qos-pub-q{QOS}", clean_session=False)
client.on_publish = on_publish
client.connect("broker.connectmq.com", 1883, keepalive=10)
client.loop_start()

for seq in range(N):
    client.publish(TOPIC, str(seq), qos=QOS)   # queued by paho while disconnected
    time.sleep(0.005)

deadline = time.time() + 60
while acked < N and time.time() < deadline:
    time.sleep(0.2)
print(f"published {N}, acknowledged {acked}")
client.disconnect()
client.loop_stop()

#!/usr/bin/env bash
LOG=${1:-received-q1.log}; N=${2:-2000}
echo "received lines : $(wc -l < "$LOG")"
echo "unique         : $(cut -d' ' -f1 "$LOG" | sort -u | wc -l)"
echo "duplicates     : $(cut -d' ' -f1 "$LOG" | sort | uniq -d | wc -l)"
echo "with DUP flag  : $(awk '$2 == 1' "$LOG" | wc -l)"
echo "missing        : $(comm -23 <(seq 0 $((N - 1)) | sort) <(cut -d' ' -f1 "$LOG" | sort -u) | wc -l)"

Starte python sub.py 1, dann python pub.py 1 und anschließend ./analyze.sh received-q1.log. Beginne mit einem Lauf ohne Fehler: Er muss 2.000 eindeutige und 0 fehlende Nachrichten zeigen. Scheitert schon die Baseline, behebe das zuerst (falsches QoS bei der Subscription, eine ACL, ein Größenlimit für Nachrichten), bevor du Fehler injizierst.

Schritt 3: mitten im Versand Dinge kaputt machen

Führe jedes Szenario aus, während pub.py sendet, und analysiere danach. Verwende pro Lauf eine neue Logdatei.

BROKER_IP=$(dig +short broker.connectmq.com | head -1)
sudo iptables -I OUTPUT -p tcp -d "$BROKER_IP" --dport 1883 -j DROP
sleep 20
sudo iptables -D OUTPUT -p tcp -d "$BROKER_IP" --dport 1883 -j DROP

Das Blackhole ist der realistischere Fehler: Es kommt kein RST an, der Client bemerkt den Abbruch also nur über Keep-Alive (brokerseitig etwa 1,5 × die 10 Sekunden Keep-Alive) und verbindet sich dann neu. Beim Reconnect sollte der Subscriber session present: True ausgeben, und der Broker sollte unbestätigte Nachrichten mit DUP=1 erneut senden, gefolgt vom Rückstau aus der Queue. Weitere Techniken findest du unter QoS mit Netzwerkunterbrechung testen.

Der Prozessabsturz ist der interessante Fall für QoS 2. paho-mqtt hält seinen Session-State im Speicher, ein kill -9 löscht also die Client-Hälfte des QoS-2-Handshakes. Je nachdem, ob eine Library die Nachricht bei PUBLISH oder bei PUBREL an deinen Code übergibt, kann ein Absturz im falschen Moment ein Duplikat oder eine fehlende Nachricht erzeugen. Exactly-once ist eine Eigenschaft des Protokolls zwischen zwei Endpunkten, die ihren State behalten – darf deine Anwendung nach einem Absturz nichts doppelt verarbeiten, dedupliziere anhand einer Nachrichten-ID in deinem eigenen Speicher.

Ergebnisse richtig lesen

• Fehlende Nachrichten bei QoS 1/2 nach einer Netzwerkunterbrechung – die Session wurde nicht beibehalten: Prüfe session present im Log, die Client-ID (muss beim Reconnect identisch sein) und die Session Expiry bei MQTT 5.
• Fehlende Nachrichten nur nach langen Ausfällen – die Queue des Brokers pro Client lief voll und er hat Nachrichten per Policy verworfen (zum Beispiel max_queued_messages in Mosquitto). Das ist ein Kapazitätslimit, kein QoS-Bug; siehe Message Queue voll.
• Duplikate bei QoS 2 – prüfe das effektive QoS beim Subscriber (q2 in mosquitto_sub -d) und suche dann nach Abstürzen oder einem zweiten Client mit derselben Client-ID. Weitere Ursachen unter doppelte Nachrichten.
• Duplikate ohne DUP-Flag bei QoS 1 – meist hat der Publisher nach dem Reconnect erneut gesendet, was aus Sicht des Brokers eine neue Nachricht ist. Bei At-least-once normal.
• acknowledged kleiner als published – der Publisher hat aufgegeben, bevor der Rückstau abgearbeitet war; erhöhe die Deadline oder prüfe ein Inflight-Limit auf Broker-Seite.

Ein schneller manueller Check

Für einen schnellen Plausibilitätscheck ohne Skripte abonnierst du im Online-MQTT-Client mit QoS 1, veröffentlichst aus einem Terminal mit mosquitto_pub -q 1 -d und vergleichst das auf beiden Seiten angezeigte QoS. Trenne dann den Browser-Client, veröffentliche ein paar weitere Nachrichten und verbinde dich mit derselben Client-ID und deaktivierter Clean Session neu: Die gepufferten Nachrichten sollten ankommen.

QoS-Zustellungen im Browser beobachten

QoS 1 ist durch null Lücken bewiesen, QoS 2 durch null Lücken und null Duplikate – gemessen mit Sequenznummern über echte Verbindungsabbrüche hinweg, nicht aus der Spezifikation angenommen. Leg das Harness in dein Repository und führe es erneut aus, wann immer du Broker, Client-Library oder Session-Einstellungen änderst.

Häufig gestellte Fragen