pinch-points 0.8.0

//! What one datagram says, and the bytes it says it in.
//!
//! Byte 0 tags the message and byte 1 is the [`PROTOCOL_VERSION`], frozen
//! across versions so a build can tell "I cannot read this" from "I
//! disagree with this".

use super::*;

#[derive(Clone, PartialEq, Eq, Debug)]
pub enum NetMsg {
    /// Handshake ping; the host learns the peer's address (and what to
    /// call them) from it, and whether a bot is driving its seat: the
    /// game sets that flag, never the name, so the table always knows.
    Hello { name: WireName, bot: bool },
    /// Handshake ping from a peer that wants to watch, not play. Repeated
    /// like `Hello` until a `Start` lands.
    ///
    /// It carries a name for the same reason `Hello` does, and the reason
    /// is sharper here: the host vets a chat line against the name it
    /// knows the sender by (see `OnlineSession::take_chat`), and a
    /// spectator it has no name for is a peer whose claim about who is
    /// talking nobody can check.
    Watch { name: WireName },
    /// Every input the sender can still be holding for a peer: the newest
    /// commit and the resend tail behind it, in one datagram.
    ///
    /// One datagram a tick, not one per seat-frame: at a six-seat table
    /// that took the host from 850 datagrams a tick to 25, and the bytes
    /// fell eightfold with them.
    ///
    /// The whole tail still goes out every tick, so a lost datagram is
    /// made good by the next one 33 ms later. A peer now gets all of a
    /// tick's tail or none of it, rather than the ragged subset a burst of
    /// loss used to leave.
    Inputs(Vec<InputMsg>),
    /// State fingerprint after `frame`, for loud desync detection.
    Hash { frame: u32, hash: u64 },
    /// Host → joiner: the match begins with `seats` seats on `terms`; you are
    /// `seat`, and the table is called `names` (empty entries fall back to
    /// seat labels). Re-sent whenever a joiner is still saying hello.
    Start {
        seats: u8,
        /// The seat this peer is given, or `None` for a peer that came to
        /// watch. On the wire that `None` is [`SPECTATOR_SEAT`], a number
        /// outside the range of real seats.
        seat: Option<u8>,
        terms: MatchTerms,
        names: [WireName; crate::sim::MAX_PLAYERS],
        /// What holds each seat (see [`kind_byte`]): the round names its
        /// bots, so every peer draws the robot on them.
        kinds: [u8; crate::sim::MAX_PLAYERS],
        /// Where the series stands as this round begins, or `None` for a
        /// single round. On the wire the absence is a round number of
        /// zero, which no series ever reaches.
        ///
        /// The host says, because seats move: a peer that leaves between
        /// rounds frees its chair and everyone behind it moves up one, so
        /// a tally kept by seat number credited the departed player's
        /// rounds to whoever moved in. The host re-deals the tally with
        /// the chairs, and a peer admitted from the queue mid-series
        /// learns the standings the same way.
        standing: Option<SeriesStanding>,
        /// The beach itself, when the host picked one it built rather than
        /// one both peers already have. A generated arena travels as a
        /// seed; a handmade one has to travel as itself, because the
        /// joiner has never seen the file. Empty for the built-in maps.
        ///
        /// Compressed with the same coder the share codes use, and it has
        /// to be: a 20x13 beach is fifteen hundred characters of text and
        /// the datagram is a kilobyte.
        beach: Vec<u8>,
    },
    /// Someone hit pause: everybody stops committing at `frame`. Re-sent
    /// every tick while paused, so a dropped datagram costs a moment of
    /// confusion rather than a stuck match.
    Pause { frame: u32 },
    /// Play on: the pause that was to freeze on `frame` is lifted. Also
    /// re-sent until the sim visibly moves again. The frame is what lets a
    /// peer tell the `Pause` echoes still in flight from before the resume
    /// (see `Lockstep::receive_pause`) from a fresh pause.
    Resume { frame: u32 },
    /// Host → the table: `seat` has stopped sending and an AI is taking
    /// the chair.
    ///
    /// Only the host says so, and everyone acts on its word rather than on
    /// their own patience. A peer that decided for itself would fill the
    /// seat on whichever frame its own timer happened to run out, and two
    /// peers filling it on different frames is a desync, which lockstep has
    /// no way to recover from.
    ///
    /// `frame` is the one the host was held up on, and every peer empties
    /// the seat from there: peers do not all hold the same inputs from a
    /// player who has gone quiet, so "from the frame you are stuck on" is
    /// not the same frame everywhere. Repeated for the rest of the round,
    /// since a lost one leaves that peer frozen while the others play on.
    Abandoned { seat: u8, frame: u32 },
    /// Host → the lobby: who is at the table right now, in seat order.
    ///
    /// A joiner has only ever spoken to the host, so without this it knows
    /// nobody else is even there until the match starts. Re-sent whenever
    /// the table changes and on a timer besides, since a roster that went
    /// missing would leave a screen wrong for good rather than briefly.
    Roster {
        seats: u8,
        names: [WireName; crate::sim::MAX_PLAYERS],
        /// The host's dials as they stand, so a joiner's card shows the
        /// match it is joining rather than its own setup screen's idea of
        /// one. Everything but the seed is meaningful before the launch.
        terms: MatchTerms,
        /// What holds each place at the table (see [`kind_byte`]), so a
        /// bot wears the robot on every screen in the room.
        kinds: [u8; crate::sim::MAX_PLAYERS],
    },
    /// Host → one peer: the host asked you to leave. The peer goes back to
    /// choosing a beach, and the host forgets it.
    Kicked,
    /// A line said in the lobby, and who said it. The sender names itself
    /// rather than the host stamping it: a joiner's greeting is the only
    /// other place the host learns a name, and a peer that never greeted
    /// would otherwise speak anonymously.
    ///
    /// Relayed by the host to the rest of the table, like an input: the
    /// spokes of the star cannot hear each other.
    Chat { name: WireName, text: WireChat },
    /// Host → everyone, once a second while it changes: how many are
    /// watching, how long a vote has left to join, and how long until the
    /// next one may be called. All in whole seconds, zero for none.
    ///
    /// Only the host counts, so only the host can say. Without it a
    /// spectator presses a key into silence: it cannot tell whether a vote
    /// is open, how many of them there are to out-vote, or why nothing
    /// happened when the crowd is still waiting.
    SpectatorTally { watching: u8, open: u8, wait: u8 },
    /// A spectator's pick for the next tide event the spectators call.
    ///
    /// Only the host counts these. It is the one peer everybody waits on
    /// anyway, so it is the one that can put the result where nobody can
    /// miss it: into an action, on a frame.
    SpectatorVote { event: u8 },
    /// A spectator's call for who wins this round: a seat, picked while the
    /// host still takes picks. Only the host counts, as with the vote, and
    /// a later pick from the same spectator replaces an earlier one.
    SpectatorPick { seat: u8 },
    /// Host → everyone, once a second while picks may still change: how
    /// many spectators have called each seat, and the whole seconds left to
    /// call one, zero once the calls are in. The players see it too; what
    /// the crowd thinks of them is half the point.
    CrowdPicks {
        picks: [u8; crate::sim::MAX_PLAYERS],
        open: u8,
    },
    /// Host → a peer that turned up after the launch: the round is under
    /// way and cannot take you, but you are in line for the next one, with
    /// `ahead` people in front of you.
    ///
    /// The answer to a greeting that used to be met with a spectator seat:
    /// lockstep replays from frame zero, so such a peer built a board
    /// nobody would ever send it inputs for and sat there, apparently
    /// connected, forever.
    Queued { ahead: u8 },
    /// Host → a spectator that arrived with the round already running: one
    /// part of the round as it stands at `frame`, the `Start` it would have
    /// been sent at the launch and the board, packed together and cut into
    /// `parts` pieces, of which this is number `part`.
    ///
    /// In parts because the board does not fit a datagram: a mid-round
    /// beach carries every crab and gull and post on it. Keyed by `frame`,
    /// so pieces of two different snapshots can never be put together; a
    /// piece that went astray is made good by the next greeting, which is
    /// answered with a whole fresh snapshot.
    CatchUp {
        frame: u32,
        part: u8,
        parts: u8,
        bytes: Vec<u8>,
    },
    /// "I speak protocol `version`, and what you sent me is not it." The
    /// answer to a datagram from another build, so a mismatched joiner is
    /// told why nothing is happening instead of greeting a host that ignores
    /// it forever.
    ///
    /// The one message exempt from the version gate, and the one whose layout
    /// is frozen along with the version byte: every build, past and future,
    /// can read `[TAG_INCOMPATIBLE, version]`.
    Incompatible { version: u8 },
}

