sectorsync_core/

gateway.rs

//! Low-level gateway session and client routing primitives.

use std::cmp::Reverse;
use std::collections::{BTreeMap, BinaryHeap, HashMap};
use std::hash::Hash;

use crate::command::{CommandEnvelope, CommandRejectReason};
use crate::ids::{ClientId, StationId, Tick};

const HASHED_GATEWAY_SESSION_MIN_ENTRIES: usize = 1_024;

#[derive(Clone, Debug)]
enum AdaptiveSessionMap<K, V> {
    Ordered(BTreeMap<K, V>),
    Hashed(HashMap<K, V>),
}

impl<K: Copy + Eq + Hash + Ord, V> AdaptiveSessionMap<K, V> {
    fn new() -> Self {
        Self::Ordered(BTreeMap::new())
    }

    fn len(&self) -> usize {
        match self {
            Self::Ordered(entries) => entries.len(),
            Self::Hashed(entries) => entries.len(),
        }
    }

    fn is_empty(&self) -> bool {
        match self {
            Self::Ordered(entries) => entries.is_empty(),
            Self::Hashed(entries) => entries.is_empty(),
        }
    }

    fn get(&self, key: &K) -> Option<&V> {
        match self {
            Self::Ordered(entries) => entries.get(key),
            Self::Hashed(entries) => entries.get(key),
        }
    }

    fn get_mut(&mut self, key: &K) -> Option<&mut V> {
        match self {
            Self::Ordered(entries) => entries.get_mut(key),
            Self::Hashed(entries) => entries.get_mut(key),
        }
    }

    fn insert(&mut self, key: K, value: V) -> Option<V> {
        let promote = match self {
            Self::Ordered(entries) => {
                entries.len() >= HASHED_GATEWAY_SESSION_MIN_ENTRIES.saturating_sub(1)
                    && !entries.contains_key(&key)
            }
            Self::Hashed(_) => false,
        };
        if promote {
            let Self::Ordered(ordered) = std::mem::replace(self, Self::Hashed(HashMap::new()))
            else {
                unreachable!("promotion starts from ordered session storage");
            };
            let mut hashed = HashMap::with_capacity(ordered.len().saturating_add(1));
            hashed.extend(ordered);
            *self = Self::Hashed(hashed);
        }
        match self {
            Self::Ordered(entries) => entries.insert(key, value),
            Self::Hashed(entries) => entries.insert(key, value),
        }
    }

    fn remove(&mut self, key: &K) -> Option<V> {
        match self {
            Self::Ordered(entries) => entries.remove(key),
            Self::Hashed(entries) => entries.remove(key),
        }
    }

    fn iter(&self) -> impl Iterator<Item = (&K, &V)> {
        match self {
            Self::Ordered(entries) => SessionMapIter::Ordered(entries.iter()),
            Self::Hashed(entries) => SessionMapIter::Hashed(entries.iter()),
        }
    }

    #[cfg(test)]
    fn is_hashed(&self) -> bool {
        matches!(self, Self::Hashed(_))
    }
}

enum SessionMapIter<'a, K, V> {
    Ordered(std::collections::btree_map::Iter<'a, K, V>),
    Hashed(std::collections::hash_map::Iter<'a, K, V>),
}

impl<'a, K, V> Iterator for SessionMapIter<'a, K, V> {
    type Item = (&'a K, &'a V);

    fn next(&mut self) -> Option<Self::Item> {
        match self {
            Self::Ordered(entries) => entries.next(),
            Self::Hashed(entries) => entries.next(),
        }
    }
}

/// Gateway/session table configuration.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GatewayConfig {
    /// Maximum tracked client sessions.
    pub max_sessions: usize,
    /// Number of ticks a disconnected session may reconnect with its current
    /// generation.
    pub reconnect_grace_ticks: u64,
    /// Maximum admitted commands per client per tick.
    pub max_commands_per_tick: usize,
}

impl Default for GatewayConfig {
    fn default() -> Self {
        Self {
            max_sessions: 65_536,
            reconnect_grace_ticks: 20 * 60,
            max_commands_per_tick: 64,
        }
    }
}

/// Gateway route snapshot for a client.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GatewayRoute {
    /// Routed client.
    pub client_id: ClientId,
    /// Current target station for client commands.
    pub station_id: StationId,
    /// Session generation used by reconnect handshakes.
    pub generation: u64,
    /// Route epoch incremented every time a client changes station route.
    pub route_epoch: u64,
}

