pinch-points 0.6.0

//! Online versus, shell side: a UDP lockstep session driving the same
//! versus mode local play uses (spec §7.6 fallback path; see sim::net for
//! why lockstep rather than GGRS rollback today).
//!
//! Dev-grade session setup: `PINCH_HOST=<port>` hosts as player 0,
//! `PINCH_JOIN=<addr:port>` joins as player 1. Both boot straight into the
//! arena; the match starts once the handshake completes.

use crate::sim::{HASH_INTERVAL, Lockstep, MAX_PLAYERS, PlayerAction};
use crate::transport::{
    Announcer, MatchTerms, NetMsg, SeriesStanding, UdpTransport, chat_from_wire, name_from_wire,
    table_from_wire, wire_table,
};
use bevy::prelude::*;
pub use peers::{Peer, PeerBook, Place};
use std::collections::VecDeque;

/// How many ticks a resume is repeated after the local session unpauses.
/// A second's worth: enough to ride out a loss burst, and harmless if the
/// peers already heard the first one.
const RESUME_ECHOES: u8 = 30;

pub struct OnlineSession {
    pub transport: UdpTransport,
    pub session: Lockstep,
    /// Seats in the match, humans plus AI. The lockstep only carries the
    /// humans, so this is not `session.player_count()`.
    pub seats: u8,
    /// What the lobby agreed the match is: AI seats, map, gulls, round
    /// length, light, team scoring, and the board's seed. Every peer builds
    /// from these, so nothing about the round is a local opinion.
    pub terms: MatchTerms,
    /// The host's handmade beach, compressed, when the round is played on
    /// one. A generated arena travels as a seed in [`Self::terms`]; a
    /// level somebody built has to travel as itself, because no peer but
    /// the host has the file. Empty for the built-in maps.
    pub beach: Vec<u8>,
    /// The peers on the socket, one row each: the chair the launch plan
    /// gave it (`Queued` for one that turned up mid-round, and for every
    /// peer a joiner knows, since a joiner keeps no plan and answers
    /// nobody), what it calls itself, whether it asked to watch, and how
    /// long since it last said anything.
    ///
    /// The silence is the difference between a player who has *gone* and
    /// one who is merely behind: a machine still talking to us has not left
    /// the room, however far its inputs have fallen back. A joiner keeps
    /// the one row it has, the host, and calls the round off when that goes
    /// quiet.
    pub(crate) peers: PeerBook,
    /// What each seat is called, agreed at the handshake so every peer
    /// shows the same table. Empty entries fall back to seat labels; local
    /// couch names never apply to an online round.
    pub names: [String; MAX_PLAYERS],
    /// The peers' state hashes against our own, and the first frame they
    /// disagreed on, if ever.
    hashes: HashCheck,
    /// Whether the picture has stopped moving, and on whom.
    stall: StallWatch,
    /// Seats given up on this round, each with the frame it was emptied
    /// from, so the shell can put an AI in each and say so once rather
    /// than every frame, and the host can keep telling the table.
    pub abandoned: Vec<(u8, u32)>,
    /// Lines said to the table since the shell last read them, each with
    /// who said it.
    ///
    /// Drained rather than kept: the round's own feed is where a line is
    /// shown, and the feed keeps its own history.
    pub heard: Vec<(String, String)>,
    /// What the people without a chair are doing: the tide vote, the
    /// crowd's size, the calls for the winner.
    pub stands: crate::app::spectators::Stands,
    /// The beach this session came from, and what it owes it.
    pub home: Home,
    /// A next round has been agreed and the session is armed for it; the
    /// shell reads this and walks everyone back into the arena. Cleared by
    /// whoever acts on it.
    pub next_round: bool,
    /// Where the series stands as the next round begins, as the host said
    /// it: the 1-based round number and the wins per seat, both re-dealt to
    /// this round's chairs. The shell folds it into its `Tournament` on the
    /// way into the arena, so a peer admitted mid-series joins the table's
    /// standings rather than starting its own. `None` outside a series.
    pub series_standing: Option<SeriesStanding>,
    /// Ticks of `Resume` still to repeat (see [`RESUME_ECHOES`]).
    resume_echo: u8,
    /// A round joined late, or owed to someone who came late to it.
    pub catch_up: catch_up::CatchUp,
}

/// The beach a session came from: its beacon, what it is called, and
/// whether there is a lobby to walk the table back to.
pub struct Home {
    /// Host side: the beacon, carried out of the lobby so the beach stays
    /// on the air while the round runs, listed as in progress with its
    /// occupancy, for anyone who wants the next one. Silent on a joiner,
    /// and on a direct `PINCH_HOST` pair, which never announced at all.
    announcer: Farewell,
    /// What the beach is called, for the beacon it keeps up while the round
    /// runs. Not a player's name: the list is choosing between games.
    pub game_name: String,
    announce_in: f32,
    /// Whether this session was formed in the beach lobby, which is the
    /// only place a finished match has to walk its table back to: the
    /// direct `PINCH_HOST` pair was never anywhere else to begin with.
    pub from_lobby: bool,
    /// Joiner side: seconds until the next greeting while the results card
    /// is up.
    ///
    /// An established session says nothing at all between rounds: inputs
    /// and hashes belong to the round that ended, and the host has nothing
    /// to send until somebody calls the next one. Silence read as evidence
    /// is then a joiner walking out of a perfectly good table twenty
    /// seconds into a results card, so it keeps saying hello on the lobby's
    /// cadence, and the host's reply is what tells it the host is there.
    greet_in: f32,
}

impl Home {
    /// Nowhere in particular: the direct pair, which never announced and
    /// has no lobby behind it.
    fn nowhere() -> Home {
        Home {
            announcer: Farewell::silent(),
            game_name: String::new(),
            announce_in: 0.0,
            from_lobby: false,
            greet_in: 0.0,
        }
    }
}

/// Whether the round has stopped moving, for how long, and who the status
/// line is naming for it.
///
/// Measured by the frame number rather than by whether the next frame's
/// inputs happen to be in, because the second question is asked at a
/// moment where the answer is always no: the sim advances until it
/// *cannot*, and the tick that watches for stalls runs after it, in
/// `Update`. So it always found the next frame's slots empty and always
/// called a healthy round stalled.
#[derive(Default)]
struct StallWatch {
    /// Seconds the round has been unable to move, and the frame it has
    /// been unable to move off.
    stalled_for: f32,
    stalled_on: u32,
    /// Who the status line is naming, and how long it has left to keep
    /// naming them.
    ///
    /// A latch rather than a reading of `stalled_for`, because that number
    /// snaps back to zero the moment one frame gets through, and a limping
    /// round gets a frame through all the time. Reading it directly put the
    /// line on and off once a second.
    waiting_on: Option<u8>,
    waiting_hold: f32,
}

