diff --git a/packages/cultnet-rs/Cargo.toml b/packages/cultnet-rs/Cargo.toml index f289d86e..d4585be2 100644 --- a/packages/cultnet-rs/Cargo.toml +++ b/packages/cultnet-rs/Cargo.toml @@ -19,6 +19,7 @@ cultcache-rs = { path = "../cultcache-rs" } ed25519-dalek = "2.2" hmac = "0.12" rand = "0.9" +reed-solomon-erasure = "6" rmp-serde = "1.3" rmpv = { version = "1.3", features = ["with-serde"] } serde = { version = "1", features = ["derive"] } diff --git a/packages/cultnet-rs/src/lib.rs b/packages/cultnet-rs/src/lib.rs index 5bfd1857..621150ba 100644 --- a/packages/cultnet-rs/src/lib.rs +++ b/packages/cultnet-rs/src/lib.rs @@ -2,6 +2,7 @@ mod content; mod contracts; mod idunn_deployment_brake; mod idunn_lifecycle_brake; +mod media_fec; mod media_stream_contracts; mod media_stream_wire; mod provider_health_contracts; @@ -24,6 +25,7 @@ pub use content::*; pub use contracts::*; pub use idunn_deployment_brake::*; pub use idunn_lifecycle_brake::*; +pub use media_fec::*; pub use media_stream_contracts::*; pub use media_stream_wire::*; pub use provider_health_contracts::*; diff --git a/packages/cultnet-rs/src/media_fec.rs b/packages/cultnet-rs/src/media_fec.rs new file mode 100644 index 00000000..29f35e8c --- /dev/null +++ b/packages/cultnet-rs/src/media_fec.rs @@ -0,0 +1,655 @@ +//! Forward error correction for CultMesh Media Stream records. +//! +//! Pure functions over the records in [`crate::media_stream_contracts`]: no +//! sockets, no clock, no randomness. A producer calls [`protect_video_frame`] +//! and [`protect_audio_block`]; a consumer calls [`recover_video_block`] and +//! [`recover_audio_block`]. Both ends link this one implementation, so a +//! producer and a consumer cannot disagree about the math without one of them +//! failing to build against the same version of this crate. +//! +//! # The code +//! +//! Every block is a systematic Reed-Solomon code over GF(2^8): `k` data shards +//! are sent unchanged and `m` parity shards are computed from them. Any `k` of +//! the `k + m` shards recover the block (the code is MDS), so any `m` erasures +//! are repaired and any more are reported as [`MediaFecError::BeyondRepair`], +//! never as wrong bytes. The construction is named on every parity record by +//! [`MEDIA_FEC_SCHEME_RS_GF256_V1`] and backed by the `reed-solomon-erasure` +//! crate. The library is not the contract: `tests/media_fec.rs` pins the parity +//! bytes of a known input, so a library that changes the matrix fails there +//! instead of on the wire, and a decoder in another runtime has a vector to +//! meet. +//! +//! # What a block is +//! +//! * Video: a frame's chunks split into `ceil(n / 16)` contiguous, near-equal +//! blocks, each protected by `max(2, ceil(k / 4))` parity shards. A block +//! never spans two frames, so one frame's deadline governs its recovery. +//! [`protect_video_frame`] returns the frame's records in send order: blocks +//! are interleaved, so a burst of consecutive losses lands as one loss per +//! block rather than many in one. +//! * Audio: `4` consecutive packets of one payload length, protected by `2` +//! parity shards. +//! +//! Every shard, data or parity, fits one datagram once wrapped for the wire +//! ([`MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES`]), so parity never reintroduces the +//! fragment amplification it exists to remove. +//! +//! # What this module does not own +//! +//! When to give up on a block is the receiver's deadline, and the `(k, m)` +//! choice is the producer's policy ([`MediaFecPolicy`]); this module only makes +//! and consumes shards. + +use std::collections::BTreeMap; +use std::fmt; + +use reed_solomon_erasure::galois_8::ReedSolomon; + +use crate::media_stream_contracts::{ + GameCultMediaAudioPacketRecord, GameCultMediaAudioParityShardRecord, + GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord, +}; +use crate::media_stream_wire::{ + GAMECULT_MEDIA_CHANNEL, GameCultMediaWireRecord, validate_audio_parity_record, + validate_audio_record, validate_video_parity_record, validate_video_record, +}; +use crate::rudp::RUDP_FIXED_HEADER_BYTES; + +/// Names the construction on every parity record: systematic Reed-Solomon over +/// GF(2^8), as built by `reed-solomon-erasure`'s `galois_8` field. A different +/// matrix is a different scheme id. +pub const MEDIA_FEC_SCHEME_RS_GF256_V1: &str = "rs-gf256-v1"; + +/// The largest UDP payload an IPv4 datagram carries on a 1500-byte MTU. +pub const MEDIA_IPV4_UDP_PAYLOAD_BYTES: usize = 1_472; + +/// The most a wrapped media record may occupy so that the RUDP header and the +/// channel name still leave it in one datagram (1,431 bytes). +pub const MEDIA_FEC_MAX_WIRE_BYTES: usize = + MEDIA_IPV4_UDP_PAYLOAD_BYTES - RUDP_FIXED_HEADER_BYTES - GAMECULT_MEDIA_CHANNEL.len(); + +/// The largest shard payload the codec produces or accepts. What is left of +/// [`MEDIA_FEC_MAX_WIRE_BYTES`] must hold the record's fields and its envelope, +/// which repeats the stream and session identifiers three times over: a parity +/// record wraps to 705 bytes around its payload when the stream id, session id +/// and codec name are 32, 32 and 16 bytes and the frame id has 20 digits. This is the largest multiple of 16 +/// that fits with identifiers of that size; the test +/// `every_record_at_the_maximum_shard_size_fits_one_datagram` measures it. A +/// producer with longer identifiers must cut its chunks smaller. +pub const MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES: usize = 720; + +/// A producer's choice of `(k, m)`. [`MediaFecPolicy::STANDARD`] is the ruled +/// default; each record carries the `(k, m)` it was made with, so a decoder +/// needs no policy. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct MediaFecPolicy { + /// The most data chunks in one video block. + pub video_max_block_data_shards: u16, + /// The fewest parity shards a video block carries. + pub video_min_parity_shards: u16, + /// A video block of `k` data shards carries at least `ceil(k / this)` parity + /// shards. + pub video_parity_divisor: u16, + /// Packets per audio block. + pub audio_data_shards: u16, + /// Parity shards per audio block. + pub audio_parity_shards: u16, +} + +impl MediaFecPolicy { + /// Video `k <= 16` with `m = max(2, ceil(k / 4))`; audio 4+2. + pub const STANDARD: Self = Self { + video_max_block_data_shards: 16, + video_min_parity_shards: 2, + video_parity_divisor: 4, + audio_data_shards: 4, + audio_parity_shards: 2, + }; + + /// `m` for a video block of `data_shards` chunks. + pub fn video_parity_shards(&self, data_shards: u16) -> u16 { + data_shards + .div_ceil(self.video_parity_divisor.max(1)) + .max(self.video_min_parity_shards) + } + + /// How a frame of `chunk_count` chunks splits: `ceil(n / max)` blocks whose + /// sizes differ by at most one, larger blocks first. + pub fn video_block_sizes(&self, chunk_count: u16) -> Vec { + let max = self.video_max_block_data_shards.max(1); + let blocks = chunk_count.div_ceil(max); + if blocks == 0 { + return Vec::new(); + } + let base = chunk_count / blocks; + let extra = chunk_count % blocks; + (0..blocks).map(|index| base + u16::from(index < extra)).collect() + } + + fn validate(&self) -> Result<(), MediaFecError> { + let widest = self.video_max_block_data_shards; + if widest == 0 + || self.video_min_parity_shards == 0 + || self.video_parity_divisor == 0 + || self.audio_data_shards == 0 + || self.audio_parity_shards == 0 + { + return Err(MediaFecError::invalid("media FEC policy fields must be non-zero")); + } + if usize::from(widest) + usize::from(self.video_parity_shards(widest)) > 256 + || usize::from(self.audio_data_shards) + usize::from(self.audio_parity_shards) > 256 + { + return Err(MediaFecError::invalid("media FEC policy exceeds 256 shards per block")); + } + Ok(()) + } +} + +impl Default for MediaFecPolicy { + fn default() -> Self { + Self::STANDARD + } +} + +/// Why a block could not be protected or recovered. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum MediaFecError { + /// Fewer than `k` of the block's shards are present. Nothing is returned: + /// a block that cannot be repaired is reported, never guessed at. + BeyondRepair { shards_present: usize, shards_needed: usize }, + /// The inputs are not one well-formed block. + Invalid(String), +} + +impl MediaFecError { + fn invalid(message: impl Into) -> Self { + Self::Invalid(message.into()) + } +} + +impl fmt::Display for MediaFecError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::BeyondRepair { shards_present, shards_needed } => write!( + f, + "media FEC block is beyond repair: {shards_present} shards present, {shards_needed} needed" + ), + Self::Invalid(message) => write!(f, "invalid media FEC block: {message}"), + } + } +} + +impl std::error::Error for MediaFecError {} + +// --------------------------------------------------------------------------- +// The code: equal-length shards in, equal-length shards out. +// --------------------------------------------------------------------------- + +fn parity_shards(data: &[Vec], parity_count: usize) -> Result>, MediaFecError> { + let code = ReedSolomon::new(data.len(), parity_count) + .map_err(|error| MediaFecError::invalid(format!("{error:?}")))?; + let mut parity = vec![vec![0_u8; data[0].len()]; parity_count]; + code.encode_sep(data, &mut parity) + .map_err(|error| MediaFecError::invalid(format!("{error:?}")))?; + Ok(parity) +} + +/// `shards` holds the block's `k` data slots then `m` parity slots, `None` for +/// each missing. Returns the `k` data shards. +fn recover_data_shards( + data_shards: usize, + mut shards: Vec>>, +) -> Result>, MediaFecError> { + let parity_shards = shards.len() - data_shards; + let present = shards.iter().flatten().count(); + if present < data_shards { + return Err(MediaFecError::BeyondRepair { + shards_present: present, + shards_needed: data_shards, + }); + } + let code = ReedSolomon::new(data_shards, parity_shards) + .map_err(|error| MediaFecError::invalid(format!("{error:?}")))?; + code.reconstruct_data(&mut shards) + .map_err(|error| MediaFecError::invalid(format!("{error:?}")))?; + Ok(shards + .into_iter() + .take(data_shards) + .map(|shard| shard.expect("reconstruct_data fills every data shard")) + .collect()) +} + +fn padded(payload: &[u8], shard_bytes: usize) -> Vec { + let mut shard = vec![0_u8; shard_bytes]; + shard[..payload.len()].copy_from_slice(payload); + shard +} + +fn from_anyhow(error: anyhow::Error) -> MediaFecError { + MediaFecError::Invalid(error.to_string()) +} + +// --------------------------------------------------------------------------- +// Video +// --------------------------------------------------------------------------- + +/// Protects one video frame. `chunks` are all of the frame's access-unit +/// records, in any order; the result is every data and parity record in send +/// order. +/// +/// Within a block, data and parity alternate; across blocks, the sequence +/// rotates one record from each block at a time. Where every block is the same +/// size, no two consecutive `block_count` records share a block. Blocks that +/// differ in size run out at the end, and the rotation then cycles over the +/// blocks remaining. +/// +/// Errors if `chunks` are not exactly one whole frame (mixed frames, a missing +/// or repeated chunk index, disagreeing framing), if a chunk is not shard +/// shaped (every chunk of a block but its last must be the same length, the +/// last no longer), or if a chunk exceeds [`MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES`]. +pub fn protect_video_frame( + chunks: &[GameCultMediaVideoAccessUnitRecord], + policy: &MediaFecPolicy, +) -> Result, MediaFecError> { + policy.validate()?; + let first = chunks + .first() + .ok_or_else(|| MediaFecError::invalid("a video frame needs at least one chunk"))?; + let chunk_count = usize::from(first.chunk_count); + if chunks.len() != chunk_count { + return Err(MediaFecError::invalid(format!( + "frame {} declares {} chunks but {} were given", + first.frame_id, + chunk_count, + chunks.len() + ))); + } + let mut by_index: BTreeMap = BTreeMap::new(); + for chunk in chunks { + validate_video_record(chunk).map_err(from_anyhow)?; + if !same_frame(first, chunk) { + return Err(MediaFecError::invalid( + "a block never spans frames: chunks disagree on stream, session, frame or framing", + )); + } + if chunk.payload.len() > MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES { + return Err(MediaFecError::invalid(format!( + "chunk payload of {} bytes exceeds the {MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES}-byte shard limit", + chunk.payload.len() + ))); + } + if by_index.insert(chunk.chunk_index, chunk).is_some() { + return Err(MediaFecError::invalid(format!( + "chunk {} appears twice", + chunk.chunk_index + ))); + } + } + + let block_sizes = policy.video_block_sizes(first.chunk_count); + let block_count = block_sizes.len() as u16; + let mut lanes: Vec> = Vec::with_capacity(block_sizes.len()); + let mut start = 0_u16; + for (block_index, &size) in block_sizes.iter().enumerate() { + let members: Vec<&GameCultMediaVideoAccessUnitRecord> = (start..start + size) + .map(|index| by_index[&index]) + .collect(); + let shard_bytes = members[0].payload.len(); + let (last, body) = members.split_last().expect("block has a member"); + if body.iter().any(|chunk| chunk.payload.len() != shard_bytes) + || last.payload.len() > shard_bytes + { + return Err(MediaFecError::invalid( + "every chunk of a block but its last must share one length, and the last no longer", + )); + } + let data: Vec> = members + .iter() + .map(|chunk| padded(&chunk.payload, shard_bytes)) + .collect(); + let parity_count = policy.video_parity_shards(size); + let parity = parity_shards(&data, usize::from(parity_count))?; + + let parity_records: Vec = parity + .into_iter() + .enumerate() + .map(|(parity_index, payload)| { + GameCultMediaWireRecord::VideoParity(GameCultMediaVideoParityShardRecord { + stream_id: first.stream_id.clone(), + session_id: first.session_id.clone(), + frame_id: first.frame_id, + codec: first.codec.clone(), + pts_ticks: first.pts_ticks, + duration_ticks: first.duration_ticks, + timebase_num: first.timebase_num, + timebase_den: first.timebase_den, + keyframe: first.keyframe, + dependency_frame_id: first.dependency_frame_id, + deadline_ticks: first.deadline_ticks, + chunk_count: first.chunk_count, + fec_scheme: MEDIA_FEC_SCHEME_RS_GF256_V1.to_string(), + block_index: block_index as u16, + block_count, + block_data_start: start, + block_data_count: size, + parity_index: parity_index as u16, + parity_count, + shard_payload_bytes: shard_bytes as u32, + last_chunk_payload_bytes: last.payload.len() as u32, + payload, + }) + }) + .collect(); + let data_records = members + .iter() + .map(|chunk| GameCultMediaWireRecord::Video((*chunk).clone())); + lanes.push(interleave(data_records.collect(), parity_records)); + start += size; + } + Ok(rotate(lanes)) +} + +fn same_frame(a: &GameCultMediaVideoAccessUnitRecord, b: &GameCultMediaVideoAccessUnitRecord) -> bool { + a.stream_id == b.stream_id + && a.session_id == b.session_id + && a.frame_id == b.frame_id + && a.codec == b.codec + && a.pts_ticks == b.pts_ticks + && a.duration_ticks == b.duration_ticks + && a.timebase_num == b.timebase_num + && a.timebase_den == b.timebase_den + && a.keyframe == b.keyframe + && a.dependency_frame_id == b.dependency_frame_id + && a.deadline_ticks == b.deadline_ticks + && a.chunk_count == b.chunk_count +} + +/// Data and parity alternate: `D0 P0 D1 P1 ...`, the longer list finishing the +/// lane. +fn interleave(data: Vec, parity: Vec) -> Vec { + let mut data = data.into_iter(); + let mut parity = parity.into_iter(); + let mut lane = Vec::new(); + loop { + let next_data = data.next(); + let next_parity = parity.next(); + if next_data.is_none() && next_parity.is_none() { + return lane; + } + lane.extend(next_data); + lane.extend(next_parity); + } +} + +/// One record from each lane in turn, until every lane is empty. +fn rotate(lanes: Vec>) -> Vec { + let mut lanes: Vec<_> = lanes.into_iter().map(Vec::into_iter).collect(); + let mut out = Vec::new(); + loop { + let before = out.len(); + for lane in &mut lanes { + out.extend(lane.next()); + } + if out.len() == before { + return out; + } + } +} + +/// Rebuilds the data chunks a block lost. +/// +/// `parity` are the block's parity shards that arrived (at least one, for its +/// geometry) and `data` its data chunks that arrived; each may be in any order +/// and may repeat. Returns the missing chunks, ordered by `chunk_index`: empty +/// if none were missing. Fails with [`MediaFecError::BeyondRepair`] if fewer +/// than `block_data_count` shards of the block are present, and with +/// [`MediaFecError::Invalid`] if the records are not from one block. +pub fn recover_video_block( + parity: &[GameCultMediaVideoParityShardRecord], + data: &[GameCultMediaVideoAccessUnitRecord], +) -> Result, MediaFecError> { + let geometry = parity + .first() + .ok_or_else(|| MediaFecError::invalid("recovery needs at least one parity shard"))?; + let k = usize::from(geometry.block_data_count); + let m = usize::from(geometry.parity_count); + let start = usize::from(geometry.block_data_start); + let shard_bytes = geometry.shard_payload_bytes as usize; + + let mut shards: Vec>> = vec![None; k + m]; + for shard in parity { + validate_video_parity_record(shard).map_err(from_anyhow)?; + if !same_block(geometry, shard) { + return Err(MediaFecError::invalid("parity shards are not from one block")); + } + shards[k + usize::from(shard.parity_index)].get_or_insert_with(|| shard.payload.clone()); + } + for chunk in data { + validate_video_record(chunk).map_err(from_anyhow)?; + let index = usize::from(chunk.chunk_index); + let framing_agrees = chunk.stream_id == geometry.stream_id + && chunk.session_id == geometry.session_id + && chunk.frame_id == geometry.frame_id + && chunk.chunk_count == geometry.chunk_count; + if !framing_agrees || index < start || index >= start + k { + return Err(MediaFecError::invalid("data chunk does not belong to the parity's block")); + } + let expected = if index == start + k - 1 { + geometry.last_chunk_payload_bytes as usize + } else { + shard_bytes + }; + if chunk.payload.len() != expected { + return Err(MediaFecError::invalid(format!( + "chunk {index} is {} bytes but the block declares {expected}", + chunk.payload.len() + ))); + } + shards[index - start].get_or_insert_with(|| padded(&chunk.payload, shard_bytes)); + } + + let missing: Vec = (0..k).filter(|&slot| shards[slot].is_none()).collect(); + if missing.is_empty() { + return Ok(Vec::new()); + } + let recovered = recover_data_shards(k, shards)?; + Ok(missing + .into_iter() + .map(|slot| { + let mut payload = recovered[slot].clone(); + if slot == k - 1 { + payload.truncate(geometry.last_chunk_payload_bytes as usize); + } + GameCultMediaVideoAccessUnitRecord { + stream_id: geometry.stream_id.clone(), + session_id: geometry.session_id.clone(), + frame_id: geometry.frame_id, + codec: geometry.codec.clone(), + pts_ticks: geometry.pts_ticks, + duration_ticks: geometry.duration_ticks, + timebase_num: geometry.timebase_num, + timebase_den: geometry.timebase_den, + keyframe: geometry.keyframe, + dependency_frame_id: geometry.dependency_frame_id, + deadline_ticks: geometry.deadline_ticks, + chunk_index: (start + slot) as u16, + chunk_count: geometry.chunk_count, + payload, + } + }) + .collect()) +} + +fn same_block(a: &GameCultMediaVideoParityShardRecord, b: &GameCultMediaVideoParityShardRecord) -> bool { + a.stream_id == b.stream_id + && a.session_id == b.session_id + && a.frame_id == b.frame_id + && a.codec == b.codec + && a.pts_ticks == b.pts_ticks + && a.duration_ticks == b.duration_ticks + && a.timebase_num == b.timebase_num + && a.timebase_den == b.timebase_den + && a.keyframe == b.keyframe + && a.dependency_frame_id == b.dependency_frame_id + && a.deadline_ticks == b.deadline_ticks + && a.chunk_count == b.chunk_count + && a.fec_scheme == b.fec_scheme + && a.block_index == b.block_index + && a.block_count == b.block_count + && a.block_data_start == b.block_data_start + && a.block_data_count == b.block_data_count + && a.parity_count == b.parity_count + && a.shard_payload_bytes == b.shard_payload_bytes + && a.last_chunk_payload_bytes == b.last_chunk_payload_bytes +} + +// --------------------------------------------------------------------------- +// Audio +// --------------------------------------------------------------------------- + +/// Protects one block of audio packets: exactly `policy.audio_data_shards` +/// packets with contiguous ids and presentation times, one payload length, and +/// one codec and timebase. Returns the parity records in `parity_index` order. +pub fn protect_audio_block( + packets: &[GameCultMediaAudioPacketRecord], + policy: &MediaFecPolicy, +) -> Result, MediaFecError> { + policy.validate()?; + if packets.len() != usize::from(policy.audio_data_shards) { + return Err(MediaFecError::invalid(format!( + "an audio block is {} packets, got {}", + policy.audio_data_shards, + packets.len() + ))); + } + let first = &packets[0]; + let shard_bytes = first.payload.len(); + for (index, packet) in packets.iter().enumerate() { + validate_audio_record(packet).map_err(from_anyhow)?; + let expected_id = first.packet_id.checked_add(index as u64); + let expected_pts = i64::from(first.duration_ticks) + .checked_mul(index as i64) + .and_then(|offset| first.pts_ticks.checked_add(offset)); + if packet.stream_id != first.stream_id + || packet.session_id != first.session_id + || packet.codec != first.codec + || packet.timebase_num != first.timebase_num + || packet.timebase_den != first.timebase_den + || packet.duration_ticks != first.duration_ticks + || Some(packet.packet_id) != expected_id + || Some(packet.pts_ticks) != expected_pts + || packet.payload.len() != shard_bytes + { + return Err(MediaFecError::invalid( + "audio block packets must be contiguous and share codec, timebase, duration and payload length", + )); + } + } + if shard_bytes > MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES { + return Err(MediaFecError::invalid(format!( + "audio payload of {shard_bytes} bytes exceeds the {MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES}-byte shard limit" + ))); + } + let data: Vec> = packets.iter().map(|packet| packet.payload.clone()).collect(); + let deadline_ticks = packets.iter().map(|packet| packet.deadline_ticks).max().unwrap_or(0); + Ok(parity_shards(&data, usize::from(policy.audio_parity_shards))? + .into_iter() + .enumerate() + .map(|(parity_index, payload)| GameCultMediaAudioParityShardRecord { + stream_id: first.stream_id.clone(), + session_id: first.session_id.clone(), + codec: first.codec.clone(), + fec_scheme: MEDIA_FEC_SCHEME_RS_GF256_V1.to_string(), + base_packet_id: first.packet_id, + base_pts_ticks: first.pts_ticks, + packet_duration_ticks: first.duration_ticks, + timebase_num: first.timebase_num, + timebase_den: first.timebase_den, + deadline_ticks, + data_shard_count: policy.audio_data_shards, + parity_index: parity_index as u16, + parity_shard_count: policy.audio_parity_shards, + shard_payload_bytes: shard_bytes as u32, + payload, + }) + .collect()) +} + +/// Rebuilds the packets an audio block lost, ordered by `packet_id`. A +/// recovered packet takes its id and presentation time from the block's +/// arithmetic and its deadline from the block's, the latest in the block. +/// Errors as [`recover_video_block`] does. +pub fn recover_audio_block( + parity: &[GameCultMediaAudioParityShardRecord], + data: &[GameCultMediaAudioPacketRecord], +) -> Result, MediaFecError> { + let geometry = parity + .first() + .ok_or_else(|| MediaFecError::invalid("recovery needs at least one parity shard"))?; + let k = usize::from(geometry.data_shard_count); + let m = usize::from(geometry.parity_shard_count); + let shard_bytes = geometry.shard_payload_bytes as usize; + + let mut shards: Vec>> = vec![None; k + m]; + for shard in parity { + validate_audio_parity_record(shard).map_err(from_anyhow)?; + if !same_audio_block(geometry, shard) { + return Err(MediaFecError::invalid("parity shards are not from one block")); + } + shards[k + usize::from(shard.parity_index)].get_or_insert_with(|| shard.payload.clone()); + } + for packet in data { + validate_audio_record(packet).map_err(from_anyhow)?; + let slot = packet + .packet_id + .checked_sub(geometry.base_packet_id) + .filter(|slot| *slot < k as u64) + .ok_or_else(|| MediaFecError::invalid("audio packet does not belong to the parity's block"))? + as usize; + if packet.stream_id != geometry.stream_id + || packet.session_id != geometry.session_id + || packet.payload.len() != shard_bytes + { + return Err(MediaFecError::invalid("audio packet disagrees with the parity's block")); + } + shards[slot].get_or_insert_with(|| packet.payload.clone()); + } + + let missing: Vec = (0..k).filter(|&slot| shards[slot].is_none()).collect(); + if missing.is_empty() { + return Ok(Vec::new()); + } + let recovered = recover_data_shards(k, shards)?; + Ok(missing + .into_iter() + .map(|slot| GameCultMediaAudioPacketRecord { + stream_id: geometry.stream_id.clone(), + session_id: geometry.session_id.clone(), + packet_id: geometry.base_packet_id + slot as u64, + codec: geometry.codec.clone(), + pts_ticks: geometry.base_pts_ticks + i64::from(geometry.packet_duration_ticks) * slot as i64, + duration_ticks: geometry.packet_duration_ticks, + timebase_num: geometry.timebase_num, + timebase_den: geometry.timebase_den, + deadline_ticks: geometry.deadline_ticks, + payload: recovered[slot].clone(), + }) + .collect()) +} + +fn same_audio_block(a: &GameCultMediaAudioParityShardRecord, b: &GameCultMediaAudioParityShardRecord) -> bool { + a.stream_id == b.stream_id + && a.session_id == b.session_id + && a.codec == b.codec + && a.fec_scheme == b.fec_scheme + && a.base_packet_id == b.base_packet_id + && a.base_pts_ticks == b.base_pts_ticks + && a.packet_duration_ticks == b.packet_duration_ticks + && a.timebase_num == b.timebase_num + && a.timebase_den == b.timebase_den + && a.deadline_ticks == b.deadline_ticks + && a.data_shard_count == b.data_shard_count + && a.parity_shard_count == b.parity_shard_count + && a.shard_payload_bytes == b.shard_payload_bytes +} diff --git a/packages/cultnet-rs/src/media_stream_contracts.rs b/packages/cultnet-rs/src/media_stream_contracts.rs index 5704912e..2991bb43 100644 --- a/packages/cultnet-rs/src/media_stream_contracts.rs +++ b/packages/cultnet-rs/src/media_stream_contracts.rs @@ -36,15 +36,15 @@ //! Nothing here specifies a delivery guarantee. Whether a stream rides a //! reliable or lossy CultNet channel, and whether it carries parity, is a //! transport decision made per stream — not a property of the payload shape. -//! Video parity shards exist for producers that choose forward error correction; -//! audio currently has no parity record, and a producer emitting audio on a -//! lossy channel needs one. +//! Video and audio parity shards exist for producers that choose forward error +//! correction; [`crate::media_fec`] owns the code that makes and consumes them. use cultcache_rs::DatabaseEntry; pub const GAMECULT_MEDIA_VIDEO_ACCESS_UNIT_SCHEMA: &str = "gamecult.media_video_access_unit.v1"; -pub const GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA: &str = "gamecult.media_video_parity_shard.v2"; +pub const GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA: &str = "gamecult.media_video_parity_shard.v3"; pub const GAMECULT_MEDIA_AUDIO_PACKET_SCHEMA: &str = "gamecult.media_audio_packet.v1"; +pub const GAMECULT_MEDIA_AUDIO_PARITY_SHARD_SCHEMA: &str = "gamecult.media_audio_parity_shard.v1"; pub const GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA: &str = "gamecult.media_receiver_feedback.v1"; pub const GAMECULT_MEDIA_STREAM_ADVERTISEMENT_SCHEMA: &str = "gamecult.media_stream_advertisement.v1"; @@ -464,15 +464,25 @@ pub struct GameCultMediaVideoAccessUnitRecord { pub payload: Vec, } -/// Forward-error-correction parity for one video access unit. Carries the -/// framing fields of the access unit it protects so a receiver can recover -/// without having seen the original, plus the shard geometry needed to run -/// recovery: `parity_index` within `parity_count`, and the chunk sizes the -/// parity was computed over. +/// Forward-error-correction parity for one block of a video access unit. +/// +/// A frame's `chunk_count` chunks are split into `block_count` contiguous +/// blocks, and each block carries `parity_count` parity shards computed over its +/// `block_data_count` data chunks (`block_data_start..block_data_start + +/// block_data_count`). Any `block_data_count` of a block's +/// `block_data_count + parity_count` shards recover it. The math is +/// [`crate::media_fec`]'s and is named by `fec_scheme`; a decoder that does not +/// know the scheme must drop the shard rather than guess. +/// +/// It carries the framing fields of the access unit it protects, so a receiver +/// can rebuild a lost chunk without having seen any chunk of the frame. +/// `shard_payload_bytes` is the padded shard length; `last_chunk_payload_bytes` +/// is the true length of the block's final data chunk, the only one that may be +/// shorter. #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] #[cultcache( type = "gamecult.media_video_parity_shard", - schema = "gamecult.media_video_parity_shard.v2" + schema = "gamecult.media_video_parity_shard.v3" )] pub struct GameCultMediaVideoParityShardRecord { #[cultcache(key = 0)] @@ -500,14 +510,26 @@ pub struct GameCultMediaVideoParityShardRecord { #[cultcache(key = 11)] pub chunk_count: u16, #[cultcache(key = 12)] - pub parity_index: u16, + pub fec_scheme: String, #[cultcache(key = 13)] - pub parity_count: u16, + pub block_index: u16, #[cultcache(key = 14)] - pub chunk_payload_bytes: u32, + pub block_count: u16, #[cultcache(key = 15)] + pub block_data_start: u16, + /// `k`: how many data chunks this block protects. + #[cultcache(key = 16)] + pub block_data_count: u16, + #[cultcache(key = 17)] + pub parity_index: u16, + /// `m`: how many parity shards this block carries. + #[cultcache(key = 18)] + pub parity_count: u16, + #[cultcache(key = 19)] + pub shard_payload_bytes: u32, + #[cultcache(key = 20)] pub last_chunk_payload_bytes: u32, - #[cultcache(key = 16, bytes)] + #[cultcache(key = 21, bytes)] pub payload: Vec, } @@ -541,6 +563,60 @@ pub struct GameCultMediaAudioPacketRecord { pub payload: Vec, } +/// Forward-error-correction parity for a block of consecutive audio packets. +/// +/// The block is `data_shard_count` packets with contiguous ids starting at +/// `base_packet_id`, all with one payload length (`shard_payload_bytes`) and +/// contiguous presentation times spaced `packet_duration_ticks` apart. A packet +/// recovered from parity takes its id and time from that arithmetic, and its +/// `deadline_ticks` from the block's, which is the latest in the block: the +/// per-packet deadline is not recoverable, so a recovered packet may be held +/// past its own but never past the block's. +/// +/// Codecs whose frames vary in length pad each one to a shared shard length +/// before parity is computed; that framing belongs to the codec, not to this +/// record. +#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] +#[cultcache( + type = "gamecult.media_audio_parity_shard", + schema = "gamecult.media_audio_parity_shard.v1" +)] +pub struct GameCultMediaAudioParityShardRecord { + #[cultcache(key = 0)] + pub stream_id: String, + #[cultcache(key = 1)] + pub session_id: String, + #[cultcache(key = 2)] + pub codec: String, + #[cultcache(key = 3)] + pub fec_scheme: String, + #[cultcache(key = 4)] + pub base_packet_id: u64, + #[cultcache(key = 5)] + pub base_pts_ticks: i64, + #[cultcache(key = 6)] + pub packet_duration_ticks: u32, + #[cultcache(key = 7)] + pub timebase_num: u32, + #[cultcache(key = 8)] + pub timebase_den: u32, + /// The latest `deadline_ticks` among the block's data packets. + #[cultcache(key = 9)] + pub deadline_ticks: i64, + /// `k`. + #[cultcache(key = 10)] + pub data_shard_count: u16, + #[cultcache(key = 11)] + pub parity_index: u16, + /// `m`. + #[cultcache(key = 12)] + pub parity_shard_count: u16, + #[cultcache(key = 13)] + pub shard_payload_bytes: u32, + #[cultcache(key = 14, bytes)] + pub payload: Vec, +} + /// What a receiver observed, published back to the producer so it can adapt: /// what decoded, what is missing, what arrived too late, and how much queue and /// jitter the receiver is carrying. diff --git a/packages/cultnet-rs/src/media_stream_wire.rs b/packages/cultnet-rs/src/media_stream_wire.rs index 9b7fbc1b..b1119ca0 100644 --- a/packages/cultnet-rs/src/media_stream_wire.rs +++ b/packages/cultnet-rs/src/media_stream_wire.rs @@ -28,27 +28,27 @@ use crate::contracts::{ CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, CultNetWireContract, decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, }; +use crate::media_fec::{MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES, MEDIA_FEC_SCHEME_RS_GF256_V1}; use crate::media_stream_contracts::{ - GAMECULT_MEDIA_AUDIO_PACKET_SCHEMA, GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA, - GAMECULT_MEDIA_VIDEO_ACCESS_UNIT_SCHEMA, GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA, - GameCultMediaAudioPacketRecord, GameCultMediaReceiverFeedbackRecord, + GAMECULT_MEDIA_AUDIO_PACKET_SCHEMA, GAMECULT_MEDIA_AUDIO_PARITY_SHARD_SCHEMA, + GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA, GAMECULT_MEDIA_VIDEO_ACCESS_UNIT_SCHEMA, + GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA, GameCultMediaAudioPacketRecord, + GameCultMediaAudioParityShardRecord, GameCultMediaReceiverFeedbackRecord, GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord, }; -/// The CultNet channel video and receiver feedback ride. +/// The CultNet channel every media record rides: video, audio, their parity, +/// and receiver feedback. Whether it is reliable is the transport profile's +/// choice per session; stale media is dropped, not retransmitted. pub const GAMECULT_MEDIA_CHANNEL: &str = "media"; -/// The CultNet channel audio rides. Audio has its own channel so one session -/// can carry it reliably with a short expiry while video stays lossy under -/// parity; the RUDP transport profile owns each channel's delivery. -pub const GAMECULT_MEDIA_AUDIO_CHANNEL: &str = "audio"; - /// One media record on its way somewhere. #[derive(Clone, Debug, PartialEq, Eq)] pub enum GameCultMediaWireRecord { Video(GameCultMediaVideoAccessUnitRecord), VideoParity(GameCultMediaVideoParityShardRecord), Audio(GameCultMediaAudioPacketRecord), + AudioParity(GameCultMediaAudioParityShardRecord), Feedback(GameCultMediaReceiverFeedbackRecord), } @@ -84,6 +84,7 @@ impl GameCultMediaWireRecord { Self::Video(_) => GAMECULT_MEDIA_VIDEO_ACCESS_UNIT_SCHEMA, Self::VideoParity(_) => GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA, Self::Audio(_) => GAMECULT_MEDIA_AUDIO_PACKET_SCHEMA, + Self::AudioParity(_) => GAMECULT_MEDIA_AUDIO_PARITY_SHARD_SCHEMA, Self::Feedback(_) => GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA, } } @@ -96,13 +97,21 @@ impl GameCultMediaWireRecord { record.stream_id, record.session_id, record.frame_id, record.chunk_index ), Self::VideoParity(record) => format!( - "{}:{}:video-parity:{}:{}", - record.stream_id, record.session_id, record.frame_id, record.parity_index + "{}:{}:video-parity:{}:{}:{}", + record.stream_id, + record.session_id, + record.frame_id, + record.block_index, + record.parity_index ), Self::Audio(record) => format!( "{}:{}:audio:{}", record.stream_id, record.session_id, record.packet_id ), + Self::AudioParity(record) => format!( + "{}:{}:audio-parity:{}:{}", + record.stream_id, record.session_id, record.base_packet_id, record.parity_index + ), Self::Feedback(record) => format!( "{}:{}:feedback:{}", record.stream_id, record.session_id, record.receiver_id @@ -124,6 +133,7 @@ impl GameCultMediaWireRecord { Self::Video(record) => rmp_serde::to_vec(record), Self::VideoParity(record) => rmp_serde::to_vec(record), Self::Audio(record) => rmp_serde::to_vec(record), + Self::AudioParity(record) => rmp_serde::to_vec(record), Self::Feedback(record) => rmp_serde::to_vec(record), } .map_err(Into::into) @@ -191,6 +201,12 @@ pub fn decode_media_wire_record(payload: &[u8]) -> Result { + let record: GameCultMediaAudioParityShardRecord = + rmp_serde::from_slice(&document.payload)?; + validate_audio_parity_record(&record)?; + GameCultMediaWireRecord::AudioParity(record) + } GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA => { let record: GameCultMediaReceiverFeedbackRecord = rmp_serde::from_slice(&document.payload)?; @@ -349,39 +365,105 @@ pub fn validate_video_parity_record(record: &GameCultMediaVideoParityShardRecord "video parity media record deadline_ticks must not precede pts_ticks" )); } + if record.fec_scheme != MEDIA_FEC_SCHEME_RS_GF256_V1 { + return Err(anyhow!( + "video parity media record fec_scheme {:?} is not supported", + record.fec_scheme + )); + } if record.chunk_count == 0 { return Err(anyhow!( "video parity media record chunk_count must be non-zero" )); } + if record.block_count == 0 || record.block_index >= record.block_count { + return Err(anyhow!( + "video parity media record block_index {} is outside block_count {}", + record.block_index, + record.block_count + )); + } + if record.block_data_count == 0 + || u32::from(record.block_data_start) + u32::from(record.block_data_count) + > u32::from(record.chunk_count) + { + return Err(anyhow!( + "video parity media record block does not lie inside the frame's chunks" + )); + } if record.parity_count == 0 || record.parity_index >= record.parity_count { return Err(anyhow!( - "video parity media record has invalid parity stripe metadata" + "video parity media record parity_index {} is outside parity_count {}", + record.parity_index, + record.parity_count )); } - if record.parity_count > record.chunk_count { + if usize::from(record.block_data_count) + usize::from(record.parity_count) > 256 { return Err(anyhow!( - "video parity media record parity_count exceeds chunk_count" + "video parity media record block exceeds 256 shards" )); } - if record.chunk_payload_bytes == 0 || record.last_chunk_payload_bytes == 0 { + if record.shard_payload_bytes == 0 + || record.shard_payload_bytes as usize > MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES + { return Err(anyhow!( - "video parity media record chunk lengths must be non-zero" + "video parity media record shard_payload_bytes must be in 1..={MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES}" )); } - if record.last_chunk_payload_bytes > record.chunk_payload_bytes { + if record.last_chunk_payload_bytes == 0 + || record.last_chunk_payload_bytes > record.shard_payload_bytes + { return Err(anyhow!( - "video parity media record last chunk length exceeds regular chunk length" + "video parity media record last chunk length must be in 1..=shard_payload_bytes" )); } - if record.payload.is_empty() { + if record.payload.len() != record.shard_payload_bytes as usize { return Err(anyhow!( - "video parity media record payload must be non-empty" + "video parity media record payload does not match declared shard length" )); } - if record.payload.len() > record.chunk_payload_bytes as usize { + Ok(()) +} + +pub fn validate_audio_parity_record(record: &GameCultMediaAudioParityShardRecord) -> Result<()> { + if record.stream_id.is_empty() || record.session_id.is_empty() || record.codec.is_empty() { return Err(anyhow!( - "video parity media record payload exceeds declared chunk length" + "audio parity media record stream_id, session_id and codec must be non-empty" + )); + } + if record.timebase_num == 0 || record.timebase_den == 0 || record.packet_duration_ticks == 0 { + return Err(anyhow!( + "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", + record.fec_scheme + )); + } + if record.data_shard_count == 0 + || record.parity_shard_count == 0 + || record.parity_index >= record.parity_shard_count + || usize::from(record.data_shard_count) + usize::from(record.parity_shard_count) > 256 + { + return Err(anyhow!("audio parity media record stripe metadata is invalid")); + } + if record.shard_payload_bytes == 0 + || record.shard_payload_bytes as usize > MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES + { + return Err(anyhow!( + "audio parity media record shard_payload_bytes must be in 1..={MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES}" + )); + } + if record.payload.len() != record.shard_payload_bytes as usize { + return Err(anyhow!( + "audio parity media record payload does not match declared shard length" )); } Ok(()) diff --git a/packages/cultnet-rs/src/rudp.rs b/packages/cultnet-rs/src/rudp.rs index 174739be..d1afca7c 100644 --- a/packages/cultnet-rs/src/rudp.rs +++ b/packages/cultnet-rs/src/rudp.rs @@ -29,7 +29,7 @@ use crate::encode_cultnet_message_to_vec; const RUDP_MAGIC: [u8; 4] = [0x43, 0x4e, 0x52, 0x30]; const RUDP_VERSION: u8 = 0; -const RUDP_FIXED_HEADER_BYTES: usize = 36; +pub(crate) const RUDP_FIXED_HEADER_BYTES: usize = 36; pub const CULTNET_RUDP_DEFAULT_MAX_PAYLOAD_BYTES: usize = 16 * 1024 * 1024; pub const CULTNET_RUDP_RELIABLE_SEND_WINDOW_PACKETS: usize = 32; const RUDP_RECEIVED_SEQUENCE_WINDOW: usize = 4_096; @@ -2215,17 +2215,6 @@ pub fn create_rudp_transport_profile( max_pending_reliable_packets: options.max_pending_reliable_packets, reliable_expire_after_ms: options.media_reliable_expire_after_ms, }, - // Audio beside lossy video on one session: reliable, but a packet - // older than the media expiry is not worth a retransmit. - CultNetTransportChannel { - channel_id: "audio".to_string(), - delivery: CultNetTransportDelivery::Reliable, - ordering: CultNetTransportOrdering::Unordered, - 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, - }, ], }], } @@ -2391,7 +2380,7 @@ fn channel_send_options( // 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" | "audio" => CultNetRudpSendOptions { + "media" => CultNetRudpSendOptions { reliable: matches!(delivery, CultNetTransportDelivery::Reliable), ordered: false, sequenced: false, diff --git a/packages/cultnet-rs/tests/cultnet.rs b/packages/cultnet-rs/tests/cultnet.rs index 426e2ed9..845ec846 100644 --- a/packages/cultnet-rs/tests/cultnet.rs +++ b/packages/cultnet-rs/tests/cultnet.rs @@ -432,12 +432,6 @@ fn rudp_transport_profile_advertises_state_and_realtime_channels() { CultNetTransportOrdering::Unordered, Some(64) ), - ( - "audio", - CultNetTransportDelivery::Reliable, - CultNetTransportOrdering::Unordered, - Some(64) - ), ] ); } @@ -3006,7 +3000,7 @@ fn accept_rudp_session(session: &mut CultNetRudpSession, connection_id: u32) -> } #[test] -fn rudp_profile_expires_only_the_media_channels() -> Result<()> { +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); @@ -3020,14 +3014,12 @@ fn rudp_profile_expires_only_the_media_channels() -> Result<()> { .map(|channel| (channel.channel_id.as_str(), channel.reliable_expire_after_ms)) .collect::>(); - for media_channel in ["media", "audio"] { - assert!( - channels.contains(&(media_channel, Some(75))), - "the {media_channel} channel advertises its expiry so a peer can see it: {channels:?}" - ); - } + 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 !matches!(*channel_id, "media" | "audio") { + if *channel_id != "media" { assert_eq!( *expiry, None, "state channel {channel_id} must never expire a reliable send" diff --git a/packages/cultnet-rs/tests/media_fec.rs b/packages/cultnet-rs/tests/media_fec.rs new file mode 100644 index 00000000..8e121b48 --- /dev/null +++ b/packages/cultnet-rs/tests/media_fec.rs @@ -0,0 +1,723 @@ +//! The media FEC codec's guarantees, proved against its public surface. +//! +//! Every test drives `protect_*` and `recover_*` the way a producer and a +//! consumer would, and asserts what a consumer is entitled to rely on: what is +//! recovered is exactly what was sent, what cannot be recovered is reported and +//! never guessed, and a shard always fits a datagram. + +use cultnet_rs::{ + GameCultMediaAudioPacketRecord, GameCultMediaAudioParityShardRecord, + GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord, + GameCultMediaWireRecord, MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES, 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, +}; + +const STANDARD: MediaFecPolicy = MediaFecPolicy::STANDARD; + +/// Deterministic, non-constant bytes so a shard mix-up cannot pass by accident. +fn bytes(seed: u32, len: usize) -> Vec { + let mut state = seed.wrapping_mul(2_654_435_761).wrapping_add(1); + (0..len) + .map(|_| { + state = state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223); + (state >> 24) as u8 + }) + .collect() +} + +/// One frame of `chunk_count` chunks: every chunk `chunk_bytes` long except the +/// last, which is `last_bytes`. +fn frame( + frame_id: u64, + chunk_count: u16, + chunk_bytes: usize, + last_bytes: usize, +) -> Vec { + (0..chunk_count) + .map(|index| GameCultMediaVideoAccessUnitRecord { + stream_id: "muninn.raven.av.rudp".to_string(), + session_id: "session-1".to_string(), + frame_id, + codec: "h264".to_string(), + pts_ticks: 90_000, + duration_ticks: 3_000, + timebase_num: 1, + timebase_den: 90_000, + keyframe: frame_id % 2 == 0, + dependency_frame_id: (frame_id % 2 == 1).then(|| frame_id - 1), + deadline_ticks: 90_000 + 22_500, + chunk_index: index, + chunk_count, + payload: bytes( + (frame_id as u32).wrapping_mul(1_000).wrapping_add(u32::from(index)), + if index + 1 == chunk_count { last_bytes } else { chunk_bytes }, + ), + }) + .collect() +} + +fn audio_block(base_packet_id: u64, payload_bytes: usize) -> Vec { + (0..4_u64) + .map(|index| GameCultMediaAudioPacketRecord { + stream_id: "muninn.raven.av.rudp".to_string(), + session_id: "session-1".to_string(), + packet_id: base_packet_id + index, + codec: "opus".to_string(), + pts_ticks: 48_000 + 960 * index as i64, + duration_ticks: 960, + timebase_num: 1, + timebase_den: 48_000, + deadline_ticks: 60_000 + 960 * index as i64, + payload: bytes((base_packet_id as u32).wrapping_mul(10).wrapping_add(index as u32), payload_bytes), + }) + .collect() +} + +fn split( + records: Vec, +) -> (Vec, Vec) { + let mut data = Vec::new(); + let mut parity = Vec::new(); + for record in records { + match record { + GameCultMediaWireRecord::Video(record) => data.push(record), + GameCultMediaWireRecord::VideoParity(record) => parity.push(record), + other => panic!("a video frame produced {}", other.schema_id()), + } + } + (data, parity) +} + +/// The block a wire record belongs to, from its own fields. +fn block_of(record: &GameCultMediaWireRecord, sizes: &[u16]) -> usize { + match record { + GameCultMediaWireRecord::Video(chunk) => { + let mut start = 0_u16; + for (block, size) in sizes.iter().enumerate() { + if chunk.chunk_index < start + size { + return block; + } + start += size; + } + panic!("chunk {} is outside every block", chunk.chunk_index) + } + GameCultMediaWireRecord::VideoParity(parity) => usize::from(parity.block_index), + other => panic!("not a video record: {}", other.schema_id()), + } +} + +// --------------------------------------------------------------------------- +// Recovery: the MDS guarantee, exhaustively +// --------------------------------------------------------------------------- + +#[test] +fn the_policy_gives_the_ruled_block_geometry() { + // m = max(2, ceil(k / 4)), written independently of the policy. + for k in 1..=16_u16 { + let expected = std::cmp::max(2, (k + 3) / 4); + assert_eq!(STANDARD.video_parity_shards(k), expected, "k = {k}"); + } + assert_eq!(STANDARD.audio_data_shards, 4); + assert_eq!(STANDARD.audio_parity_shards, 2); +} + +#[test] +fn a_frame_splits_into_ceil_n_over_16_near_equal_blocks() { + for n in 1..=200_u16 { + let sizes = STANDARD.video_block_sizes(n); + assert_eq!(sizes.len(), usize::from(n.div_ceil(16)), "n = {n}"); + assert_eq!(sizes.iter().sum::(), n, "n = {n}"); + assert!(sizes.iter().all(|size| (1..=16).contains(size)), "n = {n}: {sizes:?}"); + let (widest, narrowest) = (sizes.iter().max().unwrap(), sizes.iter().min().unwrap()); + assert!(widest - narrowest <= 1, "n = {n}: {sizes:?}"); + assert!(sizes.windows(2).all(|pair| pair[0] >= pair[1]), "larger blocks first: {sizes:?}"); + } +} + +/// Erases every pattern of at most `m` records from a `k`-chunk frame's wire +/// output and requires the data back, byte for byte, then requires every +/// `m + 1` pattern to be refused. +fn assert_every_erasure_pattern_behaves(k: u16) { + let original = frame(u64::from(k), k, 37, 21); + let records = protect_video_frame(&original, &STANDARD).expect("protects"); + let m = usize::from(STANDARD.video_parity_shards(k)); + let total = usize::from(k) + m; + assert_eq!(records.len(), total, "one block: k data and m parity records"); + + for erased in 0_u32..(1 << total) { + let erasures = erased.count_ones() as usize; + if erasures > m + 1 { + continue; + } + let mut data = Vec::new(); + let mut parity = Vec::new(); + for (slot, record) in records.iter().enumerate() { + if erased & (1 << slot) != 0 { + continue; + } + match record { + GameCultMediaWireRecord::Video(chunk) => data.push(chunk.clone()), + GameCultMediaWireRecord::VideoParity(shard) => parity.push(shard.clone()), + _ => unreachable!(), + } + } + if parity.is_empty() { + continue; // a receiver holding no parity has nothing to recover with + } + let outcome = recover_video_block(&parity, &data); + if erasures <= m { + let recovered = outcome.unwrap_or_else(|error| { + panic!("k={k} m={m} erased={erased:#b}: {error}"); + }); + data.extend(recovered); + data.sort_by_key(|chunk| chunk.chunk_index); + assert_eq!(data, original, "k={k} m={m} erased={erased:#b}"); + } else { + assert!( + matches!(outcome, Err(MediaFecError::BeyondRepair { .. })), + "k={k} m={m} erased={erased:#b}: {outcome:?} must be beyond repair, never wrong bytes" + ); + } + } +} + +#[test] +fn every_le_m_erasure_recovers_exactly_and_m_plus_one_is_refused_for_every_video_k() { + for k in 1..=16 { + assert_every_erasure_pattern_behaves(k); + } +} + +#[test] +fn every_le_m_erasure_recovers_exactly_and_m_plus_one_is_refused_for_audio_4_plus_2() { + let original = audio_block(100, 41); + let parity = protect_audio_block(&original, &STANDARD).expect("protects"); + assert_eq!(parity.len(), 2); + for erased in 0_u32..(1 << 6) { + let erasures = erased.count_ones(); + let data: Vec<_> = (0..4) + .filter(|slot| erased & (1 << slot) == 0) + .map(|slot| original[slot].clone()) + .collect(); + let present_parity: Vec<_> = (0..2) + .filter(|slot| erased & (1 << (4 + slot)) == 0) + .map(|slot| parity[slot].clone()) + .collect(); + if present_parity.is_empty() { + continue; + } + let outcome = recover_audio_block(&present_parity, &data); + if erasures <= 2 { + let mut all = data.clone(); + let block_deadline = original.iter().map(|p| p.deadline_ticks).max().unwrap(); + for mut packet in outcome.unwrap_or_else(|error| panic!("erased={erased:#b}: {error}")) { + // A recovered packet carries the block's deadline, not its own. + assert_eq!(packet.deadline_ticks, block_deadline, "erased={erased:#b}"); + packet.deadline_ticks = original[(packet.packet_id - 100) as usize].deadline_ticks; + all.push(packet); + } + all.sort_by_key(|packet| packet.packet_id); + assert_eq!(all, original, "erased={erased:#b}"); + } else { + assert!( + matches!(outcome, Err(MediaFecError::BeyondRepair { .. })), + "erased={erased:#b}: {outcome:?}" + ); + } + } +} + +#[test] +fn recovery_reports_how_short_the_block_is() { + let original = frame(9, 8, 37, 37); + let (data, parity) = split(protect_video_frame(&original, &STANDARD).unwrap()); + // 8 data + 2 parity; keep 6 data and 1 parity: 7 present, 8 needed. + let outcome = recover_video_block(&parity[..1], &data[..6]); + assert_eq!( + outcome, + Err(MediaFecError::BeyondRepair { shards_present: 7, shards_needed: 8 }) + ); +} + +#[test] +fn a_recovered_short_final_chunk_has_its_true_length() { + let original = frame(4, 5, 100, 7); + let (mut data, parity) = split(protect_video_frame(&original, &STANDARD).unwrap()); + let lost = data.pop().unwrap(); + assert_eq!(lost.payload.len(), 7); + let recovered = recover_video_block(&parity, &data).unwrap(); + assert_eq!(recovered, vec![lost]); +} + +#[test] +fn duplicate_shards_neither_count_twice_nor_break_recovery() { + let original = frame(5, 4, 50, 50); + let (mut data, parity) = split(protect_video_frame(&original, &STANDARD).unwrap()); + let lost = data.remove(1); + data.push(data[0].clone()); + let mut parity_twice = parity.clone(); + parity_twice.extend(parity); + assert_eq!(recover_video_block(&parity_twice, &data).unwrap(), vec![lost]); +} + +#[test] +fn a_complete_block_recovers_nothing() { + let original = frame(6, 4, 50, 50); + let (data, parity) = split(protect_video_frame(&original, &STANDARD).unwrap()); + assert_eq!(recover_video_block(&parity[..1], &data).unwrap(), Vec::new()); +} + +#[test] +fn recovery_refuses_records_that_are_not_one_block() { + let original = frame(7, 32, 40, 40); + let (data, parity) = split(protect_video_frame(&original, &STANDARD).unwrap()); + let block_zero: Vec<_> = parity.iter().filter(|p| p.block_index == 0).cloned().collect(); + let block_one: Vec<_> = parity.iter().filter(|p| p.block_index == 1).cloned().collect(); + assert_eq!(block_zero.len(), 4); + + let mixed = vec![block_zero[0].clone(), block_one[0].clone()]; + assert!(matches!(recover_video_block(&mixed, &[]), Err(MediaFecError::Invalid(_)))); + + // A chunk from block one handed to block zero's parity. + let stray = data.iter().find(|chunk| chunk.chunk_index == 20).unwrap().clone(); + assert!(matches!(recover_video_block(&block_zero, &[stray]), Err(MediaFecError::Invalid(_)))); + + // A chunk whose length disagrees with the block's. + let mut wrong_length = data[0].clone(); + wrong_length.payload.push(0); + assert!(matches!( + recover_video_block(&block_zero, &[wrong_length]), + Err(MediaFecError::Invalid(_)) + )); + + assert!(matches!(recover_video_block(&[], &data), Err(MediaFecError::Invalid(_)))); +} + +#[test] +fn a_frame_of_several_blocks_recovers_each_block_from_its_own_parity() { + let original = frame(11, 40, 60, 33); + let records = protect_video_frame(&original, &STANDARD).unwrap(); + let sizes = STANDARD.video_block_sizes(40); + assert_eq!(sizes, vec![14, 13, 13]); + let (data, parity) = split(records); + + // Lose the first two data chunks of every block. + let mut lost_indexes = Vec::new(); + let mut start = 0_u16; + for size in &sizes { + lost_indexes.extend([start, start + 1]); + start += size; + } + let held: Vec<_> = data.iter().filter(|c| !lost_indexes.contains(&c.chunk_index)).cloned().collect(); + let mut all = held.clone(); + for block in 0..3_u16 { + let block_parity: Vec<_> = parity.iter().filter(|p| p.block_index == block).cloned().collect(); + let lo = sizes[..usize::from(block)].iter().sum::(); + let block_data: Vec<_> = held + .iter() + .filter(|c| c.chunk_index >= lo && c.chunk_index < lo + sizes[usize::from(block)]) + .cloned() + .collect(); + all.extend(recover_video_block(&block_parity, &block_data).unwrap()); + } + all.sort_by_key(|chunk| chunk.chunk_index); + assert_eq!(all, original); +} + +// --------------------------------------------------------------------------- +// Known answers: the scheme is pinned, not the library +// --------------------------------------------------------------------------- + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|byte| format!("{byte:02x}")).collect() +} + +/// Four 37-byte audio payloads, `payload[i][j] = (i * 37 + j * 7 + 3) mod 256`. +/// If these bytes ever change, `rs-gf256-v1` changed meaning: the matrix, the +/// field polynomial, or the shard order. Bump the scheme id, do not edit them. +#[test] +fn known_answer_audio_parity_bytes() { + let mut block = audio_block(1, 37); + for (i, packet) in block.iter_mut().enumerate() { + packet.payload = (0..37).map(|j| (i * 37 + j * 7 + 3) as u8).collect(); + } + let parity = protect_audio_block(&block, &STANDARD).unwrap(); + assert_eq!(parity[0].fec_scheme, "rs-gf256-v1"); + assert_eq!( + hex(&parity[0].payload), + "f246399711f2ecefa4147ae3856317be83b9a87c0ed13bc552918473bc08e869a91dd98356", + "parity 0" + ); + assert_eq!( + hex(&parity[1].payload), + "a1532c08e588cc05da6175c61b09a704f7d6c7597d76fbef76435656bf65e2331def91dcbf", + "parity 1" + ); +} + +/// The video path pads a short final chunk with zeros before computing parity. +/// Chunks of 5, 5 and 3 bytes, `payload[i][j] = (i * 11 + j * 3 + 1)`. +#[test] +fn known_answer_video_parity_bytes_pad_a_short_final_chunk_with_zeros() { + let mut chunks = frame(2, 3, 5, 3); + for chunk in &mut chunks { + let len = chunk.payload.len(); + chunk.payload = (0..len).map(|j| (usize::from(chunk.chunk_index) * 11 + j * 3 + 1) as u8).collect(); + } + let (_, parity) = split(protect_video_frame(&chunks, &STANDARD).unwrap()); + assert_eq!(parity.len(), 2); + assert_eq!(hex(&parity[0].payload), "1a11081f15", "parity 0"); + assert_eq!(hex(&parity[1].payload), "1d18f3ce8b", "parity 1"); + assert_eq!(parity[0].shard_payload_bytes, 5); + assert_eq!(parity[0].last_chunk_payload_bytes, 3); +} + +#[test] +fn parity_is_a_pure_function_of_its_inputs() { + let original = frame(3, 20, 200, 90); + assert_eq!( + protect_video_frame(&original, &STANDARD).unwrap(), + protect_video_frame(&original, &STANDARD).unwrap() + ); + let block = audio_block(8, 64); + assert_eq!( + protect_audio_block(&block, &STANDARD).unwrap(), + protect_audio_block(&block, &STANDARD).unwrap() + ); +} + +// --------------------------------------------------------------------------- +// One datagram +// --------------------------------------------------------------------------- + +fn provenance() -> MediaWireProvenance<'static> { + MediaWireProvenance { + stored_at: "unix:1700000000000", + runtime_id: "raven-muninn-primary", + role: "muninn.media", + producer: "muninn", + } +} + +#[test] +fn the_datagram_budget_is_the_rudp_fragment_the_media_channel_leaves() { + // 1,472 bytes of UDP payload, less the 36-byte RUDP header and the 5-byte + // channel name "media". + assert_eq!(MEDIA_FEC_MAX_WIRE_BYTES, 1_431); +} + +/// Identifiers are the caller's, so the bound is stated for stream and session +/// identifiers, and codec names, of up to this many bytes each. +const IDENTIFIER_BYTES: usize = 32; + +#[test] +fn every_record_at_the_maximum_shard_size_fits_one_datagram() { + let long = |prefix: &str| format!("{prefix}{}", "x".repeat(IDENTIFIER_BYTES - prefix.len())); + let mut chunks = frame(u64::MAX - 1, 20, MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES, MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES); + for chunk in &mut chunks { + chunk.stream_id = long("stream-"); + chunk.session_id = long("session-"); + chunk.codec = long("codec-"); + } + let records = protect_video_frame(&chunks, &STANDARD).unwrap(); + assert!(records.len() > chunks.len()); + let mut largest = 0; + for record in &records { + let wire = encode_media_wire_record(record, provenance()).unwrap(); + largest = largest.max(wire.len()); + assert!( + wire.len() <= MEDIA_FEC_MAX_WIRE_BYTES, + "{} record is {} bytes, over {MEDIA_FEC_MAX_WIRE_BYTES}", + record.schema_id(), + wire.len() + ); + } + + let mut packets = audio_block(u64::MAX - 8, MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES); + for packet in &mut packets { + packet.stream_id = long("stream-"); + packet.session_id = long("session-"); + packet.codec = long("codec-"); + } + let mut audio: Vec<_> = packets.iter().cloned().map(GameCultMediaWireRecord::Audio).collect(); + audio.extend(protect_audio_block(&packets, &STANDARD).unwrap().into_iter().map(GameCultMediaWireRecord::AudioParity)); + for record in &audio { + let wire = encode_media_wire_record(record, provenance()).unwrap(); + largest = largest.max(wire.len()); + assert!(wire.len() <= MEDIA_FEC_MAX_WIRE_BYTES, "{} is {} bytes", record.schema_id(), wire.len()); + } + // The limit is not slack: the largest record leaves less than a full extra + // identifier's worth of headroom. + assert!( + largest + 96 > MEDIA_FEC_MAX_WIRE_BYTES, + "the shard limit wastes {} bytes of datagram", + MEDIA_FEC_MAX_WIRE_BYTES - largest + ); +} + +#[test] +fn a_shard_over_the_limit_is_refused_at_the_producer_and_at_the_decoder() { + let too_big = MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES + 1; + let chunks = frame(1, 3, too_big, too_big); + assert!(matches!(protect_video_frame(&chunks, &STANDARD), Err(MediaFecError::Invalid(_)))); + assert!(matches!( + protect_audio_block(&audio_block(1, too_big), &STANDARD), + Err(MediaFecError::Invalid(_)) + )); + + let ok = frame(1, 3, MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES, MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES); + let (_, mut parity) = split(protect_video_frame(&ok, &STANDARD).unwrap()); + parity[0].shard_payload_bytes = too_big as u32; + parity[0].payload = vec![0; too_big]; + let wire = encode_media_wire_record(&GameCultMediaWireRecord::VideoParity(parity[0].clone()), provenance()).unwrap(); + assert!(decode_media_wire_record(&wire).is_err()); +} + +// --------------------------------------------------------------------------- +// Framing: what a block may and may not contain +// --------------------------------------------------------------------------- + +#[test] +fn a_block_never_spans_frames() { + let mut chunks = frame(20, 4, 30, 30); + chunks[2].frame_id = 21; + let error = protect_video_frame(&chunks, &STANDARD).unwrap_err(); + assert!(matches!(error, MediaFecError::Invalid(_)), "{error}"); + + let mut chunks = frame(20, 4, 30, 30); + chunks[3].deadline_ticks += 1; + assert!(protect_video_frame(&chunks, &STANDARD).is_err(), "parity would misstate the deadline"); + + let mut chunks = frame(20, 4, 30, 30); + chunks[1].session_id = "session-2".to_string(); + assert!(protect_video_frame(&chunks, &STANDARD).is_err()); +} + +#[test] +fn a_frame_must_be_whole_and_shard_shaped() { + let mut chunks = frame(30, 4, 30, 30); + chunks.pop(); + assert!(protect_video_frame(&chunks, &STANDARD).is_err(), "a missing chunk"); + + let mut chunks = frame(30, 4, 30, 30); + chunks[3] = chunks[2].clone(); + assert!(protect_video_frame(&chunks, &STANDARD).is_err(), "a repeated chunk"); + + let mut chunks = frame(30, 4, 30, 30); + chunks[1].payload.truncate(29); + assert!(protect_video_frame(&chunks, &STANDARD).is_err(), "a short chunk that is not last"); + + let mut chunks = frame(30, 4, 30, 30); + chunks[3].payload.push(0); + assert!(protect_video_frame(&chunks, &STANDARD).is_err(), "a last chunk longer than the shard"); + + assert!(protect_video_frame(&[], &STANDARD).is_err()); +} + +#[test] +fn chunks_may_arrive_in_any_order() { + let original = frame(31, 6, 30, 12); + let mut shuffled = original.clone(); + shuffled.reverse(); + assert_eq!( + protect_video_frame(&original, &STANDARD).unwrap(), + protect_video_frame(&shuffled, &STANDARD).unwrap() + ); +} + +#[test] +fn an_audio_block_is_four_contiguous_equal_length_packets() { + assert!(protect_audio_block(&audio_block(1, 20)[..3], &STANDARD).is_err(), "three packets"); + let mut packets = audio_block(1, 20); + packets[2].packet_id += 1; + assert!(protect_audio_block(&packets, &STANDARD).is_err(), "a gap in packet ids"); + let mut packets = audio_block(1, 20); + packets[1].pts_ticks += 1; + assert!(protect_audio_block(&packets, &STANDARD).is_err(), "a gap in presentation time"); + let mut packets = audio_block(1, 20); + packets[3].payload.push(0); + assert!(protect_audio_block(&packets, &STANDARD).is_err(), "unequal payload lengths"); + let mut packets = audio_block(1, 20); + packets[0].codec = "pcm".to_string(); + assert!(protect_audio_block(&packets, &STANDARD).is_err(), "mixed codecs"); +} + +#[test] +fn parity_carries_its_frames_deadline_unchanged() { + let original = frame(40, 40, 50, 50); + let (_, parity) = split(protect_video_frame(&original, &STANDARD).unwrap()); + assert!(!parity.is_empty()); + for shard in &parity { + assert_eq!(shard.deadline_ticks, original[0].deadline_ticks); + assert_eq!(shard.pts_ticks, original[0].pts_ticks); + assert_eq!(shard.frame_id, 40); + assert_eq!(shard.fec_scheme, MEDIA_FEC_SCHEME_RS_GF256_V1); + } +} + +#[test] +fn audio_parity_carries_the_latest_deadline_in_its_block_and_recovery_uses_it() { + let mut packets = audio_block(50, 30); + packets[1].deadline_ticks = 99_999; // the block's latest, not its last + let parity = protect_audio_block(&packets, &STANDARD).unwrap(); + assert!(parity.iter().all(|shard| shard.deadline_ticks == 99_999)); + + let held = vec![packets[0].clone(), packets[2].clone(), packets[3].clone()]; + let recovered = recover_audio_block(&parity[..1], &held).unwrap(); + assert_eq!(recovered.len(), 1); + let packet = &recovered[0]; + assert_eq!(packet.packet_id, 51, "id from the block's arithmetic"); + assert_eq!(packet.pts_ticks, packets[1].pts_ticks, "pts from the block's arithmetic"); + assert_eq!(packet.deadline_ticks, 99_999, "deadline from the block"); + assert_eq!(packet.payload, packets[1].payload); + assert_eq!(packet.codec, "opus"); +} + +// --------------------------------------------------------------------------- +// Send order: a burst must land as one loss per block +// --------------------------------------------------------------------------- + +#[test] +fn send_order_is_a_permutation_of_the_frames_records() { + for n in [1_u16, 2, 15, 16, 17, 31, 32, 33, 47, 100] { + let original = frame(60, n, 20, 9); + let records = protect_video_frame(&original, &STANDARD).unwrap(); + let sizes = STANDARD.video_block_sizes(n); + let expected_parity: usize = sizes.iter().map(|&k| usize::from(STANDARD.video_parity_shards(k))).sum(); + assert_eq!(records.len(), usize::from(n) + expected_parity, "n = {n}"); + let mut data_indexes: Vec = records + .iter() + .filter_map(|r| match r { + GameCultMediaWireRecord::Video(c) => Some(c.chunk_index), + _ => None, + }) + .collect(); + data_indexes.sort_unstable(); + assert_eq!(data_indexes, (0..n).collect::>(), "n = {n}"); + } +} + +#[test] +fn equal_blocks_send_one_record_per_block_before_any_block_repeats() { + for n in [32_u16, 48, 64, 24, 20, 18] { + let sizes = STANDARD.video_block_sizes(n); + let blocks = sizes.len(); + assert!(sizes.windows(2).all(|pair| pair[0] == pair[1]), "n = {n} must split equally"); + let records = protect_video_frame(&frame(70, n, 20, 20), &STANDARD).unwrap(); + let order: Vec = records.iter().map(|r| block_of(r, &sizes)).collect(); + for window in order.windows(blocks) { + let mut seen = window.to_vec(); + seen.sort_unstable(); + seen.dedup(); + assert_eq!(seen.len(), blocks, "n = {n}: a window of {blocks} repeats a block: {order:?}"); + } + } +} + +/// Blocks that differ in size run out at the end and cannot keep the full +/// spacing. What holds: two records of one block are never closer than the +/// number of blocks still sending from the later one onward. +#[test] +fn unequal_blocks_keep_every_block_spaced_by_the_blocks_still_sending() { + for n in [17_u16, 19, 35, 40, 50, 100] { + let sizes = STANDARD.video_block_sizes(n); + assert!(sizes.windows(2).any(|pair| pair[0] != pair[1]), "n = {n} must split unequally"); + let records = protect_video_frame(&frame(71, n, 20, 20), &STANDARD).unwrap(); + let order: Vec = records.iter().map(|r| block_of(r, &sizes)).collect(); + for later in 0..order.len() { + let Some(earlier) = (0..later).rev().find(|&i| order[i] == order[later]) else { continue }; + let mut still_sending = order[later..].to_vec(); + still_sending.sort_unstable(); + still_sending.dedup(); + assert!( + later - earlier >= still_sending.len(), + "n = {n}: block {} sent at {earlier} and again at {later} with {} blocks still sending: {order:?}", + order[later], + still_sending.len() + ); + } + } +} + +// --------------------------------------------------------------------------- +// The wire round trip, and the policy +// --------------------------------------------------------------------------- + +#[test] +fn a_frame_survives_the_wire_and_recovers_from_what_arrives() { + let original = frame(80, 20, 300, 111); + let sizes = STANDARD.video_block_sizes(20); + let records = protect_video_frame(&original, &STANDARD).unwrap(); + + // The network eats a burst of four consecutive records. + let arrived: Vec = records + .iter() + .enumerate() + .filter(|(slot, _)| !(9..13).contains(slot)) + .map(|(_, record)| { + let wire = encode_media_wire_record(record, provenance()).unwrap(); + decode_media_wire_record(&wire).unwrap() + }) + .collect(); + + let mut all = Vec::new(); + for block in 0..sizes.len() { + let mut data = Vec::new(); + let mut parity = Vec::new(); + for record in arrived.iter().filter(|r| block_of(r, &sizes) == block) { + match record { + GameCultMediaWireRecord::Video(c) => data.push(c.clone()), + GameCultMediaWireRecord::VideoParity(p) => parity.push(p.clone()), + _ => unreachable!(), + } + } + let recovered = recover_video_block(&parity, &data).expect("a burst of 4 over 2 blocks is 2 per block"); + all.extend(data); + all.extend(recovered); + } + all.sort_by_key(|chunk| chunk.chunk_index); + assert_eq!(all, original); +} + +#[test] +fn a_custom_policy_sizes_blocks_by_its_own_fields() { + let policy = MediaFecPolicy { + video_max_block_data_shards: 8, + video_min_parity_shards: 1, + video_parity_divisor: 2, + ..STANDARD + }; + assert_eq!(policy.video_block_sizes(20), vec![7, 7, 6]); + assert_eq!(policy.video_parity_shards(7), 4); + assert_eq!(policy.video_parity_shards(1), 1); + let (data, parity) = split(protect_video_frame(&frame(90, 20, 20, 20), &policy).unwrap()); + assert_eq!(data.len(), 20); + assert_eq!(parity.len(), 4 + 4 + 3); + assert_eq!(MediaFecPolicy::default(), STANDARD); +} + +#[test] +fn a_policy_that_cannot_form_a_block_is_refused() { + let zero = MediaFecPolicy { video_min_parity_shards: 0, ..STANDARD }; + assert!(protect_video_frame(&frame(1, 2, 10, 10), &zero).is_err()); + let too_wide = MediaFecPolicy { video_max_block_data_shards: 255, ..STANDARD }; + assert!(protect_video_frame(&frame(1, 2, 10, 10), &too_wide).is_err()); + let audio_too_wide = MediaFecPolicy { audio_data_shards: 255, audio_parity_shards: 2, ..STANDARD }; + assert!(protect_audio_block(&audio_block(1, 10), &audio_too_wide).is_err()); +} + +#[test] +fn parity_records_survive_the_positional_encoding_unchanged() { + let (_, parity) = split(protect_video_frame(&frame(5, 5, 40, 17), &STANDARD).unwrap()); + let audio = protect_audio_block(&audio_block(5, 25), &STANDARD).unwrap(); + for shard in parity { + let bytes = rmp_serde::to_vec(&shard).unwrap(); + assert_eq!(rmp_serde::from_slice::(&bytes).unwrap(), shard); + } + for shard in audio { + let bytes = rmp_serde::to_vec(&shard).unwrap(); + assert_eq!(rmp_serde::from_slice::(&bytes).unwrap(), shard); + } +} diff --git a/packages/cultnet-rs/tests/media_stream_contracts.rs b/packages/cultnet-rs/tests/media_stream_contracts.rs index 963d78a2..08db4b30 100644 --- a/packages/cultnet-rs/tests/media_stream_contracts.rs +++ b/packages/cultnet-rs/tests/media_stream_contracts.rs @@ -5,9 +5,10 @@ use cultcache_rs::DatabaseEntry; use cultnet_rs::{ - GAMECULT_MEDIA_AUDIO_PACKET_SCHEMA, GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA, - GAMECULT_MEDIA_VIDEO_ACCESS_UNIT_SCHEMA, GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA, - GameCultMediaAudioPacketRecord, GameCultMediaReceiverFeedbackRecord, + GAMECULT_MEDIA_AUDIO_PACKET_SCHEMA, GAMECULT_MEDIA_AUDIO_PARITY_SHARD_SCHEMA, + GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA, GAMECULT_MEDIA_VIDEO_ACCESS_UNIT_SCHEMA, + GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA, GameCultMediaAudioPacketRecord, + GameCultMediaAudioParityShardRecord, GameCultMediaReceiverFeedbackRecord, GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord, }; use rmpv::Value; @@ -129,9 +130,14 @@ fn parity_shards_carry_the_geometry_recovery_needs() { dependency_frame_id: Some(4241), deadline_ticks: 108_000, chunk_count: 5, + fec_scheme: "rs-gf256-v1".to_string(), + block_index: 0, + block_count: 1, + block_data_start: 0, + block_data_count: 5, parity_index: 1, parity_count: 2, - chunk_payload_bytes: 1024, + shard_payload_bytes: 1024, last_chunk_payload_bytes: 512, payload: vec![9, 9, 9], }; @@ -141,13 +147,21 @@ fn parity_shards_carry_the_geometry_recovery_needs() { panic!("parity shards must encode as a positional array"); }; - assert_eq!(slots.len(), 17, "field count is wire contract"); - assert_eq!(slots[12].as_u64(), Some(1), "slot 12 is parity_index"); - assert_eq!(slots[13].as_u64(), Some(2), "slot 13 is parity_count"); + assert_eq!(slots.len(), 22, "field count is wire contract"); + assert_eq!(slots[12].as_str(), Some("rs-gf256-v1"), "slot 12 is fec_scheme"); + assert_eq!(slots[13].as_u64(), Some(0), "slot 13 is block_index"); + assert_eq!(slots[16].as_u64(), Some(5), "slot 16 is block_data_count (k)"); + assert_eq!(slots[17].as_u64(), Some(1), "slot 17 is parity_index"); + assert_eq!(slots[18].as_u64(), Some(2), "slot 18 is parity_count (m)"); assert_eq!( - slots[15].as_u64(), + slots[20].as_u64(), Some(512), - "slot 15 is last_chunk_payload_bytes, which recovery needs to size the tail" + "slot 20 is last_chunk_payload_bytes, which recovery needs to size the tail" + ); + assert_eq!( + slots[21].as_slice(), + Some([9, 9, 9].as_slice()), + "the parity payload is bin, not an array of integers" ); let decoded: GameCultMediaVideoParityShardRecord = @@ -155,6 +169,42 @@ fn parity_shards_carry_the_geometry_recovery_needs() { assert_eq!(decoded, shard); } +#[test] +fn audio_parity_shards_carry_the_geometry_recovery_needs() { + let shard = GameCultMediaAudioParityShardRecord { + stream_id: "raven-primary-av".to_string(), + session_id: "session-1".to_string(), + codec: "opus".to_string(), + fec_scheme: "rs-gf256-v1".to_string(), + base_packet_id: 77, + base_pts_ticks: 960, + packet_duration_ticks: 960, + timebase_num: 1, + timebase_den: 48_000, + deadline_ticks: 4_800, + data_shard_count: 4, + parity_index: 1, + parity_shard_count: 2, + shard_payload_bytes: 3, + payload: vec![7, 8, 9], + }; + + let encoded = rmp_serde::to_vec(&shard).expect("encodes"); + let Value::Array(slots) = decode_as_value(&encoded) else { + panic!("audio parity shards must encode as a positional array"); + }; + + assert_eq!(slots.len(), 15, "field count is wire contract"); + assert_eq!(slots[4].as_u64(), Some(77), "slot 4 is base_packet_id"); + assert_eq!(slots[10].as_u64(), Some(4), "slot 10 is data_shard_count (k)"); + assert_eq!(slots[12].as_u64(), Some(2), "slot 12 is parity_shard_count (m)"); + assert_eq!(slots[14].as_slice(), Some([7, 8, 9].as_slice()), "the payload is bin"); + + let decoded: GameCultMediaAudioParityShardRecord = + rmp_serde::from_slice(&encoded).expect("round-trips"); + assert_eq!(decoded, shard); +} + /// `missing_video_chunk_keys` was appended after producers were already writing /// feedback. A record written without it must still decode, or the append was a /// breaking change wearing a `default` attribute. @@ -205,6 +255,11 @@ fn published_schema_constants_match_the_derived_identity() { GameCultMediaAudioPacketRecord::SCHEMA_NAME, GAMECULT_MEDIA_AUDIO_PACKET_SCHEMA ); + assert_eq!( + GameCultMediaAudioParityShardRecord::SCHEMA_NAME, + GAMECULT_MEDIA_AUDIO_PARITY_SHARD_SCHEMA + ); + assert_eq!(GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA, "gamecult.media_video_parity_shard.v3"); assert_eq!( GameCultMediaReceiverFeedbackRecord::SCHEMA_NAME, GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA diff --git a/packages/cultnet-rs/tests/media_stream_wire.rs b/packages/cultnet-rs/tests/media_stream_wire.rs index cee31a04..8137a7cc 100644 --- a/packages/cultnet-rs/tests/media_stream_wire.rs +++ b/packages/cultnet-rs/tests/media_stream_wire.rs @@ -2,7 +2,8 @@ //! nothing on the wire. use cultnet_rs::{ - GAMECULT_MEDIA_CHANNEL, GameCultMediaAudioPacketRecord, GameCultMediaReceiverFeedbackRecord, + GAMECULT_MEDIA_CHANNEL, GameCultMediaAudioPacketRecord, GameCultMediaAudioParityShardRecord, + GameCultMediaReceiverFeedbackRecord, GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord, GameCultMediaWireRecord, MediaWireProvenance, decode_media_wire_record, encode_media_wire_record, @@ -116,12 +117,34 @@ fn every_variant_round_trips() { 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, - chunk_payload_bytes: 848, - last_chunk_payload_bytes: 400, + shard_payload_bytes: 3, + last_chunk_payload_bytes: 2, payload: vec![1, 2, 3], }), + GameCultMediaWireRecord::AudioParity(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: 960, + timebase_num: 1, + timebase_den: 48_000, + deadline_ticks: 4_000, + data_shard_count: 4, + parity_index: 0, + parity_shard_count: 2, + shard_payload_bytes: 3, + payload: vec![4, 5, 6], + }), GameCultMediaWireRecord::Audio(GameCultMediaAudioPacketRecord { stream_id: "s".to_string(), session_id: "x".to_string(), @@ -215,3 +238,148 @@ fn provenance_must_be_complete() { fn the_media_channel_is_named_once() { assert_eq!(GAMECULT_MEDIA_CHANNEL, "media"); } + +fn video_parity() -> GameCultMediaVideoParityShardRecord { + GameCultMediaVideoParityShardRecord { + stream_id: "s".to_string(), + session_id: "x".to_string(), + frame_id: 7, + codec: "h264".to_string(), + pts_ticks: 0, + duration_ticks: 1, + timebase_num: 1, + timebase_den: 90_000, + keyframe: false, + dependency_frame_id: None, + deadline_ticks: 1, + chunk_count: 20, + fec_scheme: "rs-gf256-v1".to_string(), + block_index: 1, + block_count: 2, + block_data_start: 10, + block_data_count: 10, + parity_index: 1, + parity_count: 3, + shard_payload_bytes: 4, + last_chunk_payload_bytes: 2, + payload: vec![1, 2, 3, 4], + } +} + +fn audio_parity() -> 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: 960, + timebase_num: 1, + timebase_den: 48_000, + deadline_ticks: 4_000, + data_shard_count: 4, + parity_index: 1, + parity_shard_count: 2, + shard_payload_bytes: 3, + payload: vec![4, 5, 6], + } +} + +fn decodes(record: GameCultMediaWireRecord) -> Result { + let wire = encode_media_wire_record(&record, provenance()).map_err(|error| error.to_string())?; + decode_media_wire_record(&wire).map_err(|error| error.to_string()) +} + +/// A parity key names the block as well as the shard, or two blocks' parity +/// index 0 would share an address. +#[test] +fn parity_record_keys_name_the_block_and_the_shard() { + let mut other_block = video_parity(); + other_block.block_index = 0; + let mut other_shard = video_parity(); + other_shard.parity_index = 0; + let keys = [ + GameCultMediaWireRecord::VideoParity(video_parity()).record_key(), + GameCultMediaWireRecord::VideoParity(other_block).record_key(), + GameCultMediaWireRecord::VideoParity(other_shard).record_key(), + ]; + assert_eq!(keys[0], "s:x:video-parity:7:1:1"); + assert_ne!(keys[0], keys[1]); + assert_ne!(keys[0], keys[2]); + + let audio = GameCultMediaWireRecord::AudioParity(audio_parity()).record_key(); + assert_eq!(audio, "s:x:audio-parity:8:1"); + let mut other = audio_parity(); + other.parity_index = 0; + assert_ne!(audio, GameCultMediaWireRecord::AudioParity(other).record_key()); +} + +#[test] +fn a_well_formed_parity_record_is_admitted() { + assert!(decodes(GameCultMediaWireRecord::VideoParity(video_parity())).is_ok()); + assert!(decodes(GameCultMediaWireRecord::AudioParity(audio_parity())).is_ok()); +} + +/// The decoder is where a consumer's idea of a well-formed shard is enforced. +/// Each corruption below would otherwise reach the recovery arithmetic. +#[test] +fn a_malformed_video_parity_record_is_refused_with_its_reason() { + let cases: Vec<(&str, Box)> = vec![ + ("fec_scheme", Box::new(|r| r.fec_scheme = "xor-v0".to_string())), + ("block_index", Box::new(|r| r.block_index = 2)), + ("block_index", Box::new(|r| r.block_count = 0)), + ("inside the frame", Box::new(|r| r.block_data_start = 11)), + ("inside the frame", Box::new(|r| r.block_data_count = 0)), + ("parity_index", Box::new(|r| r.parity_index = 3)), + ("parity_index", Box::new(|r| r.parity_count = 0)), + ("256 shards", Box::new(|r| { + r.block_data_count = 254; + r.chunk_count = 300; + r.parity_count = 3; + })), + ("shard_payload_bytes", Box::new(|r| { + r.shard_payload_bytes = 0; + r.payload.clear(); + })), + ("shard_payload_bytes", Box::new(|r| { + r.shard_payload_bytes = 721; + r.payload = vec![0; 721]; + })), + ("last chunk length", Box::new(|r| r.last_chunk_payload_bytes = 0)), + ("last chunk length", Box::new(|r| r.last_chunk_payload_bytes = 5)), + ("declared shard length", Box::new(|r| r.payload.push(0))), + ("chunk_count", Box::new(|r| r.chunk_count = 0)), + ("deadline_ticks", Box::new(|r| r.deadline_ticks = -1)), + ]; + for (expected, corrupt) in cases { + let mut record = video_parity(); + corrupt(&mut record); + let error = decodes(GameCultMediaWireRecord::VideoParity(record)).unwrap_err(); + assert!(error.contains(expected), "expected {expected:?} in {error:?}"); + } +} + +#[test] +fn a_malformed_audio_parity_record_is_refused_with_its_reason() { + let cases: Vec<(&str, Box)> = vec![ + ("fec_scheme", Box::new(|r| r.fec_scheme = String::new())), + ("stripe", Box::new(|r| r.parity_index = 2)), + ("stripe", Box::new(|r| r.data_shard_count = 0)), + ("stripe", Box::new(|r| r.data_shard_count = 255)), + ("timing", Box::new(|r| r.packet_duration_ticks = 0)), + ("base_pts_ticks", Box::new(|r| r.deadline_ticks = -1)), + ("shard_payload_bytes", Box::new(|r| { + r.shard_payload_bytes = 721; + r.payload = vec![0; 721]; + })), + ("declared shard length", Box::new(|r| { r.payload.pop(); })), + ("codec", Box::new(|r| r.codec = String::new())), + ]; + for (expected, corrupt) in cases { + let mut record = audio_parity(); + corrupt(&mut record); + let error = decodes(GameCultMediaWireRecord::AudioParity(record)).unwrap_err(); + assert!(error.contains(expected), "expected {expected:?} in {error:?}"); + } +}