diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 29467890..5f7084ce 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -126,8 +126,8 @@ pub struct CultNetRudpSendOptions { } /// Identifies every transport packet belonging to one non-expiring reliable -/// send. Receipts are local to the session that issued them and cannot survive -/// a peer reset. +/// send. Receipts are local to the session that issued them: a peer reset or +/// a Disconnect invalidates them for good. #[derive(Clone, Debug, PartialEq, Eq)] pub struct CultNetRudpReliableSendReceipt { session_scope: Uuid, @@ -278,6 +278,10 @@ impl CultNetRudpSession { Some(pending.packet.clone()) } + fn session_scope(&self) -> Uuid { + self.session_scope + } + fn reliable_send_status( &self, receipt: &CultNetRudpReliableSendReceipt, @@ -588,7 +592,13 @@ impl CultNetRudpSession { } if packet.packet_type == CultNetRudpPacketType::Disconnect { + // The session is over. What it still owed the peer dies with it, + // and receipts it issued can never become Acknowledged in a later + // session. self.connected = false; + self.session_scope = Uuid::new_v4(); + self.pending_reliable.clear(); + self.queued_reliable.clear(); return Ok(CultNetRudpReceiveResult { delivered: Vec::new(), ready_to_send, @@ -1920,12 +1930,7 @@ impl CultNetRudpSocketTransportConnection { &self, receipt: &CultNetRudpReliableSendReceipt, ) -> CultNetRudpReliableSendStatus { - 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 + self.session.reliable_send_status(receipt) } fn send_data_packets(&mut self, packets: &[CultNetRudpPacket]) -> Result<()> { @@ -1990,15 +1995,18 @@ impl CultNetRudpSocketTransportConnection { self.socket.set_read_timeout(Some(poll_timeout))?; let deadline = Instant::now() + timeout; + // A flush belongs to the session it started in: a Connect that + // replaces the peer mid-wait forgets the writes being waited on. + let scope = self.session.session_scope(); let mut preserved_frames = VecDeque::new(); let result = (|| { loop { // An ended session forgot its unacknowledged writes; reporting // them flushed would be a lie. - if let Some(reason) = &self.disconnect_reason { + if self.session.session_scope() != scope || self.disconnect_reason.is_some() { return Err(anyhow!( "RUDP session ended before its reliable writes were acknowledged: {}", - String::from_utf8_lossy(reason) + String::from_utf8_lossy(self.disconnect_reason.as_deref().unwrap_or(b"replaced by a new Connect")) )); } if self.outstanding_reliable_packet_count() == 0 { diff --git a/packages/cultnet-rs/tests/rudp_stray_packets.rs b/packages/cultnet-rs/tests/rudp_stray_packets.rs index bbb96395..25dd80e6 100644 --- a/packages/cultnet-rs/tests/rudp_stray_packets.rs +++ b/packages/cultnet-rs/tests/rudp_stray_packets.rs @@ -620,3 +620,132 @@ fn a_write_unacknowledged_when_the_peer_disconnects_fails_its_flush() -> Result< ); Ok(()) } + +/// Drains what a raw peer socket has been sent. +fn drain(socket: &UdpSocket) -> Result> { + let mut buffer = vec![0_u8; 65_535]; + let mut packets = Vec::new(); + while let Ok((received, _)) = socket.recv_from(&mut buffer) { + packets.push(decode_rudp_packet(&buffer[..received])?); + } + Ok(packets) +} + +/// A receipt belongs to the session it was issued in. A peer's Disconnect ends +/// that session for good: reconnecting must not bring the write back to Pending, +/// and an ack the old peer sent late must not make it Acknowledged. +#[test] +fn a_receipt_invalidated_by_a_peer_disconnect_never_becomes_acknowledged() -> 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 old_peer = raw_session(CONNECTION_ID); + let accept = old_peer.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"written before the disconnect".to_vec())?; + let data = drain(&server)? + .into_iter() + .find(|packet| packet.packet_type == CultNetRudpPacketType::Data) + .expect("the write reached the peer"); + old_peer.receive(&data, 0)?; + let late_ack = old_peer.create_ack_for_received(data.sequence); + send_to(&server, client_addr, &old_peer.create_disconnect(b"bye".to_vec()))?; + let _ = client.receive_once()?; + assert_eq!(client.reliable_send_status(&receipt), CultNetRudpReliableSendStatus::Invalidated); + + client.connect(b"hello again".to_vec())?; + assert_eq!( + client.reliable_send_status(&receipt), + CultNetRudpReliableSendStatus::Invalidated, + "reconnecting brought the write back to Pending" + ); + let connect = drain(&server)? + .into_iter() + .find(|packet| packet.packet_type == CultNetRudpPacketType::Connect) + .expect("the reconnect reached the peer"); + let mut new_peer = raw_session(CONNECTION_ID); + let new_accept = new_peer.accept_connect(&connect, 0, Vec::new())?; + send_to(&server, client_addr, &new_accept)?; + send_to(&server, client_addr, &late_ack)?; + while client.receive_once()?.is_some() {} + assert_eq!( + client.reliable_send_status(&receipt), + CultNetRudpReliableSendStatus::Invalidated, + "the old session's write was acknowledged in the new session" + ); + // The old write is not retransmitted into the new session either. + assert_eq!(client.outstanding_reliable_packet_count(), 1, "only the reconnect's own Connect is outstanding"); + Ok(()) +} + +/// A server-mode transport that accepts a new Connect has a live session again; +/// the reason the previous one ended is history, not state. +#[test] +fn server_mode_forgets_the_end_reason_once_a_new_connect_is_accepted() -> 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); + send_to(&peer, server_addr, &peer_session.create_connect(0, b"peer".to_vec())?)?; + let _ = server.receive_once()?; + send_to(&peer, server_addr, &peer_session.create_disconnect(b"bye".to_vec()))?; + let _ = server.receive_once()?; + assert_eq!(server.disconnect_reason(), Some(&b"bye"[..])); + + let mut next_peer = raw_session(CONNECTION_ID); + send_to(&peer, server_addr, &next_peer.create_connect(0, b"peer".to_vec())?)?; + let _ = server.receive_once()?; + assert!(server.connected()); + assert_eq!(server.disconnect_reason(), None, "a stale reason survives the new session"); + Ok(()) +} + +/// A flush waits on one session. A Connect from another endpoint replaces the +/// peer and forgets the writes being waited on; the flush must say so. +#[test] +fn a_flush_fails_when_a_new_endpoint_replaces_the_peer_mid_wait() -> 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_a = socket()?; + let peer_b = socket()?; + send_to(&peer_a, server_addr, &raw_session(CONNECTION_ID).create_connect(0, b"a".to_vec())?)?; + let _ = server.receive_once()?; + server.send_reliable("schema", b"for A".to_vec())?; + send_to(&peer_b, server_addr, &raw_session(CONNECTION_ID).create_connect(0, b"b".to_vec())?)?; + + let error = server.flush_reliable(Duration::from_millis(300)).expect_err("A's write was forgotten"); + assert!( + error.to_string().contains("ended before its reliable writes were acknowledged"), + "{error}" + ); + Ok(()) +} + +/// A repeated Connect still carries an acknowledgement field, and the session +/// honours it: the Accept it acknowledges is no longer owed. +#[test] +fn a_repeated_connect_acknowledges_what_it_carries() -> Result<()> { + let mut client = raw_session(CONNECTION_ID); + let mut server = raw_session(CONNECTION_ID); + let connect = client.create_connect(0, Vec::new())?; + let accept = server.accept_connect(&connect, 0, Vec::new())?; + let mut repeat = connect.clone(); + repeat.ack = accept.sequence; + let reply = server.answer_repeated_connect(&repeat, 1)?; + assert_eq!(reply.packet_type, CultNetRudpPacketType::Ack); + assert_eq!(server.outstanding_reliable_packet_count(), 0); + Ok(()) +}