Optimizing MQTT QoS Levels for Pharmaceutical Telemetry

In a validated cold chain, the delivery guarantee you assign to each MQTT message is a data-integrity decision, not a networking preference. FDA 21 CFR Part 11 §11.10(e) requires complete, contemporaneous, computer-generated audit trails, and EU GMP Annex 11 §7.1 requires that recorded data be protected from loss and unauthorized alteration. A telemetry payload silently dropped during an RF dead-spot violates the ALCOA+ Complete attribute just as surely as a deleted record does; a duplicate that lands in an unguarded time series corrupts the Accurate attribute. This guide shows how to choose an MQTT Quality of Service (QoS) level per telemetry stream, implement it in production Python, and prove the configuration to an inspector. It assumes you have already settled the polling versus push question in favour of push.

How MQTT QoS Maps to Data-Integrity Obligations

The three QoS levels differ at the wire level, and those differences map directly onto regulatory constraints:

QoS 0 (at most once) sends a single PUBLISH with no acknowledgment. A reading lost to insulated-warehouse RF attenuation simply disappears, which breaks the Complete principle that §11.10(e) audit trails depend on. Reserve it for redundant, safety-irrelevant traffic.
QoS 1 (at least once) confirms every payload with a PUBACK, creating a verifiable delivery chain. If the PUBACK is lost, the publisher resends with DUP=1, so the consumer must be idempotent. This is the workhorse level for regulated telemetry.
QoS 2 (exactly once) guarantees no duplicates through the PUBREC / PUBREL / PUBCOMP four-step handshake, satisfying Annex 11’s prohibition on duplicate-induced alteration, but the broker-side state it holds adds latency that hurts high-frequency sampling.

There is no single correct QoS for a facility; the right level is chosen per stream against the consequence of loss and duplication. The matrix below is the decision most teams converge on for refrigerated and frozen storage.

Telemetry stream	Recommended QoS	Rationale	Regulatory anchor
Temperature / humidity (1-min intervals)	QoS 1	Warehouse RF gaps require acknowledgment, but exactly-once latency is unnecessary if ingestion is idempotent	21 CFR Part 11 §11.10(e); ALCOA+ Complete
Door open / close events	QoS 1	State changes must be logged once per cycle; idempotent keying prevents access-audit corruption	EU GMP Annex 11 §7.1
Sensor heartbeat / status ping	QoS 0	Redundant by design; a single lost ping affects neither product safety nor the record	Operational efficiency only
Critical excursion alarms	QoS 1 + retained	Retained messages give newly connected dashboards the last alarm state immediately	WHO TRS 992 Annex 9 §9.3.2
Configuration / OTA updates	QoS 2	Threshold or firmware changes applied twice cause calibration drift	EU GMP Annex 11 §7.1

Prerequisites

Python 3.11 or newer (the example relies on paho-mqtt v2 and MQTT v5 properties).
Client library: pip install "paho-mqtt>=2.1,<3".
Broker: an MQTT v5 broker that honours Session Expiry Interval and retained messages — Mosquitto 2.x, EMQX 5.x, or HiveMQ. Persistent storage for the session queue must be enabled.
Transport security: mutual TLS terminated at the broker. Telemetry should reach the broker through a hardened edge node; see designing secure IoT gateways for pharma logistics for the mTLS gateway pattern this guide assumes upstream.
Access control: per-device credentials with topic-scoped ACLs so a gateway can publish only to its own zone (pharma/coldchain/<zone>/#), satisfying Annex 11’s least-privilege expectation.
Sink: a time-series database that supports an atomic UPSERT keyed on (sensor_id, timestamp).

Step-by-Step Implementation

Step 1 — Establish a persistent MQTT v5 session

A persistent session (clean_start=False) lets the broker hold unacknowledged QoS 1 messages while a gateway is offline, so a network partition does not erase the record. In paho-mqtt v2 the clean_start flag and the Session Expiry Interval are passed to connect(), not to the Client constructor.

python

import json
import logging
import time
import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

# Structured, append-only logging is the evidence layer for §11.10(e) audit trails.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[logging.FileHandler("pharma_mqtt_audit.log"), logging.StreamHandler()],
)


