Simulando redes ruins para MQTT com Toxiproxy

O Toxiproxy é um pequeno proxy TCP da Shopify que fica entre seu cliente MQTT e o broker e permite quebrar a conexão de propósito: adicionar latência e jitter, limitar a banda, congelar o tráfego ou resetar o socket. Tudo é controlado por uma API HTTP, então seus testes podem ligar e desligar falhas no código.

Este guia coloca o Toxiproxy na frente de um broker MQTT, explica os toxics que importam para MQTT e os transforma em testes automatizados de keep-alive, reconexão e comportamento de QoS.

Por que um proxy em vez de puxar o cabo

Desligar o Wi-Fi ou matar o broker são ferramentas grosseiras. É difícil acertar o momento, difícil repetir no CI, e elas simulam uma única falha: "sumiu tudo". Redes reais falham de jeitos mais sutis — um link móvel que adiciona dois segundos de latência, um NAT que descarta silenciosamente um fluxo ocioso, um uplink congestionado que entrega 5 KB por segundo. São essas condições que revelam bugs em loops de reconexão, janelas inflight e timeouts.

O Toxiproxy oferece cada uma dessas falhas como um "toxic" nomeado e reversível sobre uma única conexão TCP. O broker e seus outros clientes não são afetados, e o teste pode verificar exatamente o que seu código fez enquanto o toxic estava ativo.

Rodando o Toxiproxy na frente do broker

O jeito mais rápido é o container oficial. A porta 8474 é a API de controle; cada proxy que você cria tem sua própria porta de escuta.

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

Crie um proxy que escuta na 21883 e encaminha para o broker:

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

Aponte qualquer cliente MQTT para localhost:21883 em vez do broker e confirme que o tráfego continua passando:

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

Você pode acompanhar o mesmo topic no cliente MQTT online conectado diretamente ao broker: as mensagens publicadas pelo proxy também aparecem lá, o que deixa fácil ver quando um toxic as atrasa ou bloqueia.

Para um stack de testes, declare o proxy em um arquivo de configuração para que ele exista assim que o container subir:

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

Se o broker roda no mesmo arquivo Compose (por exemplo um container Mosquitto com Testcontainers ou um serviço chamado mosquitto), use mosquitto:1883 como upstream.

Os toxics que importam para MQTT

Toxics são adicionados por proxy e por direção. downstream é broker → cliente (CONNACK, PUBACK, mensagens entregues) e upstream é cliente → broker (CONNECT, PUBLISH, PINGREQ). O padrão é 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

Algumas observações específicas de MQTT:

• latency atrasa cada bloco de dados. Aplicado em downstream, alonga cada ida e volta: CONNACK, SUBACK e PUBACK chegam atrasados. Use para encontrar timeouts apertados demais — por exemplo, um connect timeout de 1 s falha assim que a latência passa disso.
• bandwidth (rate em KB/s) mostra o que acontece quando você publica mais rápido do que o link consegue transportar. Com QoS 1, as mensagens pendentes se acumulam na janela inflight do cliente; veja controle de fluxo no MQTT.
• timeout com timeout: 0 interrompe todos os dados mas mantém o socket aberto. Nenhum lado recebe erro TCP — é a clássica conexão half-open, e só o keep alive do MQTT vai percebê-la.
• reset_peer fecha a conexão com um TCP RST depois de timeout ms, gerando ECONNRESET no cliente.
• slicer divide os dados em segmentos pequenos com um atraso entre eles. Um cliente correto remonta os pacotes MQTT entre leituras; um parser embarcado ingênuo talvez não.
• limit_data fecha a conexão depois de bytes bytes — útil para cortar um PUBLISH grande ao meio.

Para derrubar a conexão por completo, desative o proxy: curl -X POST localhost:8474/proxies/mqtt -d '{"enabled": false}'. As conexões existentes são fechadas e as novas são recusadas até você reativá-lo.

Automatizando: um teste de keep-alive em Python

A API de controle é HTTP puro, então nenhuma biblioteca cliente é necessária. Este teste (paho-mqtt 2.x) conecta pelo proxy com keep-alive de 5 segundos, congela o link e verifica que o cliente percebe isso dentro da janela 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"

        # Congela a conexão nas duas direções sem fechá-la
        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()

Rode com pytest -s test_keepalive.py. A detecção deve levar cerca de um a dois intervalos de keep-alive, dependendo de em que ponto do ciclo de ping o congelamento acontece. Se demorar muito mais, ou nunca acontecer, seu cliente não está aplicando o keep-alive — um bug que deixaria dispositivos reais "conectados" a nada por horas. Para uma organização de fixtures mais limpa, veja fixtures do pytest para MQTT.

O mesmo padrão cobre outros cenários:

1. Reconexão — desative o proxy, espere, reative e verifique que o cliente reconecta e refaz as assinaturas (detalhes em testes de reconexão MQTT).
2. Retransmissão QoS 1 — adicione limit_data em upstream, publique uma mensagem QoS 1 grande, remova o toxic e confira se o subscriber a recebe exatamente uma vez ou pelo menos uma vez, conforme seu design espera.
3. Orçamento de timeouts — adicione 3 s de latência em downstream e verifique que o timeout de conexão ou de request-response falha de forma limpa, com um erro claro, em vez de travar.

MQTT sobre TLS e WebSockets pelo Toxiproxy

O Toxiproxy encaminha bytes crus, então TLS (8883) e WebSockets (8083, 8084 com path /mqtt) funcionam sem mudanças: crie um proxy com upstream broker.connectmq.com:8883 e o handshake passa direto.

A única coisa a resolver é a verificação de hostname. O cliente conecta em localhost, mas o certificado nomeia o broker. Opções:

• Mantenha o hostname do broker no cliente e mapeie-o para o proxy no /etc/hosts ou com extra_hosts do Docker no runner de testes.
• Defina o server name do TLS explicitamente — no MQTT.js, a opção servername; em Python, envolva o socket com server_hostname igual ao nome do broker.
• Use TCP puro nos testes de falha de rede e deixe o TLS para um teste de conexão TLS separado.

Nunca desative a verificação de certificado para fazer o proxy funcionar em um teste que também roda com as configurações de produção — esse hábito costuma vazar para o código real.

Erros comuns

• Adicionar latência só em upstream. Muitos timeouts interessantes esperam a resposta do broker. Teste as duas direções, ou downstream primeiro.
• Confundir timeout: 0 com desativar o proxy. Desativar fecha os sockets (o cliente vê uma desconexão imediata); timeout: 0 os congela (só o keep-alive salva você).
• Testar só com QoS 0. QoS 0 esconde o que acontece com mensagens não confirmadas. Repita os cenários críticos com QoS 1.
• Medir por um broker público. Sandboxes compartilhados adicionam sua própria variação. Para qualquer coisa que você vá verificar em um assert, use um broker que você controla.

Acompanhe as mensagens do proxy no navegador

O Toxiproxy transforma "às vezes desconecta em campo" em um teste que roda a cada commit. Comece com um proxy, um teste de keep-alive com o link congelado e um teste de reconexão; quando passarem, adicione perfis de latência e banda parecidos com as redes que seus dispositivos usam de verdade, como os perfis celulares de testar MQTT em redes celulares instáveis.

Perguntas frequentes