/// Session connection state.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GatewaySessionState {
    /// Client is connected and may submit commands.
    Connected,
    /// Client disconnected and may reconnect until the grace window expires.
    Disconnected {
        /// Tick at which the disconnect was observed.
        since: Tick,
    },
}

/// Tracked client gateway session.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GatewaySession {
    /// Client id.
    pub client_id: ClientId,
    /// Current target station.
    pub station_id: StationId,
    /// Tick at which this generation was first connected.
    pub connected_at: Tick,
    /// Last tick at which the client was observed.
    pub last_seen: Tick,
    /// Reconnect generation. External gateways may expose this as an opaque
    /// token after adding their own authentication.
    pub generation: u64,
    /// Route epoch incremented on station route changes.
    pub route_epoch: u64,
    /// Connection state.
    pub state: GatewaySessionState,
    last_sequence: Option<u64>,
    command_tick: Tick,
    commands_this_tick: usize,
}

impl GatewaySession {
    /// Returns a route snapshot.
    pub const fn route(&self) -> GatewayRoute {
        GatewayRoute {
            client_id: self.client_id,
            station_id: self.station_id,
            generation: self.generation,
            route_epoch: self.route_epoch,
        }
    }

    /// Returns the latest accepted command sequence.
    pub const fn last_sequence(&self) -> Option<u64> {
        self.last_sequence
    }

    /// Returns command count admitted in the currently tracked tick.
    pub const fn commands_this_tick(&self) -> usize {
        self.commands_this_tick
    }

    /// Returns whether the session is connected.
    pub const fn is_connected(&self) -> bool {
        matches!(self.state, GatewaySessionState::Connected)
    }
}

/// Result of connecting or reconnecting a client.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GatewayConnectOutcome {
    /// A new session entry was created.
    Created,
    /// An already connected session was refreshed.
    AlreadyConnected,
    /// A disconnected session reconnected within its grace window.
    Reconnected {
        /// Ticks between disconnect and reconnect.
        disconnected_for: u64,
    },
    /// A stale disconnected session was replaced with a new generation.
    ReplacedExpired {
        /// Ticks between disconnect and replacement.
        disconnected_for: u64,
    },
}

/// Connection report.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GatewayConnectReport {
    /// Outcome.
    pub outcome: GatewayConnectOutcome,
    /// Current route.
    pub route: GatewayRoute,
}

/// Result of admitting a command through gateway metadata checks.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GatewayCommandAdmission {
    /// Route that should receive the command.
    pub route: GatewayRoute,
    /// Accepted client-side command sequence.
    pub sequence: u64,
    /// Commands accepted for this client in the current tick.
    pub commands_this_tick: usize,
}

/// Gateway session table statistics.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct GatewayStats {
    /// Sessions created.
    pub sessions_created: usize,
    /// Successful reconnects.
    pub sessions_reconnected: usize,
    /// Disconnected sessions expired or replaced after grace.
    pub sessions_expired: usize,
    /// Route changes.
    pub routes_changed: usize,
    /// Commands admitted.
    pub commands_admitted: usize,
    /// Commands rejected as replay/stale.
    pub commands_rejected_replay: usize,
    /// Commands rejected by per-client rate limit.
    pub commands_rejected_rate_limit: usize,
    /// Due deadline entries examined by expiry maintenance.
    pub expiry_deadlines_popped: usize,
    /// Due entries ignored because the session reconnected or disconnected again.
    pub stale_expiry_deadlines: usize,
    /// Stale deadline heap compactions performed after bounded growth.
    pub expiry_deadline_compactions: usize,
}

/// Gateway/session error.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GatewayError {
    /// Session table is full.
    CapacityFull {
        /// Configured maximum session count.
        capacity: usize,
    },
    /// Session does not exist.
    MissingSession(ClientId),
    /// Session is disconnected.
    SessionDisconnected {
        /// Client id.
        client_id: ClientId,
        /// Disconnect tick.
        since: Tick,
    },
    /// Session reconnect generation does not match.
    BadGeneration {
        /// Expected generation.
        expected: u64,
        /// Provided generation.
        actual: u64,
    },
    /// Command sequence was stale or replayed.
    ReplayOrStale {
        /// Latest accepted sequence, if any.
        last_sequence: Option<u64>,
        /// Submitted sequence.
        sequence: u64,
    },
    /// Client exceeded the per-tick command admission limit.
    RateLimited {
        /// Configured limit.
        limit: usize,
        /// Attempted count in the tick.
        attempted: usize,
    },
}