impl StallWatch {
    /// Nothing is stuck and nobody is being waited for, as of `frame`: a
    /// new round, or a round holding still on purpose.
    fn reset(&mut self, frame: u32) {
        let Self {
            stalled_for,
            stalled_on,
            waiting_on,
            waiting_hold,
        } = self;
        *stalled_for = 0.0;
        *stalled_on = frame;
        *waiting_hold = 0.0;
        *waiting_on = None;
    }
}

/// The peers' state hashes against our own.
#[derive(Default)]
struct HashCheck {
    /// First frame where the peers' state hashes disagreed, if ever: a
    /// determinism bug surfaced loudly rather than played through.
    desync_at: Option<u32>,
    own: VecDeque<(u32, u64)>,
    /// Peer hashes for frames we may not have simulated yet; compared as
    /// our own hashes appear, so no exchanged check is skipped.
    peers: VecDeque<(u32, u64)>,
}

impl HashCheck {
    /// Our own hash for `frame`. A short window: the peers are at most a
    /// commit lead behind.
    fn record_own(&mut self, frame: u32, hash: u64) {
        self.own.push_back((frame, hash));
        while self.own.len() > 16 {
            self.own.pop_front();
        }
    }

    /// A peer's hash for `frame`, which we may not have simulated yet.
    fn record_peer(&mut self, frame: u32, hash: u64) {
        self.peers.push_back((frame, hash));
        while self.peers.len() > 64 {
            self.peers.pop_front();
        }
    }

    /// Compare what has been exchanged, and remember the first frame that
    /// disagreed.
    fn compare(&mut self) {
        for &(frame, peer) in &self.peers {
            if let Some(&(_, own)) = self.own.iter().find(|(f, _)| *f == frame)
                && own != peer
            {
                self.desync_at.get_or_insert(frame);
            }
        }
    }

    /// Last round's disagreement is last round's; a new board starts
    /// level, and the hashes that proved it are gone with it.
    fn reset(&mut self) {
        *self = HashCheck::default();
    }
}

/// The goodbye a hosted beach owes the network when its session ends.
///
/// A hosted beach stops being announced when its session goes, whichever
/// way that happens: the match ending, the player quitting to the menu, a
/// desync giving up. Holding the duty here rather than at each of those
/// exits means no path drops an announcing session without either saying
/// goodbye or taking the announcer back out through
/// [`OnlineSession::back_to_the_lobby`], the one exit where the beach is
/// not going anywhere.
struct Farewell {
    /// `None` on a joiner and on the direct pair, which never announced,
    /// and in the shell a session leaves behind when its beach goes home.
    announcer: Option<Announcer>,
    /// The game port the beacons named, taken when the announcer came
    /// aboard: the port never changes after the bind, and it has to be
    /// readable at drop time.
    port: u16,
}

impl Farewell {
    /// A session with nothing to announce and so nothing to owe.
    fn silent() -> Farewell {
        Farewell {
            announcer: None,
            port: 0,
        }
    }

    /// The announcer, while one is aboard.
    fn aboard(&self) -> Option<&Announcer> {
        self.announcer.as_ref()
    }

    /// Take the announcer back out, disarming the goodbye: the beach is
    /// going back on the air, not away.
    fn disarm(mut self) -> Option<Announcer> {
        self.announcer.take()
    }
}

impl Drop for Farewell {
    fn drop(&mut self) {
        if let Some(announcer) = &self.announcer {
            announcer.closing(self.port);
        }
    }
}

/// What survives of a session when its table walks back to the lobby
/// together at the end of a match: the sockets, still connected, and the
/// bookkeeping the lobby needs to stand the beach back up.
pub struct LobbyReturn {
    /// Host side: the beacon, going back on the air as open. `None` for a
    /// joiner.
    pub announcer: Option<Announcer>,
    pub transport: UdpTransport,
    pub game_name: String,
    /// Host side: every registered peer by its socket index, with the
    /// name its greeting carried and whether it watches, by the same rule
    /// the next round would have seated it. Nobody holds a chair any
    /// more: the lobby deals them again at the launch.
    pub peers: PeerBook,
    /// Joiner side: this peer came to watch, and greets with `Watch`.
    pub watching: bool,
    pub host: bool,
    /// The seed of the round just played. A host still on its results
    /// card re-answers every greeting with that round's `Start`, and a
    /// lobby that has just walked out of it must read the repeat as
    /// stale rather than as an invitation straight back in.
    pub played_seed: u64,
}

#[derive(Resource, Default)]
pub struct Online(pub Option<OnlineSession>);

/// Pass `msg` from peer `from` on to every other peer. Star topology: the
/// spokes cannot hear each other, so whatever one says reaches the rest
/// only by the hub repeating it, and inputs, pauses, resumes and chat
/// are all repeated this same way.
pub(crate) fn relay(transport: &UdpTransport, from: usize, msg: NetMsg) {
    for other in 0..transport.peer_count() {
        if other != from {
            transport.send_to(other, msg.clone());
        }
    }
}

/// Drain the session while the results card is up. See
/// [`OnlineSession::poll_between_rounds`] for why this cannot wait for the
/// sim: the sim is exactly what has stopped.
pub fn poll_between_rounds(time: Res<Time>, mut online: ResMut<Online>) {
    if let Some(session) = &mut online.0 {
        session.poll_between_rounds(time.delta_secs());
    }
}

pub(crate) mod catch_up;
mod crowd;
mod peers;
mod presence;
mod rounds;
pub(crate) use presence::{abandon_the_departed, catch_up_again, leave_a_hostless_round};
pub use rounds::Invitation;
pub(crate) use rounds::lockstep_for;

