Store and Forward MQTT: Buffering Edge Data During Outages and Replaying It in Order

Store and forward MQTT is the pattern that keeps industrial data intact when the network between the edge and the cloud goes down. Instead of dropping messages while the broker is unreachable, the edge device writes them to local storage and replays them, in order, once the connection returns. This article explains how the pattern works, the design decisions that matter, and gives you a runnable Python implementation backed by SQLite.

Why MQTT Alone Is Not Enough

MQTT has strong delivery features: QoS 1 and 2 acknowledgements, persistent sessions and retained messages. But they all start working only once a message has left the device. Consider what happens during a two-hour WAN outage at a plant:

• The gateway keeps polling machines and producing values.
• The MQTT client cannot connect, so there is no session to deliver into.
• Most client libraries keep unsent messages in memory, with a size limit, and lose them on a restart or power cut.

Persistent sessions on the broker help the other direction: they hold messages for an offline subscriber. They do nothing for a publisher that cannot reach the broker. That gap is what store and forward fills, and it is why buffering is listed as a core job of every IIoT edge gateway.

How Store and Forward MQTT Works

  Data source --> [ write to local queue (disk) ] --> forwarder --> MQTT broker
                         ^                               |
                         |      delete only after        |
                         +------- PUBACK received -------+

1. Every message is written to a durable local queue first, with a sequence number and the source timestamp.
2. A forwarder reads the oldest messages and publishes them with QoS 1.
3. Only after the broker acknowledges (PUBACK) is the message deleted from the queue.
4. When the connection fails, the forwarder stops, the queue grows, and data sources carry on unaffected.
5. On reconnect, the forwarder drains the backlog oldest-first at a controlled rate, then continues with live data.

Routing all data through the queue, not just data produced during outages, keeps ordering simple: there is only one path, and it is always first-in, first-out.

Design Decisions That Matter

Sizing the Buffer

Estimate it from three numbers: messages per second during normal operation, average message size including overhead, and the longest outage you want to survive. For example, 20 messages per second of 200 bytes over a 24-hour outage is about 345 MB of payload, plus storage overhead. If that is too much, reducing traffic at the source with report by exception often shrinks the requirement dramatically.

Choosing a Drop Policy

When the buffer is full, you must choose: drop the oldest data (keep recent state) or drop the newest (keep a contiguous history from the start of the outage). For monitoring, dropping oldest is usually right. For compliance or batch records, you may prefer to stop accepting new low-priority data and preserve the beginning. Either way, publish a counter of dropped messages so the gap is visible.

Replay and the Retained Last Value

Replaying hours of history to a topic that dashboards also use as "current value" can make a gauge jump through old values. Two common fixes are to send backlog to a separate topic such as .../history, or to include the source timestamp and let consumers ignore older values for display. With MQTT 5 you can also set a message expiry interval on values that are worthless once stale, so the broker discards them instead of delivering them late.

A Store and Forward Implementation in Python

The following script uses SQLite as the durable queue and paho-mqtt 2.x as the client. A producer thread generates a reading every second, regardless of connectivity. The forwarder publishes the oldest rows with QoS 1, waits for each acknowledgement, deletes delivered rows and throttles replay to stay under the public broker's limit of 10 messages per second.

import json
import sqlite3
import threading
import time
import random
from datetime import datetime, timezone

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
CLIENT_ID = "saf-gateway-2b7c"
TOPIC = "connectmq/demo/saf-2b7c/plant1/compressor1/telemetry"
DB_PATH = "saf_queue.db"
MAX_ROWS = 500_000          # buffer limit; oldest rows dropped beyond this
REPLAY_RATE = 8             # messages per second (public broker allows 10)
ACK_TIMEOUT = 10            # seconds to wait for PUBACK

db_lock = threading.Lock()
db = sqlite3.connect(DB_PATH, check_same_thread=False)
db.execute("PRAGMA journal_mode=WAL")
db.execute("""CREATE TABLE IF NOT EXISTS queue (
                seq INTEGER PRIMARY KEY AUTOINCREMENT,
                topic TEXT NOT NULL,
                payload TEXT NOT NULL)""")
db.commit()


def enqueue(topic, data):
    with db_lock:
        db.execute("INSERT INTO queue (topic, payload) VALUES (?, ?)",
                   (topic, json.dumps(data)))
        count = db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
        if count > MAX_ROWS:
            db.execute("DELETE FROM queue WHERE seq IN "
                       "(SELECT seq FROM queue ORDER BY seq LIMIT ?)", (count - MAX_ROWS,))
        db.commit()


def oldest(n=50):
    with db_lock:
        return db.execute("SELECT seq, topic, payload FROM queue ORDER BY seq LIMIT ?",
                          (n,)).fetchall()


def delete(seq):
    with db_lock:
        db.execute("DELETE FROM queue WHERE seq = ?", (seq,))
        db.commit()