class PharmaTelemetryClient:
    SESSION_EXPIRY_SECONDS = 86_400  # 24h broker-side queue retention survives long outages

    def __init__(self, broker: str, port: int, client_id: str):
        self.broker = broker
        self.port = port
        self.client = mqtt.Client(
            CallbackAPIVersion.VERSION2,
            client_id=client_id,   # stable client_id is required for session resumption
            protocol=mqtt.MQTTv5,
        )
        self.client.on_connect = self._on_connect
        self.client.on_message = self._on_message
        self.client.on_publish = self._on_publish
        self.client.on_disconnect = self._on_disconnect

    def connect(self):
        # Session Expiry travels as an MQTT v5 CONNECT property; clean_start=False
        # tells the broker to resume the prior session and flush queued QoS 1 messages.
        props = Properties(PacketTypes.CONNECT)
        props.SessionExpiryInterval = self.SESSION_EXPIRY_SECONDS
        self.client.connect(self.broker, self.port, keepalive=60,
                            clean_start=False, properties=props)
        self.client.loop_start()

Verify the library and protocol level before going further:

bash

python -c "import paho.mqtt; print(paho.mqtt.__version__)"   # expect 2.1.x or newer

Telemetry and alarms subscribe at QoS 1; alarms additionally publish with retain=True so a dashboard that reconnects after a time-series alignment restart immediately sees the last known excursion state.

python

    def _on_connect(self, client, userdata, connect_flags, reason_code, properties):
        if reason_code != 0:
            logging.error("Connect refused: %s", reason_code)
            return
        logging.info("Connected; session_present=%s", connect_flags.session_present)
        client.subscribe("pharma/coldchain/zone_a/telemetry", qos=1)  # §11.10(e) delivery chain
        client.subscribe("pharma/coldchain/zone_a/alarms", qos=1)

    def publish_telemetry(self, topic: str, payload: dict, qos: int = 1):
        retain = "alarm" in topic  # WHO TRS 992 §9.3.2: alarms must be immediately available
        info = self.client.publish(topic, json.dumps(payload), qos=qos, retain=retain)
        info.wait_for_publish(timeout=5.0)
        # Fail loudly: an un-acked critical publish is an audit-trail gap, not a warning.
        assert info.rc == mqtt.MQTT_ERR_SUCCESS, f"publish failed rc={info.rc}"
        logging.info("Published %s qos=%d mid=%s retain=%s", topic, qos, info.mid, retain)

    def _on_publish(self, client, userdata, mid, reason_code, properties):
        logging.debug("Broker acknowledged mid=%s", mid)

Confirm the retained alarm is genuinely held by the broker with a fresh subscriber:

bash

mosquitto_sub -h mqtt.broker.internal -t 'pharma/coldchain/zone_a/alarms' -C 1 -v
# A retained payload prints immediately on connect; a non-retained topic blocks.

Step 3 — Consume idempotently and never drop the DUP flag

This is the step practitioners get wrong. msg.dup=True means the broker is redelivering because a prior PUBACK was lost — it does not mean the payload is a semantic duplicate. Discarding dup=1 messages can drop the only copy that ever arrives, violating ALCOA+ Complete. Deduplication belongs at the database boundary, keyed on (sensor_id, timestamp).

python

    def _on_message(self, client, userdata, msg):
        if msg.dup:
            # Annex 11 §7.1: redelivery is expected; correctness is enforced by the
            # idempotent UPSERT below, not by dropping flagged messages here.
            logging.info("Redelivery flag on %s; deferring to idempotent ingest.", msg.topic)
        try:
            payload = json.loads(msg.payload.decode("utf-8"))
        except json.JSONDecodeError:
            logging.error("Malformed JSON on %s; quarantining.", msg.topic)
            return
        self._upsert_reading(payload)

    @staticmethod
    def _upsert_reading(payload: dict):
        # The deduplication boundary: a primary key on (sensor_id, timestamp) makes a
        # replayed QoS 1 message a no-op, satisfying ALCOA+ Accurate without losing data.
        # INSERT INTO readings (...) VALUES (...)
        #   ON CONFLICT (sensor_id, timestamp) DO UPDATE SET temp_c = EXCLUDED.temp_c;
        logging.info("Upsert ts=%s sensor=%s temp=%s",
                    payload.get("timestamp"), payload.get("sensor_id"), payload.get("temp_c"))

