// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>
// SPDX-License-Identifier: MIT

//! An MQTT client session with no I/O in it.
//!
//! Everything this library knows about being a client lives here: the packet
//! identifier counter, the QoS 1 and QoS 2 handshakes in both directions, the
//! flow control window the broker imposes, topic aliases, keepalive, and what
//! has to be sent again when a session is resumed. None of it touches a socket,
//! a file, a clock or a thread.
//!
//! The caller moves the bytes and supplies the time:
//!
//! ```zig
//! var conn: 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.
//!     try conn.tick(now());
//! }
//! ```
//!
//! **Events borrow.** A `.message` event's topic and payload point into the
//! receive buffer, and stay valid only until the next call to `nextEvent` or
//! `receive`. An application that keeps a message must copy it. This is what
//! makes delivering a message free of allocation, and it is the one sharp edge
//! in the interface.
//!
//! Time is whatever monotonic millisecond count the caller has. The connection
//! never asks what time it is; it is told, at the three points where it could
//! matter -- when bytes went out, when bytes came in, and when the caller wants
//! timers serviced.

const std = @import("std");
const Allocator = std.mem.Allocator;

const protocol = @import("protocol.zig");
const packet = protocol.packet;
const topic_rules = protocol.topic;
const session_format = @import("session.zig");

const Packet = packet.Packet;
const PacketId = protocol.types.PacketId;
const Properties = protocol.properties.Properties;
const Qos = protocol.types.Qos;
const ReasonCode = protocol.types.ReasonCode;
const Version = protocol.types.Version;

const Connection = @This();

/// How the connection is set up, decided once and never renegotiated.
pub const Options = struct {
    /// Which version to speak. A broker that does not support 5.0 refuses the
    /// CONNECT with `unsupported_protocol_version`, at which point a client
    /// that wants to fall back builds a new `Connection` with `.v3_1_1`.
    version: Version = .v5_0,

    /// The largest packet this client will accept. A broker that sends a larger
    /// one has broken the limit it was told about in the CONNECT, and the
    /// connection fails rather than growing the receive buffer to fit.
    ///
    /// In 3.1.1 there is no way to tell the broker, so this is enforced locally
    /// and a broker that exceeds it is simply disconnected from.
    max_incoming_packet_size: u32 = 1 << 20,

    /// How many QoS 1 and QoS 2 messages the broker may have in flight towards
    /// this client at once. Sent as the Receive Maximum property; 3.1.1 has no
    /// equivalent and brokers speaking it are unbounded.
    receive_maximum: u16 = 20,

    /// How many topic aliases this client will accept from the broker. Zero
    /// means none, and a broker that sends one anyway is a protocol error.
    incoming_topic_alias_maximum: u16 = 8,

    /// How many topic aliases this client is willing to establish for its own
    /// publications, capped by what the broker says it will accept. Zero -- the
    /// default -- means topics are always sent in full, which is what a client
    /// that publishes to a handful of topics wants; a client publishing to one
    /// topic thousands of times a second wants this turned on.
    outgoing_topic_alias_maximum: u16 = 0,

    /// How long the broker should keep this session after the connection ends,
    /// in seconds, sent as the Session Expiry Interval property.
    ///
    /// Null sends nothing, 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. That is the specification's default and so it
    /// is this one, but it is almost never what a client asking for
    /// `clean_start = false` meant: without it the broker throws the session
    /// away the moment the socket closes, every reconnection reports
    /// `session_present` false, and the unacknowledged messages this end is
    /// holding have nowhere to go. Measured against mosquitto, which does
    /// exactly that.
    ///
    /// So anything that persists or resumes a session wants a value here, and
    /// wants it longer than the outage it expects to survive.
    /// `0xFFFF_FFFF` means the session does not expire at all.
    ///
    /// 3.1.1 has no equivalent: `clean_session = 0` asks the broker to keep the
    /// session and the broker decides for how long, so this is ignored there.
    session_expiry_interval: ?u32 = null,

    /// How long after a PINGREQ the broker may go without answering before the
    /// connection is declared dead, as a multiple of the keep alive interval.
    /// The specification says nothing about this -- it describes only the
    /// broker's duty to disconnect a silent client -- so the number is a
    /// judgement rather than a rule.
    ping_response_factor: f32 = 1.0,
};

/// Where the session has got to.
pub const State = enum {
    /// Nothing sent yet. `connect` is the only thing to do.
    idle,
    /// CONNECT has been queued or sent, and the CONNACK has not arrived.
    connecting,
    /// The broker accepted the connection.
    connected,
    /// DISCONNECT has been queued. The caller should flush and close.
    disconnecting,
    /// Over, either because the broker refused or closed, or because something
    /// went wrong. Session state survives for a reconnection.
    closed,
};

/// What went wrong.
pub const Error = error{
    /// The bytes from the broker do not form a packet.
    MalformedPacket,
    /// The broker said something the protocol forbids -- including, for
    /// example, acknowledging a packet identifier that was never sent.
    ProtocolError,
    /// An incoming packet is larger than `Options.max_incoming_packet_size`, or
    /// an outgoing one larger than the broker said it would accept.
    PacketTooLarge,
    /// The packet cannot be expressed in the version being spoken.
    UnsupportedByVersion,
    /// A command that requires an established connection was issued before
    /// CONNACK, or after the connection ended.
    NotConnected,
    /// `connect` on a connection that is already connecting or connected.
    AlreadyConnected,
    /// The broker's CONNACK gave a Maximum QoS below the one asked for.
    QosNotSupported,
    /// The broker's CONNACK said it does not retain messages.
    RetainNotSupported,
    /// Every one of the 65535 packet identifiers is in use. Only reachable by
    /// a client with a very large Receive Maximum and a very slow broker.
    PacketIdentifiersExhausted,
    /// The broker did not answer a PINGREQ in time.
    KeepAliveTimeout,
    /// The broker used a topic alias above the maximum it was given, or one it
    /// had not established.
    TopicAliasInvalid,
    /// `auth` on a connection whose CONNECT named no authentication method.
    /// There is no exchange to continue.
    NotAuthenticating,
    /// `auth` with a reason code that is not one an endpoint may send:
    /// section 3.15.2.1 allows `continue_authentication` and `reauthenticate`
    /// from a client, and `success` only from the broker.
    BadReasonCode,
    /// `auth` naming a different authentication method from the CONNECT,
    /// which section 4.12 fixes for the life of the exchange.
    AuthenticationMethodChanged,
    OutOfMemory,
} || topic_rules.TopicError;

/// Something the broker did.
///
/// The publish and acknowledgement handshakes do not appear here: by the time a
/// `.message` is handed out its PUBACK or PUBREC has already been queued, and a
/// `.published` means the whole handshake finished. What is left is what an
/// application would want to know about.
pub const Event = union(enum) {
    /// The broker accepted the connection. Carries the CONNACK, whose
    /// properties say what limits the broker is imposing and, if the client
    /// sent an empty identifier, what it has been named.
    connected: packet.ConnAck,
    /// A message. Its topic and payload borrow from the receive buffer and are
    /// valid only until the next call to `nextEvent` or `receive`.
    message: packet.Publish,
    /// A QoS 1 or QoS 2 publication this client sent has finished its
    /// handshake, successfully or otherwise.
    published: Published,
    /// The answer to a `subscribe`, with one reason code per filter.
    subscribed: packet.SubAck,
    /// The answer to an `unsubscribe`.
    unsubscribed: packet.SubAck,
    /// The broker answered a PINGREQ.
    pong,
    /// A step in an extended authentication exchange. The caller is expected to
    /// answer with `auth`. MQTT 5.0 only.
    authenticate: packet.Auth,
    /// The broker refused the connection or closed it, and said why. The
    /// connection is `closed` and the transport should be too.
    disconnected: Disconnected,
};

/// A publication of this client's that has finished.
pub const Published = struct {
    packet_id: PacketId,
    qos: Qos,
    /// What the broker made of it. `no_matching_subscribers` is a success that
    /// is worth knowing about; anything at or above 0x80 is a failure, and the
    /// message was not delivered.
    reason_code: ReasonCode = .success,
    properties: Properties = .empty,

    pub fn succeeded(p: Published) bool {
        return !p.reason_code.isError();
    }
};

/// The end of the connection, as the broker described it.
pub const Disconnected = struct {
    reason_code: ReasonCode,
    properties: Properties = .empty,
    /// Whether this came from a DISCONNECT packet or from a CONNACK that
    /// refused the connection. The distinction matters to a client deciding
    /// whether to retry: a refused CONNECT will usually be refused again.
    origin: Origin,

    pub const Origin = enum {
        /// The broker refused the connection outright.
        connack,
        /// The broker ended an established connection.
        disconnect,
    };
};

/// What to say in the CONNECT packet.
///
/// Nothing here is retained: the strings are copied into the outgoing buffer
/// before `connect` returns, so they may be stack-allocated. A reconnection
/// calls `connect` again and supplies them again.
pub const ConnectOptions = struct {
    client_id: []const u8 = "",
    clean_start: bool = true,
    /// Seconds. Zero disables keepalive entirely, which means a connection to a
    /// broker that has silently gone away is never noticed.
    keep_alive: u16 = 60,
    username: ?[]const u8 = null,
    password: ?[]const u8 = null,
    will: ?packet.Will = null,
    /// Extra properties. Receive Maximum, Maximum Packet Size and Topic Alias
    /// Maximum are filled in from `Options` if they are not set here, so that a
    /// caller that does not care gets limits that match what this connection
    /// will actually enforce.
    properties: Properties = .empty,
};

/// A message to publish.
pub const Message = struct {
    topic: []const u8,
    payload: []const u8 = "",
    qos: Qos = .at_most_once,
    retain: bool = false,
    properties: Properties = .empty,
};

gpa: Allocator,
options: Options,
version: Version,
state: State = .idle,

/// Bytes received from the transport and not yet decoded, plus a cursor. The
/// cursor is what lets a decoded packet keep borrowing from the buffer after
/// it has been handed to the caller: nothing is discarded until the caller asks
/// for the next event.
rx: std.ArrayList(u8) = .empty,
rx_start: usize = 0,
/// The length of the packet most recently returned as an event, which is what
/// `rx_start` advances by on the next call.
rx_held: usize = 0,

/// Bytes to hand to the transport, plus a cursor over what has gone already.
tx: std.ArrayList(u8) = .empty,
tx_start: usize = 0,

next_packet_id: PacketId = 1,

/// Changes whenever the session state does.
///
/// For a caller persisting the session: keep the value from the last save, and
/// there is nothing to write while it still matches. It is generous in one
/// direction only -- it can move when a save would produce the same bytes, and
/// it never fails to move when the bytes would differ -- which is the direction
/// that matters, because the cost of the first is a redundant write and the
/// cost of the second is a lost message.
///
/// Wraps rather than saturating, because a `u64` of changes is not reachable
/// and comparing for equality does not care.
session_serial: u64 = 0,

/// Publications waiting for or undergoing an acknowledgement handshake, oldest
/// first. Bounded by the broker's Receive Maximum plus whatever the caller has
/// queued beyond it, and searched linearly, which is the right shape for the
/// tens of entries this holds in practice.
outbound: std.ArrayList(Outbound) = .empty,

/// Packet identifiers of QoS 2 messages received and not yet released, which is
/// how a retransmitted PUBLISH is recognised and not delivered twice.
inbound_qos2: std.AutoHashMapUnmanaged(PacketId, void) = .empty,

/// Topic aliases the broker has established, and the topics they stand for.
/// The topics are owned, because the PUBLISH that introduced the alias is long
/// out of the receive buffer by the time the alias is used.
incoming_aliases: std.AutoHashMapUnmanaged(u16, []u8) = .empty,

/// Topic aliases this client has established. Keys are owned.
outgoing_aliases: std.StringHashMapUnmanaged(u16) = .empty,
next_outgoing_alias: u16 = 1,

// -- what the broker said in its CONNACK -------------------------------------

/// How many QoS 1 and 2 publications may be in flight towards the broker.
/// 65535 until told otherwise, which is the specification's default.
server_receive_maximum: u16 = 65535,
server_maximum_packet_size: ?u32 = null,
server_maximum_qos: Qos = .exactly_once,
server_retain_available: bool = true,
server_topic_alias_maximum: u16 = 0,

/// The keep alive actually in force: what was asked for, unless the broker's
/// CONNACK overrode it with a Server Keep Alive.
keep_alive: u16 = 0,
last_sent_ms: i64 = 0,
last_received_ms: i64 = 0,
ping_sent_ms: i64 = 0,
ping_outstanding: bool = false,

/// The authentication method named in the CONNECT, owned, or null when this
/// connection asked for no extended exchange.
///
/// Owned rather than borrowed because it outlives the call that supplied it and
/// is compared against every AUTH for the life of the connection -- section
/// 4.12 fixes the method for the whole exchange, and a caller that rebuilt its
/// mechanism between attempts would otherwise leave this pointing at a name
/// that had been freed.
auth_method: ?[]u8 = null,