def producer():
    """Simulated data source: never blocks on the network."""
    while True:
        enqueue(TOPIC, {
            "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
            "pressure_bar": round(7 + random.gauss(0, 0.1), 3),
            "motor_temp_c": round(65 + random.gauss(0, 0.5), 2),
        })
        time.sleep(1)


def on_connect(client, userdata, flags, reason_code, properties):
    print("Connected:", reason_code)


def on_disconnect(client, userdata, flags, reason_code, properties):
    print("Disconnected:", reason_code, "- buffering locally")


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=60)
# Production: client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"); client.tls_set(); PORT = 8883
client.connect_async(BROKER, PORT, keepalive=30)
client.loop_start()

threading.Thread(target=producer, daemon=True).start()

interval = 1.0 / REPLAY_RATE
while True:
    if not client.is_connected():
        time.sleep(1)
        continue
    rows = oldest()
    if not rows:
        time.sleep(0.5)
        continue
    for seq, topic, payload in rows:
        data = json.loads(payload)
        data["seq"] = seq                      # lets consumers deduplicate
        try:
            info = client.publish(topic, json.dumps(data), qos=1)
            info.wait_for_publish(timeout=ACK_TIMEOUT)
        except (RuntimeError, ValueError) as exc:
            print("Publish failed:", exc)
            break
        if not info.is_published():
            print("No PUBACK, will retry from seq", seq)
            break                              # keep row, retry later in order
        delete(seq)
        time.sleep(interval)
    with db_lock:
        backlog = db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]
    if backlog > 1:
        print("Backlog:", backlog)

To test it, run the script, watch the topic in another terminal, then disconnect your network for a minute or two:

mosquitto_sub -h broker.connectmq.com -p 1883 -i saf-2b7c-viewer -t "connectmq/demo/saf-2b7c/#" -v

While offline, the script prints the disconnect and keeps queueing. When the network returns, the client reconnects automatically and the backlog drains in sequence order. The ts field shows when each value was measured, and seq increases without gaps. You can also stop the script entirely during the outage; on restart, the SQLite queue is still there.

Notes on the Implementation

• One message in flight at a time keeps ordering trivial and is fast enough for most gateways. For higher throughput, publish a small window and delete rows as their acknowledgements arrive.
• At least once, not exactly once. If the PUBACK is lost after the broker accepted the message, the row is resent. The seq field lets a historian or subscriber drop duplicates. MQTT QoS 2 would avoid duplicates on the wire, but not across a gateway restart, so application-level IDs are still worth having. See MQTT QoS levels explained for the trade-offs.
• WAL mode in SQLite allows reads and writes to proceed concurrently with good durability.
• Throttled replay protects the broker and downstream consumers. Shared plans have per-client limits, and the ConnectMQ free Serverless plan supports QoS 1, which is exactly what this pattern needs.

Store and Forward MQTT Checklist

1. Write every message to disk before attempting to send it.
2. Include a source timestamp and a sequence number in every payload.
3. Delete from the buffer only after the broker acknowledges.
4. Replay oldest first, throttled below broker and consumer limits.
5. Define a maximum buffer size and a drop policy, and report drops.
6. Decide how replayed history interacts with "current value" topics.
7. Test with real outages: pull the cable, reboot the gateway mid-outage, fill the buffer.

Store and Forward on Cellular and Remote Sites

Remote assets on cellular links lose connectivity more often and pay per byte. There, store and forward combines naturally with batching (several readings per message) and compression. The MQTT over cellular guide covers keep alive tuning and data budgets for those links, and the industrial IoT with MQTT guide shows where buffering fits in a full plant architecture.

A Reliable Broker on the Other End

Store and forward protects the edge side of the connection. On the cloud side, you need a broker that is reachable, secured with TLS and sized for replay bursts. A hosted MQTT broker takes broker operations off your plate, and the MQTT broker in the cloud guide explains how to compare options.

Get Started with ConnectMQ

Point your buffered gateways at ConnectMQ: TLS on port 8883, issued credentials, MQTT 5 message expiry and a live topic explorer to watch a backlog drain after an outage. Create a free Serverless broker at app.connectmq.com, no credit card required.

Frequently Asked Questions

Does MQTT have built-in store and forward?

Not for publishers that cannot reach the broker. QoS and persistent sessions work once a message reaches the broker; buffering before that is the edge application's job.

Should I buffer in memory or on disk?

On disk. Memory buffers are lost on a restart or power cut, which often happens during the same events that take the network down.

How do consumers handle duplicate messages after replay?

Include a sequence number or unique ID in each payload and let consumers ignore IDs they have already stored. Historians often deduplicate on timestamp and tag as well.

Will a large replay overload the broker?

It can hit rate limits or overwhelm subscribers if sent all at once. Throttle replay to a fixed rate, as in the example, and size the rate to your plan.