impl OnlineSession {
    pub fn new(
        transport: UdpTransport,
        session: Lockstep,
        seats: u8,
        terms: MatchTerms,
    ) -> OnlineSession {
        OnlineSession {
            transport,
            session,
            seats,
            terms,
            beach: Vec::new(),
            peers: PeerBook::default(),
            names: Default::default(),
            hashes: HashCheck::default(),
            stall: StallWatch::default(),
            abandoned: Vec::new(),
            heard: Vec::new(),
            stands: Default::default(),
            home: Home::nowhere(),
            next_round: false,
            series_standing: None,
            resume_echo: 0,
            catch_up: Default::default(),
        }
    }

    /// Keep the beach on the air while the round runs, so a lobby can list
    /// it as in progress rather than not at all. Hosts only, and the beacon
    /// says running, so nobody is offered a join that lockstep could not
    /// honour; what it offers is a place in the queue.
    ///
    /// `taken` counts the humans, not the table: an AI seat gives way to a
    /// player who wants it. The humans in line for the next round count
    /// too, since the chair each will take is spoken for, and a beacon that
    /// left the queue out let a sixth player in to be dealt a watching place with
    /// no notice.
    pub fn keep_announcing(&mut self, delta: f32) {
        if self.home.announcer.aboard().is_none()
            || !crate::app::lobby::once_a_second(&mut self.home.announce_in, delta)
        {
            return;
        }
        let taken = self.players_spoken_for();
        let (Some(announcer), Ok(addr)) =
            (self.home.announcer.aboard(), self.transport.local_addr())
        else {
            return;
        };
        announcer.running(
            addr.port(),
            crate::transport::OnAir {
                name: &self.home.game_name,
                // Seat zero is the host's, and the host is the only one that
                // announces.
                host: &self.names[0],
                taken,
                seats: MAX_PLAYERS as u8,
            },
        );
    }

    /// Host: carry the beacon out of the lobby, and with it the goodbye
    /// owed to the network from now on.
    pub fn stay_on_air(&mut self, announcer: Announcer) {
        // Port 0 if the socket will not say: the goodbye is matched by
        // the announcer's id first, so it still clears the right row.
        let port = self
            .transport
            .local_addr()
            .map(|addr| addr.port())
            .unwrap_or(0);
        self.home.announcer = Farewell {
            announcer: Some(announcer),
            port,
        };
    }

    /// Dismantle a finished match into what the lobby needs to stand the
    /// beach back up, goodbye disarmed: this table is going back to the
    /// lobby together, not away.
    pub fn back_to_the_lobby(mut self) -> LobbyReturn {
        let host = self.is_host();
        let watching = self.watching();
        // Each peer under the name its chair carried, which is the one
        // every screen showed, and the plan's chairs given up: back in the
        // lobby the plan is dealt again, and only who watches survives it.
        for peer in 0..self.peers.len() {
            if let Some(name) = self.peer_name(peer) {
                let name = name.to_string();
                self.peers.row(peer).name = name;
            }
        }
        self.peers.unseat();
        let Self {
            transport,
            session: _,
            seats: _,
            terms,
            beach: _,
            peers,
            names: _,
            hashes: _,
            stall: _,
            abandoned: _,
            heard: _,
            stands: _,
            home,
            next_round: _,
            series_standing: _,
            resume_echo: _,
            catch_up: _,
        } = self;
        let Home {
            announcer,
            game_name,
            announce_in: _,
            from_lobby: _,
            greet_in: _,
        } = home;
        LobbyReturn {
            announcer: announcer.disarm(),
            transport,
            game_name,
            peers,
            watching,
            host,
            played_seed: terms.seed,
        }
    }

    /// The first frame the peers' state hashes disagreed on, if ever.
    pub fn desync_at(&self) -> Option<u32> {
        self.hashes.desync_at
    }

    /// Call a pause, and tell the peers which frame it lands on. A
    /// spectator's Escape opens their own menu; the match plays on behind
    /// it, and they are still in step when they close it, so the lockstep
    /// names no frame and nothing is sent.
    pub fn request_pause(&mut self) {
        if let Some(frame) = self.session.request_pause() {
            self.transport.send(NetMsg::Pause { frame });
        }
    }

    /// Play on, and keep saying so for a moment (see `RESUME_ECHOES`).
    pub fn request_resume(&mut self) {
        if self.watching() {
            return;
        }
        let frame = self.session.resume();
        self.resume_echo = RESUME_ECHOES;
        self.transport.send(NetMsg::Resume { frame });
    }

    /// The `Resume` to repeat: the pause most recently lifted here.
    fn resume_msg(&self) -> NetMsg {
        NetMsg::Resume {
            frame: self.session.lifted_pause().unwrap_or(0),
        }
    }

    /// Is this session the star's hub? Seat 0 hosts and relays. Public
    /// because the shell has to know who calls the next round.
    pub fn is_host(&self) -> bool {
        self.session.seat() == Some(0)
    }

    /// Whether this peer watches rather than plays.
    pub fn watching(&self) -> bool {
        self.session.watching()
    }

    /// The humans at the table and in line for it: the players in the
    /// lockstep plus every queued peer that did not ask to watch. What the
    /// running beacon reports as taken.
    pub(crate) fn players_spoken_for(&self) -> u8 {
        let queued = self
            .peers
            .iter()
            .filter(|peer| peer.place == Place::Queued && !peer.watch)
            .count();
        (self.session.player_count() + queued).min(MAX_PLAYERS) as u8
    }