/// One publication in flight.
const Outbound = struct {
    packet_id: PacketId,
    qos: Qos,
    stage: Stage,
    /// The encoded PUBLISH, kept so that it can be sent again with the
    /// duplicate flag set if the session is resumed. Freed as soon as it can no
    /// longer be needed -- on PUBACK for QoS 1, and on PUBREC for QoS 2, since
    /// after that the handshake is carried by PUBREL and PUBCOMP alone.
    bytes: []u8,
    /// Set when a resumed session owes this exchange a PUBREL again, which is
    /// the case for a QoS 2 message whose PUBCOMP never arrived. It is a flag
    /// rather than an immediate `enqueue` so that the resend goes out in list
    /// order along with the publications being sent again.
    resend_pubrel: bool = false,

    const Stage = enum {
        /// Held back by flow control; not yet on the wire.
        queued,
        awaiting_puback,
        awaiting_pubrec,
        awaiting_pubcomp,
    };

    fn inFlight(o: Outbound) bool {
        return o.stage != .queued;
    }
};

pub fn init(gpa: Allocator, options: Options) Connection {
    return .{ .gpa = gpa, .options = options, .version = options.version };
}

pub fn deinit(c: *Connection) void {
    c.clearSession();
    if (c.auth_method) |m| c.gpa.free(m);
    c.rx.deinit(c.gpa);
    c.tx.deinit(c.gpa);
    c.outbound.deinit(c.gpa);
    c.inbound_qos2.deinit(c.gpa);
    c.incoming_aliases.deinit(c.gpa);
    c.outgoing_aliases.deinit(c.gpa);
    c.* = undefined;
}

// -- moving bytes ------------------------------------------------------------

/// Hand over bytes read from the transport.
///
/// This invalidates any event still being held from a previous `nextEvent`,
/// because it may move the buffer those borrowed slices point into.
pub fn receive(c: *Connection, bytes: []const u8, now_ms: i64) Error!void {
    c.last_received_ms = now_ms;
    try c.rx.appendSlice(c.gpa, bytes);
}

/// The bytes waiting to be written to the transport. Empty when there is
/// nothing to say.
pub fn pending(c: *const Connection) []const u8 {
    return c.tx.items[c.tx_start..];
}

/// Report that the first `n` bytes of `pending` reached the transport.
///
/// This is where the keepalive clock is reset, because what matters to the
/// broker is when bytes arrived, not when they were encoded.
pub fn sent(c: *Connection, n: usize, now_ms: i64) void {
    std.debug.assert(n <= c.pending().len);
    if (n == 0) return;
    c.tx_start += n;
    c.last_sent_ms = now_ms;
    if (c.tx_start == c.tx.items.len) {
        c.tx.clearRetainingCapacity();
        c.tx_start = 0;
    } else if (c.tx_start > c.tx.items.len / 2) {
        // Compaction is amortised: only once the cursor has passed the halfway
        // point, so a connection dribbling bytes out does not memmove the whole
        // buffer on every write.
        c.tx.replaceRangeAssumeCapacity(0, c.tx_start, &.{});
        c.tx_start = 0;
    }
}

/// Whether there is anything to write.
pub fn wantsWrite(c: *const Connection) bool {
    return c.pending().len != 0;
}

// -- events ------------------------------------------------------------------

/// Decode the next thing the broker said, or null if nothing complete has
/// arrived.
///
/// The returned event borrows from the receive buffer and is invalidated by the
/// next call to this function or to `receive`.
///
/// Packets that need no decision from the caller -- PUBREL, and the PUBACK this
/// client owes for an incoming QoS 1 message -- are handled here and produce no
/// event; the loop keeps going until there is something to report or the buffer
/// runs out.
pub fn nextEvent(c: *Connection) Error!?Event {
    while (true) {
        c.releaseHeld();
        const bytes = c.rx.items[c.rx_start..];

        const header = packet.peekFixedHeader(bytes) catch |err| switch (err) {
            error.Incomplete => return null,
            else => |e| return c.fail(e),
        };
        if (header.remaining_length > c.options.max_incoming_packet_size) {
            return c.fail(error.PacketTooLarge);
        }
        if (bytes.len < header.packetLen()) return null;

        const body = bytes[header.encodedLen()..header.packetLen()];
        const p = Packet.decode(header, body, c.version) catch |err| switch (err) {
            // `decode` is handed a complete body, so it cannot want more.
            error.Incomplete => unreachable,
            else => |e| return c.fail(e),
        };
        c.rx_held = header.packetLen();

        if (try c.handle(p)) |event| return event;
    }
}

/// Discard the packet handed out by the previous call, now that the caller has
/// had its turn with it.
fn releaseHeld(c: *Connection) void {
    c.rx_start += c.rx_held;
    c.rx_held = 0;
    if (c.rx_start == c.rx.items.len) {
        c.rx.clearRetainingCapacity();
        c.rx_start = 0;
    } else if (c.rx_start > c.rx.items.len / 2) {
        c.rx.replaceRangeAssumeCapacity(0, c.rx_start, &.{});
        c.rx_start = 0;
    }
}

/// End the connection because of `err`, so that a caller that ignores the error
/// and calls back in does not carry on as though nothing happened.
fn fail(c: *Connection, err: Error) Error {
    c.state = .closed;
    return err;
}

fn handle(c: *Connection, p: Packet) Error!?Event {
    // Section 3.2: the first packet from the broker must be a CONNACK
    // [MQTT-3.2.0-1]. An AUTH may come first instead, and only then, because
    // section 4.12 makes the extended authentication exchange part of the
    // handshake rather than something that follows it.
    //
    // This is not pedantry. Without it a broker could deliver a message before
    // the connection was established, into a session whose limits are not yet
    // known -- and `Client.connect`, which has nothing sensible to do with a
    // message, would have to handle one.
    if (c.state == .connecting and p != .connack and p != .auth) {
        return c.fail(error.ProtocolError);
    }

    return switch (p) {
        .connack => |ack| try c.handleConnack(ack),
        .publish => |m| try c.handlePublish(m),
        .puback => |a| try c.handlePuback(a),
        .pubrec => |a| try c.handlePubrec(a),
        .pubrel => |a| try c.handlePubrel(a),
        .pubcomp => |a| try c.handlePubcomp(a),
        .suback => |a| .{ .subscribed = a },
        .unsuback => |a| .{ .unsubscribed = a },
        .pingresp => blk: {
            c.ping_outstanding = false;
            break :blk .pong;
        },
        .disconnect => |d| blk: {
            c.state = .closed;
            break :blk .{ .disconnected = .{
                .reason_code = d.reason_code,
                .properties = d.properties,
                .origin = .disconnect,
            } };
        },
        .auth => |a| try c.handleAuth(a),
        // Everything else is a packet only a broker receives. A broker sending
        // one has misunderstood which end it is.
        .connect, .subscribe, .unsubscribe, .pingreq, .reserved => c.fail(error.ProtocolError),
    };
}

// -- commands ----------------------------------------------------------------

/// Queue a CONNECT.
///
/// The properties this connection needs the broker to know about -- how big a
/// packet it will accept, how many messages it will take at once, how many
/// topic aliases it will remember -- are filled in from `Options` unless the
/// caller set them, so that what is advertised is what is actually enforced.
pub fn connect(c: *Connection, options: ConnectOptions, now_ms: i64) Error!void {
    if (c.state == .connecting or c.state == .connected) return error.AlreadyConnected;

    if (options.will) |will| try topic_rules.validateName(will.topic);

    // Section 4.12: the method is fixed when the CONNECT goes out and every
    // AUTH after it must name the same one. Kept here so that this can be
    // checked rather than assumed.
    if (c.auth_method) |m| c.gpa.free(m);
    c.auth_method = null;
    if (c.version == .v5_0) {
        if (options.properties.authentication_method) |m| {
            c.auth_method = try c.gpa.dupe(u8, m);
        }
    }
    errdefer if (c.auth_method) |m| {
        c.gpa.free(m);
        c.auth_method = null;
    };

    c.state = .connecting;
    c.keep_alive = options.keep_alive;
    c.last_sent_ms = now_ms;
    c.last_received_ms = now_ms;
    c.ping_outstanding = false;

    // A clean start throws away whatever the last connection left behind,
    // before the CONNECT that asks for it goes out.
    if (options.clean_start) c.clearSession();

    var properties = options.properties;
    if (c.version == .v5_0) {
        if (properties.receive_maximum == null and c.options.receive_maximum != 65535) {
            properties.receive_maximum = c.options.receive_maximum;
        }
        if (properties.maximum_packet_size == null) {
            properties.maximum_packet_size = c.options.max_incoming_packet_size;
        }
        if (properties.topic_alias_maximum == null and c.options.incoming_topic_alias_maximum != 0) {
            properties.topic_alias_maximum = c.options.incoming_topic_alias_maximum;
        }
        if (properties.session_expiry_interval == null) {
            if (c.options.session_expiry_interval) |seconds| {
                properties.session_expiry_interval = seconds;
            }
        }
        // Whatever ends up in the packet is what will be enforced when an alias
        // arrives. A caller that set the property by hand would otherwise have
        // advertised one limit while the connection went on rejecting anything
        // above another, which is the same failure as advertising a Receive
        // Maximum that is not the one in force.
        c.options.incoming_topic_alias_maximum = properties.topic_alias_maximum orelse 0;
    }

    try c.enqueue(.{ .connect = .{
        .client_id = options.client_id,
        .clean_start = options.clean_start,
        .keep_alive = options.keep_alive,
        .username = options.username,
        .password = options.password,
        .will = options.will,
        .properties = properties,
    } });
}

/// Publish a message.
///
/// Returns the packet identifier for QoS 1 and QoS 2, which is what a later
/// `.published` event will be tagged with, and null for QoS 0, which is never
/// acknowledged and so has nothing to wait for.
///
/// A QoS 1 or 2 message beyond the broker's Receive Maximum is queued rather
/// than refused, and goes out as earlier ones are acknowledged. There is no
/// bound on that queue: a caller publishing faster than the broker acknowledges
/// is the one holding the information about what should give way.
pub fn publish(c: *Connection, message: Message) Error!?PacketId {
    if (c.state != .connected) return error.NotConnected;
    try topic_rules.validateName(message.topic);
    if (@backingInt(message.qos) > @backingInt(c.server_maximum_qos)) return error.QosNotSupported;
    if (message.retain and !c.server_retain_available) return error.RetainNotSupported;

    var p: packet.Publish = .{
        .topic = message.topic,
        .payload = message.payload,
        .qos = message.qos,
        .retain = message.retain,
        .properties = message.properties,
    };
    // Aliasing rewrites the packet, so it has to happen before the length is
    // measured against the broker's maximum packet size.
    if (c.version == .v5_0) {
        if (p.properties.topic_alias) |alias| {
            // The caller chose the alias. Sending one above what the broker
            // said it would accept is a protocol error it will disconnect over,
            // so it is refused here where it is still a returned error.
            if (alias == 0 or alias > c.server_topic_alias_maximum) return error.TopicAliasInvalid;
        } else {
            try c.applyOutgoingAlias(&p);
        }
    }

    if (message.qos == .at_most_once) {
        try c.enqueue(.{ .publish = p });
        return null;
    }

    const id = try c.takePacketId();
    p.packet_id = id;

    const bytes = try c.encodeAlloc(.{ .publish = p });
    {
        // Scoped, because once the append succeeds the list owns `bytes`. An
        // `errdefer` covering the whole function would free them again if
        // `pumpQueued` then ran out of memory, leaving an entry in the list
        // pointing at memory that had been handed back.
        errdefer c.gpa.free(bytes);
        c.touchSession();
        try c.outbound.append(c.gpa, .{
            .packet_id = id,
            .qos = message.qos,
            .stage = .queued,
            .bytes = bytes,
        });
    }
    try c.pumpQueued();
    return id;
}

/// Subscribe to one or more filters. Returns the packet identifier the SUBACK
/// will carry.
pub fn subscribe(
    c: *Connection,
    filters: []const packet.TopicFilter,
    properties: Properties,
) Error!PacketId {
    if (c.state != .connected) return error.NotConnected;
    if (filters.len == 0) return error.UnsupportedByVersion;
    for (filters) |f| try topic_rules.validateFilter(f.filter);

    const id = try c.takePacketId();
    try c.enqueue(.{ .subscribe = .{
        .packet_id = id,
        .filters = filters,
        .properties = properties,
    } });
    return id;
}

/// Unsubscribe from one or more filters. Returns the packet identifier the
/// UNSUBACK will carry.
pub fn unsubscribe(
    c: *Connection,
    filters: []const []const u8,
    properties: Properties,
) Error!PacketId {
    if (c.state != .connected) return error.NotConnected;
    if (filters.len == 0) return error.UnsupportedByVersion;
    for (filters) |f| try topic_rules.validateFilter(f);

    const id = try c.takePacketId();
    try c.enqueue(.{ .unsubscribe = .{
        .packet_id = id,
        .filters = filters,
        .properties = properties,
    } });
    return id;
}

