Multiroom MQTT: A Retained Message Can Outlive Its Sensor

1. What this experiment establishes

MQ Telemetry Transport (MQTT) is a publish/subscribe protocol: a publisher sends a payload to a named topic, a broker routes it, and a subscriber receives matching deliveries. A retained publication also replaces the broker's saved payload for that topic. That saved payload is useful when a display starts late. It does not carry an automatic measurement timestamp, expiration deadline or proof that the source remains powered.

The deliverable runs on one computer. The intended small server is an original Raspberry Pi 4 Model B; the actual supplied tests ran on an Apple Silicon host using a locally built Eclipse Mosquitto 2.1.2 broker and Eclipse Paho MQTT Python 2.1.0. They used only the loopback address 127.0.0.1. No Raspberry Pi execution, ESP32 firmware execution, radio test or physical room measurement is claimed. The two publishers generate demonstration counts, not temperatures, occupancy or door readings.

The proposed later room nodes are two Espressif ESP32-DevKitC V4 boards with ESP32-WROOM-32E modules. Their future firmware baseline is MicroPython 1.26.0, but firmware for that extension is not supplied or tested here. The complete executable lesson is the localhost experiment. Moving it onto a private wireless network requires a separate listener, access policy and tested client adapter; localhost on an ESP32 would mean that ESP32 itself, not the Pi. Do not change the supplied listener to a wildcard address to make that extension appear to work.

By the end you can reproduce a cached delivery after a subscriber restart, distinguish it from two observed live sequence values, force a publisher's Last Will, and explain why all three observations still fall short of proving sensor freshness. For the broader learning path, use the electronics course.

2. A network topology with explicit power paths

The temporary broker keeps retained payloads in random-access memory (RAM) and listens on no local area network (LAN) interface. Universal Serial Bus (USB) supplies power in the proposed hardware panel. The message arrows use Quality of Service (QoS) level 1. All general-purpose input/output (GPIO) headers in the proposed panel are unused.

The Pi uses its USB-C power connector and the recommended 5.1 volt (V), 3 ampere (A) supply. Each proposed DevKitC uses only its Micro-USB power connector and its on-board regulator. The board guide makes the Micro-USB, 5V-header and 3V3-header supply choices mutually exclusive. Leave both header supply inputs unused in this proposed arrangement. The module itself operates from 3.0–3.6 V; that is not permission to apply 5 V to its module supply or signal terminals.

There are no general-purpose input/output (GPIO) connections in this lesson, no added pull-up resistors and no board-to-board ground jumper. The USB cables include their own return conductors. Network communication is not a reason to join positive rails, and this drawing makes no claim of galvanic isolation between power supplies. Keep an ESP32 antenna clear in a future physical installation and identify the actual board/module revision before selecting firmware. The DevKitC guide's C15 note concerns earlier board boot/clock behavior; this experiment neither changes C15 nor drives the affected GPIO0 clock.

3. Give the application its own identity

The wire payload has six exact fields: schema version v, room, a sixteen-character hexadecimal boot, seq, uptime_ms and value. JavaScript Object Notation (JSON) encodes these fields. Here value is an integer demonstration count. A new publisher invocation chooses a new 64-bit random boot token and starts sequence zero. It sends at most sixteen publications, so this exercise never wraps its 16-bit allowed sequence range.

A boot token distinguishes publisher invocations; it is not authentication, silicon identity or a globally ordered epoch. A delayed old boot is not necessarily a more recent source. During one observer run, the policy keeps a bounded set of retired boots. Returning to one is a sticky fault, and exhausting the set also faults instead of silently forgetting history. Restarting the observer deliberately loses that local history. Broker replay after that restart therefore remains CACHE, regardless of the boot token inside it.

