Simular redes malas para MQTT con Toxiproxy

Toxiproxy es un pequeño proxy TCP de Shopify que se coloca entre tu cliente MQTT y el broker y te permite romper la conexión a propósito: añadir latencia y jitter, limitar el ancho de banda, congelar el tráfico o resetear el socket. Todo se controla con una API HTTP, así que tus tests pueden activar y desactivar fallos desde el código.

Esta guía configura Toxiproxy delante de un broker MQTT, repasa los toxics que importan para MQTT y los convierte en tests automatizados de keep-alive, reconexión y comportamiento de QoS.

Por qué un proxy y no desenchufar el cable

Desconectar el Wi-Fi o matar el broker son herramientas toscas. Son difíciles de sincronizar, difíciles de repetir en CI y solo simulan un fallo: "todo ha desaparecido". Las redes reales fallan de formas más sutiles: un enlace móvil que añade dos segundos de latencia, un NAT que descarta en silencio un flujo inactivo, un uplink congestionado que entrega 5 KB por segundo. Esas son las condiciones que destapan bugs en bucles de reconexión, ventanas inflight y timeouts.

Toxiproxy te da cada uno de esos fallos como un "toxic" con nombre y reversible sobre una sola conexión TCP. El broker y tus otros clientes no se ven afectados, y el test puede comprobar exactamente qué hizo tu código mientras el toxic estaba activo.

Ejecutar Toxiproxy delante del broker

Lo más rápido es el contenedor oficial. El puerto 8474 es la API de control; cada proxy que crees tiene su propio puerto de escucha.

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

Crea un proxy que escuche en 21883 y reenvíe al broker:

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

Apunta cualquier cliente MQTT a localhost:21883 en lugar del broker y comprueba que el tráfico sigue fluyendo:

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

Puedes observar el mismo topic en el cliente MQTT online conectado directamente al broker: los mensajes publicados a través del proxy también aparecen ahí, lo que facilita ver cuándo un toxic los retrasa o los bloquea.

Para un stack de test, declara el proxy en un fichero de configuración para que exista en cuanto arranque el contenedor:

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

Si el broker corre en el mismo fichero Compose (por ejemplo un contenedor Mosquitto con Testcontainers o un servicio llamado mosquitto), pon como upstream mosquitto:1883.

Los toxics que importan para MQTT

Los toxics se añaden por proxy y por dirección. downstream es broker → cliente (CONNACK, PUBACK, mensajes entregados) y upstream es cliente → broker (CONNECT, PUBLISH, PINGREQ). El valor por defecto es 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

Algunas notas específicas de MQTT:

• latency retrasa cada fragmento de datos. Aplicado en downstream alarga cada ida y vuelta: CONNACK, SUBACK y PUBACK llegan tarde. Úsalo para encontrar timeouts demasiado ajustados; por ejemplo, un connect timeout de 1 s falla en cuanto la latencia lo supera.
• bandwidth (rate en KB/s) muestra qué pasa cuando publicas más rápido de lo que el enlace puede transportar. Con QoS 1, los mensajes pendientes se acumulan en la ventana inflight del cliente; consulta el control de flujo de MQTT.
• timeout con timeout: 0 detiene todos los datos pero mantiene el socket abierto. Ningún lado recibe un error TCP: es la clásica conexión half-open, y solo el keep alive de MQTT la detectará.
• reset_peer cierra la conexión con un TCP RST tras timeout ms, lo que produce ECONNRESET en el cliente.
• slicer divide los datos en segmentos pequeños con un retardo entre ellos. Un cliente correcto reensambla los paquetes MQTT entre lecturas; un parser embebido ingenuo puede no hacerlo.
• limit_data cierra la conexión tras bytes bytes, útil para cortar un PUBLISH grande por la mitad.

Para cortar la conexión por completo, desactiva el proxy: curl -X POST localhost:8474/proxies/mqtt -d '{"enabled": false}'. Las conexiones existentes se cierran y las nuevas se rechazan hasta que lo vuelvas a activar.

Automatizarlo: un test de keep-alive en Python

La API de control es HTTP plano, así que no necesitas ninguna librería cliente. Este test (paho-mqtt 2.x) se conecta a través del proxy con un keep-alive de 5 segundos, congela el enlace y comprueba que el cliente lo detecta dentro de la ventana 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 la conexión en ambas direcciones sin cerrarla
        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()

Ejecútalo con pytest -s test_keepalive.py. La detección debería tardar aproximadamente uno o dos intervalos de keep-alive, según en qué punto del ciclo de ping caiga la congelación. Si tarda mucho más, o nunca ocurre, tu cliente no está aplicando el keep-alive: un bug que dejaría a dispositivos reales "conectados" a nada durante horas. Para una estructura de fixtures más limpia, consulta fixtures de pytest para MQTT.

El mismo patrón cubre otros escenarios:

1. Reconexión: desactiva el proxy, espera, actívalo y comprueba que el cliente se reconecta y se vuelve a suscribir (detalles en tests de reconexión MQTT).
2. Retransmisión QoS 1: añade limit_data en upstream, publica un mensaje QoS 1 grande, elimina el toxic y comprueba que el suscriptor lo recibe exactamente una vez o al menos una vez, según espere tu diseño.
3. Presupuesto de timeouts: añade 3 s de latencia en downstream y comprueba que tu timeout de conexión o de request-response falla limpiamente con un error claro en lugar de quedarse colgado.

MQTT sobre TLS y WebSockets a través de Toxiproxy

Toxiproxy reenvía bytes crudos, así que TLS (8883) y WebSockets (8083, 8084 con path /mqtt) funcionan sin cambios: crea un proxy cuyo upstream sea broker.connectmq.com:8883 y el handshake pasa directamente.

Lo único que debes resolver es la verificación del hostname. El cliente se conecta a localhost, pero el certificado nombra al broker. Opciones:

• Mantén el hostname del broker en el cliente y mapéalo al proxy en /etc/hosts o con extra_hosts de Docker en el runner de tests.
• Fija el server name de TLS explícitamente: en MQTT.js, la opción servername; en Python, envuelve el socket con server_hostname igual al nombre del broker.
• Usa TCP plano para los tests de fallos de red y deja TLS para un test de conexión TLS aparte.

Nunca desactives la verificación de certificados para que el proxy funcione en un test que también se ejecuta con la configuración de producción: ese hábito suele colarse en el código real.

Errores comunes

• Añadir latencia solo en upstream. Muchos timeouts interesantes esperan la respuesta del broker. Prueba ambas direcciones, o downstream primero.
• Confundir timeout: 0 con desactivar el proxy. Desactivarlo cierra los sockets (el cliente ve una desconexión inmediata); timeout: 0 los congela (solo te salva el keep-alive).
• Probar solo con QoS 0. QoS 0 oculta lo que pasa con los mensajes sin confirmar. Repite los escenarios críticos con QoS 1.
• Medir a través de un broker público. Los sandboxes compartidos añaden su propia variabilidad. Para todo lo que vayas a comprobar en un assert, usa un broker que controles.

Observa los mensajes del proxy en tu navegador

Toxiproxy convierte "a veces se desconecta en campo" en un test que puedes ejecutar en cada commit. Empieza con un proxy, un test de keep-alive con el enlace congelado y un test de reconexión; cuando pasen, añade perfiles de latencia y ancho de banda que se parezcan a las redes que usan realmente tus dispositivos, como los perfiles celulares de probar MQTT en redes celulares inestables.

Preguntas frecuentes