/// Where a series stands as a round begins: its 1-based number, and the
/// rounds each *seat* has won so far.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct SeriesStanding {
    pub round: u8,
    pub wins: [u8; crate::sim::MAX_PLAYERS],
}

impl SeriesStanding {
    /// The standing at the top of a series: round one, nothing won.
    pub fn opening() -> SeriesStanding {
        SeriesStanding {
            round: 1,
            wins: [0; crate::sim::MAX_PLAYERS],
        }
    }
}

impl NetMsg {
    /// A greeting carrying `name` in wire form.
    pub fn hello(name: &str) -> NetMsg {
        NetMsg::Hello {
            name: wire_name(name),
            bot: false,
        }
    }

    /// The greeting of a game whose seat a bot will drive.
    pub fn hello_bot(name: &str) -> NetMsg {
        NetMsg::Hello {
            name: wire_name(name),
            bot: true,
        }
    }

    /// The same greeting from somebody who came to watch. Named, because
    /// the host has to be able to call the crowd something and to check
    /// what the crowd says it is called.
    pub fn watch(name: &str) -> NetMsg {
        NetMsg::Watch {
            name: wire_name(name),
        }
    }

    /// A line of chat as `who` said it, both halves in wire form. An empty
    /// `who` is the lobby itself speaking.
    pub fn chat(who: &str, line: &str) -> NetMsg {
        NetMsg::Chat {
            name: wire_name(who),
            text: wire_chat(line),
        }
    }
}

// Byte 0 of every datagram.
//
// A new tag is three lines, not one: the tag, `HIGHEST_TAG` below it, and
// a `PROTOCOL_VERSION` bump. Nothing stops compiling if you skip the bump,
// and then two builds claim the same version while one sends a tag the
// other reads as noise: chat, the roster and the abandonment notice all
// shipped under version 4 before anybody noticed.
const TAG_HELLO: u8 = 0;
/// Retired in version 11, when a tick's inputs became one datagram rather
/// than thirty-four. The number stays spoken for and is never handed to
/// another message: a version 10 peer still sends it, and `peek_version`
/// only answers `Incompatible` for a tag inside the range, which is what
/// tells that peer why its round never starts.
const TAG_RETIRED_INPUT: u8 = 1;
const TAG_HASH: u8 = 2;
const TAG_START: u8 = 3;
const TAG_PAUSE: u8 = 4;
const TAG_RESUME: u8 = 5;
const TAG_WATCH: u8 = 6;
const TAG_INCOMPATIBLE: u8 = 7;
const TAG_QUEUED: u8 = 8;
const TAG_CHAT: u8 = 9;
const TAG_ROSTER: u8 = 10;
const TAG_ABANDONED: u8 = 11;
const TAG_INPUTS: u8 = 12;
const TAG_SPECTATOR_VOTE: u8 = 13;
const TAG_SPECTATOR_TALLY: u8 = 14;
const TAG_CATCH_UP: u8 = 15;
const TAG_SPECTATOR_PICK: u8 = 16;
const TAG_CROWD_PICKS: u8 = 17;
const TAG_KICKED: u8 = 18;
/// The last of them, which `peek_version` uses to tell one of ours from
/// stray traffic on the port.
///
/// It said `TAG_INPUTS` through two more tags, so a build of another
/// version sending the spectators' two was ignored rather than told why.
const HIGHEST_TAG: u8 = TAG_KICKED;

/// Inputs one datagram may carry.
///
/// The cap is the buffer: `2 + 1 + 127 * INPUT_BYTES` is 1019 bytes, inside
/// [`MAX_DATAGRAM`] and inside any MTU worth worrying about. A datagram past
/// the path MTU is fragmented, and a fragment lost is the whole datagram
/// lost, so a batch that needed two IP packets would be a resend tail that
/// fails twice as often as the thing it repairs.
///
/// A tail never comes close: `2 * resend_span(DEFAULT_DELAY)` is 66. The
/// host's relay does, at a full table, and chunks.
pub const MAX_INPUTS_PER_DATAGRAM: usize = 127;

/// The most of a snapshot one `CatchUp` carries: the datagram less its
/// header (tag, version, frame, part, parts and a length), with room to
/// spare under a 1500-byte MTU so no part is ever fragmented.
pub const CATCH_UP_CHUNK: usize = 900;

/// The most parts a snapshot may be cut into. A crowded XL beach packs to
/// a few; this is room for many times that, and a bound on what a
/// receiver will hold for a snapshot still arriving.
pub const CATCH_UP_PARTS: u8 = 32;

/// How a peer that came to watch is written down in a `Start`: outside
/// the range of real seats, so it cannot collide with one. A wire detail:
/// everything above this file says `None`.
const SPECTATOR_SEAT: u8 = u8::MAX;

/// What holds a seat, as one byte on the wire: 0 a person, 1 the game's
/// AI, 2 a bot. Anything else reads as a person, which is what a seat is
/// unless somebody says otherwise.
pub fn kind_byte(kind: crate::sim::SeatKind) -> u8 {
    match kind {
        crate::sim::SeatKind::Human => 0,
        crate::sim::SeatKind::Ai => 1,
        crate::sim::SeatKind::Bot => 2,
    }
}