    /// One fixed-tick's worth of network pumping: commit and (re)send local
    /// input, drain the socket (the host relays joiner inputs to the other
    /// joiners and re-answers stray hellos with their seat), then simulate
    /// every frame that has complete inputs via `tick`. Records hashes on
    /// the [`HASH_INTERVAL`] cadence. Returns whether the local action was
    /// committed (false = at the commit lead; retry it next tick).
    pub fn pump(&mut self, local_action: PlayerAction, mut tick: impl FnMut(&mut Self)) -> bool {
        let committed = self.session.commit_local(local_action).is_some();
        // The newest commit and the whole resend tail behind it, in one
        // datagram: see `NetMsg::Inputs` for what that is worth at a full
        // table. To the table and the watchers, never to the queue: a peer in
        // line is not simulating anything (`PeerBook::follows_the_round`).
        let peers = &self.peers;
        self.transport
            .send_inputs(self.session.recent_commits(), |peer| {
                peers.follows_the_round(peer)
            });
        // Pause state is repeated every tick rather than sent once: UDP
        // drops, and the peer that misses a Pause would otherwise sit
        // watching a frozen beach with no card, while a missed Resume would
        // leave the session stopped for good. Both self-heal here: the
        // resume repeats until frames actually start moving again.
        match self.session.pause_frame() {
            Some(frame) => self.transport.send(NetMsg::Pause { frame }),
            None if self.resume_echo > 0 => {
                self.resume_echo -= 1;
                self.transport.send(self.resume_msg());
            }
            None => {}
        }
        let host = self.is_host();
        // Set when the host's call for the next round arrives in this very
        // pump, which is possible for a joiner or spectator still on the
        // last frames of the round the host has finished. The lockstep is
        // fresh from frame zero then, and the board is still the old one:
        // ticking would play the new round's first frames out on it, and
        // the joiner would enter the arena two frames ahead of every other
        // table, a desync at the first hash.
        let mut rearmed = false;
        // Likewise every seat given up on, for the rest of the round: a
        // joiner that misses the one notice waits on that seat forever,
        // and the host never notices, because a joiner that is waiting is
        // still talking. Five bytes a seat a tick.
        if host {
            for &(seat, frame) in &self.abandoned {
                self.transport.send(NetMsg::Abandoned { seat, frame });
            }
        }
        for (msg, from) in self.transport.recv_all() {
            // Whatever it was, that peer is still there, which is all
            // `abandon_stalled` and `host_gone` go on.
            self.mark_heard(from);
            match msg {
                // A peer that missed the Start datagram keeps greeting us;
                // repeat what it is until it stops. A watcher is told it is
                // watching, so the seat it gets is not one.
                NetMsg::Hello { name } => {
                    if host {
                        let answer = self.answer_greeting(from, &name_from_wire(&name), false);
                        self.transport.send_to(from, answer);
                    }
                }
                NetMsg::Watch { name } => {
                    // Somebody who came to watch after the launch is sent
                    // the round as it stands, once this tick's frames are
                    // in, and watches from there (`catch_up`). One from the
                    // launch is told its place again, as ever. A greeting
                    // repeats until the watcher is in, so a snapshot that
                    // went astray is simply sent again, fresher.
                    if host && self.late_watcher(from) {
                        let told = name_from_wire(&name);
                        if !told.is_empty() {
                            self.remember_peer_name(from, &told);
                        }
                        self.note_watch_wish(from);
                        self.peers.row(from).place = Place::LateWatching;
                        if !self.catch_up.owed.contains(&from) {
                            self.catch_up.owed.push(from);
                        }
                    } else if host {
                        let answer = self.answer_greeting(from, &name_from_wire(&name), true);
                        self.transport.send_to(from, answer);
                    }
                }
                NetMsg::Inputs(mut inputs) => {
                    // The host knows which seat each peer was given, and
                    // takes inputs for that seat alone: a peer speaking for
                    // another's seat (a bug, or a spectator with ideas)
                    // would otherwise be believed by whoever heard it
                    // first, and the seats would diverge. A joiner hears
                    // only the host, which has already done this. The
                    // direct `PINCH_HOST` pair keeps no plan, having never
                    // been through a lobby; with nothing to check against
                    // it takes any seat but its own.
                    //
                    // Weeded before anything is believed *or* passed on:
                    // what the host relays is what it accepted.
                    if host {
                        let planned = self.peers.planned();
                        let own = self.session.seat();
                        let theirs = self.seat_of(from);
                        inputs.retain(|input| match planned {
                            0 => Some(input.player) != own,
                            _ => theirs == Some(input.player),
                        });
                    }
                    for &input in &inputs {
                        self.session.receive(input);
                    }
                    if host {
                        let peers = &self.peers;
                        self.transport.send_inputs(&inputs, |peer| {
                            peer != from && peers.follows_the_round(peer)
                        });
                    }
                }
                NetMsg::Hash { frame, hash } => {
                    // The peer may be ahead of us; buffer until we simulate
                    // that frame ourselves.
                    self.hashes.record_peer(frame, hash);
                }
                NetMsg::Pause { frame } => {
                    self.session.receive_pause(frame);
                    if host {
                        relay(&self.transport, from, NetMsg::Pause { frame });
                    }
                }
                NetMsg::Resume { frame } => {
                    let frame = self.session.receive_resume(frame);
                    self.resume_echo = RESUME_ECHOES;
                    if host {
                        relay(&self.transport, from, NetMsg::Resume { frame });
                    }
                }
                NetMsg::Start {
                    seats,
                    seat,
                    terms,
                    names,
                    standing,
                    beach,
                } if !host && self.is_next_round(&terms) => {
                    // The host has called the next round. A fresh seed is
                    // what says so; everything else about the round may
                    // well be the same. Taking it up here only rearms the
                    // session; the board is rebuilt on the way back into
                    // the arena, by the same path a first round uses.
                    self.take_up(Invitation {
                        seats,
                        seat,
                        terms,
                        names,
                        standing,
                        beach,
                    });
                    self.next_round = true;
                    rearmed = true;
                }
                NetMsg::Start { names, .. } => {
                    // Stale as a launch signal, but a joiner in the direct
                    // PINCH_HOST/PINCH_JOIN pair, which never waits in a
                    // lobby, learns the table's names from it.
                    if !host {
                        for (slot, name) in self.names.iter_mut().zip(table_from_wire(&names)) {
                            if !name.is_empty() {
                                *slot = name;
                            }
                        }
                    }
                }
                // Someone on another build is talking to this port. Every
                // peer in a running match paired before it started, so this
                // is a stranger, not a member: the match plays on without it.
                NetMsg::Incompatible { .. } => {}
                // Only a host hands these out, and a host never receives
                // one: a joiner in the queue is still in the lobby, and so
                // is a watcher being caught up.
                NetMsg::Queued { .. } | NetMsg::CatchUp { .. } => {}
                // A spectator talking. Players have both hands on the keys
                // and no way to answer, which is the point: the people
                // with nothing to hold are the ones who can say something,
                // and the round's own feed is where it lands.
                NetMsg::Chat { name, text } => {
                    let said = self.take_chat(host, from, name, text);
                    if let Some(said) = said {
                        self.heard.push(said);
                    }
                }
                // Only the host counts. Every peer hears the answer, and
                // hears it as an action on a frame rather than as a claim
                // about one, so there is nothing here for anyone else to
                // agree with or miss.
                // The host's word on where the vote stands. It counts, so
                // it says; nobody else has anything to compare it against.
                NetMsg::SpectatorTally {
                    watching,
                    open,
                    wait,
                } => {
                    if !host {
                        self.stands.crowd = crate::app::spectators::Crowd {
                            watching,
                            open,
                            wait,
                        };
                    }
                }
                // Counted only from someone who is actually watching this
                // round. The sender was never looked at, so a seated
                // player could call tide events down on the table it was
                // playing at, and a peer in line for the next round, which
                // is sent no frames and has no board on screen, could vote
                // on this one. With nobody watching at all, one stray
                // datagram had the beach announcing what the spectators
                // had called.
                NetMsg::SpectatorVote { event } => self.take_vote(host, from, event),
                NetMsg::SpectatorPick { seat } => self.take_pick(host, from, seat),
                // The host's word on the calls, like the tally.
                NetMsg::CrowdPicks { picks, open } => {
                    if !host {
                        self.stands.picks = crate::app::spectators::Picks {
                            counts: picks,
                            open,
                        };
                    }
                }
                NetMsg::Abandoned { seat, frame } => {
                    // The host's word, not our own patience. Idempotent,
                    // because it is repeated against packet loss.
                    if !host && !self.abandoned.iter().any(|(gone, _)| *gone == seat) {
                        self.session.abandon(seat, frame);
                        self.abandoned.push((seat, frame));
                    }
                }
                // The lobby's business, and the lobby is behind us.
                NetMsg::Roster { .. } => {}
            }
        }
        // One row per peer the socket has, and never a row for one it
        // does not: rows are grown for a peer the socket just indexed and
        // dropped with it when it is forgotten. A row past the socket's
        // list is somebody else's seat, name or watch wish.
        debug_assert!(
            self.peers.len() <= self.transport.peer_count(),
            "{} peer rows for {} peers",
            self.peers.len(),
            self.transport.peer_count()
        );
        self.hashes.compare();
        if rearmed {
            return committed;
        }
        // At most a couple of frames per fixed tick: a lagging peer catches
        // up gradually instead of spiralling.
        for _ in 0..2 {
            tick(self);
        }
        self.hashes.compare();
        committed
    }

