diff --git a/packages/cultmesh-rs/src/rudp_document_server.rs b/packages/cultmesh-rs/src/rudp_document_server.rs index 524d17c7..5536add0 100644 --- a/packages/cultmesh-rs/src/rudp_document_server.rs +++ b/packages/cultmesh-rs/src/rudp_document_server.rs @@ -411,17 +411,17 @@ where now: u64, ) -> Result { let reply = match self.sessions.get_mut(&key) { - // The same key is a retransmitted Connect for the existing epoch: - // repeat the Accept still awaiting acknowledgement, or acknowledge - // if none is. It queues nothing, so a Connect storm cannot grow the - // session's reliable queue. A genuinely fresh client incarnation - // must choose a fresh id. + // The same key is a retransmitted Connect for the existing epoch. + // A genuinely fresh client incarnation must choose a fresh id. Some(entry) => { entry.last_activity_monotonic_millis = now; - entry - .session - .pending_accept_for_resend(now) - .unwrap_or_else(|| entry.session.create_ack()) + match entry.session.answer_repeated_connect(packet, now) { + Ok(reply) => reply, + Err(_) => { + self.sessions.remove(&key); + return Ok(false); + } + } } None => { if self.sessions.len() >= self.options.max_sessions { diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index 980bf04b..d3ee1de8 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -581,7 +581,7 @@ fn a_connect_storm_and_stray_frames_are_dropped_not_fatal() -> Result<()> { max_pending_reliable_packets: None, }) }; - let mut send_raw = |server: &mut Server, packet: &cultnet_rs::CultNetRudpPacket| -> Result<()> { + let send_raw = |server: &mut Server, packet: &cultnet_rs::CultNetRudpPacket| -> Result<()> { raw.send_to(&encode_rudp_packet(packet)?, target)?; server.poll_once()?; Ok(()) diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index f91838a2..00e0e61b 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -271,9 +271,7 @@ impl CultNetRudpSession { self.pending_reliable.len() + self.queued_reliable.len() } - /// 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 { + fn pending_accept_for_resend(&mut self, now_ms: u64) -> Option { let pending = self .pending_reliable .values_mut() @@ -397,6 +395,30 @@ impl CultNetRudpSession { Ok(response) } + /// Answers a Connect from a peer this session has already accepted. The + /// packet's sequence is remembered and nothing is queued, so a Connect storm + /// cannot grow the reliable queue. The reply is the Accept still awaiting + /// acknowledgement, or an Ack once it was acknowledged. + pub fn answer_repeated_connect( + &mut self, + packet: &CultNetRudpPacket, + now_ms: u64, + ) -> Result { + self.require_connection(packet)?; + if packet.packet_type != CultNetRudpPacketType::Connect { + return Err(anyhow!( + "Expected RUDP connect packet, got {:?}", + packet.packet_type + )); + } + self.apply_acknowledgements(packet); + self.remember_received(packet.sequence); + self.last_received_at_ms = Some(now_ms); + Ok(self + .pending_accept_for_resend(now_ms) + .unwrap_or_else(|| self.create_ack())) + } + pub fn send( &mut self, channel_id: &str, @@ -1594,14 +1616,10 @@ impl CultNetRudpServerHub { if let Some(peer) = self.peers.get_mut(&remote_addr) && peer.context.connect_payload == packet.payload { - if peer.session.receive(&packet, now_ms()).is_err() { + let Ok(reply) = peer.session.answer_repeated_connect(&packet, now_ms()) else { self.end_refusing_peer(remote_addr)?; return Ok(true); - } - let reply = peer - .session - .pending_accept_for_resend(now_ms()) - .unwrap_or_else(|| peer.session.create_ack()); + }; self.send_packet(remote_addr, &reply)?; return Ok(true); } @@ -1712,6 +1730,9 @@ impl CultNetRudpServerHub { 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()); self.pending_events .push_back(CultNetRudpServerEvent::Disconnected { @@ -2068,10 +2089,18 @@ impl CultNetRudpSocketTransportConnection { // 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; + // 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); - self.send_packet(&goodbye)?; + let sent = self.send_packet(&goodbye); + if self.mode == CultNetRudpSocketMode::Server { + self.remote_addr = None; + } + sent?; return Ok(false); }; if let Some(reply) = result.reply { diff --git a/packages/cultnet-rs/tests/rudp_stray_packets.rs b/packages/cultnet-rs/tests/rudp_stray_packets.rs index 7a371e9e..161773f6 100644 --- a/packages/cultnet-rs/tests/rudp_stray_packets.rs +++ b/packages/cultnet-rs/tests/rudp_stray_packets.rs @@ -388,9 +388,12 @@ fn client_ends_its_session_when_it_refuses_a_packet() -> Result<()> { 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); + let mut told = false; + while let Ok((received, _)) = server.recv_from(&mut buffer) { + told |= decode_rudp_packet(&buffer[..received])?.packet_type + == CultNetRudpPacketType::Disconnect; + } + assert!(told, "the server must be told the session ended"); Ok(()) }