MQTT QoS Levels and Duplicate Sensor Messages #

The short answer is QoS 1 with a persistent session and an idempotent sink, and most of the difficulty in getting there comes from a single misunderstanding: MQTT’s quality-of-service levels describe the guarantee between one publisher and one broker, or between one broker and one subscriber — not end to end. A reading published at QoS 2 can still be lost by a subscriber that acknowledged it and crashed before writing. This guide sets out what each level actually buys for environmental telemetry, why duplicates are the cheap failure to design for, and how to absorb them in the sink. It extends the client configuration built in MQTT broker integration for environmental sensors.

What Each Level Costs, in Sensor Terms #

QoS 0 sends the packet once with no acknowledgement. On a stable link it is nearly lossless; on a cellular gateway it loses every message in flight during a reconnect, and reconnects are the norm rather than the exception. Its legitimate use is high-rate diagnostic data where a gap is irrelevant — a per-second battery voltage feed, for instance.

QoS 1 adds a PUBACK: the sender retransmits until acknowledged. Delivery is guaranteed and duplicates are expected, because a lost PUBACK causes a retransmit of a message that already arrived. Two round trips instead of one, and one extra broker state entry per in-flight message.

QoS 2 adds a four-step handshake that guarantees exactly-once delivery to the broker or to the subscribing client. It quadruples the round trips and doubles the broker’s per-message state. For telemetry it solves a problem that a unique index solves better: the four-step handshake protects the transport, but nothing protects you from a device that resends a reading after a power cycle, which arrives as a legitimately new QoS 2 message and a genuine duplicate reading.

That last point is the argument. Duplicate messages have several causes, and only one of them — transport retransmission — is what QoS 2 prevents. Since you need idempotency in the sink for the other causes anyway, paying four round trips to eliminate one of them is a poor trade. The Kafka guides in this section reach the same conclusion from the other direction.

Production-Ready Implementation #

The subscriber configuration first. Three settings decide whether QoS 1 delivers on its promise:

# python 3.11 · paho-mqtt==2.1.0 · psycopg[binary]==3.1.18
import paho.mqtt.client as mqtt

def build_subscriber(client_id: str) -> mqtt.Client:
    """Persistent-session subscriber: the broker queues QoS 1 messages across outages."""
    client = mqtt.Client(
        mqtt.CallbackAPIVersion.VERSION2,
        client_id=client_id,          # STABLE — a random id gets a fresh, empty session
        clean_session=False,          # keep the subscription and queue across disconnects
        protocol=mqtt.MQTTv311,
    )
    client.max_inflight_messages_set(100)
    client.max_queued_messages_set(0)     # 0 = unbounded client-side outbound queue
    return client

A stable client_id with clean_session=False is the whole mechanism: the broker recognises the returning client, replays what it queued, and the outage becomes latency rather than loss. Generating a random client id per process — the default in many examples — silently converts every restart into a data gap.

The sink is where duplicates are absorbed. The natural key is the reading’s identity, not the message’s:

CREATE TABLE reading (
    device_id    text        NOT NULL,
    observed_at  timestamptz NOT NULL,
    metric       text        NOT NULL,
    value        double precision NOT NULL,
    received_at  timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (device_id, observed_at, metric)
);

INSERT_SQL = """
INSERT INTO reading (device_id, observed_at, metric, value)
VALUES (%(device_id)s, %(observed_at)s, %(metric)s, %(value)s)
ON CONFLICT (device_id, observed_at, metric) DO NOTHING
RETURNING 1
"""

def write_batch(conn, rows: list[dict]) -> int:
    """Insert readings idempotently. Returns how many were genuinely new."""
    with conn.cursor() as cur:
        cur.executemany(INSERT_SQL, rows, returning=True)
        inserted = 0
        while True:
            if cur.fetchone() is not None:
                inserted += 1
            if not cur.nextset():
                break
    conn.commit()
    return inserted

Counting the genuinely-new rows is not decoration: len(rows) - inserted is your duplicate rate, and it is the single most useful number for diagnosing this layer. A steady 1–3% is normal QoS 1 retransmission. A sudden jump to 50% means a device is republishing its buffer, and a drop to zero alongside falling volume means messages are being lost before they reach the sink.

The message handler must not acknowledge before the write succeeds, which in paho means doing the write inside the callback (or handing it to a queue whose failure path re-raises):

def on_message(client, userdata, msg):
    reading = decode(msg.payload)            # raises on malformed payloads
    try:
        write_batch(userdata["conn"], [reading])
    except psycopg.OperationalError:
        client.loop_stop()                   # stop consuming; the broker keeps the queue
        raise

Stopping the loop on a sink failure is what makes the broker your overflow buffer. The alternative — catching the exception, logging it, and returning — acknowledges a message you never stored.

Parameter Tuning Guide #

The last row is worth internalising: during a buffered backlog replay almost everything arriving is a duplicate, by design. If your monitoring alerts on a high duplicate rate, it will alert on every successful recovery.

Verification and Testing #

The test that matters simulates the failure QoS 1 is meant to survive: an acknowledgement lost after the write.

# python 3.11 · pytest==8.2.0
def test_redelivery_after_lost_ack_does_not_duplicate_the_reading(conn):
    reading = {
        "device_id": "eui-70b3d5",
        "observed_at": "2026-08-01T10:00:00+00:00",
        "metric": "pm25",
        "value": 18.4,
    }
    assert write_batch(conn, [reading]) == 1      # first delivery: genuinely new
    assert write_batch(conn, [reading]) == 0      # redelivery: absorbed, not duplicated

    with conn.cursor() as cur:
        cur.execute("SELECT count(*) FROM reading WHERE device_id = 'eui-70b3d5'")
        assert cur.fetchone()[0] == 1

Then verify the persistent session end to end, because it is the setting most often wrong: connect a subscriber with clean_session=False, disconnect it, publish three messages at QoS 1, reconnect, and assert all three arrive. Run the same test with clean_session=True and watch all three vanish — it is the fastest way to convince a team that the flag is not a detail.

Gotchas #

A randomly generated client id defeats the persistent session. Libraries and container orchestrators both encourage per-process randomness. Derive the client id from a stable identity — the hostname, the pod’s ordinal in a StatefulSet, the device serial — or the broker will treat every restart as a new client with an empty queue.

Two subscribers sharing one client id. The broker allows only one connection per client id and will disconnect the incumbent when the second connects. Two replicas configured identically produce an endless disconnect loop that looks like a network fault. Use MQTT 5 shared subscriptions for horizontal scaling, not duplicate client ids.

Deduplicating on received_at instead of observed_at. A duplicate arrives at a different wall-clock time, so a key that includes the receive time deduplicates nothing. The natural key must be composed only of facts about the reading itself.

Unbounded max_queued_messages on the broker for an offline subscriber. The broker will queue until it runs out of memory. Set a limit per client that reflects how long an outage you intend to cover, and accept that beyond it the queue drops the oldest — which is the honest failure and the one your gap detection will notice.

FAQ #

Related #

• MQTT Broker Integration for Environmental Sensors — the broker integration these settings belong to
• MQTT TLS and ACL Hardening for Sensor Fleets — the security half of the same client configuration
• Building an At-Least-Once Faust Agent for Sensor Streams — the same at-least-once plus idempotent-sink pattern, in Kafka