MQTT and AMQP in OxiDB: a PUBACK means on disk

OxiDB ships two message brokers: an MQTT 3.1.1 broker and an AMQP 0-9-1 broker (the protocol RabbitMQ speaks). Both live in the same binary, in the same process, and both get their durability from the same place: the document engine's fsync'd write path. This post walks through the whole layer: why I put a broker inside a database, how I made the MQTT broker actually keep its "QoS 1" promise, why I added AMQP on top, how an MQTT sensor's readings end up in an AMQP worker's queue, and which data structures make all of it work. Every example here was run on my own machine (Apple M4, macOS, local) against a real oxidb-server 0.44.2 process, and nothing on the client side is OxiDB-specific: mosquitto_pub/mosquitto_sub, paho-mqtt 2.1.0 and pika 1.4.4, stock and unmodified.

Why is there a broker inside a database?

The MQTT broker started life for an IoT scenario. In the ColdChain demo, temperature probes send readings over MQTT and those readings need to be stored and queried. The classic setup needs three moving parts for that: a broker (Mosquitto, EMQX), a bridge service, and a database. I put the broker at the edge of the database instead: a reading can be written to the document engine the moment it arrives, fed into the time-series engine, and seen by OxiMem (the Redis-compatible layer) subscribers too.

At some point the obvious question came up: since there is a broker anyway, should OxiDB grow a separate Kafka- or RabbitMQ-style messaging engine? The answer is written down in ADR-0015 (one of the project's architecture decision records), and it is no. Kafka's value is per-partition leadership with ISR replication. Putting that on top of OxiDB's single-leader Raft cluster means either routing every message through one Raft leader (throwing away the partition parallelism that is the whole point) or fighting the consensus model. It would have been fake Kafka or a cluster liability. So I decided to make the broker I already had honest, and later added AMQP on top of it (ADR-0016).

The core idea of this post: in OxiDB, the moment a message is acknowledged (PUBACK in MQTT, a publisher confirm in AMQP), it is on disk, fsync'd. The broker has no new file format. Sessions, queued messages and retained values are ordinary documents in the _mqtt and _amqp collections.

Turning it on: a handful of environment variables

Every protocol listener is off by default, and costs nothing while off. For the experiments in this post I started the server like this (ports from my assigned range):

OXIDB_ADDR=127.0.0.1:15800 \
OXIDB_DATA=$PWD/data \
OXIDB_MQTT_PORT=15804 \
OXIDB_AMQP_PORT=15805 \
OXIDB_MQTT_PERSIST=1 \
OXIDB_MQTT_INGEST="sensors/#:sensor_readings,fleet/+/pos:pings" \
oxidb-server

The startup log says which parts are active (the server uses an em dash in these lines; I replaced it with a hyphen here, everything else is verbatim):

MQTT listening on 0.0.0.0:15804
MQTT persistence on - recovered 0 sessions, 0 pending messages
MQTT ingest bridge on - 2 route(s)
AMQP listening on 0.0.0.0:15805 - recovered 0 durable queues, 0 messages

One thing worth noticing: the MQTT and AMQP listeners bind 0.0.0.0, so the 127.0.0.1 in OXIDB_ADDR does not cover them. In production, turn on authentication or firewall the ports.

MQTT 3.1.1: the ordinary part first

The basics any broker should do: topics, wildcard subscriptions and retained messages. I publish two room temperatures as retained, subscribe with home/+/temp, then publish two more messages:

mosquitto_pub -p 15804 -t 'home/livingroom/temp' -m '22.4' -r -q 1
mosquitto_pub -p 15804 -t 'home/kitchen/temp'    -m '24.1' -r -q 1
mosquitto_sub -p 15804 -t 'home/+/temp' -v -q 1 &
mosquitto_pub -p 15804 -t 'home/bedroom/temp'     -m '19.8' -q 1
mosquitto_pub -p 15804 -t 'home/bedroom/humidity' -m '55'   -q 1

home/livingroom/temp 22.4
home/kitchen/temp 24.1
home/bedroom/temp 19.8

The first two lines are the retained values delivered the moment the subscriber connects; the third is live. home/bedroom/humidity does not match the filter. + matches exactly one level, a trailing # matches the rest (including nothing).

Every broker does this. What makes OxiDB's different is what QoS means.

Honest QoS: the story of ADR-0015

MQTT has three delivery levels. QoS 0 is "at most once": fire and forget. QoS 1 is "at least once": the broker must keep a message until the subscriber PUBACKs it, retransmit it with the DUP flag if the connection breaks, and, for a client that connected with clean_session=false, keep the messages that arrive while it is offline and replay them when it comes back. QoS 2 is "exactly once": a two-phase PUBREC/PUBREL/PUBCOMP handshake plus deduplication by packet id.

OxiDB's first MQTT broker did not keep that promise, but it behaved as if it did. When I audited my own code while writing ADR-0015, this is what I found: the delivery loop set the QoS 1 flag and allocated a packet id, and the website said "QoS 1 delivery with packet ids". But:

• a delivered QoS 1 message was not stored anywhere while waiting for its PUBACK, so it could never be redelivered;
• a PUBACK from a subscriber was read and thrown away, because there was no record for it to acknowledge;
• a publish only reached subscribers with a live channel at that instant, so a subscriber that blinked offline lost the message entirely;
• clean_session=false did nothing; CONNACK always said session_present=0.

The wire said at-least-once, the behaviour was at-most-once. I classified it as a correctness bug: the system claimed a guarantee it did not keep. First I corrected the docs to match reality, then I raised the behaviour to the docs' old claim. The rule is simple: a broker that under-promises and over-delivers is acceptable; the reverse is the bug itself.

Persistent sessions: the receiver belongs to the session, not the socket

The fix hinges on one small design pivot. In the old broker, SUBSCRIBE handed the connection an mpsc::Receiver; when the socket closed, the receiver dropped and anything sent to the channel evaporated. In the new design the receiver is owned by the session. An mpsc channel keeps buffering while its receiver is alive with nobody reading, so a publish to an offline session keeps landing in the channel. While a connection is attached, the connection drains it; while detached, one broker-wide reaper thread (every 500 ms) drains it into the session's bounded queue.

A session is a struct keyed by client id: the subscription list, outbound messages awaiting PUBACK (in a BTreeMap by packet id, so redelivery is ordered), the queue of messages that arrived while offline, two sets for QoS 2, and a generation counter. When a second CONNECT arrives with the same client id (MQTT-3.1.4-2: the newcomer wins), the generation is bumped; the older connection that lost the takeover race sees it move and exits instead of double-delivering.

The offline queue is bounded: MAX_QUEUED = 4096 per session. When it is full, the oldest message is dropped and a counter (dropped) goes up. For telemetry the freshest reading is the one worth keeping, and QoS 1 promises "at least once", not "never lost under unbounded backlog against a dead consumer". A slow consumer must not be able to run the broker out of memory.

OXIDB_MQTT_PERSIST: testing it with SIGKILL

In-memory sessions protect against client disconnects, but a server restart wipes them. With OXIDB_MQTT_PERSIST=1, sessions, queued messages and retained values are written through the document engine's WAL. I took the ADR's own rule seriously here: "a test that passes without a SIGKILL'd subprocess proves nothing". So I ran this one with kill -9.

The scenario: a cold-room worker subscribes to plant/+/alarm at QoS 1 with a persistent session, then disconnects. A probe publishes three alarms and waits for a PUBACK on each. Then I kill the server, restart it, and reconnect the worker. The Python side uses paho-mqtt:

import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion

def client(cid, clean):
    return mqtt.Client(CallbackAPIVersion.VERSION2, client_id=cid, clean_session=clean)

# 1) subscribe, then drop
c = client("cold-room-worker", False)
def on_connect(c, u, flags, rc, props):
    print(f"CONNACK rc={rc} session_present={flags.session_present}")
    c.subscribe("plant/+/alarm", qos=1)
def on_subscribe(c, u, mid, rcs, props):
    print(f"SUBACK granted={[r.value for r in rcs]}")
    c.disconnect()

# 2) publisher: wait for the PUBACK of every message
p = client("probe-17", True)
info = p.publish("plant/room1/alarm", '{"seq": 1, "temp_c": -12.5}', qos=1)
info.wait_for_publish()

Subscribe-and-drop, then three publishes:

CONNACK rc=Success session_present=False
SUBACK granted=[1]
PUBACK mid=1 topic=plant/room1/alarm
PUBACK mid=2 topic=plant/room2/alarm
PUBACK mid=3 topic=plant/room1/alarm
PUBACK mid=4 topic=plant/status (retained)

At this point I looked at the _mqtt collection with OxiDB's own Python client over the native wire protocol. The broker's entire state is right there, as plain documents:

{"_kind": "retain", "topic": "home/livingroom/temp", "payload": "22.4", "_id": 1, "_version": 1}
{"_kind": "retain", "topic": "home/kitchen/temp", "payload": "24.1", "_id": 2, "_version": 1}
{"_kind": "sub", "client": "cold-room-worker", "filter": "plant/+/alarm", "qos": 1, "_id": 3, "_version": 1}
{"_kind": "msg", "client": "cold-room-worker", "seq": 1, "topic": "plant/room1/alarm", "payload": "{\"seq\": 1, \"temp_c\": -12.5}", "qos": 1, "_id": 4, "_version": 1}
{"_kind": "msg", "client": "cold-room-worker", "seq": 2, "topic": "plant/room2/alarm", "payload": "{\"seq\": 2, \"temp_c\": -3.1}", "qos": 1, "_id": 5, "_version": 1}
{"_kind": "msg", "client": "cold-room-worker", "seq": 3, "topic": "plant/room1/alarm", "payload": "{\"seq\": 3, \"temp_c\": 4.8}", "qos": 1, "_id": 6, "_version": 1}
{"_kind": "retain", "topic": "plant/status", "payload": "online", "_id": 7, "_version": 1}

Now kill -9, a restart, and the worker coming back. The startup log:

MQTT persistence on - recovered 1 sessions, 3 pending messages

The worker reconnects with the same client id and clean_session=false; separately, a clean-session "dashboard" client subscribes to plant/status:

CONNACK rc=Success session_present=True
PUBLISH topic=plant/room1/alarm qos=1 dup=0 payload={"seq": 1, "temp_c": -12.5}
PUBLISH topic=plant/room2/alarm qos=1 dup=0 payload={"seq": 2, "temp_c": -3.1}
PUBLISH topic=plant/room1/alarm qos=1 dup=0 payload={"seq": 3, "temp_c": 4.8}
PUBLISH topic=plant/status retain=1 payload=online

session_present=True finally means something: the subscription came back from a killed process. The three alarms were delivered in seq order, and the retained online value survived the restart too. As the worker PUBACKs each message, its _kind: "msg" record is deleted; after the run the same query came back empty.

There is a small but important detail in that delete: on PUBACK the record is deleted by (client, seq), not by content. Deleting by content collapses duplicates: publish the same payload twice, ack once, and a content match deletes both records, quietly breaking at-least-once for the second copy. That is why every in-memory inflight message carries the sequence number of its on-disk record.

Recovery has its own bound too. A crash window can leave more than MAX_QUEUED records on disk; at startup the newest ones are loaded and the excess is deleted from disk as well, otherwise a restart would resurrect messages the bound had already discarded.

QoS 2: exactly once

QoS 2 works in both directions. The -d flag of the mosquitto clients shows the packet flow. A meter publishes to meter/42/kwh at QoS 2, and a billing client is subscribed to meter/+/kwh at QoS 2:

Client meter-42 sending CONNECT
Client meter-42 received CONNACK (0)
Client meter-42 sending PUBLISH (d0, q2, r0, m1, 'meter/42/kwh', ... (6 bytes))
Client meter-42 received PUBREC (Mid: 1)
Client meter-42 sending PUBREL (m1)
Client meter-42 received PUBCOMP (Mid: 1, RC:0)
Client meter-42 sending DISCONNECT

Client billing sending SUBSCRIBE (Mid: 1, Topic: meter/+/kwh, QoS: 2, Options: 0x00)
Client billing received SUBACK
Subscribed (mid: 1): 2
Client billing received PUBLISH (d0, q2, r0, m1, 'meter/42/kwh', ... (6 bytes))
Client billing sending PUBREC (m1, rc0)
Client billing received PUBREL (Mid: 1)
Client billing sending PUBCOMP (m1)
1532.7

Two sets carry this. On the inbound side, rx2: packet ids of publishes I have already fanned out but whose PUBREL I have not seen yet. If the publisher retransmits with the same id (DUP), the id is in the set, so I only repeat the PUBREC and do not fan the message out again; that set is the entire difference between "exactly once" and "at least once with a fancier handshake". On the outbound side, pubrel_pending: ids I have received a PUBREC for and owe a PUBCOMP on. The message record is deleted at PUBREC (the receiver owns the message from that point), but the PUBREL obligation has to survive, or the subscriber's state machine wedges on a forever-pending id. With persistence on, both sets go to disk as _kind: "rx2" and _kind: "rel" documents; exactly-once has to survive a crash too.

No ack before the write. What does it cost?

The order of the publish path is spelled out in mqtt.rs: fan out to the in-memory bus, then the fsync'd insert for persistent sessions, then hand it to the AMQP bridge and the ingest bridge, and only then send the PUBACK (PUBREC for QoS 2). So a PUBACK does not mean "I received it"; it means "it is on disk for every durable subscriber that will need it".

That has to cost something, so I measured it. Same machine (Apple M4, data directory on the internal SSD), 500 sequential QoS 1 publishes with paho, waiting for each PUBACK. In the first run no persistent subscriber matched the topic; in the second one did (offline):

bench/nobody: 500 sequential QoS1 publishes, p50=0.059 ms p99=0.185 ms total=0.03 s
bench/durable: 500 sequential QoS1 publishes, p50=4.164 ms p99=9.006 ms total=2.25 s

Roughly 70x, and all of it is fsync. On macOS OxiDB uses a full flush to stable storage (F_FULLFSYNC), which costs a few milliseconds. If no persistent subscriber matches, there is no fsync at all; if the server has no persistent sessions whatsoever, the whole path is skipped with one atomic load (the has_durable flag). Only those who need it pay for it. These are worst-case numbers, sequential on one connection; with several publishers the picture changes, and the group commit I describe in the AMQP section exists exactly for that.

OXIDB_MQTT_INGEST: from a topic to a queryable document

The ingest bridge is the most practical piece I wrote for people using MQTT as a data intake. OXIDB_MQTT_INGEST is a comma-separated list of topic-filter:collection routes. Every matching publish becomes a document in that collection:

• if the payload is a JSON object, the object itself becomes the document (so {"value": ..} is queryable as value, not payload.value);
• anything else is wrapped as {"payload": ...};
• either way the document gains _topic (the concrete topic it was published to; with a wildcard route that is how rows are told apart) and _received_at (epoch milliseconds).

The parser splits each entry on its last :, because an MQTT topic level may contain one itself. A malformed route refuses to start the server: a filter with # in the middle (a/#/b) matches nothing per the MQTT spec, so it is a typo, and I did not want it to become a bridge that silently never fires.

Five publishes, one of them unmapped:

mosquitto_pub -p 15804 -q 1 -t sensors/greenhouse/temp     -m '{"value": 27.3, "unit": "C"}'
mosquitto_pub -p 15804 -q 1 -t sensors/greenhouse/humidity -m '{"value": 71, "unit": "%"}'
mosquitto_pub -p 15804 -q 1 -t sensors/garage/temp         -m '18.9'
mosquitto_pub -p 15804 -q 1 -t fleet/u42/pos -m '{"lat": 41.0151, "lon": 28.9795, "speed": 34}'
mosquitto_pub -p 15804 -q 1 -t chat/general  -m '{"msg": "not mapped"}'

Then a look at the collections with the native client:

with OxiDbClient(port=15800) as db:
    print("collections:", sorted(...db.list_collections()))
    for d in db.find("sensor_readings", {}):
        print(json.dumps(d, ensure_ascii=False))
    for d in db.find("sensor_readings", {"value": {"$gte": 25}}):
        print(d["_topic"], d["value"], d["unit"])
    for d in db.find("pings", {}):
        print(json.dumps(d))

collections: ['_alerts', '_amqp', '_mqtt', '_schedules', 'pings', 'sensor_readings']
{"value": 27.3, "unit": "C", "_topic": "sensors/greenhouse/temp", "_received_at": 1791754335817, "_id": 1, "_version": 1}
{"value": 71, "unit": "%", "_topic": "sensors/greenhouse/humidity", "_received_at": 1791754335830, "_id": 2, "_version": 1}
{"payload": 18.9, "_topic": "sensors/garage/temp", "_received_at": 1791754335839, "_id": 3, "_version": 1}
--- find value >= 25
sensors/greenhouse/temp 27.3 C
sensors/greenhouse/humidity 71 %
--- pings
{"lat": 41.0151, "lon": 28.9795, "speed": 34, "_topic": "fleet/u42/pos", "_received_at": 1791754335848, "_id": 1, "_version": 1}

chat/general went nowhere. A bare number like 18.9 was wrapped under payload. Everything else is now an ordinary document: you can index it, aggregate it and use the whole query language of the document engine. The pings collection, with its coordinates, is ready for geospatial queries.

The insert happens before the broker's QoS ack, so an acknowledged publish is durably in the collection too. When the bridge is not configured, its cost per publish is one atomic load (OnceLock::get).

AMQP 0-9-1: what MQTT cannot express

In ADR-0015 I had deferred, not rejected, a RabbitMQ-style durable queue: "once persistent sessions, an inflight store and WAL-backed redelivery exist, a durable-queue surface is an increment on top, not a separate engine". Then the question came back in protocol form: can existing RabbitMQ clients talk to OxiDB?

AMQP is not an MQTT extension; it is a second wire protocol with one genuinely new semantic: competing consumers. An MQTT subscription copies every message to every matching subscriber. An AMQP queue gives each message to exactly one of its consumers. That is work distribution, the reason RabbitMQ sits between producers and worker pools. OxiDB had pub/sub (MQTT, OxiMem) but no work queue.

I kept the scope deliberately narrow and drew the line honestly. Supported: the connection handshake (PLAIN), channels, heartbeats, the default exchange ("", routes by queue name), direct, fanout and topic exchanges, Queue.Declare (durable, exclusive, auto-delete, server-named, passive), Queue.Bind, multi-frame bodies, Basic.Consume/Deliver/Get/Cancel, Basic.Ack (including multiple), Basic.Nack/Reject, Basic.Qos prefetch, the mandatory flag with Basic.Return, and publisher confirms. Out of scope: the tx class, headers exchanges, exchange-to-exchange bindings, per-message TTL and priority, dead-letter arguments, alternate exchanges, AMQP 1.0 (a different protocol entirely), streams, and any management UI.

Exchanges: direct, fanout, topic

All of them in one pika script:

import pika
conn = pika.BlockingConnection(pika.ConnectionParameters("127.0.0.1", 15805))
ch = conn.channel()

ch.exchange_declare("orders", exchange_type="direct")
for q, key in [("q.paid", "paid"), ("q.refund", "refunded")]:
    ch.queue_declare(q); ch.queue_bind(q, "orders", routing_key=key)
for key, body in [("paid", "order-1001"), ("refunded", "order-0999"),
                  ("paid", "order-1002"), ("cancelled", "order-1003")]:
    ch.basic_publish("orders", key, body)

ch.exchange_declare("cache-bust", exchange_type="fanout")
for q in ("node-a", "node-b", "node-c"):
    ch.queue_declare(q); ch.queue_bind(q, "cache-bust")
ch.basic_publish("cache-bust", "ignored", "invalidate:/products/42")

ch.exchange_declare("logs", exchange_type="topic")
binds = {"all-errors": "*.error", "eu-everything": "eu.#", "eu-west-db": "eu.west.db.*"}
for q, pat in binds.items():
    ch.queue_declare(q); ch.queue_bind(q, "logs", routing_key=pat)
for key in ["api.error", "eu.west.db.error", "eu.west.db.slow", "eu", "us.east.api.info"]:
    ch.basic_publish("logs", key, key)

Draining the queues with basic_get:

server product: OxiDB

== direct exchange
q.paid   -> ['order-1001', 'order-1002']
q.refund -> ['order-0999']

== fanout exchange
node-a -> ['invalidate:/products/42']
node-b -> ['invalidate:/products/42']
node-c -> ['invalidate:/products/42']

== topic exchange
all-errors     *.error        -> ['api.error']
eu-everything  eu.#           -> ['eu.west.db.error', 'eu.west.db.slow', 'eu']
eu-west-db     eu.west.db.*   -> ['eu.west.db.error', 'eu.west.db.slow']

The cancelled order matched no binding and was dropped (normal AMQP behaviour without mandatory). In the topic exchange, * matches exactly one word and # zero or more: eu.# also caught the bare eu key, while *.error did not take the four-word eu.west.db.error.

Confirms, mandatory, nack, and an honest refusal

ch.confirm_delivery()
ch.queue_declare("jobs")
ch.basic_publish("", "jobs", "resize:img-1.png")
try:
    ch.basic_publish("", "no-such-queue", "lost?", mandatory=True)
except pika.exceptions.UnroutableError as e:
    ...
m, _, body = ch.basic_get("jobs");  ch.basic_nack(m.delivery_tag, requeue=True)
m, _, body = ch.basic_get("jobs");  ch.basic_ack(m.delivery_tag)
ch.exchange_declare("hdr", exchange_type="headers")

== publisher confirms + mandatory
confirmed: resize:img-1.png
UnroutableError: ['NO_ROUTE'] [312]

== nack with requeue
got resize:img-1.png redelivered = False -> basic_nack(requeue=True)
got resize:img-1.png redelivered = True -> basic_ack

== unsupported exchange type
ChannelClosedByBroker: 540 exchange type 'headers' is outside ADR-0016's scope; supported: direct, fanout, topic

The last line is my favourite part of the design. A client that uses something I do not implement gets a channel error naming the ADR (540 NOT_IMPLEMENTED), not silent acceptance. A headers exchange that is quietly accepted and never routes anything is the kind of problem that takes days to debug.

A detail about mandatory: the Basic.Return is sent before the confirm. pika collects the return, then sees the ack and raises UnroutableError, which is what RabbitMQ does too. The .NET client taught me that a Return with no ack behind it must not sit in the buffer until the next heartbeat, so every batch now ends in exactly one flush.

Competing consumers and prefetch

This is the feature AMQP was added for. Two workers consume the same queue: a fast one (50 ms per job) and a slow one (500 ms). I put 12 jobs on the queue, first with no prefetch, then with basic_qos(prefetch_count=1):

def worker(name, cost):
    conn = pika.BlockingConnection(P); ch = conn.channel()
    ch.queue_declare(Q, durable=True)
    if prefetch: ch.basic_qos(prefetch_count=prefetch)
    def on_msg(ch, m, props, body):
        time.sleep(cost)
        log.append((name, body.decode()))
        ch.basic_ack(m.delivery_tag)
    ch.basic_consume(Q, on_msg)
    ...

prefetch=unlimited fast:  6 jobs ['job-02', 'job-04', 'job-06', 'job-08', 'job-10', 'job-12']
prefetch=unlimited slow:  6 jobs ['job-01', 'job-03', 'job-05', 'job-07', 'job-09', 'job-11']
prefetch=1 fast: 10 jobs ['job-01', 'job-03', 'job-04', 'job-05', 'job-06', 'job-07', 'job-08', 'job-09', 'job-10', 'job-12']
prefetch=1 slow:  2 jobs ['job-02', 'job-11']

Without prefetch it is pure round-robin: each job goes to the next consumer in turn, and the fast worker sits idle while the slow one spends three seconds on its six. With prefetch=1, each worker may hold at most one unacknowledged message; a consumer at its limit has its turn passed to one with capacity. The fast worker got 10 jobs, the slow one 2. That is the whole point of prefetch: a busy worker must not stall the queue behind it.

Unacknowledged messages go back to the front of the queue, flagged redelivered=true, when their channel or connection dies. Their on-disk records were never deleted, so a crash in that window changes nothing.

Durability follows the protocol: a durable queue and kill -9

MQTT needed OXIDB_MQTT_PERSIST because MQTT 3.1.1 has no wire-level concept of storage. AMQP does: the client declares a queue durable and publishes with delivery_mode=2 (persistent). It states its intent in the protocol. So there is no OXIDB_AMQP_PERSIST: a durable queue with persistent messages is durable, period. A transient queue lives in memory, because the client asked for that by not saying durable.

The test: three persistent invoices into the durable invoices queue, plus one delivery_mode=1 draft, plus one message into the transient scratch queue, all with publisher confirms. Then the _amqp collection:

{"_kind": "queue", "name": "invoices", "_id": 31, "_version": 1}
{"_kind": "qmsg", "queue": "invoices", "seq": 25, "exchange": "", "rkey": "invoices", "props": "kAAKdGV4dC9wbGFpbgI=", "body": "aW52b2ljZS0x", "_id": 32, "_version": 1}
{"_kind": "qmsg", "queue": "invoices", "seq": 26, "exchange": "", "rkey": "invoices", "props": "kAAKdGV4dC9wbGFpbgI=", "body": "aW52b2ljZS0y", "_id": 33, "_version": 1}
{"_kind": "qmsg", "queue": "invoices", "seq": 27, "exchange": "", "rkey": "invoices", "props": "kAAKdGV4dC9wbGFpbgI=", "body": "aW52b2ljZS0z", "_id": 34, "_version": 1}

(The collection also held queue records from earlier experiments; I kept only the invoices lines here.) The draft and the scratch message are not on disk: one is not persistent, the other's queue is not durable. Body and properties are stored base64-encoded. props is the content header's property section, verbatim from the wire, so the broker returns content_type, correlation ids and application headers on delivery without modelling any of them.

Then kill -9 and a restart:

The persistent messages came back with content_type=text/plain intact, and the transient queue is gone (404). Every recovered message is redelivered=True. That is deliberately conservative: "this message may have been delivered before the crash" is the promise the flag exists to keep, and tracking exact delivery state would cost a disk write per delivery.

I also noticed a small thing here: the log says "recovered 5 durable queues", but there were only three distinct durable queues. The reason is that Queue.Declare writes its record as "delete, then insert" without holding the broker lock, so when three connections declared the same queue at the same moment (the two workers and the setup connection in the prefetch test), two records ended up on disk. Recovery deduplicates with entry().or_insert, so behaviour is unaffected; only the startup counter counts records. A small flaw I will fix in the next release.

The MQTT to AMQP bridge

The last phase of ADR-0016 connects the two protocols inside one binary, the way RabbitMQ's own MQTT plugin does: every MQTT publish is routed into the amq.topic topic exchange with / turned into ., and every AMQP publish to amq.topic comes out on the MQTT side with . turned back into /. amq.topic is pre-declared at startup, so binding to it needs no exchange_declare.

The scenario: a boiler sensor publishes over MQTT to sensors/boiler/temp; an AMQP worker has bound its durable temp-alerts queue to amq.topic with the pattern sensors.*.temp:

ch.queue_declare("temp-alerts", durable=True)
ch.queue_bind("temp-alerts", "amq.topic", routing_key="sensors.*.temp")   # amq.topic is pre-declared
ch.basic_qos(prefetch_count=10)
def on_msg(ch, m, props, body):
    print(f"[amqp worker] exchange={m.exchange} routing_key={m.routing_key} "
          f"delivery_mode={props.delivery_mode} body={body.decode()}")
    ch.basic_ack(m.delivery_tag)
ch.basic_consume("temp-alerts", on_msg)

mosquitto_pub -p 15804 -i boiler-probe -q 1 -t sensors/boiler/temp -m '{"value": 91.5, "unit": "C"}' -d
mosquitto_pub -p 15804 -q 1 -t sensors/boiler/pressure -m '{"value": 2.1, "unit": "bar"}'

Client boiler-probe sending CONNECT
Client boiler-probe received CONNACK (0)
Client boiler-probe sending PUBLISH (d0, q1, r0, m1, 'sensors/boiler/temp', ... (28 bytes))
Client boiler-probe received PUBACK (Mid: 1, RC:0)

[amqp worker] waiting on temp-alerts <- amq.topic 'sensors.*.temp'
[amqp worker] exchange=amq.topic routing_key=sensors.boiler.temp delivery_mode=2 body={"value": 91.5, "unit": "C"}

The pressure reading did not match the pattern, so the worker never saw it. A QoS 1 MQTT publish crosses into AMQP as delivery_mode=2: a message that arrived with a durability promise keeps it across the bridge. And because the same publish also matched the sensors/# ingest route, it was written to sensor_readings at the same time:

{"value": 91.5, "unit": "C", "_topic": "sensors/boiler/temp", "_received_at": 1791754410602, "_id": 4, "_version": 1}

One MQTT PUBLISH: to live MQTT subscribers, into an AMQP work queue, and into a queryable collection. All of it before the PUBACK.

The reverse direction works as well. I published a command from AMQP to amq.topic with the key devices.valve7.cmd, and a mosquitto_sub on devices/+/cmd received it:

[amqp publisher] confirmed amq.topic devices.valve7.cmd
devices/valve7/cmd {"open": false}

The bridge has loop protection: a publish that enters amq.topic from MQTT is marked bridge: false and never forwarded back out to MQTT.

Finally I tried the bridge's hardest case: with the worker down, the sensor publishes two QoS 1 readings, the server dies with kill -9, comes back, and only then does the worker connect:

published 2 (worker offline)
AMQP listening on 0.0.0.0:15805 - recovered 6 durable queues, 2 messages
[amqp worker] waiting on temp-alerts <- amq.topic 'sensors.*.temp'
[amqp worker] exchange=amq.topic routing_key=sensors.boiler.temp delivery_mode=2 body={"value": 92.8, "unit": "C"}
[amqp worker] exchange=amq.topic routing_key=sensors.boiler.temp delivery_mode=2 body={"value": 95.1, "unit": "C"}

The PUBACK the sensor got meant the copy in the AMQP queue was on disk too, and it was. Because amq.topic counts as durable, a durable queue's binding to it (_kind: "bind") is written to disk as well; the binding survived the restart along with the messages.

How it works inside

Storage: no new format, just documents

Both brokers keep their state in two separate collections, as documents discriminated by a _kind field:

As I wrote in ADR-0015, the reason is that I did not want a second durability path to test. The document engine's collection and WAL machinery already solves crash recovery, checkpoints and PITR, and the broker gets all of it for free. The server starts in "strict ACID-D" mode: every commit is fsync'd. The collections are separate so neither broker can corrupt the other's recovery.

Recovery runs at startup, before the listener accepts its first connection: find queries by _kind, messages sorted by seq and re-enqueued, and the sequence counter resumed at one past the highest seq seen.

Topic matching: two protocols, two rule sets

MQTT and AMQP wildcards look alike but are not, and the code has two separate matchers:

• MQTT (/-separated levels): + is exactly one level; # is only valid as the last level and means "the rest, including nothing". The broker compiles each subscription filter once into an anchored (^...$) regex: + becomes [^/]+, a trailing /# becomes (/.*)? (which is why sensors/# also matches plain sensors). A filter without wildcards becomes an exact match through regex::escape. The ingest bridge applies the same rules without regex, with a small recursive function over level slices.
• AMQP (.-separated words): * is exactly one word, # is zero or more words and may appear anywhere in the pattern. topic_matches handles # with a two-branch recursion: consume the #, or swallow one word of the key and stay on the #.

The bridge joins the two worlds by swapping only the separator (/ and .), exactly like RabbitMQ's MQTT plugin. An AMQP binding uses its own rules as its pattern (sensors.*.temp), and an MQTT filter uses its own.

Even when several bindings on an exchange match the same queue, the queue gets one copy: the target list is sorted and deduplicated (RabbitMQ semantics).

The AMQP queue: a shared object

In MQTT the durable state belongs to a client (the session). In AMQP a queue is a shared object with a consumer list and a round-robin pointer. Each connection runs on its own thread and asks the broker "which deliveries do you owe me?" (poll). Each ready message goes to the next consumer in turn; if that turn belongs to another connection, the message waits for that connection's own pass. That is what makes consumers actually compete instead of the fastest poller taking everything. A consumer at its prefetch limit is skipped.

The first version of this design had 51 ms end-to-end latency because of a 50 ms poll tick. According to the project's notes it dropped to 1.15 ms with an adaptive 1 ms tick, then to 0.02 ms with a cross-thread wakeup: every consuming connection owns a nonblocking pipe whose read end sits in its poll(2) set next to the TCP socket; when a message lands in a queue, the publisher writes one byte into the pipe of each of that queue's consumers, and their kernel wait ends immediately. (Unix only; other platforms keep the adaptive tick.)

Group commit: not one fsync per message

The first version of durable publishing did one fsync per message; per the project's measurements that was 264 messages per second on a single connection. Two layers changed that:

1. Per-connection batching. The connection routes the publishes of a pipeline burst with prepare_publish and collects them (up to 256, or until the socket buffer is drained), then writes all of them with one insert_many and sends the confirms in publish order. The order matters: the disk write comes first. An ack racing the enqueue and deleting a record that has not been written yet would resurrect a consumed message after a crash; the old per-message path had exactly that window wrong.
2. Cross-connection group commit. If no commit is on the disk, the arriving connection becomes the leader and writes everything pending, round after round; connections that arrive while it is on the disk share the next round and wait for their generation on a Condvar. The batch window is the fsync's own duration, so a lone publisher leads its own round immediately and pays nothing for this.

Bounds

Both brokers follow the same discipline: a slow or dead consumer must not exhaust the broker's memory, and loss must never be silent. The MQTT session queue is capped at 4096 messages, the AMQP ready queue at 100,000; in both, the oldest is dropped, a counter goes up, and the dropped message's on-disk record is deleted too. That last part matters: if the bound did not apply to the disk, a dead consumer would grow the collection forever and a restart would resurrect messages the bound had already discarded. I learned that on the MQTT side first and did it right from the start on AMQP.

Project benchmarks: compared with RabbitMQ

The repo has a benchmark that drives OxiDB and RabbitMQ with the same Go client (rabbitmq/amqp091-go) and the same scenarios. I did not run these numbers for this post; they are the project's own measurements from 2026-07-17, on an arm64 MacBook, OxiDB 0.36.12 against RabbitMQ 4.x (brew), 100-byte bodies:

The one losing row is an honest loss: every durable OxiDB confirm sits behind a real F_FULLFSYNC of the batch it arrived in. According to the project's notes, RabbitMQ classic queues flush lazily on an interval, so its persistent confirm does not by itself prove the message is on stable storage. With eight connections the group commit kicks in and the result flips.

Limits and honest trade-offs

• Node-local. MQTT sessions and AMQP queues live on the one node the connections live on. A client that reconnects to a different node after a failover does not resume its session. That is not an accident: a cross-node queue is the shared-log (Kafka) design both ADRs decline. The durable data behind the queue (collections, time series) is already replicated by the engines.
• MQTT 3.1.1, not MQTT 5. No shared subscriptions, message expiry or reason codes. If you want competing consumers, the answer is an AMQP queue.
• An AMQP subset. The out-of-scope list above is firm: no tx, headers exchanges, TTL, priority, dead-lettering, AMQP 1.0 or management UI. Clients that use them get an explicit error.
• Bounded queues drop the oldest. Right for telemetry, debatable for a payments queue. A queue approaching its bound is something to alert on.
• Conservative redelivery. Every AMQP message recovered after a crash arrives with redelivered=true. Consumers should be idempotent anyway; that is what AMQP's at-least-once implies.
• The fsync cost is real. In my single-publisher, sequential, durable run the difference was about 70x. For telemetry that does not need durability, QoS 0 or a clean session is still the cheapest path.
• Simple authentication. MQTT username/password and AMQP PLAIN; when unset, the listeners are open and bound to 0.0.0.0.

How it fits with the rest of OxiDB

The messaging layer is not a product on its own; it is OxiDB's front door. MQTT publishes also reach Redis-compatible SUBSCRIBE clients of OxiMem through the in-memory bus (and the other way round). The ingest bridge writes readings into the document engine, where the SQL engine can query them and from where long-lived metrics can move into the time-series engine. An AMQP worker pool can consume work that arrived over MQTT and write its results back into OxiDB. Logs and alerting get their own post in this series: logs and alerting.

For me the real lesson of this layer was that a broker's most important property is not its speed but that its promise on the wire and its behaviour are the same thing. In OxiDB a PUBACK now means "on disk", and I prove it with a kill -9 every time.

Trying it

Prebuilt binaries and Docker images are on the GitHub releases page. Two environment variables turn MQTT and AMQP on:

OXIDB_MQTT_PORT=1883 OXIDB_AMQP_PORT=5672 OXIDB_MQTT_PERSIST=1 oxidb-server

After that, use the tools you already have: mosquitto_pub, paho-mqtt, pika, amqp091-go or RabbitMQ.Client. None of them were modified for OxiDB. Next in the series is the logs and alerting engine.