Taking in MQTT and LoRaWAN telemetry in .NET without losing it
Persistent sessions, a bounded channel, batched writes and idempotency keys, so a deploy or a slow database never costs you a reading.
The easiest MQTT listener deserialises each message in the handler and saves it to the database before returning. It looks fine until the database has a slow minute. The handler blocks, the client stops reading the socket, keep-alives time out, the broker drops the connection, and whatever was in flight is gone if the session was not persistent. A deploy loses another window. Nobody notices until a customer asks why their graph has a gap at 14:05.
This is how I approach ingestion in LiquidInsights, the IoT and energy operations platform I build at Savtech, which takes telemetry from LoRaWAN devices through ChirpStack and from devices that publish MQTT directly. The design is simplified from that production platform to show the shape, not copied from it, and uses MQTTnet 4.x on .NET 8.
It comes down to one rule, which I wrote down more formally in the spec for Flux, a separate telemetry platform I am designing.
Key ideaA message may be lost only if it was never acknowledged.
Everything below exists to make that rule true without making the handler slow.
The client: a session the broker remembers
QoS 1 alone doesn't protect you. The broker only queues messages for a client it remembers: a stable client id and a session kept on disconnect. With MQTT 3.1.1 that's CleanSession = false. With MQTT 5 it is clean start false plus a session expiry interval long enough to cover a deploy.
My listener is a background service that connects over TLS with a fixed client id and a 30-second keep-alive, and subscribes at QoS 1 to ChirpStack's uplink events and to a devices/+/telemetry topic for direct devices. Its reconnect loop is deliberately boring: check every five seconds, connect if needed, subscribe again. Resubscribing costs nothing and picks up a changed topic filter even when the broker restored an old session.
For LoRaWAN, QoS has to be right in a second place. ChirpStack's MQTT integration publishes uplinks with its own QoS setting, and a subscriber gets the lower of the publish and subscribe levels. If ChirpStack publishes at QoS 0, your persistent session protects nothing. Check qos under the MQTT integration in the ChirpStack configuration.
The handler does almost nothing
MQTTnet awaits the handler before moving to the next message, so anything slow in it slows the whole connection. Mine parses, normalises and hands off to a channel. No database, no HTTP calls, no rule evaluation.
The setting that matters is AutoAcknowledge, which I turn off. By default MQTTnet sends the PUBACK when the handler returns, telling the broker to forget a message that is still only in memory. Acknowledging from the writer after the batch commits moves the ack to the point where the reading is durable. If the process dies with a full buffer, nothing in it was acknowledged, and the broker redelivers it all on reconnect.
That has a cost. The broker limits how many unacknowledged QoS 1 messages it sends one session at a time: Receive Maximum in MQTT 5, a per-broker in-flight setting otherwise. With ack-after-commit that window caps throughput. A window of 32 and a 500 ms batch wait gives around 64 messages a second from that session, however fast the database. Raise the window well above your batch size, or run several listeners on a shared subscription, each with its own window.
An unparseable message is acknowledged at once and counted, because redelivery will never fix it. Acks go out in arrival order, since the writer reads the channel in order, which is what MQTT expects of a QoS 1 receiver.
One envelope for every source
ChirpStack uplinks and direct MQTT payloads look nothing alike. The normaliser turns both into one envelope, so nothing after it branches on where a message came from. It also chooses the idempotency key, because only here do we know enough about the source to pick a good one.
ChirpStack v4 gives one deduplicationId per uplink however many gateways heard it. An uplink with no decoded object means the device profile has no codec, so it is rejected. For direct MQTT I prefer a device-supplied msgId and fall back to a SHA-256 hash of the exact payload bytes. Unparseable messages and unknown topics are rejected too.
The hash fallback has a known weakness: two separate readings with identical bytes collapse into one. It is safe only if the device includes its own timestamp, another reason to ask device makers for one.
The envelope keeps the payload as sent. Mapping raw keys onto named variables happens downstream, so a mapping mistake can be fixed and replayed from what was stored rather than lost at the edge.
Bounded, and why Wait is the right policy
The buffer is a Channel with a fixed capacity of 5,000 and a single reader. An unbounded channel just moves the outage from the database to the process's memory. The capacity sits above the broker's total in-flight window, so in normal running the handler never waits. If it ever does, Wait means we stop reading and the broker keeps the rest.
DropOldest and DropWrite are tempting because the handler stays fast no matter what. For telemetry they are silent data loss with extra steps. Nothing in the channel is acknowledged, so waiting is safe: the broker is a far better place for a backlog than my process.
Batching writes
The writer drains the channel into batches of at most 1,000 messages or 500 ms, whichever comes first, so quiet periods flush quickly and busy ones make fewer, larger round trips.
Each batch goes to the raw telemetry table through SqlBulkCopy in one transaction. A failed write is retried with backoff of up to 30 seconds; nothing is acked yet, so the channel fills and then the broker holds the rest. After commit the writer acks each message in arrival order. If an ack fails because the delivering connection is gone, the broker redelivers and the unique index drops the copy.
The writer stores the raw envelope and nothing else. Mapping into variables, pushing live values over SignalR and running the event engine all happen after commit, off the ingestion path, so a slow rule never slows a write.
At-least-once means duplicates are normal
A crash between commit and ack causes a redelivery of data you already stored. So does every reconnect with unacknowledged messages, and a replayed LoRaWAN uplink. Duplicates are not an edge case; they are the price of not losing anything. So the database, not the code, decides what is new.
The raw table's unique index on source, device key and message key uses IGNORE_DUP_KEY, which turns a duplicate into a skipped row with a warning instead of an error that fails the whole batch.
The key includes source and device because message ids are only unique within a device. A msgId longer than the 64-character column is rejected in the normaliser, not truncated, because a truncated key is a collision waiting to happen.
What I measure
- ›Buffer depth. Near zero in steady state. A rising line means the writer is behind, long before any loss.
- ›Batch size. Always tiny means the time window does all the work; always full means you are at the ceiling, so look at the in-flight window or the database.
- ›Commit lag. Time from receipt to commit for the oldest message in each batch. This is the number to alert on.
- ›Device-to-receipt lag. Receive time minus device time. For LoRaWAN it exposes network server and gateway backhaul delays that look like ingestion problems.
- ›Connects and rejects. More than one connect per deploy means the client is flapping. A jump in rejects usually means a firmware or codec change.
Where I'd take this next
The Flux spec takes the same rule further. Gateways publish each envelope to RabbitMQ quorum queues and acknowledge the device only after the broker confirms, so the in-memory channel disappears. A separate writer batches into PostgreSQL with TimescaleDB and acknowledges the queue only after commit, with ON CONFLICT DO NOTHING as the dedupe. ChirpStack gets its own bridge with a frame counter replay guard and a bucket for devices the network server knows but the platform doesn't. The design targets tens of thousands of readings per second on one node; targets to prove with its load harness, not measurements.
What I'd do differently
- ›Count duplicates from the start.
IGNORE_DUP_KEYmakes duplicates vanish quietly, hiding how many there are. Staging into a temp table and inserting with aNOT EXISTScheck gives a row count you can graph, and a duplicate spike is often the first sign of a reconnect loop. - ›Size the in-flight window before the batch. I tuned batch size first and throughput didn't move. The broker's window was the ceiling all along.
- ›Require a device timestamp and message id in the direct MQTT contract. The hash fallback works, but it is a guess about identity. A short published payload contract would have removed it.
- ›Write a kill test early. Stopping the service mid-load and counting rows against what was published is the only convincing proof of the rule above, and it is cheap to automate.