diff --git a/packages/cultnet-rs/src/media_fec.rs b/packages/cultnet-rs/src/media_fec.rs index 88a3b811..490eaaed 100644 --- a/packages/cultnet-rs/src/media_fec.rs +++ b/packages/cultnet-rs/src/media_fec.rs @@ -95,7 +95,7 @@ use crate::media_stream_contracts::{ GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord, }; use crate::media_stream_wire::{ - GAMECULT_MEDIA_CHANNEL, GameCultMediaWireRecord, MediaWireProvenance, + GAMECULT_MEDIA_CHANNEL, GAMECULT_MEDIA_MAX_WIRE_BYTES, GameCultMediaWireRecord, MediaWireProvenance, encode_media_wire_record, validate_audio_parity_record, validate_audio_record, validate_video_parity_record, validate_video_record, }; @@ -118,6 +118,13 @@ pub const MEDIA_FEC_MAX_WIRE_BYTES: usize = /// A producer's choice of `(k, m)` and datagram budget. /// [`MediaFecPolicy::STANDARD`] is the ruled default; each record carries the /// `(k, m)` it was made with, so a decoder needs no policy. +/// +/// Build one by overriding fields of the default, `MediaFecPolicy { +/// max_wire_bytes: 1_211, ..MediaFecPolicy::STANDARD }`, not by naming every +/// field: a field added later then cannot break, or silently mis-set, the +/// caller. `max_wire_bytes` may not exceed +/// [`GAMECULT_MEDIA_MAX_WIRE_BYTES`], the ceiling a decoder holds every record +/// to. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct MediaFecPolicy { /// The most data chunks in one video block. @@ -183,6 +190,12 @@ impl MediaFecPolicy { { return Err(MediaFecError::invalid("media FEC policy exceeds 256 shards per block")); } + if self.max_wire_bytes > GAMECULT_MEDIA_MAX_WIRE_BYTES { + return Err(MediaFecError::invalid(format!( + "media FEC policy max_wire_bytes {} exceeds the {GAMECULT_MEDIA_MAX_WIRE_BYTES}-byte media record ceiling", + self.max_wire_bytes + ))); + } Ok(()) } } diff --git a/packages/cultnet-rs/src/media_stream_wire.rs b/packages/cultnet-rs/src/media_stream_wire.rs index 5c401571..78915a75 100644 --- a/packages/cultnet-rs/src/media_stream_wire.rs +++ b/packages/cultnet-rs/src/media_stream_wire.rs @@ -42,6 +42,19 @@ use crate::media_stream_contracts::{ /// choice per session; stale media is dropped, not retransmitted. pub const GAMECULT_MEDIA_CHANNEL: &str = "media"; +/// The largest wrapped media record the decoder accepts, and the most the RUDP +/// `media` channel reassembles: 64 KiB. +/// +/// It is the bound a receiver holds a peer to. A parity shard's length is +/// whatever its producer's [`MediaFecPolicy::max_wire_bytes`](crate::MediaFecPolicy::max_wire_bytes) allowed, and +/// recovery work grows with the square of the block size times that length, so +/// without a ceiling one hostile record could name any length RUDP would +/// reassemble (16 MiB). 64 KiB is one maximum UDP datagram: it covers every +/// path MTU in use, jumbo frames (9,000 bytes) with a wide margin, and a +/// record is still one message rather than a stream. A policy may not ask for +/// more; see [`MediaFecPolicy`](crate::MediaFecPolicy). +pub const GAMECULT_MEDIA_MAX_WIRE_BYTES: usize = 64 * 1024; + /// One media record on its way somewhere. #[derive(Clone, Debug, PartialEq, Eq)] pub enum GameCultMediaWireRecord { @@ -174,6 +187,12 @@ pub fn encode_media_wire_record( /// Unwraps a media record from the wire. pub fn decode_media_wire_record(payload: &[u8]) -> Result { + if payload.len() > GAMECULT_MEDIA_MAX_WIRE_BYTES { + return Err(anyhow!( + "media record is {} bytes, over the {GAMECULT_MEDIA_MAX_WIRE_BYTES}-byte ceiling", + payload.len() + )); + } let message = decode_cultnet_message_from_slice(payload, CultNetWireContract::CultNetSchemaV0)?; let CultNetMessage::DocumentPutRaw { document, .. } = message else { return Err(anyhow!("expected cultnet.document_put_raw.v0")); @@ -434,11 +453,6 @@ pub fn validate_audio_parity_record(record: &GameCultMediaAudioParityShardRecord "audio parity media record timing metadata must be non-zero" )); } - if record.deadline_ticks < record.base_pts_ticks { - return Err(anyhow!( - "audio parity media record deadline_ticks must not precede base_pts_ticks" - )); - } if record.fec_scheme != MEDIA_FEC_SCHEME_RS_GF256_V1 { return Err(anyhow!( "audio parity media record fec_scheme {:?} is not supported", @@ -452,17 +466,23 @@ pub fn validate_audio_parity_record(record: &GameCultMediaAudioParityShardRecord { return Err(anyhow!("audio parity media record stripe metadata is invalid")); } - if audio_slot( + let Some((_, last_pts_ticks)) = audio_slot( record.base_packet_id, record.base_pts_ticks, record.packet_duration_ticks, u64::from(record.data_shard_count) - 1, - ) - .is_none() - { + ) else { return Err(anyhow!( "audio parity media record block runs past the packet id or pts range" )); + }; + // Recovery gives every recovered packet the block's deadline, so it must + // not precede the last packet's pts or a recovered packet would fail + // `validate_audio_record`. + if record.deadline_ticks < last_pts_ticks { + return Err(anyhow!( + "audio parity media record deadline_ticks must not precede the block's last pts_ticks" + )); } if record.shard_payload_bytes == 0 { return Err(anyhow!( diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index d1afca7c..d5cc512b 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -23,6 +23,8 @@ use crate::CultNetTransportProfile; use crate::CultNetTransportProtocol; use crate::CultNetTransportStats; use crate::CultNetWireContract; +use crate::GAMECULT_MEDIA_CHANNEL; +use crate::GAMECULT_MEDIA_MAX_WIRE_BYTES; use crate::create_reconnect_policy; use crate::decode_cultnet_message_from_slice; use crate::encode_cultnet_message_to_vec; @@ -429,7 +431,7 @@ impl CultNetRudpSession { if options.ordered && !options.reliable { return Err(anyhow!("RUDP ordered delivery requires reliability")); } - self.require_payload_size(payload.len())?; + self.require_payload_size(channel_id, payload.len())?; // Reclaim deadline-passed sends before measuring window capacity, so an // expiring stream cannot be refused for a backlog it no longer owns. self.purge_expired_reliable(options.now_ms); @@ -958,7 +960,7 @@ impl CultNetRudpSession { packet: &CultNetRudpPacket, ) -> Result> { if packet.fragment_count == 0 { - self.require_payload_size(packet.payload.len())?; + self.require_payload_size(&packet.channel_id, packet.payload.len())?; return Ok(Some(( CultNetRudpDeliveredFrame { channel_id: packet.channel_id.clone(), @@ -1010,7 +1012,7 @@ impl CultNetRudpSession { .sum::() }) .unwrap_or(0); - if let Err(error) = self.require_payload_size(buffered_bytes + packet.payload.len()) { + if let Err(error) = self.require_payload_size(&packet.channel_id, buffered_bytes + packet.payload.len()) { self.fragment_buffers.remove(&key); return Err(error); } @@ -1068,11 +1070,12 @@ impl CultNetRudpSession { ))) } - fn require_payload_size(&self, payload_bytes: usize) -> Result<()> { - if self - .max_payload_bytes - .is_some_and(|max_payload_bytes| payload_bytes > max_payload_bytes) - { + /// The session's cap, tightened on the media channel to the ceiling every + /// media record is held to (`GAMECULT_MEDIA_MAX_WIRE_BYTES`), so media + /// reassembly never buffers more than a record may be. + fn require_payload_size(&self, channel_id: &str, payload_bytes: usize) -> Result<()> { + let limit = channel_payload_limit(channel_id, self.max_payload_bytes); + if limit.is_some_and(|limit| payload_bytes > limit) { return Err(anyhow!("RUDP payload exceeds max_payload_bytes")); } Ok(()) @@ -2156,6 +2159,18 @@ pub struct RudpTransportProfileOptions { pub media_delivery: Option, } +/// The largest payload a channel accepts: the session-wide cap, tightened on +/// the media channel to the media ceiling. +fn channel_payload_limit(channel_id: &str, session_limit: Option) -> Option { + if channel_id == GAMECULT_MEDIA_CHANNEL { + Some(session_limit.map_or(GAMECULT_MEDIA_MAX_WIRE_BYTES, |limit| { + limit.min(GAMECULT_MEDIA_MAX_WIRE_BYTES) + })) + } else { + session_limit + } +} + pub fn create_rudp_transport_profile( runtime_id: impl Into, options: RudpTransportProfileOptions, @@ -2205,12 +2220,16 @@ pub fn create_rudp_transport_profile( reliable_expire_after_ms: None, }, CultNetTransportChannel { - channel_id: "media".to_string(), + channel_id: GAMECULT_MEDIA_CHANNEL.to_string(), delivery: options .media_delivery .unwrap_or(CultNetTransportDelivery::Reliable), ordering: CultNetTransportOrdering::Unordered, - max_payload_bytes: options.max_payload_bytes, + max_payload_bytes: channel_payload_limit( + GAMECULT_MEDIA_CHANNEL, + options.max_payload_bytes.map(|limit| limit as usize), + ) + .map(|limit| limit as u32), 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, diff --git a/packages/cultnet-rs/tests/media_fec.rs b/packages/cultnet-rs/tests/media_fec.rs index 661fe449..4af9f243 100644 --- a/packages/cultnet-rs/tests/media_fec.rs +++ b/packages/cultnet-rs/tests/media_fec.rs @@ -8,7 +8,7 @@ use cultnet_rs::{ GameCultMediaAudioPacketRecord, GameCultMediaAudioParityShardRecord, GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord, - GameCultMediaWireRecord, MEDIA_FEC_MAX_WIRE_BYTES, + GAMECULT_MEDIA_MAX_WIRE_BYTES, GameCultMediaWireRecord, MEDIA_FEC_MAX_WIRE_BYTES, MEDIA_FEC_SCHEME_RS_GF256_V1, MediaFecError, MediaFecPolicy, MediaWireProvenance, decode_media_wire_record, encode_media_wire_record, protect_audio_block, protect_video_frame, recover_audio_block, recover_video_block, @@ -1052,3 +1052,92 @@ fn the_worst_burst_that_always_recovers_is_blocks_times_the_smallest_m() { ); } } + +// --------------------------------------------------------------------------- +// The ceiling, and what the audio budget check relies on +// --------------------------------------------------------------------------- + +/// Only the parity is measured against the budget, on the claim that it is the +/// largest record of an audio block. It carries every field a packet does plus +/// the stripe geometry, so this holds across the id, pts, duration and deadline +/// widths that vary the encoding. +#[test] +fn an_audio_parity_record_is_larger_on_the_wire_than_every_packet_it_protects() { + let bases: [u64; 8] = [0, 0x7c, 0xfc, 0xfffc, 0xffff_fffc, 99_997, u64::MAX - 3, 1]; + let starts: [i64; 7] = [i64::MIN, -1, 0, 125, 65_530, 4_294_967_290, i64::MAX - 3 * 4_000_000_000]; + let durations: [u32; 5] = [1, 120, 960, 65_535, 4_000_000_000]; + for base in bases { + for start in starts { + for duration in durations { + for wide_deadline in [false, true] { + let mut packets = audio_block(base, 40); + for (index, packet) in packets.iter_mut().enumerate() { + packet.pts_ticks = start + i64::from(duration) * index as i64; + packet.duration_ticks = duration; + packet.deadline_ticks = if wide_deadline { i64::MAX } else { packet.pts_ticks }; + } + let parity = protect_audio_block(&packets, &STANDARD, provenance()).unwrap(); + let parity_len = wire_len(&GameCultMediaWireRecord::AudioParity(parity[0].clone()), provenance()); + for packet in &packets { + let packet_len = wire_len(&GameCultMediaWireRecord::Audio(packet.clone()), provenance()); + assert!( + packet_len < parity_len, + "base {base} start {start} duration {duration}: packet {packet_len} >= parity {parity_len}" + ); + } + } + } + } + } +} + +/// A parity whose deadline precedes its last packet's pts would recover a packet +/// that fails `validate_audio_record`. Recovery refuses it, and so does the wire. +#[test] +fn audio_parity_with_a_deadline_before_its_last_packet_recovers_nothing() { + let packets = audio_block(10, 8); + let mut parity = protect_audio_block(&packets, &STANDARD, provenance()).unwrap(); + for shard in &mut parity { + shard.deadline_ticks = shard.base_pts_ticks; + } + let error = recover_audio_block(&parity, &packets[..2]).unwrap_err(); + assert!(is_invalid_mentioning(&error, "last pts_ticks"), "{error}"); + let wire = encode_media_wire_record( + &GameCultMediaWireRecord::AudioParity(parity[0].clone()), + provenance(), + ) + .unwrap(); + assert!(decode_media_wire_record(&wire).is_err()); +} + +/// The datagram budget is the producer's, but no budget may promise a record the +/// decoder would refuse. +#[test] +fn a_datagram_budget_over_the_media_record_ceiling_is_refused() { + let over = MediaFecPolicy { max_wire_bytes: GAMECULT_MEDIA_MAX_WIRE_BYTES + 1, ..STANDARD }; + let error = protect_audio_block(&audio_block(1, 100), &over, provenance()).unwrap_err(); + assert!(is_invalid_mentioning(&error, "ceiling"), "{error}"); + let error = protect_video_frame(&frame(2, 3, 100, 100), &over, provenance()).unwrap_err(); + assert!(is_invalid_mentioning(&error, "ceiling"), "{error}"); +} + +/// A budget at the ceiling admits shards up to it, and the decoder accepts +/// what the policy admitted. +#[test] +fn a_budget_at_the_media_record_ceiling_admits_large_shards_end_to_end() { + let at_ceiling = MediaFecPolicy { max_wire_bytes: GAMECULT_MEDIA_MAX_WIRE_BYTES, ..STANDARD }; + let records = + protect_video_frame(&frame(2, 3, 60_000, 60_000), &at_ceiling, provenance()).unwrap(); + for record in &records { + let wire = encode_media_wire_record(record, provenance()).unwrap(); + assert!(wire.len() <= GAMECULT_MEDIA_MAX_WIRE_BYTES); + decode_media_wire_record(&wire).unwrap(); + } + let parity = protect_audio_block(&audio_block(1, 60_000), &at_ceiling, provenance()).unwrap(); + let wire = encode_media_wire_record( + &GameCultMediaWireRecord::AudioParity(parity[0].clone()), + provenance(), + ) + .unwrap(); + decode_media_wire_record(&wire).unwrap(); +} diff --git a/packages/cultnet-rs/tests/media_stream_wire.rs b/packages/cultnet-rs/tests/media_stream_wire.rs index 762fa78d..530123f7 100644 --- a/packages/cultnet-rs/tests/media_stream_wire.rs +++ b/packages/cultnet-rs/tests/media_stream_wire.rs @@ -2,6 +2,8 @@ //! nothing on the wire. use cultnet_rs::{ + CultNetRudpDeliveredFrame, CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions, + GAMECULT_MEDIA_MAX_WIRE_BYTES, RudpTransportProfileOptions, create_rudp_transport_profile, GAMECULT_MEDIA_CHANNEL, GameCultMediaAudioPacketRecord, GameCultMediaAudioParityShardRecord, GameCultMediaReceiverFeedbackRecord, GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord, @@ -420,3 +422,220 @@ fn records_on_the_edge_of_each_bound_are_admitted() { audio.deadline_ticks = i64::MAX; assert!(decodes(GameCultMediaWireRecord::AudioParity(audio)).is_ok(), "the last pts is i64::MAX"); } + +// --------------------------------------------------------------------------- +// The ceiling on one media record +// --------------------------------------------------------------------------- + +fn audio_parity(payload_bytes: usize, deadline_ticks: i64) -> GameCultMediaAudioParityShardRecord { + GameCultMediaAudioParityShardRecord { + stream_id: "s".to_string(), + session_id: "x".to_string(), + codec: "opus".to_string(), + fec_scheme: "rs-gf256-v1".to_string(), + base_packet_id: 8, + base_pts_ticks: 0, + packet_duration_ticks: 100, + timebase_num: 1, + timebase_den: 48_000, + deadline_ticks, + data_shard_count: 4, + parity_index: 0, + parity_shard_count: 2, + shard_payload_bytes: payload_bytes as u32, + payload: vec![7; payload_bytes], + } +} + +fn video_parity(payload_bytes: usize) -> GameCultMediaVideoParityShardRecord { + GameCultMediaVideoParityShardRecord { + stream_id: "s".to_string(), + session_id: "x".to_string(), + frame_id: 1, + codec: "h264".to_string(), + pts_ticks: 0, + duration_ticks: 1, + timebase_num: 1, + timebase_den: 90_000, + keyframe: false, + dependency_frame_id: Some(0), + deadline_ticks: 1, + chunk_count: 4, + fec_scheme: "rs-gf256-v1".to_string(), + block_index: 0, + block_count: 1, + block_data_start: 0, + block_data_count: 4, + parity_index: 1, + parity_count: 2, + shard_payload_bytes: payload_bytes as u32, + last_chunk_payload_bytes: 2, + payload: vec![7; payload_bytes], + } +} + +fn wire_len(record: &GameCultMediaWireRecord) -> usize { + encode_media_wire_record(record, provenance()).unwrap().len() +} + +/// The payload length whose wrapped record is exactly `wire_bytes` long, found +/// from the record's own linear growth in its payload. +fn payload_for_wire_len( + wire_bytes: usize, + wrap: &dyn Fn(usize) -> GameCultMediaWireRecord, +) -> usize { + let probe = 1_000; + let payload = probe + wire_bytes - wire_len(&wrap(probe)); + assert_eq!(wire_len(&wrap(payload)), wire_bytes, "the record grows one byte per payload byte here"); + payload +} + +/// RUDP would reassemble a 16 MiB record, and recovery work scales with the +/// shard length, so the decoder holds every record to one ceiling. +#[test] +fn a_media_record_one_byte_over_the_ceiling_is_refused_at_decode() { + let ceiling = GAMECULT_MEDIA_MAX_WIRE_BYTES; + let wraps: [(&str, Box GameCultMediaWireRecord>); 2] = [ + ( + "audio parity", + Box::new(|payload| GameCultMediaWireRecord::AudioParity(audio_parity(payload, 300))), + ), + ( + "video parity", + Box::new(|payload| GameCultMediaWireRecord::VideoParity(video_parity(payload))), + ), + ]; + for (name, wrap) in &wraps { + let at = payload_for_wire_len(ceiling, wrap.as_ref()); + let wire = encode_media_wire_record(&wrap(at), provenance()).unwrap(); + assert_eq!(wire.len(), ceiling); + decode_media_wire_record(&wire) + .unwrap_or_else(|error| panic!("{name} at the ceiling is refused: {error}")); + + let over = encode_media_wire_record(&wrap(at + 1), provenance()).unwrap(); + assert_eq!(over.len(), ceiling + 1); + let error = decode_media_wire_record(&over).unwrap_err(); + assert!(error.to_string().contains("ceiling"), "{name}: {error}"); + } +} + +#[test] +fn a_jumbo_frame_sized_record_is_accepted() { + let wire = encode_media_wire_record( + &GameCultMediaWireRecord::AudioParity(audio_parity(8_800, 300)), + provenance(), + ) + .unwrap(); + assert!(wire.len() > 8_800 && wire.len() < GAMECULT_MEDIA_MAX_WIRE_BYTES); + decode_media_wire_record(&wire).expect("a jumbo-frame record decodes"); +} + +/// Recovery hands every recovered packet the block's deadline, so a parity +/// whose deadline precedes the last packet's pts would recover a packet that +/// fails `validate_audio_record`. +#[test] +fn audio_parity_must_not_have_a_deadline_before_its_last_packets_pts() { + // Four packets, 100 ticks each from 0: the last plays at 300. + let decodes = |deadline_ticks: i64| { + let wire = encode_media_wire_record( + &GameCultMediaWireRecord::AudioParity(audio_parity(8, deadline_ticks)), + provenance(), + ) + .unwrap(); + decode_media_wire_record(&wire) + }; + assert!(decodes(300).is_ok()); + assert!(decodes(301).is_ok()); + let error = decodes(299).unwrap_err(); + assert!(error.to_string().contains("last pts_ticks"), "{error}"); + let error = decodes(0).unwrap_err(); + assert!(error.to_string().contains("last pts_ticks"), "{error}"); +} + +// --------------------------------------------------------------------------- +// The media channel's own payload cap +// --------------------------------------------------------------------------- + +fn connected_session() -> CultNetRudpSession { + let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions { + connection_id: 9, + ..CultNetRudpSessionOptions::default() + }); + session.assume_connected(0); + session +} + +const FRAGMENT_BYTES: usize = 16 * 1024; + +fn media_payload_limit(session_limit: Option) -> Option { + let profile = create_rudp_transport_profile( + "raven", + RudpTransportProfileOptions { + max_payload_bytes: session_limit, + ..RudpTransportProfileOptions::default() + }, + ); + profile.transports[0] + .channels + .iter() + .find(|channel| channel.channel_id == GAMECULT_MEDIA_CHANNEL) + .expect("the profile advertises the media channel") + .max_payload_bytes +} + +#[test] +fn the_media_channel_advertises_the_ceiling_or_a_tighter_session_cap() { + let ceiling = GAMECULT_MEDIA_MAX_WIRE_BYTES as u32; + assert_eq!(media_payload_limit(None), Some(ceiling)); + assert_eq!(media_payload_limit(Some(16 * 1024 * 1024)), Some(ceiling)); + assert_eq!(media_payload_limit(Some(4_096)), Some(4_096)); +} + +#[test] +fn a_session_refuses_to_send_more_than_the_ceiling_on_the_media_channel_only() { + let ceiling = GAMECULT_MEDIA_MAX_WIRE_BYTES; + let send = |channel: &str, payload: usize| { + connected_session().send_many( + channel, + vec![0; payload], + CultNetRudpSendOptions::default(), + Some(FRAGMENT_BYTES), + ) + }; + assert!(send(GAMECULT_MEDIA_CHANNEL, ceiling).is_ok()); + let error = send(GAMECULT_MEDIA_CHANNEL, ceiling + 1).unwrap_err(); + assert!(error.to_string().contains("max_payload_bytes"), "{error}"); + assert!(send("schema", ceiling + 1).is_ok(), "other channels keep the session cap"); +} + +/// Reassembly is where memory is spent, so the receiver enforces the ceiling +/// itself, not only the sender. +#[test] +fn a_session_refuses_to_reassemble_more_than_the_ceiling_on_the_media_channel() { + let ceiling = GAMECULT_MEDIA_MAX_WIRE_BYTES; + let packets = connected_session() + .send_many( + "schema", + vec![0; ceiling + 1], + CultNetRudpSendOptions::default(), + Some(FRAGMENT_BYTES), + ) + .unwrap(); + assert!(packets.len() > 1); + + let deliver = |channel: &str| -> anyhow::Result> { + let mut receiver = connected_session(); + let mut delivered = Vec::new(); + for packet in &packets { + let mut packet = packet.clone(); + packet.channel_id = channel.to_string(); + delivered.extend(receiver.receive(&packet, 0)?.delivered); + } + Ok(delivered) + }; + let frames = deliver("schema").expect("the same fragments reassemble on another channel"); + assert_eq!(frames.len(), 1); + assert_eq!(frames[0].payload.len(), ceiling + 1); + let error = deliver(GAMECULT_MEDIA_CHANNEL).unwrap_err(); + assert!(error.to_string().contains("max_payload_bytes"), "{error}"); +}