zig-mqtt #

An MQTT client for Zig that speaks 5.0 and 3.1.1 and does not know what a socket is. It builds with Zig 0.17; for Zig 0.16, use the zig-0.16 branch, where the 0.1 releases live.

The protocol lives in one module and the transport in another. The protocol module takes bytes in, gives bytes out, and is told what time it is; it has no reader, no writer, no thread and no clock. That is what makes it usable over TCP, over TLS, over a WebSocket, over a serial line, or over a buffer in a test, and it is why the state machine can be driven through every one of its corners without a socket in sight.

The API documentation is generated from the doc comments, which carry most of the explanation, and is published at https://jeff.jcollie.page/zig-mqtt/.

var read_buffer: [4096]u8 = undefined;
var write_buffer: [4096]u8 = undefined;
var tcp: mqtt_client.Transport.Tcp =
    try .initHost(io, "broker.example", 1883, &read_buffer, &write_buffer);

var client: mqtt_client.Reconnecting = .init(gpa, io, tcp.transport(), .{
    .connect = .{ .client_id = "sensor-7" },
    .subscriptions = &.{.{ .filter = "commands/#", .qos = .at_least_once }},
});
defer client.deinit();

while (try client.next()) |event| switch (event) {
    .message => |m| std.debug.print("{s}: {s}\n", .{ m.topic, m.payload }),
    // The connection was made, or made again. `session_present` says whether
    // the broker still had the session, and the client has already re-sent the
    // subscriptions if it did not.
    .connected => {},
    else => {},
};

That is the form most programs want. Underneath it is a Client over a reader and a writer, and underneath that a Connection that does the protocol and no I/O at all; both are public, and either can be used on its own.

The two modules #

Module	Depends on	What it is
mqtt	nothing	The wire format and the session state machine. No I/O of any kind.
mqtt_client	mqtt	Carries those bytes along a std.Io.Reader and a std.Io.Writer, and reconnects when one breaks.

They are separate rather than one module with two halves, because the reason to want a sans-I/O core is usually that you have an event loop already. A program that does never compiles a line of mqtt_client and never sees std.Io.Reader in its dependency graph.

What it does #

Both versions of the protocol, in full on the client side:

• All fifteen control packets, encoded and decoded in both directions. Decoding a CONNECT or a SUBSCRIBE is of no use to a client, and is there because the fuzz targets need it and because a broker built on this layer would.
• QoS 0, 1 and 2, in both directions, including the receiver-side deduplication that makes QoS 2 mean what it says.
• Sessions. Unacknowledged publications survive a broken connection and are sent again, in order, with the duplicate flag set, as soon as the broker confirms the session. A QoS 2 exchange that got as far as PUBREC resumes with a PUBREL rather than the message.
• Flow control. The broker's Receive Maximum is honoured: publications beyond the window are queued and go out as earlier ones are acknowledged.
• MQTT 5.0 properties, everywhere they appear, validated against the per-packet tables — a property in the wrong packet, or repeated where it may not be, is a protocol error rather than something quietly ignored.
• Topic aliases, in both directions. Incoming ones are remembered and resolved; outgoing ones are established once and then used, which is the point of them.
• Shared subscriptions, subscription identifiers, and the rest of the 5.0 subscription options.
• Keepalive, including the Server Keep Alive that overrides what the client asked for.
• Extended authentication (section 4.12), driven as a SASL exchange through zig-sasl: the method goes in the CONNECT, every challenge is answered, and a mechanism's mutual half is checked before the connection is called established. Which mechanism is the caller's choice, including across reconnections.
• Automatic reconnection, with exponential backoff and jitter, session resumption, and re-subscription when the broker says the session did not survive.
• Sessions that survive the process, written down and read back, so that a QoS 2 message left half-delivered by one run is finished by the next.
• TLS, through tls.zig: TLS 1.2 and 1.3, with the broker's certificate checked against the system trust store or against an authority of your own; client certificates, for a broker that insists on mutual TLS; and ALPN, for one that multiplexes MQTT onto port 443 by protocol name.
• MQTT over WebSocket (section 6), through zig-websocket, for a broker reachable only through a proxy that forwards HTTP — and wss, which is the same framing with a TLS session under it instead of a socket.

Version differences are refused, not papered over #

A 3.1.1 connection cannot carry a property block, a reason code on an acknowledgement, a DISCONNECT that says why, a subscription option beyond QoS, or an AUTH packet. Encoding any of those under 3.1.1 returns error.UnsupportedByVersion rather than dropping the part that does not fit, because a packet that the broker accepts and that does something other than what was asked for is worse than an error.

Where the two versions can be reconciled they are. A 3.1.1 CONNACK's return code comes from a six-value table that overlaps the 5.0 reason codes numerically and means different things — 4 is "bad user name or password" in one and "disconnect with will message" in the other — so it is translated. A 3.1.1 SUBACK's return codes are a subset of the 5.0 table and are passed through unchanged, which is why a SUBACK's reason codes can be borrowed from the packet rather than copied.

Using it #

The stream client #

Client is a Connection with the byte-shuffling done for you. It is handed a reader and a writer that the caller opened however it likes, and it knows nothing about what is underneath them — with one exception, which is how it waits.

const mqtt = @import("mqtt");
const mqtt_client = @import("mqtt_client");

var read_buffer: [4096]u8 = undefined;
var write_buffer: [4096]u8 = undefined;
var tcp: mqtt_client.Transport.Tcp = .init(address, &read_buffer, &write_buffer);
const transport = tcp.transport();
defer transport.close(io);

const streams = try transport.open(io);

// The session is the caller's, because it outlives this transport and any
// number of later ones.
var conn: mqtt.Connection = .init(gpa, .{ .version = .v5_0 });
defer conn.deinit();

// `initOver` rather than `init`: taking the transport is what lets the client
// bound its wait by the keepalive deadline, and what lets it push bytes that
// are buffered below the writer. Without the first an idle connection is
// closed by the broker for never having pinged; without the second, TLS
// encrypts every packet and sends none of them.
var client: mqtt_client.Client = .initOver(&conn, io, transport, streams);

const ack = try client.connect(.{
    .client_id = "sensor-7",
    .keep_alive = 60,
    .will = .{ .topic = "sensor/7/status", .payload = "offline", .retain = true },
});
if (ack.properties.assigned_client_identifier) |name| {
    std.debug.print("the broker called us {s}\n", .{name});
}

publish, subscribe and unsubscribe return as soon as the packet is on the transport, handing back the packet identifier that a later event will carry. They do not wait for the acknowledgement, because a client that waits can only have one message in flight and the flow control window exists precisely so that it can have more.

Authenticating #

A user name and a password go in the CONNECT and need nothing from this section. Section 4.12 is for the rest: a broker that wants a real exchange — SCRAM, Kerberos, a one-time token — carries one in MQTT's own packets, with the method named in the CONNECT, each challenge in an AUTH, and the last message from the broker on the CONNACK.

That is a SASL exchange in an MQTT-shaped envelope, so the envelope is here and the mechanism is not:

var plain: mqtt_client.sasl.Plain = .init("sensor-7", "hunter2");
var buffer: [512]u8 = undefined;
client.authentication = .{ .mechanism = plain.client(), .buffer = &buffer };

_ = try client.connect(.{ .client_id = "sensor-7" });

sasl.Client is the interface, so PLAIN, EXTERNAL, CRAM-MD5, XOAUTH2 and OAUTHBEARER come from zig-sasl and SCRAM comes from zig-scram's scram-sasl adapter — three lines either way, and nothing here knows the difference. zig-sasl has no dependencies of its own, so a consumer that wants none of this pays for none of it.

The last message is the one that matters. For a mechanism with a mutual half — SCRAM, DIGEST-MD5 — the final message is the broker proving it knows the secret, and MQTT puts it on the CONNACK. A client that stopped at the CONNACK would report a connection to something that never proved anything, which is most of why such a mechanism is chosen at all. So connect feeds that data back to the mechanism and asks satisfied() before returning; a broker that accepted without proving itself is error.ServerNotAuthenticated and the connection is closed.

Connection enforces what section 4.12 says even if you drive it yourself: an AUTH must belong to an exchange the CONNECT asked for, every AUTH must name the same method, a broker cannot end the first exchange with an AUTH success, and a CONNACK that accepts a connection whose CONNECT named a method must name it too — which is what stops the exchange being skipped.

Reconnection needs one thing more, because a SASL exchange is a state machine that runs once and Reconnecting opens many connections. It takes a way to make a mechanism rather than a mechanism:

.authenticator = .{ .context = &credentials, .begin = Credentials.begin, .buffer = &buffer },

begin is called once per attempt, before the CONNECT goes out, and end releases whatever it built. A mechanism that cannot be built — a password SASLprep rejects — is error.AuthenticatorFailed and is not retried.

Re-authentication (section 4.12.1) is reauthenticate on either client, and it is the same operation as reconnecting minus the reconnecting — the session, the subscriptions and the unacknowledged messages all carry on across it, which is what it is for: a credential with an expiry need not cost a connection.

The two differ in where the exchange runs. Client.reauthenticate waits for it and drops whatever else arrives meanwhile, like the other round trips there. Reconnecting.reauthenticate sends the first message and returns, and the rest happens inside next alongside everything else — so a subscriber renewing a credential loses nothing, which is the case that makes the feature worth having. It takes the new mechanism from the authenticator, which is what an authenticator is already for.

