From 36f31920ce69a80ad4c340a7bdfb23d63a9273e9 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 10 Sep 2026 00:49:32 +0200 Subject: [PATCH] Let a caller choose what the media channel does MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `media` was hardcoded Reliable in two places — the profile builder and the per-channel send options — so no caller could ask for anything else. Measured on Raven through `cultnet-impair`, 848-byte packets at 12 Mbps, which is Muninn's real media shape: profile mode delivered ack bytes clean unreliable 100.00% 43 clean reliable 69.10% 131,322 loss 1% unreliable 98.95% 43 loss 1% reliable 72.00% 135,837 loss 3% unreliable 96.73% 43 loss 3% reliable 73.30% 138,030 A reliable media channel loses 31% of a clean 12 Mbps offer, and its delivery barely moves as loss rises, so loss is not the binding constraint — it cannot carry the rate. An unreliable one holds everything on a clean link and then degrades by about the loss injected. The reverse channel differs by three orders of magnitude. Muninn wanted the unreliable lane and had no way to say so. It signalled the intent the only way the API allowed, by passing no reliable expiry, and its own test has asserted Unreliable and failed since it was written. With this change it passes. Delivery now comes from the advertised profile, resolved by `channel_delivery` alongside the `channel_reliable_expire_after_ms` that already worked this way. That removes the send path's private opinion about reliability, so the profile a peer was shown and the behaviour it gets cannot disagree. The channels whose semantics are fixed by design — schema reliable and ordered, realtime unreliable — are untouched and pinned by a test. `media_delivery: None` means Reliable, so no existing consumer changes behaviour. It is still source-breaking for anyone constructing the three options types with a struct literal rather than `::client(...)`: two sites in cultmesh-rs, four in Muninn, seven across the test suites here, all mechanical. An earlier draft claimed Muninn was unaffected; the compiler disagreed. The `examples/media_delivery_probe.rs` added here is what produced the table, so the claim can be re-run rather than believed. Note its sender-side wire-cost figure is unsound for the reliable case: the sender stops draining after five seconds while retransmission is still in flight, so those byte counts under-report. The delivery percentages are receiver-measured and stand. Co-Authored-By: Claude Opus 5 --- packages/cultmesh-rs/src/lib.rs | 1 + .../cultmesh-rs/tests/rudp_document_server.rs | 1 + .../examples/media_delivery_probe.rs | 177 ++++++++++++++++++ packages/cultnet-rs/src/cultmesh.rs | 2 + packages/cultnet-rs/src/rudp.rs | 43 ++++- packages/cultnet-rs/tests/cultnet.rs | 7 + .../tests/media_channel_delivery.rs | 92 +++++++++ 7 files changed, 321 insertions(+), 2 deletions(-) create mode 100644 packages/cultnet-rs/examples/media_delivery_probe.rs create mode 100644 packages/cultnet-rs/tests/media_channel_delivery.rs diff --git a/packages/cultmesh-rs/src/lib.rs b/packages/cultmesh-rs/src/lib.rs index 7999f18b..135314bb 100644 --- a/packages/cultmesh-rs/src/lib.rs +++ b/packages/cultmesh-rs/src/lib.rs @@ -955,6 +955,7 @@ fn connect_rudp_client( socket.set_read_timeout(Some(poll_interval))?; let mut client = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: runtime_id.to_string(), socket, mode: cultnet_rs::CultNetRudpSocketMode::Client, diff --git a/packages/cultmesh-rs/tests/rudp_document_server.rs b/packages/cultmesh-rs/tests/rudp_document_server.rs index c0f191b4..df14b4ea 100644 --- a/packages/cultmesh-rs/tests/rudp_document_server.rs +++ b/packages/cultmesh-rs/tests/rudp_document_server.rs @@ -163,6 +163,7 @@ fn client(target: SocketAddr, id: u32) -> Result +//! media_delivery_probe send +//! ``` +//! +//! The sender reports amplification: wire bytes divided by offered payload +//! bytes. At 1.0 nothing was retransmitted. Well above 1.0 under loss is the +//! pathology. + +use std::env; +use std::net::{SocketAddr, UdpSocket}; +use std::time::{Duration, Instant}; + +use anyhow::{Result, anyhow}; +use cultnet_rs::{ + CultNetRudpServerHubOptions, CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportOptions, CultNetTransportDelivery, +}; + +const CHANNEL: &str = "media"; +const DRAIN: Duration = Duration::from_secs(5); + +fn main() -> Result<()> { + let args: Vec = env::args().collect(); + match args.get(1).map(String::as_str) { + Some("recv") => recv( + args.get(2).ok_or_else(|| anyhow!("bind addr"))?.parse()?, + args.get(3).ok_or_else(|| anyhow!("expect count"))?.parse()?, + ), + Some("send") => send( + args.get(2).ok_or_else(|| anyhow!("endpoint"))?.parse()?, + match args.get(3).map(String::as_str) { + Some("unreliable") => CultNetTransportDelivery::Unreliable, + Some("reliable") => CultNetTransportDelivery::Reliable, + other => return Err(anyhow!("delivery must be reliable|unreliable, got {other:?}")), + }, + args.get(4).ok_or_else(|| anyhow!("payload bytes"))?.parse()?, + args.get(5).ok_or_else(|| anyhow!("count"))?.parse()?, + args.get(6).map(|v| v.parse()).transpose()?.unwrap_or(12.0), + ), + _ => Err(anyhow!( + "usage: media_delivery_probe recv | send " + )), + } +} + +fn recv(bind: SocketAddr, expect: u64) -> Result<()> { + let socket = UdpSocket::bind(bind)?; + socket.set_nonblocking(true)?; + let mut hub = cultnet_rs::CultNetRudpServerHub::new(CultNetRudpServerHubOptions::new( + "probe-recv", + socket, + 0x0BE0_0001, + ))?; + + println!("listening on {bind}, expecting {expect} frames"); + let started = Instant::now(); + let mut frames = 0u64; + let mut payload_bytes = 0u64; + let mut last_progress = Instant::now(); + + loop { + while let Some(event) = hub.receive_event_once()? { + if let cultnet_rs::CultNetRudpServerEvent::Frame { frame, .. } = event + && frame.channel_id == CHANNEL + { + frames += 1; + payload_bytes += frame.payload.len() as u64; + last_progress = Instant::now(); + } + } + if frames >= expect || last_progress.elapsed() > DRAIN { + break; + } + std::thread::sleep(Duration::from_micros(200)); + } + + let stats = hub.stats(); + let elapsed = started.elapsed().as_secs_f64(); + println!("RESULT recv frames={frames} expected={expect} payload_bytes={payload_bytes}"); + println!( + "RESULT recv wire_bytes_in={} wire_bytes_out={} elapsed_s={elapsed:.2} delivered_pct={:.2}", + stats.bytes_received, + stats.bytes_sent, + (frames as f64 / expect as f64) * 100.0 + ); + Ok(()) +} + +fn send( + endpoint: SocketAddr, + delivery: CultNetTransportDelivery, + payload_bytes: usize, + count: u64, + target_mbps: f64, +) -> Result<()> { + let socket = UdpSocket::bind("0.0.0.0:0")?; + socket.set_nonblocking(true)?; + let mut options = + CultNetRudpSocketTransportOptions::client("probe-send", socket, endpoint, 0x0BE0_0001); + options.media_delivery = Some(delivery); + options.media_reliable_expire_after_ms = None; + + let mut transport = CultNetRudpSocketTransportConnection::new(options)?; + transport.connect(b"probe".to_vec())?; + let deadline = Instant::now() + Duration::from_secs(10); + while !transport.connected() { + let _ = transport.receive_once()?; + transport.poll_resends()?; + if Instant::now() >= deadline { + return Err(anyhow!("timed out connecting to {endpoint}")); + } + std::thread::sleep(Duration::from_millis(2)); + } + + // Pace to a realistic media rate. An unpaced blast overruns the receiver's + // socket buffer and manufactures loss that has nothing to do with the link. + let per_packet = Duration::from_secs_f64((payload_bytes as f64 * 8.0) / (target_mbps * 1e6)); + let payload = vec![0xA5u8; payload_bytes]; + let offered = payload_bytes as u64 * count; + let started = Instant::now(); + let mut sent = 0u64; + let mut would_block = 0u64; + + while sent < count { + match transport.send(CHANNEL, payload.clone()) { + Ok(()) => sent += 1, + Err(error) if error.to_string().contains("would block") => { + would_block += 1; + transport.poll_resends()?; + let _ = transport.receive_once()?; + std::thread::sleep(Duration::from_micros(500)); + } + Err(error) => return Err(error), + } + if sent % 64 == 0 { + let _ = transport.receive_once()?; + transport.poll_resends()?; + } + let due = started + per_packet.mul_f64(sent as f64); + if let Some(wait) = due.checked_duration_since(Instant::now()) { + std::thread::sleep(wait); + } + } + let send_elapsed = started.elapsed(); + + // Let retransmission run: this is where a reliable channel spends what an + // unreliable one does not. + let drain_start = Instant::now(); + while drain_start.elapsed() < DRAIN { + let _ = transport.receive_once()?; + transport.poll_resends()?; + std::thread::sleep(Duration::from_millis(1)); + } + + let stats = transport.stats(); + let amplification = stats.bytes_sent as f64 / offered as f64; + println!( + "RESULT send delivery={delivery:?} payload_bytes={payload_bytes} count={count} offered={offered} target_mbps={target_mbps}" + ); + println!( + "RESULT send wire_bytes={} amplification={amplification:.3} send_s={:.2} total_s={:.2} would_block={would_block} reliable_expired={}", + stats.bytes_sent, + send_elapsed.as_secs_f64(), + started.elapsed().as_secs_f64(), + stats.reliable_packets_expired + ); + Ok(()) +} diff --git a/packages/cultnet-rs/src/cultmesh.rs b/packages/cultnet-rs/src/cultmesh.rs index efc8acad..14998396 100644 --- a/packages/cultnet-rs/src/cultmesh.rs +++ b/packages/cultnet-rs/src/cultmesh.rs @@ -319,6 +319,7 @@ impl CultMesh { let socket = bind_rudp_socket(&options)?; CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { media_reliable_expire_after_ms: None, + media_delivery: None, runtime_id: runtime_id.into(), socket, mode: CultNetRudpSocketMode::Server, @@ -343,6 +344,7 @@ impl CultMesh { let socket = bind_rudp_socket(&options)?; CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { media_reliable_expire_after_ms: None, + media_delivery: None, runtime_id: runtime_id.into(), socket, mode: CultNetRudpSocketMode::Client, diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index f1fd4ca4..b7019b1b 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -1210,6 +1210,8 @@ pub struct CultNetRudpSocketTransportOptions { pub max_pending_reliable_packets: Option, pub reconnect_policy: Option, pub media_reliable_expire_after_ms: Option, + /// `None` means the historical default, `Reliable`. + pub media_delivery: Option, } impl CultNetRudpSocketTransportOptions { @@ -1221,6 +1223,7 @@ impl CultNetRudpSocketTransportOptions { ) -> Self { Self { media_reliable_expire_after_ms: Some(DEFAULT_MEDIA_RELIABLE_EXPIRE_AFTER_MS), + media_delivery: None, runtime_id: runtime_id.into(), socket, mode: CultNetRudpSocketMode::Client, @@ -1239,6 +1242,7 @@ impl CultNetRudpSocketTransportOptions { pub fn server(runtime_id: impl Into, socket: UdpSocket, connection_id: u32) -> Self { Self { media_reliable_expire_after_ms: Some(DEFAULT_MEDIA_RELIABLE_EXPIRE_AFTER_MS), + media_delivery: None, runtime_id: runtime_id.into(), socket, mode: CultNetRudpSocketMode::Server, @@ -1280,12 +1284,15 @@ pub struct CultNetRudpServerHubOptions { pub max_pending_reliable_packets: Option, pub max_peers: usize, pub media_reliable_expire_after_ms: Option, + /// `None` means the historical default, `Reliable`. + pub media_delivery: Option, } impl CultNetRudpServerHubOptions { pub fn new(runtime_id: impl Into, socket: UdpSocket, connection_id: u32) -> Self { Self { media_reliable_expire_after_ms: Some(DEFAULT_MEDIA_RELIABLE_EXPIRE_AFTER_MS), + media_delivery: None, runtime_id: runtime_id.into(), socket, connection_id, @@ -1372,6 +1379,7 @@ impl CultNetRudpServerHub { options.runtime_id, RudpTransportProfileOptions { media_reliable_expire_after_ms: options.media_reliable_expire_after_ms, + media_delivery: options.media_delivery, transport_id: options.transport_id, host: Some(local_addr.ip().to_string()), port: Some(local_addr.port()), @@ -1451,6 +1459,7 @@ impl CultNetRudpServerHub { channel_id, now_ms(), channel_reliable_expire_after_ms(&self.profile, channel_id), + channel_delivery(&self.profile, channel_id), ), self.max_fragment_bytes, )? @@ -1671,6 +1680,7 @@ impl CultNetRudpSocketTransportConnection { options.runtime_id, RudpTransportProfileOptions { media_reliable_expire_after_ms: options.media_reliable_expire_after_ms, + media_delivery: options.media_delivery, transport_id: options.transport_id, host: Some(local_addr.ip().to_string()), port: Some(local_addr.port()), @@ -1738,6 +1748,7 @@ impl CultNetRudpSocketTransportConnection { channel_id, now_ms(), channel_reliable_expire_after_ms(&self.profile, channel_id), + channel_delivery(&self.profile, channel_id), ); let packets = self.session @@ -1759,6 +1770,7 @@ impl CultNetRudpSocketTransportConnection { channel_id, now_ms(), channel_reliable_expire_after_ms(&self.profile, channel_id), + channel_delivery(&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. @@ -2073,6 +2085,8 @@ pub struct RudpTransportProfileOptions { pub max_pending_reliable_packets: Option, pub reconnect_policy: Option, pub media_reliable_expire_after_ms: Option, + /// `None` means the historical default, `Reliable`. + pub media_delivery: Option, } pub fn create_rudp_transport_profile( @@ -2125,7 +2139,9 @@ pub fn create_rudp_transport_profile( }, CultNetTransportChannel { channel_id: "media".to_string(), - delivery: CultNetTransportDelivery::Reliable, + delivery: options + .media_delivery + .unwrap_or(CultNetTransportDelivery::Reliable), ordering: CultNetTransportOrdering::Unordered, max_payload_bytes: options.max_payload_bytes, max_fragment_bytes: options.max_fragment_bytes, @@ -2254,10 +2270,30 @@ fn channel_reliable_expire_after_ms( .and_then(|channel| channel.reliable_expire_after_ms) } +/// The advertised profile owns each channel's delivery, for the same reason it +/// owns expiry: a send path that hardcodes reliability can disagree with the +/// profile the peer was shown, and then no one can say what the channel does. +/// +/// Falls back to `Reliable` for a channel the profile does not describe, which +/// is what every channel got before delivery was configurable. +fn channel_delivery( + profile: &CultNetTransportProfile, + channel_id: &str, +) -> CultNetTransportDelivery { + profile + .transports + .iter() + .flat_map(|transport| transport.channels.iter()) + .find(|channel| channel.channel_id == channel_id) + .map(|channel| channel.delivery) + .unwrap_or(CultNetTransportDelivery::Reliable) +} + fn channel_send_options( channel_id: &str, now_ms: u64, reliable_expire_after_ms: Option, + delivery: CultNetTransportDelivery, ) -> CultNetRudpSendOptions { match channel_id { "schema" => CultNetRudpSendOptions { @@ -2274,8 +2310,11 @@ fn channel_send_options( now_ms, reliable_expire_after_ms: None, }, + // The one channel whose delivery a caller chooses. Media that has gone + // stale is worth dropping, not retransmitting: a reliable media channel + // under loss adds load exactly when the link has least to give. "media" => CultNetRudpSendOptions { - reliable: true, + reliable: matches!(delivery, CultNetTransportDelivery::Reliable), ordered: false, sequenced: false, now_ms, diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 851646cf..7be6ea18 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -359,6 +359,7 @@ fn rudp_transport_profile_advertises_state_and_realtime_channels() { let profile = create_rudp_transport_profile( "rust-rudp", cultnet_rs::RudpTransportProfileOptions { + media_delivery: None, transport_id: Some("public-rudp".to_string()), host: Some("127.0.0.1".to_string()), port: Some(7777), @@ -1248,6 +1249,7 @@ fn rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames() let connection_id = 0x1020_3040; let mut server = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-server".to_string(), socket: server_socket, mode: CultNetRudpSocketMode::Server, @@ -1264,6 +1266,7 @@ fn rudp_socket_transport_handshakes_and_carries_reliable_ordered_schema_frames() })?; let mut client = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-client".to_string(), socket: client_socket, mode: CultNetRudpSocketMode::Client, @@ -1340,6 +1343,7 @@ fn rudp_socket_transport_carries_fragmented_reliable_ordered_schema_frames() -> let connection_id = 0x1020_3041; let mut server = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-fragment-server".to_string(), socket: server_socket, mode: CultNetRudpSocketMode::Server, @@ -1356,6 +1360,7 @@ fn rudp_socket_transport_carries_fragmented_reliable_ordered_schema_frames() -> })?; let mut client = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-fragment-client".to_string(), socket: client_socket, mode: CultNetRudpSocketMode::Client, @@ -1470,6 +1475,7 @@ fn rudp_socket_flush_waits_for_large_fragment_delivery() -> Result<()> { let connection_id = 0x1020_3043; let mut server = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-window-server".into(), socket: server_socket, mode: CultNetRudpSocketMode::Server, @@ -1486,6 +1492,7 @@ fn rudp_socket_flush_waits_for_large_fragment_delivery() -> Result<()> { })?; let mut client = CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions { + media_delivery: None, runtime_id: "rust-rudp-window-client".into(), socket: client_socket, mode: CultNetRudpSocketMode::Client, diff --git a/packages/cultnet-rs/tests/media_channel_delivery.rs b/packages/cultnet-rs/tests/media_channel_delivery.rs new file mode 100644 index 00000000..499be0d3 --- /dev/null +++ b/packages/cultnet-rs/tests/media_channel_delivery.rs @@ -0,0 +1,92 @@ +//! Media delivery is the caller's choice, and the advertised profile is where +//! that choice lives. +//! +//! Before this, `media` was hardcoded `Reliable` in two places — the profile +//! builder and the per-channel send options — so no caller could say otherwise. +//! A producer asking for an unreliable media lane got a reliable one that +//! retransmits under loss, which adds load exactly when the link has least to +//! give. Muninn asked for that lane by passing no expiry, meaning "unreliable, +//! so expiry is moot", and got reliable-with-no-expiry instead: retransmit +//! forever. +//! +//! The default stays `Reliable` so every existing consumer is untouched. + +use std::net::UdpSocket; + +use cultnet_rs::{ + CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, + CultNetTransportDelivery, +}; + +fn media_channel_delivery( + media_delivery: Option, +) -> (CultNetTransportDelivery, Option) { + let socket = UdpSocket::bind("127.0.0.1:0").expect("binds"); + let endpoint = "127.0.0.1:5204".parse().expect("parses"); + let mut options = CultNetRudpSocketTransportOptions::client("test-media", socket, endpoint, 7); + options.media_delivery = media_delivery; + options.media_reliable_expire_after_ms = None; + + let transport = CultNetRudpSocketTransportConnection::new(options).expect("opens"); + let channel = transport + .profile + .transports + .first() + .expect("a transport") + .channels + .iter() + .find(|channel| channel.channel_id == "media") + .expect("a media channel"); + (channel.delivery, channel.reliable_expire_after_ms) +} + +#[test] +fn media_defaults_to_reliable_so_existing_consumers_do_not_move() { + let (delivery, _) = media_channel_delivery(None); + assert_eq!(delivery, CultNetTransportDelivery::Reliable); +} + +#[test] +fn a_caller_can_ask_for_an_unreliable_media_lane() { + let (delivery, expiry) = media_channel_delivery(Some(CultNetTransportDelivery::Unreliable)); + assert_eq!(delivery, CultNetTransportDelivery::Unreliable); + assert_eq!( + expiry, None, + "an unreliable channel has nothing to expire; the pairing should stay honest" + ); +} + +#[test] +fn asking_for_reliable_explicitly_matches_the_default() { + let (explicit, _) = media_channel_delivery(Some(CultNetTransportDelivery::Reliable)); + let (defaulted, _) = media_channel_delivery(None); + assert_eq!(explicit, defaulted); +} + +/// The channels whose semantics are fixed by design must not drift with this. +#[test] +fn the_other_channels_keep_their_fixed_delivery() { + let socket = UdpSocket::bind("127.0.0.1:0").expect("binds"); + let endpoint = "127.0.0.1:5204".parse().expect("parses"); + let mut options = CultNetRudpSocketTransportOptions::client("test-media", socket, endpoint, 8); + options.media_delivery = Some(CultNetTransportDelivery::Unreliable); + + let transport = CultNetRudpSocketTransportConnection::new(options).expect("opens"); + let channels = &transport + .profile + .transports + .first() + .expect("a transport") + .channels; + + let delivery_of = |id: &str| { + channels + .iter() + .find(|channel| channel.channel_id == id) + .unwrap_or_else(|| panic!("channel {id}")) + .delivery + }; + + assert_eq!(delivery_of("schema"), CultNetTransportDelivery::Reliable); + assert_eq!(delivery_of("realtime"), CultNetTransportDelivery::Unreliable); +}