/// A table's kinds in wire form, a byte a seat; places past the table
/// are people, which is what a byte of nought says.
pub fn wire_kinds(kinds: &[crate::sim::SeatKind]) -> [u8; crate::sim::MAX_PLAYERS] {
    let mut out = [0u8; crate::sim::MAX_PLAYERS];
    for (slot, kind) in out.iter_mut().zip(kinds) {
        *slot = kind_byte(*kind);
    }
    out
}

/// The kind a byte names; see [`kind_byte`].
pub fn kind_from_byte(byte: u8) -> crate::sim::SeatKind {
    match byte {
        1 => crate::sim::SeatKind::Ai,
        2 => crate::sim::SeatKind::Bot,
        _ => crate::sim::SeatKind::Human,
    }
}

/// Everything about a match that every peer has to agree on, or the boards
/// diverge (or, for `teams`, two peers score the same round differently).
///
/// Held as plain numbers rather than the app's enums: this layer is the wire
/// and knows nothing about menus. The app maps them at the edge.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct MatchTerms {
    /// Seats the AI holds, counting down from the top seat.
    pub bots: u8,
    /// AI difficulty index (easy/normal/fierce).
    pub bot_level: u8,
    /// Map choice index.
    pub map: u8,
    /// Gull pressure index.
    pub gulls: u8,
    /// Round length index.
    pub round: u8,
    /// How the round is scored, as a team-mode index (free-for-all, pairs,
    /// trios).
    pub teams: u8,
    /// The board's PRNG seed, so every peer builds the same beach. Also
    /// what tells a fresh `Start` from the stale one a host re-answers a
    /// stray greeting with: a new seed is a new round.
    pub seed: u64,
    /// How long the series runs, as a `SeriesLength` index: one round,
    /// best of three, best of five. Every peer keeps its own tally, and
    /// they agree because they are counting the same deterministic boards,
    /// but only if they all know how many rounds take it.
    pub series: u8,
    /// How many arrows each player may have standing. A count, not an
    /// index, and not trusted: the app clamps it to its own dial's range.
    pub posts: u8,
}

/// Written out because the arrows are not zero by default: terms nobody
/// set play the versus rule's three, as every round did before the count
/// travelled.
impl Default for MatchTerms {
    fn default() -> Self {
        MatchTerms {
            bots: 0,
            bot_level: 0,
            map: 0,
            gulls: 0,
            round: 0,
            teams: 0,
            seed: 0,
            series: 0,
            posts: crate::sim::MAX_SIGNPOSTS_PER_PLAYER as u8,
        }
    }
}

impl MatchTerms {
    /// Whether these terms call for a series at all. Index 0 is the dial's
    /// "single round"; every other index is some best-of. The only reading
    /// of the byte the app should make: comparing against a particular
    /// index once left best-of-five launching as a single round on every
    /// joiner.
    pub fn is_series(self) -> bool {
        self.series != 0
    }

    /// The humans at a table of `seats`: the AI holds the top seats, so
    /// the humans are the low `seats - bots` of them, and the lockstep is
    /// built over those alone. Never fewer than one, since seat zero is
    /// the host's whatever the dials say.
    pub fn humans(self, seats: u8) -> u8 {
        seats.saturating_sub(self.bots).max(1)
    }
}

impl MatchTerms {
    const BYTES: usize = 16;

    fn encode(self) -> [u8; Self::BYTES] {
        let mut out = [0u8; Self::BYTES];
        out[0] = self.bots;
        out[1] = self.bot_level;
        out[2] = self.map;
        out[3] = self.gulls;
        out[4] = self.round;
        out[5] = self.teams;
        out[6..14].copy_from_slice(&self.seed.to_le_bytes());
        out[14] = self.series;
        out[15] = self.posts;
        out
    }

    fn decode(bytes: &[u8]) -> Option<MatchTerms> {
        let seed = u64::from_le_bytes(bytes.get(6..14)?.try_into().ok()?);
        Some(MatchTerms {
            bots: *bytes.first()?,
            bot_level: *bytes.get(1)?,
            map: *bytes.get(2)?,
            gulls: *bytes.get(3)?,
            round: *bytes.get(4)?,
            teams: *bytes.get(5)?,
            seed,
            series: *bytes.get(14)?,
            posts: *bytes.get(15)?,
        })
    }
}

/// A batch of inputs straight into the caller's buffer: byte for byte what
/// `NetMsg::Inputs(..).encode()` writes, without the `Vec` that would take.
/// Answers the length written.
///
/// `msgs` must be within [`MAX_INPUTS_PER_DATAGRAM`]; the one caller chunks,
/// and `the_two_encoders_agree` holds the two spellings to each other.
pub(super) fn encode_inputs(msgs: &[InputMsg], buf: &mut [u8; MAX_DATAGRAM]) -> usize {
    debug_assert!(
        msgs.len() <= MAX_INPUTS_PER_DATAGRAM,
        "{} inputs in one datagram",
        msgs.len()
    );
    let count = msgs.len().min(MAX_INPUTS_PER_DATAGRAM);
    buf[0] = TAG_INPUTS;
    buf[1] = PROTOCOL_VERSION;
    buf[2] = count as u8;
    for (index, msg) in msgs[..count].iter().enumerate() {
        let at = 3 + index * INPUT_BYTES;
        buf[at..at + INPUT_BYTES].copy_from_slice(&msg.encode());
    }
    3 + count * INPUT_BYTES
}

