Something went wrong. Try again.
An MQTT 5.0 and 3.1.1 client for Zig: a sans-I/O core, TCP, TLS, WebSocket and wss transports, SASL authentication, and reconnection
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! Writing a session down and reading it back, so that it survives the process//! that made it.//!//! MQTT's session is the reason QoS 1 and QoS 2 mean anything. A message//! published at QoS 1 is "delivered at least once" only because the client//! keeps sending it until the broker says it arrived, and a client that forgets//! what it was owed the moment it exits has downgraded every one of those//! messages to QoS 0 with extra packets. Section 4.1 is explicit that a session//! may outlive a network connection; nothing in it says the session may not//! outlive the process, and for a sensor that publishes and is restarted by a//! watchdog that is exactly what is wanted.//!//! ## What is in it//!//! Section 4.1's list, from the client's side://!//! * QoS 1 and QoS 2 publications sent and not completely acknowledged, with//! the bytes that were sent, so that they can go again with the duplicate//! flag where the specification requires it.//! * The packet identifiers of QoS 2 messages received and not yet released,//! which is how a retransmitted PUBLISH is recognised rather than delivered//! twice.//! * Where the packet identifier counter had got to, so that a restart cannot//! hand out an identifier that is still in flight.//!//! And one thing beyond that list: publications the caller handed over which//! had not yet gone out, because the broker's Receive Maximum was full. Section//! 4.1 does not call those session state -- they were never sent -- but losing//! them would be losing a message the caller believes it published, which is a//! worse failure than being slightly more generous than the standard requires.//!//! ## What is deliberately not in it//!//! **Topic aliases.** Section 3.3.2.3.4 makes them last only for the network//! connection they were established on, so a restored alias would be a//! reference to a topic the broker has never heard of.//!//! **Subscriptions.** They are the broker's session state, not the client's.//! `Reconnecting` re-sends them when a CONNACK says the session did not//! survive, which covers a restart as well as a reconnection.//!//! **Anything the CONNACK establishes** -- the Receive Maximum, the Maximum//! Packet Size, the keep alive in force. All of it is negotiated afresh on//! every connection, so storing it would be storing an answer to a question//! that is about to be asked again.//!//! ## Using it//!//! ```zig//! var conn: mqtt.Connection = .init(gpa, .{});//! defer conn.deinit();//!//! // Before connecting, and with `clean_start = false`, or the broker will//! // throw away its half of what was just restored.//! if (readSessionFile()) |bytes| {//! var reader: std.Io.Reader = .fixed(bytes);//! try conn.loadSession(&reader, "sensor-7");//! }//! ```//!//! Saving is the caller's to schedule, and `Connection.session_serial` says//! when there is anything to save: it changes whenever the session does, and//! not otherwise, so a caller that remembers the value it last wrote knows//! whether writing again would accomplish anything.//!//! **Write it atomically.** A session half-written by a process that was killed//! is the one realistic corruption, and every field here is length-prefixed, so//! a truncated file is caught -- but a file truncated *in place* over a good//! one has destroyed the good one. Write a temporary file and rename it.
const std = @import("std");const Io = std.Io;
const types = @import("protocol/types.zig");const PacketId = types.PacketId;const Qos = types.Qos;const Version = types.Version;
/// What every session file begins with, so that pointing this at the wrong/// file is an error rather than a parse.pub const magic = "MQTTSESS";
/// The layout below. Bumped if it ever changes, which makes an old file/// `error.UnsupportedFormat` rather than a wrong answer.pub const format_version: u8 = 1;
/// Integers are big-endian, as everything else in this repository is: MQTT's/// own wire format is, and one endianness in one project is one fewer thing to/// remember.const endian: std.builtin.Endian = .big;
pub const SaveError = error{ OutOfMemory, /// The client identifier was empty. A client that lets the broker name it /// gets a new name every time it connects, so it can never resume /// anything, and a session written under no name is a session that will /// never be loaded. SessionNotResumable,} || Io.Writer.Error;
pub const LoadError = error{ /// The bytes do not begin with `magic`. NotASession, /// Written by a version of this that laid the fields out differently. UnsupportedFormat, /// The file ended early, or its checksum does not match what it holds. CorruptSession, /// Saved by a client with a different identifier. MQTT identifies a /// session by the client identifier and nothing else, so restoring one /// under another name would be claiming somebody else's unacknowledged /// messages. SessionBelongsToAnotherClient, /// Saved by a connection speaking the other protocol version. The packets /// in it are encoded for that one and would not mean the same thing. SessionVersionMismatch, /// A field says it is longer than anything this will read, which a good /// file never does. SessionTooLarge, OutOfMemory,} || Io.Reader.Error;
/// The largest a single stored PUBLISH may be, and the largest client/// identifier. Both are bounds on what a *corrupt* file can make this/// allocate; a real one is far below them.pub const max_stored_packet_len = 256 * 1024 * 1024;pub const max_client_id_len = 65535;
/// One publication in flight, as it is written down. The stage and the bytes/// together are what a resumed session owes: a PUBLISH to send again, or a/// PUBREL to finish a handshake whose PUBLISH is long gone.pub const StoredOutbound = struct { packet_id: PacketId, qos: Qos, /// The same four states `Connection` keeps, written as a number so that /// reordering the enum cannot silently change what an old file means. stage: Stage, resend_pubrel: bool, /// The encoded PUBLISH, with the duplicate flag already set if it has been /// sent once. Empty once the PUBREC has arrived and only the PUBREL is /// owed. /// /// Mutable because a `Connection` that takes one over will set the /// duplicate bit in it when it sends it again, which is a rewrite of one /// nibble rather than a re-encode. bytes: []u8,
pub const Stage = enum(u8) { queued = 0, awaiting_puback = 1, awaiting_pubrec = 2, awaiting_pubcomp = 3, };};
/// Everything a session is, in a shape that can be written and read.////// A struct rather than a pair of methods on `Connection` so that the format/// can be tested against values built by hand, including the ones a `Connection`/// would never produce.pub const Snapshot = struct { version: Version, client_id: []const u8, next_packet_id: PacketId, outbound: []const StoredOutbound, inbound_qos2: []const PacketId,};
/// Write a snapshot, checksum and all.////// Built in memory first, because the checksum covers everything before it and/// a `std.Io.Writer` cannot be read back. That is affordable because the thing/// being written is bounded: it is the in-flight window, which the broker's/// Receive Maximum caps, plus whatever the caller queued behind it.pub fn write(gpa: std.mem.Allocator, w: *Io.Writer, snapshot: Snapshot) SaveError!void { if (snapshot.client_id.len == 0) return error.SessionNotResumable;
var body: Io.Writer.Allocating = .init(gpa); defer body.deinit(); const out = &body.writer;
try out.writeAll(magic); try out.writeByte(format_version); try out.writeByte(@backingInt(snapshot.version)); try writeBytes(out, snapshot.client_id); try out.writeInt(PacketId, snapshot.next_packet_id, endian);
try out.writeInt(u32, @intCast(snapshot.outbound.len), endian); for (snapshot.outbound) |o| { try out.writeInt(PacketId, o.packet_id, endian); try out.writeByte(@backingInt(o.qos)); try out.writeByte(@backingInt(o.stage)); try out.writeByte(@intFromBool(o.resend_pubrel)); try writeBytes(out, o.bytes); }
try out.writeInt(u32, @intCast(snapshot.inbound_qos2.len), endian); for (snapshot.inbound_qos2) |id| try out.writeInt(PacketId, id, endian);
const bytes = body.written(); try w.writeAll(bytes); // Over everything before it, and so not over itself. try w.writeInt(u32, std.hash.Crc32.hash(bytes), endian);}
/// Read a snapshot. Everything it points at is allocated from `gpa` and is/// freed by `Snapshot.deinit` below, or taken over by `Connection.loadSession`.pub fn read(gpa: std.mem.Allocator, r: *Io.Reader, buffer: *Buffer) LoadError!Snapshot { // Everything read is fed to the checksum as it goes, which is why each // helper takes it: there is no second pass over bytes a `std.Io.Reader` // has already handed out. var crc: std.hash.Crc32 = .init();
var header: [magic.len]u8 = undefined; r.readSliceAll(&header) catch |err| switch (err) { error.EndOfStream => return error.NotASession, else => |e| return e, }; if (!std.mem.eql(u8, &header, magic)) return error.NotASession; crc.update(&header); if (try byte(r, &crc) != format_version) return error.UnsupportedFormat;
const version: Version = switch (try byte(r, &crc)) { 4 => .v3_1_1, 5 => .v5_0, else => return error.CorruptSession, }; buffer.client_id = try readBytes(gpa, r, max_client_id_len, &crc); const next_packet_id = try int(PacketId, r, &crc);
const outbound_count = try int(u32, r, &crc); // A count is not a promise: a corrupt one claiming four billion entries // must not become four billion allocations before the stream runs out. // Each entry is at least eight bytes, so this is what the file could // possibly hold. try buffer.outbound.ensureTotalCapacity(gpa, @min(outbound_count, 1024)); for (0..outbound_count) |_| { const packet_id = try int(PacketId, r, &crc); const qos: Qos = switch (try byte(r, &crc)) { 0 => .at_most_once, 1 => .at_least_once, 2 => .exactly_once, else => return error.CorruptSession, }; const stage: StoredOutbound.Stage = switch (try byte(r, &crc)) { 0 => .queued, 1 => .awaiting_puback, 2 => .awaiting_pubrec, 3 => .awaiting_pubcomp, else => return error.CorruptSession, }; const resend_pubrel = switch (try byte(r, &crc)) { 0 => false, 1 => true, else => return error.CorruptSession, }; const bytes = try readBytes(gpa, r, max_stored_packet_len, &crc); try buffer.outbound.append(gpa, .{ .packet_id = packet_id, .qos = qos, .stage = stage, .resend_pubrel = resend_pubrel, .bytes = bytes, }); }
const inbound_count = try int(u32, r, &crc); try buffer.inbound.ensureTotalCapacity(gpa, @min(inbound_count, 1024)); for (0..inbound_count) |_| { try buffer.inbound.append(gpa, try int(PacketId, r, &crc)); }
// The checksum is the last four bytes and covers nothing but what came // before it. Checked after the fields are read rather than before, because // reading them is what proves the lengths were survivable in the first // place. const expected = crc.final(); const stored = try int(u32, r, &crc); if (stored != expected) return error.CorruptSession;
return .{ .version = version, .client_id = buffer.client_id, .next_packet_id = next_packet_id, .outbound = buffer.outbound.items, .inbound_qos2 = buffer.inbound.items, };}
/// Where `read` puts what it allocates, so that a failure half way through/// frees what it had rather than leaking it.pub const Buffer = struct { client_id: []u8 = &.{}, outbound: std.ArrayList(StoredOutbound) = .empty, inbound: std.ArrayList(PacketId) = .empty,
pub fn deinit(b: *Buffer, gpa: std.mem.Allocator) void { gpa.free(b.client_id); for (b.outbound.items) |o| gpa.free(o.bytes); b.outbound.deinit(gpa); b.inbound.deinit(gpa); b.* = undefined; }
/// Give up ownership of the outbound bytes, for a caller that has taken /// them over. Everything else is still this buffer's to free. pub fn releaseOutbound(b: *Buffer) void { b.outbound.clearRetainingCapacity(); }};
// -- the small pieces --------------------------------------------------------
fn writeBytes(w: *Io.Writer, bytes: []const u8) Io.Writer.Error!void { try w.writeInt(u32, @intCast(bytes.len), endian); try w.writeAll(bytes);}
fn byte(r: *Io.Reader, crc: *std.hash.Crc32) LoadError!u8 { var one: [1]u8 = undefined; r.readSliceAll(&one) catch |err| switch (err) { error.EndOfStream => return error.CorruptSession, else => |e| return e, }; crc.update(&one); return one[0];}
fn int(comptime T: type, r: *Io.Reader, crc: *std.hash.Crc32) LoadError!T { var bytes: [@divExact(@typeInfo(T).int.bits, 8)]u8 = undefined; r.readSliceAll(&bytes) catch |err| switch (err) { error.EndOfStream => return error.CorruptSession, else => |e| return e, }; crc.update(&bytes); return std.mem.readInt(T, &bytes, endian);}
fn readBytes( gpa: std.mem.Allocator, r: *Io.Reader, limit: usize, crc: *std.hash.Crc32,) LoadError![]u8 { const len = try int(u32, r, crc); if (len > limit) return error.SessionTooLarge; const bytes = try gpa.alloc(u8, len); errdefer gpa.free(bytes); r.readSliceAll(bytes) catch |err| switch (err) { error.EndOfStream => return error.CorruptSession, else => |e| return e, }; crc.update(bytes); return bytes;}
// -- tests -------------------------------------------------------------------
const testing = std.testing;
fn roundTrip(gpa: std.mem.Allocator, snapshot: Snapshot, buffer: *Buffer) !Snapshot { var out: Io.Writer.Allocating = .init(gpa); defer out.deinit(); try write(gpa, &out.writer, snapshot);
var in: Io.Reader = .fixed(out.written()); return read(gpa, &in, buffer);}
test "a session survives being written down and read back" { var publish = [_]u8{ 0x32, 0x0a, 0x00, 0x03, 'a', '/', 'b', 0x00, 0x07, 'h' }; const snapshot: Snapshot = .{ .version = .v5_0, .client_id = "sensor-7", .next_packet_id = 4242, .outbound = &.{ .{ .packet_id = 7, .qos = .at_least_once, .stage = .awaiting_puback, .resend_pubrel = false, .bytes = &publish, }, // The shape a QoS 2 handshake leaves behind once its PUBREC has // arrived: no publish left to send, only a PUBREL owed. .{ .packet_id = 8, .qos = .exactly_once, .stage = .awaiting_pubcomp, .resend_pubrel = true, .bytes = &.{}, }, }, .inbound_qos2 = &.{ 11, 12, 13 }, };
var buffer: Buffer = .{}; defer buffer.deinit(testing.allocator); const got = try roundTrip(testing.allocator, snapshot, &buffer);
try testing.expectEqual(Version.v5_0, got.version); try testing.expectEqualStrings("sensor-7", got.client_id); try testing.expectEqual(@as(PacketId, 4242), got.next_packet_id); try testing.expectEqual(@as(usize, 2), got.outbound.len); try testing.expectEqual(@as(PacketId, 7), got.outbound[0].packet_id); try testing.expectEqualSlices(u8, &publish, got.outbound[0].bytes); try testing.expectEqual(StoredOutbound.Stage.awaiting_pubcomp, got.outbound[1].stage); try testing.expect(got.outbound[1].resend_pubrel); try testing.expectEqual(@as(usize, 0), got.outbound[1].bytes.len); try testing.expectEqualSlices(PacketId, &.{ 11, 12, 13 }, got.inbound_qos2);}
test "an empty session is still a session" { const snapshot: Snapshot = .{ .version = .v3_1_1, .client_id = "x", .next_packet_id = 1, .outbound = &.{}, .inbound_qos2 = &.{}, }; var buffer: Buffer = .{}; defer buffer.deinit(testing.allocator); const got = try roundTrip(testing.allocator, snapshot, &buffer); try testing.expectEqual(Version.v3_1_1, got.version); try testing.expectEqual(@as(usize, 0), got.outbound.len);}
test "a client that cannot be named cannot be resumed" { // An empty client identifier means the broker chooses one, and it chooses // a different one every time -- so there is nothing for a later connection // to resume, and writing the file would be writing something that can only // ever be rejected. var out: Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); try testing.expectError(error.SessionNotResumable, write(testing.allocator, &out.writer, .{ .version = .v5_0, .client_id = "", .next_packet_id = 1, .outbound = &.{}, .inbound_qos2 = &.{}, }));}
test "what is not a session is not read as one" { var buffer: Buffer = .{}; defer buffer.deinit(testing.allocator);
var empty: Io.Reader = .fixed(""); try testing.expectError(error.NotASession, read(testing.allocator, &empty, &buffer));
var wrong: Io.Reader = .fixed("not a session at all"); try testing.expectError(error.NotASession, read(testing.allocator, &wrong, &buffer));
// The right magic and a format number from the future. var future_bytes: [magic.len + 1]u8 = undefined; @memcpy(future_bytes[0..magic.len], magic); future_bytes[magic.len] = format_version + 1; var future: Io.Reader = .fixed(&future_bytes); try testing.expectError(error.UnsupportedFormat, read(testing.allocator, &future, &buffer));}
test "a session that was half written is refused rather than half believed" { const snapshot: Snapshot = .{ .version = .v5_0, .client_id = "sensor-7", .next_packet_id = 9, .outbound = &.{}, .inbound_qos2 = &.{ 1, 2, 3 }, }; var out: Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); try write(testing.allocator, &out.writer, snapshot); const whole = out.written();
// Every truncation of it, which is what a process killed mid-write leaves // behind. None may be read as a shorter but valid session. var at: usize = 0; while (at < whole.len) : (at += 1) { var buffer: Buffer = .{}; defer buffer.deinit(testing.allocator); var r: Io.Reader = .fixed(whole[0..at]); try testing.expectError( if (at < magic.len) error.NotASession else error.CorruptSession, read(testing.allocator, &r, &buffer), ); }}
test "a byte flipped anywhere in a session is caught" { // What the checksum is for. Every field is range-checked on the way in, so // most corruption is caught without it -- but "most" is not a property // worth relying on when what gets published is the result. const snapshot: Snapshot = .{ .version = .v5_0, .client_id = "sensor-7", .next_packet_id = 9, .outbound = &.{}, .inbound_qos2 = &.{ 1, 2, 3 }, }; var out: Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); try write(testing.allocator, &out.writer, snapshot);
const damaged = try testing.allocator.dupe(u8, out.written()); defer testing.allocator.free(damaged);
for (0..damaged.len) |i| { const original = damaged[i]; damaged[i] = original ^ 0x40; defer damaged[i] = original;
var buffer: Buffer = .{}; defer buffer.deinit(testing.allocator); var r: Io.Reader = .fixed(damaged); if (read(testing.allocator, &r, &buffer)) |_| { std.debug.print("byte {d} flipped and the session was still read\n", .{i}); return error.TestUnexpectedResult; } else |_| {} }}
test "a length nobody could have written is refused before it is allocated" { // A corrupt file claiming a four-gigabyte client identifier must not // become a four-gigabyte allocation on the way to failing. var bytes: Io.Writer.Allocating = .init(testing.allocator); defer bytes.deinit(); try bytes.writer.writeAll(magic); try bytes.writer.writeByte(format_version); try bytes.writer.writeByte(@backingInt(Version.v5_0)); try bytes.writer.writeInt(u32, 0xffff_ffff, endian);
var buffer: Buffer = .{}; defer buffer.deinit(testing.allocator); var r: Io.Reader = .fixed(bytes.written()); try testing.expectError(error.SessionTooLarge, read(testing.allocator, &r, &buffer));}