/// Queue a DISCONNECT and stop accepting commands.
///
/// The caller should flush `pending` before closing the transport: a
/// DISCONNECT that never arrives leaves the broker to publish the will message,
/// which is the opposite of what a clean shutdown means.
pub fn disconnect(c: *Connection, reason_code: ReasonCode, properties: Properties) Error!void {
    if (c.state != .connected and c.state != .connecting) return error.NotConnected;
    try c.enqueue(.{ .disconnect = .{ .reason_code = reason_code, .properties = properties } });
    c.state = .disconnecting;
}

/// Queue an AUTH, continuing an extended authentication exchange. MQTT 5.0 only.
///
/// The authentication method is filled in from what the CONNECT named, because
/// section 4.12 requires every AUTH to carry it and there is exactly one right
/// answer. A caller that sets it to something else has changed the method
/// mid-exchange, which is what the rule exists to forbid.
///
/// `reason_code` is `continue_authentication` to answer a challenge, or
/// `reauthenticate` to start the exchange again on a connection that is
/// already up (section 4.12.1).
pub fn auth(c: *Connection, reason_code: ReasonCode, properties: Properties) Error!void {
    if (c.state != .connected and c.state != .connecting) return error.NotConnected;
    const method = c.auth_method orelse return error.NotAuthenticating;
    switch (reason_code) {
        .continue_authentication => {},
        // Section 4.12.1: only meaningful once the connection is up.
        .reauthenticate => if (c.state != .connected) return error.NotConnected,
        else => return error.BadReasonCode,
    }
    var p = properties;
    if (p.authentication_method) |named| {
        if (!std.mem.eql(u8, named, method)) return error.AuthenticationMethodChanged;
    } else {
        p.authentication_method = method;
    }
    try c.enqueue(.{ .auth = .{ .reason_code = reason_code, .properties = p } });
}

/// Queue a PINGREQ regardless of whether one is due. `tick` sends these on its
/// own; this is for a caller that wants to probe the connection now.
pub fn ping(c: *Connection, now_ms: i64) Error!void {
    if (c.state != .connected) return error.NotConnected;
    try c.enqueue(.pingreq);
    if (!c.ping_outstanding) {
        c.ping_outstanding = true;
        c.ping_sent_ms = now_ms;
    }
}

// -- timers ------------------------------------------------------------------

/// Service the clock: send a PINGREQ if one is due, and give up if the broker
/// has stopped answering them.
///
/// Calling this more often than necessary is harmless. Calling it not at all
/// means the broker will eventually disconnect this client for being silent.
pub fn tick(c: *Connection, now_ms: i64) Error!void {
    if (c.state != .connected) return;
    if (c.keep_alive == 0) return;

    const interval_ms: i64 = @as(i64, c.keep_alive) * 1000;

    if (c.ping_outstanding) {
        const grace: i64 = @intFromFloat(@as(f32, @floatFromInt(interval_ms)) * c.options.ping_response_factor);
        if (now_ms - c.ping_sent_ms > grace) return c.fail(error.KeepAliveTimeout);
        return;
    }

    if (now_ms - c.last_sent_ms >= interval_ms) try c.ping(now_ms);
}

/// When `tick` next wants to be called, as an absolute time on the caller's
/// clock, or null if there is nothing to wait for.
///
/// A caller with an event loop uses this as the timeout on its read; a caller
/// happy to block on the transport can ignore it, since the PINGREQ that falls
/// due will be answered and so will wake the read anyway.
pub fn deadline(c: *const Connection) ?i64 {
    if (c.state != .connected or c.keep_alive == 0) return null;
    const interval_ms: i64 = @as(i64, c.keep_alive) * 1000;
    if (c.ping_outstanding) {
        const grace: i64 = @intFromFloat(@as(f32, @floatFromInt(interval_ms)) * c.options.ping_response_factor);
        return c.ping_sent_ms + grace;
    }
    return c.last_sent_ms + interval_ms;
}

/// The transport went away without a DISCONNECT.
///
/// Throws away the half-read packet and anything queued to write, and keeps the
/// session: the unacknowledged publications, the packet identifier counter, and
/// the record of which QoS 2 messages have been received. Reconnecting with
/// `clean_start = false` then resumes where this left off, and the broker
/// confirming the session in its CONNACK is what causes the unacknowledged
/// publications to be sent again.
pub fn transportClosed(c: *Connection) void {
    c.state = .closed;
    c.rx.clearRetainingCapacity();
    c.rx_start = 0;
    c.rx_held = 0;
    c.tx.clearRetainingCapacity();
    c.tx_start = 0;
    c.ping_outstanding = false;

    // Aliases are a property of one network connection, not of the session, and
    // both ends forget them when it ends.
    c.clearIncomingAliases();
    c.clearOutgoingAliases();

    // Anything on the wire is back to being merely queued, so that resuming the
    // session sends it again in order. That is session state, so a caller
    // persisting it has something new to write.
    c.touchSession();
    for (c.outbound.items) |*o| {
        if (o.stage == .awaiting_puback) o.stage = .queued;
        if (o.stage == .awaiting_pubrec) o.stage = .queued;
    }
}

/// How many publications are waiting for an acknowledgement or for a window to
/// go out in. Mostly useful to a caller deciding whether to publish more.
pub fn inFlightCount(c: *const Connection) usize {
    return c.outbound.items.len;
}

/// Whether bytes of an unfinished packet are being held.
///
/// The question a transport layer asks when the stream ends: between packets is
/// a conversation that finished, and part-way through one is a conversation
/// that was cut off.
pub fn hasPartialPacket(c: *const Connection) bool {
    return c.rx.items.len != c.rx_start;
}

// -- incoming packets --------------------------------------------------------

fn handleConnack(c: *Connection, ack: packet.ConnAck) Error!?Event {
    if (c.state != .connecting) return c.fail(error.ProtocolError);

    // Section 4.12: a CONNACK that accepts a connection whose CONNECT named an
    // authentication method must name the same method. Refusing a success that
    // does not is not pedantry -- it is the whole of what stops the exchange
    // being skipped. A broker that simply ignored the method and said yes
    // would otherwise leave a client believing it had authenticated by a
    // mechanism that never ran.
    if (!ack.reason_code.isError()) {
        if (c.auth_method) |method| {
            const named = ack.properties.authentication_method orelse
                return c.fail(error.ProtocolError);
            if (!std.mem.eql(u8, named, method)) return c.fail(error.ProtocolError);
        }
    }

    if (ack.reason_code.isError()) {
        c.state = .closed;
        return .{ .disconnected = .{
            .reason_code = ack.reason_code,
            .properties = ack.properties,
            .origin = .connack,
        } };
    }

    const p = ack.properties;
    if (p.receive_maximum) |v| c.server_receive_maximum = v;
    if (p.maximum_packet_size) |v| c.server_maximum_packet_size = v;
    if (p.maximum_qos) |v| c.server_maximum_qos = v;
    if (p.retain_available) |v| c.server_retain_available = v;
    if (p.topic_alias_maximum) |v| c.server_topic_alias_maximum = v;
    // Section 3.2.2.3.14: a Server Keep Alive replaces what the client asked
    // for, and the client must use it.
    if (p.server_keep_alive) |v| c.keep_alive = v;

    c.state = .connected;

    if (!ack.session_present) {
        // A fresh session: whatever was left over belongs to a session the
        // broker has forgotten, and sending it would be sending it into
        // nothing.
        c.clearSession();
    } else {
        // Section 4.4: everything unacknowledged goes again, in the order it
        // was first sent, with the duplicate flag set. Both kinds of debt are
        // only marked here; `pumpQueued` is what puts them on the wire, so that
        // a PUBLISH and a PUBREL owed by the same session go out in the order
        // the messages were first published rather than in two separate passes.
        c.touchSession();
        for (c.outbound.items) |*o| {
            switch (o.stage) {
                .awaiting_puback, .awaiting_pubrec, .queued => o.stage = .queued,
                // The PUBLISH is long gone; what is owed is the PUBREL.
                .awaiting_pubcomp => o.resend_pubrel = true,
            }
        }
    }
    try c.pumpQueued();

    return .{ .connected = ack };
}

/// Section 4.12, which is the whole of what makes an AUTH meaningful.
///
/// An AUTH carries no packet identifier and nothing that ties it to anything,
/// so the only thing distinguishing a step of an exchange from a packet
/// somebody made up is that the exchange was asked for and the method matches.
fn handleAuth(c: *Connection, a: packet.Auth) Error!?Event {
    // An AUTH belongs to an exchange the CONNECT started. A broker sending one
    // to a client that named no method has invented a conversation, and a
    // caller has nothing to answer it with.
    const method = c.auth_method orelse return c.fail(error.ProtocolError);

    // The method is fixed for the life of the exchange, so a broker naming a
    // different one -- or none -- is either confused or trying to move the
    // client onto something weaker.
    const named = a.properties.authentication_method orelse return c.fail(error.ProtocolError);
    if (!std.mem.eql(u8, named, method)) return c.fail(error.ProtocolError);

    switch (a.reason_code) {
        // The ordinary step: the broker wants another message.
        .continue_authentication => {},
        // Section 4.12.1 ends a *re-authentication* with this. The first
        // exchange ends with a CONNACK instead, so a success while still
        // connecting is a broker skipping the CONNACK -- and a client that
        // took it would consider itself connected without ever having been
        // told the session limits.
        .success => if (c.state != .connected) return c.fail(error.ProtocolError),
        // Only a client sends this, to ask for re-authentication. A broker
        // sending it has the direction backwards.
        .reauthenticate => return c.fail(error.ProtocolError),
        else => return c.fail(error.ProtocolError),
    }
    return .{ .authenticate = a };
}

fn handlePublish(c: *Connection, incoming: packet.Publish) Error!?Event {
    var message = incoming;

    if (c.version == .v5_0) {
        c.resolveIncomingAlias(&message) catch |err| switch (err) {
            // Running out of memory says nothing about the broker, and leaves
            // the connection perfectly able to carry on if the caller can find
            // some. Everything else is the broker breaking the protocol.
            error.OutOfMemory => return err,
            else => return c.fail(err),
        };
    }

    switch (message.qos) {
        .at_most_once => return .{ .message = message },
        .at_least_once => {
            const id = message.packet_id orelse return c.fail(error.ProtocolError);
            try c.enqueue(.{ .puback = .{ .packet_id = id } });
            return .{ .message = message };
        },
        .exactly_once => {
            const id = message.packet_id orelse return c.fail(error.ProtocolError);
            c.touchSession();
            const entry = try c.inbound_qos2.getOrPut(c.gpa, id);
            try c.enqueue(.{ .pubrec = .{ .packet_id = id } });
            // Already held: this is a retransmission of something delivered the
            // first time, and delivering it again is exactly what QoS 2 exists
            // to prevent.
            if (entry.found_existing) return null;
            return .{ .message = message };
        },
    }
}

fn handlePuback(c: *Connection, ack: packet.Ack) Error!?Event {
    c.touchSession();
    const index = c.findOutbound(ack.packet_id, .awaiting_puback) orelse
        return c.fail(error.ProtocolError);
    const o = c.outbound.orderedRemove(index);
    c.gpa.free(o.bytes);
    try c.pumpQueued();
    return .{ .published = .{
        .packet_id = ack.packet_id,
        .qos = .at_least_once,
        .reason_code = ack.reason_code,
        .properties = ack.properties,
    } };
}

fn handlePubrec(c: *Connection, ack: packet.Ack) Error!?Event {
    c.touchSession();
    const index = c.findOutbound(ack.packet_id, .awaiting_pubrec) orelse
        return c.fail(error.ProtocolError);
    const o = &c.outbound.items[index];

    // The PUBLISH cannot be needed again either way: the broker has it, or has
    // refused it.
    c.gpa.free(o.bytes);
    o.bytes = &.{};

    if (ack.reason_code.isError()) {
        // Section 4.3.3: a PUBREC that reports a failure ends the handshake
        // there. No PUBREL, no PUBCOMP.
        const id = ack.packet_id;
        _ = c.outbound.orderedRemove(index);
        try c.pumpQueued();
        return .{ .published = .{
            .packet_id = id,
            .qos = .exactly_once,
            .reason_code = ack.reason_code,
            .properties = ack.properties,
        } };
    }

    o.stage = .awaiting_pubcomp;
    try c.enqueue(.{ .pubrel = .{ .packet_id = ack.packet_id } });
    return null;
}

fn handlePubrel(c: *Connection, ack: packet.Ack) Error!?Event {
    c.touchSession();
    // The broker is releasing a message this client received at QoS 2. Whether
    // or not it is still held -- a PUBCOMP that was lost means this arrives
    // again after the identifier was forgotten -- the answer is the same.
    _ = c.inbound_qos2.remove(ack.packet_id);
    try c.enqueue(.{ .pubcomp = .{ .packet_id = ack.packet_id } });
    return null;
}

fn handlePubcomp(c: *Connection, ack: packet.Ack) Error!?Event {
    c.touchSession();
    const index = c.findOutbound(ack.packet_id, .awaiting_pubcomp) orelse
        return c.fail(error.ProtocolError);
    const o = c.outbound.orderedRemove(index);
    std.debug.assert(o.bytes.len == 0);
    try c.pumpQueued();
    return .{ .published = .{
        .packet_id = ack.packet_id,
        .qos = .exactly_once,
        .reason_code = ack.reason_code,
        .properties = ack.properties,
    } };
}