impl NetMsg {
    pub fn encode(self) -> Vec<u8> {
        // Byte 0 tags the message, byte 1 says who wrote it; the payload
        // starts at byte 2.
        let mut bytes = match self {
            NetMsg::Hello { .. } => vec![TAG_HELLO],
            NetMsg::Watch { .. } => vec![TAG_WATCH],
            NetMsg::Queued { .. } => vec![TAG_QUEUED],
            NetMsg::Chat { .. } => vec![TAG_CHAT],
            NetMsg::Roster { .. } => vec![TAG_ROSTER],
            NetMsg::Abandoned { .. } => vec![TAG_ABANDONED],
            NetMsg::Inputs(_) => vec![TAG_INPUTS],
            NetMsg::Hash { .. } => vec![TAG_HASH],
            NetMsg::Start { .. } => vec![TAG_START],
            NetMsg::Pause { .. } => vec![TAG_PAUSE],
            NetMsg::SpectatorVote { .. } => vec![TAG_SPECTATOR_VOTE],
            NetMsg::SpectatorTally { .. } => vec![TAG_SPECTATOR_TALLY],
            NetMsg::Resume { .. } => vec![TAG_RESUME],
            NetMsg::CatchUp { .. } => vec![TAG_CATCH_UP],
            NetMsg::SpectatorPick { .. } => vec![TAG_SPECTATOR_PICK],
            NetMsg::CrowdPicks { .. } => vec![TAG_CROWD_PICKS],
            NetMsg::Kicked => vec![TAG_KICKED],
            NetMsg::Incompatible { version } => return vec![TAG_INCOMPATIBLE, version],
        };
        bytes.push(PROTOCOL_VERSION);
        match self {
            NetMsg::Incompatible { .. } | NetMsg::Kicked => {}
            NetMsg::Resume { frame } => bytes.extend_from_slice(&frame.to_le_bytes()),
            NetMsg::Queued { ahead } => bytes.push(ahead),
            NetMsg::Abandoned { seat, frame } => {
                bytes.push(seat);
                bytes.extend_from_slice(&frame.to_le_bytes());
            }
            NetMsg::Chat { name, text } => {
                bytes.extend_from_slice(&name);
                bytes.extend_from_slice(&text);
            }
            NetMsg::Roster {
                seats,
                names,
                terms,
                kinds,
            } => {
                bytes.push(seats);
                for name in names {
                    bytes.extend_from_slice(&name);
                }
                bytes.extend_from_slice(&terms.encode());
                bytes.extend_from_slice(&kinds);
            }
            NetMsg::Hello { name, bot } => {
                bytes.extend_from_slice(&name);
                bytes.push(u8::from(bot));
            }
            NetMsg::Watch { name } => bytes.extend_from_slice(&name),
            NetMsg::Inputs(ref inputs) => {
                // Count first, then that many fixed-width inputs. A batch
                // past the cap is truncated rather than sent whole: the
                // alternative is a datagram the receiver never sees.
                // Senders chunk, so nothing here does.
                debug_assert!(
                    inputs.len() <= MAX_INPUTS_PER_DATAGRAM,
                    "{} inputs in one datagram",
                    inputs.len()
                );
                let count = inputs.len().min(MAX_INPUTS_PER_DATAGRAM);
                bytes.push(count as u8);
                for input in &inputs[..count] {
                    bytes.extend_from_slice(&input.encode());
                }
            }
            NetMsg::Hash { frame, hash } => {
                bytes.extend_from_slice(&frame.to_le_bytes());
                bytes.extend_from_slice(&hash.to_le_bytes());
            }
            NetMsg::Start {
                seats,
                seat,
                terms,
                names,
                kinds,
                standing,
                beach,
            } => {
                bytes.push(seats);
                bytes.push(seat.unwrap_or(SPECTATOR_SEAT));
                bytes.extend_from_slice(&terms.encode());
                for name in names {
                    bytes.extend_from_slice(&name);
                }
                bytes.extend_from_slice(&kinds);
                // Round zero is "no series", as `SPECTATOR_SEAT` is "no
                // seat": both are numbers the real thing never takes.
                let SeriesStanding { round, wins } = standing.unwrap_or(SeriesStanding {
                    round: 0,
                    wins: [0; crate::sim::MAX_PLAYERS],
                });
                bytes.push(round);
                bytes.extend_from_slice(&wins);
                // Length-prefixed and last, so the fixed part above stays
                // where it was and a beach of any size is one read.
                let len = u16::try_from(beach.len()).unwrap_or(0);
                bytes.extend_from_slice(&len.to_le_bytes());
                bytes.extend_from_slice(&beach[..usize::from(len)]);
            }
            NetMsg::Pause { frame } => bytes.extend_from_slice(&frame.to_le_bytes()),
            NetMsg::CatchUp {
                frame,
                part,
                parts,
                bytes: ref piece,
            } => {
                debug_assert!(piece.len() <= CATCH_UP_CHUNK, "{} bytes", piece.len());
                let len = piece.len().min(CATCH_UP_CHUNK);
                bytes.extend_from_slice(&frame.to_le_bytes());
                bytes.push(part);
                bytes.push(parts);
                bytes.extend_from_slice(&(len as u16).to_le_bytes());
                bytes.extend_from_slice(&piece[..len]);
            }
            NetMsg::SpectatorVote { event } => bytes.push(event),
            NetMsg::SpectatorPick { seat } => bytes.push(seat),
            NetMsg::CrowdPicks { picks, open } => {
                bytes.extend_from_slice(&picks);
                bytes.push(open);
            }
            NetMsg::SpectatorTally {
                watching,
                open,
                wait,
            } => bytes.extend_from_slice(&[watching, open, wait]),
        }
        bytes
    }

    /// The protocol version a datagram was written by, whatever else it
    /// says, and `None` for anything that is not one of our messages at
    /// all. Byte 1 is frozen across versions so this always answers, and
    /// the tag is checked first so stray traffic on the port draws no reply.
    pub fn peek_version(bytes: &[u8]) -> Option<u8> {
        let tag = *bytes.first()?;
        (tag <= HIGHEST_TAG)
            .then(|| bytes.get(1).copied())
            .flatten()
    }