Assert the idempotency contract in a unit test before release:

python

# Replaying an identical reading must not change the row count.
before = count_rows()
client._upsert_reading({"sensor_id": "T-8842", "timestamp": 1_700_000_000, "temp_c": 2.4})
client._upsert_reading({"sensor_id": "T-8842", "timestamp": 1_700_000_000, "temp_c": 2.4})
assert count_rows() == before + 1, "UPSERT is not idempotent — QoS 1 redelivery will duplicate"

Step 4 — Preserve order when the broker flushes a backlog

When a gateway reconnects after an outage, the broker delivers the queued QoS 1 messages in a burst. Persist them by their original payload timestamp, never by broker arrival time, or the burst will record out-of-order rows and break contemporaneous sequencing. For high-volume bursts, route the flush through async batching strategies so a flood of buffered readings does not saturate the connection pool.

python

    def _on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties):
        # Unacked QoS 1 messages remain queued broker-side because the session persists,
        # so disconnect is recoverable rather than a data-loss event.
        logging.warning("Disconnected (%s); queued messages await resume.", reason_code)


if __name__ == "__main__":
    c = PharmaTelemetryClient("mqtt.broker.internal", 1883, "pharma_gateway_01")
    c.connect()
    time.sleep(5)
    c.publish_telemetry(
        "pharma/coldchain/zone_a/telemetry",
        {"timestamp": int(time.time()), "temp_c": 2.4, "humidity_pct": 45.1, "sensor_id": "T-8842"},
        qos=1,
    )

Reproduce the gap-recovery path and confirm no readings are lost:

bash

# Sever the broker for 15 minutes, publish throughout, then restore and reconcile.
sudo tc qdisc add dev eth0 root netem loss 100%   # simulate WAN loss
sleep 900 && sudo tc qdisc del dev eth0 root       # restore; broker flushes the queue

Compliance Validation Checklist

Run this checklist as part of computerized-system validation; each item is something an auditor can independently confirm.

Delivery guarantee documented — the validation protocol states the QoS chosen per stream and the loss/duplication consequence that justifies it.
Persistent session proven — clean_start=False with SessionExpiryInterval longer than the worst-case expected outage, evidenced by a network-degradation test that delivers every queued QoS 1 message on restoration.
Idempotency enforced — a unit test replays an identical (sensor_id, timestamp) payload and asserts no new row, demonstrating §11.10(e) integrity under redelivery.
DUP handling reviewed — code review confirms msg.dup=True messages are processed, not discarded.
Retained alarms verified — a fresh subscriber receives the last excursion alarm immediately (WHO TRS 992 §9.3.2).
Audit log reconciled — pharma_mqtt_audit.log maps 1:1 to database commits for every critical stream.
Broker resources bounded — max_queued_messages and retained-topic counts are capped so QoS 2 or alarm backlogs cannot exhaust memory.
Payload conformance mapped — each field traces to a regulatory requirement per how to map 21 CFR Part 11 requirements to MQTT payloads.

Troubleshooting

Symptom	Root cause	Fix
Missing telemetry during RF dropouts	QoS 0 configured, or `clean_start=True` discarded the session	Switch the stream to QoS 1, set `clean_start=False`, and set `SessionExpiryInterval` longer than the longest expected outage
Duplicate rows in the time-series DB	Broker redelivery after a lost `PUBACK`	Add an idempotent UPSERT on `(sensor_id, timestamp)`; do not drop `msg.dup=True` — that flag signals redelivery, not semantic duplication
Broker memory exhaustion	Excess QoS 2 in-flight state or a retained-alarm backlog	Audit retained topics, downgrade non-critical streams to QoS 1, and set `max_queued_messages`
Delayed alarm delivery	High-frequency QoS 2 traffic blocking broker threads	Isolate alarm topics, keep alerts at QoS 1, and monitor event-loop health from `loop_start`
Out-of-order rows after reconnect	Persisting by broker arrival time instead of payload time	Sort by the source `timestamp` before commit and quarantine readings older than the last committed value

For architectural context, see Polling vs Push Architectures for Pharma IoT Sensors, part of the broader IoT Sensor Data Ingestion & Time-Series Synchronization section.