Validate on decode, where both ends inherit it
Moving the envelope moved less than it should have. Muninn's decoder validated
every record and checked the envelope's key against the key derived from the
record's own contents; the version that landed here did neither, so the move was
quietly a downgrade. Six of Muninn's tests failed on the switch and said so,
which is the only reason this is a paragraph rather than a bug.
The four record validators and the video chunk key move here with it. They check
the contract, not one producer's use of it: non-empty identifiers, non-zero
timebases, sane pressure figures, well-formed chunk keys. Left in Muninn they
would have protected the producer and left every consumer to invent its own idea
of a well-formed record — which is how a C++ receiver and a Rust sender came to
disagree while each looked correct alone.
The record-key check is the sharper one. A key is derived from the record it
addresses, so a mismatch means envelope and payload disagree about what the
thing is. Admitting that would let a consumer address media by a key its
producer never meant.
VideoChunkKey comes too. Its string form ("42:1") travels in receiver feedback
and is parsed at the other end, so it was already contract living in a producer.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
a58968e7ac
commit
96138ffcfa
1 changed files with 276 additions and 14 deletions
|
|
@ -167,19 +167,281 @@ pub fn decode_media_wire_record(payload: &[u8]) -> Result<GameCultMediaWireRecor
|
|||
return Err(anyhow!("media document payload must be MessagePack"));
|
||||
}
|
||||
|
||||
match document.schema_id.as_str() {
|
||||
GAMECULT_MEDIA_VIDEO_ACCESS_UNIT_SCHEMA => 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<Self> {
|
||||
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 '<frame_id>:<chunk_index>'"))?;
|
||||
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<String>) -> Result<Vec<String>> {
|
||||
let mut parsed = keys
|
||||
.iter()
|
||||
.map(|key| VideoChunkKey::parse(key))
|
||||
.collect::<Result<Vec<_>>>()?;
|
||||
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(())
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue