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

Bring the media delivery choice alongside the contract

Browse source 
Both branches are the same body of work seen from two angles: what a media
stream is, and what its channel does. Muninn needs both to build, so they merge
here rather than making a consumer track two refs.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-10 02:07:13 +02:00
commit a58968e7ac
7 changed files with 321 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/cultmesh-rs/src/lib.rs
View file

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

1
packages/cultmesh-rs/tests/rudp_document_server.rs
View file

@ -163,6 +163,7 @@ fn client(target: SocketAddr, id: u32) -> Result<CultNetRudpSocketTransportConne
let socket = UdpSocket::bind("127.0.0.1:0")?;
socket.set_nonblocking(true)?;
CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions {
media_delivery: None,
runtime_id: "test-client".into(),
socket,
mode: CultNetRudpSocketMode::Client,

177
packages/cultnet-rs/examples/media_delivery_probe.rs Normal file
View file

@ -0,0 +1,177 @@
//! Measures what a media channel actually costs on the wire under loss.
//!
//! The claim under test: a reliable media channel retransmits lost packets, so
//! it adds load exactly when the link has least to give, while an unreliable one
//! drops them and holds its rate. Run both halves through `cultnet-impair` with
//! a `loss_basis_points` profile and compare bytes on the wire against bytes of
//! payload offered.
//!
//! ```text
//! media_delivery_probe recv <bind_addr> <expect_count>
//! media_delivery_probe send <endpoint> <reliable|unreliable> <payload_bytes> <count>
//! ```
//!
//! 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<String> = 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 <bind> <count> | send <endpoint> <reliable|unreliable> <bytes> <count>"
)),
}
}
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(())
}

2
packages/cultnet-rs/src/cultmesh.rs
View file

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

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

@ -1210,6 +1210,8 @@ pub struct CultNetRudpSocketTransportOptions {
pub max_pending_reliable_packets: Option<u32>,
pub reconnect_policy: Option<CultNetReconnectPolicy>,
pub media_reliable_expire_after_ms: Option<u64>,
/// `None` means the historical default, `Reliable`.
pub media_delivery: Option<CultNetTransportDelivery>,
}
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<String>, 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<u32>,
pub max_peers: usize,
pub media_reliable_expire_after_ms: Option<u64>,
/// `None` means the historical default, `Reliable`.
pub media_delivery: Option<CultNetTransportDelivery>,
}
impl CultNetRudpServerHubOptions {
pub fn new(runtime_id: impl Into<String>, 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<u32>,
pub reconnect_policy: Option<CultNetReconnectPolicy>,
pub media_reliable_expire_after_ms: Option<u64>,
/// `None` means the historical default, `Reliable`.
pub media_delivery: Option<CultNetTransportDelivery>,
}
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<u64>,
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,

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

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

92
packages/cultnet-rs/tests/media_channel_delivery.rs Normal file
View file

@ -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>,
) -> (CultNetTransportDelivery, Option<u64>) {
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);
}