    /// Decode a datagram this build can act on. A message from another
    /// protocol version is refused here rather than half-understood. The
    /// sole exception is the refusal message itself, which every version
    /// can read by construction.
    pub fn decode(bytes: &[u8]) -> Option<NetMsg> {
        let tag = *bytes.first()?;
        let version = *bytes.get(1)?;
        if tag == TAG_INCOMPATIBLE {
            return Some(NetMsg::Incompatible { version });
        }
        if version != PROTOCOL_VERSION {
            return None;
        }
        let body = bytes.get(2..)?;
        match tag {
            TAG_HELLO => Some(NetMsg::Hello {
                name: body.get(..WIRE_NAME)?.try_into().ok()?,
                bot: *body.get(WIRE_NAME)? != 0,
            }),
            TAG_KICKED => Some(NetMsg::Kicked),
            TAG_WATCH => Some(NetMsg::Watch {
                name: body.get(..WIRE_NAME)?.try_into().ok()?,
            }),
            TAG_INPUTS => {
                // A count past the cap is refused rather than clamped: the
                // encoder would write back fewer than it read, and a message
                // that does not survive its own round trip is one the host
                // could relay as something other than what it was told.
                let count = usize::from(*body.first()?);
                if count > MAX_INPUTS_PER_DATAGRAM {
                    return None;
                }
                let packed = body.get(1..1 + count * INPUT_BYTES)?;
                Some(NetMsg::Inputs(
                    packed
                        .as_chunks::<INPUT_BYTES>()
                        .0
                        .iter()
                        .map(|chunk| InputMsg::decode(*chunk))
                        .collect(),
                ))
            }
            // Spoken for and unread: a peer on this version never sends
            // it, and one on an older version is answered by the caller.
            TAG_RETIRED_INPUT => None,
            TAG_HASH => {
                let frame = u32::from_le_bytes(body.get(..4)?.try_into().ok()?);
                let hash = u64::from_le_bytes(body.get(4..12)?.try_into().ok()?);
                Some(NetMsg::Hash { frame, hash })
            }
            TAG_START => {
                let terms = MatchTerms::decode(body.get(2..)?)?;
                let names = read_table(body.get(2 + MatchTerms::BYTES..)?)?;
                let (seats, seat) = (*body.first()?, *body.get(1)?);
                // A table this build cannot sit at is refused outright
                // rather than squeezed into range: the seat number goes on
                // to index the lockstep's own slots. Watching is the one
                // seat legitimately outside the range.
                //
                // The AI holds the top seats, so the joiner builds its
                // lockstep over the low `seats - bots`. A `Start` seating
                // us in an AI's chair would have it play a session it is
                // not a player of, which the lockstep meets with a panic.
                let humans = terms.humans(seats);
                if !(2..=crate::sim::MAX_PLAYERS as u8).contains(&seats)
                    || (seat != SPECTATOR_SEAT && seat >= humans)
                {
                    return None;
                }
                let seat = (seat != SPECTATOR_SEAT).then_some(seat);
                debug_assert!(seat.is_none_or(|seat| seat < seats));
                let kinds_at = 2 + MatchTerms::BYTES + WIRE_NAME * crate::sim::MAX_PLAYERS;
                let kinds: [u8; crate::sim::MAX_PLAYERS] = body
                    .get(kinds_at..kinds_at + crate::sim::MAX_PLAYERS)?
                    .try_into()
                    .ok()?;
                let series_at = kinds_at + crate::sim::MAX_PLAYERS;
                let round = *body.get(series_at)?;
                let wins: [u8; crate::sim::MAX_PLAYERS] = body
                    .get(series_at + 1..series_at + 1 + crate::sim::MAX_PLAYERS)?
                    .try_into()
                    .ok()?;
                let standing = (round != 0).then_some(SeriesStanding { round, wins });
                let after = series_at + 1 + crate::sim::MAX_PLAYERS;
                // Every encoder of this version writes the length, so a
                // datagram that ends before it is not one of ours.
                let len = usize::from(u16::from_le_bytes(
                    body.get(after..after + 2)?.try_into().ok()?,
                ));
                let beach = body.get(after + 2..after + 2 + len)?.to_vec();
                Some(NetMsg::Start {
                    seats,
                    seat,
                    terms,
                    names,
                    kinds,
                    standing,
                    beach,
                })
            }
            TAG_PAUSE => Some(NetMsg::Pause {
                frame: u32::from_le_bytes(body.get(..4)?.try_into().ok()?),
            }),
            TAG_CATCH_UP => {
                let frame = u32::from_le_bytes(body.get(..4)?.try_into().ok()?);
                let (part, parts) = (*body.get(4)?, *body.get(5)?);
                let len = usize::from(u16::from_le_bytes(body.get(6..8)?.try_into().ok()?));
                // A piece of a snapshot nobody could finish is refused
                // here, before a receiver sets aside room for it.
                if parts == 0 || parts > CATCH_UP_PARTS || part >= parts || len > CATCH_UP_CHUNK {
                    return None;
                }
                Some(NetMsg::CatchUp {
                    frame,
                    part,
                    parts,
                    bytes: body.get(8..8 + len)?.to_vec(),
                })
            }
            TAG_SPECTATOR_VOTE => Some(NetMsg::SpectatorVote {
                event: *body.first()?,
            }),
            // A seat off the table is nobody to call, and would index off
            // the end of the count.
            TAG_SPECTATOR_PICK => {
                let seat = *body.first()?;
                (seat < crate::sim::MAX_PLAYERS as u8).then_some(NetMsg::SpectatorPick { seat })
            }
            TAG_CROWD_PICKS => Some(NetMsg::CrowdPicks {
                picks: body.get(..crate::sim::MAX_PLAYERS)?.try_into().ok()?,
                open: *body.get(crate::sim::MAX_PLAYERS)?,
            }),
            TAG_SPECTATOR_TALLY => Some(NetMsg::SpectatorTally {
                watching: *body.first()?,
                open: *body.get(1)?,
                wait: *body.get(2)?,
            }),
            TAG_RESUME => Some(NetMsg::Resume {
                frame: u32::from_le_bytes(body.get(..4)?.try_into().ok()?),
            }),
            TAG_QUEUED => Some(NetMsg::Queued {
                ahead: *body.first()?,
            }),
            TAG_ABANDONED => {
                let seat = *body.first()?;
                let frame = u32::from_le_bytes(body.get(1..5)?.try_into().ok()?);
                // A seat outside the table is not one anything can be done
                // about, and acting on it would index off the end.
                (seat < crate::sim::MAX_PLAYERS as u8).then_some(NetMsg::Abandoned { seat, frame })
            }
            TAG_ROSTER => {
                let table = body.get(1..)?;
                let names = read_table(table)?;
                let terms_at = WIRE_NAME * crate::sim::MAX_PLAYERS;
                let terms = MatchTerms::decode(table.get(terms_at..)?)?;
                let kinds_at = terms_at + MatchTerms::BYTES;
                let kinds = table
                    .get(kinds_at..kinds_at + crate::sim::MAX_PLAYERS)?
                    .try_into()
                    .ok()?;
                Some(NetMsg::Roster {
                    seats: *body.first()?,
                    names,
                    terms,
                    kinds,
                })
            }
            TAG_CHAT => Some(NetMsg::Chat {
                name: body.get(..WIRE_NAME)?.try_into().ok()?,
                text: body
                    .get(WIRE_NAME..WIRE_NAME + WIRE_CHAT)?
                    .try_into()
                    .ok()?,
            }),
            _ => None,
        }
    }
}

