GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Let the advertised profile own each channel's reliable expiry

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-04 03:57:44 +02:00
commit cbde940df2
4 changed files with 78 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

6
packages/cultnet-rs/src/contracts.rs
View file

@ -142,6 +142,12 @@ pub struct CultNetTransportChannel {
pub max_fragment_bytes: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_pending_reliable_packets: Option<u32>,
/// 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<u64>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]

48
packages/cultnet-rs/src/rudp.rs
View file

@ -1266,7 +1266,6 @@ pub struct CultNetRudpSocketTransportConnection {
max_fragment_bytes: Option<usize>,
disconnect_reason: Option<Vec<u8>>,
pong_payloads: VecDeque<Vec<u8>>,
media_reliable_expire_after_ms: Option<u64>,
}
pub struct CultNetRudpServerHubOptions {
@ -1351,7 +1350,6 @@ pub struct CultNetRudpServerHub {
max_payload_bytes: usize,
max_fragment_bytes: Option<usize>,
max_peers: usize,
media_reliable_expire_after_ms: Option<u64>,
next_session_generation: u64,
peers: BTreeMap<SocketAddr, CultNetRudpServerPeer>,
pending_events: VecDeque<CultNetRudpServerEvent>,
@ -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<u8>) -> 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<u8>,
) -> Result<CultNetRudpReliableSendReceipt> {
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<u32>,
pub max_pending_reliable_packets: Option<u32>,
pub reconnect_policy: Option<CultNetReconnectPolicy>,
pub media_reliable_expire_after_ms: Option<u64>,
}
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<u64> {
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<u64>,
reliable_expire_after_ms: Option<u64>,
) -> 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,

1
packages/cultnet-rs/src/transport.rs
View file

@ -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,
}],
}],
}

32
packages/cultnet-rs/tests/cultnet.rs
View file

@ -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::<Vec<_>>();
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(())
}