The policy accepts the first live payload for a boot as BASELINE. A later live payload with a strictly larger sequence and nondecreasing publisher uptime establishes PROGRESS. A jump can indicate omitted candidate publications, but does not identify where they were lost. An exact duplicate does not advance progress. The same sequence with different data, a decreasing sequence, decreasing uptime or an invalid envelope causes INVALID for the rest of that run. A different unseen boot starts a new baseline and clears the previous progress age.

Quality of Service (QoS) level 1 provides an at-least-once protocol delivery mechanism. Its duplicate (DUP) bit belongs to the transport packet, not to the identity of a business event. The application sequence is required even if DUP is false. Paho's documented reconnect behavior also makes it unsafe to assume that a clean session prevents every application duplicate. This program uses clean sessions and disables automatic reconnect; any loop failure terminates the experiment. A new run subscribes anew and treats replay as cache.

The local token is deliberately not a security mechanism. Another local process can publish a plausible envelope or replace a retained topic. Simultaneous publishers claiming the same room break the ownership premise. No code here proves the existence, condition or recent sampling of a real sensor.

4. Two ages that answer different questions

All observer ages use the same computer's monotonic clock. They are milliseconds (ms) since an observed callback: delivery_age_ms since the last valid envelope, including a duplicate or retained delivery, and progress_age_ms since the last accepted forward live sequence. Neither is a source measurement age. Publisher uptime comes from a different process origin; subtracting it from observer uptime would manufacture a timestamp.

delivery_age_ms = observer_now_ms - last_valid_delivery_ms
progress_age_ms = observer_now_ms - last_forward_delivery_ms
STALE when progress_age_ms > 5000

Example: forward delivery at 2000 ms; duplicate at 4000 ms
at 7001 ms: delivery age = 3001 ms; progress age = 5001 ms → STALE

The five-second threshold is an application choice for this demonstration, not a property of MQTT or a physical sensor's maximum age. At exactly 5000 ms the status remains PROGRESS; one millisecond later it is STALE. Scheduling delays, a paused observer and backlog can all change the observation. A newly delivered nonretained packet can itself contain old data. Two forward sequences establish only progress among deliveries seen by this observer.

Availability is deliberately separate. The publisher registers an offline Last Will before connecting and retains an online hint after connection. A normal finish explicitly publishes offline and sends DISCONNECT. An abrupt process exit skips that normal path, allowing the broker to send the Will when it detects the lost connection. Keep Alive is five seconds here, but this is not a guaranteed failure-detection deadline across server failure or arbitrary delay. An old retained online hint may survive its source. Hints never switch the observer's data state, refresh progress, or establish ordering across the two topics.

5. Start a disposable broker and two rooms

Use a Python 3 environment and Mosquitto 2.1.2. On a Pi, first inspect mosquitto -h; distribution packages may contain another version. Use the official release source if necessary and follow its build instructions. The author used an isolated source build, with Transport Layer Security (TLS) support compiled in because the macOS build needs an available random-number provider. The actual listener uses plain Transmission Control Protocol (TCP) only on localhost. This is a protocol laboratory, not a network deployment configuration. No system service is installed or enabled.

Create a dedicated directory, save the program and configuration below, and install the pinned Paho package in an isolated Python virtual environment. The standard Python package installer pip command changes only that environment. The program's client pins MQTT 3.1.1 and Paho's version-2 callback Application Programming Interface (API).

python3 -m venv .venv
.venv/bin/python -m pip install paho-mqtt==2.1.0
mosquitto -h
mosquitto -c mosquitto-loopback.conf

Run the last command in a foreground terminal, after confirming port 18883 is free. A bind failure is a stop condition; do not remove the explicit address or reuse an unknown running broker. Keep that terminal open. The configuration is complete:

# Foreground disposable lesson broker. No LAN listener, password or persistence.
listener 18883 127.0.0.1
allow_anonymous true
persistence false
log_dest stderr

In a second terminal run the observer. In a third and fourth terminal start rooms A and B. Open all terminals in the same directory. The two room processes own different topics even though they share the broker.