impl GatewayError {
    /// Maps gateway metadata errors into generic command reject reasons.
    pub const fn command_reject_reason(&self) -> Option<CommandRejectReason> {
        match self {
            Self::ReplayOrStale { .. } => Some(CommandRejectReason::ReplayOrStale),
            Self::RateLimited { .. } => Some(CommandRejectReason::RateLimited),
            Self::MissingSession(_)
            | Self::SessionDisconnected { .. }
            | Self::BadGeneration { .. }
            | Self::CapacityFull { .. } => None,
        }
    }
}

impl core::fmt::Display for GatewayError {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            Self::CapacityFull { capacity } => {
                write!(f, "gateway session table is full at capacity {capacity}")
            }
            Self::MissingSession(client_id) => {
                write!(
                    f,
                    "gateway session for client {} is missing",
                    client_id.get()
                )
            }
            Self::SessionDisconnected { client_id, since } => write!(
                f,
                "gateway session for client {} disconnected at tick {}",
                client_id.get(),
                since.get()
            ),
            Self::BadGeneration { expected, actual } => write!(
                f,
                "gateway reconnect generation mismatch: expected {expected}, actual {actual}"
            ),
            Self::ReplayOrStale {
                last_sequence,
                sequence,
            } => write!(
                f,
                "gateway command sequence {sequence} is not newer than {last_sequence:?}"
            ),
            Self::RateLimited { limit, attempted } => write!(
                f,
                "gateway command rate limited: limit {limit}, attempted {attempted}"
            ),
        }
    }
}

impl std::error::Error for GatewayError {}

/// Bounded in-memory gateway session and route table.
#[derive(Clone, Debug)]
pub struct GatewaySessionTable {
    config: GatewayConfig,
    sessions: AdaptiveSessionMap<ClientId, GatewaySession>,
    expiry_deadlines: BinaryHeap<Reverse<(u64, ClientId, u64)>>,
    stats: GatewayStats,
}

impl GatewaySessionTable {
    /// Creates an empty gateway session table.
    pub fn new(config: GatewayConfig) -> Self {
        Self {
            config,
            sessions: AdaptiveSessionMap::new(),
            expiry_deadlines: BinaryHeap::new(),
            stats: GatewayStats::default(),
        }
    }

    /// Returns configuration.
    pub const fn config(&self) -> GatewayConfig {
        self.config
    }

    /// Returns statistics.
    pub const fn stats(&self) -> GatewayStats {
        self.stats
    }

    /// Returns tracked session count.
    pub fn len(&self) -> usize {
        self.sessions.len()
    }

    /// Returns whether no sessions are tracked.
    pub fn is_empty(&self) -> bool {
        self.sessions.is_empty()
    }

    /// Returns a session by client id.
    pub fn session(&self, client_id: ClientId) -> Option<&GatewaySession> {
        self.sessions.get(&client_id)
    }

    /// Returns a connected route by client id.
    pub fn route(&self, client_id: ClientId) -> Result<GatewayRoute, GatewayError> {
        let session = self
            .sessions
            .get(&client_id)
            .ok_or(GatewayError::MissingSession(client_id))?;
        match session.state {
            GatewaySessionState::Connected => Ok(session.route()),
            GatewaySessionState::Disconnected { since } => {
                Err(GatewayError::SessionDisconnected { client_id, since })
            }
        }
    }

    /// Connects a client, reconnecting within grace or replacing stale sessions.
    pub fn connect(
        &mut self,
        client_id: ClientId,
        station_id: StationId,
        now: Tick,
    ) -> Result<GatewayConnectReport, GatewayError> {
        let reconnect_grace_ticks = self.config.reconnect_grace_ticks;
        if let Some(session) = self.sessions.get_mut(&client_id) {
            return Ok(Self::connect_existing(
                session,
                station_id,
                now,
                reconnect_grace_ticks,
                &mut self.stats,
            ));
        }

        if self.sessions.len() >= self.config.max_sessions {
            return Err(GatewayError::CapacityFull {
                capacity: self.config.max_sessions,
            });
        }

        let session = GatewaySession {
            client_id,
            station_id,
            connected_at: now,
            last_seen: now,
            generation: 1,
            route_epoch: 1,
            state: GatewaySessionState::Connected,
            last_sequence: None,
            command_tick: now,
            commands_this_tick: 0,
        };
        let route = session.route();
        self.sessions.insert(client_id, session);
        self.stats.sessions_created = self.stats.sessions_created.saturating_add(1);
        Ok(GatewayConnectReport {
            outcome: GatewayConnectOutcome::Created,
            route,
        })
    }

