Como testar a expiração de sessão e as filas offline no MQTT

Uma sessão MQTT persistente permite que um dispositivo fique offline e ainda receba todas as mensagens QoS 1 que perdeu. Se isso funciona depende de quatro configurações fáceis de errar: o client ID, o Clean Start, o Session Expiry Interval e os limites de fila do broker. O jeito confiável de testar: assine com uma sessão persistente, desconecte, publique, reconecte e valide o Session Present e as mensagens exatas recebidas. Depois repita quando a sessão já deveria ter expirado.

Este guia traz um teste manual com mosquitto_sub, um script Python que verifica os três comportamentos e uma lista dos padrões de falha que você vai encontrar.

O que precisa ser verdade para a sessão sobreviver

O padrão do Session Expiry Interval no MQTT 5 é 0. Um cliente que define Clean Start como false mas esquece o intervalo perde a sessão no momento em que desconecta. A diferença entre as duas versões do protocolo está em Clean Session vs Clean Start, e o conceito geral em sessões persistentes no MQTT.

O broker também pode sobrescrever seu valor: se ele permitir menos do que você pediu, um broker MQTT 5 devolve o próprio Session Expiry Interval no CONNACK. Sempre registre esse valor.

O que o broker guarda numa sessão

• As assinaturas do cliente (com QoS e opções).
• Mensagens QoS 1 e QoS 2 que chegaram enquanto o cliente estava offline.
• Mensagens QoS 1 e QoS 2 em trânsito que não foram totalmente confirmadas.
• Opcionalmente, mensagens QoS 0, se o broker estiver configurado para enfileirá-las.

O que estiver fora dessa lista, como mensagens retidas, vive independente da sessão. Uma mensagem com a própria expiração de mensagem pode sumir da fila antes da sessão.

Teste manual com mosquitto_sub

Escolha um client ID e um topic únicos e crie a sessão:

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

Com -d você vê todos os pacotes trocados. Nem toda versão do Mosquitto imprime a flag Session Present, então a evidência aqui é o acumulado: msg 1 a msg 5, em ordem, logo depois do CONNACK. O teste em Python abaixo verifica a flag explicitamente. Agora pare o cliente, publique de novo, espere mais de 120 segundos e rode o passo 3 outra vez: a sessão precisa ter sumido e nenhuma das mensagens novas pode chegar.

Duas coisas confundem aqui. Primeiro, o topic do passo 3 não importa para o acumulado: quem entrega é a assinatura armazenada. Segundo, rodar o passo 3 enquanto outro cliente usa o mesmo ID derruba um deles. Isso é uma tomada de sessão (session takeover), não um teste de sessão.

Um teste de sessão automatizado em Python

Este script com paho-mqtt 2.x roda três testes de uma vez: a sessão sobrevive e enfileira mensagens em ordem, a sessão realmente expira, e o que acontece quando a fila transborda.

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

Rode com python session_test.py ou passe uma rajada maior, por exemplo python session_test.py 5000. Num broker que mantém sessões, uma execução saudável fica assim:

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

A última linha é informativa, não pass/fail. Ela mostra o limite efetivo da fila e a política de descarte: first=0 last=999 significa que o broker manteve as mensagens mais antigas e descartou as mais novas; first=1000 last=1999 significaria o contrário. Se expiry mostrar um número, o broker sobrescreveu o intervalo que você pediu.

O script encerra cada sessão com Session Expiry Interval = 0 no DISCONNECT, para que os testes não deixem sessões órfãs. Só dá para baixar o intervalo para 0 dessa forma se ele era diferente de zero no CONNECT. Aumentá-lo a partir de 0 no DISCONNECT é erro de protocolo.

Testar limites de fila no seu próprio broker

Todo broker limita quanto guarda por cliente offline, porque mil dispositivos mortos com filas ilimitadas esgotariam a memória. No Mosquitto, as configurações relevantes são:

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 o limite que você configurou, não o que você supõe: publique uma rajada de max_queued_messages + 500, reconecte e compare. Depois decida do que seus dispositivos precisam. Um sensor que só se importa com o último comando deveria usar expiração de mensagem ou estado retido, não uma fila enorme. Quando os clientes recebem erros de fila cheia, fila de mensagens MQTT cheia mostra as correções. O gerador de configuração do Mosquitto pode produzir essas configurações para você.

Teste também um reinício do broker enquanto o cliente está offline. Sem persistência ativada, o Mosquitto guarda as sessões só em memória, e um reinício perde todas.

Padrões de falha e o que significam

Para um passo a passo de depuração mais amplo, veja por que a sessão MQTT não persiste.

Testar no navegador

Para uma checagem visual rápida, abra o cliente MQTT online, conecte com um client ID fixo, Clean Start desligado e uma expiração de sessão, assine em QoS 1 e desconecte. Publique algumas mensagens pelo terminal, reconecte no navegador e veja o acumulado chegar.

Experimente uma sessão persistente no navegador

Um teste de sessão leva um minuto para rodar e economiza horas investigando relatos de "meu dispositivo perdeu o comando". Mantenha o script no seu repositório, rode-o em todo broker que você colocar no ar e trate a linha do limite de fila como parte do seu planejamento de capacidade.

Perguntas frequentes