// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie // 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)); }