/// A full table of names off the front of `bytes`, or `None` if the
/// datagram ends before the table does. Both messages that carry one
/// (`Start`, `Roster`) read it this way.
fn read_table(bytes: &[u8]) -> Option<[WireName; crate::sim::MAX_PLAYERS]> {
    let mut names = [[0u8; WIRE_NAME]; crate::sim::MAX_PLAYERS];
    for (i, name) in names.iter_mut().enumerate() {
        *name = bytes
            .get(i * WIRE_NAME..(i + 1) * WIRE_NAME)?
            .try_into()
            .ok()?;
    }
    Some(names)
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Every message must encode within the receive buffer, or the kernel
    /// truncates it on arrival and decode drops it without a trace, which
    /// is how a 64-byte buffer once swallowed the named `Start` whole.
    #[test]
    fn every_message_fits_the_receive_buffer() {
        let widest = wire_name("WWWWWWWWWWWWWWWWWWWWWWWW");
        for msg in [
            NetMsg::Hello {
                name: widest,
                bot: true,
            },
            NetMsg::Kicked,
            NetMsg::Chat {
                name: widest,
                text: wire_chat(&"W".repeat(CHAT_CHARS)),
            },
            NetMsg::Start {
                seats: 6,
                seat: Some(5),
                terms: MatchTerms::default(),
                names: [widest; crate::sim::MAX_PLAYERS],
                kinds: [0; crate::sim::MAX_PLAYERS],
                // The widest a Start gets: a full table of longest names
                // and the largest beach the sender will hand it.
                standing: None,
                beach: vec![0xAB; MAX_BEACH_BYTES],
            },
            NetMsg::Hash {
                frame: u32::MAX,
                hash: u64::MAX,
            },
            NetMsg::CatchUp {
                frame: u32::MAX,
                part: CATCH_UP_PARTS - 1,
                parts: CATCH_UP_PARTS,
                bytes: vec![0xAB; CATCH_UP_CHUNK],
            },
            // A batch filled to the cap, which is the one message whose
            // size a caller chooses rather than the format fixing it.
            NetMsg::Inputs(vec![
                InputMsg {
                    player: 5,
                    frame: u32::MAX,
                    action: PlayerAction::Place {
                        x: 19,
                        y: 12,
                        dir: Direction::Left,
                    },
                };
                MAX_INPUTS_PER_DATAGRAM
            ]),
        ] {
            let len = msg.clone().encode().len();
            assert!(len <= MAX_DATAGRAM, "{len} bytes: {msg:?}");
        }
    }

    /// The whole point of the batch: a tail as long as the lockstep can
    /// ever hold it must be one datagram, not two. `resend_span` keeps
    /// commits from `frame - span` while they run to `frame + span - 1`, so
    /// twice the span is the most [`Lockstep::recent_commits`] can answer,
    /// and it is nowhere near the cap.
    ///
    /// If this ever fails the fix is not a larger buffer but chunking, as
    /// the host's relay already does. See [`MAX_INPUTS_PER_DATAGRAM`].
    #[test]
    fn a_full_resend_tail_is_one_datagram() {
        let span = (crate::sim::DEFAULT_DELAY + crate::sim::MAX_COMMIT_LEAD) as usize;
        let tail = 2 * span;
        assert!(
            tail <= MAX_INPUTS_PER_DATAGRAM,
            "a {tail}-message tail does not fit {MAX_INPUTS_PER_DATAGRAM} per datagram",
        );
        let len = NetMsg::Inputs(vec![
            InputMsg {
                player: 5,
                frame: u32::MAX,
                action: PlayerAction::None,
            };
            tail
        ])
        .encode()
        .len();
        assert!(len <= MAX_DATAGRAM, "{len} bytes");
    }

    /// The batch has two encoders: the one every message uses, and the one
    /// the send path uses to keep a `Vec` off the hot loop. They write the
    /// same bytes or they are two protocols.
    #[test]
    fn the_two_encoders_agree() {
        for count in [0usize, 1, 34, MAX_INPUTS_PER_DATAGRAM] {
            let msgs: Vec<InputMsg> = (0..count)
                .map(|i| InputMsg {
                    player: (i % crate::sim::MAX_PLAYERS) as u8,
                    frame: i as u32 * 7,
                    action: PlayerAction::Place {
                        x: (i % 20) as u8,
                        y: (i % 13) as u8,
                        dir: Direction::Right,
                    },
                })
                .collect();
            let mut buf = [0u8; MAX_DATAGRAM];
            let len = encode_inputs(&msgs, &mut buf);
            assert_eq!(
                &buf[..len],
                NetMsg::Inputs(msgs.clone()).encode().as_slice(),
                "{count} inputs"
            );
            assert_eq!(
                NetMsg::decode(&buf[..len]),
                Some(NetMsg::Inputs(msgs)),
                "{count} inputs"
            );
        }
    }

    /// The retired tag earns its keep at exactly one moment: a version 10
    /// peer dials in, sends the inputs it has always sent, and has to be
    /// told why nothing happens. That answer only goes out for a tag
    /// inside the range, so the number stays spoken for.
    #[test]
    fn a_version_ten_input_is_refused_and_answered() {
        let mut old = vec![TAG_RETIRED_INPUT, 10];
        old.extend_from_slice(&[0u8; INPUT_BYTES]);
        assert_eq!(NetMsg::decode(&old), None, "not read as anything");
        assert_eq!(
            NetMsg::peek_version(&old),
            Some(10),
            "and answered with why"
        );
    }

    /// A bot is said to be one in every message that names a seat: the
    /// greeting's flag, and the kinds the roster and the start carry.
    #[test]
    fn what_holds_a_seat_survives_the_wire() {
        for bot in [false, true] {
            let hello = NetMsg::Hello {
                name: wire_name("Greedy (Ana)"),
                bot,
            };
            assert_eq!(NetMsg::decode(&hello.clone().encode()), Some(hello));
        }
        let kinds = wire_kinds(&[
            crate::sim::SeatKind::Human,
            crate::sim::SeatKind::Bot,
            crate::sim::SeatKind::Ai,
        ]);
        assert_eq!(kinds, [0, 2, 1, 0, 0, 0]);
        let roster = NetMsg::Roster {
            seats: 3,
            names: [wire_name("x"); crate::sim::MAX_PLAYERS],
            terms: MatchTerms::default(),
            kinds,
        };
        assert_eq!(NetMsg::decode(&roster.clone().encode()), Some(roster));
        let start = NetMsg::Start {
            seats: 3,
            seat: Some(1),
            terms: MatchTerms::default(),
            names: [wire_name("x"); crate::sim::MAX_PLAYERS],
            kinds,
            standing: None,
            beach: vec![1, 2, 3],
        };
        assert_eq!(NetMsg::decode(&start.clone().encode()), Some(start));
        assert_eq!(
            NetMsg::decode(&NetMsg::Kicked.encode()),
            Some(NetMsg::Kicked)
        );
        assert_eq!(kind_from_byte(2), crate::sim::SeatKind::Bot);
        assert_eq!(
            kind_from_byte(99),
            crate::sim::SeatKind::Human,
            "unknown reads as a person"
        );
    }

    /// [`MAX_BEACH_BYTES`] is what the sender trusts, so it has to be a
    /// number this encoder agrees with: the widest possible invitation
    /// carrying the largest allowed beach must still fit the buffer, with
    /// room to spare for a field somebody adds to `Start` later.
    #[test]
    fn a_start_carrying_the_largest_beach_still_fits() {
        let widest = wire_name("WWWWWWWWWWWWWWWWWWWWWWWW");
        let len = NetMsg::Start {
            seats: 6,
            seat: Some(5),
            terms: MatchTerms::default(),
            names: [widest; crate::sim::MAX_PLAYERS],
            kinds: [0; crate::sim::MAX_PLAYERS],
            standing: None,
            beach: vec![0xAB; MAX_BEACH_BYTES],
        }
        .encode()
        .len();
        assert!(len <= MAX_DATAGRAM, "{len} bytes");
        let spare = MAX_DATAGRAM - len;
        assert!(spare >= 16, "only {spare} bytes of slack left");
    }

    /// The series byte is an *index*, and every index but zero names some
    /// best-of. Read any narrower and it lies: `== 1` is best-of-three and
    /// nothing else, which once launched every best-of-five as a single
    /// round on each joiner while the host went on counting a series. The
    /// whole byte, because a peer on another build picks the number.
    #[test]
    fn every_series_index_but_zero_names_a_series() {
        let dialled = |series| MatchTerms {
            series,
            ..MatchTerms::default()
        };
        assert!(!dialled(0).is_series(), "the dial's single round");
        for index in 1..=u8::MAX {
            assert!(dialled(index).is_series(), "index {index}");
        }
    }

    /// The humans at a table: the range a joiner builds its lockstep over,
    /// and the range this file's own decoder refuses a seat outside of.
    ///
    /// The AI holds the top chairs, so the humans are what is left
    /// underneath - but never none. Seat zero is the host's whatever the
    /// dials say, and a `bots` byte from a broken or hostile build must
    /// come to a floor rather than wrapping the count back up under it.
    #[test]
    fn the_humans_are_what_the_ai_leaves_and_never_none() {
        let with = |bots| MatchTerms {
            bots,
            ..MatchTerms::default()
        };
        assert_eq!(with(0).humans(6), 6, "no AI, every chair a player's");
        assert_eq!(with(4).humans(6), 2, "four bots behind two people");
        assert_eq!(with(5).humans(6), 1, "the host alone");
        assert_eq!(with(6).humans(6), 1, "and still the host alone");
        assert_eq!(with(u8::MAX).humans(6), 1, "however wild the byte");
        assert_eq!(with(0).humans(0), 1, "even a table of nobody seats one");
        // Whatever it answers is a count the per-seat arrays can hold, which
        // is what the `Start` decoder leans on when it refuses a seat.
        for seats in 0..=u8::MAX {
            for bots in [0, 1, 5, 6, 7, u8::MAX] {
                let humans = with(bots).humans(seats);
                assert!(humans >= 1, "{seats} seats, {bots} bots");
                assert!(humans <= seats.max(1), "{seats} seats, {bots} bots");
            }
        }
    }

    /// A call for a seat no table has is refused on the way in.
    #[test]
    fn a_pick_off_the_table_is_refused() {
        let mut bytes = NetMsg::SpectatorPick { seat: 1 }.encode();
        *bytes.last_mut().expect("a seat byte") = crate::sim::MAX_PLAYERS as u8;
        assert!(NetMsg::decode(&bytes).is_none());
    }

    #[test]
    fn decode_rejects_garbage() {
        assert!(NetMsg::decode(&[]).is_none());
        assert!(NetMsg::decode(&[0xFF, 1, 2, 3]).is_none());
        assert!(NetMsg::decode(b"PNCH?").is_none());
    }

    use crate::sim::{Direction, PlayerAction};

    #[test]
    fn messages_round_trip() {
        for msg in [
            NetMsg::hello("Anna"),
            NetMsg::hello("Überlang-Name-über-die-Kappe-hinaus"),
            NetMsg::watch("Dee"),
            NetMsg::Resume { frame: 0 },
            NetMsg::Resume { frame: 70_000 },
            NetMsg::Pause { frame: 7 },
            NetMsg::SpectatorVote { event: 8 },
            NetMsg::SpectatorTally {
                watching: 3,
                open: 2,
                wait: 0,
            },
            NetMsg::SpectatorPick { seat: 5 },
            NetMsg::CrowdPicks {
                picks: [0, 3, 1, 0, 0, 2],
                open: 27,
            },
            NetMsg::CatchUp {
                frame: 901,
                part: 1,
                parts: 3,
                bytes: vec![7; 40],
            },
            NetMsg::Inputs(vec![InputMsg {
                player: 1,
                frame: 42,
                action: PlayerAction::Place {
                    x: 3,
                    y: 8,
                    dir: Direction::Down,
                },
            }]),
            // An empty batch is never sent, and still has to come back as
            // one: the fuzzer will build the three bytes that say so.
            NetMsg::Inputs(Vec::new()),
            // The XL beach is 20 wide: a placement out in column 18 has to
            // survive the trip, which it did not while a tile was a nibble.
            // Behind it, the shape a real tail has: several frames of one
            // seat in one datagram.
            NetMsg::Inputs(vec![
                InputMsg {
                    player: 5,
                    frame: 4000,
                    action: PlayerAction::Place {
                        x: 18,
                        y: 11,
                        dir: Direction::Left,
                    },
                },
                InputMsg {
                    player: 5,
                    frame: 4001,
                    action: PlayerAction::None,
                },
                InputMsg {
                    player: 5,
                    frame: 4002,
                    action: PlayerAction::Place {
                        x: 0,
                        y: 0,
                        dir: Direction::Up,
                    },
                },
            ]),
            NetMsg::Hash {
                frame: 990,
                hash: 0xDEAD_BEEF_0BAD_F00D,
            },
            NetMsg::Start {
                seats: 6,
                seat: Some(3),
                terms: MatchTerms {
                    bots: 2,
                    teams: 1,
                    seed: 0x1234_5678_9ABC_DEF0,
                    ..MatchTerms::default()
                },
                names: std::array::from_fn(|i| wire_name(&format!("Seat {i}"))),
                kinds: [0; crate::sim::MAX_PLAYERS],
                standing: None,
                beach: b"a handmade beach".to_vec(),
            },
            // Mid-series: the round and the tally come back as they went,
            // and as a presence rather than as the zero that means none.
            NetMsg::Start {
                seats: 4,
                seat: Some(2),
                terms: MatchTerms {
                    series: 2,
                    ..MatchTerms::default()
                },
                names: [[0u8; WIRE_NAME]; crate::sim::MAX_PLAYERS],
                kinds: [0; crate::sim::MAX_PLAYERS],
                standing: Some(SeriesStanding {
                    round: 3,
                    wins: [1, 0, 1, 0, 0, 0],
                }),
                beach: Vec::new(),
            },
            NetMsg::Incompatible { version: 9 },
            NetMsg::Queued { ahead: 0 },
            NetMsg::Queued { ahead: 4 },
            NetMsg::Chat {
                name: wire_name("Anna"),
                text: wire_chat("wait for me!"),
            },
            NetMsg::Roster {
                seats: 4,
                names: std::array::from_fn(|i| wire_name(&format!("P{i}"))),
                kinds: [0; crate::sim::MAX_PLAYERS],
                terms: MatchTerms {
                    bots: 1,
                    map: 3,
                    ..MatchTerms::default()
                },
            },
            NetMsg::Abandoned { seat: 0, frame: 0 },
            NetMsg::Abandoned {
                seat: crate::sim::MAX_PLAYERS as u8 - 1,
                frame: 123_456,
            },
        ] {
            assert_eq!(NetMsg::decode(&msg.clone().encode()), Some(msg));
        }
    }

    /// A `Start` is the one message that sizes the table, and a joiner acts
    /// on it before anything else has looked at it: the seat count becomes
    /// the length every per-seat loop runs to, and the seat number indexes
    /// the lockstep's slots directly. Both are refused here rather than
    /// trusted, since a host on a broken build can say anything.
    #[test]
    fn a_start_that_seats_more_than_the_table_is_refused() {
        let good = NetMsg::Start {
            seats: 6,
            seat: Some(5),
            terms: MatchTerms::default(),
            names: [[0u8; WIRE_NAME]; crate::sim::MAX_PLAYERS],
            kinds: [0; crate::sim::MAX_PLAYERS],
            standing: None,
            beach: Vec::new(),
        };
        assert_eq!(NetMsg::decode(&good.clone().encode()), Some(good.clone()));
        // Byte 2 is `seats` and byte 3 is `seat`, just past the tag and
        // version. Nothing outside the table survives the trip.
        for (seats, seat) in [(7, 0), (255, 0), (1, 0), (0, 0), (4, 4), (4, 200)] {
            let mut bytes = good.clone().encode();
            bytes[2] = seats;
            bytes[3] = seat;
            assert_eq!(NetMsg::decode(&bytes), None, "{seats} seats, sat at {seat}");
        }
        // Watching is outside the seat range on purpose and stays legal,
        // and comes back as the absence it means rather than as the number
        // it travelled in.
        let mut watching = good.clone().encode();
        watching[3] = SPECTATOR_SEAT;
        assert!(matches!(
            NetMsg::decode(&watching),
            Some(NetMsg::Start { seat: None, .. })
        ));
        // An AI's chair is not one a joiner can be sat in either: the
        // lockstep it builds carries only the humans, and being seated
        // outside it was a panic, not a refusal.
        let with_bots = |seats, bots, seat| {
            NetMsg::decode(
                &NetMsg::Start {
                    seats,
                    seat: Some(seat),
                    terms: MatchTerms {
                        bots,
                        ..MatchTerms::default()
                    },
                    names: [[0u8; WIRE_NAME]; crate::sim::MAX_PLAYERS],
                    kinds: [0; crate::sim::MAX_PLAYERS],
                    standing: None,
                    beach: Vec::new(),
                }
                .encode(),
            )
        };
        assert!(with_bots(6, 4, 1).is_some(), "the last human seat");
        assert!(with_bots(6, 4, 2).is_none(), "the first AI seat");
        assert!(with_bots(6, 4, 3).is_none());
        assert!(with_bots(6, 6, 1).is_none(), "more bots than chairs");
        assert!(with_bots(6, 6, 0).is_some(), "seat zero always stands");
    }

    /// The whole point of the version byte: a datagram from another build is
    /// refused rather than half-understood.
    #[test]
    fn another_build_is_refused_and_told_so() {
        let mut hello = NetMsg::hello("Anna").encode();
        assert_eq!(hello[1], PROTOCOL_VERSION, "the version byte is byte 1");
        hello[1] = PROTOCOL_VERSION.wrapping_add(1);
        assert_eq!(NetMsg::decode(&hello), None, "not ours, so not acted on");
        assert_eq!(
            NetMsg::peek_version(&hello),
            Some(PROTOCOL_VERSION.wrapping_add(1)),
            "but it can still be identified, which is what gets it answered"
        );
        // The refusal itself is exempt from the gate: it is the one message
        // every version must be able to read, whatever the sender speaks.
        let refusal = NetMsg::Incompatible { version: 77 }.encode();
        assert_eq!(refusal.len(), 2, "its layout is frozen at two bytes");
        assert_eq!(
            NetMsg::decode(&refusal),
            Some(NetMsg::Incompatible { version: 77 })
        );
        // Stray traffic on the port is not a version clash and draws no reply.
        assert_eq!(NetMsg::peek_version(&[0xFE, 3]), None);
        assert_eq!(NetMsg::peek_version(b"PNCH1"), None, "an announcement");
    }
}

