MQTT Soak Testing: Find Memory Leaks and Slow Degradation

A soak test runs an MQTT system at normal load for a long time — usually 24 to 72 hours — to catch problems that a ten-minute benchmark never sees: memory that grows a few megabytes per hour, file descriptors that are never released, queues for sessions that never come back, and latency that slowly creeps up until devices start timing out.

This guide shows what to measure, gives you a Python harness that records everything to CSV, and explains how to read the result so that "it looks fine" becomes a number.

Soak test vs load test

If you have not measured capacity yet, start with MQTT broker load testing. A soak test assumes the system handles the load now and asks whether it still will on day three.

What degrades slowly

Most long-run failures fall into a handful of patterns:

• Client memory leaks — callbacks that append to a list, a growing in-flight queue because the client publishes faster than acknowledgements arrive, or a new client object created on every reconnect without the old one being stopped. See MQTT client memory leaks.
• Unbounded broker queues — persistent sessions (clean_session=False / session expiry) for devices that never reconnect keep collecting QoS 1 and 2 messages until the broker runs out of memory. The classic result is described in MQTT broker crashes and OOM.
• Retained message growth — topics with unique IDs in the path (.../session-8f3a/state) published with retain=true accumulate forever.
• File descriptor and socket leaks — reconnect paths that do not close the old socket. The process eventually fails with Too many open files.
• Time-based events — log rotation, certificate renewal, token expiry, daily backups or a disk filling up with persistence files. These only happen if the test runs long enough to cross them.

The metrics to record

Record client-side and broker-side numbers on the same timeline, once per minute:

A Python soak harness

This script (paho-mqtt 2.x and psutil) publishes a sequence-numbered message at a fixed rate, subscribes to its own topic, and writes one CSV row per minute with throughput, gaps, duplicates, latency percentiles, its own memory and open file descriptors. It resubscribes in on_connect, so it survives reconnects instead of silently going deaf.

import csv, json, os, threading, time, uuid
import paho.mqtt.client as mqtt
import psutil

HOST, PORT = os.environ.get("MQTT_HOST", "broker.connectmq.com"), 1883
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/soak/{RUN}/data"
RATE = 20            # messages per second
DURATION_H = 72      # total run time
WINDOW_S = 60        # one CSV row per minute

lock = threading.Lock()
st = {"recv": 0, "gaps": 0, "dups": 0, "last": -1, "lat": [], "disc": 0}

def on_connect(client, userdata, flags, reason_code, properties):
    if client is sub:
        client.subscribe(TOPIC, qos=1)

def on_disconnect(client, userdata, flags, reason_code, properties):
    with lock:
        st["disc"] += 1

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    with lock:
        st["recv"] += 1
        st["lat"].append((time.monotonic() - data["t"]) * 1000)
        if data["seq"] <= st["last"]:
            st["dups"] += 1
        else:
            st["gaps"] += data["seq"] - st["last"] - 1
            st["last"] = data["seq"]

def make_client(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"soak-{RUN}-{name}")
    c.username_pw_set("username", "password")
    c.on_connect, c.on_disconnect = on_connect, on_disconnect
    c.reconnect_delay_set(min_delay=1, max_delay=30)
    return c

sub = make_client("sub")
sub.on_message = on_message
sub.connect(HOST, PORT, keepalive=30)
sub.loop_start()
pub = make_client("pub")
pub.connect(HOST, PORT, keepalive=30)
pub.loop_start()
time.sleep(2)

proc = psutil.Process()
pct = lambda xs, q: xs[max(0, int(len(xs) * q) - 1)] if xs else 0.0

with open(f"soak-{RUN}.csv", "w", newline="") as f:
    w = csv.writer(f)
    w.writerow(["time", "sent", "recv", "gaps", "dups", "disconnects",
                "p50_ms", "p99_ms", "rss_mb", "fds"])
    end = time.monotonic() + DURATION_H * 3600
    next_row = time.monotonic() + WINDOW_S
    seq = 0
    while time.monotonic() < end:
        pub.publish(TOPIC, json.dumps({"seq": seq, "t": time.monotonic()}), qos=1)
        seq += 1
        time.sleep(1 / RATE)
        if time.monotonic() >= next_row:
            with lock:
                lat, st["lat"] = sorted(st["lat"]), []
                counts = [st["recv"], st["gaps"], st["dups"], st["disc"]]
            w.writerow([time.strftime("%Y-%m-%dT%H:%M:%S"), seq, *counts,
                        round(pct(lat, 0.50), 1), round(pct(lat, 0.99), 1),
                        round(proc.memory_info().rss / 1e6, 1), proc.num_fds()])
            f.flush()
            next_row += WINDOW_S

Run it with nohup python soak.py & (or in tmux) on a machine that will stay up for three days. The CSV is flushed every minute, so you can inspect it while the test is still running. num_fds() is available on Linux and macOS; on Windows use proc.num_handles() instead. A short trial against the shared broker.connectmq.com endpoint is fine, but for the real multi-day run set MQTT_HOST to your own broker instance: the shared endpoint is meant for quick tests, is rate-limited and keeps no persistent sessions, so the session-related failure modes below only show up on a broker of your own.

Watch the broker side too

Client numbers tell you that something degraded; broker numbers tell you where. If your broker publishes $SYS topics (Mosquitto does), subscribe to the counters that reveal growth:

mosquitto_sub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "$MQTT_PASS" -v \
  -t '$SYS/broker/clients/connected' \
  -t '$SYS/broker/messages/stored' \
  -t '$SYS/broker/retained messages/count' \
  -t '$SYS/broker/heap/current'

If the broker runs in Docker on a host you control, sample its memory and CPU once a minute:

while true; do
  docker stats --no-stream --format "$(date -u +%FT%TZ),{{.Name}},{{.MemUsage}},{{.CPUPerc}}" mosquitto
  sleep 60
done >> broker-stats.csv

For anything you run longer than a test, export the same values to a time-series database — Prometheus and Grafana monitoring and $SYS topic monitoring cover the setup.

Turning the CSV into a verdict

Eyeballing a 4,000-row CSV does not work. Fit a trend line instead. Skip the first hour (caches, JIT, connection pools warming up) and compute the slope of memory over time with the standard library:

import csv, statistics, sys

rows = list(csv.DictReader(open(sys.argv[1])))[60:]   # skip 1 h warm-up
hours = [i / 60 for i in range(len(rows))]
for col in ("rss_mb", "fds", "p99_ms"):
    ys = [float(r[col]) for r in rows]
    slope, _ = statistics.linear_regression(hours, ys)
    print(f"{col:8} start {ys[0]:8.1f}  end {ys[-1]:8.1f}  trend {slope:+.3f}/hour")

last = rows[-1]
print(f"sent {last['sent']}  recv {last['recv']}  gaps {last['gaps']}  "
      f"dups {last['dups']}  disconnects {last['disconnects']}")

How to interpret it:

• RSS trend near zero with a sawtooth in the raw data — normal garbage collection. A steady positive slope at constant load is a leak; multiply it by your deployment uptime to see whether it matters (5 MB/hour is 3.6 GB per month).
• File descriptors rising — something opens sockets or files without closing them, typically in the reconnect path.
• p99 rising while p50 stays flat — growing queues, longer garbage-collection pauses or a broker that is accumulating stored messages.
• Gaps that line up with disconnects — expected at QoS 0, a bug at QoS 1 with a persistent session. Gaps without any disconnect point to the broker dropping messages (queue limits) or to a subscriber that cannot keep up.
• Broker messages/stored climbing — usually persistent sessions for clients that are gone. Set a session expiry (MQTT 5) or clean up stale client IDs.

Make the soak test realistic

A constant stream from two clients finds client leaks. To find broker leaks, the traffic must exercise the code paths that allocate memory:

1. Connection churn — a fraction of clients disconnect and reconnect every few minutes, some cleanly and some by dropping the socket so the Last Will fires.
2. Session variety — mix clean sessions with persistent ones, and include a few persistent clients that go offline for hours.
3. Topic variety — publish to thousands of distinct topics, not one; include retained messages and wildcard subscribers.
4. Real payload sizes — generate them with an MQTT test data generator instead of fixed strings.
5. Network faults — inject a few short outages to confirm recovery; MQTT reconnect testing shows how.

Before you start the three-day run, check the topic and credentials with the online MQTT client and confirm messages arrive.

Watch soak traffic in your browser

A soak test is cheap: one small machine, one script, and patience. Run it before every major release and after every broker or library upgrade, keep the CSVs, and compare the trend lines — slow degradation is only invisible when nobody measures it.

Frequently asked questions