How to Build a Manufacturing Traceability System with MQTT Events

A manufacturing traceability system answers two questions quickly and with evidence: what went into this product and how was it made, and which other products share the same suspect material or conditions. This guide shows how to design traceability events, publish them over MQTT from every station, store them in a database and query them when a customer complaint or a supplier recall arrives.

What Traceability Really Requires

Traceability is often reduced to "we print a serial number". A useful system records much more:

1. Identity. A unique serial for each unit, or a batch or lot ID where unit-level tracking is impractical.
2. Genealogy. Which component serials or material lots were consumed into which unit. This is what enables backward and forward tracing.
3. Route. Which stations the unit passed through, when, and on which line.
4. Process data. The parameters in effect while the unit was processed: torque and angle for a screw, temperature and time for a cure, pressure for a press fit.
5. Results. Test and inspection outcomes, measured values, and rework history.

With these five records you can trace backward (from a failed unit to its materials, machines and parameters) and forward (from a suspect material lot or a bad time window to every unit affected). Forward tracing is what turns a recall from "everything made in March" into a precise list.

Why Event-Based Traceability Works

Traditional traceability is often built by having each station write directly into a central database or MES over a custom interface. That works until you add a station, change a vendor or need the data somewhere else. An event-based approach decouples the producers from the consumers: every station publishes what happened as a small message, and any number of systems subscribe.

MQTT suits this well. Stations and edge gateways publish over a lightweight protocol, QoS 1 ensures each event is delivered at least once, and new consumers, such as a quality dashboard or a real-time SPC service, can be added without touching the stations. If you are new to the pattern, read what an MQTT broker does first.

One rule matters above all: the broker transports events, the database stores them. An MQTT broker is not an archive, so the traceability database is a subscriber that persists every event.

Designing the Events

Keep the event vocabulary small and explicit. These six types cover most discrete manufacturing:

Every event carries an event_id (a UUID generated at the source) and a timestamp from a synchronised clock. The event ID makes storage idempotent: QoS 1 may occasionally deliver a message twice, and the database simply ignores the duplicate.

Topic structure

Use an ISA-95 style hierarchy so events are naturally grouped by plant, area, line and station:

acme/plant1/assembly/line2/station30/trace/event

A subscriber that needs every traceability event in the plant subscribes to acme/plant1/+/+/+/trace/event. The ISA-95 MQTT topics article covers this naming in depth, and the ConnectMQ portal recognises such topics and groups devices by level automatically.

Example payloads

{
  "event_id": "8f1c2e7a-5b9d-4c11-9e0f-2a7d61c3b4f0",
  "event": "component_consumed",
  "ts": "2026-10-16T07:52:13.118Z",
  "serial": "PMP-26-0041877",
  "station": "station20",
  "part_number": "MOT-0.75KW",
  "component_serial": "M2609-118422"
}

{
  "event_id": "c3a90b5e-0d2f-47b8-8d6e-91f4a7e02b1c",
  "event": "station_result",
  "ts": "2026-10-16T07:55:40.902Z",
  "serial": "PMP-26-0041877",
  "station": "station30",
  "result": "pass",
  "values": {"torque_nm": 12.4, "angle_deg": 186.0, "leak_rate_ml_min": 0.8},
  "program": "SCREW-P7",
  "operator": "OP114"
}

Building the Traceability Subscriber

The Python service below subscribes to all traceability events under a prefix, validates them and writes them to SQLite. It keeps the full event as JSON for completeness and extracts the fields you query most often into indexed columns. SQLite keeps the example self-contained; the same schema works in PostgreSQL for production. It uses paho-mqtt 2.x and the public ConnectMQ test broker.

import json
import sqlite3
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/a41b06"
db = sqlite3.connect("traceability.db", check_same_thread=False)
db.executescript("""
CREATE TABLE IF NOT EXISTS events (
    event_id TEXT PRIMARY KEY,
    event TEXT NOT NULL,
    ts TEXT NOT NULL,
    serial TEXT NOT NULL,
    station TEXT,
    topic TEXT,
    payload TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS ix_serial ON events(serial);
CREATE INDEX IF NOT EXISTS ix_station_ts ON events(station, ts);
CREATE TABLE IF NOT EXISTS genealogy (
    serial TEXT NOT NULL,
    component TEXT NOT NULL,
    part_number TEXT,
    PRIMARY KEY (serial, component)
);
CREATE INDEX IF NOT EXISTS ix_component ON genealogy(component);
""")

REQUIRED = {"event_id", "event", "ts", "serial"}

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/+/+/trace/event", qos=1)