// -- flow control and identifiers --------------------------------------------

fn findOutbound(c: *Connection, id: PacketId, stage: Outbound.Stage) ?usize {
    for (c.outbound.items, 0..) |o, i| {
        if (o.packet_id == id and o.stage == stage) return i;
    }
    return null;
}

/// Send as many queued publications as the broker's Receive Maximum allows.
///
/// Oldest first, which is what makes a resumed session replay in order, and
/// what keeps a slow broker from starving the earliest message indefinitely.
fn pumpQueued(c: *Connection) Error!void {
    if (c.state != .connected) return;

    var in_flight: usize = 0;
    for (c.outbound.items) |o| {
        if (o.inFlight()) in_flight += 1;
    }

    for (c.outbound.items) |*o| {
        // A PUBREL a resumed session owes goes out whatever the window says:
        // its exchange is already counted against the window, and finishing it
        // is what frees the space.
        if (o.resend_pubrel) {
            c.touchSession();
            try c.enqueue(.{ .pubrel = .{ .packet_id = o.packet_id } });
            o.resend_pubrel = false;
            continue;
        }
        if (o.stage != .queued) continue;
        if (in_flight >= c.server_receive_maximum) continue;

        // A resend carries the duplicate flag. The first send does not, and
        // `bytes` was encoded without it, so setting the bit is a rewrite of
        // one nibble of the fixed header rather than a re-encode.
        c.touchSession();
        try c.writeRaw(o.bytes);
        o.bytes[0] |= 0x08;

        o.stage = switch (o.qos) {
            .at_most_once => unreachable,
            .at_least_once => .awaiting_puback,
            .exactly_once => .awaiting_pubrec,
        };
        in_flight += 1;
    }
}

/// The next unused packet identifier.
///
/// Zero is not a valid identifier, so the counter skips it when it wraps. The
/// scan for a collision is linear over the in-flight list, which is bounded by
/// the broker's Receive Maximum.
fn takePacketId(c: *Connection) Error!PacketId {
    var tried: u32 = 0;
    while (tried <= std.math.maxInt(PacketId)) : (tried += 1) {
        const id = c.next_packet_id;
        c.next_packet_id = if (id == std.math.maxInt(PacketId)) 1 else id + 1;
        c.touchSession();

        var in_use = false;
        for (c.outbound.items) |o| {
            if (o.packet_id == id) in_use = true;
        }
        if (!in_use) return id;
    }
    return error.PacketIdentifiersExhausted;
}

// -- topic aliases -----------------------------------------------------------

/// Replace a repeated topic with the alias the broker has been told to use for
/// it, or establish a new alias if there is room.
///
/// The policy is the simplest one that is correct: hand out aliases until the
/// broker's limit is reached, then stop. Nothing is evicted, because an
/// eviction would have to be announced with a full-topic PUBLISH anyway, and a
/// client publishing to more distinct topics than the broker will alias is
/// better served by not aliasing than by thrashing.
fn applyOutgoingAlias(c: *Connection, p: *packet.Publish) Error!void {
    const limit = @min(c.options.outgoing_topic_alias_maximum, c.server_topic_alias_maximum);
    if (limit == 0) return;

    if (c.outgoing_aliases.get(p.topic)) |alias| {
        // Established: the topic is now redundant, and leaving it out is the
        // whole point.
        p.properties.topic_alias = alias;
        p.topic = "";
        return;
    }

    if (c.next_outgoing_alias > limit) return;

    const alias = c.next_outgoing_alias;
    const owned = try c.gpa.dupe(u8, p.topic);
    errdefer c.gpa.free(owned);
    try c.outgoing_aliases.put(c.gpa, owned, alias);
    c.next_outgoing_alias += 1;

    // The packet that establishes an alias carries both the topic and the
    // alias; only later ones may leave the topic out.
    p.properties.topic_alias = alias;
}

/// Resolve an incoming topic alias, rewriting `p.topic` to the topic the alias
/// stands for.
///
/// Every failure here is a protocol error on the broker's part, and the caller
/// closes the connection over it. That it *is* the caller who closes, rather
/// than this function, was a bug the fuzzer found: written to hand the error
/// back as a value, it threw it instead, and a broker sending an alias it had
/// never established left the connection reporting an error and still calling
/// itself connected.
fn resolveIncomingAlias(c: *Connection, p: *packet.Publish) Error!void {
    const alias = p.properties.topic_alias orelse {
        // Section 3.3.2.1: with no alias, the topic is the only thing that says
        // where the message goes, and it may not be empty.
        if (p.topic.len == 0) return error.ProtocolError;
        return;
    };
    if (alias > c.options.incoming_topic_alias_maximum) return error.TopicAliasInvalid;

    if (p.topic.len != 0) {
        // Establishing, or re-establishing, the alias.
        const entry = try c.incoming_aliases.getOrPut(c.gpa, alias);
        if (entry.found_existing) c.gpa.free(entry.value_ptr.*);
        entry.value_ptr.* = c.gpa.dupe(u8, p.topic) catch |err| {
            // Leaving a live entry pointing at freed memory would be worse than
            // losing the alias.
            _ = c.incoming_aliases.remove(alias);
            return err;
        };
        return;
    }

    p.topic = c.incoming_aliases.get(alias) orelse return error.TopicAliasInvalid;
}

fn clearIncomingAliases(c: *Connection) void {
    var it = c.incoming_aliases.valueIterator();
    while (it.next()) |topic| c.gpa.free(topic.*);
    c.incoming_aliases.clearRetainingCapacity();
}

fn clearOutgoingAliases(c: *Connection) void {
    var it = c.outgoing_aliases.keyIterator();
    while (it.next()) |topic| c.gpa.free(topic.*);
    c.outgoing_aliases.clearRetainingCapacity();
    c.next_outgoing_alias = 1;
}

/// Throw away everything that belongs to the session rather than to the
/// connection.
/// Note that the session has changed. See `session_serial`.
fn touchSession(c: *Connection) void {
    c.session_serial +%= 1;
}

/// Write the session down, so that it can outlive this process.
///
/// `client_id` is what will be sent in the CONNECT that resumes it. MQTT
/// identifies a session by the client identifier and nothing else, so it is
/// written into the file and checked when it is read back; an empty one -- a
/// client that lets the broker choose its name -- is `SessionNotResumable`,
/// because such a client gets a different name every time and can never resume
/// anything.
///
/// May be called at any point. What is in the outgoing buffer is not session
/// state and does not need to survive: anything unacknowledged in it is sent
/// again when the session resumes, which is the same guarantee that covers a
/// connection that broke mid-write.
///
/// See `session` for the format and for what is deliberately left out of it.
pub fn saveSession(
    c: *Connection,
    w: *std.Io.Writer,
    client_id: []const u8,
) session_format.SaveError!void {
    var stored: std.ArrayList(session_format.StoredOutbound) = .empty;
    defer stored.deinit(c.gpa);
    try stored.ensureTotalCapacity(c.gpa, c.outbound.items.len);
    for (c.outbound.items) |o| stored.appendAssumeCapacity(.{
        .packet_id = o.packet_id,
        .qos = o.qos,
        .stage = switch (o.stage) {
            .queued => .queued,
            .awaiting_puback => .awaiting_puback,
            .awaiting_pubrec => .awaiting_pubrec,
            .awaiting_pubcomp => .awaiting_pubcomp,
        },
        .resend_pubrel = o.resend_pubrel,
        .bytes = o.bytes,
    });

    var ids: std.ArrayList(PacketId) = .empty;
    defer ids.deinit(c.gpa);
    try ids.ensureTotalCapacity(c.gpa, c.inbound_qos2.count());
    var it = c.inbound_qos2.keyIterator();
    while (it.next()) |id| ids.appendAssumeCapacity(id.*);
    // Sorted, because a hash map's iteration order is not a fact about the
    // session: without this the same session writes different bytes on
    // different runs, and a caller comparing what it wrote last would write
    // again forever.
    std.mem.sort(PacketId, ids.items, {}, std.sort.asc(PacketId));

    return session_format.write(c.gpa, w, .{
        .version = c.version,
        .client_id = client_id,
        .next_packet_id = c.next_packet_id,
        .outbound = stored.items,
        .inbound_qos2 = ids.items,
    });
}

pub const LoadSessionError = session_format.LoadError || error{
    /// `loadSession` after the CONNECT has gone out. The session has to be in
    /// place before the broker is asked to resume it.
    AlreadyConnected,
};

/// Read a session back, replacing whatever this connection was holding.
///
/// Before `connect`, and with `clean_start` false -- a CONNECT that asks for a
/// clean start tells the broker to throw away its half of what has just been
/// restored, which leaves the two disagreeing about every message in flight.
///
/// What comes back is what was owed: publications go out again, with the
/// duplicate flag where they had already been sent once, as soon as a CONNACK
/// says the session is still there. A CONNACK that says it is not clears all of
/// this, which is correct and worth expecting -- a broker that forgot the
/// session will not accept the second half of a handshake it has no record of.
pub fn loadSession(
    c: *Connection,
    r: *std.Io.Reader,
    client_id: []const u8,
) LoadSessionError!void {
    if (c.state == .connecting or c.state == .connected) return error.AlreadyConnected;

    var buffer: session_format.Buffer = .{};
    defer buffer.deinit(c.gpa);
    const snapshot = try session_format.read(c.gpa, r, &buffer);

    if (snapshot.version != c.version) return error.SessionVersionMismatch;
    if (!std.mem.eql(u8, snapshot.client_id, client_id)) {
        return error.SessionBelongsToAnotherClient;
    }

    // Room for everything first, so that a failure part way through does not
    // leave half a session behind.
    try c.outbound.ensureTotalCapacity(c.gpa, snapshot.outbound.len);
    try c.inbound_qos2.ensureTotalCapacity(c.gpa, @intCast(snapshot.inbound_qos2.len));

    c.clearSession();
    for (snapshot.outbound) |o| c.outbound.appendAssumeCapacity(.{
        .packet_id = o.packet_id,
        .qos = o.qos,
        .stage = switch (o.stage) {
            .queued => .queued,
            .awaiting_puback => .awaiting_puback,
            .awaiting_pubrec => .awaiting_pubrec,
            .awaiting_pubcomp => .awaiting_pubcomp,
        },
        .bytes = o.bytes,
        .resend_pubrel = o.resend_pubrel,
    });
    // Taken over, so the buffer must not free them.
    buffer.releaseOutbound();

    for (snapshot.inbound_qos2) |id| c.inbound_qos2.putAssumeCapacity(id, {});
    c.next_packet_id = snapshot.next_packet_id;
    c.touchSession();
}

fn clearSession(c: *Connection) void {
    c.touchSession();
    for (c.outbound.items) |o| c.gpa.free(o.bytes);
    c.outbound.clearRetainingCapacity();
    c.inbound_qos2.clearRetainingCapacity();
    c.clearIncomingAliases();
    c.clearOutgoingAliases();
    c.next_packet_id = 1;
}

// -- writing -----------------------------------------------------------------

/// Encode a packet into the outgoing buffer.
///
/// The length is measured first and the capacity reserved, so the writer it is
/// handed cannot run out of room. That turns what would otherwise be a
/// `WriteFailed` -- an allocation failure wearing a disguise -- back into an
/// `OutOfMemory` the caller can act on.
fn enqueue(c: *Connection, p: Packet) Error!void {
    const len = try p.encodedLen(c.version);
    if (c.server_maximum_packet_size) |max| {
        if (len > max) return error.PacketTooLarge;
    }
    try c.tx.ensureUnusedCapacity(c.gpa, len);
    var w: std.Io.Writer = .fixed(c.tx.unusedCapacitySlice());
    p.encode(&w, c.version) catch |err| switch (err) {
        error.WriteFailed => unreachable, // capacity was reserved above
        else => |e| return e,
    };
    c.tx.items.len += w.end;
}

/// Encode a packet into memory of its own, for a publication that may have to
/// be sent more than once.
fn encodeAlloc(c: *Connection, p: Packet) Error![]u8 {
    const len = try p.encodedLen(c.version);
    if (c.server_maximum_packet_size) |max| {
        if (len > max) return error.PacketTooLarge;
    }
    const bytes = try c.gpa.alloc(u8, len);
    errdefer c.gpa.free(bytes);
    var w: std.Io.Writer = .fixed(bytes);
    p.encode(&w, c.version) catch |err| switch (err) {
        error.WriteFailed => unreachable,
        else => |e| return e,
    };
    std.debug.assert(w.end == len);
    return bytes;
}

fn writeRaw(c: *Connection, bytes: []const u8) Error!void {
    try c.tx.appendSlice(c.gpa, bytes);
}

// -- tests -------------------------------------------------------------------

const testing = std.testing;

