GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

cultnet-rs: end the session of a peer whose packet the session refuses

Browse source 
CultNetRudpSession::receive records a reliable sequence before reassembly, so
keeping the session after it refuses a frame made the sender's retransmit be
acknowledged and the frame silently lost. The hub and the socket transport now
end that peer's session, send it a Disconnect, and surface the end to the
caller (a Disconnected event, disconnect_reason()). The hub and socket
constructors reject initial_sequence u32::MAX and max_pending_reliable_packets
Some(0), so the accept_connect drops are unreachable. pending_accept_for_resend
becomes public for the document server.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-29 19:33:30 +02:00
commit f6dbbf77ef
2 changed files with 189 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

67
packages/cultnet-rs/src/rudp.rs
View file

@ -271,7 +271,9 @@ impl CultNetRudpSession {
self.pending_reliable.len() + self.queued_reliable.len()
}
fn pending_accept_for_resend(&mut self, now_ms: u64) -> Option<CultNetRudpPacket> {
/// The unacknowledged Accept, stamped as just sent, for a peer that
/// repeats its Connect. `None` once the peer has acknowledged it.
pub fn pending_accept_for_resend(&mut self, now_ms: u64) -> Option<CultNetRudpPacket> {
let pending = self
.pending_reliable
.values_mut()
@ -1399,6 +1401,10 @@ pub struct CultNetRudpServerHub {
impl CultNetRudpServerHub {
pub fn new(options: CultNetRudpServerHubOptions) -> Result<Self> {
validate_rudp_session_limits(
options.initial_sequence,
options.max_pending_reliable_packets,
)?;
let local_addr = options.socket.local_addr()?;
if options.max_peers == 0 {
return Err(anyhow!(
@ -1589,7 +1595,7 @@ impl CultNetRudpServerHub {
&& peer.context.connect_payload == packet.payload
{
if peer.session.receive(&packet, now_ms()).is_err() {
self.stats.packets_dropped += 1;
self.end_refusing_peer(remote_addr)?;
return Ok(true);
}
let reply = peer
@ -1650,7 +1656,7 @@ impl CultNetRudpServerHub {
return Ok(true);
};
let Ok(result) = peer.session.receive(&packet, now_ms()) else {
self.stats.packets_dropped += 1;
self.end_refusing_peer(remote_addr)?;
return Ok(true);
};
let context = peer.context.clone();
@ -1696,6 +1702,25 @@ impl CultNetRudpServerHub {
Ok(true)
}
/// An admitted peer sent a packet its session refuses. `receive` has
/// already recorded the packet's reliable sequence, so the session cannot be
/// kept: a retransmit would be acknowledged and the frame silently lost.
/// The session ends, the peer is told, and the caller sees `Disconnected`.
fn end_refusing_peer(&mut self, remote_addr: SocketAddr) -> Result<()> {
self.stats.packets_dropped += 1;
let Some(mut peer) = self.peers.remove(&remote_addr) else {
return Ok(());
};
let reason = RUDP_REFUSED_PACKET_REASON.to_vec();
let goodbye = peer.session.create_disconnect(reason.clone());
self.pending_events
.push_back(CultNetRudpServerEvent::Disconnected {
session: peer.context,
reason,
});
self.send_packet(remote_addr, &goodbye)
}
pub fn poll_resends(&mut self) -> Result<()> {
let now = now_ms();
let mut packets = Vec::new();
@ -1743,6 +1768,10 @@ impl CultNetRudpServerHub {
impl CultNetRudpSocketTransportConnection {
pub fn new(options: CultNetRudpSocketTransportOptions) -> Result<Self> {
validate_rudp_session_limits(
options.initial_sequence,
options.max_pending_reliable_packets,
)?;
let local_addr = options.socket.local_addr()?;
let max_payload_bytes = options
.max_payload_bytes
@ -2035,8 +2064,15 @@ impl CultNetRudpSocketTransportConnection {
}
let Ok(result) = self.session.receive(&packet, now_ms()) else {
// `receive` has already recorded the packet's reliable sequence, so
// the session cannot be kept: a retransmit would be acknowledged
// and the frame silently lost. End it and tell the peer.
self.stats.packets_dropped += 1;
return Ok(true);
let reason = RUDP_REFUSED_PACKET_REASON.to_vec();
let goodbye = self.session.create_disconnect(reason.clone());
self.disconnect_reason = Some(reason);
self.send_packet(&goodbye)?;
return Ok(false);
};
if let Some(reply) = result.reply {
self.send_packet(&reply)?;
@ -2181,6 +2217,29 @@ where
}
}
/// Sent to a peer whose session refused one of its packets, and reported to
/// the local caller as the disconnect reason.
const RUDP_REFUSED_PACKET_REASON: &[u8] = b"session refused a packet";
/// A session cannot admit a Connect when its sequence space starts exhausted or
/// its reliable queue holds nothing, so neither is a usable configuration.
fn validate_rudp_session_limits(
initial_sequence: u32,
max_pending_reliable_packets: Option<u32>,
) -> Result<()> {
if initial_sequence == u32::MAX {
return Err(anyhow!(
"RUDP initial_sequence must leave room for a reliable packet"
));
}
if max_pending_reliable_packets == Some(0) {
return Err(anyhow!(
"RUDP max_pending_reliable_packets must be greater than zero"
));
}
Ok(())
}
#[derive(Clone, Debug, Default)]
pub struct RudpTransportProfileOptions {
pub transport_id: Option<String>,

182
packages/cultnet-rs/tests/rudp_stray_packets.rs
View file

@ -273,46 +273,101 @@ per_stray!(
server_mode_survives_foreign_connection_ids
);
#[test]
fn hub_drops_a_packet_its_admitted_peer_session_refuses() -> Result<()> {
let mut hub = CultNetRudpServerHub::new(CultNetRudpServerHubOptions::new(
"hub",
socket()?,
CONNECTION_ID,
))?;
let hub_addr = hub.local_addr()?;
let peer = socket()?;
let mut peer_session = raw_session(CONNECTION_ID);
send_to(&peer, hub_addr, &peer_session.create_connect(0, b"peer".to_vec())?)?;
assert!(matches!(
hub.receive_event_once()?,
Some(CultNetRudpServerEvent::Connected { .. })
));
let mut buffer = vec![0_u8; 65_535];
let (received, _) = peer.recv_from(&mut buffer)?;
peer_session.receive(&decode_rudp_packet(&buffer[..received])?, 0)?;
/// A session that refuses a packet has already recorded its reliable sequence,
/// so keeping it would acknowledge the sender's retransmit and lose the frame.
/// The session ends and the sender is told.
fn refused_frame_options() -> (u32, Vec<u8>) {
(64, vec![7_u8; 100])
}
send_to(&peer, hub_addr, &poisoned(&mut peer_session)?)?;
assert!(hub.receive_event_once()?.is_none());
assert_eq!(hub.stats().packets_dropped, 1);
assert_eq!(hub.sessions().len(), 1);
fn client_with_fast_resend(server_addr: SocketAddr) -> Result<CultNetRudpSocketTransportConnection> {
let mut options =
CultNetRudpSocketTransportOptions::client("client", socket()?, server_addr, CONNECTION_ID);
options.resend_delay_ms = 1;
CultNetRudpSocketTransportConnection::new(options)
}
send_to(
&peer,
hub_addr,
&peer_session.send("schema", b"still here".to_vec(), reliable())?,
)?;
match hub.receive_event_once()? {
Some(CultNetRudpServerEvent::Frame { frame, .. }) => {
assert_eq!(frame.payload, b"still here");
}
other => panic!("the peer must still be served, got {other:?}"),
}
/// The client's view after the server refused its frame: told the session is
/// over, and never shown the frame as acknowledged, even after it retransmits.
fn assert_client_sees_refusal(
client: &mut CultNetRudpSocketTransportConnection,
receipt: &CultNetRudpReliableSendReceipt,
mut serve: impl FnMut() -> Result<()>,
) -> Result<()> {
serve()?;
let _ = client.receive_once()?;
assert!(
client.disconnect_reason().is_some(),
"the client must be told the session ended"
);
std::thread::sleep(Duration::from_millis(10));
client.poll_resends()?;
serve()?;
let _ = client.receive_once()?;
assert_ne!(
client.reliable_send_status(receipt),
CultNetRudpReliableSendStatus::Acknowledged,
"a refused frame must never be acknowledged"
);
Ok(())
}
#[test]
fn client_drops_a_packet_its_session_refuses() -> Result<()> {
fn hub_ends_the_session_of_a_peer_whose_reliable_frame_it_refuses() -> Result<()> {
let (limit, payload) = refused_frame_options();
let mut options = CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID);
options.max_payload_bytes = Some(limit);
let mut hub = CultNetRudpServerHub::new(options)?;
let hub_addr = hub.local_addr()?;
let mut client = client_with_fast_resend(hub_addr)?;
client.connect(b"peer".to_vec())?;
assert!(matches!(
hub.receive_event_once()?,
Some(CultNetRudpServerEvent::Connected { .. })
));
let _ = client.receive_once()?;
assert!(client.connected());
let receipt = client.send_reliable("schema", payload)?;
let mut ended = false;
assert_client_sees_refusal(&mut client, &receipt, || {
while let Some(event) = hub.receive_event_once()? {
ended |= matches!(event, CultNetRudpServerEvent::Disconnected { .. });
}
Ok(())
})?;
assert!(ended, "the hub must report the ended session");
assert_eq!(hub.sessions().len(), 0);
assert!(hub.stats().packets_dropped >= 1);
Ok(())
}
#[test]
fn server_mode_ends_the_session_of_a_peer_whose_reliable_frame_it_refuses() -> Result<()> {
let (limit, payload) = refused_frame_options();
let server_socket = socket()?;
let server_addr = server_socket.local_addr()?;
let mut options = CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID);
options.max_payload_bytes = Some(limit);
let mut server = CultNetRudpSocketTransportConnection::new(options)?;
let mut client = client_with_fast_resend(server_addr)?;
client.connect(b"peer".to_vec())?;
let _ = server.receive_once()?;
let _ = client.receive_once()?;
assert!(client.connected());
let receipt = client.send_reliable("schema", payload)?;
assert_client_sees_refusal(&mut client, &receipt, || {
let _ = server.receive_once()?;
Ok(())
})?;
assert!(server.disconnect_reason().is_some());
assert!(server.stats().packets_dropped >= 1);
Ok(())
}
#[test]
fn client_ends_its_session_when_it_refuses_a_packet() -> Result<()> {
let server = socket()?;
let server_addr = server.local_addr()?;
let mut client = CultNetRudpSocketTransportConnection::new(
@ -330,17 +385,17 @@ fn client_drops_a_packet_its_session_refuses() -> Result<()> {
send_to(&server, client_addr, &poisoned(&mut server_session)?)?;
assert!(client.receive_once()?.is_none());
assert_eq!(client.stats().packets_dropped, 1);
send_to(
&server,
client_addr,
&server_session.send("schema", b"real".to_vec(), reliable())?,
)?;
assert_eq!(client.receive_once()?.expect("the real frame").payload, b"real");
assert!(client.disconnect_reason().is_some());
// The server is told, not left holding a live session.
let (received, _) = server.recv_from(&mut buffer)?;
let told = decode_rudp_packet(&buffer[..received])?;
assert_eq!(told.packet_type, CultNetRudpPacketType::Disconnect);
Ok(())
}
#[test]
fn server_mode_drops_what_is_not_its_peer_or_what_its_session_refuses() -> Result<()> {
fn server_mode_ends_the_session_when_the_session_refuses_a_packet() -> Result<()> {
let server_socket = socket()?;
let server_addr = server_socket.local_addr()?;
let mut server = CultNetRudpSocketTransportConnection::new(
@ -367,16 +422,45 @@ fn server_mode_drops_what_is_not_its_peer_or_what_its_session_refuses() -> Resul
assert!(server.receive_once()?.is_none());
assert_eq!(server.stats().packets_dropped, 6);
// A packet from the peer that the session itself refuses.
// A packet from the peer that the session itself refuses ends the session
// and the peer is told.
send_to(&peer, server_addr, &poisoned(&mut peer_session)?)?;
assert!(server.receive_once()?.is_none());
assert_eq!(server.stats().packets_dropped, 7);
send_to(
&peer,
server_addr,
&peer_session.send("schema", b"real".to_vec(), reliable())?,
)?;
assert_eq!(server.receive_once()?.expect("the real frame").payload, b"real");
assert!(server.disconnect_reason().is_some());
let (received, _) = peer.recv_from(&mut buffer)?;
assert_eq!(
decode_rudp_packet(&buffer[..received])?.packet_type,
CultNetRudpPacketType::Disconnect
);
Ok(())
}
#[test]
fn constructors_reject_limits_that_make_a_connect_unadmittable() -> Result<()> {
let mut hub = CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID);
hub.initial_sequence = u32::MAX;
assert!(CultNetRudpServerHub::new(hub).is_err());
let mut hub = CultNetRudpServerHubOptions::new("hub", socket()?, CONNECTION_ID);
hub.max_pending_reliable_packets = Some(0);
assert!(CultNetRudpServerHub::new(hub).is_err());
for build in [
(|s, a| CultNetRudpSocketTransportOptions::client("c", s, a, CONNECTION_ID))
as fn(UdpSocket, SocketAddr) -> CultNetRudpSocketTransportOptions,
|s, _| CultNetRudpSocketTransportOptions::server("s", s, CONNECTION_ID),
] {
let addr = "127.0.0.1:9".parse()?;
let mut options = build(socket()?, addr);
options.initial_sequence = u32::MAX;
assert!(CultNetRudpSocketTransportConnection::new(options).is_err());
let mut options = build(socket()?, addr);
options.max_pending_reliable_packets = Some(0);
assert!(CultNetRudpSocketTransportConnection::new(options).is_err());
let mut options = build(socket()?, addr);
options.initial_sequence = u32::MAX - 1;
options.max_pending_reliable_packets = Some(1);
assert!(CultNetRudpSocketTransportConnection::new(options).is_ok());
}
Ok(())
}