Media wire: bound every record by one ceiling, and validate audio parity against its last pts
Removing the 720-byte shard cap left the decoder with no per-record size bound: RUDP reassembles up to 16 MiB and the media channel set no cap, so a 4 MiB parity record decoded and recovery work scaled with it. GAMECULT_MEDIA_MAX_WIRE_BYTES (64 KiB, one maximum UDP datagram, ~7x a jumbo frame) is now the ceiling. decode_media_wire_record refuses a larger record, the RUDP session caps the media channel to it on send and on reassembly (and the profile advertises it), and MediaFecPolicy::validate refuses a max_wire_bytes above it so no producer can promise a record the decoder drops. validate_audio_parity_record now checks the deadline against the last slot's pts, not the base, so recovery can no longer produce a packet that violates deadline_ticks >= pts_ticks. Tests: ceiling +-1 at decode for audio and video parity; jumbo record accepted; session send and reassembly on the media channel only; profile advertisement; the audio deadline boundary and recovery refusal; a policy over the ceiling refused; a budget at the ceiling end to end; audio parity larger than every packet it protects (the budget check relies on it). MediaFecPolicy docs tell callers to build by struct update from STANDARD. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
b90fdc45aa
commit
51e9a550f9
5 changed files with 381 additions and 21 deletions
|
|
@ -95,7 +95,7 @@ use crate::media_stream_contracts::{
|
|||
GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord,
|
||||
};
|
||||
use crate::media_stream_wire::{
|
||||
GAMECULT_MEDIA_CHANNEL, GameCultMediaWireRecord, MediaWireProvenance,
|
||||
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,
|
||||
};
|
||||
|
|
@ -118,6 +118,13 @@ pub const MEDIA_FEC_MAX_WIRE_BYTES: usize =
|
|||
/// 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.
|
||||
|
|
@ -183,6 +190,12 @@ impl MediaFecPolicy {
|
|||
{
|
||||
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(())
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -42,6 +42,19 @@ use crate::media_stream_contracts::{
|
|||
/// choice per session; stale media is dropped, not retransmitted.
|
||||
pub const GAMECULT_MEDIA_CHANNEL: &str = "media";
|
||||
|
||||
/// 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)]
|
||||
pub enum GameCultMediaWireRecord {
|
||||
|
|
@ -174,6 +187,12 @@ pub fn encode_media_wire_record(
|
|||
|
||||
/// 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"));
|
||||
|
|
@ -434,11 +453,6 @@ pub fn validate_audio_parity_record(record: &GameCultMediaAudioParityShardRecord
|
|||
"audio parity media record timing metadata must be non-zero"
|
||||
));
|
||||
}
|
||||
if record.deadline_ticks < record.base_pts_ticks {
|
||||
return Err(anyhow!(
|
||||
"audio parity media record deadline_ticks must not precede base_pts_ticks"
|
||||
));
|
||||
}
|
||||
if record.fec_scheme != MEDIA_FEC_SCHEME_RS_GF256_V1 {
|
||||
return Err(anyhow!(
|
||||
"audio parity media record fec_scheme {:?} is not supported",
|
||||
|
|
@ -452,17 +466,23 @@ pub fn validate_audio_parity_record(record: &GameCultMediaAudioParityShardRecord
|
|||
{
|
||||
return Err(anyhow!("audio parity media record stripe metadata is invalid"));
|
||||
}
|
||||
if audio_slot(
|
||||
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,
|
||||
)
|
||||
.is_none()
|
||||
{
|
||||
) 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!(
|
||||
|
|
|
|||
|
|
@ -23,6 +23,8 @@ 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;
|
||||
|
|
@ -429,7 +431,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);
|
||||
|
|
@ -958,7 +960,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(),
|
||||
|
|
@ -1010,7 +1012,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);
|
||||
}
|
||||
|
|
@ -1068,11 +1070,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(())
|
||||
|
|
@ -2156,6 +2159,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,
|
||||
|
|
@ -2205,12 +2220,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_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,
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@
|
|||
use cultnet_rs::{
|
||||
GameCultMediaAudioPacketRecord, GameCultMediaAudioParityShardRecord,
|
||||
GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord,
|
||||
GameCultMediaWireRecord, MEDIA_FEC_MAX_WIRE_BYTES,
|
||||
GAMECULT_MEDIA_MAX_WIRE_BYTES, GameCultMediaWireRecord, MEDIA_FEC_MAX_WIRE_BYTES,
|
||||
MEDIA_FEC_SCHEME_RS_GF256_V1, MediaFecError, MediaFecPolicy, MediaWireProvenance,
|
||||
decode_media_wire_record, encode_media_wire_record, protect_audio_block, protect_video_frame,
|
||||
recover_audio_block, recover_video_block,
|
||||
|
|
@ -1052,3 +1052,92 @@ fn the_worst_burst_that_always_recovers_is_blocks_times_the_smallest_m() {
|
|||
);
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// The ceiling, and what the audio budget check relies on
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Only the parity is measured against the budget, on the claim that it is the
|
||||
/// largest record of an audio block. It carries every field a packet does plus
|
||||
/// the stripe geometry, so this holds across the id, pts, duration and deadline
|
||||
/// widths that vary the encoding.
|
||||
#[test]
|
||||
fn an_audio_parity_record_is_larger_on_the_wire_than_every_packet_it_protects() {
|
||||
let bases: [u64; 8] = [0, 0x7c, 0xfc, 0xfffc, 0xffff_fffc, 99_997, u64::MAX - 3, 1];
|
||||
let starts: [i64; 7] = [i64::MIN, -1, 0, 125, 65_530, 4_294_967_290, i64::MAX - 3 * 4_000_000_000];
|
||||
let durations: [u32; 5] = [1, 120, 960, 65_535, 4_000_000_000];
|
||||
for base in bases {
|
||||
for start in starts {
|
||||
for duration in durations {
|
||||
for wide_deadline in [false, true] {
|
||||
let mut packets = audio_block(base, 40);
|
||||
for (index, packet) in packets.iter_mut().enumerate() {
|
||||
packet.pts_ticks = start + i64::from(duration) * index as i64;
|
||||
packet.duration_ticks = duration;
|
||||
packet.deadline_ticks = if wide_deadline { i64::MAX } else { packet.pts_ticks };
|
||||
}
|
||||
let parity = protect_audio_block(&packets, &STANDARD, provenance()).unwrap();
|
||||
let parity_len = wire_len(&GameCultMediaWireRecord::AudioParity(parity[0].clone()), provenance());
|
||||
for packet in &packets {
|
||||
let packet_len = wire_len(&GameCultMediaWireRecord::Audio(packet.clone()), provenance());
|
||||
assert!(
|
||||
packet_len < parity_len,
|
||||
"base {base} start {start} duration {duration}: packet {packet_len} >= parity {parity_len}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A parity whose deadline precedes its last packet's pts would recover a packet
|
||||
/// that fails `validate_audio_record`. Recovery refuses it, and so does the wire.
|
||||
#[test]
|
||||
fn audio_parity_with_a_deadline_before_its_last_packet_recovers_nothing() {
|
||||
let packets = audio_block(10, 8);
|
||||
let mut parity = protect_audio_block(&packets, &STANDARD, provenance()).unwrap();
|
||||
for shard in &mut parity {
|
||||
shard.deadline_ticks = shard.base_pts_ticks;
|
||||
}
|
||||
let error = recover_audio_block(&parity, &packets[..2]).unwrap_err();
|
||||
assert!(is_invalid_mentioning(&error, "last pts_ticks"), "{error}");
|
||||
let wire = encode_media_wire_record(
|
||||
&GameCultMediaWireRecord::AudioParity(parity[0].clone()),
|
||||
provenance(),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(decode_media_wire_record(&wire).is_err());
|
||||
}
|
||||
|
||||
/// The datagram budget is the producer's, but no budget may promise a record the
|
||||
/// decoder would refuse.
|
||||
#[test]
|
||||
fn a_datagram_budget_over_the_media_record_ceiling_is_refused() {
|
||||
let over = MediaFecPolicy { max_wire_bytes: GAMECULT_MEDIA_MAX_WIRE_BYTES + 1, ..STANDARD };
|
||||
let error = protect_audio_block(&audio_block(1, 100), &over, provenance()).unwrap_err();
|
||||
assert!(is_invalid_mentioning(&error, "ceiling"), "{error}");
|
||||
let error = protect_video_frame(&frame(2, 3, 100, 100), &over, provenance()).unwrap_err();
|
||||
assert!(is_invalid_mentioning(&error, "ceiling"), "{error}");
|
||||
}
|
||||
|
||||
/// A budget at the ceiling admits shards up to it, and the decoder accepts
|
||||
/// what the policy admitted.
|
||||
#[test]
|
||||
fn a_budget_at_the_media_record_ceiling_admits_large_shards_end_to_end() {
|
||||
let at_ceiling = MediaFecPolicy { max_wire_bytes: GAMECULT_MEDIA_MAX_WIRE_BYTES, ..STANDARD };
|
||||
let records =
|
||||
protect_video_frame(&frame(2, 3, 60_000, 60_000), &at_ceiling, provenance()).unwrap();
|
||||
for record in &records {
|
||||
let wire = encode_media_wire_record(record, provenance()).unwrap();
|
||||
assert!(wire.len() <= GAMECULT_MEDIA_MAX_WIRE_BYTES);
|
||||
decode_media_wire_record(&wire).unwrap();
|
||||
}
|
||||
let parity = protect_audio_block(&audio_block(1, 60_000), &at_ceiling, provenance()).unwrap();
|
||||
let wire = encode_media_wire_record(
|
||||
&GameCultMediaWireRecord::AudioParity(parity[0].clone()),
|
||||
provenance(),
|
||||
)
|
||||
.unwrap();
|
||||
decode_media_wire_record(&wire).unwrap();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,6 +2,8 @@
|
|||
//! nothing on the wire.
|
||||
|
||||
use cultnet_rs::{
|
||||
CultNetRudpDeliveredFrame, CultNetRudpSendOptions, CultNetRudpSession, CultNetRudpSessionOptions,
|
||||
GAMECULT_MEDIA_MAX_WIRE_BYTES, RudpTransportProfileOptions, create_rudp_transport_profile,
|
||||
GAMECULT_MEDIA_CHANNEL, GameCultMediaAudioPacketRecord, GameCultMediaAudioParityShardRecord,
|
||||
GameCultMediaReceiverFeedbackRecord,
|
||||
GameCultMediaVideoAccessUnitRecord, GameCultMediaVideoParityShardRecord,
|
||||
|
|
@ -420,3 +422,220 @@ fn records_on_the_edge_of_each_bound_are_admitted() {
|
|||
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 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 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(audio_parity(payload, 300))),
|
||||
),
|
||||
(
|
||||
"video parity",
|
||||
Box::new(|payload| GameCultMediaWireRecord::VideoParity(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}"));
|
||||
|
||||
let over = encode_media_wire_record(&wrap(at + 1), provenance()).unwrap();
|
||||
assert_eq!(over.len(), ceiling + 1);
|
||||
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(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(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}");
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue