What if your device fleet ran on one log?

A factory floor is covered in sensors. Each one publishes a reading every few seconds over MQTT, because that's what its firmware speaks. Somewhere a model reads those readings and makes decisions: slow down line 3, close a valve, flag a bearing that sounds wrong. The decisions go back to the devices, over MQTT again.

The usual way to build this takes three systems. An MQTT broker such as Mosquitto holds the device connections. A bridge copies every message into Kafka, because the training pipeline wants months of history and the MQTT broker isn't built to keep it. Another job writes each device's latest state into Valkey, so a dashboard or the model can ask what device 42 is doing right now.

Each of those systems is good at its job. The trouble is in the bridges. Each one has its own idea of order, retries and failure. When the bridge into Kafka restarts, did that reading get copied once, twice, or not at all? When Valkey says the valve is open and the history in Kafka says it closed, which one do you believe?

Felix doesn't speak MQTT. This post is a thought experiment: what if it did? Most of what a fleet needs is already in Felix, and what's missing is the front door. There's a design for that front door. Everything below about MQTT is how it could work, not something you can run.

What's already there

A log you can read again

A durable stream in Felix is a sharded, ordered log. Every record gets an offset. A reader can start at the oldest record, at an offset, or at a time, since offset_for_time finds the first record written at or after a given moment. A publish is acknowledged once it's in the log under the broker's fsync policy, and on a Quorum stream once a majority of replicas hold it.

For a model, that log is the training and evaluation record. A training set is a range of offsets. If a new model does something odd, you can run it over exactly the records the last one saw and compare the two.

The Kafka API

A broker started with FELIX_KAFKA_LISTEN also serves Kafka clients. A consumer can list topics, look up offsets by time and fetch records from durable streams, and a producer can write to them. There are no consumer groups, so a consumer assigns its own partitions and keeps its own offsets, which suits a job that reads a fixed range of history. Tools that already read Kafka can read a Felix stream with no bridge in between. Point your Kafka client at Felix has the details.

State first, then every change

A log-backed cache in Felix can be watched, by key or by key prefix. A watch picks up from an offset. If the history it asks for is gone, the broker answers with the current value of every matching key and then the live changes. The watch is registered before anything is read, so no write can slip in between the snapshot and the changes.

That's MQTT's retained message: a new subscriber gets the last value on a topic, then every update after it. It's also what fleet platforms call a device shadow, the last known state of each device, readable by anything that wants to know what the fleet looks like right now.

One slow reader is its own problem

Each subscriber has its own bounded queue on the broker. When the queue fills, the default policy drops new records for that subscriber alone and moves on. The publisher never waits for a reader, and the other readers don't notice.

A drop is only acceptable if you can see it. On a durable stream every delivered record carries its offset, so a gap in offsets is a drop. A client that asks for it is also told straight away: the broker ends the subscription at its first drop and says which offset to resume from. The reader catches up from the log. One slow subscriber shouldn't slow anyone else goes through how that works.

A disk that doesn't fill

An edge site has a disk, and the disk has an end. Retention bounds a stream by size or by age. When the thing that knows what's safe to delete is a program, like one that ships history to a central cluster, a head trim lets it delete everything below an offset it has finished with. The trim is conditional. It applies only if the stream's tail is where the caller expects, and if a publish has moved the tail, the trim is refused with the new tail and the caller tries again. A reader that asks for a trimmed offset is told it's too old rather than handed a short read. The trim that hid the fence is about why the condition is there.

What MQTT would add

The design adds an MQTT listener to every broker, next to the Kafka one and off unless an operator turns it on. A device would connect to it, authenticate with a Felix token, and publish and subscribe as it does today.

Topics mounted on streams

MQTT topics spring into existence when a device publishes to one. Felix streams are created by an operator. A mount would join the two: an operator mounts a durable stream at a topic prefix such as plant/, and every topic under that prefix becomes a record in that stream. Devices keep their own topic scheme, and nobody creates plant/7/line/3/temp by hand.

Each record would carry the full topic as its key and MQTT's message properties as headers, and keep the payload exactly as the device sent it:

MQTT	Felix record
Topic	Key, also used to pick the shard
Message properties	Headers
Payload	Payload, untouched

Because the topic picks the shard, every message on one topic would land in one ordered log, which is how MQTT's per-topic ordering would hold. And because the payload is untouched, a Kafka consumer or a Felix client reading the mounted stream would see exactly what the device sent.

Acknowledged when Felix would acknowledge

A QoS 1 PUBACK would be sent at the point Felix would acknowledge the write for that stream, never earlier. Each MQTT session would get a Felix idempotent producer. If a broker died mid-write, the listener would send the same batch to the shard's new owner, and the log would answer a duplicate with the offset it was first written at instead of writing it again. That's the bridge's question, once, twice or not at all, answered by the log.

A sensor whose reading is stale a second later could publish at QoS 0 and skip the wait altogether.

Retained messages in the cache

Each mount would name a cache for its retained messages, keyed by topic. A retained publish would write the record to the stream and then the entry to the cache, with a conditional put that only replaces an entry from an older offset. Two devices racing to retain on one topic would end up agreeing with the stream's order. A new subscription would register first and read the retained values second: the same state-then-changes contract as a cache watch.

A device that goes to sleep

Isolation would carry over to devices. Each MQTT session would be fed through its own bounded queue. A QoS 1 session whose queue filled would leave the shared feed and read the log from its own position at its own pace. A QoS 0 subscription would drop, as a slow Felix subscriber does.

A persistent session would keep a position per shard instead of a queue of copies. While a device is offline nothing piles up in memory for it, since its messages are already in the log. When it reconnects it reads on from its bookmark. How far back it can catch up depends on how much history the site keeps. If retention or a trim has passed a session's bookmark, the session would resume at the oldest record left and the broker would count the gap, because neither MQTT version has a way to tell the device.

Closing the loop with a model

The model is the part of this loop that is hardest to explain afterwards. When a valve closes at 02:14, someone will ask what the device saw, which model was running, and what the model said.

Running the model again doesn't answer that. Inference on a GPU usually isn't bit-for-bit reproducible. Kernels add numbers in parallel, in an order that changes from run to run, and floating point addition depends on the order. Libraries pick kernels by hardware, driver and batch size, and sampling draws random numbers. The same reading can come back with a different answer when two scores are close.

So the idea is to treat inference as an effect, like a call to any outside service: ask once, record the answer, and never compute it again. The loop would go like this:

1. A device publishes a reading over MQTT, and it lands on the telemetry stream at an offset.
2. A program reading that stream asks the model, with a request record that names the reading and the model version.
3. The model's response is written to the log: which model, with which parameters, and what it said.
4. The program turns the response into a decision and publishes it to a stream mounted at cmd/. A device subscribed to cmd/42/# would get the decisions meant for it.

Felix's part is what makes this cheap. Readings, responses and decisions would sit in one ordered, durable log, each with an offset and an append time, readable again by offset or by time. Recording the answer is one more publish.

Because the response is on the log, any decision could be replayed later without calling the model: read the readings and the recorded responses back, and run the same steps over them. A new model could be shadowed over the same recorded history before it takes over. Send it the recorded requests, write its answers to a separate stream, and compare its decisions with the ones the fleet actually made.

What if every machine decision could be replayed? on the NineLives blog explores this in depth.

What it isn't

Only durable streams could be mounted. Bookmarks, persistent sessions and re-sent writes all depend on the log, so a mount without one wouldn't keep MQTT's promises.

A log also costs something. A QoS 1 publish would wait for the write under the stream's fsync policy, like any acknowledged Felix publish. For a fleet that only ever needs the latest reading and never looks back, a plain MQTT broker is simpler.

None of these pieces is new. MQTT brokers, Kafka and key-value stores have run fleets for years. The question is whether one log, with a cache projected from it and a reader per subscriber, can do the work of all three, and get rid of the bridges while it's at it.

Join in

The design is in the Felix repository, and the discussion is on #1685.

docker compose up -d