.venv/bin/python mqtt_room_lab.py observe --seconds 20
.venv/bin/python mqtt_room_lab.py publish --room a --count 4 --interval 1
.venv/bin/python mqtt_room_lab.py publish --room b --count 4 --interval 1

The four one-second publications create only three possible forward transitions after the first live baseline. If a publisher finishes before the observer subscribes, the observer sees cache instead. After a publisher stops, restart only the observer: its new subscription should receive the retained final state as CACHE, with no invented forward count. Stop the broker with Control-C at the end; persistence is disabled, so its saved topics are not deliberately written to disk. Starting a new broker process under this configuration loses that memory. This is different from restarting just a subscriber.

6. Complete program and checked failure boundaries

Save the following complete file as mqtt_room_lab.py. The callbacks execute through the manual network loop; there is no background Paho thread in this program. Waiting for a publish acknowledgement inside such a callback would block the loop that must receive it, so checked publishing happens only in the publisher's outer flow. The helper checks both the immediate return code and is_published(); a timeout-return from wait_for_publish() alone would not prove completion. A broker acknowledgement is still not a subscriber's processing acknowledgement.

"""Localhost-only MQTT 3.1.1 experiment; synthetic room counts, not sensor data."""
import argparse
import json
import math
import os
import secrets
import time

PREFIX = 'tricky097/lab'
ROOMS = ('a', 'b')
STALE_MS = 5000


def packet(room, boot, seq, uptime_ms, value):
    return json.dumps(dict(v=1, room=room, boot=boot, seq=seq,
                           uptime_ms=uptime_ms, value=value), separators=(',', ':'))


def decode(raw, room):
    if not isinstance(raw, bytes) or not 1 <= len(raw) <= 256:
        raise ValueError('PAYLOAD_SIZE')
    obj = json.loads(raw.decode('ascii'))
    if not isinstance(obj, dict) or set(obj) != {'v', 'room', 'boot', 'seq', 'uptime_ms', 'value'}:
        raise ValueError('FIELDS')
    if type(obj['v']) is not int or obj['v'] != 1 or obj['room'] != room:
        raise ValueError('IDENTITY')
    boot = obj['boot']
    if not isinstance(boot, str) or len(boot) != 16 or any(c not in '0123456789abcdef' for c in boot):
        raise ValueError('BOOT')
    for key, maximum in [('seq', 65535), ('uptime_ms', 3600000), ('value', 1000)]:
        if type(obj[key]) is not int or not 0 <= obj[key] <= maximum:
            raise ValueError(key)
    return obj


class Room:
    def __init__(self, name):
        if name not in ROOMS:
            raise ValueError('ROOM')
        self.name, self.boot, self.last, self.cache = name, None, None, None
        self.retired = set()
        self.delivered_at = self.progress_at = self.clock = None
        self.forward = 0
        self.fault = None
        self.availability_hint = None
        self.last_delivery_retained = None

    def invalidate(self, reason):
        self.fault = self.fault or reason

    def at(self, now):
        if type(now) is not int or now < 0 or (self.clock is not None and now < self.clock):
            self.invalidate('CLOCK')
            return False
        self.clock = now
        return True

    def accept(self, raw, retained, now):
        if not self.at(now) or self.fault:
            return
        try:
            obj = decode(raw, self.name)
            if type(retained) is not bool:
                raise ValueError('RETAIN_FLAG')
        except (ValueError, UnicodeError, TypeError):
            self.invalidate('PAYLOAD')
            return
        self.delivered_at = now
        self.last_delivery_retained = retained
        if retained:
            self.cache = obj
            return  # A replay is never a live baseline or observed progress.
        if obj['boot'] != self.boot:
            if obj['boot'] in self.retired or len(self.retired) >= 16:
                self.invalidate('BOOT_REUSE_OR_LIMIT')
                return
            if self.boot is not None:
                self.retired.add(self.boot)
            self.boot, self.last = obj['boot'], obj
            self.forward, self.progress_at = 0, None
            return
        if obj['seq'] == self.last['seq']:
            if obj != self.last:
                self.invalidate('SEQ_CONFLICT')
            return  # Exact duplicates cannot refresh progress age.
        if obj['seq'] < self.last['seq'] or obj['uptime_ms'] < self.last['uptime_ms']:
            self.invalidate('ORDER')
            return
        self.last, self.progress_at = obj, now
        self.forward += 1

    def snapshot(self, now):
        self.at(now)
        age = None if self.progress_at is None else now - self.progress_at
        status = ('INVALID' if self.fault else 'CACHE' if self.boot is None and self.cache else
                  'UNKNOWN' if self.boot is None else 'BASELINE' if age is None else
                  'STALE' if age > STALE_MS else 'PROGRESS')
        return dict(room=self.name, status=status, fault=self.fault, boot=self.boot,
                    seq=None if self.last is None else self.last['seq'],
                    value=None if self.last is None else self.last['value'],
                    delivery_age_ms=None if self.delivered_at is None else now-self.delivered_at,
                    progress_age_ms=age, observed_forward=self.forward,
                    availability_hint=self.availability_hint,
                    last_delivery_retained=self.last_delivery_retained,
                    cached_payload=self.cache)