/// A connection and the scaffolding a test needs around it: somewhere to put
/// the bytes going each way, and a clock that only moves when told.
const Harness = struct {
    conn: Connection,
    /// Bytes taken off the outgoing queue, which is where a decoded outgoing
    /// packet borrows from. Copying them out means marking them sent -- and so
    /// letting the connection compact its buffer -- cannot pull the ground out
    /// from under an assertion.
    scratch: [4096]u8 = undefined,
    now: i64 = 0,

    fn init(options: Options) Harness {
        return .{ .conn = .init(testing.allocator, options) };
    }

    fn deinit(h: *Harness) void {
        h.conn.deinit();
    }

    /// Decode the next packet the connection wants to send, marking it sent.
    fn take(h: *Harness) !?Packet {
        const out = h.conn.pending();
        if (out.len == 0) return null;
        const header = try packet.peekFixedHeader(out);
        const len = header.packetLen();
        @memcpy(h.scratch[0..len], out[0..len]);
        h.conn.sent(len, h.now);
        return try Packet.decode(header, h.scratch[header.encodedLen()..len], h.conn.version);
    }

    /// Assert that exactly nothing is queued to send.
    fn expectQuiet(h: *Harness) !void {
        try testing.expectEqual(@as(usize, 0), h.conn.pending().len);
    }

    /// Hand the connection a packet, as the broker would.
    fn deliver(h: *Harness, p: Packet) !void {
        var buf: [4096]u8 = undefined;
        var w: std.Io.Writer = .fixed(&buf);
        try p.encode(&w, h.conn.version);
        try h.conn.receive(w.buffered(), h.now);
    }

    /// Drive the handshake to the point where publishing is allowed.
    fn connectAndAccept(h: *Harness, ack: packet.ConnAck) !void {
        try h.conn.connect(.{ .client_id = "test", .keep_alive = 60 }, h.now);
        _ = try h.take();
        try h.deliver(.{ .connack = ack });
        const event = (try h.conn.nextEvent()).?;
        try testing.expectEqual(State.connected, h.conn.state);
        try testing.expect(event == .connected);
    }

    fn advance(h: *Harness, ms: i64) void {
        h.now += ms;
    }
};

test "a connection starts idle and refuses commands until it is connected" {
    var h: Harness = .init(.{});
    defer h.deinit();

    try testing.expectEqual(State.idle, h.conn.state);
    try testing.expectError(error.NotConnected, h.conn.publish(.{ .topic = "a" }));
    try testing.expectError(error.NotConnected, h.conn.subscribe(&.{.{ .filter = "a" }}, .empty));

    try h.conn.connect(.{ .client_id = "test" }, h.now);
    try testing.expectEqual(State.connecting, h.conn.state);
    // Still not connected: the CONNACK has not arrived.
    try testing.expectError(error.NotConnected, h.conn.publish(.{ .topic = "a" }));
    try testing.expectError(error.AlreadyConnected, h.conn.connect(.{}, h.now));
}

test "the connect packet advertises the limits this connection enforces" {
    var h: Harness = .init(.{
        .receive_maximum = 5,
        .max_incoming_packet_size = 4096,
        .incoming_topic_alias_maximum = 3,
    });
    defer h.deinit();

    try h.conn.connect(.{ .client_id = "sensor", .keep_alive = 30 }, h.now);
    const p = (try h.take()).?.connect;

    try testing.expectEqualStrings("sensor", p.client_id);
    try testing.expectEqual(@as(u16, 30), p.keep_alive);
    // Advertising a limit the connection would not actually enforce is the
    // failure mode this guards against.
    try testing.expectEqual(@as(u16, 5), p.properties.receive_maximum.?);
    try testing.expectEqual(@as(u32, 4096), p.properties.maximum_packet_size.?);
    try testing.expectEqual(@as(u16, 3), p.properties.topic_alias_maximum.?);
}

test "the broker's connack limits are adopted" {
    var h: Harness = .init(.{});
    defer h.deinit();

    try h.connectAndAccept(.{ .properties = .{
        .receive_maximum = 2,
        .maximum_qos = .at_least_once,
        .retain_available = false,
        .server_keep_alive = 15,
        .maximum_packet_size = 64,
    } });

    try testing.expectEqual(@as(u16, 2), h.conn.server_receive_maximum);
    // Section 3.2.2.3.14: the broker's keep alive replaces what was asked for.
    try testing.expectEqual(@as(u16, 15), h.conn.keep_alive);

    // And the limits are enforced rather than merely recorded.
    try testing.expectError(error.QosNotSupported, h.conn.publish(.{
        .topic = "a",
        .qos = .exactly_once,
    }));
    try testing.expectError(error.RetainNotSupported, h.conn.publish(.{
        .topic = "a",
        .retain = true,
    }));
    try testing.expectError(error.PacketTooLarge, h.conn.publish(.{
        .topic = "a",
        .payload = &@as([128]u8, @splat(0)),
    }));
}

test "a refused connection is a disconnection with a reason" {
    var h: Harness = .init(.{});
    defer h.deinit();

    try h.conn.connect(.{ .client_id = "test" }, h.now);
    _ = try h.take();
    try h.deliver(.{ .connack = .{
        .reason_code = .bad_user_name_or_password,
        .properties = .{ .reason_string = "no" },
    } });

    const event = (try h.conn.nextEvent()).?;
    try testing.expectEqual(ReasonCode.bad_user_name_or_password, event.disconnected.reason_code);
    try testing.expectEqual(Disconnected.Origin.connack, event.disconnected.origin);
    try testing.expectEqualStrings("no", event.disconnected.properties.reason_string.?);
    try testing.expectEqual(State.closed, h.conn.state);
}

test "a qos 0 publish is sent and forgotten" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    try testing.expectEqual(@as(?PacketId, null), try h.conn.publish(.{
        .topic = "a/b",
        .payload = "hello",
    }));
    const p = (try h.take()).?.publish;
    try testing.expectEqualStrings("a/b", p.topic);
    try testing.expectEqualStrings("hello", p.payload);
    try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());
}

test "a qos 1 publish completes on its puback" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    const id = (try h.conn.publish(.{
        .topic = "a/b",
        .payload = "x",
        .qos = .at_least_once,
    })).?;
    const sent_packet = (try h.take()).?.publish;
    try testing.expectEqual(id, sent_packet.packet_id.?);
    try testing.expect(!sent_packet.duplicate);
    try testing.expectEqual(@as(usize, 1), h.conn.inFlightCount());

    try h.deliver(.{ .puback = .{ .packet_id = id, .reason_code = .no_matching_subscribers } });
    const event = (try h.conn.nextEvent()).?;
    try testing.expectEqual(id, event.published.packet_id);
    try testing.expectEqual(ReasonCode.no_matching_subscribers, event.published.reason_code);
    // Delivered to nobody is still delivered, as far as the handshake goes.
    try testing.expect(event.published.succeeded());
    try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());
}

test "a qos 2 publish takes four packets" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    const id = (try h.conn.publish(.{
        .topic = "a/b",
        .payload = "x",
        .qos = .exactly_once,
    })).?;
    try testing.expectEqual(id, (try h.take()).?.publish.packet_id.?);

    // PUBREC gets a PUBREL back, and no event: nothing has finished.
    try h.deliver(.{ .pubrec = .{ .packet_id = id } });
    try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent());
    try testing.expectEqual(id, (try h.take()).?.pubrel.packet_id);
    try testing.expectEqual(@as(usize, 1), h.conn.inFlightCount());

    try h.deliver(.{ .pubcomp = .{ .packet_id = id } });
    const event = (try h.conn.nextEvent()).?;
    try testing.expectEqual(id, event.published.packet_id);
    try testing.expectEqual(Qos.exactly_once, event.published.qos);
    try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());
}

test "a pubrec that reports a failure ends the handshake there" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    const id = (try h.conn.publish(.{ .topic = "a/b", .qos = .exactly_once })).?;
    _ = try h.take();

    try h.deliver(.{ .pubrec = .{ .packet_id = id, .reason_code = .not_authorized } });
    const event = (try h.conn.nextEvent()).?;
    try testing.expect(!event.published.succeeded());
    try testing.expectEqual(ReasonCode.not_authorized, event.published.reason_code);
    // Section 4.3.3: no PUBREL follows a rejected PUBREC.
    try h.expectQuiet();
    try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());
}

test "an acknowledgement for a packet that was never sent is a protocol error" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    try h.deliver(.{ .puback = .{ .packet_id = 999 } });
    try testing.expectError(error.ProtocolError, h.conn.nextEvent());
    // And the connection does not carry on as though nothing happened.
    try testing.expectEqual(State.closed, h.conn.state);
}

test "an incoming qos 1 message is acknowledged before it is handed over" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    try h.deliver(.{ .publish = .{
        .topic = "a/b",
        .payload = "hi",
        .qos = .at_least_once,
        .packet_id = 5,
    } });
    const event = (try h.conn.nextEvent()).?;
    try testing.expectEqualStrings("a/b", event.message.topic);
    try testing.expectEqualStrings("hi", event.message.payload);
    try testing.expectEqual(@as(PacketId, 5), (try h.take()).?.puback.packet_id);
}

test "a retransmitted qos 2 message is acknowledged but not delivered twice" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    const incoming: Packet = .{ .publish = .{
        .topic = "a/b",
        .payload = "once",
        .qos = .exactly_once,
        .packet_id = 7,
    } };

    try h.deliver(incoming);
    try testing.expectEqualStrings("once", (try h.conn.nextEvent()).?.message.payload);
    try testing.expectEqual(@as(PacketId, 7), (try h.take()).?.pubrec.packet_id);

    // The PUBREC was lost, so the broker sends the message again. It must be
    // acknowledged again -- and must not reach the application again, which is
    // the entire purpose of QoS 2.
    try h.deliver(incoming);
    try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent());
    try testing.expectEqual(@as(PacketId, 7), (try h.take()).?.pubrec.packet_id);

    // PUBREL releases it, and the identifier may be used again afterwards.
    try h.deliver(.{ .pubrel = .{ .packet_id = 7 } });
    try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent());
    try testing.expectEqual(@as(PacketId, 7), (try h.take()).?.pubcomp.packet_id);

    try h.deliver(incoming);
    try testing.expectEqualStrings("once", (try h.conn.nextEvent()).?.message.payload);
}

test "a pubrel for a message already forgotten is still answered" {
    // A lost PUBCOMP means the broker sends PUBREL again after the identifier
    // has been released. Answering is what stops it retrying forever.
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    try h.deliver(.{ .pubrel = .{ .packet_id = 42 } });
    try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent());
    try testing.expectEqual(@as(PacketId, 42), (try h.take()).?.pubcomp.packet_id);
}

test "the broker's receive maximum holds publications back" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{ .properties = .{ .receive_maximum = 2 } });

    const first = (try h.conn.publish(.{ .topic = "a", .qos = .at_least_once })).?;
    const second = (try h.conn.publish(.{ .topic = "b", .qos = .at_least_once })).?;
    const third = (try h.conn.publish(.{ .topic = "c", .qos = .at_least_once })).?;

    // All three are accepted; only two are on the wire.
    try testing.expectEqual(@as(usize, 3), h.conn.inFlightCount());
    try testing.expectEqualStrings("a", (try h.take()).?.publish.topic);
    try testing.expectEqualStrings("b", (try h.take()).?.publish.topic);
    try h.expectQuiet();

    // Acknowledging the first opens the window for the third, in order.
    try h.deliver(.{ .puback = .{ .packet_id = first } });
    _ = try h.conn.nextEvent();
    try testing.expectEqualStrings("c", (try h.take()).?.publish.topic);

    try h.deliver(.{ .puback = .{ .packet_id = second } });
    _ = try h.conn.nextEvent();
    try h.deliver(.{ .puback = .{ .packet_id = third } });
    _ = try h.conn.nextEvent();
    try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());
}

test "a resumed session sends the unacknowledged publications again" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    const qos1 = (try h.conn.publish(.{ .topic = "a", .qos = .at_least_once })).?;
    const qos2 = (try h.conn.publish(.{ .topic = "b", .qos = .exactly_once })).?;
    _ = try h.take();
    _ = try h.take();

    // The QoS 2 one gets as far as PUBREC before the connection dies, so what
    // it is owed on resumption is a PUBREL and not the message again.
    try h.deliver(.{ .pubrec = .{ .packet_id = qos2 } });
    _ = try h.conn.nextEvent();
    _ = try h.take();

    h.conn.transportClosed();
    try testing.expectEqual(@as(usize, 2), h.conn.inFlightCount());

    try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now);
    _ = try h.take();
    try h.deliver(.{ .connack = .{ .session_present = true } });
    _ = try h.conn.nextEvent();

    // Section 4.4: in the order they were first sent, and flagged as repeats.
    const again = (try h.take()).?.publish;
    try testing.expectEqualStrings("a", again.topic);
    try testing.expectEqual(qos1, again.packet_id.?);
    try testing.expect(again.duplicate);

    try testing.expectEqual(qos2, (try h.take()).?.pubrel.packet_id);
    try h.expectQuiet();
}

test "a session the broker has forgotten takes its publications with it" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    _ = try h.conn.publish(.{ .topic = "a", .qos = .at_least_once });
    _ = try h.take();
    h.conn.transportClosed();

    try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now);
    _ = try h.take();
    // The broker says it has no session. Sending the message again would be
    // sending it into a session that no longer exists.
    try h.deliver(.{ .connack = .{ .session_present = false } });
    _ = try h.conn.nextEvent();

    try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());
    try h.expectQuiet();
}

