Session Expiry und Offline-Queues in MQTT testen

Eine persistente MQTT-Session erlaubt es einem Gerät, offline zu gehen und trotzdem jede verpasste QoS-1-Nachricht zu erhalten. Ob das klappt, hängt von vier leicht falsch gesetzten Einstellungen ab: Client-ID, Clean Start, Session Expiry Interval und die Queue-Limits des Brokers. Der verlässliche Test: Mit persistenter Session abonnieren, trennen, veröffentlichen, neu verbinden und dann Session Present sowie die genau empfangenen Nachrichten prüfen. Anschließend wiederholen, nachdem die Session abgelaufen sein sollte.

Dieser Guide liefert einen manuellen Test mit mosquitto_sub, ein Python-Skript, das alle drei Verhaltensweisen prüft, und eine Liste der Fehlermuster, denen du begegnen wirst.

Was stimmen muss, damit eine Session überlebt

Der MQTT-5-Standardwert für Session Expiry Interval ist 0. Ein Client, der Clean Start auf false setzt, das Intervall aber vergisst, verliert seine Session trotzdem im Moment des Disconnects. Den Unterschied zwischen beiden Protokollversionen erklärt Clean Session vs Clean Start, das Grundkonzept persistente MQTT-Sessions.

Der Broker kann deinen Wert auch überschreiben: Erlaubt er weniger als angefragt, schickt ein MQTT-5-Broker sein eigenes Session Expiry Interval im CONNACK zurück. Logge es immer.

Was der Broker in einer Session speichert

• Die Subscriptions des Clients (mit QoS und Optionen).
• QoS-1- und QoS-2-Nachrichten, die eintrafen, während der Client offline war.
• In-Flight-QoS-1- und QoS-2-Nachrichten, die nicht vollständig bestätigt wurden.
• Optional QoS-0-Nachrichten, wenn der Broker so konfiguriert ist.

Alles außerhalb dieser Liste, etwa Retained Messages, existiert unabhängig von der Session. Eine Nachricht mit eigenem Message Expiry kann vor der Session aus der Queue verschwinden.

Manueller Test mit mosquitto_sub

Wähle eine eindeutige Client-ID und ein Topic und lege die Session an:

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

Mit -d siehst du jedes ausgetauschte Paket. Nicht jede Mosquitto-Version gibt das Session-Present-Flag aus, der Beweis ist hier also der Rückstau: msg 1 bis msg 5, in Reihenfolge, direkt nach dem CONNACK. Der Python-Test unten prüft das Flag explizit. Beende jetzt den Client, veröffentliche erneut, warte länger als 120 Sekunden und führe Schritt 3 noch einmal aus: Die Session muss weg sein, und keine der neuen Nachrichten darf ankommen.

Zwei Dinge sorgen hier oft für Verwirrung. Erstens spielt das Topic in Schritt 3 für den Rückstau keine Rolle: Die gespeicherte Subscription liefert ihn. Zweitens wirft Schritt 3 einen der beiden Clients raus, wenn ein anderer Client dieselbe ID nutzt. Das ist eine Session-Übernahme (Session Takeover), kein Session-Test.

Ein automatisierter Session-Test in Python

Dieses paho-mqtt-2.x-Skript führt drei Tests in einem Durchlauf aus: Die Session überlebt und reiht Nachrichten in Reihenfolge ein, die Session läuft tatsächlich ab, und was passiert, wenn die Queue überläuft.

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

Starte es mit python session_test.py oder übergib einen größeren Burst, etwa python session_test.py 5000. Auf einem Broker, der Sessions hält, sieht ein gesunder Lauf so aus:

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

Die letzte Zeile ist Information, kein Pass/Fail. Sie zeigt das effektive Queue-Limit und die Verwerfungsstrategie: first=0 last=999 heißt, der Broker hat die ältesten Nachrichten behalten und die neuesten verworfen; first=1000 last=1999 hieße das Gegenteil. Zeigt expiry eine Zahl, hat der Broker dein angefragtes Intervall überschrieben.

Das Skript beendet jede Session mit Session Expiry Interval = 0 im DISCONNECT, damit Testläufe keine verwaisten Sessions hinterlassen. Auf 0 senken geht so nur, wenn der Wert im CONNECT ungleich null war. Ihn im DISCONNECT von 0 aus zu erhöhen, ist ein Protokollfehler.

Queue-Limits auf deinem eigenen Broker testen

Jeder Broker begrenzt, wie viel er pro Offline-Client speichert, denn tausend tote Geräte mit unbegrenzten Queues würden seinen Speicher erschöpfen. In Mosquitto sind das die relevanten Einstellungen:

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

Teste das Limit, das du konfiguriert hast, nicht das, das du vermutest: Veröffentliche einen Burst von max_queued_messages + 500, verbinde neu und vergleiche. Dann entscheide, was deine Geräte brauchen. Ein Sensor, den nur der letzte Befehl interessiert, sollte Message Expiry oder Retained State nutzen, keine riesige Queue. Wenn Clients Fehler wegen voller Queues sehen, hilft MQTT Message Queue voll. Der Mosquitto-Config-Generator kann diese Einstellungen für dich erzeugen.

Teste auch einen Broker-Neustart, während der Client offline ist. Ohne aktivierte Persistenz hält Mosquitto Sessions nur im Speicher, und ein Neustart verliert sie alle.

Fehlermuster und ihre Bedeutung

Eine ausführlichere Fehlersuche findest du unter warum eine MQTT-Session nicht erhalten bleibt.

Im Browser testen

Für eine schnelle visuelle Prüfung öffne den Online-MQTT-Client, verbinde mit fester Client-ID, ausgeschaltetem Clean Start und einer Session Expiry, abonniere mit QoS 1 und trenne die Verbindung. Veröffentliche ein paar Nachrichten aus einem Terminal, verbinde im Browser neu und sieh zu, wie der Rückstau ankommt.

Persistente Session im Browser ausprobieren

Ein Session-Test dauert eine Minute und spart Stunden beim Debuggen von "mein Gerät hat den Befehl verpasst". Leg das Skript in dein Repository, führe es gegen jeden Broker aus, den du in Betrieb nimmst, und behandle die Queue-Limit-Zeile als Teil deiner Kapazitätsplanung.

Häufig gestellte Fragen