#[cfg(test)]
mod wire_fuzz_probe {
    use super::*;

    /// Every byte string a hostile or broken peer could put on the port,
    /// through both decoders. Neither may panic, hang, or allocate wildly:
    /// this is the one surface the LAN can reach without being invited.
    #[test]
    fn no_datagram_can_break_a_decoder() {
        let mut rng = crate::sim::Pcg32::new(0xDEAD_BEEF, 0x1357);
        // Seed with real messages, so mutations land near valid ones.
        let mut seeds: Vec<Vec<u8>> = vec![
            NetMsg::hello("Anna").encode(),
            NetMsg::watch("Dee").encode(),
            NetMsg::Queued { ahead: 3 }.encode(),
            NetMsg::Chat {
                name: wire_name("Bo"),
                text: wire_chat("ready?"),
            }
            .encode(),
            NetMsg::Start {
                seats: 6,
                seat: Some(2),
                terms: MatchTerms::default(),
                names: [[0u8; WIRE_NAME]; crate::sim::MAX_PLAYERS],
                kinds: [0; crate::sim::MAX_PLAYERS],
                standing: None,
                beach: Vec::new(),
            }
            .encode(),
            NetMsg::Hash { frame: 7, hash: 9 }.encode(),
            // The count byte is a length off the wire, and the one field
            // in the protocol that says how much to read.
            NetMsg::Inputs(vec![
                InputMsg {
                    player: 1,
                    frame: 9,
                    action: crate::sim::PlayerAction::None,
                };
                6
            ])
            .encode(),
            ANNOUNCE_MAGIC.to_vec(),
            Vec::new(),
        ];
        // And a beacon, built the way the announcer builds one.
        let mut beacon = ANNOUNCE_MAGIC.to_vec();
        beacon.extend_from_slice(&49213u16.to_le_bytes());
        beacon.push(BEACON_RUNNING);
        beacon.extend_from_slice(&wire_name("Room 3"));
        beacon.push(4);
        beacon.push(6);
        beacon.extend_from_slice(&0x5EA5u64.to_le_bytes());
        seeds.push(beacon);

        for round in 0..80_000u32 {
            let mut bytes = seeds[(round as usize) % seeds.len()].clone();
            for _ in 0..(rng.next_u32() % 6) + 1 {
                if bytes.is_empty() {
                    bytes.push((rng.next_u32() % 256) as u8);
                    continue;
                }
                let at = (rng.next_u32() as usize) % bytes.len();
                match rng.next_u32() % 4 {
                    0 => bytes[at] = (rng.next_u32() % 256) as u8,
                    1 => drop(bytes.remove(at)),
                    2 => bytes.insert(at, (rng.next_u32() % 256) as u8),
                    _ => bytes.truncate(at),
                }
            }
            // The message decoder, and the version peek that answers strays.
            if let Some(msg) = NetMsg::decode(&bytes) {
                // Anything decoded must survive a round trip, or the host
                // would relay something other than what it was told.
                assert_eq!(
                    NetMsg::decode(&msg.clone().encode()),
                    Some(msg.clone()),
                    "{bytes:?}"
                );
                // And a Start that decoded must be one we can actually seat.
                if let NetMsg::Start { seats, seat, .. } = msg {
                    assert!((2..=crate::sim::MAX_PLAYERS as u8).contains(&seats));
                    assert!(seat.is_none_or(|seat| seat < seats));
                }
                // A seat that decoded is a seat something will index by.
                if let NetMsg::Abandoned { seat, .. } = msg {
                    assert!(usize::from(seat) < crate::sim::MAX_PLAYERS);
                }
            }
            let _ = NetMsg::peek_version(&bytes);
            // Both names the beacon carries: the beach's, and the host's
            // behind it, which is the one that runs off a short packet.
            let _ = beacon_name(&bytes, bytes.len(), BEACON_NAME_AT);
            let _ = beacon_name(&bytes, bytes.len(), BEACON_HOST_AT);
        }
    }
}