From c71a1bc5da8873b9df02fb4d028d891502c851cf Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 22:42:36 +0100 Subject: [PATCH] Port RUDP reliability core to Rust --- docs/cultnet-transport-parity.md | 4 +- packages/cultnet-rs/src/rudp.rs | 396 +++++++++++++++++++++++++++ packages/cultnet-rs/tests/cultnet.rs | 265 ++++++++++++++++++ 3 files changed, 663 insertions(+), 2 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 5dca5f0c..069eb0de 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -160,7 +160,7 @@ Current progress: the RUDP runtime yet; it is the binary packet language the runtimes must converge on before resend loops, windows, and timeout behavior are allowed to claim parity. -- TypeScript and C# now share the first deterministic RUDP reliability state +- TypeScript, C#, and Rust now share the first deterministic RUDP reliability state machine: connect/accept handshake, packet-level ack/ack-mask accounting, reliable resend scheduling, duplicate suppression, and reliable ordered channel delivery. It is in-memory and socket-free so the behavior can be @@ -168,7 +168,7 @@ Current progress: - The remaining parity work is to add the equivalent port to Kotlin, deepen Python's server-side use of its port, and move each runtime's existing TCP/LiteNetLib/WebSocket bodies behind those ports, port the RUDP reliability - state machine into Rust/Python, then bind it to UDP sockets. + state machine into Python, then bind it to UDP sockets. ## RUDP Packet Contract V0 diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index dc35559f..2a6c0332 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -1,5 +1,7 @@ use anyhow::Result; use anyhow::anyhow; +use std::collections::BTreeMap; +use std::collections::BTreeSet; use crate::CultNetTransportChannel; use crate::CultNetTransportDelivery; @@ -40,6 +42,400 @@ pub struct CultNetRudpPacket { pub payload: Vec, } +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CultNetRudpDeliveredFrame { + pub channel_id: String, + pub payload: Vec, + pub sequence: u32, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CultNetRudpReceiveResult { + pub delivered: Vec, + pub reply: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CultNetRudpSessionOptions { + pub connection_id: u32, + pub initial_sequence: u32, + pub resend_delay_ms: u64, +} + +impl Default for CultNetRudpSessionOptions { + fn default() -> Self { + Self { + connection_id: 0, + initial_sequence: 1, + resend_delay_ms: 250, + } + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct CultNetRudpSendOptions { + pub reliable: bool, + pub ordered: bool, + pub sequenced: bool, + pub now_ms: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct PendingReliablePacket { + packet: CultNetRudpPacket, + last_sent_at_ms: u64, +} + +pub struct CultNetRudpSession { + connection_id: u32, + resend_delay_ms: u64, + next_sequence: u32, + connected: bool, + highest_received_sequence: Option, + received_sequences: BTreeSet, + pending_reliable: BTreeMap, + ordered_next_sequence_by_channel: BTreeMap, + ordered_buffers: BTreeMap>, +} + +impl CultNetRudpSession { + pub fn new(options: CultNetRudpSessionOptions) -> Self { + Self { + connection_id: options.connection_id, + resend_delay_ms: options.resend_delay_ms, + next_sequence: options.initial_sequence, + connected: false, + highest_received_sequence: None, + received_sequences: BTreeSet::new(), + pending_reliable: BTreeMap::new(), + ordered_next_sequence_by_channel: BTreeMap::new(), + ordered_buffers: BTreeMap::new(), + } + } + + pub fn connection_id(&self) -> u32 { + self.connection_id + } + + pub fn resend_delay_ms(&self) -> u64 { + self.resend_delay_ms + } + + pub fn connected(&self) -> bool { + self.connected + } + + pub fn pending_reliable_sequences(&self) -> Vec { + self.pending_reliable.keys().copied().collect() + } + + pub fn create_connect(&mut self, now_ms: u64, payload: Vec) -> CultNetRudpPacket { + let packet = self.create_packet( + CultNetRudpPacketType::Connect, + "control", + payload, + true, + true, + false, + ); + self.track_reliable(packet.clone(), now_ms); + packet + } + + pub fn accept_connect( + &mut self, + packet: &CultNetRudpPacket, + now_ms: u64, + payload: Vec, + ) -> Result { + self.require_connection(packet)?; + if packet.packet_type != CultNetRudpPacketType::Connect { + return Err(anyhow!( + "Expected RUDP connect packet, got {:?}", + packet.packet_type + )); + } + + self.remember_received(packet.sequence); + self.connected = true; + let response = self.create_packet( + CultNetRudpPacketType::Accept, + "control", + payload, + true, + true, + false, + ); + self.track_reliable(response.clone(), now_ms); + Ok(response) + } + + pub fn send( + &mut self, + channel_id: &str, + payload: Vec, + options: CultNetRudpSendOptions, + ) -> Result { + if !self.connected { + return Err(anyhow!( + "Cannot send RUDP data before the session is connected" + )); + } + + let packet = self.create_packet( + CultNetRudpPacketType::Data, + channel_id, + payload, + options.reliable, + options.ordered, + options.sequenced, + ); + if packet.reliable { + self.track_reliable(packet.clone(), options.now_ms); + } + Ok(packet) + } + + pub fn receive( + &mut self, + packet: &CultNetRudpPacket, + _now_ms: u64, + ) -> Result { + self.require_connection(packet)?; + self.apply_acknowledgements(packet); + + if packet.packet_type == CultNetRudpPacketType::Accept { + self.remember_received(packet.sequence); + self.connected = true; + return Ok(CultNetRudpReceiveResult { + delivered: Vec::new(), + reply: None, + }); + } + + if packet.packet_type == CultNetRudpPacketType::Ping { + self.remember_received(packet.sequence); + return Ok(CultNetRudpReceiveResult { + delivered: Vec::new(), + reply: Some(self.create_packet( + CultNetRudpPacketType::Pong, + "control", + packet.payload.clone(), + false, + false, + false, + )), + }); + } + + if packet.packet_type == CultNetRudpPacketType::Ack + || packet.packet_type == CultNetRudpPacketType::Pong + { + self.remember_received(packet.sequence); + return Ok(CultNetRudpReceiveResult { + delivered: Vec::new(), + reply: None, + }); + } + + if packet.packet_type != CultNetRudpPacketType::Data { + return Ok(CultNetRudpReceiveResult { + delivered: Vec::new(), + reply: None, + }); + } + + let duplicate = self.received_sequences.contains(&packet.sequence); + self.remember_received(packet.sequence); + if duplicate { + return Ok(CultNetRudpReceiveResult { + delivered: Vec::new(), + reply: None, + }); + } + + let frame = CultNetRudpDeliveredFrame { + channel_id: packet.channel_id.clone(), + payload: packet.payload.clone(), + sequence: packet.sequence, + }; + let delivered = if packet.ordered { + self.deliver_ordered(frame) + } else { + vec![frame] + }; + Ok(CultNetRudpReceiveResult { + delivered, + reply: None, + }) + } + + pub fn create_ack(&mut self) -> CultNetRudpPacket { + self.create_packet( + CultNetRudpPacketType::Ack, + "control", + Vec::new(), + false, + false, + false, + ) + } + + pub fn due_resends(&mut self, now_ms: u64) -> Vec { + let mut due = Vec::new(); + for pending in self.pending_reliable.values_mut() { + if now_ms.saturating_sub(pending.last_sent_at_ms) >= self.resend_delay_ms { + pending.last_sent_at_ms = now_ms; + due.push(pending.packet.clone()); + } + } + due.sort_by_key(|packet| packet.sequence); + due + } + + fn create_packet( + &mut self, + packet_type: CultNetRudpPacketType, + channel_id: &str, + payload: Vec, + reliable: bool, + ordered: bool, + sequenced: bool, + ) -> CultNetRudpPacket { + let sequence = self.next_sequence; + self.next_sequence = self + .next_sequence + .checked_add(1) + .expect("sequence overflow"); + let (ack, ack_mask) = self.ack_state(); + CultNetRudpPacket { + packet_type, + connection_id: self.connection_id, + sequence, + ack, + ack_mask, + channel_id: channel_id.to_string(), + reliable, + ordered, + sequenced, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload, + } + } + + fn track_reliable(&mut self, packet: CultNetRudpPacket, now_ms: u64) { + self.pending_reliable.insert( + packet.sequence, + PendingReliablePacket { + packet, + last_sent_at_ms: now_ms, + }, + ); + } + + fn apply_acknowledgements(&mut self, packet: &CultNetRudpPacket) { + self.pending_reliable.remove(&packet.ack); + for bit in 0..32 { + if (packet.ack_mask & (1_u32 << bit)) != 0 && packet.ack > bit { + self.pending_reliable.remove(&(packet.ack - bit - 1)); + } + } + } + + fn remember_received(&mut self, sequence: u32) { + self.received_sequences.insert(sequence); + if self + .highest_received_sequence + .is_none_or(|highest| sequence > highest) + { + self.highest_received_sequence = Some(sequence); + } + } + + fn ack_state(&self) -> (u32, u32) { + let ack = self.highest_received_sequence.unwrap_or(0); + let mut ack_mask = 0_u32; + for bit in 0..32 { + if ack > bit && self.received_sequences.contains(&(ack - bit - 1)) { + ack_mask |= 1_u32 << bit; + } + } + (ack, ack_mask) + } + + fn deliver_ordered( + &mut self, + frame: CultNetRudpDeliveredFrame, + ) -> Vec { + let channel_id = frame.channel_id.clone(); + let Some(next) = self + .ordered_next_sequence_by_channel + .get(&channel_id) + .copied() + else { + self.ordered_next_sequence_by_channel + .insert(channel_id.clone(), frame.sequence + 1); + let mut delivered = vec![frame]; + delivered.extend(self.drain_ordered(&channel_id)); + return delivered; + }; + + if frame.sequence < next { + return Vec::new(); + } + + if frame.sequence > next { + self.ordered_buffers + .entry(channel_id) + .or_default() + .insert(frame.sequence, frame); + return Vec::new(); + } + + self.ordered_next_sequence_by_channel + .insert(channel_id.clone(), next + 1); + let mut delivered = vec![frame]; + delivered.extend(self.drain_ordered(&channel_id)); + delivered + } + + fn drain_ordered(&mut self, channel_id: &str) -> Vec { + let mut delivered = Vec::new(); + loop { + let Some(next) = self + .ordered_next_sequence_by_channel + .get(channel_id) + .copied() + else { + break; + }; + let Some(buffer) = self.ordered_buffers.get_mut(channel_id) else { + break; + }; + let Some(frame) = buffer.remove(&next) else { + break; + }; + delivered.push(frame); + self.ordered_next_sequence_by_channel + .insert(channel_id.to_string(), next + 1); + } + delivered + } + + fn require_connection(&self, packet: &CultNetRudpPacket) -> Result<()> { + if packet.connection_id != self.connection_id { + return Err(anyhow!( + "RUDP packet connection id {} does not match {}", + packet.connection_id, + self.connection_id + )); + } + Ok(()) + } +} + #[derive(Clone, Debug, Default)] pub struct RudpTransportProfileOptions { pub transport_id: Option, diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index a0037fc4..e06dc052 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -13,6 +13,9 @@ use cultnet_rs::CultNetMessage; use cultnet_rs::CultNetMutationAuthority; use cultnet_rs::CultNetRudpPacket; use cultnet_rs::CultNetRudpPacketType; +use cultnet_rs::CultNetRudpSendOptions; +use cultnet_rs::CultNetRudpSession; +use cultnet_rs::CultNetRudpSessionOptions; use cultnet_rs::CultNetSchemaKind; use cultnet_rs::CultNetSchemaRegistry; use cultnet_rs::CultNetSecret; @@ -265,6 +268,268 @@ fn rudp_transport_profile_advertises_state_and_realtime_channels() { ); } +#[test] +fn rudp_session_handshake_acks_reliable_connect_and_accept_packets() -> Result<()> { + let mut client = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 0x0a0b0c0d, + initial_sequence: 1, + resend_delay_ms: 50, + }); + let mut server = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 0x0a0b0c0d, + initial_sequence: 100, + resend_delay_ms: 50, + }); + + let connect = client.create_connect(0, b"join".to_vec()); + assert_eq!(connect.packet_type, CultNetRudpPacketType::Connect); + assert_eq!(connect.sequence, 1); + assert_eq!(client.pending_reliable_sequences(), vec![1]); + + let accept = server.accept_connect(&connect, 10, b"ok".to_vec())?; + assert_eq!(accept.packet_type, CultNetRudpPacketType::Accept); + assert_eq!(accept.ack, 1); + assert!(server.connected()); + assert_eq!(server.pending_reliable_sequences(), vec![100]); + + client.receive(&accept, 20)?; + assert!(client.connected()); + assert!(client.pending_reliable_sequences().is_empty()); + + let ack = client.create_ack(); + assert_eq!(ack.ack, 100); + server.receive(&ack, 30)?; + assert!(server.pending_reliable_sequences().is_empty()); + Ok(()) +} + +#[test] +fn rudp_session_computes_ack_masks_and_clears_pending_reliable_packets() -> Result<()> { + let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 7, + initial_sequence: 10, + resend_delay_ms: 100, + }); + let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 7, + initial_sequence: 200, + resend_delay_ms: 100, + }); + sender.receive( + &CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Accept, + connection_id: 7, + sequence: 1, + ack: 0, + ack_mask: 0, + channel_id: "control".to_string(), + reliable: false, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: Vec::new(), + }, + 0, + )?; + receiver.receive( + &CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Accept, + connection_id: 7, + sequence: 2, + ack: 0, + ack_mask: 0, + channel_id: "control".to_string(), + reliable: false, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: Vec::new(), + }, + 0, + )?; + + let options = CultNetRudpSendOptions { + reliable: true, + ordered: true, + ..CultNetRudpSendOptions::default() + }; + let first = sender.send("schema", b"first".to_vec(), options.clone())?; + let second = sender.send("schema", b"second".to_vec(), options.clone())?; + let third = sender.send("schema", b"third".to_vec(), options)?; + assert_eq!(sender.pending_reliable_sequences(), vec![10, 11, 12]); + + receiver.receive(&first, 0)?; + receiver.receive(&third, 0)?; + let ack_with_gap = receiver.create_ack(); + assert_eq!(ack_with_gap.ack, 12); + assert_eq!(ack_with_gap.ack_mask, 0b10 | (1 << 9)); + sender.receive(&ack_with_gap, 0)?; + assert_eq!(sender.pending_reliable_sequences(), vec![11]); + + receiver.receive(&second, 0)?; + let full_ack = receiver.create_ack(); + assert_eq!(full_ack.ack, 12); + assert_eq!(full_ack.ack_mask, 0b11 | (1 << 9)); + sender.receive(&full_ack, 0)?; + assert!(sender.pending_reliable_sequences().is_empty()); + Ok(()) +} + +#[test] +fn rudp_session_schedules_reliable_resends_until_acked() -> Result<()> { + let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 99, + initial_sequence: 1, + resend_delay_ms: 100, + }); + session.receive( + &CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Accept, + connection_id: 99, + sequence: 50, + ack: 0, + ack_mask: 0, + channel_id: "control".to_string(), + reliable: false, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: Vec::new(), + }, + 0, + )?; + let sent = session.send( + "schema", + b"payload".to_vec(), + CultNetRudpSendOptions { + reliable: true, + ordered: true, + now_ms: 10, + ..CultNetRudpSendOptions::default() + }, + )?; + + assert!(session.due_resends(90).is_empty()); + assert_eq!( + session + .due_resends(110) + .iter() + .map(|packet| packet.sequence) + .collect::>(), + vec![sent.sequence] + ); + assert!(session.due_resends(150).is_empty()); + + session.receive( + &CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Ack, + connection_id: 99, + sequence: 51, + ack: sent.sequence, + ack_mask: 0, + channel_id: "control".to_string(), + reliable: false, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: Vec::new(), + }, + 0, + )?; + assert!(session.due_resends(250).is_empty()); + Ok(()) +} + +#[test] +fn rudp_session_suppresses_duplicates_and_delivers_reliable_ordered_payloads_in_sequence() +-> Result<()> { + let mut sender = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 123, + initial_sequence: 1, + ..CultNetRudpSessionOptions::default() + }); + let mut receiver = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 123, + initial_sequence: 100, + ..CultNetRudpSessionOptions::default() + }); + sender.receive( + &CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Accept, + connection_id: 123, + sequence: 90, + ack: 0, + ack_mask: 0, + channel_id: "control".to_string(), + reliable: false, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: Vec::new(), + }, + 0, + )?; + receiver.receive( + &CultNetRudpPacket { + packet_type: CultNetRudpPacketType::Accept, + connection_id: 123, + sequence: 91, + ack: 0, + ack_mask: 0, + channel_id: "control".to_string(), + reliable: false, + ordered: false, + sequenced: false, + fragment_id: 0, + fragment_index: 0, + fragment_count: 0, + payload: Vec::new(), + }, + 0, + )?; + + let options = CultNetRudpSendOptions { + reliable: true, + ordered: true, + ..CultNetRudpSendOptions::default() + }; + let first = sender.send("schema", b"first".to_vec(), options.clone())?; + let second = sender.send("schema", b"second".to_vec(), options.clone())?; + let third = sender.send("schema", b"third".to_vec(), options)?; + + assert_eq!( + receiver + .receive(&first, 0)? + .delivered + .iter() + .map(|frame| String::from_utf8(frame.payload.clone()).unwrap()) + .collect::>(), + vec!["first"] + ); + assert!(receiver.receive(&third, 0)?.delivered.is_empty()); + assert!(receiver.receive(&first, 0)?.delivered.is_empty()); + assert_eq!( + receiver + .receive(&second, 0)? + .delivered + .iter() + .map(|frame| String::from_utf8(frame.payload.clone()).unwrap()) + .collect::>(), + vec!["second", "third"] + ); + Ok(()) +} + #[test] fn legacy_gamecult_networking_contract_round_trips_login_union() -> Result<()> { let message = CultNetMessage::Login {