def on_message(client, userdata, msg):
    try:
        e = json.loads(msg.payload)
    except json.JSONDecodeError:
        print("invalid JSON on", msg.topic)
        return
    if not REQUIRED.issubset(e):
        print("missing fields", e)
        return
    with db:
        db.execute("INSERT OR IGNORE INTO events VALUES (?,?,?,?,?,?,?)",
                   (e["event_id"], e["event"], e["ts"], e["serial"],
                    e.get("station"), msg.topic, json.dumps(e)))
        if e["event"] == "component_consumed":
            component = e.get("component_serial") or e.get("lot")
            db.execute("INSERT OR IGNORE INTO genealogy VALUES (?,?,?)",
                       (e["serial"], component, e.get("part_number")))

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="trace-writer-a41b06",
                     clean_session=False)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Two settings are deliberate. clean_session=False with a fixed client ID asks the broker to keep the subscription and queue QoS 1 messages while the writer is briefly offline, within the broker's session limits. INSERT OR IGNORE on the event ID turns duplicate deliveries into no-ops.

Publish a test event from a terminal:

mosquitto_pub -h broker.connectmq.com -p 1883 -i station30-a41b06 -q 1 \
  -t "connectmq/demo/a41b06/line2/station30/trace/event" \
  -m '{"event_id":"c3a90b5e-0d2f-47b8-8d6e-91f4a7e02b1c","event":"station_result","ts":"2026-10-16T07:55:40Z","serial":"PMP-26-0041877","station":"station30","result":"pass"}'

Querying the Traceability Data

The value of the system shows when someone asks a hard question. These three queries cover most investigations.

Backward trace: everything that happened to one unit.

SELECT ts, station, event, payload FROM events
WHERE serial = 'PMP-26-0041877' ORDER BY ts;

Forward trace: every unit that consumed a suspect component lot.

SELECT serial FROM genealogy WHERE component = 'LOT-SEAL-2609-07';

Time-window trace: every unit processed at a station while it was out of control.

SELECT DISTINCT serial FROM events
WHERE station = 'station30'
  AND ts BETWEEN '2026-10-16T07:30:00Z' AND '2026-10-16T08:15:00Z';

Combine the forward and time-window queries with shipment events and you can tell customer service exactly which pallets to hold.

Implementation Checklist

1. Define the serialisation point. Decide where serials are created and make that station publish unit_created.
2. Scan every traceable component. Genealogy is only as complete as the scanning discipline at assembly.
3. Synchronise clocks. Use NTP on every gateway and station PC; time-window queries depend on it.
4. Generate event IDs at the source. Never in the database, or duplicates become indistinguishable.
5. Publish with QoS 1. Use persistent sessions for the writer.
6. Buffer at the edge. Stations should queue events locally while the network is down and publish them on reconnect.
7. Validate and monitor. Count events per station per hour and alert when a station goes silent.
8. Define retention. Agree how long records must be kept for your industry and customers, and plan database storage accordingly.
9. Secure the pipeline with private credentials and TLS, following MQTT security best practices.

Connecting Traceability to Quality

Traceability is most powerful when linked to live quality data. When IoT quality control in manufacturing detects a failure, the traceability database tells you which upstream lots and machines were involved. When SPC flags an out-of-control period, the time-window query lists every unit to quarantine. The smart manufacturing guide shows how these systems share one data backbone.

Production Hosting

Traceability events are business records, so they belong on a private, authenticated broker, never on a public test broker. A fully managed MQTT service provides TLS endpoints and issued credentials for every station without you operating broker infrastructure. Plants with stricter isolation requirements can use a dedicated cluster or run the broker in their own cloud account under BYOC while still having it managed.

Build Your Traceability Backbone on ConnectMQ

ConnectMQ provides MQTT 3.1.1 and 5.0 with TLS, persistent sessions, QoS 1 and 2 on paid plans, ISA-95 topic detection and a live topic explorer to verify every station's events. Create a broker on the free Serverless plan at app.connectmq.com, no credit card required, and connect your first station today.

Frequently Asked Questions

What is a manufacturing traceability system?

It is a system that records the identity, components, route, process parameters and results of every unit or batch so you can trace both backward from a product and forward from a material or event.

Should traceability be by serial or by batch?

Use unit serials where products are high value or safety relevant, and batch or lot IDs where unit marking is impractical. Many plants combine both, with serialised units consuming batch-tracked materials.

Can the MQTT broker store traceability history?

No. The broker transports events. A subscriber must write every event into a database, which is the system of record.

How do I avoid duplicate records with QoS 1?

Generate a unique event ID at the source and make the database insert idempotent on that ID, so redelivered messages are ignored.