test "a clean start throws the session away before the connect goes out" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    _ = try h.conn.publish(.{ .topic = "a", .qos = .at_least_once });
    _ = try h.take();
    h.conn.transportClosed();

    try h.conn.connect(.{ .client_id = "test", .clean_start = true }, h.now);
    try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());
}

test "keepalive sends a ping when the connection has been quiet" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{ .properties = .{ .server_keep_alive = 10 } });

    try h.conn.tick(h.now);
    try h.expectQuiet();
    try testing.expectEqual(h.now + 10_000, h.conn.deadline().?);

    h.advance(10_000);
    try h.conn.tick(h.now);
    try testing.expectEqual(@as(Packet, .pingreq), (try h.take()).?);

    // One ping at a time: until it is answered, more time passing adds nothing.
    h.advance(1);
    try h.conn.tick(h.now);
    try h.expectQuiet();

    try h.deliver(.pingresp);
    try testing.expectEqual(@as(Event, .pong), (try h.conn.nextEvent()).?);
    try testing.expect(!h.conn.ping_outstanding);
}

test "a broker that stops answering pings is given up on" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{ .properties = .{ .server_keep_alive = 10 } });

    h.advance(10_000);
    try h.conn.tick(h.now);
    _ = try h.take();

    h.advance(10_000);
    try h.conn.tick(h.now); // exactly at the deadline is still in time
    h.advance(1);
    try testing.expectError(error.KeepAliveTimeout, h.conn.tick(h.now));
    try testing.expectEqual(State.closed, h.conn.state);
}

test "a keep alive of zero turns the mechanism off entirely" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.conn.connect(.{ .client_id = "test", .keep_alive = 0 }, h.now);
    _ = try h.take();
    try h.deliver(.{ .connack = .{} });
    _ = try h.conn.nextEvent();

    try testing.expectEqual(@as(?i64, null), h.conn.deadline());
    h.advance(1_000_000);
    try h.conn.tick(h.now);
    try h.expectQuiet();
}

test "sending anything at all postpones the next ping" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{ .properties = .{ .server_keep_alive = 10 } });

    h.advance(9_000);
    _ = try h.conn.publish(.{ .topic = "a" });
    _ = try h.take(); // marking it sent is what moves the clock

    h.advance(2_000); // 11 seconds since the CONNACK, 2 since the publish
    try h.conn.tick(h.now);
    try h.expectQuiet();
}

test "subscribe and unsubscribe are answered by their own packet identifiers" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    const sub_id = try h.conn.subscribe(&.{
        .{ .filter = "a/#", .qos = .at_least_once },
        .{ .filter = "b/+" },
    }, .{ .subscription_identifiers = &.{9} });

    const sub = (try h.take()).?.subscribe;
    try testing.expectEqual(sub_id, sub.packet_id);
    try testing.expectEqual(@as(u32, 9), sub.properties.subscriptionIdentifier().?);

    try h.deliver(.{ .suback = .{
        .packet_id = sub_id,
        .reason_codes = &.{ .granted_qos_1, .not_authorized },
    } });
    const suback = (try h.conn.nextEvent()).?.subscribed;
    try testing.expectEqual(Qos.at_least_once, suback.reason_codes[0].grantedQos().?);
    try testing.expect(suback.reason_codes[1].isError());

    const unsub_id = try h.conn.unsubscribe(&.{"a/#"}, .empty);
    try testing.expectEqual(unsub_id, (try h.take()).?.unsubscribe.packet_id);
    try h.deliver(.{ .unsuback = .{ .packet_id = unsub_id, .reason_codes = &.{.success} } });
    try testing.expectEqual(unsub_id, (try h.conn.nextEvent()).?.unsubscribed.packet_id);

    // Identifiers are not reused while something is outstanding.
    try testing.expect(sub_id != unsub_id);
}

test "a bad topic is caught before anything is sent" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    try testing.expectError(error.WildcardInTopicName, h.conn.publish(.{ .topic = "a/#" }));
    try testing.expectError(error.MalformedWildcard, h.conn.subscribe(&.{.{ .filter = "a/#/b" }}, .empty));
    try testing.expectError(error.Empty, h.conn.unsubscribe(&.{""}, .empty));
    try h.expectQuiet();
}

test "an incoming topic alias is remembered and then resolved" {
    var h: Harness = .init(.{ .incoming_topic_alias_maximum = 4 });
    defer h.deinit();
    try h.connectAndAccept(.{});

    // The packet that establishes the alias carries the topic as well.
    try h.deliver(.{ .publish = .{
        .topic = "sensor/7/temperature",
        .payload = "20",
        .properties = .{ .topic_alias = 2 },
    } });
    try testing.expectEqualStrings("sensor/7/temperature", (try h.conn.nextEvent()).?.message.topic);

    // Later ones leave it out, and the connection puts it back.
    try h.deliver(.{ .publish = .{ .payload = "21", .properties = .{ .topic_alias = 2 } } });
    const second = (try h.conn.nextEvent()).?.message;
    try testing.expectEqualStrings("sensor/7/temperature", second.topic);
    try testing.expectEqualStrings("21", second.payload);
}

test "an alias above the advertised maximum, or one never established, is refused" {
    var h: Harness = .init(.{ .incoming_topic_alias_maximum = 2 });
    defer h.deinit();
    try h.connectAndAccept(.{});

    try h.deliver(.{ .publish = .{ .topic = "a", .properties = .{ .topic_alias = 3 } } });
    try testing.expectError(error.TopicAliasInvalid, h.conn.nextEvent());

    var h2: Harness = .init(.{ .incoming_topic_alias_maximum = 2 });
    defer h2.deinit();
    try h2.connectAndAccept(.{});
    try h2.deliver(.{ .publish = .{ .properties = .{ .topic_alias = 1 } } });
    try testing.expectError(error.TopicAliasInvalid, h2.conn.nextEvent());
}

test "a publish with neither a topic nor an alias is a protocol error" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    try h.deliver(.{ .publish = .{ .topic = "" } });
    try testing.expectError(error.ProtocolError, h.conn.nextEvent());
}

test "outgoing aliases are established once and used thereafter" {
    var h: Harness = .init(.{ .outgoing_topic_alias_maximum = 2 });
    defer h.deinit();
    try h.connectAndAccept(.{ .properties = .{ .topic_alias_maximum = 10 } });

    // First time: topic and alias together, so the broker learns the mapping.
    _ = try h.conn.publish(.{ .topic = "sensor/7/temperature", .payload = "20" });
    const first = (try h.take()).?.publish;
    try testing.expectEqualStrings("sensor/7/temperature", first.topic);
    try testing.expectEqual(@as(u16, 1), first.properties.topic_alias.?);

    // Afterwards: the alias alone, which is the saving the feature exists for.
    _ = try h.conn.publish(.{ .topic = "sensor/7/temperature", .payload = "21" });
    const second = (try h.take()).?.publish;
    try testing.expectEqualStrings("", second.topic);
    try testing.expectEqual(@as(u16, 1), second.properties.topic_alias.?);

    // A second topic gets the next alias; a third exhausts the limit and is
    // sent in full rather than evicting anything.
    _ = try h.conn.publish(.{ .topic = "sensor/8/temperature" });
    try testing.expectEqual(@as(u16, 2), (try h.take()).?.publish.properties.topic_alias.?);
    _ = try h.conn.publish(.{ .topic = "sensor/9/temperature" });
    const third = (try h.take()).?.publish;
    try testing.expectEqualStrings("sensor/9/temperature", third.topic);
    try testing.expectEqual(@as(?u16, null), third.properties.topic_alias);
}

test "outgoing aliases are not used when the broker will not take any" {
    var h: Harness = .init(.{ .outgoing_topic_alias_maximum = 8 });
    defer h.deinit();
    // No Topic Alias Maximum in the CONNACK means zero, and a client that sent
    // an alias anyway would be disconnected for it.
    try h.connectAndAccept(.{});

    _ = try h.conn.publish(.{ .topic = "a/b" });
    const p = (try h.take()).?.publish;
    try testing.expectEqualStrings("a/b", p.topic);
    try testing.expectEqual(@as(?u16, null), p.properties.topic_alias);
}

test "aliases are forgotten when the network connection ends" {
    // Section 3.3.2.3.4: topic alias mappings belong to the network connection,
    // not the session, so a resumed session must re-establish them.
    var h: Harness = .init(.{ .outgoing_topic_alias_maximum = 2, .incoming_topic_alias_maximum = 2 });
    defer h.deinit();
    try h.connectAndAccept(.{ .properties = .{ .topic_alias_maximum = 10 } });

    _ = try h.conn.publish(.{ .topic = "a/b" });
    _ = try h.take();
    h.conn.transportClosed();

    try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now);
    _ = try h.take();
    try h.deliver(.{ .connack = .{ .session_present = true, .properties = .{ .topic_alias_maximum = 10 } } });
    _ = try h.conn.nextEvent();

    _ = try h.conn.publish(.{ .topic = "a/b" });
    const p = (try h.take()).?.publish;
    try testing.expectEqualStrings("a/b", p.topic);
    try testing.expectEqual(@as(u16, 1), p.properties.topic_alias.?);
}

test "a packet arriving one byte at a time is decoded once it is whole" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    var buf: [128]u8 = undefined;
    var w: std.Io.Writer = .fixed(&buf);
    try (Packet{ .publish = .{ .topic = "a/b", .payload = "hello" } }).encode(&w, .v5_0);
    const bytes = w.buffered();

    for (bytes[0 .. bytes.len - 1]) |byte| {
        try h.conn.receive(&.{byte}, h.now);
        try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent());
    }
    try h.conn.receive(bytes[bytes.len - 1 ..], h.now);
    try testing.expectEqualStrings("hello", (try h.conn.nextEvent()).?.message.payload);
}

test "several packets in one read are decoded one after another" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    var buf: [256]u8 = undefined;
    var w: std.Io.Writer = .fixed(&buf);
    for ([_][]const u8{ "one", "two", "three" }) |payload| {
        try (Packet{ .publish = .{ .topic = "a", .payload = payload } }).encode(&w, .v5_0);
    }
    try h.conn.receive(w.buffered(), h.now);

    try testing.expectEqualStrings("one", (try h.conn.nextEvent()).?.message.payload);
    try testing.expectEqualStrings("two", (try h.conn.nextEvent()).?.message.payload);
    try testing.expectEqualStrings("three", (try h.conn.nextEvent()).?.message.payload);
    try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent());
}

test "a packet larger than this client will accept ends the connection" {
    var h: Harness = .init(.{ .max_incoming_packet_size = 32 });
    defer h.deinit();
    try h.connectAndAccept(.{});

    // Only the fixed header is needed to know it is too big, so this never
    // grows the receive buffer to hold the rest.
    try h.conn.receive(&.{ 0x30, 0xff, 0x01 }, h.now);
    try testing.expectError(error.PacketTooLarge, h.conn.nextEvent());
}

test "rubbish from the broker ends the connection rather than being skipped" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    // Packet type 0 is reserved.
    try h.conn.receive(&.{ 0x00, 0x00 }, h.now);
    try testing.expectError(error.ProtocolError, h.conn.nextEvent());
    try testing.expectEqual(State.closed, h.conn.state);
}

test "a packet only a broker receives is a protocol error coming the other way" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    try h.deliver(.pingreq);
    try testing.expectError(error.ProtocolError, h.conn.nextEvent());
}

test "the broker can end the connection and say why" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    try h.deliver(.{ .disconnect = .{
        .reason_code = .server_shutting_down,
        .properties = .{ .server_reference = "other.example:1883" },
    } });
    const event = (try h.conn.nextEvent()).?.disconnected;
    try testing.expectEqual(ReasonCode.server_shutting_down, event.reason_code);
    try testing.expectEqual(Disconnected.Origin.disconnect, event.origin);
    try testing.expectEqualStrings("other.example:1883", event.properties.server_reference.?);
    try testing.expectEqual(State.closed, h.conn.state);
}

test "disconnecting stops the connection taking commands" {
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    try h.conn.disconnect(.normal_disconnection, .empty);
    const p = (try h.take()).?.disconnect;
    try testing.expectEqual(ReasonCode.success, p.reason_code);
    try testing.expectEqual(State.disconnecting, h.conn.state);
    try testing.expectError(error.NotConnected, h.conn.publish(.{ .topic = "a" }));
}