    /// Reconnects a disconnected client by generation.
    pub fn reconnect(
        &mut self,
        client_id: ClientId,
        generation: u64,
        now: Tick,
    ) -> Result<GatewayConnectReport, GatewayError> {
        let session = self
            .sessions
            .get_mut(&client_id)
            .ok_or(GatewayError::MissingSession(client_id))?;
        if session.generation != generation {
            return Err(GatewayError::BadGeneration {
                expected: session.generation,
                actual: generation,
            });
        }

        match session.state {
            GatewaySessionState::Connected => {
                session.last_seen = now;
                Ok(GatewayConnectReport {
                    outcome: GatewayConnectOutcome::AlreadyConnected,
                    route: session.route(),
                })
            }
            GatewaySessionState::Disconnected { since } => {
                let disconnected_for = now.get().saturating_sub(since.get());
                if disconnected_for > self.config.reconnect_grace_ticks {
                    session.connected_at = now;
                    session.last_seen = now;
                    session.generation = session.generation.saturating_add(1);
                    session.route_epoch = session.route_epoch.saturating_add(1);
                    session.state = GatewaySessionState::Connected;
                    session.last_sequence = None;
                    session.command_tick = now;
                    session.commands_this_tick = 0;
                    self.stats.sessions_expired = self.stats.sessions_expired.saturating_add(1);
                    Ok(GatewayConnectReport {
                        outcome: GatewayConnectOutcome::ReplacedExpired { disconnected_for },
                        route: session.route(),
                    })
                } else {
                    session.last_seen = now;
                    session.state = GatewaySessionState::Connected;
                    self.stats.sessions_reconnected =
                        self.stats.sessions_reconnected.saturating_add(1);
                    Ok(GatewayConnectReport {
                        outcome: GatewayConnectOutcome::Reconnected { disconnected_for },
                        route: session.route(),
                    })
                }
            }
        }
    }

    /// Marks a connected session disconnected.
    pub fn disconnect(&mut self, client_id: ClientId, now: Tick) -> Result<(), GatewayError> {
        let session = self
            .sessions
            .get_mut(&client_id)
            .ok_or(GatewayError::MissingSession(client_id))?;
        session.last_seen = now;
        session.state = GatewaySessionState::Disconnected { since: now };
        self.expiry_deadlines.push(Reverse((
            expiry_deadline(now, self.config.reconnect_grace_ticks),
            client_id,
            now.get(),
        )));
        self.compact_expiry_deadlines_if_needed();
        Ok(())
    }

    /// Removes disconnected sessions whose grace window has expired.
    pub fn expire_disconnected(&mut self, now: Tick) -> usize {
        let mut expired = 0_usize;
        while self
            .expiry_deadlines
            .peek()
            .is_some_and(|Reverse((deadline, _, _))| *deadline <= now.get())
        {
            let Reverse((_, client_id, expected_since)) = self
                .expiry_deadlines
                .pop()
                .expect("peeked deadline remains available");
            self.stats.expiry_deadlines_popped =
                self.stats.expiry_deadlines_popped.saturating_add(1);
            let current_matches = self.sessions.get(&client_id).is_some_and(|session| {
                matches!(
                    session.state,
                    GatewaySessionState::Disconnected { since }
                        if since.get() == expected_since
                            && now.get().saturating_sub(since.get())
                                > self.config.reconnect_grace_ticks
                )
            });
            if current_matches {
                self.sessions.remove(&client_id);
                expired = expired.saturating_add(1);
            } else {
                self.stats.stale_expiry_deadlines =
                    self.stats.stale_expiry_deadlines.saturating_add(1);
            }
        }
        self.stats.sessions_expired = self.stats.sessions_expired.saturating_add(expired);
        expired
    }

    /// Deadline entries retained for caller-driven expiry maintenance.
    pub fn expiry_deadline_len(&self) -> usize {
        self.expiry_deadlines.len()
    }

    /// Allocated deadline-entry capacity retained by the table.
    pub fn expiry_deadline_capacity(&self) -> usize {
        self.expiry_deadlines.capacity()
    }

    /// Changes the target station for a connected client.
    pub fn reroute(
        &mut self,
        client_id: ClientId,
        station_id: StationId,
        now: Tick,
    ) -> Result<GatewayRoute, GatewayError> {
        let session = self
            .sessions
            .get_mut(&client_id)
            .ok_or(GatewayError::MissingSession(client_id))?;
        match session.state {
            GatewaySessionState::Connected => {
                session.last_seen = now;
                if session.station_id != station_id {
                    session.station_id = station_id;
                    session.route_epoch = session.route_epoch.saturating_add(1);
                    self.stats.routes_changed = self.stats.routes_changed.saturating_add(1);
                }
                Ok(session.route())
            }
            GatewaySessionState::Disconnected { since } => {
                Err(GatewayError::SessionDisconnected { client_id, since })
            }
        }
    }

    /// Applies session metadata checks to a command and returns the route that
    /// should receive it.
    pub fn admit_command(
        &mut self,
        command: &CommandEnvelope,
    ) -> Result<GatewayCommandAdmission, GatewayError> {
        self.admit_sequence(command.client_id, command.sequence, command.received_at)
    }

    /// Applies session metadata checks to a client command sequence.
    pub fn admit_sequence(
        &mut self,
        client_id: ClientId,
        sequence: u64,
        now: Tick,
    ) -> Result<GatewayCommandAdmission, GatewayError> {
        let session = self
            .sessions
            .get_mut(&client_id)
            .ok_or(GatewayError::MissingSession(client_id))?;
        match session.state {
            GatewaySessionState::Connected => {}
            GatewaySessionState::Disconnected { since } => {
                return Err(GatewayError::SessionDisconnected { client_id, since });
            }
        }

        if session
            .last_sequence
            .is_some_and(|last_sequence| sequence <= last_sequence)
        {
            self.stats.commands_rejected_replay =
                self.stats.commands_rejected_replay.saturating_add(1);
            return Err(GatewayError::ReplayOrStale {
                last_sequence: session.last_sequence,
                sequence,
            });
        }

        if session.command_tick != now {
            session.command_tick = now;
            session.commands_this_tick = 0;
        }
        let attempted = session.commands_this_tick.saturating_add(1);
        if attempted > self.config.max_commands_per_tick {
            self.stats.commands_rejected_rate_limit =
                self.stats.commands_rejected_rate_limit.saturating_add(1);
            return Err(GatewayError::RateLimited {
                limit: self.config.max_commands_per_tick,
                attempted,
            });
        }

        session.commands_this_tick = attempted;
        session.last_sequence = Some(sequence);
        session.last_seen = now;
        self.stats.commands_admitted = self.stats.commands_admitted.saturating_add(1);
        Ok(GatewayCommandAdmission {
            route: session.route(),
            sequence,
            commands_this_tick: attempted,
        })
    }

    fn connect_existing(
        session: &mut GatewaySession,
        station_id: StationId,
        now: Tick,
        reconnect_grace_ticks: u64,
        stats: &mut GatewayStats,
    ) -> GatewayConnectReport {
        match session.state {
            GatewaySessionState::Connected => {
                session.last_seen = now;
                if session.station_id != station_id {
                    session.station_id = station_id;
                    session.route_epoch = session.route_epoch.saturating_add(1);
                    stats.routes_changed = stats.routes_changed.saturating_add(1);
                }
                GatewayConnectReport {
                    outcome: GatewayConnectOutcome::AlreadyConnected,
                    route: session.route(),
                }
            }
            GatewaySessionState::Disconnected { since } => {
                let disconnected_for = now.get().saturating_sub(since.get());
                if disconnected_for > reconnect_grace_ticks {
                    session.station_id = station_id;
                    session.connected_at = now;
                    session.last_seen = now;
                    session.generation = session.generation.saturating_add(1);
                    session.route_epoch = session.route_epoch.saturating_add(1);
                    session.state = GatewaySessionState::Connected;
                    session.last_sequence = None;
                    session.command_tick = now;
                    session.commands_this_tick = 0;
                    stats.sessions_expired = stats.sessions_expired.saturating_add(1);
                    GatewayConnectReport {
                        outcome: GatewayConnectOutcome::ReplacedExpired { disconnected_for },
                        route: session.route(),
                    }
                } else {
                    session.station_id = station_id;
                    session.last_seen = now;
                    session.state = GatewaySessionState::Connected;
                    stats.sessions_reconnected = stats.sessions_reconnected.saturating_add(1);
                    GatewayConnectReport {
                        outcome: GatewayConnectOutcome::Reconnected { disconnected_for },
                        route: session.route(),
                    }
                }
            }
        }
    }

