Add the media FEC codec, video parity v3 and audio parity records to cultnet-rs
The codec owns the erasure math for both ends: protect_video_frame and protect_audio_block on the producer, recover_video_block and recover_audio_block on the consumer. Each block is systematic Reed-Solomon over GF(2^8) (reed-solomon-erasure 6, default features), named on every parity record by the scheme id rs-gf256-v1. Any <= m erasures recover exactly; more report BeyondRepair. Video frames split into ceil(n/16) near-equal blocks with m = max(2, ceil(k/4)); audio is 4+2. A frame's records come back in send order, rotating one record from each block so a burst lands as one loss per block. Deleted first: gamecult.media_video_parity_shard.v2 and its XOR-stripe validator, and the reliable "audio" channel (profile entry, send-options arm, constant, tests). Audio now rides "media" with parity. Added: gamecult.media_video_parity_shard.v3 (block geometry, scheme id, shard length), gamecult.media_audio_parity_shard.v1, the AudioParity wire variant with validation, and the parity record key now names the block. Every shard is bounded at MEDIA_FEC_MAX_SHARD_PAYLOAD_BYTES (720) so its wrapped record fits MEDIA_FEC_MAX_WIRE_BYTES (1431), measured with identifiers of up to 32 bytes. Tests pin: exhaustive recovery for every shipped (k, m) and 4+2, m+1 refusal, known-answer parity bytes for both paths, the datagram bound on every record kind, send order, deadline carriage and decoder validation. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
5adbcaa0c8
commit
f7c58ce5bc
10 changed files with 1818 additions and 75 deletions
|
|
@ -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"] }
|
||||
|
|
|
|||
|
|
@ -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::*;
|
||||
|
|
|
|||
655
packages/cultnet-rs/src/media_fec.rs
Normal file
655
packages/cultnet-rs/src/media_fec.rs
Normal file
|
|
@ -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<u16> {
|
||||
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<String>) -> 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<u8>], parity_count: usize) -> Result<Vec<Vec<u8>>, 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<Option<Vec<u8>>>,
|
||||
) -> Result<Vec<Vec<u8>>, 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<u8> {
|
||||
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<Vec<GameCultMediaWireRecord>, 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<u16, &GameCultMediaVideoAccessUnitRecord> = 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<GameCultMediaWireRecord>> = 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<Vec<u8>> = 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<GameCultMediaWireRecord> = 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<T>(data: Vec<T>, parity: Vec<T>) -> Vec<T> {
|
||||
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<T>(lanes: Vec<Vec<T>>) -> Vec<T> {
|
||||
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<Vec<GameCultMediaVideoAccessUnitRecord>, 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<Option<Vec<u8>>> = 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<usize> = (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<Vec<GameCultMediaAudioParityShardRecord>, 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<Vec<u8>> = 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<Vec<GameCultMediaAudioPacketRecord>, 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<Option<Vec<u8>>> = 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<usize> = (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
|
||||
}
|
||||
|
|
@ -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<u8>,
|
||||
}
|
||||
|
||||
/// 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<u8>,
|
||||
}
|
||||
|
||||
|
|
@ -541,6 +563,60 @@ pub struct GameCultMediaAudioPacketRecord {
|
|||
pub payload: Vec<u8>,
|
||||
}
|
||||
|
||||
/// 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<u8>,
|
||||
}
|
||||
|
||||
/// 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.
|
||||
|
|
|
|||
|
|
@ -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<GameCultMediaWireRecor
|
|||
validate_audio_record(&record)?;
|
||||
GameCultMediaWireRecord::Audio(record)
|
||||
}
|
||||
GAMECULT_MEDIA_AUDIO_PARITY_SHARD_SCHEMA => {
|
||||
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(())
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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::<Vec<_>>();
|
||||
|
||||
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"
|
||||
|
|
|
|||
723
packages/cultnet-rs/tests/media_fec.rs
Normal file
723
packages/cultnet-rs/tests/media_fec.rs
Normal file
|
|
@ -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<u8> {
|
||||
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<GameCultMediaVideoAccessUnitRecord> {
|
||||
(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<GameCultMediaAudioPacketRecord> {
|
||||
(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<GameCultMediaWireRecord>,
|
||||
) -> (Vec<GameCultMediaVideoAccessUnitRecord>, Vec<GameCultMediaVideoParityShardRecord>) {
|
||||
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::<u16>(), 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::<u16>();
|
||||
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<u16> = 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::<Vec<_>>(), "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<usize> = 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<usize> = 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<GameCultMediaWireRecord> = 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::<GameCultMediaVideoParityShardRecord>(&bytes).unwrap(), shard);
|
||||
}
|
||||
for shard in audio {
|
||||
let bytes = rmp_serde::to_vec(&shard).unwrap();
|
||||
assert_eq!(rmp_serde::from_slice::<GameCultMediaAudioParityShardRecord>(&bytes).unwrap(), shard);
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<GameCultMediaWireRecord, String> {
|
||||
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<dyn Fn(&mut GameCultMediaVideoParityShardRecord)>)> = 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<dyn Fn(&mut GameCultMediaAudioParityShardRecord)>)> = 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:?}");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue