6 MQTT Session Management

The Mistake: Developers configure clean_session=false on publishing devices (sensors), expecting the broker to buffer their outbound messages when the network is down.

Why It Happens: The term “persistent session” suggests the broker will collect publications that a disconnected device has not yet sent. It does not. The client and broker may retain state for QoS 1/2 exchanges already in progress, but a new reading created while the publisher is offline must be buffered by that publisher if it needs later delivery.

The Fix: Implement local message buffering on the publisher side. When mqttClient.connected() returns false, store messages locally (flash or a bounded RAM buffer) and publish them on reconnection. The sketch below uses PubSubClient, which publishes at QoS 0; for acknowledged QoS 1/2 delivery, choose a client library with those publish modes and retain records until acknowledgment.

// ESP32 local buffering pattern
#define BUFFER_SIZE 100
struct BufferedMessage {
  char topic[64];
  char payload[256];
};
BufferedMessage buffer[BUFFER_SIZE];
int bufferIndex = 0;

void publishWithBuffer(const char* topic, const char* payload) {
  if (bufferIndex == BUFFER_SIZE) return;  // Explicit overflow policy needed in production
  strncpy(buffer[bufferIndex].topic, topic, 63);
  strncpy(buffer[bufferIndex].payload, payload, 255);
  buffer[bufferIndex].topic[63] = '\0';
  buffer[bufferIndex].payload[255] = '\0';
  bufferIndex++;

  if (mqttClient.connected()) {
    int sent = 0;
    while (sent < bufferIndex && mqttClient.publish(buffer[sent].topic, buffer[sent].payload)) sent++;
    for (int i = sent; i < bufferIndex; i++) buffer[i - sent] = buffer[i];
    bufferIndex -= sent;
  }
}

MQTT 3.1.1 Spec Reference: §3.1.2.4 defines CleanSession behavior and §4.1 lists session state on both peers, including in-flight QoS exchanges. Neither makes the broker a buffer for publications the offline client has not sent.

The Mistake: Developers enable persistent sessions (clean_session=false) but use auto-generated or random client IDs like ESP32_ + random() or allow the library to generate one.

Why It Happens: Many MQTT libraries generate unique client IDs automatically to avoid ID collisions. Developers don’t realize this breaks persistent session restoration. Each reconnection creates a new session with different ID, so queued messages and subscriptions from the previous session are orphaned and eventually expire.

The Fix: Use a stable, unique client ID derived from hardware identifiers. For ESP32, use the MAC address or chip ID. In MQTT 5, a broker may assign an ID to an empty-ID client; persist the Assigned Client Identifier and reuse it on reconnect with a nonzero Session Expiry Interval.

// MQTT 3.1.1: Derive stable ID from hardware
char clientId[32];
uint64_t chipId = ESP.getEfuseMac();  // Unique per chip
snprintf(clientId, sizeof(clientId), "ESP32_%04X%08X",
         (uint16_t)(chipId >> 32), (uint32_t)chipId);

// Connect with persistent session
if (mqttClient.connect(clientId, user, pass,
                       willTopic, willQos, willRetain, willMessage,
                       false)) {  // clean_session = false
  // Check Session Present; stored subscriptions and queues exist only if resumed.
}

# MQTT 5: request an assigned ID, if the server permits it.
from pathlib import Path
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

id_file = Path("mqtt-client-id.txt")
stored_id = id_file.read_text().strip() if id_file.exists() else ""
connect_properties = Properties(PacketTypes.CONNECT)
connect_properties.SessionExpiryInterval = 3600  # seconds after disconnect
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=stored_id,  # Empty ID initially requests assignment
                     protocol=mqtt.MQTTv5)

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code == 0:
        assigned = getattr(properties, "AssignedClientIdentifier", None)
        if assigned:
            id_file.write_text(assigned)
        if not flags.session_present:
            client.subscribe("devices/commands", qos=1)

client.on_connect = on_connect
client.connect(broker, port, clean_start=not bool(stored_id),
               properties=connect_properties)
client.loop_start()
# On a later process start, the saved ID and clean_start=False request that
# session; Session Present in CONNACK says whether it was actually resumed.

This example can resume broker-side subscriptions and eligible queued messages. Paho Python does not persist its own in-flight QoS state across a process restart, so a stable ID alone does not guarantee loss-free recovery of unfinished client-side exchanges.

Broker Configuration (Mosquitto): Set persistent_client_expiration to control how long orphaned sessions are kept. Default is infinite, which can consume broker memory if clients use random IDs.