    /// Called by the tick closure after simulating a frame, with the fresh
    /// state hash.
    pub fn after_frame(&mut self, hash: u64) {
        let frame = self.session.frame();
        if frame.is_multiple_of(HASH_INTERVAL) {
            self.hashes.record_own(frame, hash);
            self.transport.send(NetMsg::Hash { frame, hash });
        }
    }
}

/// Parse the dev env vars into a ready session, if requested.
/// `PINCH_HOST=47777` or `PINCH_JOIN=192.168.1.10:47777`.
pub fn session_from_env() -> Option<OnlineSession> {
    // The direct hooks skip the lobby, so there is no Start to agree on:
    // both processes must be told the same terms. `PINCH_BOTS=n` seats n AI
    // players behind the two humans and has to be set on both sides.
    let bots: u8 = crate::app::dev::bots().unwrap_or(0);
    // Two humans and the AI behind them; MAX_PLAYERS is the ceiling, and
    // the clamp is on the bots so a wild PINCH_BOTS cannot wrap the sum
    // back under the two humans.
    let humans = 2u8;
    let bots = bots.min(MAX_PLAYERS as u8 - humans);
    let seats = humans + bots;
    let terms = MatchTerms {
        bots,
        ..crate::app::match_setup::terms(
            &crate::app::match_setup::MatchConfig::default(),
            crate::app::teams::TeamMode::Solo,
            0,
        )
    };
    // The name this player goes by online: their P1 name from settings,
    // read straight off disk because the dev hooks run before the app's
    // resources are wired up.
    let own = crate::app::settings::GameSettings::load().0.names[0].clone();
    if let Some(port) = crate::app::dev::direct_host() {
        let port: u16 = port.parse().ok()?;
        let transport = UdpTransport::host(port).ok()?;
        info!("hosting on UDP port {port}, waiting for a peer ({seats} seats)");
        let mut session = OnlineSession::new(
            transport,
            Lockstep::new(0, vec![0, 1], crate::sim::DEFAULT_DELAY),
            seats,
            terms,
        );
        session.names[0] = own;
        return Some(session);
    }
    if let Some(addr) = crate::app::dev::direct_join() {
        let transport = UdpTransport::join(addr.as_str()).ok()?;
        transport.send(NetMsg::hello(&own));
        info!("joining {addr} ({seats} seats)");
        let mut session = OnlineSession::new(
            transport,
            Lockstep::new(1, vec![0, 1], crate::sim::DEFAULT_DELAY),
            seats,
            terms,
        );
        session.names[1] = own;
        return Some(session);
    }
    None
}

#[cfg(test)]
mod homecoming_tests {
    use super::*;
    use crate::sim::DEFAULT_DELAY;
    use crate::transport::{Beacon, Discovery};

    fn hosting_session(seed: u64) -> OnlineSession {
        OnlineSession::new(
            UdpTransport::host(0).expect("game socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            MatchTerms {
                seed,
                ..MatchTerms::default()
            },
        )
    }

    /// Beacons from other tests share the machine, so every packet is
    /// judged by whether it names our game port.
    fn heard_for(discovery: &mut Discovery, port: u16) -> Vec<Beacon> {
        let mut heard = Vec::new();
        for _ in 0..20 {
            std::thread::sleep(std::time::Duration::from_millis(10));
            heard.extend(
                discovery
                    .poll()
                    .into_iter()
                    .filter(|(addr, _)| addr.port() == port)
                    .map(|(_, beacon)| beacon),
            );
            if !heard.is_empty() {
                break;
            }
        }
        heard
    }

    /// The goodbye still fires on every ordinary way out: quitting to the
    /// menu, a desync giving up, the process letting go.
    #[test]
    fn a_dropped_session_still_says_goodbye() {
        let mut discovery = Discovery::bind().expect("lobby port");
        let mut session = hosting_session(0);
        let port = session.transport.local_addr().expect("addr").port();
        session.stay_on_air(Announcer::new(0xD00D).expect("announcer"));
        drop(session);
        let heard = heard_for(&mut discovery, port);
        let beacon = heard.first().expect("the goodbye went out");
        assert!(matches!(beacon, Beacon::Closing { .. }), "{beacon:?}");
    }

