diff --git a/packages/cultnet-rs/src/contracts.rs b/packages/cultnet-rs/src/contracts.rs index e637883c..b5c89ff7 100644 --- a/packages/cultnet-rs/src/contracts.rs +++ b/packages/cultnet-rs/src/contracts.rs @@ -142,6 +142,12 @@ pub struct CultNetTransportChannel { pub max_fragment_bytes: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub max_pending_reliable_packets: Option, + /// Drop a reliable packet on this channel instead of retransmitting it once + /// this many milliseconds have passed since the send. Omitted when absent, + /// so a profile that does not set it is byte-identical to one from a + /// runtime that does not know the field. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reliable_expire_after_ms: Option, } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index e54cae54..f1fd4ca4 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -1266,7 +1266,6 @@ pub struct CultNetRudpSocketTransportConnection { max_fragment_bytes: Option, disconnect_reason: Option>, pong_payloads: VecDeque>, - media_reliable_expire_after_ms: Option, } pub struct CultNetRudpServerHubOptions { @@ -1351,7 +1350,6 @@ pub struct CultNetRudpServerHub { max_payload_bytes: usize, max_fragment_bytes: Option, max_peers: usize, - media_reliable_expire_after_ms: Option, next_session_generation: u64, peers: BTreeMap, pending_events: VecDeque, @@ -1373,6 +1371,7 @@ impl CultNetRudpServerHub { let profile = create_rudp_transport_profile( options.runtime_id, RudpTransportProfileOptions { + media_reliable_expire_after_ms: options.media_reliable_expire_after_ms, transport_id: options.transport_id, host: Some(local_addr.ip().to_string()), port: Some(local_addr.port()), @@ -1383,7 +1382,6 @@ impl CultNetRudpServerHub { }, ); Ok(Self { - media_reliable_expire_after_ms: options.media_reliable_expire_after_ms, socket: options.socket, connection_id: options.connection_id, initial_sequence: options.initial_sequence, @@ -1449,7 +1447,11 @@ impl CultNetRudpServerHub { peer.session.send_many( channel_id, payload, - channel_send_options(channel_id, now_ms(), self.media_reliable_expire_after_ms), + channel_send_options( + channel_id, + now_ms(), + channel_reliable_expire_after_ms(&self.profile, channel_id), + ), self.max_fragment_bytes, )? }; @@ -1668,6 +1670,7 @@ impl CultNetRudpSocketTransportConnection { let profile = create_rudp_transport_profile( options.runtime_id, RudpTransportProfileOptions { + media_reliable_expire_after_ms: options.media_reliable_expire_after_ms, transport_id: options.transport_id, host: Some(local_addr.ip().to_string()), port: Some(local_addr.port()), @@ -1688,7 +1691,6 @@ impl CultNetRudpSocketTransportConnection { }); session.set_max_payload_bytes(max_payload_bytes.map(|value| value as usize)); Ok(Self { - media_reliable_expire_after_ms: options.media_reliable_expire_after_ms, socket: options.socket, session, mode: options.mode, @@ -1732,7 +1734,11 @@ impl CultNetRudpSocketTransportConnection { pub fn send(&mut self, channel_id: &str, payload: Vec) -> Result<()> { let options = - channel_send_options(channel_id, now_ms(), self.media_reliable_expire_after_ms); + channel_send_options( + channel_id, + now_ms(), + channel_reliable_expire_after_ms(&self.profile, channel_id), + ); let packets = self.session .send_many(channel_id, payload, options, self.max_fragment_bytes)?; @@ -1749,7 +1755,11 @@ impl CultNetRudpSocketTransportConnection { payload: Vec, ) -> Result { let options = - channel_send_options(channel_id, now_ms(), self.media_reliable_expire_after_ms); + channel_send_options( + channel_id, + now_ms(), + channel_reliable_expire_after_ms(&self.profile, channel_id), + ); // This lineage has no reliable-expiry concept, so every reliable channel is // already non-expiring. Restore the expiry half of this guard if one is added. if !options.reliable { @@ -2062,6 +2072,7 @@ pub struct RudpTransportProfileOptions { pub max_fragment_bytes: Option, pub max_pending_reliable_packets: Option, pub reconnect_policy: Option, + pub media_reliable_expire_after_ms: Option, } pub fn create_rudp_transport_profile( @@ -2092,6 +2103,7 @@ pub fn create_rudp_transport_profile( max_payload_bytes: options.max_payload_bytes, max_fragment_bytes: options.max_fragment_bytes, max_pending_reliable_packets: options.max_pending_reliable_packets, + reliable_expire_after_ms: None, }, CultNetTransportChannel { channel_id: "latest".to_string(), @@ -2100,6 +2112,7 @@ pub fn create_rudp_transport_profile( max_payload_bytes: options.max_payload_bytes, max_fragment_bytes: options.max_fragment_bytes, max_pending_reliable_packets: options.max_pending_reliable_packets, + reliable_expire_after_ms: None, }, CultNetTransportChannel { channel_id: "realtime".to_string(), @@ -2108,6 +2121,7 @@ pub fn create_rudp_transport_profile( max_payload_bytes: options.max_payload_bytes, max_fragment_bytes: options.max_fragment_bytes, max_pending_reliable_packets: options.max_pending_reliable_packets, + reliable_expire_after_ms: None, }, CultNetTransportChannel { channel_id: "media".to_string(), @@ -2116,6 +2130,7 @@ pub fn create_rudp_transport_profile( max_payload_bytes: options.max_payload_bytes, max_fragment_bytes: options.max_fragment_bytes, max_pending_reliable_packets: options.max_pending_reliable_packets, + reliable_expire_after_ms: options.media_reliable_expire_after_ms, }, ], }], @@ -2224,10 +2239,25 @@ fn packet_type_to_code(packet_type: CultNetRudpPacketType) -> u8 { } } +/// The advertised profile owns each channel's expiry. Resolving it here keeps +/// one answer for what a channel does, rather than a runtime field that can +/// disagree with the profile the peer was shown. +fn channel_reliable_expire_after_ms( + profile: &CultNetTransportProfile, + channel_id: &str, +) -> Option { + profile + .transports + .iter() + .flat_map(|transport| transport.channels.iter()) + .find(|channel| channel.channel_id == channel_id) + .and_then(|channel| channel.reliable_expire_after_ms) +} + fn channel_send_options( channel_id: &str, now_ms: u64, - media_reliable_expire_after_ms: Option, + reliable_expire_after_ms: Option, ) -> CultNetRudpSendOptions { match channel_id { "schema" => CultNetRudpSendOptions { @@ -2249,7 +2279,7 @@ fn channel_send_options( ordered: false, sequenced: false, now_ms, - reliable_expire_after_ms: media_reliable_expire_after_ms, + reliable_expire_after_ms, }, _ => CultNetRudpSendOptions { reliable: false, diff --git a/packages/cultnet-rs/src/transport.rs b/packages/cultnet-rs/src/transport.rs index f10ad87d..cf28079e 100644 --- a/packages/cultnet-rs/src/transport.rs +++ b/packages/cultnet-rs/src/transport.rs @@ -204,6 +204,7 @@ pub fn create_tcp_framed_transport_profile( max_payload_bytes: options.max_payload_bytes, max_fragment_bytes: options.max_fragment_bytes, max_pending_reliable_packets: None, + reliable_expire_after_ms: None, }], }], } diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index de84d2d9..851646cf 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -258,6 +258,7 @@ fn cultnet_schema_messages_round_trip_through_messagepack_frames() -> Result<()> max_payload_bytes: None, max_fragment_bytes: None, max_pending_reliable_packets: None, + reliable_expire_after_ms: None, }], }], }]), @@ -365,6 +366,7 @@ fn rudp_transport_profile_advertises_state_and_realtime_channels() { max_fragment_bytes: Some(1000), max_pending_reliable_packets: Some(64), reconnect_policy: None, + media_reliable_expire_after_ms: None, }, ); @@ -2817,3 +2819,33 @@ fn accept_rudp_session(session: &mut CultNetRudpSession, connection_id: u32) -> )?; Ok(()) } + +#[test] +fn rudp_profile_expires_only_the_media_channel() -> Result<()> { + let socket = std::net::UdpSocket::bind("127.0.0.1:0")?; + let peer = socket.local_addr()?; + let options = CultNetRudpSocketTransportOptions::client("expiry-profile", socket, peer, 11); + let connection = CultNetRudpSocketTransportConnection::new(options)?; + + let channels = connection + .profile + .transports + .iter() + .flat_map(|transport| transport.channels.iter()) + .map(|channel| (channel.channel_id.as_str(), channel.reliable_expire_after_ms)) + .collect::>(); + + assert!( + channels.contains(&("media", Some(75))), + "the media channel advertises its expiry so a peer can see it: {channels:?}" + ); + for (channel_id, expiry) in &channels { + if *channel_id != "media" { + assert_eq!( + *expiry, None, + "state channel {channel_id} must never expire a reliable send" + ); + } + } + Ok(()) +}