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(),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GatewayConfig {
pub max_sessions: usize,
pub reconnect_grace_ticks: u64,
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,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GatewayRoute {
pub client_id: ClientId,
pub station_id: StationId,
pub generation: u64,
pub route_epoch: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GatewaySessionState {
Connected,
Disconnected {
since: Tick,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GatewaySession {
pub client_id: ClientId,
pub station_id: StationId,
pub connected_at: Tick,
pub last_seen: Tick,
pub generation: u64,
pub route_epoch: u64,
pub state: GatewaySessionState,
last_sequence: Option<u64>,
command_tick: Tick,
commands_this_tick: usize,
}
impl GatewaySession {
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,
}
}
pub const fn last_sequence(&self) -> Option<u64> {
self.last_sequence
}
pub const fn commands_this_tick(&self) -> usize {
self.commands_this_tick
}
pub const fn is_connected(&self) -> bool {
matches!(self.state, GatewaySessionState::Connected)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GatewayConnectOutcome {
Created,
AlreadyConnected,
Reconnected {
disconnected_for: u64,
},
ReplacedExpired {
disconnected_for: u64,
},
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GatewayConnectReport {
pub outcome: GatewayConnectOutcome,
pub route: GatewayRoute,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GatewayCommandAdmission {
pub route: GatewayRoute,
pub sequence: u64,
pub commands_this_tick: usize,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct GatewayStats {
pub sessions_created: usize,
pub sessions_reconnected: usize,
pub sessions_expired: usize,
pub routes_changed: usize,
pub commands_admitted: usize,
pub commands_rejected_replay: usize,
pub commands_rejected_rate_limit: usize,
pub expiry_deadlines_popped: usize,
pub stale_expiry_deadlines: usize,
pub expiry_deadline_compactions: usize,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum GatewayError {
CapacityFull {
capacity: usize,
},
MissingSession(ClientId),
SessionDisconnected {
client_id: ClientId,
since: Tick,
},
BadGeneration {
expected: u64,
actual: u64,
},
ReplayOrStale {
last_sequence: Option<u64>,
sequence: u64,
},
RateLimited {
limit: usize,
attempted: usize,
},
}
impl GatewayError {
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 {}
#[derive(Clone, Debug)]
pub struct GatewaySessionTable {
config: GatewayConfig,
sessions: AdaptiveSessionMap<ClientId, GatewaySession>,
expiry_deadlines: BinaryHeap<Reverse<(u64, ClientId, u64)>>,
stats: GatewayStats,
}
impl GatewaySessionTable {
pub fn new(config: GatewayConfig) -> Self {
Self {
config,
sessions: AdaptiveSessionMap::new(),
expiry_deadlines: BinaryHeap::new(),
stats: GatewayStats::default(),
}
}
pub const fn config(&self) -> GatewayConfig {
self.config
}
pub const fn stats(&self) -> GatewayStats {
self.stats
}
pub fn len(&self) -> usize {
self.sessions.len()
}
pub fn is_empty(&self) -> bool {
self.sessions.is_empty()
}
pub fn session(&self, client_id: ClientId) -> Option<&GatewaySession> {
self.sessions.get(&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 })
}
}
}
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,
})
}
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(),
})
}
}
}
}
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(())
}
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
}
pub fn expiry_deadline_len(&self) -> usize {
self.expiry_deadlines.len()
}
pub fn expiry_deadline_capacity(&self) -> usize {
self.expiry_deadlines.capacity()
}
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 })
}
}
}
pub fn admit_command(
&mut self,
command: &CommandEnvelope,
) -> Result<GatewayCommandAdmission, GatewayError> {
self.admit_sequence(command.client_id, command.sequence, command.received_at)
}
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 });
}
}