# mosquitto.conf - expire orphaned sessions after 7 days
persistent_client_expiration 7d

The Mistake: Developers configure QoS 2 on the publisher side, expecting guaranteed exactly-once delivery to subscribers, but subscribers connect with QoS 0 or 1. They’re confused when messages are duplicated or lost at the subscriber despite using QoS 2 for publishing.

Why It Happens: MQTT QoS applies separately to publisher-to-broker and broker-to-subscriber segments. The broker sends at the minimum of the original publication QoS and the maximum QoS it grants for the matching subscription. A QoS 2 publication can therefore reach a subscriber at QoS 0.

The Fix: Request QoS 2 on the subscription and verify the broker grants it when exactly-once transfer on that MQTT leg is required. Both MQTT legs must use QoS 2 to avoid protocol-level duplicate delivery across the route. Application-level execution still needs an idempotency key and outcome check.

# Publisher: QoS 2 for critical command
client.publish("factory/line1/emergency_stop", "STOP", qos=2)

# Subscriber requests QoS 2 so the broker can deliver at QoS 2
def on_connect(client, userdata, flags, reason_code, properties):
    # WRONG: QoS 0 subscription downgrades all deliveries
    # client.subscribe("factory/line1/emergency_stop", qos=0)

    # Request QoS 2 on this MQTT leg; verify granted QoS in SUBACK
    client.subscribe("factory/line1/emergency_stop", qos=2)

# Delivery QoS = min(original publication QoS, broker-granted subscription QoS)
# QoS 2 publish + QoS 0 subscribe = QoS 0 delivery (NO guarantee!)
# QoS 2 publish + granted QoS 2 = QoS 2 delivery on this MQTT leg

Production impact: In a manufacturing system, configuring emergency-stop publishers at QoS 2 does not guarantee exactly-once delivery to PLC subscribers that use QoS 0. During a network glitch, that mismatch can still produce lost or repeated stop events because the subscriber leg has been downgraded. Document and test the QoS contract on both sides of every critical topic.

The Mistake: Developers deploy hundreds of IoT devices with persistent sessions (clean_session=false) and long keep-alive intervals (300+ seconds). After a broker restart or failover, synchronized client retries can overload recovery with CONNECT packets and queued message delivery.

Why It Happens: Persistent sessions can survive a client disconnect while the broker remains running; durable storage is needed for those sessions to survive a broker restart. After a restart, 1,000 devices with 50 queued messages each have 50,000 messages to drain as they reconnect. Whether this creates a bottleneck depends on persistence settings, client retry timing and measured broker capacity.

The Fix: Implement staggered reconnection with exponential backoff and jitter. In Mosquitto, configure max_queued_messages to limit each client’s queue. For MQTT 5.0, use Session Expiry Interval to clean up stale sessions.

import random
import time
import paho.mqtt.client as mqtt

class MQTTClientWithBackoff:
    def __init__(self, client_id):
        self.client_id = client_id
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id, clean_session=False)
        self.base_delay = 1.0  # 1 second base
        self.max_delay = 120.0  # 2 minute cap
        self.attempt = 0

    def connect_with_backoff(self, broker, port):
        while True:
            try:
                self.client.connect(broker, port)
                self.client.loop_start()
                self.attempt = 0  # Reset on success
                return
            except Exception as e:
                self.attempt += 1
                # Exponential backoff: 1s, 2s, 4s, 8s... capped at 120s
                delay = min(self.base_delay * (2 ** (self.attempt - 1)), self.max_delay)
                # Add jitter: random 0-50% of delay to spread reconnections
                jitter = random.uniform(0, delay * 0.5)
                total_delay = delay + jitter
                print(f"Reconnect attempt {self.attempt} in {total_delay:.1f}s")
                time.sleep(total_delay)

# Broker configuration (mosquitto.conf)
# Limit queue buildup per client (Mosquitto 2.x)
# max_queued_messages 1000
# max_inflight_messages 20
#
# Mosquitto-specific expiration for persistent clients that do not reconnect:
# persistent_client_expiration 1h
# For MQTT 5, the client requests a Session Expiry Interval in CONNECT properties.

Sizing Guide: For N devices with Q queued messages and M payload bytes per message, N x Q x M gives a payload subtotal. For 1,000 devices x 100 messages x 500 bytes, that is 50,000,000 bytes (50 MB decimal). Add measured queue metadata, session state and broker overhead. Delivery rate after restart depends on reconnect timing, broker flow control and client Receive Maximum; it need not drain immediately. Set memory headroom from a measured recovery drill.