def mqtt_module():
    import paho.mqtt.client as mqtt
    import paho.mqtt
    if paho.mqtt.__version__ != '2.1.0':
        raise RuntimeError('This lesson pins paho-mqtt 2.1.0')
    return mqtt


def open_client(port, client_id, on_message=None, subscriptions=(), will=None):
    """Manual network loop; never spawn a worker or reconnect automatically."""
    if type(port) is not int or not 1024 <= port <= 65535:
        raise ValueError('PORT')
    mqtt = mqtt_module()
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id,
                         clean_session=True, protocol=mqtt.MQTTv311,
                         reconnect_on_failure=False)
    client.connect_timeout = 2.0
    connected, subscribed = [], set()
    def on_connect(c, userdata, flags, reason, properties):
        if reason.is_failure:
            raise RuntimeError('CONNACK')
        connected.append(True)
        for topic in subscriptions:
            rc, mid = c.subscribe(topic, qos=1)
            if rc != mqtt.MQTT_ERR_SUCCESS:
                raise RuntimeError('SUBSCRIBE')
            subscribed.add(mid)
    def on_subscribe(c, userdata, mid, reasons, properties):
        if any(x.is_failure for x in reasons):
            raise RuntimeError('SUBACK')
        subscribed.discard(mid)
    client.on_connect, client.on_subscribe = on_connect, on_subscribe
    client.on_message = on_message
    if will is not None:
        client.will_set(*will, qos=1, retain=True)
    try:
        client.connect('127.0.0.1', port, keepalive=5)
        deadline = time.monotonic() + 3.0
        while not connected or subscribed:
            if time.monotonic() > deadline:
                raise TimeoutError('CONNECT/SUBACK')
            pump(client)
        return client
    except BaseException:
        client.disconnect()
        raise


def pump(client):
    rc = client.loop(timeout=0.05)
    if rc != 0:
        raise OSError('MQTT_LOOP_%s' % rc)


def publish_checked(client, topic, payload, retain=True):
    """QoS 1 broker acknowledgement; never call inside a network callback."""
    info = client.publish(topic, payload, qos=1, retain=retain)
    if info.rc != 0:
        raise OSError('PUBLISH_%s' % info.rc)
    deadline = time.monotonic() + 2.0
    while not info.is_published():
        if time.monotonic() >= deadline:
            raise TimeoutError('PUBACK_NOT_OBSERVED')
        pump(client)


def close_client(client):
    # DISCONNECT suppresses the Will; it does not clear retained messages.
    rc = client.disconnect()
    if rc != 0:
        raise OSError('DISCONNECT_%s' % rc)


