diff --git a/packages/cultnet-rs/src/media_stream_wire.rs b/packages/cultnet-rs/src/media_stream_wire.rs index 2959eef8..7083d927 100644 --- a/packages/cultnet-rs/src/media_stream_wire.rs +++ b/packages/cultnet-rs/src/media_stream_wire.rs @@ -167,19 +167,281 @@ pub fn decode_media_wire_record(payload: &[u8]) -> Result Ok(GameCultMediaWireRecord::Video( - rmp_serde::from_slice(&document.payload)?, - )), - GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA => Ok(GameCultMediaWireRecord::VideoParity( - rmp_serde::from_slice(&document.payload)?, - )), - GAMECULT_MEDIA_AUDIO_PACKET_SCHEMA => Ok(GameCultMediaWireRecord::Audio( - rmp_serde::from_slice(&document.payload)?, - )), - GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA => Ok(GameCultMediaWireRecord::Feedback( - rmp_serde::from_slice(&document.payload)?, - )), - other => Err(anyhow!("unknown media schema {other}")), + let record = match document.schema_id.as_str() { + GAMECULT_MEDIA_VIDEO_ACCESS_UNIT_SCHEMA => { + let record: GameCultMediaVideoAccessUnitRecord = + rmp_serde::from_slice(&document.payload)?; + validate_video_record(&record)?; + GameCultMediaWireRecord::Video(record) + } + GAMECULT_MEDIA_VIDEO_PARITY_SHARD_SCHEMA => { + let record: GameCultMediaVideoParityShardRecord = + rmp_serde::from_slice(&document.payload)?; + validate_video_parity_record(&record)?; + GameCultMediaWireRecord::VideoParity(record) + } + GAMECULT_MEDIA_AUDIO_PACKET_SCHEMA => { + let record: GameCultMediaAudioPacketRecord = + rmp_serde::from_slice(&document.payload)?; + validate_audio_record(&record)?; + GameCultMediaWireRecord::Audio(record) + } + GAMECULT_MEDIA_RECEIVER_FEEDBACK_SCHEMA => { + let record: GameCultMediaReceiverFeedbackRecord = + rmp_serde::from_slice(&document.payload)?; + validate_feedback_record(&record)?; + GameCultMediaWireRecord::Feedback(record) + } + other => return Err(anyhow!("unsupported media schema {other}")), + }; + + // The key is derived from the record's own contents, so a mismatch means the + // envelope and its payload disagree about what this is. Admitting it would + // let a consumer address media by a key its producer never meant. + let expected = record.record_key(); + if document.record_key != expected { + return Err(anyhow!( + "record key mismatch: envelope says {}, record derives {expected}", + document.record_key + )); + } + + Ok(record) +} + +// --------------------------------------------------------------------------- +// Validation +// +// Moved here from Muninn with the envelope. A decoder that does not validate +// leaves each consumer to invent its own idea of a well-formed record, which is +// how the C++ receiver and the Rust sender came to disagree while both looked +// correct in isolation. Muninn's tests for these came too; they now cover the +// contract rather than one producer's copy of it. +// --------------------------------------------------------------------------- + +/// Addresses one chunk of a video frame. The string form appears in receiver +/// feedback and is parsed by consumers, so its shape is contract. +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub struct VideoChunkKey { + pub frame_id: u64, + pub chunk_index: u16, +} + +impl VideoChunkKey { + pub fn new(frame_id: u64, chunk_index: u16) -> Self { + Self { + frame_id, + chunk_index, + } + } + + pub fn parse(value: &str) -> Result { + if value.matches(':').count() != 1 { + return Err(anyhow!( + "video chunk key must contain exactly one ':' separator" + )); + } + let (frame_id, chunk_index) = value + .split_once(':') + .ok_or_else(|| anyhow!("video chunk key must be ':'"))?; + Ok(Self { + frame_id: frame_id + .parse() + .map_err(|_| anyhow!("video chunk key frame_id must be an integer"))?, + chunk_index: chunk_index + .parse() + .map_err(|_| anyhow!("video chunk key chunk_index must be an integer"))?, + }) + } + + pub fn as_feedback_key(&self) -> String { + format!("{}:{}", self.frame_id, self.chunk_index) } } + +pub fn video_chunk_feedback_key(frame_id: u64, chunk_index: u16) -> String { + VideoChunkKey::new(frame_id, chunk_index).as_feedback_key() +} + +pub fn normalize_video_chunk_feedback_keys(keys: Vec) -> Result> { + let mut parsed = keys + .iter() + .map(|key| VideoChunkKey::parse(key)) + .collect::>>()?; + parsed.sort_unstable(); + parsed.dedup(); + Ok(parsed + .into_iter() + .map(|key| key.as_feedback_key()) + .collect()) +} + +pub fn validate_video_record(record: &GameCultMediaVideoAccessUnitRecord) -> Result<()> { + if record.stream_id.is_empty() { + return Err(anyhow!("video media record stream_id must be non-empty")); + } + if record.session_id.is_empty() { + return Err(anyhow!("video media record session_id must be non-empty")); + } + if record.codec.is_empty() { + return Err(anyhow!("video media record codec must be non-empty")); + } + if record.timebase_num == 0 || record.timebase_den == 0 { + return Err(anyhow!("video media record timebase must be non-zero")); + } + if record.duration_ticks == 0 { + return Err(anyhow!( + "video media record duration_ticks must be greater than zero" + )); + } + if record.deadline_ticks < record.pts_ticks { + return Err(anyhow!( + "video media record deadline_ticks must not precede pts_ticks" + )); + } + if record.chunk_count == 0 { + return Err(anyhow!("video media record chunk_count must be non-zero")); + } + if record.chunk_index >= record.chunk_count { + return Err(anyhow!( + "video media record chunk_index {} is outside chunk_count {}", + record.chunk_index, + record.chunk_count + )); + } + if record.payload.is_empty() { + return Err(anyhow!("video media record payload must be non-empty")); + } + Ok(()) +} + +pub fn validate_video_parity_record(record: &GameCultMediaVideoParityShardRecord) -> Result<()> { + if record.stream_id.is_empty() { + return Err(anyhow!( + "video parity media record stream_id must be non-empty" + )); + } + if record.session_id.is_empty() { + return Err(anyhow!( + "video parity media record session_id must be non-empty" + )); + } + if record.codec.is_empty() { + return Err(anyhow!("video parity media record codec must be non-empty")); + } + if record.timebase_num == 0 || record.timebase_den == 0 { + return Err(anyhow!( + "video parity media record timebase must be non-zero" + )); + } + if record.duration_ticks == 0 { + return Err(anyhow!( + "video parity media record duration_ticks must be greater than zero" + )); + } + if record.deadline_ticks < record.pts_ticks { + return Err(anyhow!( + "video parity media record deadline_ticks must not precede pts_ticks" + )); + } + if record.chunk_count == 0 { + return Err(anyhow!( + "video parity media record chunk_count must be non-zero" + )); + } + if record.parity_count == 0 || record.parity_index >= record.parity_count { + return Err(anyhow!( + "video parity media record has invalid parity stripe metadata" + )); + } + if record.parity_count > record.chunk_count { + return Err(anyhow!( + "video parity media record parity_count exceeds chunk_count" + )); + } + if record.chunk_payload_bytes == 0 || record.last_chunk_payload_bytes == 0 { + return Err(anyhow!( + "video parity media record chunk lengths must be non-zero" + )); + } + if record.last_chunk_payload_bytes > record.chunk_payload_bytes { + return Err(anyhow!( + "video parity media record last chunk length exceeds regular chunk length" + )); + } + if record.payload.is_empty() { + return Err(anyhow!( + "video parity media record payload must be non-empty" + )); + } + if record.payload.len() > record.chunk_payload_bytes as usize { + return Err(anyhow!( + "video parity media record payload exceeds declared chunk length" + )); + } + Ok(()) +} + +pub fn validate_audio_record(record: &GameCultMediaAudioPacketRecord) -> Result<()> { + if record.stream_id.is_empty() { + return Err(anyhow!("audio media record stream_id must be non-empty")); + } + if record.session_id.is_empty() { + return Err(anyhow!("audio media record session_id must be non-empty")); + } + if record.codec.is_empty() { + return Err(anyhow!("audio media record codec must be non-empty")); + } + if record.timebase_num == 0 || record.timebase_den == 0 { + return Err(anyhow!("audio media record timebase must be non-zero")); + } + if record.duration_ticks == 0 { + return Err(anyhow!( + "audio media record duration_ticks must be greater than zero" + )); + } + if record.deadline_ticks < record.pts_ticks { + return Err(anyhow!( + "audio media record deadline_ticks must not precede pts_ticks" + )); + } + if record.payload.is_empty() { + return Err(anyhow!("audio media record payload must be non-empty")); + } + Ok(()) +} + +pub fn validate_feedback_record(record: &GameCultMediaReceiverFeedbackRecord) -> Result<()> { + if record.stream_id.is_empty() { + return Err(anyhow!( + "receiver feedback media record stream_id must be non-empty" + )); + } + if record.session_id.is_empty() { + return Err(anyhow!( + "receiver feedback media record session_id must be non-empty" + )); + } + if record.receiver_id.is_empty() { + return Err(anyhow!( + "receiver feedback media record receiver_id must be non-empty" + )); + } + if record.observed_at.is_empty() { + return Err(anyhow!( + "receiver feedback media record observed_at must be non-empty" + )); + } + if record.jitter_us < 0 { + return Err(anyhow!( + "receiver feedback media record jitter_us must be non-negative" + )); + } + if record.decode_queue_us < 0 { + return Err(anyhow!( + "receiver feedback media record decode_queue_us must be non-negative" + )); + } + normalize_video_chunk_feedback_keys(record.missing_video_chunk_keys.clone())?; + Ok(()) +}