test "a 3.1.1 connection speaks 3.1.1" {
    var h: Harness = .init(.{ .version = .v3_1_1 });
    defer h.deinit();

    try h.conn.connect(.{ .client_id = "old", .keep_alive = 60 }, h.now);
    const connect_packet = (try h.take()).?.connect;
    try testing.expectEqualStrings("old", connect_packet.client_id);
    // The properties this connection would have advertised in 5.0 have nowhere
    // to go, and are left out rather than making the packet unencodable.
    try testing.expect(connect_packet.properties.isEmpty());

    try h.deliver(.{ .connack = .{} });
    _ = try h.conn.nextEvent();

    const id = (try h.conn.publish(.{ .topic = "a/b", .qos = .at_least_once })).?;
    try testing.expectEqual(id, (try h.take()).?.publish.packet_id.?);
    try h.deliver(.{ .puback = .{ .packet_id = id } });
    try testing.expectEqual(id, (try h.conn.nextEvent()).?.published.packet_id);

    // A 3.1.1 broker's CONNACK has no properties to carry a keep alive
    // override, so what was asked for is what is in force.
    try testing.expectEqual(@as(u16, 60), h.conn.keep_alive);
}

test "the buffers do not grow without bound" {
    // A long conversation in small pieces: without compaction on both sides
    // this grows until it runs out of memory rather than staying flat.
    var h: Harness = .init(.{});
    defer h.deinit();
    try h.connectAndAccept(.{});

    for (0..2000) |i| {
        _ = try h.conn.publish(.{ .topic = "a/b", .payload = "x" });
        _ = try h.take();
        try h.deliver(.{ .publish = .{ .topic = "c/d", .payload = "y" } });
        _ = try h.conn.nextEvent();
        _ = i;
    }
    // Drain the last event so the receive buffer can be released.
    _ = try h.conn.nextEvent();

    try testing.expect(h.conn.rx.capacity < 1024);
    try testing.expect(h.conn.tx.capacity < 1024);
}

test "nothing may precede the connack" {
    // Section 3.2, [MQTT-3.2.0-1]. A broker that delivers a message before the
    // connection is established has skipped telling the client what its limits
    // are, and `Client.connect` has nowhere to put the message.
    var h: Harness = .init(.{});
    defer h.deinit();

    try h.conn.connect(.{ .client_id = "test" }, h.now);
    _ = try h.take();
    try h.deliver(.{ .publish = .{ .topic = "a/b", .payload = "too soon" } });

    try testing.expectError(error.ProtocolError, h.conn.nextEvent());
    try testing.expectEqual(State.closed, h.conn.state);
}

test "an unasked-for authentication exchange is refused" {
    // Section 4.12: an AUTH is a step in an exchange the CONNECT started. A
    // broker sending one to a client that named no method has invented a
    // conversation, and there is nothing to answer it with.
    var h: Harness = .init(.{});
    defer h.deinit();

    try h.conn.connect(.{ .client_id = "test" }, h.now);
    _ = try h.take();
    try h.deliver(.{ .auth = .{
        .reason_code = .continue_authentication,
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    } });
    try testing.expectError(error.ProtocolError, h.conn.nextEvent());
}

test "the authentication method cannot change mid-exchange" {
    // Section 4.12 fixes the method when the CONNECT goes out. A broker naming
    // a different one is either confused or moving the client onto something
    // weaker, and the second is the reason this is checked rather than
    // tolerated.
    var h: Harness = .init(.{});
    defer h.deinit();

    try h.conn.connect(.{
        .client_id = "test",
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    }, h.now);
    _ = try h.take();
    try h.deliver(.{ .auth = .{
        .reason_code = .continue_authentication,
        .properties = .{ .authentication_method = "PLAIN" },
    } });
    try testing.expectError(error.ProtocolError, h.conn.nextEvent());
}

test "an authentication step with no method at all is refused" {
    var h: Harness = .init(.{});
    defer h.deinit();

    try h.conn.connect(.{
        .client_id = "test",
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    }, h.now);
    _ = try h.take();
    try h.deliver(.{ .auth = .{ .reason_code = .continue_authentication } });
    try testing.expectError(error.ProtocolError, h.conn.nextEvent());
}

test "a broker cannot end the first exchange with an AUTH success" {
    // Section 4.12.1 ends a *re-authentication* that way. The first exchange
    // ends with a CONNACK, and a client that took a bare success for one would
    // consider itself connected without ever having been told the session's
    // limits.
    var h: Harness = .init(.{});
    defer h.deinit();

    try h.conn.connect(.{
        .client_id = "test",
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    }, h.now);
    _ = try h.take();
    try h.deliver(.{ .auth = .{
        .reason_code = .success,
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    } });
    try testing.expectError(error.ProtocolError, h.conn.nextEvent());
}

test "a broker cannot ask the client to re-authenticate" {
    // `reauthenticate` travels one way only. A broker that wants another
    // exchange closes the connection and lets the client start one.
    var h: Harness = .init(.{});
    defer h.deinit();

    try h.conn.connect(.{
        .client_id = "test",
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    }, h.now);
    _ = try h.take();
    try h.deliver(.{ .auth = .{
        .reason_code = .reauthenticate,
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    } });
    try testing.expectError(error.ProtocolError, h.conn.nextEvent());
}

test "a success that skipped the exchange is refused" {
    // The downgrade this exists to stop: the CONNECT asked for a mechanism,
    // and something answered yes without running it. A client that accepted
    // that would believe it had authenticated by a method that never ran.
    var h: Harness = .init(.{});
    defer h.deinit();

    try h.conn.connect(.{
        .client_id = "test",
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    }, h.now);
    _ = try h.take();
    try h.deliver(.{ .connack = .{} });
    try testing.expectError(error.ProtocolError, h.conn.nextEvent());

    // And a CONNACK that names the method is accepted, so the refusal above is
    // about the missing property and not about the exchange having happened.
    var ok: Harness = .init(.{});
    defer ok.deinit();
    try ok.conn.connect(.{
        .client_id = "test",
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    }, ok.now);
    _ = try ok.take();
    try ok.deliver(.{ .connack = .{
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    } });
    try testing.expect((try ok.conn.nextEvent()).? == .connected);

    // A refusal need not carry it: there is no exchange left to protect.
    var refused: Harness = .init(.{});
    defer refused.deinit();
    try refused.conn.connect(.{
        .client_id = "test",
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    }, refused.now);
    _ = try refused.take();
    try refused.deliver(.{ .connack = .{ .reason_code = .not_authorized } });
    try testing.expectEqual(
        ReasonCode.not_authorized,
        (try refused.conn.nextEvent()).?.disconnected.reason_code,
    );
}

test "an AUTH the caller sends carries the method it does not have to remember" {
    var h: Harness = .init(.{});
    defer h.deinit();

    // Nothing to continue: the CONNECT named no method.
    try h.conn.connect(.{ .client_id = "test" }, h.now);
    _ = try h.take();
    try testing.expectError(error.NotAuthenticating, h.conn.auth(.continue_authentication, .empty));

    var e: Harness = .init(.{});
    defer e.deinit();
    try e.conn.connect(.{
        .client_id = "test",
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    }, e.now);
    _ = try e.take();

    // Filled in, because there is exactly one right answer and section 4.12
    // requires it on every AUTH.
    try e.conn.auth(.continue_authentication, .{ .authentication_data = "client-final" });
    const outgoing = (try e.take()).?.auth;
    try testing.expectEqualStrings("SCRAM-SHA-256", outgoing.properties.authentication_method.?);
    try testing.expectEqualStrings("client-final", outgoing.properties.authentication_data.?);

    // And a caller that names a different one has done the thing the rule
    // forbids, rather than having its value quietly replaced.
    try testing.expectError(error.AuthenticationMethodChanged, e.conn.auth(
        .continue_authentication,
        .{ .authentication_method = "PLAIN" },
    ));

    // `success` is the broker's to send, not ours.
    try testing.expectError(error.BadReasonCode, e.conn.auth(.success, .empty));

    // And re-authentication needs a connection to re-authenticate on.
    try testing.expectError(error.NotConnected, e.conn.auth(.reauthenticate, .empty));
}

test "re-authentication runs on a connection that is already up" {
    // Section 4.12.1: the session is untouched and the exchange happens over
    // it, which is what makes a credential that expires cost nothing.
    var h: Harness = .init(.{});
    defer h.deinit();

    try h.conn.connect(.{
        .client_id = "test",
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    }, h.now);
    _ = try h.take();
    try h.deliver(.{ .connack = .{
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    } });
    _ = try h.conn.nextEvent();

    try h.conn.auth(.reauthenticate, .{ .authentication_data = "client-first" });
    try testing.expectEqual(ReasonCode.reauthenticate, (try h.take()).?.auth.reason_code);

    try h.deliver(.{ .auth = .{
        .reason_code = .continue_authentication,
        .properties = .{
            .authentication_method = "SCRAM-SHA-256",
            .authentication_data = "server-first",
        },
    } });
    try testing.expectEqualStrings(
        "server-first",
        (try h.conn.nextEvent()).?.authenticate.properties.authentication_data.?,
    );

    // And this time a `success` on an AUTH is the right way to end it.
    try h.deliver(.{ .auth = .{
        .reason_code = .success,
        .properties = .{ .authentication_method = "SCRAM-SHA-256" },
    } });
    try testing.expectEqual(
        ReasonCode.success,
        (try h.conn.nextEvent()).?.authenticate.reason_code,
    );
    // The session carried on throughout: publishing was never interrupted.
    _ = try h.conn.publish(.{ .topic = "a/b", .payload = "still here" });
    try testing.expectEqualStrings("still here", (try h.take()).?.publish.payload);
}

test "an authentication exchange is handed to the caller" {
    var h: Harness = .init(.{});
    defer h.deinit();

    try h.conn.connect(.{
        .client_id = "test",
        .properties = .{ .authentication_method = "SCRAM-SHA-256", .authentication_data = "client-first" },
    }, h.now);
    _ = try h.take();

    try h.deliver(.{ .auth = .{
        .reason_code = .continue_authentication,
        .properties = .{ .authentication_method = "SCRAM-SHA-256", .authentication_data = "server-first" },
    } });
    const event = (try h.conn.nextEvent()).?.authenticate;
    try testing.expectEqualStrings("server-first", event.properties.authentication_data.?);

    try h.conn.auth(.continue_authentication, .{
        .authentication_method = "SCRAM-SHA-256",
        .authentication_data = "client-final",
    });
    try testing.expectEqualStrings(
        "client-final",
        (try h.take()).?.auth.properties.authentication_data.?,
    );
}

test "a connection that reports an error has stopped" {
    // The invariant the fuzzer checks after every failure, and which one path
    // -- an unknown incoming topic alias -- used to break: whatever the error,
    // the connection is closed afterwards and will not take another command.
    // A caller that ignores an error must not be able to keep publishing into
    // a session the broker is about to drop.
    const Case = struct { bytes: []const u8, expected: anyerror };
    const cases = [_]Case{
        // An alias that was never established.
        .{ .bytes = &.{ 0x30, 0x08, 0x00, 0x00, 0x03, 0x23, 0x00, 0x01, 'y', 'o' }, .expected = error.TopicAliasInvalid },
        // An alias above the advertised maximum.
        .{ .bytes = &.{ 0x30, 0x08, 0x00, 0x00, 0x03, 0x23, 0x00, 0x63, 'y', 'o' }, .expected = error.TopicAliasInvalid },
        // Neither a topic nor an alias.
        .{ .bytes = &.{ 0x30, 0x03, 0x00, 0x00, 0x00 }, .expected = error.ProtocolError },
        // A reserved packet type.
        .{ .bytes = &.{ 0x00, 0x00 }, .expected = error.ProtocolError },
        // An acknowledgement for something never sent.
        .{ .bytes = &.{ 0x40, 0x02, 0x03, 0xe7 }, .expected = error.ProtocolError },
    };

    for (cases) |case| {
        var h: Harness = .init(.{ .incoming_topic_alias_maximum = 4 });
        defer h.deinit();
        try h.connectAndAccept(.{});

        try h.conn.receive(case.bytes, h.now);
        try testing.expectError(case.expected, h.conn.nextEvent());
        try testing.expectEqual(State.closed, h.conn.state);
        try testing.expectError(error.NotConnected, h.conn.publish(.{ .topic = "a" }));
    }
}

test "what the connect advertises is what the connection enforces" {
    // A caller that sets the Topic Alias Maximum property by hand must get a
    // connection that honours that number, not the one in `Options`.
    var h: Harness = .init(.{ .incoming_topic_alias_maximum = 2 });
    defer h.deinit();

    try h.conn.connect(.{ .client_id = "test", .properties = .{ .topic_alias_maximum = 5 } }, h.now);
    try testing.expectEqual(@as(u16, 5), (try h.take()).?.connect.properties.topic_alias_maximum.?);
    try h.deliver(.{ .connack = .{} });
    _ = try h.conn.nextEvent();

    // Alias 5 was advertised as acceptable, so it must be accepted.
    try h.deliver(.{ .publish = .{ .topic = "a/b", .properties = .{ .topic_alias = 5 } } });
    try testing.expectEqualStrings("a/b", (try h.conn.nextEvent()).?.message.topic);
}

