diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 5536add0..b99c7145 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -325,7 +325,7 @@ where let result = match result { Ok(result) => result, Err(_) => { - self.sessions.remove(&key); + self.end_refused_session(key)?; self.packets_dropped += 1; return Ok(CultMeshRudpPollOutcome::Handled); } @@ -418,7 +418,7 @@ where match entry.session.answer_repeated_connect(packet, now) { Ok(reply) => reply, Err(_) => { - self.sessions.remove(&key); + self.end_refused_session(key)?; return Ok(false); } } @@ -592,6 +592,16 @@ where Ok(None) } + /// The session refused a packet: it ends, and the client is told so it does + /// not wait on a session the server no longer holds. + fn end_refused_session(&mut self, key: CultMeshRudpSessionKey) -> Result<()> { + let Some(mut entry) = self.sessions.remove(&key) else { + return Ok(()); + }; + let goodbye = entry.session.end_refused_session(); + self.send_packet(key.remote_addr, &goodbye) + } + fn payload_budget_allows(&self, key: CultMeshRudpSessionKey, bytes: usize) -> bool { let Some(session_bytes) = self .sessions diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index d3ee1de8..fa741827 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -627,6 +627,15 @@ fn a_connect_storm_and_stray_frames_are_dropped_not_fatal() -> Result<()> { send_raw(&mut server, &poison)?; assert_eq!(server.packets_dropped(), 3); assert_eq!(server.session_count(), 1); + // ...and the refused client is told, after the Accepts its storm drew. + raw.set_read_timeout(Some(Duration::from_millis(200)))?; + let mut buffer = vec![0_u8; 65_535]; + let mut told = false; + while let Ok(received) = raw.recv(&mut buffer) { + told |= cultnet_rs::decode_rudp_packet(&buffer[..received])?.packet_type + == cultnet_rs::CultNetRudpPacketType::Disconnect; + } + assert!(told, "the refused client must be sent a goodbye"); send( &mut real, diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 00e0e61b..55a8d8cf 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -334,11 +334,12 @@ impl CultNetRudpSession { Ok(()) } + /// Forgets everything learned from the peer. Sequence numbers and fragment + /// ids already issued stay issued: a peer that still remembers them must + /// never see one reused. pub fn reset_peer_state(&mut self) { self.session_scope = Uuid::new_v4(); - self.next_sequence = self.initial_sequence; self.next_sequenced_by_channel.clear(); - self.next_fragment_id = 1; self.connected = false; self.last_received_at_ms = None; self.highest_received_sequence = None; @@ -351,6 +352,16 @@ impl CultNetRudpSession { self.fragment_buffers.clear(); } + /// Ends a session that refused a packet. `receive` has already recorded the + /// refused packet's reliable sequence, so the session cannot be kept: a + /// retransmit would be acknowledged and the frame silently lost. Returns the + /// goodbye for the peer, built after the reset or its ack field would + /// acknowledge the very frame that was refused. + pub fn end_refused_session(&mut self) -> CultNetRudpPacket { + self.reset_peer_state(); + self.create_disconnect(RUDP_REFUSED_PACKET_REASON.to_vec()) + } + pub fn create_connect(&mut self, now_ms: u64, payload: Vec) -> Result { self.ensure_reliable_capacity(1)?; let packet = self.create_packet( @@ -1729,15 +1740,11 @@ impl CultNetRudpServerHub { let Some(mut peer) = self.peers.remove(&remote_addr) else { return Ok(()); }; - let reason = RUDP_REFUSED_PACKET_REASON.to_vec(); - // Forget the peer's sequences first: the goodbye's ack field would - // otherwise acknowledge the very frame the session refused. - peer.session.reset_peer_state(); - let goodbye = peer.session.create_disconnect(reason.clone()); + let goodbye = peer.session.end_refused_session(); self.pending_events .push_back(CultNetRudpServerEvent::Disconnected { session: peer.context, - reason, + reason: goodbye.payload.clone(), }); self.send_packet(remote_addr, &goodbye) } @@ -1861,6 +1868,7 @@ impl CultNetRudpSocketTransportConnection { )); } let packet = self.session.create_connect(now_ms(), payload)?; + self.disconnect_reason = None; self.send_packet(&packet) } @@ -1914,7 +1922,12 @@ impl CultNetRudpSocketTransportConnection { &self, receipt: &CultNetRudpReliableSendReceipt, ) -> CultNetRudpReliableSendStatus { - self.session.reliable_send_status(receipt) + let status = self.session.reliable_send_status(receipt); + // A write still unacknowledged when the session ended never will be. + if status == CultNetRudpReliableSendStatus::Pending && self.disconnect_reason.is_some() { + return CultNetRudpReliableSendStatus::Invalidated; + } + status } fn send_data_packets(&mut self, packets: &[CultNetRudpPacket]) -> Result<()> { @@ -1981,7 +1994,18 @@ impl CultNetRudpSocketTransportConnection { let deadline = Instant::now() + timeout; let mut preserved_frames = VecDeque::new(); let result = (|| { - while self.outstanding_reliable_packet_count() > 0 { + loop { + // An ended session forgot its unacknowledged writes; reporting + // them flushed would be a lie. + if let Some(reason) = &self.disconnect_reason { + return Err(anyhow!( + "RUDP session ended before its reliable writes were acknowledged: {}", + String::from_utf8_lossy(reason) + )); + } + if self.outstanding_reliable_packet_count() == 0 { + return Ok(()); + } if Instant::now() >= deadline { return Err(anyhow!( "RUDP reliable flush timed out with {} packets outstanding", @@ -1993,7 +2017,6 @@ impl CultNetRudpSocketTransportConnection { } self.poll_resends()?; } - Ok(()) })(); preserved_frames.append(&mut self.delivered_frames); @@ -2076,6 +2099,7 @@ impl CultNetRudpSocketTransportConnection { && packet.packet_type == CultNetRudpPacketType::Connect { self.session.reset_peer_state(); + self.disconnect_reason = None; let Ok(accept) = self.session.accept_connect(&packet, now_ms(), Vec::new()) else { self.stats.packets_dropped += 1; return Ok(true); @@ -2085,17 +2109,11 @@ 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. + // End the session and tell the peer. A server-mode transport also + // forgets its peer: only a Connect can claim it again. self.stats.packets_dropped += 1; - // The goodbye is built after the reset, or its ack field would - // acknowledge the very frame the session refused. A server-mode - // transport also forgets its peer: only a Connect can claim it again. - let reason = RUDP_REFUSED_PACKET_REASON.to_vec(); - self.session.reset_peer_state(); - let goodbye = self.session.create_disconnect(reason.clone()); - self.disconnect_reason = Some(reason); + let goodbye = self.session.end_refused_session(); + self.disconnect_reason = Some(goodbye.payload.clone()); let sent = self.send_packet(&goodbye); if self.mode == CultNetRudpSocketMode::Server { self.remote_addr = None; diff --git a/packages/cultnet-rs/tests/rudp_stray_packets.rs b/packages/cultnet-rs/tests/rudp_stray_packets.rs index 161773f6..f60cf582 100644 --- a/packages/cultnet-rs/tests/rudp_stray_packets.rs +++ b/packages/cultnet-rs/tests/rudp_stray_packets.rs @@ -467,3 +467,118 @@ fn constructors_reject_limits_that_make_a_connect_unadmittable() -> Result<()> { } Ok(()) } + +/// The distinct reliable Accept sequences a peer has been sent so far. +fn accept_sequences(peer: &UdpSocket) -> Result> { + let mut buffer = vec![0_u8; 65_535]; + let mut sequences = std::collections::BTreeSet::new(); + while let Ok((received, _)) = peer.recv_from(&mut buffer) { + let packet = decode_rudp_packet(&buffer[..received])?; + if packet.packet_type == CultNetRudpPacketType::Accept && packet.reliable { + sequences.insert(packet.sequence); + } + } + Ok(sequences) +} + +/// A Connect that repeats (a moved flow, a retransmit, a storm) is answered +/// with the Accept already owed, never a fresh reliable Accept per Connect. +#[test] +fn hub_owes_one_reliable_accept_however_many_connects_repeat() -> 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); + let connect = peer_session.create_connect(0, b"peer".to_vec())?; + for _ in 0..20 { + send_to(&peer, hub_addr, &connect)?; + while hub.receive_event_once()?.is_some() {} + } + assert_eq!(accept_sequences(&peer)?.len(), 1); + Ok(()) +} + +#[test] +fn server_mode_owes_one_reliable_accept_however_many_connects_repeat() -> Result<()> { + let server_socket = socket()?; + let server_addr = server_socket.local_addr()?; + let mut server = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID), + )?; + let peer = socket()?; + let mut peer_session = raw_session(CONNECTION_ID); + let connect = peer_session.create_connect(0, b"peer".to_vec())?; + for _ in 0..20 { + send_to(&peer, server_addr, &connect)?; + let _ = server.receive_once()?; + } + assert!(server.outstanding_reliable_packet_count() <= 1); + Ok(()) +} + +/// A late duplicate Connect resets what the server learned of the peer, never +/// what it already issued: a reused sequence is dropped by the peer as a +/// duplicate. +#[test] +fn server_mode_never_reuses_a_sequence_after_a_late_duplicate_connect() -> Result<()> { + let server_socket = socket()?; + let server_addr = server_socket.local_addr()?; + let mut server = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::server("server", server_socket, CONNECTION_ID), + )?; + let peer = socket()?; + let mut peer_session = raw_session(CONNECTION_ID); + let connect = peer_session.create_connect(0, b"peer".to_vec())?; + send_to(&peer, server_addr, &connect)?; + let _ = server.receive_once()?; + server.send_reliable("schema", b"before".to_vec())?; + send_to(&peer, server_addr, &connect)?; + let _ = server.receive_once()?; + server.send_reliable("schema", b"after".to_vec())?; + + let mut buffer = vec![0_u8; 65_535]; + let mut delivered = Vec::new(); + while let Ok((received, _)) = peer.recv_from(&mut buffer) { + let result = peer_session.receive(&decode_rudp_packet(&buffer[..received])?, 0)?; + delivered.extend(result.delivered.into_iter().map(|frame| frame.payload)); + } + assert!(delivered.contains(&b"before".to_vec())); + assert!( + delivered.contains(&b"after".to_vec()), + "the frame sent after the duplicate Connect was dropped as a reused sequence" + ); + Ok(()) +} + +/// A session that ended forgot its unacknowledged writes; a flush that then +/// reports success would claim writes the peer never acknowledged. +#[test] +fn a_write_unacknowledged_when_the_session_is_refused_fails_its_flush() -> Result<()> { + let server = socket()?; + let server_addr = server.local_addr()?; + let mut client = CultNetRudpSocketTransportConnection::new( + CultNetRudpSocketTransportOptions::client("client", socket()?, server_addr, CONNECTION_ID), + )?; + client.connect(b"hello".to_vec())?; + let mut buffer = vec![0_u8; 65_535]; + let (received, client_addr) = server.recv_from(&mut buffer)?; + let mut server_session = raw_session(CONNECTION_ID); + let accept = + server_session.accept_connect(&decode_rudp_packet(&buffer[..received])?, 0, Vec::new())?; + send_to(&server, client_addr, &accept)?; + let _ = client.receive_once()?; + + let receipt = client.send_reliable("schema", b"never acknowledged".to_vec())?; + send_to(&server, client_addr, &poisoned(&mut server_session)?)?; + let _ = client.receive_once()?; + assert!(client.disconnect_reason().is_some()); + + let flushed = client.flush_reliable(Duration::from_millis(200)); + assert!(flushed.is_err(), "the write was never acknowledged"); + assert_eq!( + client.reliable_send_status(&receipt), + CultNetRudpReliableSendStatus::Invalidated + ); + Ok(()) +}