Merge media FEC Cut 3: one Reed-Solomon codec for audio and video

rs-gf256-v1 (construction documented, independent implementation matches
every KAT); audio parity with checked block arithmetic; every encoded
record must fit the producer's datagram budget (default 1431), and no
media record exceeds 64 KiB on encode, decode, send or reassembly. The
"audio" channel is retired. Muninn and Ratatoskr bump in lockstep at
Cuts 4-5. Three Soul passes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 09:05:11 +02:00
commit 29e60353f0
26 changed files with 3931 additions and 85 deletions

View file

@ -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"] }
@ -35,3 +36,8 @@ windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Securit
[dev-dependencies]
pretty_assertions = "1.4"
tempfile = "3"
[[example]]
name = "cultnet_impair"
path = "examples/cultnet_impair/main.rs"
test = true

View file

@ -0,0 +1,336 @@
//! `cultnet-impair`: a deterministic bidirectional UDP proxy for pressure-testing
//! real CultNet RUDP sessions. It owns transport impairment only; what a stream
//! does with the loss is the stream's business.
//!
//! The proxy listens at the endpoint a client dials and forwards to `--upstream`
//! through one upstream socket per client, so the reverse ACK and feedback
//! traffic crosses the same schedule. The schedule is `policy.rs`.
//!
//! ```text
//! cargo run --release --example cultnet_impair -- \
//! --listen 0.0.0.0:17890 --upstream 127.0.0.1:17990 \
//! --profile tests/impairment/loss-1pct.toml --seed 424242 --metrics loss-1pct.csv
//! ```
//!
//! Profiles are a small TOML subset of unsigned integers; see
//! `tests/impairment/`. The receiver binds the upstream endpoint and the client
//! dials the proxy's; never one port for both. For a two-host test, run the proxy
//! on the receiver host so reverse traffic is impaired too.
mod policy;
use std::collections::HashMap;
use std::env;
use std::fs::{self, File};
use std::io::{self, Write};
use std::net::{SocketAddr, UdpSocket};
use std::path::{Path, PathBuf};
use std::thread;
use std::time::{Duration, Instant};
use policy::{Direction, Profile, Scheduler, Stats};
const MAX_DATAGRAM: usize = 65_535;
const MAX_CLIENT_FLOWS: usize = 64;
const CLIENT_FLOW_IDLE_TIMEOUT_MS: u64 = 30_000;
const MAX_DATAGRAMS_PER_FLOW_PER_TURN: usize = 16;
#[derive(Debug)]
struct UpstreamFlow {
socket: UdpSocket,
last_activity_ms: u64,
}
fn open_upstream_flow(upstream: SocketAddr, now_ms: u64) -> io::Result<UpstreamFlow> {
let socket = UdpSocket::bind("0.0.0.0:0")?;
socket.connect(upstream)?;
socket.set_nonblocking(true)?;
Ok(UpstreamFlow {
socket,
last_activity_ms: now_ms,
})
}
fn ensure_upstream_flow(
flows: &mut HashMap<SocketAddr, UpstreamFlow>,
client: SocketAddr,
upstream: SocketAddr,
now_ms: u64,
stats: &mut Stats,
) -> io::Result<()> {
if flows.contains_key(&client) {
return Ok(());
}
if flows.len() >= MAX_CLIENT_FLOWS
&& let Some(oldest) = flows
.iter()
.min_by_key(|(_, flow)| flow.last_activity_ms)
.map(|(address, _)| *address)
{
flows.remove(&oldest);
stats.flow_evicted += 1;
}
flows.insert(client, open_upstream_flow(upstream, now_ms)?);
stats.max_flows = stats.max_flows.max(flows.len());
Ok(())
}
fn reap_idle_flows(
flows: &mut HashMap<SocketAddr, UpstreamFlow>,
now_ms: u64,
stats: &mut Stats,
) {
let before = flows.len();
flows.retain(|_, flow| {
now_ms.saturating_sub(flow.last_activity_ms) < CLIENT_FLOW_IDLE_TIMEOUT_MS
});
stats.flow_evicted += (before - flows.len()) as u64;
}
#[derive(Debug)]
struct Options {
listen: SocketAddr,
upstream: SocketAddr,
profile: PathBuf,
seed: u64,
metrics: Option<PathBuf>,
}
fn parse_options() -> Result<Options, String> {
let mut args = env::args().skip(1);
let mut listen = None;
let mut upstream = None;
let mut profile = None;
let mut seed = 1;
let mut metrics = None;
while let Some(arg) = args.next() {
let mut value = || args.next().ok_or_else(|| format!("{arg} requires a value"));
match arg.as_str() {
"--listen" => {
listen = Some(
value()?
.parse()
.map_err(|_| "--listen must be a socket address")?,
)
}
"--upstream" => {
upstream = Some(
value()?
.parse()
.map_err(|_| "--upstream must be a socket address")?,
)
}
"--profile" => profile = Some(PathBuf::from(value()?)),
"--seed" => {
seed = value()?
.parse()
.map_err(|_| "--seed must be an unsigned integer")?
}
"--metrics" => metrics = Some(PathBuf::from(value()?)),
"--help" | "-h" => return Err(usage().into()),
_ => return Err(format!("unknown argument {arg}\n{}", usage())),
}
}
Ok(Options {
listen: listen.ok_or("--listen is required")?,
upstream: upstream.ok_or("--upstream is required")?,
profile: profile.ok_or("--profile is required")?,
seed,
metrics,
})
}
fn usage() -> &'static str {
"usage: cultnet-impair --listen HOST:PORT --upstream HOST:PORT --profile PATH [--seed N] [--metrics PATH]"
}
fn elapsed_ms(start: Instant) -> u64 {
start.elapsed().as_millis().try_into().unwrap_or(u64::MAX)
}
fn write_metrics(path: &Path, stats: &Stats, active_flows: usize) -> io::Result<()> {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
let mut file = File::create(path)?;
writeln!(
file,
"received,forwarded,dropped,duplicated,reordered,stalled,queue_overflow,max_queue,flow_evicted,max_flows,active_flows"
)?;
writeln!(
file,
"{},{},{},{},{},{},{},{},{},{},{}",
stats.received,
stats.forwarded,
stats.dropped,
stats.duplicated,
stats.reordered,
stats.stalled,
stats.queue_overflow,
stats.max_queue,
stats.flow_evicted,
stats.max_flows,
active_flows
)
}
fn run(options: Options) -> Result<(), String> {
let profile = Profile::load(&options.profile)?;
let client_socket = UdpSocket::bind(options.listen)
.map_err(|error| format!("binding {}: {error}", options.listen))?;
client_socket
.set_nonblocking(true)
.map_err(|error| error.to_string())?;
println!(
"cultnet-impair listen={} upstream={} seed={} profile={}",
options.listen,
options.upstream,
options.seed,
options.profile.display()
);
let start = Instant::now();
let mut scheduler = Scheduler::new(profile, options.seed);
let mut flows = HashMap::<SocketAddr, UpstreamFlow>::new();
let mut last_flow_reap_at = 0_u64;
let mut buffer = vec![0; MAX_DATAGRAM];
loop {
let now = elapsed_ms(start);
let mut did_work = false;
for _ in 0..MAX_DATAGRAMS_PER_FLOW_PER_TURN {
match client_socket.recv_from(&mut buffer) {
Ok((size, address)) => {
did_work = true;
ensure_upstream_flow(
&mut flows,
address,
options.upstream,
now,
&mut scheduler.stats,
)
.map_err(|error| format!("opening upstream flow for {address}: {error}"))?;
if let Some(flow) = flows.get_mut(&address) {
flow.last_activity_ms = now;
}
scheduler.admit(
now,
Direction::ClientToUpstream,
&buffer[..size],
address,
);
}
Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
Err(error) => return Err(error.to_string()),
}
}
for (address, flow) in &mut flows {
for _ in 0..MAX_DATAGRAMS_PER_FLOW_PER_TURN {
match flow.socket.recv(&mut buffer) {
Ok(size) => {
did_work = true;
flow.last_activity_ms = now;
scheduler.admit(
now,
Direction::UpstreamToClient,
&buffer[..size],
*address,
);
}
Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
Err(error) => return Err(error.to_string()),
}
}
}
while let Some(item) = scheduler.pop_due(now) {
did_work = true;
let result = match item.direction {
Direction::ClientToUpstream => match flows.get_mut(&item.client) {
Some(flow) => {
flow.last_activity_ms = now;
flow.socket.send(&item.bytes)
}
None => {
scheduler.stats.dropped += 1;
continue;
}
},
Direction::UpstreamToClient => client_socket.send_to(&item.bytes, item.client),
};
if result.is_ok() {
scheduler.stats.forwarded += 1;
}
}
if let Some(path) = options.metrics.as_deref() {
if now % 1000 == 0 {
let _ = write_metrics(path, &scheduler.stats, flows.len());
}
}
if now.saturating_sub(last_flow_reap_at) >= 1_000 {
reap_idle_flows(&mut flows, now, &mut scheduler.stats);
last_flow_reap_at = now;
}
if did_work {
thread::yield_now();
} else {
thread::sleep(Duration::from_millis(1));
}
}
}
fn main() {
match parse_options().and_then(run) {
Ok(()) => {}
Err(error) => {
eprintln!("{error}");
std::process::exit(2);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn independent_clients_keep_independent_upstream_udp_identities() {
let mut flows = HashMap::new();
let mut stats = Stats::default();
let upstream = "127.0.0.1:54321".parse().unwrap();
let first = "127.0.0.1:12001".parse().unwrap();
let second = "127.0.0.1:12002".parse().unwrap();
ensure_upstream_flow(&mut flows, first, upstream, 1, &mut stats).unwrap();
ensure_upstream_flow(&mut flows, second, upstream, 2, &mut stats).unwrap();
assert_eq!(flows.len(), 2);
assert_ne!(
flows[&first].socket.local_addr().unwrap(),
flows[&second].socket.local_addr().unwrap()
);
assert_eq!(stats.max_flows, 2);
assert_eq!(stats.flow_evicted, 0);
}
#[test]
fn idle_client_flows_are_reaped_without_touching_active_flows() {
let mut flows = HashMap::new();
let mut stats = Stats::default();
let upstream = "127.0.0.1:54321".parse().unwrap();
let idle = "127.0.0.1:12001".parse().unwrap();
let active = "127.0.0.1:12002".parse().unwrap();
ensure_upstream_flow(&mut flows, idle, upstream, 0, &mut stats).unwrap();
ensure_upstream_flow(
&mut flows,
active,
upstream,
CLIENT_FLOW_IDLE_TIMEOUT_MS - 1,
&mut stats,
)
.unwrap();
reap_idle_flows(&mut flows, CLIENT_FLOW_IDLE_TIMEOUT_MS, &mut stats);
assert!(!flows.contains_key(&idle));
assert!(flows.contains_key(&active));
assert_eq!(stats.flow_evicted, 1);
}
}

View file

@ -0,0 +1,298 @@
//! The impairment policy: what happens to each datagram, and nothing about
//! sockets. `Profile` says what to do, `Scheduler` decides per datagram and
//! holds the delayed ones, and every decision is a pure function of the
//! profile, the seed and the datagram's position, so a schedule replays
//! exactly. The proxy in `main.rs` and the loss-matrix test in
//! `tests/media_fec_loss.rs` both take their schedule from here.
use std::collections::VecDeque;
use std::fs;
use std::net::SocketAddr;
use std::path::Path;
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Profile {
pub loss_basis_points: u32,
pub burst_every: u64,
pub burst_length: u64,
pub duplicate_every: u64,
pub reorder_every: u64,
pub reorder_delay_ms: u64,
pub delay_ms: u64,
pub jitter_ms: u64,
pub stall_at_ms: u64,
pub stall_for_ms: u64,
pub max_scheduled: usize,
}
impl Default for Profile {
fn default() -> Self {
Self {
loss_basis_points: 0,
burst_every: 0,
burst_length: 0,
duplicate_every: 0,
reorder_every: 0,
reorder_delay_ms: 0,
delay_ms: 0,
jitter_ms: 0,
stall_at_ms: 0,
stall_for_ms: 0,
max_scheduled: 4096,
}
}
}
impl Profile {
pub fn load(path: &Path) -> Result<Self, String> {
let text = fs::read_to_string(path)
.map_err(|error| format!("reading {}: {error}", path.display()))?;
let mut profile = Self::default();
for (index, raw) in text.lines().enumerate() {
let line = raw.split('#').next().unwrap_or("").trim();
if line.is_empty() || line.starts_with('[') {
continue;
}
let (key, value) = line
.split_once('=')
.ok_or_else(|| format!("{}:{}: expected key = value", path.display(), index + 1))?;
let key = key.trim();
let value = value.trim().trim_matches('"');
let number = value.parse::<u64>().map_err(|_| {
format!(
"{}:{}: {key} must be an unsigned integer",
path.display(),
index + 1
)
})?;
match key {
"loss_basis_points" => {
profile.loss_basis_points = number
.try_into()
.map_err(|_| "loss_basis_points is too large")?
}
"burst_every" => profile.burst_every = number,
"burst_length" => profile.burst_length = number,
"duplicate_every" => profile.duplicate_every = number,
"reorder_every" => profile.reorder_every = number,
"reorder_delay_ms" => profile.reorder_delay_ms = number,
"delay_ms" => profile.delay_ms = number,
"jitter_ms" => profile.jitter_ms = number,
"stall_at_ms" => profile.stall_at_ms = number,
"stall_for_ms" => profile.stall_for_ms = number,
"max_scheduled" => {
profile.max_scheduled = number
.try_into()
.map_err(|_| "max_scheduled is too large")?
}
_ => {
return Err(format!(
"{}:{}: unknown profile key {key}",
path.display(),
index + 1
));
}
}
}
if profile.loss_basis_points > 10_000 {
return Err("loss_basis_points must be <= 10000".into());
}
if profile.max_scheduled == 0 {
return Err("max_scheduled must be greater than zero".into());
}
Ok(profile)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Direction {
ClientToUpstream,
UpstreamToClient,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Decision {
pub drop: bool,
pub copies: u8,
pub delay_ms: u64,
pub reordered: bool,
pub stalled: bool,
}
#[derive(Clone, Debug)]
pub struct DeterministicRng(u64);
impl DeterministicRng {
pub fn new(seed: u64) -> Self {
Self(if seed == 0 { 0x9e3779b97f4a7c15 } else { seed })
}
pub fn next(&mut self) -> u64 {
let mut x = self.0;
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
self.0 = x;
x
}
}
#[derive(Debug)]
pub struct Scheduler {
pub profile: Profile,
pub rng: DeterministicRng,
pub seen: u64,
pub burst_remaining: u64,
pub queue: VecDeque<Scheduled>,
pub stats: Stats,
}
#[derive(Debug)]
pub struct Scheduled {
pub due_ms: u64,
pub direction: Direction,
pub bytes: Vec<u8>,
pub client: SocketAddr,
}
#[derive(Debug, Default)]
pub struct Stats {
pub received: u64,
pub forwarded: u64,
pub dropped: u64,
pub duplicated: u64,
pub reordered: u64,
pub stalled: u64,
pub queue_overflow: u64,
pub max_queue: usize,
pub flow_evicted: u64,
pub max_flows: usize,
}
impl Scheduler {
pub fn new(profile: Profile, seed: u64) -> Self {
Self {
profile,
rng: DeterministicRng::new(seed),
seen: 0,
burst_remaining: 0,
queue: VecDeque::new(),
stats: Stats::default(),
}
}
pub fn decide(&mut self, now_ms: u64) -> Decision {
self.seen += 1;
let stalled = self.profile.stall_for_ms > 0
&& now_ms >= self.profile.stall_at_ms
&& now_ms
< self
.profile
.stall_at_ms
.saturating_add(self.profile.stall_for_ms);
if stalled {
return Decision {
drop: true,
copies: 0,
delay_ms: 0,
reordered: false,
stalled: true,
};
}
if self.burst_remaining == 0
&& self.profile.burst_every > 0
&& self.seen % self.profile.burst_every == 0
{
self.burst_remaining = self.profile.burst_length;
}
if self.burst_remaining > 0 {
self.burst_remaining -= 1;
return Decision {
drop: true,
copies: 0,
delay_ms: 0,
reordered: false,
stalled: false,
};
}
if self.profile.loss_basis_points > 0
&& self.rng.next() % 10_000 < u64::from(self.profile.loss_basis_points)
{
return Decision {
drop: true,
copies: 0,
delay_ms: 0,
reordered: false,
stalled: false,
};
}
let duplicate =
self.profile.duplicate_every > 0 && self.seen % self.profile.duplicate_every == 0;
let reordered =
self.profile.reorder_every > 0 && self.seen % self.profile.reorder_every == 0;
let jitter = if self.profile.jitter_ms == 0 {
0
} else {
self.rng.next() % (self.profile.jitter_ms + 1)
};
Decision {
drop: false,
copies: if duplicate { 2 } else { 1 },
delay_ms: self.profile.delay_ms
+ jitter
+ if reordered {
self.profile.reorder_delay_ms
} else {
0
},
reordered,
stalled: false,
}
}
pub fn admit(&mut self, now_ms: u64, direction: Direction, bytes: &[u8], client: SocketAddr) {
self.stats.received += 1;
let decision = self.decide(now_ms);
if decision.drop {
self.stats.dropped += 1;
if decision.stalled {
self.stats.stalled += 1;
}
return;
}
if decision.copies == 2 {
self.stats.duplicated += 1;
}
if decision.reordered {
self.stats.reordered += 1;
}
for copy in 0..decision.copies {
if self.queue.len() >= self.profile.max_scheduled {
self.stats.queue_overflow += 1;
self.stats.dropped += 1;
continue;
}
let delay = decision.delay_ms + u64::from(copy);
let item = Scheduled {
due_ms: now_ms.saturating_add(delay),
direction,
bytes: bytes.to_vec(),
client,
};
let position = self
.queue
.iter()
.position(|queued| queued.due_ms > item.due_ms)
.unwrap_or(self.queue.len());
self.queue.insert(position, item);
self.stats.max_queue = self.stats.max_queue.max(self.queue.len());
}
}
pub fn pop_due(&mut self, now_ms: u64) -> Option<Scheduled> {
if self.queue.front().is_some_and(|item| item.due_ms <= now_ms) {
self.queue.pop_front()
} else {
None
}
}
}

View file

@ -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::*;

View file

@ -0,0 +1,749 @@
//! 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: the construction below is, and
//! `tests/media_fec.rs` pins the parity bytes of known inputs, so a library that
//! changes the matrix fails there instead of on the wire, and a decoder in
//! another runtime has vectors to meet.
//!
//! # The `rs-gf256-v1` construction
//!
//! Enough to implement it in another runtime without this crate.
//!
//! * The field is GF(2^8) over the polynomial `x^8 + x^4 + x^3 + x^2 + 1`
//! (`0x11D`). Addition is XOR; multiplication is carry-less multiplication
//! reduced by `0x11D`.
//! * For a block of `k` data shards and `m` parity shards, let `n = k + m`
//! (`n <= 256`) and build the `n x k` Vandermonde matrix `V` whose entry in
//! row `r`, column `c` is `r^c`, where the row number `r` (0 through
//! `n - 1`) is itself read as a field element and `0^0 = 1`. The evaluation
//! points are the integers 0 through `n - 1`, not powers of a generator.
//! * The encoding matrix is `E = V * inverse(top)`, where `top` is the first
//! `k` rows of `V`. Its first `k` rows are the identity, so data shards are
//! sent unchanged; its last `m` rows generate parity.
//! * Every shard of a block has the same length, `shard_payload_bytes`. A
//! shorter final video chunk is zero-padded to that length before encoding.
//! Byte `j` of parity shard `p` is the XOR over data shards `i` of
//! `E[k + p][i] * data[i][j]`.
//! * Any `k` surviving shards recover the block: invert the `k x k` submatrix
//! of `E` for the surviving rows and multiply.
//!
//! The known-answer vectors in `tests/media_fec.rs` were checked against an
//! independent implementation of exactly this text.
//!
//! # 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.
//!
//! # One datagram
//!
//! Every record the protect functions return, data or parity, fits one
//! datagram once wrapped for the wire, so parity never reintroduces the
//! fragment amplification it exists to remove. The protect functions enforce
//! that against [`MediaFecPolicy::max_wire_bytes`]: they encode each record
//! with the caller's provenance and refuse the block if any is larger.
//! The default budget is [`MEDIA_FEC_MAX_WIRE_BYTES`], an IPv4 path with a
//! 1500-byte MTU. Pass a smaller budget when the path carries less: IPv6 over
//! 1500 leaves a UDP payload of 1452 bytes, and a WireGuard or Tailscale
//! tunnel about 1252, so subtract the 41 bytes of RUDP header and channel name
//! from the path's UDP payload limit.
//!
//! The budget is measured with the provenance given, so a producer whose
//! `stored_at` or identifiers grow later must pass the widest it will use.
//!
//! # Audio playout
//!
//! A recovered audio packet has the block's latest `deadline_ticks`, so a
//! packet recovered early in the block reports a deadline later than its own.
//! Audio consumers must gate playout on `pts_ticks`, which recovery derives
//! exactly, and use `deadline_ticks` only to give up on the block as a whole.
//!
//! # 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, GAMECULT_MEDIA_MAX_WIRE_BYTES, GameCultMediaWireRecord, MediaWireProvenance,
encode_media_wire_record, validate_audio_parity_record, validate_audio_record,
validate_video_parity_record, validate_video_record,
};
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 default datagram budget: the most a wrapped media record may occupy so
/// that the RUDP header and the channel name still leave it in one IPv4
/// datagram on a 1500-byte MTU (1,431 bytes).
pub const MEDIA_FEC_MAX_WIRE_BYTES: usize =
MEDIA_IPV4_UDP_PAYLOAD_BYTES - RUDP_FIXED_HEADER_BYTES - GAMECULT_MEDIA_CHANNEL.len();
/// A producer's choice of `(k, m)` and datagram budget.
/// [`MediaFecPolicy::STANDARD`] is the ruled default; each record carries the
/// `(k, m)` it was made with, so a decoder needs no policy.
///
/// Build one by overriding fields of the default, `MediaFecPolicy {
/// max_wire_bytes: 1_211, ..MediaFecPolicy::STANDARD }`, not by naming every
/// field: a field added later then cannot break, or silently mis-set, the
/// caller. `max_wire_bytes` may not exceed
/// [`GAMECULT_MEDIA_MAX_WIRE_BYTES`], the ceiling a decoder holds every record
/// to.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct MediaFecPolicy {
/// The most data chunks in one video block.
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,
/// The most a wrapped record may occupy. The protect functions refuse a
/// block with any record over it. See the module docs for smaller paths.
pub max_wire_bytes: usize,
}
impl MediaFecPolicy {
/// Video `k <= 16` with `m = max(2, ceil(k / 4))`; audio 4+2; the IPv4
/// datagram budget.
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,
max_wire_bytes: MEDIA_FEC_MAX_WIRE_BYTES,
};
/// `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"));
}
if self.max_wire_bytes > GAMECULT_MEDIA_MAX_WIRE_BYTES {
return Err(MediaFecError::invalid(format!(
"media FEC policy max_wire_bytes {} exceeds the {GAMECULT_MEDIA_MAX_WIRE_BYTES}-byte media record ceiling",
self.max_wire_bytes
)));
}
Ok(())
}
}
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())
}
/// Refuses a record that would not fit one datagram once wrapped.
fn fits_datagram(
record: &GameCultMediaWireRecord,
provenance: MediaWireProvenance<'_>,
policy: &MediaFecPolicy,
) -> Result<(), MediaFecError> {
let wire = encode_media_wire_record(record, provenance).map_err(from_anyhow)?;
if wire.len() > policy.max_wire_bytes {
return Err(MediaFecError::invalid(format!(
"a {} record wraps to {} bytes, over the {}-byte datagram budget: cut chunks smaller or raise max_wire_bytes",
record.schema_id(),
wire.len(),
policy.max_wire_bytes
)));
}
Ok(())
}
/// Where slot `slot` of an audio block lands: its packet id and presentation
/// time, or `None` if either overflows. The one place block arithmetic lives;
/// the protect path, the wire validator and recovery all ask it.
pub(crate) fn audio_slot(
base_packet_id: u64,
base_pts_ticks: i64,
packet_duration_ticks: u32,
slot: u64,
) -> Option<(u64, i64)> {
let packet_id = base_packet_id.checked_add(slot)?;
let pts_ticks = i64::from(packet_duration_ticks)
.checked_mul(i64::try_from(slot).ok()?)?
.checked_add(base_pts_ticks)?;
Some((packet_id, pts_ticks))
}
// ---------------------------------------------------------------------------
// 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 any resulting record, wrapped with `provenance`,
/// exceeds `policy.max_wire_bytes`.
pub fn protect_video_frame(
chunks: &[GameCultMediaVideoAccessUnitRecord],
policy: &MediaFecPolicy,
provenance: MediaWireProvenance<'_>,
) -> 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 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;
}
let records = rotate(lanes);
for record in &records {
fits_datagram(record, provenance, policy)?;
}
Ok(records)
}
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();
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.
/// Errors if a parity record, wrapped with `provenance`, exceeds
/// `policy.max_wire_bytes`; the packets are smaller than their parity.
pub fn protect_audio_block(
packets: &[GameCultMediaAudioPacketRecord],
policy: &MediaFecPolicy,
provenance: MediaWireProvenance<'_>,
) -> 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 =
audio_slot(first.packet_id, first.pts_ticks, first.duration_ticks, index as u64);
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, packet.pts_ticks)) != expected
|| packet.payload.len() != shard_bytes
{
return Err(MediaFecError::invalid(
"audio block packets must be contiguous and share codec, timebase, duration and payload length",
));
}
}
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);
let parity: Vec<GameCultMediaAudioParityShardRecord> =
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();
// A parity record repeats every field of a packet and adds the scheme and
// the geometry, so it is the larger of the block's records: if it fits, the
// packets do.
for shard in &parity {
fits_datagram(&GameCultMediaWireRecord::AudioParity(shard.clone()), provenance, policy)?;
}
Ok(parity)
}
/// 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:
/// gate playout on `pts_ticks`, not `deadline_ticks`. 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();
let recovered = recover_data_shards(k, shards)?;
missing
.into_iter()
.map(|slot| {
let (packet_id, pts_ticks) = audio_slot(
geometry.base_packet_id,
geometry.base_pts_ticks,
geometry.packet_duration_ticks,
slot as u64,
)
.ok_or_else(|| MediaFecError::invalid("audio block runs past the packet id or pts range"))?;
Ok(GameCultMediaAudioPacketRecord {
stream_id: geometry.stream_id.clone(),
session_id: geometry.session_id.clone(),
packet_id,
codec: geometry.codec.clone(),
pts_ticks,
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
}

View file

@ -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.

View file

@ -28,20 +28,32 @@ use crate::contracts::{
CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, CultNetWireContract,
decode_cultnet_message_from_slice, encode_cultnet_message_to_vec,
};
use crate::media_fec::{MEDIA_FEC_SCHEME_RS_GF256_V1, audio_slot};
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";
/// The largest wrapped media record the decoder accepts, and the most the RUDP
/// `media` channel reassembles: 64 KiB.
///
/// It is the bound a receiver holds a peer to. A parity shard's length is
/// whatever its producer's [`MediaFecPolicy::max_wire_bytes`](crate::MediaFecPolicy::max_wire_bytes) allowed, and
/// recovery work grows with the square of the block size times that length, so
/// without a ceiling one hostile record could name any length RUDP would
/// reassemble (16 MiB). 64 KiB is one maximum UDP datagram: it covers every
/// path MTU in use, jumbo frames (9,000 bytes) with a wide margin, and a
/// record is still one message rather than a stream. A policy may not ask for
/// more; see [`MediaFecPolicy`](crate::MediaFecPolicy).
pub const GAMECULT_MEDIA_MAX_WIRE_BYTES: usize = 64 * 1024;
/// One media record on its way somewhere.
#[derive(Clone, Debug, PartialEq, Eq)]
@ -49,6 +61,7 @@ pub enum GameCultMediaWireRecord {
Video(GameCultMediaVideoAccessUnitRecord),
VideoParity(GameCultMediaVideoParityShardRecord),
Audio(GameCultMediaAudioPacketRecord),
AudioParity(GameCultMediaAudioParityShardRecord),
Feedback(GameCultMediaReceiverFeedbackRecord),
}
@ -84,6 +97,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 +110,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 +146,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)
@ -159,11 +182,24 @@ pub fn encode_media_wire_record(
},
};
encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).map_err(Into::into)
let wire = encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?;
if wire.len() > GAMECULT_MEDIA_MAX_WIRE_BYTES {
return Err(anyhow!(
"media record is {} bytes, over the {GAMECULT_MEDIA_MAX_WIRE_BYTES}-byte ceiling",
wire.len()
));
}
Ok(wire)
}
/// Unwraps a media record from the wire.
pub fn decode_media_wire_record(payload: &[u8]) -> Result<GameCultMediaWireRecord> {
if payload.len() > GAMECULT_MEDIA_MAX_WIRE_BYTES {
return Err(anyhow!(
"media record is {} bytes, over the {GAMECULT_MEDIA_MAX_WIRE_BYTES}-byte ceiling",
payload.len()
));
}
let message = decode_cultnet_message_from_slice(payload, CultNetWireContract::CultNetSchemaV0)?;
let CultNetMessage::DocumentPutRaw { document, .. } = message else {
return Err(anyhow!("expected cultnet.document_put_raw.v0"));
@ -191,6 +227,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 +391,114 @@ 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 {
return Err(anyhow!(
"video parity media record chunk lengths must be non-zero"
"video parity media record shard_payload_bytes must be non-zero"
));
}
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.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"));
}
let Some((_, last_pts_ticks)) = audio_slot(
record.base_packet_id,
record.base_pts_ticks,
record.packet_duration_ticks,
u64::from(record.data_shard_count) - 1,
) else {
return Err(anyhow!(
"audio parity media record block runs past the packet id or pts range"
));
};
// Recovery gives every recovered packet the block's deadline, so it must
// not precede the last packet's pts or a recovered packet would fail
// `validate_audio_record`.
if record.deadline_ticks < last_pts_ticks {
return Err(anyhow!(
"audio parity media record deadline_ticks must not precede the block's last pts_ticks"
));
}
if record.shard_payload_bytes == 0 {
return Err(anyhow!(
"audio parity media record shard_payload_bytes must be non-zero"
));
}
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(())

View file

@ -23,13 +23,15 @@ use crate::CultNetTransportProfile;
use crate::CultNetTransportProtocol;
use crate::CultNetTransportStats;
use crate::CultNetWireContract;
use crate::GAMECULT_MEDIA_CHANNEL;
use crate::GAMECULT_MEDIA_MAX_WIRE_BYTES;
use crate::create_reconnect_policy;
use crate::decode_cultnet_message_from_slice;
use crate::encode_cultnet_message_to_vec;
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;
@ -468,7 +470,7 @@ impl CultNetRudpSession {
if options.ordered && !options.reliable {
return Err(anyhow!("RUDP ordered delivery requires reliability"));
}
self.require_payload_size(payload.len())?;
self.require_payload_size(channel_id, payload.len())?;
// Reclaim deadline-passed sends before measuring window capacity, so an
// expiring stream cannot be refused for a backlog it no longer owns.
self.purge_expired_reliable(options.now_ms);
@ -1003,7 +1005,7 @@ impl CultNetRudpSession {
packet: &CultNetRudpPacket,
) -> Result<Option<(CultNetRudpDeliveredFrame, bool, u32)>> {
if packet.fragment_count == 0 {
self.require_payload_size(packet.payload.len())?;
self.require_payload_size(&packet.channel_id, packet.payload.len())?;
return Ok(Some((
CultNetRudpDeliveredFrame {
channel_id: packet.channel_id.clone(),
@ -1055,7 +1057,7 @@ impl CultNetRudpSession {
.sum::<usize>()
})
.unwrap_or(0);
if let Err(error) = self.require_payload_size(buffered_bytes + packet.payload.len()) {
if let Err(error) = self.require_payload_size(&packet.channel_id, buffered_bytes + packet.payload.len()) {
self.fragment_buffers.remove(&key);
return Err(error);
}
@ -1113,11 +1115,12 @@ impl CultNetRudpSession {
)))
}
fn require_payload_size(&self, payload_bytes: usize) -> Result<()> {
if self
.max_payload_bytes
.is_some_and(|max_payload_bytes| payload_bytes > max_payload_bytes)
{
/// The session's cap, tightened on the media channel to the ceiling every
/// media record is held to (`GAMECULT_MEDIA_MAX_WIRE_BYTES`), so media
/// reassembly never buffers more than a record may be.
fn require_payload_size(&self, channel_id: &str, payload_bytes: usize) -> Result<()> {
let limit = channel_payload_limit(channel_id, self.max_payload_bytes);
if limit.is_some_and(|limit| payload_bytes > limit) {
return Err(anyhow!("RUDP payload exceeds max_payload_bytes"));
}
Ok(())
@ -2309,6 +2312,18 @@ pub struct RudpTransportProfileOptions {
pub media_delivery: Option<CultNetTransportDelivery>,
}
/// The largest payload a channel accepts: the session-wide cap, tightened on
/// the media channel to the media ceiling.
fn channel_payload_limit(channel_id: &str, session_limit: Option<usize>) -> Option<usize> {
if channel_id == GAMECULT_MEDIA_CHANNEL {
Some(session_limit.map_or(GAMECULT_MEDIA_MAX_WIRE_BYTES, |limit| {
limit.min(GAMECULT_MEDIA_MAX_WIRE_BYTES)
}))
} else {
session_limit
}
}
pub fn create_rudp_transport_profile(
runtime_id: impl Into<String>,
options: RudpTransportProfileOptions,
@ -2358,23 +2373,16 @@ pub fn create_rudp_transport_profile(
reliable_expire_after_ms: None,
},
CultNetTransportChannel {
channel_id: "media".to_string(),
channel_id: GAMECULT_MEDIA_CHANNEL.to_string(),
delivery: options
.media_delivery
.unwrap_or(CultNetTransportDelivery::Reliable),
ordering: CultNetTransportOrdering::Unordered,
max_payload_bytes: options.max_payload_bytes,
max_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,
},
// 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_payload_bytes: channel_payload_limit(
GAMECULT_MEDIA_CHANNEL,
options.max_payload_bytes.map(|limit| limit as usize),
)
.map(|limit| limit as u32),
max_fragment_bytes: options.max_fragment_bytes,
max_pending_reliable_packets: options.max_pending_reliable_packets,
reliable_expire_after_ms: options.media_reliable_expire_after_ms,
@ -2544,7 +2552,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,

View file

@ -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"

View file

@ -0,0 +1,3 @@
burst_every = 200
burst_length = 4
max_scheduled = 4096

View file

@ -0,0 +1,3 @@
burst_every = 200
burst_length = 8
max_scheduled = 4096

View file

@ -0,0 +1,2 @@
# No impairment. The proxy itself remains in the measured path.
max_scheduled = 4096

View file

@ -0,0 +1,3 @@
delay_ms = 5
jitter_ms = 10
max_scheduled = 4096

View file

@ -0,0 +1,2 @@
duplicate_every = 50
max_scheduled = 4096

View file

@ -0,0 +1,2 @@
loss_basis_points = 100
max_scheduled = 4096

View file

@ -0,0 +1,2 @@
loss_basis_points = 300
max_scheduled = 4096

View file

@ -0,0 +1,2 @@
loss_basis_points = 500
max_scheduled = 4096

View file

@ -0,0 +1,37 @@
profile chunks frames whole/no-fec whole/fec recovered beyond
clean 8 200 200 200 0 0
clean 16 200 200 200 0 0
clean 17 200 200 200 0 0
clean 24 200 200 200 0 0
clean 32 200 200 200 0 0
clean 48 200 200 200 0 0
loss-1pct 8 200 183 200 17 0
loss-1pct 16 200 170 200 30 0
loss-1pct 17 200 170 200 31 0
loss-1pct 24 200 154 200 48 0
loss-1pct 32 200 144 200 59 0
loss-1pct 48 200 137 200 75 0
loss-3pct 8 200 158 199 41 1
loss-3pct 16 200 126 200 74 0
loss-3pct 17 200 119 199 90 0
loss-3pct 24 200 101 200 116 0
loss-3pct 32 200 78 200 147 0
loss-3pct 48 200 57 200 208 0
loss-5pct 8 200 130 197 67 3
loss-5pct 16 200 93 199 106 1
loss-5pct 17 200 92 198 132 1
loss-5pct 24 200 59 198 180 2
loss-5pct 32 200 41 200 214 0
loss-5pct 48 200 21 199 318 1
burst-4 8 200 181 191 10 9
burst-4 16 200 161 200 39 0
burst-4 17 200 175 200 46 0
burst-4 24 200 161 200 68 0
burst-4 32 200 121 200 118 0
burst-4 48 200 121 200 177 0
burst-8 8 200 181 191 10 0
burst-8 16 200 161 181 20 19
burst-8 17 200 171 183 20 30
burst-8 24 200 161 171 19 49
burst-8 32 200 121 200 118 0
burst-8 48 200 121 200 197 0

View file

@ -0,0 +1,3 @@
reorder_every = 20
reorder_delay_ms = 12
max_scheduled = 4096

View file

@ -0,0 +1,3 @@
stall_at_ms = 10000
stall_for_ms = 250
max_scheduled = 4096

View file

@ -0,0 +1,219 @@
//! The impairment policy is deterministic, or a loss experiment cannot be
//! replayed. These pin each decision the proxy example makes, against the same
//! `policy.rs` the example compiles.
#[allow(dead_code)]
#[path = "../examples/cultnet_impair/policy.rs"]
mod policy;
use std::net::SocketAddr;
use policy::{Direction, Profile, Scheduler};
fn client() -> SocketAddr {
"127.0.0.1:1234".parse().unwrap()
}
#[test]
fn iid_decisions_are_seed_reproducible() {
let profile = Profile {
loss_basis_points: 2500,
..Default::default()
};
let mut a = Scheduler::new(profile.clone(), 42);
let mut b = Scheduler::new(profile, 42);
let left: Vec<_> = (0..100).map(|_| a.decide(0)).collect();
let right: Vec<_> = (0..100).map(|_| b.decide(0)).collect();
assert_eq!(left, right);
assert!(left.iter().any(|d| d.drop));
assert!(left.iter().any(|d| !d.drop));
}
#[test]
fn burst_drop_is_exact() {
let profile = Profile {
burst_every: 5,
burst_length: 3,
..Default::default()
};
let mut scheduler = Scheduler::new(profile, 1);
let drops: Vec<_> = (1..=9).filter(|_| scheduler.decide(0).drop).collect();
assert_eq!(drops, vec![5, 6, 7], "a burst starts on the Nth datagram and runs burst_length");
}
#[test]
fn duplicate_delay_and_reorder_are_deterministic() {
let profile = Profile {
duplicate_every: 2,
reorder_every: 3,
reorder_delay_ms: 20,
delay_ms: 5,
..Default::default()
};
let mut scheduler = Scheduler::new(profile, 1);
assert_eq!(scheduler.decide(0).copies, 1);
assert_eq!(scheduler.decide(0).copies, 2);
let third = scheduler.decide(0);
assert!(third.reordered);
assert_eq!(third.delay_ms, 25);
}
#[test]
fn stall_drops_only_inside_window() {
let profile = Profile {
stall_at_ms: 100,
stall_for_ms: 50,
..Default::default()
};
let mut scheduler = Scheduler::new(profile, 1);
assert!(!scheduler.decide(99).drop);
let stalled = scheduler.decide(100);
assert!(stalled.drop && stalled.stalled);
assert!(scheduler.decide(149).drop);
assert!(!scheduler.decide(150).drop);
}
#[test]
fn reorder_delay_places_packet_behind_later_packet() {
let profile = Profile {
reorder_every: 2,
reorder_delay_ms: 20,
..Default::default()
};
let mut scheduler = Scheduler::new(profile, 1);
scheduler.admit(0, Direction::ClientToUpstream, b"first", client());
scheduler.admit(0, Direction::ClientToUpstream, b"held", client());
scheduler.admit(1, Direction::ClientToUpstream, b"later", client());
assert_eq!(scheduler.pop_due(1).unwrap().bytes, b"first");
assert_eq!(scheduler.pop_due(1).unwrap().bytes, b"later");
assert!(scheduler.pop_due(19).is_none());
assert_eq!(scheduler.pop_due(20).unwrap().bytes, b"held");
}
#[test]
fn scheduled_queue_is_bounded_and_counts_overflow() {
let profile = Profile {
delay_ms: 100,
max_scheduled: 2,
..Default::default()
};
let mut scheduler = Scheduler::new(profile, 1);
for value in 0..5 {
scheduler.admit(0, Direction::ClientToUpstream, &[value], client());
}
assert_eq!(scheduler.queue.len(), 2);
assert_eq!(scheduler.stats.max_queue, 2);
assert_eq!(scheduler.stats.queue_overflow, 3);
assert_eq!(scheduler.stats.dropped, 3);
}
fn profile_path(name: &str) -> std::path::PathBuf {
std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.join("tests/impairment")
.join(format!("{name}.toml"))
}
#[test]
fn every_committed_profile_loads_with_its_named_meaning() {
let load = |name: &str| Profile::load(&profile_path(name)).unwrap_or_else(|e| panic!("{name}: {e}"));
assert_eq!(load("clean"), Profile::default());
assert_eq!(load("loss-1pct").loss_basis_points, 100);
assert_eq!(load("loss-3pct").loss_basis_points, 300);
assert_eq!(load("loss-5pct").loss_basis_points, 500);
let burst4 = load("burst-4");
assert_eq!((burst4.burst_every, burst4.burst_length), (200, 4));
let burst8 = load("burst-8");
assert_eq!((burst8.burst_every, burst8.burst_length), (200, 8));
let jitter = load("delay-jitter");
assert_eq!((jitter.delay_ms, jitter.jitter_ms), (5, 10));
assert_eq!(load("duplicate").duplicate_every, 50);
let reorder = load("reorder");
assert_eq!((reorder.reorder_every, reorder.reorder_delay_ms), (20, 12));
let stall = load("stall-250ms");
assert_eq!((stall.stall_at_ms, stall.stall_for_ms), (10_000, 250));
}
#[test]
fn a_malformed_profile_is_refused_with_its_line() {
let dir = tempfile::tempdir().unwrap();
let write = |name: &str, text: &str| {
let path = dir.path().join(name);
std::fs::write(&path, text).unwrap();
Profile::load(&path).unwrap_err()
};
assert!(write("unknown.toml", "jitter = 3
").contains("unknown profile key jitter"));
assert!(write("word.toml", "delay_ms = soon
").contains("unsigned integer"));
assert!(write("shape.toml", "delay_ms
").contains("key = value"));
assert!(write("loss.toml", "loss_basis_points = 10001
").contains("<= 10000"));
assert!(write("queue.toml", "max_scheduled = 0
").contains("greater than zero"));
assert!(write("line.toml", "
delay_ms = x
").contains(":3:"));
}
#[test]
fn admission_counts_what_it_did_to_each_datagram() {
let mut scheduler = Scheduler::new(
Profile { duplicate_every: 2, reorder_every: 3, reorder_delay_ms: 10, ..Default::default() },
1,
);
for value in 0..6 {
scheduler.admit(0, Direction::ClientToUpstream, &[value], client());
}
// Datagrams 2, 4 and 6 are duplicated, 3 and 6 are reordered (6 is both).
assert_eq!(scheduler.stats.received, 6);
assert_eq!(scheduler.stats.duplicated, 3);
assert_eq!(scheduler.stats.reordered, 2);
assert_eq!(scheduler.queue.len(), 9);
let mut stalled = Scheduler::new(Profile { stall_at_ms: 0, stall_for_ms: 10, ..Default::default() }, 1);
stalled.admit(5, Direction::UpstreamToClient, b"x", client());
assert_eq!((stalled.stats.stalled, stalled.stats.dropped, stalled.queue.len()), (1, 1, 0));
}
#[test]
fn jitter_is_seeded_uniform_within_its_bound() {
let profile = Profile { delay_ms: 7, jitter_ms: 5, ..Default::default() };
let mut scheduler = Scheduler::new(profile, 9);
let delays: Vec<u64> = (0..300).map(|_| scheduler.decide(0).delay_ms).collect();
assert!(delays.iter().all(|delay| (7..=12).contains(delay)), "{delays:?}");
for delay in 7..=12 {
assert!(delays.contains(&delay), "delay {delay} never drawn");
}
}
#[test]
fn the_loss_ceiling_is_inclusive() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("all.toml");
std::fs::write(&path, "loss_basis_points = 10000\n").unwrap();
let profile = Profile::load(&path).unwrap();
let mut scheduler = Scheduler::new(profile, 1);
assert!((0..50).all(|_| scheduler.decide(0).drop), "10000 basis points drops everything");
}
/// The seeded stream is part of the contract: the same seed must drop the same
/// datagrams on every run, on every host. 100,000 draws of xorshift64 from seed
/// 42, taken modulo 10,000 and compared against 100, give 1,008.
#[test]
fn seeded_loss_drops_the_exact_datagrams_the_stream_dictates() {
let mut scheduler = Scheduler::new(Profile { loss_basis_points: 100, ..Default::default() }, 42);
let dropped = (0..100_000).filter(|_| scheduler.decide(0).drop).count();
assert_eq!(dropped, 1_008);
}
#[test]
fn datagrams_due_together_leave_in_arrival_order() {
let mut scheduler = Scheduler::new(Profile { delay_ms: 5, ..Default::default() }, 1);
for value in 0..4_u8 {
scheduler.admit(0, Direction::ClientToUpstream, &[value], client());
}
let order: Vec<u8> = std::iter::from_fn(|| scheduler.pop_due(5)).map(|item| item.bytes[0]).collect();
assert_eq!(order, vec![0, 1, 2, 3]);
}

View file

@ -90,3 +90,121 @@ fn the_other_channels_keep_their_fixed_delivery() {
assert_eq!(delivery_of("schema"), CultNetTransportDelivery::Reliable);
assert_eq!(delivery_of("realtime"), CultNetTransportDelivery::Unreliable);
}
// ---------------------------------------------------------------------------
// What the media channel puts on the wire
// ---------------------------------------------------------------------------
use std::thread::sleep;
use std::time::Duration;
use cultnet_rs::{
CultNetRudpPacket, CultNetRudpSession, CultNetRudpSessionOptions, decode_rudp_packet,
encode_rudp_packet,
};
/// A connected client transport and the raw socket standing in for its peer.
struct Link {
client: CultNetRudpSocketTransportConnection,
peer: UdpSocket,
}
impl Link {
fn open(delivery: Option<CultNetTransportDelivery>, expire_after_ms: Option<u64>) -> Self {
let peer = UdpSocket::bind("127.0.0.1:0").expect("binds");
peer.set_read_timeout(Some(Duration::from_millis(500))).unwrap();
let socket = UdpSocket::bind("127.0.0.1:0").expect("binds");
socket.set_read_timeout(Some(Duration::from_millis(300))).unwrap();
let mut options = CultNetRudpSocketTransportOptions::client(
"test-media",
socket,
peer.local_addr().unwrap(),
7,
);
options.media_delivery = delivery;
options.media_reliable_expire_after_ms = expire_after_ms;
options.resend_delay_ms = 10;
let mut client = CultNetRudpSocketTransportConnection::new(options).expect("opens");
client.connect(Vec::new()).expect("sends connect");
let mut wire = vec![0_u8; 2_048];
let (received, client_addr) = peer.recv_from(&mut wire).expect("the connect arrives");
let connect = decode_rudp_packet(&wire[..received]).expect("decodes");
let mut server = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: 7,
..CultNetRudpSessionOptions::default()
});
let accept = server.accept_connect(&connect, 0, Vec::new()).expect("accepts");
peer.send_to(&encode_rudp_packet(&accept).unwrap(), client_addr).unwrap();
client.receive_once().expect("takes the accept");
assert!(client.connected());
let mut link = Self { client, peer };
link.datagrams(Duration::from_millis(100));
link
}
/// Everything the peer receives within `window`.
fn datagrams(&mut self, window: Duration) -> Vec<CultNetRudpPacket> {
self.peer.set_read_timeout(Some(window)).unwrap();
let mut packets = Vec::new();
let mut wire = vec![0_u8; 2_048];
while let Ok((received, _)) = self.peer.recv_from(&mut wire) {
packets.push(decode_rudp_packet(&wire[..received]).expect("decodes"));
}
packets
}
fn send_media(&mut self, window: Duration) -> Vec<CultNetRudpPacket> {
self.client.send("media", vec![1, 2, 3]).expect("sends");
self.datagrams(window)
}
}
fn media_only(packets: Vec<CultNetRudpPacket>) -> Vec<CultNetRudpPacket> {
packets.into_iter().filter(|packet| packet.channel_id == "media").collect()
}
/// The channel the profile describes as reliable is sent reliable, unordered
/// and unsequenced; asking for an unreliable lane sends it unreliable.
#[test]
fn the_media_channel_is_sent_with_the_delivery_the_profile_advertises() {
let mut reliable = Link::open(None, None);
let sent = media_only(reliable.send_media(Duration::from_millis(100)));
assert_eq!(sent.len(), 1);
assert!(sent[0].reliable, "the default media lane is reliable");
assert!(!sent[0].ordered && !sent[0].sequenced, "media is neither ordered nor sequenced");
let mut unreliable = Link::open(Some(CultNetTransportDelivery::Unreliable), None);
let sent = media_only(unreliable.send_media(Duration::from_millis(100)));
assert_eq!(sent.len(), 1);
assert!(!sent[0].reliable, "an unreliable media lane is not retransmitted");
assert!(!sent[0].ordered && !sent[0].sequenced);
}
/// A reliable media packet with an expiry is dropped once it has passed, and
/// one without is retransmitted until acknowledged. The session reclaims
/// expired packets on the next send, so the probe sends a second packet.
fn first_packet_is_resent_after_a_later_send(expire_after_ms: Option<u64>) -> bool {
let mut link = Link::open(None, expire_after_ms);
let first = media_only(link.send_media(Duration::from_millis(50)));
assert_eq!(first.len(), 1);
sleep(Duration::from_millis(150));
assert_eq!(media_only(link.send_media(Duration::from_millis(50))).len(), 1);
sleep(Duration::from_millis(15));
link.client.poll_resends().unwrap();
media_only(link.datagrams(Duration::from_millis(100)))
.iter()
.any(|packet| packet.sequence == first[0].sequence)
}
#[test]
fn the_media_channels_expiry_is_applied_to_its_packets() {
assert!(
!first_packet_is_resent_after_a_later_send(Some(20)),
"an expired media packet is dropped, not retransmitted"
);
assert!(
first_packet_is_resent_after_a_later_send(None),
"without an expiry the unacknowledged packet is retransmitted"
);
}

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,208 @@
//! Video frames protected by the media FEC codec, sent through the impairment
//! schedules `cultnet_impair` applies to real sockets, with no sockets: each
//! record of a frame in send order takes the next datagram decision, and what
//! survives is handed to `recover_video_block`.
//!
//! Two kinds of claim. The guarantee is asserted: a block recovers exactly when
//! at most `m` of its records were lost, and a burst short enough that the send
//! order spreads it under `m` per block always leaves every frame whole. The
//! recovered-frame ratios are evidence, not assertions; `--nocapture` prints the
//! table and `tests/impairment/loss-matrix.txt` is that output (200 frames per cell, seed 20260930).
#[allow(dead_code)]
#[path = "../examples/cultnet_impair/policy.rs"]
mod policy;
use std::fmt::Write as _;
use std::path::Path;
use cultnet_rs::{
GameCultMediaVideoAccessUnitRecord, GameCultMediaWireRecord, MediaFecError, MediaFecPolicy,
MediaWireProvenance, protect_video_frame, recover_video_block,
};
use policy::{Profile, Scheduler};
const POLICY: MediaFecPolicy = MediaFecPolicy::STANDARD;
const SEED: u64 = 20_260_930;
const FRAMES: u64 = 200;
fn provenance() -> MediaWireProvenance<'static> {
MediaWireProvenance {
stored_at: "unix:1700000000000",
runtime_id: "raven-muninn-primary",
role: "muninn.media",
producer: "muninn",
}
}
fn frame(frame_id: u64, chunk_count: u16) -> 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: frame_id as i64 * 3_000,
duration_ticks: 3_000,
timebase_num: 1,
timebase_den: 90_000,
keyframe: false,
dependency_frame_id: frame_id.checked_sub(1),
deadline_ticks: frame_id as i64 * 3_000 + 22_500,
chunk_index: index,
chunk_count,
payload: (0..if index + 1 == chunk_count { 60 } else { 100 })
.map(|byte| (frame_id as usize * 31 + usize::from(index) * 7 + byte) as u8)
.collect(),
})
.collect()
}
#[derive(Default)]
struct Tally {
frames: u64,
whole_without_parity: u64,
whole_with_parity: u64,
blocks_recovered: u64,
blocks_beyond_repair: u64,
}
/// Runs one shape through one profile's schedule and asserts the per-block
/// guarantee on every block of every frame.
fn run(profile: &Profile, chunk_count: u16) -> Tally {
let mut scheduler = Scheduler::new(profile.clone(), SEED);
let sizes = POLICY.video_block_sizes(chunk_count);
let mut tally = Tally::default();
for frame_id in 0..FRAMES {
let original = frame(frame_id, chunk_count);
let records = protect_video_frame(&original, &POLICY, provenance()).unwrap();
let survivors: Vec<GameCultMediaWireRecord> = records
.into_iter()
.filter(|_| !scheduler.decide(0).drop)
.collect();
let mut whole = true;
let mut whole_without_parity = true;
let mut start = 0_u16;
for (block, &size) in sizes.iter().enumerate() {
let m = usize::from(POLICY.video_parity_shards(size));
let mut data = Vec::new();
let mut parity = Vec::new();
for record in &survivors {
match record {
GameCultMediaWireRecord::Video(chunk)
if chunk.chunk_index >= start && chunk.chunk_index < start + size =>
{
data.push(chunk.clone())
}
GameCultMediaWireRecord::VideoParity(shard) if usize::from(shard.block_index) == block => {
parity.push(shard.clone())
}
_ => {}
}
}
let lost = usize::from(size) + m - data.len() - parity.len();
let missing_data = usize::from(size) - data.len();
whole_without_parity &= missing_data == 0;
let block_whole = if missing_data == 0 {
true
} else if parity.is_empty() {
assert!(lost > m, "no parity survived, so more than m shards were lost");
false
} else {
match recover_video_block(&parity, &data) {
Ok(recovered) => {
assert!(lost <= m, "block {block} of frame {frame_id} recovered {lost} losses with m = {m}");
assert_eq!(recovered.len(), missing_data);
for chunk in &recovered {
assert_eq!(
chunk, &original[usize::from(chunk.chunk_index)],
"a recovered chunk must be the chunk that was sent"
);
}
tally.blocks_recovered += 1;
true
}
Err(MediaFecError::BeyondRepair { .. }) => {
assert!(lost > m, "block {block} of frame {frame_id} refused a recoverable {lost} losses with m = {m}");
tally.blocks_beyond_repair += 1;
false
}
Err(other) => panic!("{other}"),
}
};
whole &= block_whole;
start += size;
}
tally.frames += 1;
tally.whole_with_parity += u64::from(whole);
tally.whole_without_parity += u64::from(whole_without_parity);
assert!(whole || !whole_without_parity, "parity never makes a whole frame worse");
}
tally
}
fn profile(name: &str) -> Profile {
Profile::load(&Path::new(env!("CARGO_MANIFEST_DIR")).join(format!("tests/impairment/{name}.toml")))
.unwrap()
}
/// Every shape here splits into equal blocks, so a burst of `L` consecutive
/// datagrams costs each of `B` blocks at most `ceil(L / B)` of its records.
const EQUAL_BLOCK_SHAPES: [u16; 5] = [8, 16, 24, 32, 48];
#[test]
fn the_loss_matrix() {
let cells = ["clean", "loss-1pct", "loss-3pct", "loss-5pct", "burst-4", "burst-8"];
let shapes = [8_u16, 16, 17, 24, 32, 48];
let mut table = String::new();
writeln!(
table,
"{:<10} {:>6} {:>7} {:>13} {:>12} {:>10} {:>9}",
"profile", "chunks", "frames", "whole/no-fec", "whole/fec", "recovered", "beyond"
)
.unwrap();
for cell in cells {
let loaded = profile(cell);
for &chunks in &shapes {
let tally = run(&loaded, chunks);
writeln!(
table,
"{:<10} {:>6} {:>7} {:>13} {:>12} {:>10} {:>9}",
cell,
chunks,
tally.frames,
tally.whole_without_parity,
tally.whole_with_parity,
tally.blocks_recovered,
tally.blocks_beyond_repair
)
.unwrap();
if cell == "clean" {
assert_eq!(tally.whole_with_parity, tally.frames, "a clean link loses nothing");
}
let burst = match cell {
"burst-4" => 4_usize,
"burst-8" => 8,
_ => continue,
};
if EQUAL_BLOCK_SHAPES.contains(&chunks) {
let sizes = POLICY.video_block_sizes(chunks);
let per_block = burst.div_ceil(sizes.len());
let m = usize::from(POLICY.video_parity_shards(sizes[0]));
if per_block <= m {
assert_eq!(
tally.whole_with_parity, tally.frames,
"{cell} over {chunks} chunks costs each of {} blocks at most {per_block} of m = {m}: no frame may be lost",
sizes.len()
);
}
}
}
}
println!("{table}");
}

View file

@ -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

View file

@ -2,7 +2,10 @@
//! nothing on the wire.
use cultnet_rs::{
GAMECULT_MEDIA_CHANNEL, GameCultMediaAudioPacketRecord, GameCultMediaReceiverFeedbackRecord,
CultNetRudpDeliveredFrame, CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions,
GAMECULT_MEDIA_MAX_WIRE_BYTES, RudpTransportProfileOptions, create_rudp_transport_profile,
GAMECULT_MEDIA_CHANNEL, GameCultMediaAudioPacketRecord, GameCultMediaAudioParityShardRecord,
GameCultMediaReceiverFeedbackRecord,
GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord,
GameCultMediaWireRecord, MediaWireProvenance, decode_media_wire_record,
encode_media_wire_record,
@ -116,12 +119,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 +240,435 @@ 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();
})),
("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)),
("256 shards", Box::new(|r| {
r.block_data_count = 256;
r.chunk_count = 300;
r.parity_count = 1;
r.parity_index = 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)),
("last pts_ticks", Box::new(|r| r.deadline_ticks = -1)),
("packet id or pts range", Box::new(|r| r.base_packet_id = u64::MAX - 2)),
("packet id or pts range", Box::new(|r| {
r.base_pts_ticks = i64::MAX - 960 * 3 + 1;
r.deadline_ticks = i64::MAX;
})),
("declared shard length", Box::new(|r| { r.payload.pop(); })),
("codec", Box::new(|r| r.codec = String::new())),
("stream_id", Box::new(|r| r.stream_id = String::new())),
("stream_id", Box::new(|r| r.session_id = String::new())),
("timing", Box::new(|r| r.timebase_num = 0)),
("timing", Box::new(|r| r.timebase_den = 0)),
];
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:?}");
}
}
/// The 256-shard ceiling and the deadline floor are inclusive on the side that
/// keeps a valid record valid.
#[test]
fn records_on_the_edge_of_each_bound_are_admitted() {
let mut video = video_parity();
video.chunk_count = 300;
video.block_data_count = 254;
video.parity_count = 2;
video.parity_index = 1;
assert!(decodes(GameCultMediaWireRecord::VideoParity(video)).is_ok(), "254 + 2 = 256 shards");
let mut audio = audio_parity();
audio.data_shard_count = 254;
audio.deadline_ticks = i64::MAX;
audio.parity_shard_count = 2;
assert!(decodes(GameCultMediaWireRecord::AudioParity(audio)).is_ok(), "254 + 2 = 256 shards");
let mut audio = audio_parity();
audio.deadline_ticks = audio.base_pts_ticks + 3 * i64::from(audio.packet_duration_ticks);
assert!(decodes(GameCultMediaWireRecord::AudioParity(audio)).is_ok(), "a deadline at the last pts");
let mut audio = audio_parity();
audio.base_packet_id = u64::MAX - 3;
assert!(decodes(GameCultMediaWireRecord::AudioParity(audio)).is_ok(), "the last packet id is u64::MAX");
let mut audio = audio_parity();
audio.base_pts_ticks = i64::MAX - 960 * 3;
audio.deadline_ticks = i64::MAX;
assert!(decodes(GameCultMediaWireRecord::AudioParity(audio)).is_ok(), "the last pts is i64::MAX");
}
// ---------------------------------------------------------------------------
// The ceiling on one media record
// ---------------------------------------------------------------------------
fn sized_audio_parity(payload_bytes: usize, deadline_ticks: i64) -> GameCultMediaAudioParityShardRecord {
GameCultMediaAudioParityShardRecord {
stream_id: "s".to_string(),
session_id: "x".to_string(),
codec: "opus".to_string(),
fec_scheme: "rs-gf256-v1".to_string(),
base_packet_id: 8,
base_pts_ticks: 0,
packet_duration_ticks: 100,
timebase_num: 1,
timebase_den: 48_000,
deadline_ticks,
data_shard_count: 4,
parity_index: 0,
parity_shard_count: 2,
shard_payload_bytes: payload_bytes as u32,
payload: vec![7; payload_bytes],
}
}
fn sized_video_parity(payload_bytes: usize) -> GameCultMediaVideoParityShardRecord {
GameCultMediaVideoParityShardRecord {
stream_id: "s".to_string(),
session_id: "x".to_string(),
frame_id: 1,
codec: "h264".to_string(),
pts_ticks: 0,
duration_ticks: 1,
timebase_num: 1,
timebase_den: 90_000,
keyframe: false,
dependency_frame_id: Some(0),
deadline_ticks: 1,
chunk_count: 4,
fec_scheme: "rs-gf256-v1".to_string(),
block_index: 0,
block_count: 1,
block_data_start: 0,
block_data_count: 4,
parity_index: 1,
parity_count: 2,
shard_payload_bytes: payload_bytes as u32,
last_chunk_payload_bytes: 2,
payload: vec![7; payload_bytes],
}
}
fn wire_len(record: &GameCultMediaWireRecord) -> usize {
encode_media_wire_record(record, provenance()).unwrap().len()
}
/// The payload length whose wrapped record is exactly `wire_bytes` long, found
/// from the record's own linear growth in its payload.
fn payload_for_wire_len(
wire_bytes: usize,
wrap: &dyn Fn(usize) -> GameCultMediaWireRecord,
) -> usize {
let probe = 1_000;
let payload = probe + wire_bytes - wire_len(&wrap(probe));
assert_eq!(wire_len(&wrap(payload)), wire_bytes, "the record grows one byte per payload byte here");
payload
}
/// RUDP would reassemble a 16 MiB record, and recovery work scales with the
/// shard length, so the decoder holds every record to one ceiling.
#[test]
fn a_media_record_one_byte_over_the_ceiling_is_refused_at_decode() {
let ceiling = GAMECULT_MEDIA_MAX_WIRE_BYTES;
let wraps: [(&str, Box<dyn Fn(usize) -> GameCultMediaWireRecord>); 2] = [
(
"audio parity",
Box::new(|payload| GameCultMediaWireRecord::AudioParity(sized_audio_parity(payload, 300))),
),
(
"video parity",
Box::new(|payload| GameCultMediaWireRecord::VideoParity(sized_video_parity(payload))),
),
];
for (name, wrap) in &wraps {
let at = payload_for_wire_len(ceiling, wrap.as_ref());
let wire = encode_media_wire_record(&wrap(at), provenance()).unwrap();
assert_eq!(wire.len(), ceiling);
decode_media_wire_record(&wire)
.unwrap_or_else(|error| panic!("{name} at the ceiling is refused: {error}"));
// The encoder refuses to make an over-ceiling record, so one byte of
// padding stands in for a peer that sent one.
let mut over = wire.clone();
over.push(0);
let error = decode_media_wire_record(&over).unwrap_err();
assert!(error.to_string().contains("ceiling"), "{name}: {error}");
}
}
#[test]
fn a_jumbo_frame_sized_record_is_accepted() {
let wire = encode_media_wire_record(
&GameCultMediaWireRecord::AudioParity(sized_audio_parity(8_800, 300)),
provenance(),
)
.unwrap();
assert!(wire.len() > 8_800 && wire.len() < GAMECULT_MEDIA_MAX_WIRE_BYTES);
decode_media_wire_record(&wire).expect("a jumbo-frame record decodes");
}
/// Recovery hands every recovered packet the block's deadline, so a parity
/// whose deadline precedes the last packet's pts would recover a packet that
/// fails `validate_audio_record`.
#[test]
fn audio_parity_must_not_have_a_deadline_before_its_last_packets_pts() {
// Four packets, 100 ticks each from 0: the last plays at 300.
let decodes = |deadline_ticks: i64| {
let wire = encode_media_wire_record(
&GameCultMediaWireRecord::AudioParity(sized_audio_parity(8, deadline_ticks)),
provenance(),
)
.unwrap();
decode_media_wire_record(&wire)
};
assert!(decodes(300).is_ok());
assert!(decodes(301).is_ok());
let error = decodes(299).unwrap_err();
assert!(error.to_string().contains("last pts_ticks"), "{error}");
let error = decodes(0).unwrap_err();
assert!(error.to_string().contains("last pts_ticks"), "{error}");
}
// ---------------------------------------------------------------------------
// The media channel's own payload cap
// ---------------------------------------------------------------------------
fn connected_session() -> CultNetRudpSession {
let mut session = CultNetRudpSession::new(CultNetRudpSessionOptions {
connection_id: 9,
..CultNetRudpSessionOptions::default()
});
session.assume_connected(0);
session
}
const FRAGMENT_BYTES: usize = 16 * 1024;
fn media_payload_limit(session_limit: Option<u32>) -> Option<u32> {
let profile = create_rudp_transport_profile(
"raven",
RudpTransportProfileOptions {
max_payload_bytes: session_limit,
..RudpTransportProfileOptions::default()
},
);
profile.transports[0]
.channels
.iter()
.find(|channel| channel.channel_id == GAMECULT_MEDIA_CHANNEL)
.expect("the profile advertises the media channel")
.max_payload_bytes
}
#[test]
fn the_media_channel_advertises_the_ceiling_or_a_tighter_session_cap() {
let ceiling = GAMECULT_MEDIA_MAX_WIRE_BYTES as u32;
assert_eq!(media_payload_limit(None), Some(ceiling));
assert_eq!(media_payload_limit(Some(16 * 1024 * 1024)), Some(ceiling));
assert_eq!(media_payload_limit(Some(4_096)), Some(4_096));
}
#[test]
fn a_session_refuses_to_send_more_than_the_ceiling_on_the_media_channel_only() {
let ceiling = GAMECULT_MEDIA_MAX_WIRE_BYTES;
let send = |channel: &str, payload: usize| {
connected_session().send_many(
channel,
vec![0; payload],
CultNetRudpSendOptions::default(),
Some(FRAGMENT_BYTES),
)
};
assert!(send(GAMECULT_MEDIA_CHANNEL, ceiling).is_ok());
let error = send(GAMECULT_MEDIA_CHANNEL, ceiling + 1).unwrap_err();
assert!(error.to_string().contains("max_payload_bytes"), "{error}");
assert!(send("schema", ceiling + 1).is_ok(), "other channels keep the session cap");
}
/// Reassembly is where memory is spent, so the receiver enforces the ceiling
/// itself, not only the sender.
#[test]
fn a_session_refuses_to_reassemble_more_than_the_ceiling_on_the_media_channel() {
let ceiling = GAMECULT_MEDIA_MAX_WIRE_BYTES;
let packets = connected_session()
.send_many(
"schema",
vec![0; ceiling + 1],
CultNetRudpSendOptions::default(),
Some(FRAGMENT_BYTES),
)
.unwrap();
assert!(packets.len() > 1);
let deliver = |channel: &str| -> anyhow::Result<Vec<CultNetRudpDeliveredFrame>> {
let mut receiver = connected_session();
let mut delivered = Vec::new();
for packet in &packets {
let mut packet = packet.clone();
packet.channel_id = channel.to_string();
delivered.extend(receiver.receive(&packet, 0)?.delivered);
}
Ok(delivered)
};
let frames = deliver("schema").expect("the same fragments reassemble on another channel");
assert_eq!(frames.len(), 1);
assert_eq!(frames[0].payload.len(), ceiling + 1);
let error = deliver(GAMECULT_MEDIA_CHANNEL).unwrap_err();
assert!(error.to_string().contains("max_payload_bytes"), "{error}");
}
/// A producer learns at encode time that a record is over the ceiling, rather
/// than the receiver dropping it.
#[test]
fn the_encoder_refuses_a_record_over_the_ceiling() {
let wrap = |payload: usize| GameCultMediaWireRecord::AudioParity(sized_audio_parity(payload, 300));
let at = payload_for_wire_len(GAMECULT_MEDIA_MAX_WIRE_BYTES, &wrap);
assert!(encode_media_wire_record(&wrap(at), provenance()).is_ok());
let error = encode_media_wire_record(&wrap(at + 1), provenance()).unwrap_err();
assert!(error.to_string().contains("ceiling"), "{error}");
}
/// A record that arrives unfragmented is held to the same ceiling.
#[test]
fn a_session_refuses_an_unfragmented_media_packet_over_the_ceiling() {
let packets = connected_session()
.send_many(
"schema",
vec![0; GAMECULT_MEDIA_MAX_WIRE_BYTES + 1],
CultNetRudpSendOptions::default(),
None,
)
.unwrap();
assert_eq!(packets.len(), 1, "sent whole");
let mut packet = packets[0].clone();
assert!(connected_session().receive(&packet, 0).is_ok());
packet.channel_id = GAMECULT_MEDIA_CHANNEL.to_string();
let error = connected_session().receive(&packet, 0).unwrap_err();
assert!(error.to_string().contains("max_payload_bytes"), "{error}");
}