test "an outgoing alias the broker will not take is refused before it is sent" {
    var h: Harness = .init(.{});
    defer h.deinit();
    // No Topic Alias Maximum in the CONNACK means zero: the broker takes none.
    try h.connectAndAccept(.{});

    try testing.expectError(error.TopicAliasInvalid, h.conn.publish(.{
        .topic = "a/b",
        .properties = .{ .topic_alias = 1 },
    }));
    try h.expectQuiet();

    // With a broker that does take them, the caller's own alias goes through.
    var h2: Harness = .init(.{});
    defer h2.deinit();
    try h2.connectAndAccept(.{ .properties = .{ .topic_alias_maximum = 4 } });
    _ = try h2.conn.publish(.{ .topic = "a/b", .properties = .{ .topic_alias = 3 } });
    try testing.expectEqual(@as(u16, 3), (try h2.take()).?.publish.properties.topic_alias.?);
}

test "a publish that runs out of memory part-way leaves nothing dangling" {
    // `encodeAlloc` succeeds, the append succeeds, and then `pumpQueued` fails.
    // The in-flight entry owns the encoded packet by then, so an `errdefer`
    // reaching that far would hand the same memory back twice.
    //
    // Rather than arranging that one ordering by hand, every allocation in the
    // sequence is failed in turn and the connection is torn down each time; the
    // testing allocator underneath is what reports a double free or a leak.
    for (0..24) |budget| {
        var failing: std.testing.FailingAllocator = .init(testing.allocator, .{ .fail_index = budget });
        var conn: Connection = .init(failing.allocator(), .{});
        defer conn.deinit();

        conn.connect(.{ .client_id = "oom", .keep_alive = 0 }, 0) catch continue;
        conn.sent(conn.pending().len, 0);
        conn.receive(&.{ 0x20, 0x03, 0x00, 0x00, 0x00 }, 0) catch continue;
        _ = conn.nextEvent() catch continue;

        _ = conn.publish(.{
            .topic = "a/b",
            .payload = "hello",
            .qos = .at_least_once,
        }) catch {};
    }
}

/// What the session looked like last time it was asked, so that a test can say
/// "the bytes changed" and "the serial changed" and compare the two.
const SessionWitness = struct {
    serial: u64 = 0,
    /// Whether `bytes` holds a session or merely has not been filled in yet,
    /// which are different things and only the first is worth comparing.
    primed: bool = false,
    bytes: std.ArrayList(u8) = .empty,

    fn deinit(w: *SessionWitness) void {
        w.bytes.deinit(testing.allocator);
    }

    /// The guarantee `session_serial` makes: it may move when the session did
    /// not change, and it may never fail to move when the session did.
    fn check(w: *SessionWitness, c: *Connection) !void {
        var out: std.Io.Writer.Allocating = .init(testing.allocator);
        defer out.deinit();
        try c.saveSession(&out.writer, "test");

        if (w.primed and
            !std.mem.eql(u8, out.written(), w.bytes.items) and
            c.session_serial == w.serial)
        {
            return error.SessionSerialMissedAChange;
        }
        w.primed = true;
        w.serial = c.session_serial;
        w.bytes.clearRetainingCapacity();
        try w.bytes.appendSlice(testing.allocator, out.written());
    }
};

test "the session serial moves whenever the session does" {
    // The check that keeps `session_serial` worth having. A caller that skips
    // a save because the serial did not move must be right to skip it, so
    // every step of a conversation is measured both ways: what the session
    // serialises to, and what the counter says.
    var h: Harness = .init(.{});
    defer h.deinit();
    var witness: SessionWitness = .{};
    defer witness.deinit();

    try witness.check(&h.conn);
    try h.connectAndAccept(.{});
    try witness.check(&h.conn);

    // QoS 1, all the way through.
    const one = (try h.conn.publish(.{ .topic = "a/b", .payload = "x", .qos = .at_least_once })).?;
    try witness.check(&h.conn);
    _ = try h.take();
    try witness.check(&h.conn);
    try h.deliver(.{ .puback = .{ .packet_id = one } });
    _ = try h.conn.nextEvent();
    try witness.check(&h.conn);

    // QoS 2, all four packets.
    const two = (try h.conn.publish(.{ .topic = "c/d", .payload = "y", .qos = .exactly_once })).?;
    try witness.check(&h.conn);
    _ = try h.take();
    try witness.check(&h.conn);
    try h.deliver(.{ .pubrec = .{ .packet_id = two } });
    _ = try h.conn.nextEvent();
    try witness.check(&h.conn);
    _ = try h.take();
    try witness.check(&h.conn);
    try h.deliver(.{ .pubcomp = .{ .packet_id = two } });
    _ = try h.conn.nextEvent();
    try witness.check(&h.conn);

    // And the other direction: a QoS 2 message arriving is session state too,
    // because its identifier has to be remembered until the PUBREL.
    try h.deliver(.{ .publish = .{
        .topic = "e/f",
        .payload = "z",
        .qos = .exactly_once,
        .packet_id = 55,
    } });
    _ = try h.conn.nextEvent();
    try witness.check(&h.conn);
    _ = try h.take();
    try witness.check(&h.conn);
    try h.deliver(.{ .pubrel = .{ .packet_id = 55 } });
    _ = try h.conn.nextEvent();
    try witness.check(&h.conn);

    // Something left in flight, and then the transport goes.
    _ = try h.conn.publish(.{ .topic = "g/h", .qos = .at_least_once });
    try witness.check(&h.conn);
    _ = try h.take();
    try witness.check(&h.conn);
    h.conn.transportClosed();
    try witness.check(&h.conn);

    // A resumed session marks it for sending again, which is a change, and so
    // is the duplicate flag that goes on it when it goes.
    try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now);
    _ = try h.take();
    try witness.check(&h.conn);
    try h.deliver(.{ .connack = .{ .session_present = true } });
    _ = try h.conn.nextEvent();
    try witness.check(&h.conn);
    _ = try h.take();
    try witness.check(&h.conn);

    // And a session the broker has forgotten is emptied.
    h.conn.transportClosed();
    try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now);
    _ = try h.take();
    try h.deliver(.{ .connack = .{ .session_present = false } });
    _ = try h.conn.nextEvent();
    try witness.check(&h.conn);
}

test "a session outlives the process that made it" {
    // The whole point. A QoS 2 publication whose PUBCOMP never arrived is a
    // message the broker has been told to hold and not yet told to deliver;
    // a client that forgot it on exit has left it there forever.
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    var packet_id: PacketId = undefined;

    {
        var h: Harness = .init(.{});
        defer h.deinit();
        try h.connectAndAccept(.{});

        packet_id = (try h.conn.publish(.{
            .topic = "a/b",
            .payload = "important",
            .qos = .exactly_once,
        })).?;
        _ = try h.take();
        try h.deliver(.{ .pubrec = .{ .packet_id = packet_id } });
        _ = try h.conn.nextEvent();
        // The PUBREL goes out, and then the process dies before the PUBCOMP.
        _ = try h.take();

        try h.conn.saveSession(&out.writer, "test");
    }

    // A new process, a new connection, and the same session.
    var h: Harness = .init(.{});
    defer h.deinit();

    var reader: std.Io.Reader = .fixed(out.written());
    try h.conn.loadSession(&reader, "test");

    // Resumed, not cleaned: a clean start would tell the broker to forget the
    // half of this that it is holding.
    try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now);
    _ = try h.take();
    try h.deliver(.{ .connack = .{ .session_present = true } });
    _ = try h.conn.nextEvent();

    // The PUBREL is owed, and goes again.
    const owed = (try h.take()).?;
    try testing.expectEqual(packet_id, owed.pubrel.packet_id);

    try h.deliver(.{ .pubcomp = .{ .packet_id = packet_id } });
    const done = (try h.conn.nextEvent()).?.published;
    try testing.expectEqual(packet_id, done.packet_id);

    // And the identifier counter came back too, so the next message cannot
    // collide with one the broker is still holding.
    const next = (try h.conn.publish(.{ .topic = "c/d", .qos = .at_least_once })).?;
    try testing.expect(next != packet_id);
}

test "a publication that never went out survives as well" {
    // Not section 4.1's session state -- it was never sent -- but losing it
    // would be losing a message the caller believes it published, which is a
    // worse failure than storing slightly more than the standard requires.
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();

    {
        var h: Harness = .init(.{});
        defer h.deinit();
        // A window of one, so the second publication is held back.
        try h.connectAndAccept(.{ .properties = .{ .receive_maximum = 1 } });
        _ = try h.conn.publish(.{ .topic = "a/b", .qos = .at_least_once });
        _ = try h.conn.publish(.{ .topic = "c/d", .payload = "waiting", .qos = .at_least_once });
        _ = try h.take();
        try h.expectQuiet();
        try h.conn.saveSession(&out.writer, "test");
    }

    var h: Harness = .init(.{});
    defer h.deinit();
    var reader: std.Io.Reader = .fixed(out.written());
    try h.conn.loadSession(&reader, "test");
    try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now);
    _ = try h.take();
    try h.deliver(.{ .connack = .{ .session_present = true } });
    _ = try h.conn.nextEvent();

    // Both go out: the one that had been sent carries the duplicate flag, and
    // the one that never did does not.
    const first = (try h.take()).?.publish;
    try testing.expectEqualStrings("a/b", first.topic);
    try testing.expect(first.duplicate);
    const second = (try h.take()).?.publish;
    try testing.expectEqualStrings("c/d", second.topic);
    try testing.expect(!second.duplicate);
}

test "a session belongs to one client and one protocol version" {
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    {
        var h: Harness = .init(.{});
        defer h.deinit();
        try h.connectAndAccept(.{});
        _ = try h.conn.publish(.{ .topic = "a/b", .qos = .at_least_once });
        try h.conn.saveSession(&out.writer, "test");
    }

    // Another client's unacknowledged messages are not this client's to send.
    {
        var h: Harness = .init(.{});
        defer h.deinit();
        var reader: std.Io.Reader = .fixed(out.written());
        try testing.expectError(
            error.SessionBelongsToAnotherClient,
            h.conn.loadSession(&reader, "somebody-else"),
        );
    }

    // And the packets in it are encoded for the version that wrote them.
    {
        var h: Harness = .init(.{ .version = .v3_1_1 });
        defer h.deinit();
        var reader: std.Io.Reader = .fixed(out.written());
        try testing.expectError(
            error.SessionVersionMismatch,
            h.conn.loadSession(&reader, "test"),
        );
    }

    // Loading is something that happens before the CONNECT, because the broker
    // has to be asked to resume what has just been restored.
    {
        var h: Harness = .init(.{});
        defer h.deinit();
        try h.connectAndAccept(.{});
        var reader: std.Io.Reader = .fixed(out.written());
        try testing.expectError(error.AlreadyConnected, h.conn.loadSession(&reader, "test"));
    }
}

test "a broker that forgot the session throws away what was restored" {
    // The honest outcome rather than a hidden one: a broker with no record of
    // the session will not accept the second half of a handshake it never saw,
    // so sending the restored messages again would be sending them into
    // nothing.
    var out: std.Io.Writer.Allocating = .init(testing.allocator);
    defer out.deinit();
    {
        var h: Harness = .init(.{});
        defer h.deinit();
        try h.connectAndAccept(.{});
        _ = try h.conn.publish(.{ .topic = "a/b", .qos = .exactly_once });
        _ = try h.take();
        try h.conn.saveSession(&out.writer, "test");
    }

    var h: Harness = .init(.{});
    defer h.deinit();
    var reader: std.Io.Reader = .fixed(out.written());
    try h.conn.loadSession(&reader, "test");
    try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now);
    _ = try h.take();
    try h.deliver(.{ .connack = .{ .session_present = false } });
    _ = try h.conn.nextEvent();

    try h.expectQuiet();
}

test "asking the broker to keep a session is something you have to ask for" {
    // Section 3.1.2.11.2: a missing Session Expiry Interval means zero, and
    // zero means the session ends with the network connection. So
    // `clean_start = false` on its own asks the broker to resume a session it
    // is about to throw away, which is a trap worth having a knob for --
    // measured against mosquitto, which does exactly that.
    {
        var h: Harness = .init(.{});
        defer h.deinit();
        try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now);
        const out = (try h.take()).?.connect;
        try testing.expectEqual(@as(?u32, null), out.properties.session_expiry_interval);
    }

    {
        var h: Harness = .init(.{ .session_expiry_interval = 3600 });
        defer h.deinit();
        try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now);
        const out = (try h.take()).?.connect;
        try testing.expectEqual(@as(?u32, 3600), out.properties.session_expiry_interval);
    }

    // What the caller put in the CONNECT wins, including a zero that means
    // "drop it after all".
    {
        var h: Harness = .init(.{ .session_expiry_interval = 3600 });
        defer h.deinit();
        try h.conn.connect(.{
            .client_id = "test",
            .clean_start = false,
            .properties = .{ .session_expiry_interval = 0 },
        }, h.now);
        const out = (try h.take()).?.connect;
        try testing.expectEqual(@as(?u32, 0), out.properties.session_expiry_interval);
    }

    // 3.1.1 has no such property: `clean_session = 0` asks the broker to keep
    // the session and the broker decides for how long.
    {
        var h: Harness = .init(.{ .version = .v3_1_1, .session_expiry_interval = 3600 });
        defer h.deinit();
        try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now);
        const out = (try h.take()).?.connect;
        try testing.expectEqual(@as(?u32, null), out.properties.session_expiry_interval);
    }
}