    /// The one exit where the beach is not going anywhere: walking the
    /// table back to the lobby keeps the socket, keeps the beacon, and
    /// says no goodbye, because the same beach is about to announce
    /// itself open again.
    #[test]
    fn the_way_back_to_the_lobby_says_no_goodbye() {
        let mut discovery = Discovery::bind().expect("lobby port");
        let mut session = hosting_session(42);
        let port = session.transport.local_addr().expect("addr").port();
        session.stay_on_air(Announcer::new(0xB0A7).expect("announcer"));
        session.home.game_name = "Room 3".into();
        session.home.from_lobby = true;
        let returned = session.back_to_the_lobby();
        assert!(returned.host);
        assert!(!returned.watching);
        assert_eq!(returned.played_seed, 42, "the round it walked out of");
        assert_eq!(returned.game_name, "Room 3");
        assert_eq!(
            returned.transport.local_addr().expect("addr").port(),
            port,
            "the same socket every peer is still talking to"
        );
        assert!(
            returned.announcer.is_some(),
            "the beacon rides home to announce again"
        );
        std::thread::sleep(std::time::Duration::from_millis(40));
        assert!(
            discovery
                .poll()
                .into_iter()
                .all(|(addr, _)| addr.port() != port),
            "no goodbye for a beach that is coming home"
        );
    }

    /// Who counts as watching this round, which decides both whose vote
    /// the host takes and what number the table is shown.
    ///
    /// Not `Peer::watches`, which is a wish as much as a place: it is true
    /// of somebody in line for the *next* round, who is sent no frames and
    /// sees nothing, and of a seated player that has asked to watch next
    /// time while still playing this one. Counting either had a player in
    /// the audience for its own round, and let a peer with no board vote
    /// on what happened on it.
    #[test]
    fn only_someone_in_the_round_without_a_chair_is_watching_it() {
        let mut session = hosting_session(11);
        // Three peers in the launch plan: one seated, one watching, one
        // seated but asking to watch next round. Then a fourth in line.
        session.peers.row(0).place = Place::Seated(1);
        session.peers.row(1).place = Place::Watching;
        session.peers.row(2).place = Place::Seated(2);
        session.peers.row(2).watch = true;
        session.peers.row(3).place = Place::Queued;
        session.peers.row(3).watch = true;

        assert!(!session.watching_this_round(0), "a player");
        assert!(session.watching_this_round(1), "a watcher");
        assert!(
            !session.watching_this_round(2),
            "still playing, however it means to spend the next round"
        );
        assert!(
            !session.watching_this_round(3),
            "in line, sent no frames, watching nothing"
        );
        assert_eq!(session.watchers_in_round(), 1, "one of the four");
    }

    /// A pick is counted only from someone watching this round, and only
    /// by the host, which is the only peer that counts anything.
    #[test]
    fn a_call_for_the_winner_is_taken_from_a_watcher_while_the_calls_are_open() {
        use crate::sim::InputMsg;
        let mut session = hosting_session(5);
        session.peers.row(0).place = Place::Seated(1);
        session.peers.row(1).place = Place::Watching;
        session.peers.row(2).place = Place::LateWatching;
        session.peers.row(3).place = Place::Queued;
        session.take_pick(true, 0, 0);
        session.take_pick(true, 1, 1);
        session.take_pick(true, 2, 1);
        session.take_pick(true, 3, 0);
        session.take_pick(true, 1, 9);
        let picks = session.crowd_picks_now();
        assert_eq!(
            picks.counts,
            [0, 2, 0, 0, 0, 0],
            "two watchers; the player and the queue unheard"
        );
        assert!(picks.open > 0);
        // A spectator may change its mind, and a spoke counts nothing.
        session.take_pick(true, 1, 0);
        session.take_pick(false, 2, 0);
        assert_eq!(session.crowd_picks_now().counts, [1, 1, 0, 0, 0, 0]);

        // Past the first half minute the calls are in.
        let close = crate::app::spectators::PICKS_OPEN_FOR;
        for frame in 0..close {
            session.session.commit_local(PlayerAction::None);
            session.session.receive(InputMsg {
                player: 1,
                frame: DEFAULT_DELAY + frame,
                action: PlayerAction::None,
            });
            while session.session.advance().is_some() {}
        }
        assert!(session.session.frame() >= close);
        assert_eq!(session.crowd_picks_now().open, 0);
        session.take_pick(true, 1, 1);
        assert_eq!(
            session.crowd_picks_now().counts,
            [1, 1, 0, 0, 0, 0],
            "a call after the close is not taken"
        );
    }

    #[test]
    fn a_vote_is_taken_only_from_someone_watching() {
        let sand = crate::sim::TideEvent::FreshSand.index() as u8;
        let open = |session: &OnlineSession| session.stands.votes.open().is_some();

        // Seated: a player may not call the tide down on its own table.
        let mut session = hosting_session(1);
        session.peers.row(0).place = Place::Seated(1);
        session.take_vote(true, 0, sand);
        assert!(!open(&session), "a player at the table");

        // In line for the next round: sent no frames, sees no board.
        let mut session = hosting_session(2);
        session.peers.row(0).place = Place::Queued;
        session.take_vote(true, 0, sand);
        assert!(!open(&session), "a peer in the queue");

        // Watching: heard.
        let mut session = hosting_session(3);
        session.peers.row(0).place = Place::Watching;
        session.take_vote(true, 0, sand);
        assert!(open(&session), "a spectator");

        // And a spoke counts nothing, whoever sent it: only the hub tallies.
        let mut session = hosting_session(4);
        session.peers.row(0).place = Place::Watching;
        session.take_vote(false, 0, sand);
        assert!(!open(&session), "not the host");
    }

    /// A direct `PINCH_HOST` pair keeps no launch plan, and
    /// `follows_the_round` reads an empty plan as "everybody hears" so the
    /// two keep talking to the one peer they have. Spectating is the place
    /// a peer was dealt, not something derived from that: read the other
    /// way, the host had its opponent down as a spectator, counted in the
    /// audience and allowed to vote on the game it was playing.
    #[test]
    fn a_direct_pair_has_no_spectators_in_it() {
        let mut session = hosting_session(5);
        // No plan was ever dealt, so the one peer keeps the default place.
        session.peers.reach(1);
        assert_eq!(session.peers.planned(), 0, "no plan at all");
        assert!(
            session.peers.follows_the_round(0),
            "and it is sent the round regardless, which is the point"
        );

        assert!(!session.watching_this_round(0), "it is the other player");
        assert_eq!(session.watchers_in_round(), 0, "an audience of nobody");

        session.take_vote(true, 0, crate::sim::TideEvent::FreshSand.index() as u8);
        assert!(
            session.stands.votes.open().is_none(),
            "and it cannot call the tide onto the game it is playing"
        );
    }

    /// A spectator that walks away is forgotten, which nothing used to do:
    /// the round gives up only on *seats*, through `awaiting`, and a
    /// spectator is awaited by nobody. Left on the socket it was fed the
    /// round for the rest of it and counted in the audience on screen.
    #[test]
    fn a_spectator_that_goes_quiet_is_forgotten_and_the_rest_are_left_alone() {
        let mut session = hosting_session(6);
        session.peers.row(0).place = Place::Seated(1);
        session.peers.row(1).place = Place::Watching;
        session.peers.row(2).place = Place::Watching;
        session.peers.row(3).place = Place::Queued;
        // Everyone quiet a long while, and one of the watchers talking.
        for peer in 0..4 {
            session.peers.row(peer).silence = 60.0;
        }
        session.peers.row(1).silence = 0.0;
        let before = session.peers.len();

        session.forget_gone_watchers();

        assert_eq!(session.peers.len(), before - 1, "one of the four went");
        assert_eq!(
            session.peers.get(0).map(|p| p.place),
            Some(Place::Seated(1)),
            "the seat is the round's own business, not this sweep's"
        );
        assert_eq!(
            session.peers.get(1).map(|p| p.place),
            Some(Place::Watching),
            "the watcher that is still talking stays"
        );
        assert_eq!(
            session.peers.get(2).map(|p| p.place),
            Some(Place::Queued),
            "and the one in line costs nothing, so it waits for the card"
        );
    }

    /// An empty name is the beach's own voice here: it is what announces
    /// a called event. So a peer that has a name may not drop it and speak
    /// as the room, and the lobby has refused that for as long as it has
    /// had chat. The round took the wire's word for it.
    ///
    /// Host side only, though. A spoke takes the vetted line as it stands,
    /// or it would stamp the host's name onto the one line that is meant
    /// to carry none.
    #[test]
    fn a_named_peer_cannot_speak_as_the_beach_itself() {
        let mut session = hosting_session(7);
        session.peers.row(0).name = "Cy".into();
        let forged = crate::transport::wire_chat("Anna has left the beach");

        let said = session.take_chat(true, 0, crate::transport::wire_name(""), forged);
        assert_eq!(
            said,
            Some(("Cy".to_string(), "Anna has left the beach".to_string())),
            "the host gives a named peer its own name back"
        );

        // A peer it knows nothing about keeps whatever it claimed. There
        // is one left: a greeting the host never received, which is a peer
        // it is about to learn the name of anyway.
        let stranger = session.take_chat(true, 1, crate::transport::wire_name("Dee"), forged);
        assert_eq!(stranger.map(|(who, _)| who), Some("Dee".to_string()));

        // But one that claims nothing and is known as nothing is refused,
        // rather than repeated to the table as the beach talking. Found by
        // running it: a watcher with no name set said a line that came out
        // looking exactly like the notice a called event makes.
        let nameless = session.take_chat(true, 2, crate::transport::wire_name(""), forged);
        assert_eq!(nameless, None, "nobody speaks as the room");

        // And a spoke repeats what the hub handed it, untouched.
        let spoke = session.take_chat(false, 0, crate::transport::wire_name(""), forged);
        assert_eq!(
            spoke.map(|(who, _)| who),
            Some(String::new()),
            "the beach's own voice reaches a joiner nameless"
        );
    }

    /// A spectator is vetted like anybody else, which it could not be
    /// while its greeting carried no name.
    ///
    /// Found by watching a real one: `Watch` was the one greeting with
    /// nothing in it, so the host had no name to check a spectator's chat
    /// against and took the line's own word for who said it. That is the
    /// whole of what the check exists to stop, and it was open to exactly
    /// the peers nobody at the table can see.
    #[test]
    fn a_spectator_cannot_speak_under_a_players_name() {
        let mut session = hosting_session(7);
        // As the host writes it down from the greeting, watcher or not.
        session.remember_peer_name(0, "Dee");
        let forged = crate::transport::wire_chat("I fold, take my castle");

        let said = session.take_chat(true, 0, crate::transport::wire_name("Bo"), forged);
        assert_eq!(
            said,
            Some(("Dee".to_string(), "I fold, take my castle".to_string())),
            "the crowd speaks under the name it turned up with"
        );
    }

    /// What the lobby gets back to seat people with: each peer's name as
    /// its greeting carried it, and the watchers by the same rule the next
    /// round would have used - a spectator's `None` in the launch plan, or
    /// a watch wish greeted mid-round.
    #[test]
    fn the_table_walks_back_with_its_names_and_watchers() {
        let mut session = hosting_session(0);
        let port = session.transport.local_addr().expect("addr").port();
        // Two peers, registered one at a time so their indices are known:
        // Bo plays, and somebody nameless watches.
        let bo = UdpTransport::join(("127.0.0.1", port)).expect("join");
        bo.send(NetMsg::hello("Bo"));
        let watcher = UdpTransport::join(("127.0.0.1", port)).expect("join");
        for want in [1, 2] {
            if want == 2 {
                watcher.send(NetMsg::watch("Dee"));
            }
            for _ in 0..40 {
                std::thread::sleep(std::time::Duration::from_millis(5));
                for (msg, from) in session.transport.recv_all() {
                    match msg {
                        NetMsg::Hello { name } => {
                            let told = name_from_wire(&name);
                            session.remember_peer_name(from, &told);
                        }
                        NetMsg::Watch { .. } => session.note_watch_wish(from),
                        NetMsg::Inputs(_)
                        | NetMsg::Hash { .. }
                        | NetMsg::Start { .. }
                        | NetMsg::Pause { .. }
                        | NetMsg::Resume { .. }
                        | NetMsg::Queued { .. }
                        | NetMsg::SpectatorVote { .. }
                        | NetMsg::SpectatorTally { .. }
                        | NetMsg::Chat { .. }
                        | NetMsg::Roster { .. }
                        | NetMsg::Abandoned { .. }
                        | NetMsg::CatchUp { .. }
                        | NetMsg::SpectatorPick { .. }
                        | NetMsg::CrowdPicks { .. }
                        | NetMsg::Incompatible { .. } => {}
                    }
                }
                if session.transport.peer_count() >= want {
                    break;
                }
            }
            assert_eq!(session.transport.peer_count(), want, "peer registered");
        }
        session.peers.deal(&[Some(1), None]);
        let returned = session.back_to_the_lobby();
        let names: Vec<&str> = returned.peers.iter().map(|p| p.name.as_str()).collect();
        assert_eq!(
            names,
            ["Bo", ""],
            "each chair keeps the name its greeting carried"
        );
        let watching: Vec<bool> = returned.peers.iter().map(Peer::watches).collect();
        assert_eq!(watching, [false, true], "and a watcher stays a watcher");
    }