    fn compact_expiry_deadlines_if_needed(&mut self) {
        let limit = self.config.max_sessions.max(1).saturating_mul(2);
        if self.expiry_deadlines.len() <= limit {
            return;
        }
        let mut deadlines = BinaryHeap::with_capacity(self.sessions.len());
        for (client_id, session) in self.sessions.iter() {
            if let GatewaySessionState::Disconnected { since } = session.state {
                deadlines.push(Reverse((
                    expiry_deadline(since, self.config.reconnect_grace_ticks),
                    *client_id,
                    since.get(),
                )));
            }
        }
        self.expiry_deadlines = deadlines;
        self.stats.expiry_deadline_compactions =
            self.stats.expiry_deadline_compactions.saturating_add(1);
    }
}

const fn expiry_deadline(since: Tick, grace: u64) -> u64 {
    since.get().saturating_add(grace).saturating_add(1)
}

impl Default for GatewaySessionTable {
    fn default() -> Self {
        Self::new(GatewayConfig::default())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::command::{CommandEnvelope, CommandPriority};
    use crate::ids::{CommandId, EntityId};

    fn command(client_id: ClientId, sequence: u64, tick: u64) -> CommandEnvelope {
        CommandEnvelope {
            id: CommandId::new(sequence),
            client_id,
            entity_id: EntityId::new(10),
            sequence,
            received_at: Tick::new(tick),
            kind: 1,
            priority: CommandPriority::Normal,
            payload: Vec::new(),
        }
    }

    #[test]
    fn connects_routes_and_reroutes_sessions() {
        let client_id = ClientId::new(7);
        let mut table = GatewaySessionTable::new(GatewayConfig {
            max_sessions: 4,
            reconnect_grace_ticks: 3,
            max_commands_per_tick: 4,
        });

        let connected = table
            .connect(client_id, StationId::new(1), Tick::new(10))
            .expect("connect should work");
        assert_eq!(connected.outcome, GatewayConnectOutcome::Created);
        assert_eq!(connected.route.station_id, StationId::new(1));
        assert_eq!(connected.route.generation, 1);
        assert_eq!(connected.route.route_epoch, 1);

        let route = table
            .reroute(client_id, StationId::new(2), Tick::new(11))
            .expect("reroute should work");
        assert_eq!(route.station_id, StationId::new(2));
        assert_eq!(route.route_epoch, 2);
        assert_eq!(table.stats().routes_changed, 1);
        assert_eq!(
            table
                .route(client_id)
                .expect("route should exist")
                .station_id,
            StationId::new(2)
        );
    }

    #[test]
    fn gateway_sessions_promote_without_losing_route_admission_or_expiry_state() {
        let mut table = GatewaySessionTable::new(GatewayConfig {
            max_sessions: HASHED_GATEWAY_SESSION_MIN_ENTRIES + 1,
            reconnect_grace_ticks: 2,
            max_commands_per_tick: 4,
        });
        for index in 0..HASHED_GATEWAY_SESSION_MIN_ENTRIES - 1 {
            table
                .connect(
                    ClientId::new(u64::try_from(index).expect("test client id fits u64")),
                    StationId::new(u32::try_from(index % 4).expect("test station id fits u32")),
                    Tick::new(0),
                )
                .expect("session should connect");
        }
        assert!(!table.sessions.is_hashed());
        table
            .connect(ClientId::new(0), StationId::new(3), Tick::new(1))
            .expect("existing session should refresh without promotion");
        assert!(!table.sessions.is_hashed());
        table
            .admit_sequence(ClientId::new(0), 1, Tick::new(1))
            .expect("command should admit before promotion");
        table
            .disconnect(ClientId::new(1), Tick::new(1))
            .expect("session should disconnect before promotion");

        let final_client = ClientId::new(
            u64::try_from(HASHED_GATEWAY_SESSION_MIN_ENTRIES - 1).expect("test client id fits u64"),
        );
        table
            .connect(final_client, StationId::new(2), Tick::new(2))
            .expect("threshold session should connect");
        assert!(table.sessions.is_hashed());
        assert_eq!(
            table
                .route(ClientId::new(0))
                .expect("route should survive")
                .station_id,
            StationId::new(3)
        );
        assert_eq!(
            table
                .admit_sequence(ClientId::new(0), 2, Tick::new(2))
                .expect("admission should survive")
                .sequence,
            2
        );
        assert_eq!(table.expire_disconnected(Tick::new(4)), 1);
        assert!(table.sessions.is_hashed());
        assert_eq!(table.len(), HASHED_GATEWAY_SESSION_MIN_ENTRIES - 1);
        assert_eq!(
            table.stats.sessions_created,
            HASHED_GATEWAY_SESSION_MIN_ENTRIES
        );
        assert_eq!(table.stats.sessions_expired, 1);
        assert_eq!(table.stats.commands_admitted, 2);
    }

    #[test]
    fn reconnects_with_generation_and_expires_disconnected_sessions() {
        let client_id = ClientId::new(7);
        let mut table = GatewaySessionTable::new(GatewayConfig {
            max_sessions: 4,
            reconnect_grace_ticks: 3,
            max_commands_per_tick: 4,
        });
        let connected = table
            .connect(client_id, StationId::new(1), Tick::new(10))
            .expect("connect should work");
        table
            .disconnect(client_id, Tick::new(12))
            .expect("disconnect should work");
        assert!(matches!(
            table.route(client_id),
            Err(GatewayError::SessionDisconnected { .. })
        ));

        let bad = table
            .reconnect(client_id, connected.route.generation + 1, Tick::new(13))
            .expect_err("bad generation should fail");
        assert_eq!(
            bad,
            GatewayError::BadGeneration {
                expected: 1,
                actual: 2
            }
        );

        let reconnected = table
            .reconnect(client_id, connected.route.generation, Tick::new(14))
            .expect("reconnect should work");
        assert_eq!(
            reconnected.outcome,
            GatewayConnectOutcome::Reconnected {
                disconnected_for: 2
            }
        );
        assert_eq!(reconnected.route.generation, 1);

        table
            .disconnect(client_id, Tick::new(15))
            .expect("disconnect should work");
        assert_eq!(table.expire_disconnected(Tick::new(19)), 1);
        assert_eq!(table.len(), 0);
        assert_eq!(table.stats().sessions_expired, 1);
    }

    #[test]
    fn stale_expiry_deadline_cannot_remove_reconnected_session() {
        let client_id = ClientId::new(7);
        let mut table = GatewaySessionTable::new(GatewayConfig {
            max_sessions: 1,
            reconnect_grace_ticks: 3,
            max_commands_per_tick: 4,
        });
        let connected = table
            .connect(client_id, StationId::new(1), Tick::new(0))
            .expect("connect");
        table
            .disconnect(client_id, Tick::new(1))
            .expect("disconnect");
        table
            .reconnect(client_id, connected.route.generation, Tick::new(2))
            .expect("reconnect");

        assert_eq!(table.expire_disconnected(Tick::new(10)), 0);
        assert!(
            table
                .session(client_id)
                .is_some_and(GatewaySession::is_connected)
        );
        assert_eq!(table.stats().expiry_deadlines_popped, 1);
        assert_eq!(table.stats().stale_expiry_deadlines, 1);
    }

    #[test]
    fn repeated_disconnect_deadlines_compact_at_bounded_growth() {
        let client_id = ClientId::new(7);
        let mut table = GatewaySessionTable::new(GatewayConfig {
            max_sessions: 1,
            reconnect_grace_ticks: 10,
            max_commands_per_tick: 4,
        });
        let generation = table
            .connect(client_id, StationId::new(1), Tick::new(0))
            .expect("connect")
            .route
            .generation;
        for tick in 1..=3 {
            table
                .disconnect(client_id, Tick::new(tick * 2))
                .expect("disconnect");
            table
                .reconnect(client_id, generation, Tick::new(tick * 2 + 1))
                .expect("reconnect");
        }

        assert_eq!(table.stats().expiry_deadline_compactions, 1);
        assert!(table.expiry_deadline_len() <= table.config().max_sessions);
        assert!(
            table
                .session(client_id)
                .is_some_and(GatewaySession::is_connected)
        );
    }

    #[test]
    fn expiry_retains_connected_and_grace_boundary_sessions() {
        let mut table = GatewaySessionTable::new(GatewayConfig {
            max_sessions: 4,
            reconnect_grace_ticks: 3,
            max_commands_per_tick: 4,
        });
        for client in 1..=3 {
            table
                .connect(ClientId::new(client), StationId::new(1), Tick::new(10))
                .expect("session should connect");
        }
        table
            .disconnect(ClientId::new(2), Tick::new(12))
            .expect("boundary session disconnects");
        table
            .disconnect(ClientId::new(3), Tick::new(11))
            .expect("expired session disconnects");

        assert_eq!(table.expire_disconnected(Tick::new(15)), 1);
        assert!(table.session(ClientId::new(1)).is_some());
        assert!(table.session(ClientId::new(2)).is_some());
        assert!(table.session(ClientId::new(3)).is_none());
        assert_eq!(table.stats().sessions_expired, 1);
        assert_eq!(table.expire_disconnected(Tick::new(16)), 1);
        assert_eq!(table.len(), 1);
        assert_eq!(table.stats().sessions_expired, 2);
    }

    #[test]
    fn connect_replaces_stale_disconnected_session() {
        let client_id = ClientId::new(7);
        let mut table = GatewaySessionTable::new(GatewayConfig {
            max_sessions: 4,
            reconnect_grace_ticks: 3,
            max_commands_per_tick: 4,
        });
        let connected = table
            .connect(client_id, StationId::new(1), Tick::new(10))
            .expect("connect should work");
        table
            .admit_sequence(client_id, 10, Tick::new(11))
            .expect("first command should admit");
        table
            .disconnect(client_id, Tick::new(12))
            .expect("disconnect should work");

        let replaced = table
            .connect(client_id, StationId::new(2), Tick::new(20))
            .expect("stale reconnect should replace generation");
        assert_eq!(
            replaced.outcome,
            GatewayConnectOutcome::ReplacedExpired {
                disconnected_for: 8
            }
        );
        assert_eq!(replaced.route.generation, connected.route.generation + 1);
        assert_eq!(replaced.route.station_id, StationId::new(2));
        assert_eq!(
            table
                .admit_sequence(client_id, 1, Tick::new(21))
                .expect("new generation should reset sequence")
                .sequence,
            1
        );
    }

    #[test]
    fn command_admission_rejects_replay_and_rate_limit() {
        let client_id = ClientId::new(7);
        let mut table = GatewaySessionTable::new(GatewayConfig {
            max_sessions: 4,
            reconnect_grace_ticks: 3,
            max_commands_per_tick: 2,
        });
        table
            .connect(client_id, StationId::new(1), Tick::new(10))
            .expect("connect should work");

        let first = table
            .admit_command(&command(client_id, 1, 10))
            .expect("first command should admit");
        assert_eq!(first.commands_this_tick, 1);
        assert_eq!(first.route.station_id, StationId::new(1));

        let replay = table
            .admit_command(&command(client_id, 1, 10))
            .expect_err("same sequence should reject");
        assert_eq!(
            replay,
            GatewayError::ReplayOrStale {
                last_sequence: Some(1),
                sequence: 1
            }
        );
        assert_eq!(
            replay.command_reject_reason(),
            Some(CommandRejectReason::ReplayOrStale)
        );

        table
            .admit_command(&command(client_id, 2, 10))
            .expect("second command should admit");
        let limited = table
            .admit_command(&command(client_id, 3, 10))
            .expect_err("third same-tick command should rate limit");
        assert_eq!(
            limited,
            GatewayError::RateLimited {
                limit: 2,
                attempted: 3
            }
        );
        assert_eq!(
            limited.command_reject_reason(),
            Some(CommandRejectReason::RateLimited)
        );

        let next_tick = table
            .admit_command(&command(client_id, 3, 11))
            .expect("next tick should reset rate count");
        assert_eq!(next_tick.commands_this_tick, 1);
        assert_eq!(table.stats().commands_admitted, 3);
        assert_eq!(table.stats().commands_rejected_replay, 1);
        assert_eq!(table.stats().commands_rejected_rate_limit, 1);
    }

    #[test]
    fn capacity_limit_is_enforced() {
        let mut table = GatewaySessionTable::new(GatewayConfig {
            max_sessions: 1,
            reconnect_grace_ticks: 3,
            max_commands_per_tick: 4,
        });
        table
            .connect(ClientId::new(1), StationId::new(1), Tick::new(0))
            .expect("first session should fit");
        let error = table
            .connect(ClientId::new(2), StationId::new(1), Tick::new(0))
            .expect_err("second session should exceed capacity");
        assert_eq!(error, GatewayError::CapacityFull { capacity: 1 });
    }
}