From cbde940df2f5ee0ccac50e8dbfe9f36ffbebb585 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Fri, 4 Sep 2026 03:57:44 +0200 Subject: [PATCH] Let the advertised profile own each channel's reliable expiry The first pass put the media deadline on the transport options and carried it as a runtime field on the connection and hub. That is a second answer to what a channel does, able to disagree with the profile the peer was actually shown. CultNetTransportChannel now carries reliable_expire_after_ms and the send path resolves the deadline from the profile, so there is one answer. The runtime fields are deleted rather than left as a shadow copy. The transport option remains as the knob that populates the profile, not as a parallel authority. Odin's fork had the field in exactly this place, which is how the gap surfaced: its muninn test reads it off the channel descriptor. This is wire-visible, unlike the rest of the expiry work. The field is optional and skipped when absent, so a profile that does not set it is byte-identical to one from a runtime that has never heard of it -- but the RUDP profile does set it on the media channel, so Rust peers now advertise it. Verified against the interop lane: TypeScript, C# and Kotlin all accept the profile unchanged, in five passing lanes each. The Python lanes cannot run here because this machine's Python runtime has no msgpack, and they fail identically before and after this change. The other four runtimes should grow the field so a profile round-trips through them without losing the channel's declared behaviour. cultnet-rs: 107 tests. Co-Authored-By: Claude Opus 5 --- packages/cultnet-rs/src/contracts.rs | 6 ++++ packages/cultnet-rs/src/rudp.rs | 48 ++++++++++++++++++++++------ packages/cultnet-rs/src/transport.rs | 1 + packages/cultnet-rs/tests/cultnet.rs | 32 +++++++++++++++++++ 4 files changed, 78 insertions(+), 9 deletions(-) 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(()) +}