    /// A joiner's way back: no beacon to carry, but the socket, the seed
    /// it played on, and whether it was watching all come home with it.
    #[test]
    fn a_watching_joiner_comes_home_still_watching() {
        let session = OnlineSession::new(
            UdpTransport::join(("127.0.0.1", 47999)).expect("join"),
            Lockstep::observer(vec![0, 1], DEFAULT_DELAY),
            2,
            MatchTerms {
                seed: 7,
                ..MatchTerms::default()
            },
        );
        let returned = session.back_to_the_lobby();
        assert!(!returned.host);
        assert!(returned.watching, "the watcher is remembered");
        assert!(returned.announcer.is_none(), "a joiner has no beacon");
        assert_eq!(returned.played_seed, 7);
    }
}

#[cfg(test)]
mod table_tests {
    use super::*;
    use crate::sim::DEFAULT_DELAY;

    fn session(local: Option<u8>, port: u16) -> OnlineSession {
        let transport = match local {
            Some(0) => UdpTransport::host(0).expect("host socket"),
            _ => UdpTransport::join(("127.0.0.1", port)).expect("join"),
        };
        OnlineSession::new(
            transport,
            match local {
                Some(seat) => Lockstep::new(seat, vec![0, 1], DEFAULT_DELAY),
                None => Lockstep::observer(vec![0, 1], DEFAULT_DELAY),
            },
            2,
            MatchTerms::default(),
        )
    }

    /// The running beacon says how many chairs are spoken for: the humans
    /// in the lockstep and everyone in line who did not ask to watch.
    #[test]
    fn the_beacon_counts_the_queue_as_taken() {
        let mut host = session(Some(0), 0);
        assert_eq!(host.players_spoken_for(), 2, "the two at the table");
        host.peers.row(0).name = "Bo".into();
        assert_eq!(host.players_spoken_for(), 3, "one in line for a chair");
        host.peers.row(1).watch = true;
        assert_eq!(host.players_spoken_for(), 3, "a watcher wants no chair");
        for peer in 2..8 {
            host.peers.row(peer).name = "Cy".into();
        }
        assert_eq!(
            host.players_spoken_for(),
            MAX_PLAYERS as u8,
            "and never more chairs than the table has"
        );
    }

    /// A joiner reads its own seat in the host's abandonment notice: that
    /// is how it learns it was dropped, rather than waiting out the host
    /// and being told the host had left. Somebody else's seat, or the
    /// host's own view, says nothing of the kind.
    #[test]
    fn a_seat_told_of_its_own_abandonment_knows_it_was_dropped() {
        let host = session(Some(0), 0);
        let port = host.transport.local_addr().expect("addr").port();
        let mut joiner = session(Some(1), port);
        assert!(!joiner.dropped(), "nothing said yet");
        joiner.abandoned.push((0, 30));
        assert!(!joiner.dropped(), "another seat's abandonment is theirs");
        joiner.abandoned.push((1, 30));
        assert!(joiner.dropped(), "its own is the word it was waiting for");
        let mut host = host;
        host.abandoned.push((0, 30));
        assert!(!host.dropped(), "the host drops nobody it is");
        let mut watcher = session(None, port);
        watcher.abandoned.push((1, 30));
        assert!(
            !watcher.dropped(),
            "a spectator holds no seat to be dropped from"
        );
    }
}

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

    /// The desync check has to work across the gap between two peers, and
    /// the gap runs both ways: a peer a commit lead ahead of us sends its
    /// hash for a frame we have not simulated yet.
    ///
    /// So a peer's hash is kept and weighed once our own for that frame
    /// appears. Comparing only what was already in hand skipped exactly the
    /// checks a lagging peer sends.
    #[test]
    fn a_peer_hash_that_arrives_early_is_still_weighed() {
        let mut hashes = HashCheck::default();
        // It arrives first, and there is nothing yet to hold it against.
        hashes.record_peer(60, 0xBAD_B0A7);
        hashes.compare();
        assert_eq!(hashes.desync_at, None, "nothing to compare it to yet");
        // A frame in between, which both sides agree on, is not a desync.
        hashes.record_own(30, 0xC0FFEE);
        hashes.record_peer(30, 0xC0FFEE);
        hashes.compare();
        assert_eq!(hashes.desync_at, None, "and they agree about frame 30");
        // We reach 60 at last, and the hash that has been waiting is what
        // catches it.
        hashes.record_own(60, 0x600D_B0A7);
        hashes.compare();
        assert_eq!(hashes.desync_at, Some(60), "the waiting hash caught it");
    }

    /// A round that agreed all the way through never raises one, however
    /// many hashes crossed: a loud check that cries on a healthy round is
    /// one nobody believes on a sick one.
    #[test]
    fn a_round_that_agrees_is_never_flagged_and_a_new_one_starts_level() {
        let mut hashes = HashCheck::default();
        for frame in (0..300).step_by(HASH_INTERVAL as usize) {
            let hash = u64::from(frame) * 0x9E37_79B9;
            hashes.record_own(frame, hash);
            hashes.record_peer(frame, hash);
            hashes.compare();
        }
        assert_eq!(hashes.desync_at, None, "every frame agreed");

        // Last round's disagreement is last round's. A new board starts
        // level, or one bad round would paint every round after it.
        hashes.record_own(30, 1);
        hashes.record_peer(30, 2);
        hashes.compare();
        assert_eq!(hashes.desync_at, Some(30));
        hashes.reset();
        hashes.compare();
        assert_eq!(hashes.desync_at, None, "and the new board is level");
    }
}