Keeping the session #

A session is what makes QoS 1 and QoS 2 mean anything: "delivered at least once" is a promise kept by sending the message again until the broker says it arrived. A client that forgets what it was owed when it exits has quietly downgraded every one of those messages to QoS 0 with extra packets, which is the worst kind of wrong — it works in every test where nothing restarts.

So the session can be written down:

// Before connecting, and with `clean_start` false, or the broker throws away
// its half of what was just restored.
var reader: std.Io.Reader = .fixed(saved_bytes);
try conn.loadSession(&reader, "sensor-7");

// ...and afterwards, whenever there is something new to write.
if (conn.session_serial != last_saved) try conn.saveSession(&writer, "sensor-7");

Reconnecting has the same pair without the identifier, which it already knows, and refuses to load a session that clean_start would immediately discard — a mistake the shape of the options makes easy, because naming any field of connect replaces the default that made it safe.

What is stored is section 4.1's list: publications sent and not completely acknowledged, with the bytes that were sent so the duplicate flag is right when they go again; the packet identifiers of QoS 2 messages received and not yet released; and where the identifier counter had got to. Plus one thing beyond that list — publications the caller handed over that had not gone out yet, because losing a message the caller believes it published is worse than storing slightly more than the standard asks for.

What is not stored is anything that belongs to a connection rather than a session: topic aliases (section 3.3.2.3.4 makes them last one connection), subscriptions (the broker's session state, and Reconnecting re-sends them), and everything the CONNACK negotiates.

session_serial changes whenever the session does. It is generous in one direction only — it can move when a save would write the same bytes, and it never fails to move when they would differ — which is the direction that matters, since the cost of the first is a redundant write and the cost of the second is a lost message. A test drives a whole conversation and compares the counter against the bytes at every step, so it stays that way.

Both halves have to be asked for. Persisting this end accomplishes nothing unless the broker keeps the other end, and MQTT 5.0 reads a missing Session Expiry Interval as zero — section 3.1.2.11.2, the session ends when the network connection does. So clean_start = false on its own asks the broker to resume a session it is about to throw away. Measured against mosquitto, which does exactly that: two connections with the same client identifier, no expiry, and session_present false both times.

Connection.Options.session_expiry_interval is what sends one, filled into the CONNECT the same way the other advertised limits are, and Reconnecting defaults it to an hour because without it everything that type says about resuming sessions would be untrue. 3.1.1 has no such property: clean_session = 0 asks the broker to keep the session and the broker decides for how long.

Write the file atomically. A process killed mid-write is the failure this exists to survive, and every field is length-prefixed and the whole is checksummed, so a truncated or damaged file is refused rather than half believed — but a file truncated in place over a good one has destroyed the good one. Write a temporary file and rename it; src/main.zig does.

Staying connected #

Reconnecting owns a transport, opens it when it needs one, opens it again with backoff when it breaks, and resumes the session across the gap.

var read_buffer: [4096]u8 = undefined;
var write_buffer: [4096]u8 = undefined;
var tcp: mqtt_client.Transport.Tcp =
    try .initHost(io, "broker.example", 1883, &read_buffer, &write_buffer);

var client: mqtt_client.Reconnecting = .init(gpa, io, tcp.transport(), .{
    .connect = .{ .client_id = "sensor-7" },
    .subscriptions = &.{.{ .filter = "commands/#", .qos = .at_least_once }},
});
defer client.deinit();

while (try client.next()) |event| switch (event) {
    .message => |m| handle(m),
    // A reconnection is an event, not a silence: this is how a caller learns
    // there was a gap and whether the session survived it.
    .connected => |ack| log(ack.session_present),
    else => {},
};

Four things make that more than a retry loop. The session is not thrown away and rebuilt, so a QoS 1 message that was unacknowledged when the socket died goes again rather than being lost — which only works because clean_start defaults to false here, against Connection's own default of true. Subscriptions live in the broker's session, so when it reports that the session did not survive they are re-sent; a client that did not do this would stay connected and silently stop receiving, which is the worst way for this to fail. A CONNACK that refuses the connection is returned as an error rather than retried, because not_authorized will still be not_authorized in thirty seconds.

And not every transport failure is worth waiting out either. A refused socket is; a certificate for the wrong name is not, and both arrive here as an error from open. So the transport is asked — Transport.VTable.retry, which Tls and WebSocket fill in and Tcp leaves null because every way a TCP connection fails is a way it might not fail in a minute:

// Stops, rather than backing off to thirty seconds and staying there.
client.next() catch |err| switch (err) {
    error.TransportFailedPermanently => log(client.last_transport_error.?),
    else => return err,
};

Tls calls a certificate that does not verify, a TLS alert naming the certificate or the protocol version, and a local configuration the other end cannot meet fatal; everything else — a reset, a timeout, a truncated stream — is worth another attempt. It also reads the session's own record of what went wrong, because ReadFailed says nothing and the alert underneath it says everything. WebSocket draws the line where §4.2.2 already draws it: a 4xx is an answer about the request, a 5xx is a proxy in front of a broker that is restarting — and anything that is not the handshake's own it hands to the layer below, which is how wss inherits the whole of Tls's judgement. Options.retry overrides all of it, in either direction.

Receiving and publishing on one task #

next waits for as long as it takes, and a PINGREQ falling due is answered inside the wait rather than ending it — so a subscriber sitting on a quiet topic is parked indefinitely. That is what most subscribers want, and it is exactly wrong for a program that also has something to send, because this type is not safe from two tasks at once: publish and next both drive the same Connection, which owns the transmit buffer, the packet identifier counter and the in-flight window. Locking does not help when the wait is unbounded.

nextBefore gives the wait a deadline, so one task can do both:

while (true) {
    for (mailbox.drain()) |message| _ = try client.publish(message);

    const event = try client.nextBefore(.{ .duration = .{
        .raw = .fromMilliseconds(100),
        .clock = .boot,
    } }) orelse continue; // nothing yet; go round and send whatever has queued
    switch (event) {
        .message => |m| handle(m),
        else => {},
    }
}

Null is "nothing yet, ask again", and it is not ambiguous the way it would be on Client: a clean end of stream is a reconnection to make rather than news for the caller, so Reconnecting never returns null for one. Keepalive is unaffected — the wait is the earlier of the caller's deadline and whatever the connection wants next, so a ping falling due inside a long deadline still goes out on time. A deadline already past is a poll: everything in hand is still decoded and the connection is still ticked, and the transport is not waited on at all.

A reconnection is not interrupted by the deadline. While there is no connection, nextBefore waits out the backoff and the CONNECT round trip however long they take, because returning early would let a caller polling on a short deadline keep a connection from ever being made. A caller that needs its task back sooner than a broker being down allows wants a lower Backoff.max_ms.

Client has the same method, with three outcomes rather than two — .event, .end_of_stream, .timeout — because at that layer the end of the stream is the caller's news to have. Either way the deadline is applied to the transport's readable hook, so it needs a client built with initOver; a plain init over a bare reader has nothing to bound, since a std.Io.Reader cannot be asked to wait only so long.

Finding the broker #

Every transport here is built from an Io.net.IpAddress, because that is what a socket is opened to and what a reconnection opens again without asking anybody anything a second time. A caller holding a host name turns it into one with Transport.resolveHost — which is where the bare address in the other snippets comes from — or lets Tcp.initHost do both at once.

const address = try mqtt_client.Transport.resolveHost(io, "broker.example", 1883);

Worth having here rather than left to the caller, because the obvious way does not work. Io.net.IpAddress.resolve asks no resolver anything: it parses an IPv4 literal, then an IPv6 one, and the resolving it does is of an IPv6 scope. Every host name reaches it as neither and comes back error.ParseFailed — localhost included, which is how this was noticed. Io.net.HostName.lookup is the one that reads /etc/hosts and asks DNS.

An address literal is taken as one and never looked up, so a broker named by address needs no resolver at all. A name is resolved once, when the transport is made, and not again on each reconnection: a broker that has moved wants the program restarted, and a name looked up per attempt would turn a reconnect loop into a resolver loop.

TLS #

Transport.Tls is the same thing with a TLS session in the middle, and nothing above it changes:

var tls: mqtt_client.Transport.Tls = try .init(gpa, address, "broker.example", .system, .{
    .ciphertext_read = &ciphertext_read_buffer,
    .ciphertext_write = &ciphertext_write_buffer,
    .plaintext_read = &plaintext_read_buffer,
    .plaintext_write = &plaintext_write_buffer,
}, .{});
defer tls.deinit(io);

var client: mqtt_client.Reconnecting = .init(gpa, io, tls.transport(), .{ ... });

The name passed to init is what goes in SNI and what the certificate is checked against, and it is separate from the address connected to — verifying broker.example while connecting to an address you already have is an ordinary thing to want, and the alternative is a resolver in the middle of a transport.

How the certificate is judged is the fourth argument: .system for the operating system's trust store, .{ .file = "ca.pem" } for an authority of your own, .{ .bundle = ... } for one you loaded yourself, and .insecure for neither the name nor the issuer. Only the first three authenticate anything.

The last argument carries the rest. auth presents a client certificate, which is mutual TLS:

var auth: mqtt_client.tls.config.CertKeyPair =
    try .fromFilePath(gpa, io, .cwd(), "client.pem", "client.key");
defer auth.deinit(gpa);
// ... .{ .auth = &auth, .alpn_protocols = &.{"mqtt"} }

and alpn_protocols advertises protocol names, for a broker reached on 443 by name rather than by port. Transport.Tls.alpn_protocol is what the broker picked, and lastError() is worth reaching for whenever a read or write comes back as the interface's opaque ReadFailed — over TLS the session knows what actually happened, and for mutual TLS it usually happened after the handshake, because TLS 1.3 has the client send its Finished before the broker has judged its certificate.

The two ciphertext buffers must hold a whole TLS record and init refuses anything smaller. The two plaintext ones have no minimum at all, which is worth saying because the previous implementation — std.crypto.tls.Client — decrypted a whole record into the plaintext read buffer in one piece and silently lost whatever did not fit.

WebSocket #

Transport.WebSocket is section 6: the same packets in the same order, inside binary WebSocket messages, after an RFC 6455 opening handshake. The reason to want it is the network rather than the protocol — port 8883 is blocked by a great many corporate firewalls, and a WebSocket is how a byte stream gets through a proxy that forwards nothing else.

It is built on a Transport rather than on a socket, which is what makes wss the same code with a different thing underneath:

// ws: the framing over a plain socket.
var tcp: mqtt_client.Transport.Tcp = .init(address, &socket_read, &socket_write);
var ws: mqtt_client.Transport.WebSocket = try .init(tcp.transport(), "broker.example:8083", .{
    .message = &message,   // where an incoming message is assembled
    .read = &read,         // between the framing and MQTT
    .write = &write,
}, .{});

// wss: the same framing, over a TLS session instead.
var tls: mqtt_client.Transport.Tls = try .init(gpa, address, "broker.example", .system, .{
    .ciphertext_read = &ciphertext_read,
    .ciphertext_write = &ciphertext_write,
    .plaintext_read = &plaintext_read,
    .plaintext_write = &plaintext_write,
}, .{});
var wss: mqtt_client.Transport.WebSocket = try .init(tls.transport(), "broker.example:443", .{
    .message = &message,
    .read = &read,
    .write = &write,
}, .{});

var client: mqtt_client.Reconnecting = .init(gpa, io, wss.transport(), .{ ... });

The composition is not only where the bytes go. Waiting for something to read asks the layer below when the framing has nothing pending, so a TLS record that arrived whole is noticed without the socket being touched; and an error the handshake does not recognise is handed down to be judged, so a certificate that will never verify stops a reconnecting client even though what it was doing at the time was opening a WebSocket.

The second argument is the authority for the Host header. The path defaults to /mqtt, which is the convention rather than a requirement — section 6 says nothing about it, and mosquitto serves every path. Section 6 does require the subprotocol to be mqtt at both ends, so a broker that answers the handshake without choosing it is refused with error.SubprotocolNotMqtt rather than spoken to. When a handshake fails for an ordinary HTTP reason, WebSocket.status holds the status code: a 401 or 403 means credentials were wanted in a header — Options.headers is where those go — and a 404 usually means the path is wrong.

Message boundaries carry no meaning, which section 6 says in as many words, so a packet larger than the write buffer is sent as several messages and a message may hold several packets. Nothing above the transport sees either.

Transport itself is five function pointers — open, close, and optionally "wait until readable", "push what is buffered below the writer" and "is this failure worth another attempt" — and Tcp, Tls and WebSocket are the same five with more inside them. Transport.Tcp is the plain one.

Driving it yourself #

Connection is the whole protocol and none of the I/O. The caller moves the bytes and supplies the time:

var conn: mqtt.Connection = .init(gpa, .{});
defer conn.deinit();

try conn.connect(.{ .client_id = "sensor-7" }, now());

while (true) {
    // Anything the connection wants to say.
    const out = conn.pending();
    if (out.len != 0) conn.sent(try stream.write(out), now());

    // Anything the broker said.
    const n = try stream.read(&buf);
    try conn.receive(buf[0..n], now());
    while (try conn.nextEvent()) |event| switch (event) {
        .message => |m| handle(m),
        else => {},
    };

    // Keepalive, retransmission, flow control. `deadline()` says when this
    // next wants calling, which is the timeout for the poll above.
    try conn.tick(now());
}

Time is whatever monotonic millisecond count you have. The connection never asks what time it is; it is told, at the three points where it could matter.

Events borrow #

A .message event's topic and payload point into the receive buffer and are valid only until the next call to nextEvent or receive. An application that keeps a message must copy it.

This is the one sharp edge in the interface, and it is deliberate: it is what makes delivering a message cost nothing. The whole protocol layer allocates nothing at all — a decoded packet borrows every string from the bytes it was decoded from — and Connection allocates only where it must, to keep an unacknowledged message until it is acknowledged and to remember a topic alias after the packet that introduced it is gone.

What it does not do #

• A Client built over a bare reader and writer cannot keep itself alive. A std.Io.Reader has no deadline, so next blocks in it for as long as the transport does and a PINGREQ falling due meanwhile is never sent — which on an idle subscription is a disconnection every keep-alive-and-a-half. Client.initOver takes the Transport and gets from it both a way to wait with a deadline and a way to push anything buffered below the writer; that is what Reconnecting and the demo use. Plain init is for a reader and writer from somewhere else entirely, and there both are the caller's problem — Connection.deadline says when the keepalive falls due.
• No server. The packet layer decodes everything a broker would receive, and the end-to-end tests contain a few hundred lines that answer a client, but there is no broker here and none is planned.
• No broker anywhere tests the authentication exchange. mosquitto has no enhanced authentication and FlashMQ needs a C++ plugin for it, so section 4.12 is the one feature here judged only by a broker in this repository — tests/e2e.zig runs zig-sasl's PLAIN against its own server half, and a mechanism written for the purpose covers the mutual case. Everything else has a NixOS machine and somebody else's implementation behind it.
• No channel binding, so no SCRAM-SHA-256-PLUS. tls.zig exposes no keying-material exporter — no tls-exporter (RFC 9266), no Finished message — so there is nothing to bind a mechanism to. Plain SCRAM-SHA-256 works.
• A reconnecting client waits out a certificate renewal only if you ask it to. The transport judges its own failures — Transport.retryAfter — so a certificate for the wrong name, an unknown authority, or a WebSocket handshake answered with 403 stops with error.TransportFailedPermanently instead of backing off forever, and last_transport_error says which. That is a judgement call, and a program that would rather sit through a renewal than exit overrides it with Options.retry. What no transport can classify is a TLS alert that arrives as a reset rather than as an alert, which is a race: a broker refusing a client certificate produces one or the other, and only the first is recognisable.
• Outgoing topic aliases are never evicted. Aliases are handed out until the broker's limit is reached and then the topic is sent in full. A client publishing to more distinct topics than the broker will alias is better served by that than by thrashing.

Dependencies #

Three, all pinned by commit in build.zig.zon. Everything else is the standard library.

tls.zig is the TLS: a fork of ianic/tls.zig, pinned to its quic-0.17 branch, which is upstream's main with QUIC added and is the tls.zig that zig-http and zig-quic build on too. What this library needs from it is the alert handling: an alert the broker sent is kept and reported rather than discarded, and only failures that are this end's fault produce one. Transport.Tls.lastError is built on it, and without it a broker refusing a client certificate arrives as a bare ReadFailed with nothing to tell a user.

It is there rather than std.crypto.tls.Client for four reasons, three of them capabilities this library exposes — client certificates, ALPN, and a handshake that works with servers which omit the optional TLS 1.3 middlebox-compatibility record. The fourth is that it has no minimum size for the plaintext read buffer, where the standard library's client truncated large messages in silence.

For Nix, build.zig.zon.nix is generated by zon2nix and committed, so the sandboxed build is handed its dependency rather than fetching one it has no network for. Regenerate it whenever build.zig.zon changes:

$ nix develop -c zon2nix --16 --nix=build.zig.zon.nix build.zig.zon

A consumer does not need a dependency of its own: mqtt_client.tls re-exports the library, which is where tls.config.CertKeyPair comes from.

zig-websocket is the RFC 6455 framing and the client half of its opening handshake. Only its websocket module is taken: the server half of the handshake lives in a second module there, written against an HTTP server library, and a client has no use for one. As with TLS, a consumer needs no dependency of its own: mqtt_client.websocket re-exports it, which is where the header type Options.headers takes comes from.

zig-sasl is the authentication interface, and the interface only — which mechanism to speak is the caller's choice and a library that picked one would be wrong for everybody else. It has no dependencies at all, so this costs a consumer nothing but the mechanisms it already wanted, and mqtt_client.sasl re-exports it.

Building and testing #

$ nix develop                    # a Zig, the brokers, openssl, reuse, git-pages-cli
$ zig build test --summary all   # 219 tests
$ zig build                      # the demo client
$ zig build docs-serve           # read the API documentation at :8000

There are four kinds of test, and each covers something the others cannot.

Unit tests, beside the code, in src/. Encoding and decoding checked against bytes written out of the specification, and the state machine driven through every handshake by handing it packets and reading back what it wants to send.

End-to-end tests, in tests/e2e.zig. A real client and a small broker on two ends of a real loopback socket, run concurrently, because what the state machine believes it is sending and what comes out of the other end of a socket are two different claims.

Fuzz targets, in tests/fuzz.zig: the packet decoders, the property blocks, the topic rules, the variable byte integers, the connection fed broker bytes, the connection driven by application commands, the reconnecting client meeting a transport that keeps breaking, and the encoder measured against itself. The connection target runs with and without an authentication exchange asked for, because section 4.12's rules are only in the path when one was. Each is an ordinary test as well, so zig build test runs the properties against the seeds beside them.

That last one guards something sharp. Connection.enqueue asks the encoder how much room a packet needs, reserves exactly that, and treats a short write as unreachable — so a measuring pass that disagrees with the writing pass by one byte is a panic in a checked build and a heap overflow in an unchecked one. The target builds packets out of fuzz data, with wills and user properties and long filter lists that no decoded packet would have, and checks the two passes agree.

$ zig build fuzz --fuzz                     # Zig's fuzzer, with a web interface
$ zig build fuzz --fuzz=1M                  # a bounded run, then a report
$ zig build fuzz-run                        # about a minute of each target
$ zig build fuzz-run -- --seconds 600 --target connection

zig build fuzz --fuzz is Zig's own fuzzer, steered by coverage. That needs the fuzz tests compiled by LLVM — the self-hosted backend Debug uses by default emits no coverage instrumentation — so build.zig sets use_llvm on them. zig build fuzz-run drives the same targets through the loop in tools/fuzz.zig instead: seeded, bounded, and an ordinary executable, which is what CI runs and what reproduces a run exactly.

A word of warning if you write targets like these. std.testing.Smith's value and its ranged forms return the range's minimum whenever the bytes they are reading fall outside it, and random bytes almost always do — measured here at 2000 out of 2000. Four of these targets picked their parameters that way and so ran one point of their parameter space for entire campaigns: reconnection never reconnected, properties only ever looked at CONNECT, the connection target never spoke 3.1.1, and the command target never met a small flow control window. All of them report a large number of iterations while doing it. Smith.bytes reads raw bytes and is uniform; everything here uses it now, and the change turned up a real encoder bug within eighty thousand iterations.

Between them the targets have found four real bugs, each of which is now a named test:

• a 3.1.1 CONNACK's return code was widened into the 5.0 table on the way in and written back out as the 5.0 number, so server_unavailable went to a 3.1.1 client as 0x88 — a code that table does not have;
• a topic name beginning with $share/ was accepted by validateName, and no filter could ever match it, so a message published there would have gone nowhere;
• the variable byte integer decoder accepted non-minimal encodings, which section 1.5.5 forbids and which would have made a packet's length non-canonical;
• a protocol error while resolving an incoming topic alias propagated without closing the connection, so a caller that ignored it could keep publishing into a session the broker was about to drop.

The worst one the fuzzer could never have found, because it needed a real TLS session: the plaintext read buffer was one record short, so a message larger than 16 KiB arrived truncated, with no error anywhere — 20,000 bytes published, 1,156 read. Transport.Tls.init now refuses a buffer that small.

Three more came out of the interoperability tests and of reading back what had been written. A broker could send a packet before its CONNACK and reach an unreachable in Client.connect; section 3.2 forbids that, and now so does Connection. Client flushed only when it was about to block, so a PUBACK could sit in the buffer while a caller drained a burst of events and a broker with a small Receive Maximum waited for it. And an idle subscription was being disconnected at a keep-alive and a half, because the blocking read meant the PINGREQ that fell due during it was never sent — which is what Client.initOver and Transport.VTable.readable exist to fix, and which nothing but a broker and a clock was ever going to catch.

Interoperability tests, in tests/nixos/, run as nix flake check. Each boots a machine with a real broker on it and talks to it with the demo client. There are four, and they are chosen for what they disagree about:

• mosquitto leaves most of the CONNACK out — no Retain Available, no Maximum QoS, no Server Keep Alive — so every one of those falls to the specification's default, and this is where the client is proved to apply them. It is also where the client meets a second client implementation, with messages sent both ways through mosquitto_pub and mosquitto_sub, and where a session is proved to survive a restart: one process publishes at QoS 2 without finishing the handshake, and a second one with nothing but the session file finishes it and the message is delivered exactly once.

• FlashMQ fills the CONNACK in, including a Server Keep Alive that overrides what the client asked for, and a Receive Maximum of 512 against mosquitto's 20 — so the flow control window is effectively open, which is a path the other machine never takes.

• mosquitto again, over TLS, which is a machine of its own because almost everything interesting about TLS is what should fail: a certificate for another name, one from an authority we do not trust, and one that did not sign itself where --self-signed was asked for. It also publishes 100,000 bytes through a real TLS session, which is the only thing that catches a message truncated at a record boundary. A second listener on it requires a client certificate, because mutual TLS is the capability the switch to tls.zig was for.

• mosquitto again, over a WebSocket, which is where section 6 is judged by somebody else's implementation of it. The same broker serves plain MQTT on 1883 beside it and wss on 8884, so messages cross between all three and any difference in behaviour belongs to the framing. The wss listener is also where the layering is checked rather than assumed: a certificate for the wrong name has to be reported by the TLS layer and has to stop a reconnecting client, both of which happen underneath a WebSocket handshake that never gets sent. It is also the only test anywhere that has zig-websocket's client handshake answered by a server that was not written alongside it.

Each boots a NixOS guest, so running them needs a machine that can nest virtualisation — /dev/kvm has to be there. They are what found the keepalive bug and the TLS truncation bug, neither of which any amount of testing this repository against itself was ever going to.

The demo client #

src/main.zig is not the point of the repository; it is there so that zig build run -- sub -t '#' answers the question "does any of this work against mosquitto", which no unit test can.

$ zig build run -- info                                     # what the broker says about itself
$ zig build run -- sub -t 'sensor/#'
$ zig build run -- pub -t sensor/7/temperature -m 21.5 -q 1 -r
$ zig build run -- sub -R -i my-client -t 'commands/#'      # and stay subscribed
$ zig build run -- pub -3 -t old/style -m hello             # MQTT 3.1.1
$ zig build run -- info -S -H broker.example               # over TLS, system trust store
$ zig build run -- sub -S -H 127.0.0.1 --tls-host broker.example \
      --cafile ca.pem -t 'sensor/#'                        # and with your own authority
$ zig build run -- info -S -H broker.example \
      --cert client.pem --key client.key                   # mutual TLS
$ zig build run -- info -S -H broker.example -p 443 --alpn mqtt
$ zig build run -- info -W -H broker.example                # over a WebSocket, port 8083
$ zig build run -- sub -W -H broker.example --ws-path / -t 'sensor/#'
$ zig build run -- info -W -S -H broker.example             # wss, port 8884
$ zig build run -- pub -i s7 --session s.dat -t a/b -m hi -q 2 --no-wait
$ zig build run -- resume -i s7 --session s.dat             # finish what that owed
$ zig build run -- info --sasl plain -u sensor-7 -P hunter2 # a SASL exchange
$ zig build run -- info -S --cert client.pem --key client.key --sasl external

Where this lives #

Three places, and they are all the same repository.

$ git clone https://git.jcollie.dev/jeff/zig-mqtt.git

is the one to use. It is also mirrored on Tangled, at https://tangled.org/jcollie.dev/zig-mqtt.

And it is on Radicle, where a repository is findable only by its identifier, so here it is:

$ rad clone rad:z3S9yifrR4sAA5am2yoJVXn2eJhPs

Licensing #

MIT, and REUSE compliant — every file carries its copyright and licence, and reuse lint says so.

References cited #

• [MQTT-5.0] Banks, A., Briggs, E., Borgendale, K., and R. Gupta, Eds., "MQTT Version 5.0", OASIS Standard, 07 March 2019, https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html. The specification this implements. Section numbers in the comments throughout src/ refer to it unless they say otherwise; the ones worth knowing are 1.5.4 and 1.5.5 for the data representations, 2.2.2 for properties, 3.x for the packets, 4.1 for what a session is and 4.4 for what a resumed one owes, 4.7 for topic matching and 4.8.2 for shared subscriptions.
• [MQTT-3.1.1] Banks, A. and R. Gupta, Eds., "MQTT Version 3.1.1", OASIS Standard Incorporating Approved Errata 01, 10 December 2015, https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html. The older protocol, which most brokers still accept and some still require. Read for the differences: no properties, no reason codes, a six-value CONNACK return code table, and reserved bits in the subscription options byte that 5.0 gave meanings to. Section 4.12 is extended authentication and 4.12.1 is re-authenticating a connection that is already up; 3.15 is the AUTH packet they are carried in. Section 6 is MQTT over WebSocket, and it is three paragraphs: binary frames, the mqtt subprotocol at both ends, and an explicit warning that packets do not align with message boundaries.
• [RFC 4422] Melnikov, A., Ed. and K. Zeilenga, Ed., "Simple Authentication and Security Layer (SASL)", RFC 4422, DOI 10.17487/RFC4422, June 2006, https://www.rfc-editor.org/info/rfc4422. What section 4.12 carries. MQTT names no mechanism and defines no framing for one, so what an authentication method is comes from here; the mechanisms themselves live in zig-sasl.
• [RFC 6455] Fette, I. and A. Melnikov, "The WebSocket Protocol", RFC 6455, DOI 10.17487/RFC6455, December 2011, https://www.rfc-editor.org/info/rfc6455. What MQTT section 6 rides on. Section 4.1 is the client's opening handshake and 4.2.2 is what the response must contain for it to be accepted; the framing itself is implemented in zig-websocket rather than here.