def run_publisher(port, room, count=4, interval=1.0, abrupt=False, emit=print):
    if room not in ROOMS or type(count) is not int or not 1 <= count <= 16:
        raise ValueError('ROOM/COUNT')
    if type(interval) not in (float, int) or not math.isfinite(interval) or not 0.1 <= interval <= 5:
        raise ValueError('INTERVAL')
    boot = secrets.token_hex(8)
    state_topic = PREFIX + '/' + room + '/state'
    hint_topic = PREFIX + '/' + room + '/availability'
    offline = json.dumps(dict(boot=boot, hint='offline'))
    client = open_client(port, 'pub-'+room+'-'+boot, will=(hint_topic, offline))
    try:
        started = time.monotonic()
        publish_checked(client, hint_topic, json.dumps(dict(boot=boot, hint='online')))
        for seq in range(count):
            data = packet(room, boot, seq, int((time.monotonic()-started)*1000), seq)
            publish_checked(client, state_topic, data)
            emit(data)
            until = time.monotonic() + interval
            while time.monotonic() < until:
                pump(client)
        if abrupt:
            os._exit(23)  # Intentional test of connection loss, not a sensor result.
        publish_checked(client, hint_topic, offline)
    finally:
        close_client(client)


def observe(port, seconds=20, emit=print):
    if type(seconds) is not int or not 1 <= seconds <= 120:
        raise ValueError('SECONDS')
    rooms = {r: Room(r) for r in ROOMS}
    origin = time.monotonic_ns()
    def now():
        return (time.monotonic_ns()-origin)//1000000
    def on_message(client, userdata, msg):
        parts = msg.topic.split('/')
        if len(parts) != 4 or parts[:2] != PREFIX.split('/') or parts[2] not in ROOMS:
            raise ValueError('TOPIC')
        room = rooms[parts[2]]
        if parts[3] == 'state':
            room.accept(bytes(msg.payload), bool(msg.retain), now())
        elif parts[3] == 'availability':
            # Raw bounded hint is diagnostic only. It never changes progress policy.
            room.availability_hint = dict(
                payload=bytes(msg.payload[:128]).decode('ascii', errors='replace'),
                retained=bool(msg.retain), received_elapsed_ms=now())
    client = open_client(port, 'obs-'+secrets.token_hex(8), on_message,
                         (PREFIX+'/+/state', PREFIX+'/+/availability'))
    last_output = -1000
    try:
        while now() < seconds*1000:
            pump(client)
            stamp = now()
            if stamp-last_output >= 1000:
                emit(json.dumps(dict(elapsed_ms=stamp,
                                     rooms=[r.snapshot(stamp) for r in rooms.values()])))
                last_output = stamp
        return dict(closed_at_ms=now(), rooms=[r.snapshot(now()) for r in rooms.values()])
    finally:
        close_client(client)


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('mode', choices=['observe', 'publish'])
    p.add_argument('--port', type=int, default=18883)
    p.add_argument('--room', choices=ROOMS, default='a')
    p.add_argument('--count', type=int, default=4)
    p.add_argument('--interval', type=float, default=1)
    p.add_argument('--seconds', type=int, default=20)
    p.add_argument('--abrupt', action='store_true')
    a = p.parse_args()
    if a.mode == 'observe':
        print(json.dumps(observe(a.port, a.seconds)))
    else:
        run_publisher(a.port, a.room, a.count, a.interval, a.abrupt)


if __name__ == '__main__':
    main()

The observer runs for a requested 1–120 seconds, and a publisher attempts 1–16 demonstration values. These are software limits, not hard real-time watchdogs. Network calls, console output or operating-system scheduling can delay completion. A raised exception aborts the run and attempts the normal disconnect path. Some lines may already have been printed: they remain historical observations, not a successful final return or current room health. A clock failure immediately after connection is inside the publisher's cleanup boundary. Even cleanup can fail; an exception is not evidence that every remote resource is already closed.

For the intentional abrupt-loss case run .venv/bin/python mqtt_room_lab.py publish --room a --count 1 --interval 0.1 --abrupt while an observer is subscribed. The process exits with code 23 without normal cleanup. This deliberately tests a broker-detected connection loss; it does not simulate every failure of a real board. Do not use it as a normal shutdown command.

7. Record observations without upgrading them to measurements

1. Verify the two software versions and the loopback-only listener. Note host operating system and whether the process really stopped at the end. Start with a fresh foreground broker.
2. Start the observer before room A. Record baseline and forward deliveries. Repeat for B and confirm that its boot/sequence history is independent.
3. Let both publishers finish; restart only the observer. Record retained replay and CACHE. Check that a replay does not increment observed_forward.
4. Restart one publisher while the observer stays open. Its new boot starts BASELINE; a later live sequence may establish progress. That boot change is not a counted room event.
5. Observe an abrupt exit and then a normal exit. Record the availability hints actually received, their boot identifiers and their delivery metadata. Do not infer exact disconnection time from an offline hint.
6. Run the supplied host tests. They additionally send exact duplicate application envelopes, conflicting sequences, retired boots and malformed data; no sensor has to produce these faults.
7. Stop the broker and all client processes. Check that the disposable listener is gone. Keep raw observed values separate from the constructed graph.

The blank observation worksheet has no filled sample rows. Enter receiver elapsed time and callback flags; leave source measurement time unknown. A future real sensor adapter must state when it samples, how it handles its own stale readings and restarts, and how the selected client handles reconnects. A Wi-Fi connection, boot token or online hint cannot supply those missing facts.

The supplied host test archive contains this exact program, the actual-code tests, the broker configuration and a README. Set MOSQUITTO_BIN to the executable you have verified as 2.1.2, then run python test_mqtt_room.py in the Paho environment. The tests start a temporary broker bound to 127.0.0.1 on a free high port, terminate it and close their resources. They do not configure a permanent service. The earlier unrelated lessons on endpoint calibration, log rotation, Wi-Fi retry budgets and Bluetooth observation age do not establish these broker behaviors.

8. Challenge: why is “online” beside STALE reasonable?

Room A delivered boot A/sequence 10 at two seconds. At four seconds it retransmits exactly that envelope. At six seconds an old retained online hint arrives, and at 7.001 seconds the observer prints STALE. Should either later delivery restore PROGRESS? What should a newly started observer display if it receives only the retained state?

9. References and downloads

The protocol rules were checked against the Organization for the Advancement of Structured Information Standards (OASIS) MQTT 3.1.1 standard dated 29 October 2014; the implementation and configuration were checked against the exact pinned sources. Current product documentation and the ESP32 revision-specific errata entry point were reviewed for the proposed board extension, which remains unimplemented. No general claim of “no errata” or hardware qualification follows from this software test.

• OASIS MQTT 3.1.1, §§3.1.2.4–3.1.2.11, 3.3.1 and 3.14: clean sessions, Will, Keep Alive, retained delivery and disconnect.
• Eclipse Mosquitto official download and 2.1.2 configuration source: listener address, anonymous local access and persistence.
• Paho MQTT 2.1.0 release, pinned client source and documented limitations.
• Raspberry Pi hardware documentation and the 15 W USB-C power supply product brief: Pi 4 supply recommendation and 5.1 V / 3 A output.
• ESP32-DevKitC V4 guide, ESP32-WROOM-32E datasheet v2.1 and ESP32 revision-specific errata.
• MicroPython 1.26.0 ESP32 reference, proposed extension baseline only; no MQTT adapter or target execution is supplied.

Complete Python program · Local broker configuration · Blank worksheet · Actual-code host tests. Both diagrams also have editable SVG downloads beside their full-size Portable Network Graphics (PNG) links.