1//! Bounded receiver feedback and adaptive-bitrate decisions.2//!3//! The feedback ALPN uses one direction for receiver reports and the other for4//! host decisions. Both records are fixed width so a peer cannot influence an5//! allocation or leave a decoder waiting for a peer-declared payload length.67use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};89use crate::{PROTOCOL_VERSION, ProtocolError, Result};1011const MEDIA_FEEDBACK_REPORT_V1_MAGIC: [u8; 4] = *b"SGR1";12const ADAPTIVE_BITRATE_DECISION_V1_MAGIC: [u8; 4] = *b"SGD1";1314/// Size of one receiver feedback report in bytes.15pub const MEDIA_FEEDBACK_REPORT_V1_LEN: usize = 64;16const MEDIA_FEEDBACK_REPORT_V1_LEN_U16: u16 = MEDIA_FEEDBACK_REPORT_V1_LEN as u16;17/// Size of one host adaptive-bitrate decision in bytes.18pub const ADAPTIVE_BITRATE_DECISION_V1_LEN: usize = 40;19const ADAPTIVE_BITRATE_DECISION_V1_LEN_U16: u16 = ADAPTIVE_BITRATE_DECISION_V1_LEN as u16;2021const MIN_FEEDBACK_INTERVAL_MS: u16 = 250;22const MAX_FEEDBACK_INTERVAL_MS: u16 = 5_000;23const MAX_FEEDBACK_QUEUE_CAPACITY: u8 = 16;24const OPTIONAL_SEQUENCE_NONE: u64 = u64::MAX;25const OPTIONAL_LATENCY_NONE: u16 = u16::MAX;26const MAX_FEEDBACK_LATENCY_MS: u16 = 60_000;27const MIN_ADAPTIVE_BITRATE_KBPS: u32 = 1_000;28const MAX_ADAPTIVE_BITRATE_KBPS: u32 = 100_000;2930/// State flags reported by Portal.3132pub struct MediaFeedbackFlags(u8);3334impl MediaFeedbackFlags {35 pub const NONE: Self = Self(0);36 pub const RESYNC_ACTIVE: Self = Self(1 << 0);3738 const KNOWN: u8 = Self::RESYNC_ACTIVE.0;3940 pub const fn from_bits(bits: u8) -> Option<Self> {41 if bits & !Self::KNOWN == 0 {42 Some(Self(bits))43 } else {44 None45 }46 }4748 pub const fn bits(self) -> u8 {49 self.050 }5152 pub const fn contains(self, flag: Self) -> bool {53 self.0 & flag.0 == flag.054 }5556 pub const fn union(self, other: Self) -> Self {57 Self(self.0 | other.0)58 }59}6061/// One bounded snapshot of receiver-side video pressure.6263pub struct MediaFeedbackReportV1 {64 pub report_id: u64,65 pub interval_ms: u16,66 pub flags: MediaFeedbackFlags,67 pub last_sequence: Option<u64>,68 pub transport_dropped_delta: u32,69 pub frontend_dropped_delta: u32,70 pub decoder_dropped_delta: u32,71 pub presenter_dropped_delta: u32,72 pub frontend_queue_depth: u8,73 pub frontend_queue_capacity: u8,74 pub decode_queue_depth: u8,75 pub decode_queue_capacity: u8,76 pub presenter_queue_depth: u8,77 pub presenter_queue_capacity: u8,78 pub transport_delivery_p95_ms: Option<u16>,79 pub decode_p95_ms: Option<u16>,80 pub presentation_p95_ms: Option<u16>,81}8283impl MediaFeedbackReportV1 {84 pub fn validate(&self) -> Result<()> {85 if self.report_id == 0 {86 return Err(invalid_report("report ID must be nonzero"));87 }88 if !(MIN_FEEDBACK_INTERVAL_MS..=MAX_FEEDBACK_INTERVAL_MS).contains(&self.interval_ms) {89 return Err(invalid_report("interval must be 250..=5000 milliseconds"));90 }91 if MediaFeedbackFlags::from_bits(self.flags.bits()).is_none() {92 return Err(invalid_report("unsupported feedback flag bits"));93 }94 if self.last_sequence == Some(OPTIONAL_SEQUENCE_NONE) {95 return Err(invalid_report(96 "the all-ones last sequence is reserved for none",97 ));98 }99 validate_queue(self.frontend_queue_depth, self.frontend_queue_capacity)?;100 validate_queue(self.decode_queue_depth, self.decode_queue_capacity)?;101 validate_queue(self.presenter_queue_depth, self.presenter_queue_capacity)?;102 for latency in [103 self.transport_delivery_p95_ms,104 self.decode_p95_ms,105 self.presentation_p95_ms,106 ] {107 if latency.is_some_and(|latency| latency > MAX_FEEDBACK_LATENCY_MS) {108 return Err(invalid_report(109 "p95 latency must be at most 60000 milliseconds",110 ));111 }112 }113 Ok(())114 }115116 pub fn encode(&self) -> Result<[u8; MEDIA_FEEDBACK_REPORT_V1_LEN]> {117 self.validate()?;118 let mut wire = [0_u8; MEDIA_FEEDBACK_REPORT_V1_LEN];119 wire[0..4].copy_from_slice(&MEDIA_FEEDBACK_REPORT_V1_MAGIC);120 wire[4..6].copy_from_slice(&PROTOCOL_VERSION.to_be_bytes());121 wire[6..8].copy_from_slice(&MEDIA_FEEDBACK_REPORT_V1_LEN_U16.to_be_bytes());122 wire[8..16].copy_from_slice(&self.report_id.to_be_bytes());123 wire[16..18].copy_from_slice(&self.interval_ms.to_be_bytes());124 wire[18] = self.flags.bits();125 // Byte 19 is reserved and remains zero.126 wire[20..28].copy_from_slice(127 &self128 .last_sequence129 .unwrap_or(OPTIONAL_SEQUENCE_NONE)130 .to_be_bytes(),131 );132 wire[28..32].copy_from_slice(&self.transport_dropped_delta.to_be_bytes());133 wire[32..36].copy_from_slice(&self.frontend_dropped_delta.to_be_bytes());134 wire[36..40].copy_from_slice(&self.decoder_dropped_delta.to_be_bytes());135 wire[40..44].copy_from_slice(&self.presenter_dropped_delta.to_be_bytes());136 wire[44] = self.frontend_queue_depth;137 wire[45] = self.frontend_queue_capacity;138 wire[46] = self.decode_queue_depth;139 wire[47] = self.decode_queue_capacity;140 wire[48] = self.presenter_queue_depth;141 wire[49] = self.presenter_queue_capacity;142 wire[50..52].copy_from_slice(143 &self144 .transport_delivery_p95_ms145 .unwrap_or(OPTIONAL_LATENCY_NONE)146 .to_be_bytes(),147 );148 wire[52..54].copy_from_slice(149 &self150 .decode_p95_ms151 .unwrap_or(OPTIONAL_LATENCY_NONE)152 .to_be_bytes(),153 );154 wire[54..56].copy_from_slice(155 &self156 .presentation_p95_ms157 .unwrap_or(OPTIONAL_LATENCY_NONE)158 .to_be_bytes(),159 );160 // Bytes 56..64 are reserved and remain zero.161 Ok(wire)162 }163164 pub fn decode(wire: &[u8; MEDIA_FEEDBACK_REPORT_V1_LEN]) -> Result<Self> {165 let magic: [u8; 4] = wire[0..4].try_into().expect("fixed slice length");166 if magic != MEDIA_FEEDBACK_REPORT_V1_MAGIC {167 return Err(invalid_report("invalid report magic"));168 }169 validate_fixed_prefix(170 wire,171 MEDIA_FEEDBACK_REPORT_V1_LEN_U16,172 "media feedback report",173 )?;174 if wire[19] != 0 || wire[56..64].iter().any(|byte| *byte != 0) {175 return Err(invalid_report("reserved fields must be zero"));176 }177 let flags = MediaFeedbackFlags::from_bits(wire[18])178 .ok_or_else(|| invalid_report("unsupported feedback flag bits"))?;179 let last_sequence =180 u64::from_be_bytes(wire[20..28].try_into().expect("fixed slice length"));181 let report = Self {182 report_id: u64::from_be_bytes(wire[8..16].try_into().expect("fixed slice length")),183 interval_ms: u16::from_be_bytes(wire[16..18].try_into().expect("fixed slice length")),184 flags,185 last_sequence: (last_sequence != OPTIONAL_SEQUENCE_NONE).then_some(last_sequence),186 transport_dropped_delta: u32::from_be_bytes(187 wire[28..32].try_into().expect("fixed slice length"),188 ),189 frontend_dropped_delta: u32::from_be_bytes(190 wire[32..36].try_into().expect("fixed slice length"),191 ),192 decoder_dropped_delta: u32::from_be_bytes(193 wire[36..40].try_into().expect("fixed slice length"),194 ),195 presenter_dropped_delta: u32::from_be_bytes(196 wire[40..44].try_into().expect("fixed slice length"),197 ),198 frontend_queue_depth: wire[44],199 frontend_queue_capacity: wire[45],200 decode_queue_depth: wire[46],201 decode_queue_capacity: wire[47],202 presenter_queue_depth: wire[48],203 presenter_queue_capacity: wire[49],204 transport_delivery_p95_ms: decode_optional_latency(wire, 50),205 decode_p95_ms: decode_optional_latency(wire, 52),206 presentation_p95_ms: decode_optional_latency(wire, 54),207 };208 report.validate()?;209 Ok(report)210 }211}212213/// Direction selected by the host controller.214215216pub enum AdaptiveBitrateStateV1 {217 Hold = 0,218 Decrease = 1,219 Increase = 2,220}221222impl TryFrom<u8> for AdaptiveBitrateStateV1 {223 type Error = ProtocolError;224225 fn try_from(value: u8) -> Result<Self> {226 match value {227 0 => Ok(Self::Hold),228 1 => Ok(Self::Decrease),229 2 => Ok(Self::Increase),230 _ => Err(invalid_decision("unsupported controller state")),231 }232 }233}234235/// Bounded reasons contributing to a bitrate decision.236237pub struct AdaptiveBitrateReasonFlagsV1(u16);238239impl AdaptiveBitrateReasonFlagsV1 {240 pub const NONE: Self = Self(0);241 pub const RTT_INFLATION: Self = Self(1 << 0);242 pub const LOSS_OR_CANCELLATION: Self = Self(1 << 1);243 pub const SENDER_BACKPRESSURE: Self = Self(1 << 2);244 pub const RECEIVER_QUEUE: Self = Self(1 << 3);245 pub const DECODE_BACKLOG: Self = Self(1 << 4);246 pub const DELIVERY_LATENCY: Self = Self(1 << 5);247 pub const CLEAN_RECOVERY: Self = Self(1 << 6);248 pub const FEEDBACK_STALE: Self = Self(1 << 7);249250 const KNOWN: u16 = Self::RTT_INFLATION.0251 | Self::LOSS_OR_CANCELLATION.0252 | Self::SENDER_BACKPRESSURE.0253 | Self::RECEIVER_QUEUE.0254 | Self::DECODE_BACKLOG.0255 | Self::DELIVERY_LATENCY.0256 | Self::CLEAN_RECOVERY.0257 | Self::FEEDBACK_STALE.0;258259 pub const fn from_bits(bits: u16) -> Option<Self> {260 if bits & !Self::KNOWN == 0 {261 Some(Self(bits))262 } else {263 None264 }265 }266267 pub const fn bits(self) -> u16 {268 self.0269 }270271 pub const fn contains(self, flag: Self) -> bool {272 self.0 & flag.0 == flag.0273 }274275 pub const fn union(self, other: Self) -> Self {276 Self(self.0 | other.0)277 }278}279280/// One bounded host controller decision corresponding to a receiver report.281282pub struct AdaptiveBitrateDecisionV1 {283 pub decision_id: u64,284 pub report_id: u64,285 pub target_kbps: u32,286 pub floor_kbps: u32,287 pub ceiling_kbps: u32,288 pub state: AdaptiveBitrateStateV1,289 pub reasons: AdaptiveBitrateReasonFlagsV1,290 pub applied: bool,291}292293impl AdaptiveBitrateDecisionV1 {294 pub fn validate(&self) -> Result<()> {295 if self.decision_id == 0 {296 return Err(invalid_decision("decision ID must be nonzero"));297 }298 if self.report_id == 0 {299 return Err(invalid_decision("report ID must be nonzero"));300 }301 for bitrate in [self.target_kbps, self.floor_kbps, self.ceiling_kbps] {302 if !(MIN_ADAPTIVE_BITRATE_KBPS..=MAX_ADAPTIVE_BITRATE_KBPS).contains(&bitrate) {303 return Err(invalid_decision(304 "bitrates must be between 1000 and 100000 kilobits per second",305 ));306 }307 }308 if self.floor_kbps > self.target_kbps || self.target_kbps > self.ceiling_kbps {309 return Err(invalid_decision(310 "bitrate floor must not exceed target and target must not exceed ceiling",311 ));312 }313 if AdaptiveBitrateReasonFlagsV1::from_bits(self.reasons.bits()).is_none() {314 return Err(invalid_decision("unsupported decision reason bits"));315 }316 Ok(())317 }318319 pub fn encode(&self) -> Result<[u8; ADAPTIVE_BITRATE_DECISION_V1_LEN]> {320 self.validate()?;321 let mut wire = [0_u8; ADAPTIVE_BITRATE_DECISION_V1_LEN];322 wire[0..4].copy_from_slice(&ADAPTIVE_BITRATE_DECISION_V1_MAGIC);323 wire[4..6].copy_from_slice(&PROTOCOL_VERSION.to_be_bytes());324 wire[6..8].copy_from_slice(&ADAPTIVE_BITRATE_DECISION_V1_LEN_U16.to_be_bytes());325 wire[8..16].copy_from_slice(&self.decision_id.to_be_bytes());326 wire[16..24].copy_from_slice(&self.report_id.to_be_bytes());327 wire[24..28].copy_from_slice(&self.target_kbps.to_be_bytes());328 wire[28..32].copy_from_slice(&self.floor_kbps.to_be_bytes());329 wire[32..36].copy_from_slice(&self.ceiling_kbps.to_be_bytes());330 wire[36] = self.state as u8;331 wire[37] = u8::from(self.applied);332 wire[38..40].copy_from_slice(&self.reasons.bits().to_be_bytes());333 Ok(wire)334 }335336 pub fn decode(wire: &[u8; ADAPTIVE_BITRATE_DECISION_V1_LEN]) -> Result<Self> {337 let magic: [u8; 4] = wire[0..4].try_into().expect("fixed slice length");338 if magic != ADAPTIVE_BITRATE_DECISION_V1_MAGIC {339 return Err(invalid_decision("invalid decision magic"));340 }341 validate_fixed_prefix(342 wire,343 ADAPTIVE_BITRATE_DECISION_V1_LEN_U16,344 "adaptive bitrate decision",345 )?;346 if wire[37] > 1 {347 return Err(invalid_decision("applied must be encoded as zero or one"));348 }349 let reason_bits = u16::from_be_bytes(wire[38..40].try_into().expect("fixed slice length"));350 let decision = Self {351 decision_id: u64::from_be_bytes(wire[8..16].try_into().expect("fixed slice length")),352 report_id: u64::from_be_bytes(wire[16..24].try_into().expect("fixed slice length")),353 target_kbps: u32::from_be_bytes(wire[24..28].try_into().expect("fixed slice length")),354 floor_kbps: u32::from_be_bytes(wire[28..32].try_into().expect("fixed slice length")),355 ceiling_kbps: u32::from_be_bytes(wire[32..36].try_into().expect("fixed slice length")),356 state: AdaptiveBitrateStateV1::try_from(wire[36])?,357 reasons: AdaptiveBitrateReasonFlagsV1::from_bits(reason_bits)358 .ok_or_else(|| invalid_decision("unsupported decision reason bits"))?,359 applied: wire[37] == 1,360 };361 decision.validate()?;362 Ok(decision)363 }364}365366/// Read one fixed-width report. Clean EOF before the next record returns `None`.367pub async fn read_media_feedback_report_v1<R>(368 reader: &mut R,369) -> Result<Option<MediaFeedbackReportV1>>370where371 R: AsyncRead + Unpin,372{373 let mut wire = [0_u8; MEDIA_FEEDBACK_REPORT_V1_LEN];374 if reader.read(&mut wire[..1]).await? == 0 {375 return Ok(None);376 }377 reader.read_exact(&mut wire[1..]).await?;378 MediaFeedbackReportV1::decode(&wire).map(Some)379}380381/// Validate and write one fixed-width report.382pub async fn write_media_feedback_report_v1<W>(383 writer: &mut W,384 report: &MediaFeedbackReportV1,385) -> Result<()>386where387 W: AsyncWrite + Unpin,388{389 writer.write_all(&report.encode()?).await?;390 writer.flush().await?;391 Ok(())392}393394/// Read one fixed-width decision. Clean EOF before the next record returns `None`.395pub async fn read_adaptive_bitrate_decision_v1<R>(396 reader: &mut R,397) -> Result<Option<AdaptiveBitrateDecisionV1>>398where399 R: AsyncRead + Unpin,400{401 let mut wire = [0_u8; ADAPTIVE_BITRATE_DECISION_V1_LEN];402 if reader.read(&mut wire[..1]).await? == 0 {403 return Ok(None);404 }405 reader.read_exact(&mut wire[1..]).await?;406 AdaptiveBitrateDecisionV1::decode(&wire).map(Some)407}408409/// Validate and write one fixed-width decision.410pub async fn write_adaptive_bitrate_decision_v1<W>(411 writer: &mut W,412 decision: &AdaptiveBitrateDecisionV1,413) -> Result<()>414where415 W: AsyncWrite + Unpin,416{417 writer.write_all(&decision.encode()?).await?;418 writer.flush().await?;419 Ok(())420}421422fn validate_queue(depth: u8, capacity: u8) -> Result<()> {423 if !(1..=MAX_FEEDBACK_QUEUE_CAPACITY).contains(&capacity) {424 return Err(invalid_report("queue capacity must be 1..=16"));425 }426 if depth > capacity {427 return Err(invalid_report("queue depth must not exceed capacity"));428 }429 Ok(())430}431432fn decode_optional_latency<const N: usize>(wire: &[u8; N], offset: usize) -> Option<u16> {433 let value = u16::from_be_bytes(434 wire[offset..offset + 2]435 .try_into()436 .expect("fixed slice length"),437 );438 (value != OPTIONAL_LATENCY_NONE).then_some(value)439}440441fn validate_fixed_prefix<const N: usize>(442 wire: &[u8; N],443 expected_len: u16,444 message_type: &'static str,445) -> Result<()> {446 let version = u16::from_be_bytes(wire[4..6].try_into().expect("fixed slice length"));447 if version != PROTOCOL_VERSION {448 return Err(ProtocolError::UnsupportedVersion {449 expected: PROTOCOL_VERSION,450 actual: version,451 });452 }453 let actual_len = u16::from_be_bytes(wire[6..8].try_into().expect("fixed slice length"));454 if actual_len != expected_len {455 return Err(ProtocolError::InvalidMessage {456 message_type,457 reason: "fixed record length does not match the protocol",458 });459 }460 Ok(())461}462463fn invalid_report(reason: &'static str) -> ProtocolError {464 ProtocolError::InvalidMessage {465 message_type: "media feedback report",466 reason,467 }468}469470fn invalid_decision(reason: &'static str) -> ProtocolError {471 ProtocolError::InvalidMessage {472 message_type: "adaptive bitrate decision",473 reason,474 }475}476477478mod tests {479 use std::io::ErrorKind;480481 use tokio::io::{AsyncWriteExt, duplex};482483 use super::*;484485 fn report() -> MediaFeedbackReportV1 {486 MediaFeedbackReportV1 {487 report_id: 0x0102_0304_0506_0708,488 interval_ms: 1_000,489 flags: MediaFeedbackFlags::RESYNC_ACTIVE,490 last_sequence: Some(0x1112_1314_1516_1718),491 transport_dropped_delta: 0x0102_0304,492 frontend_dropped_delta: 0x1112_1314,493 decoder_dropped_delta: 0x2122_2324,494 presenter_dropped_delta: 0x3132_3334,495 frontend_queue_depth: 1,496 frontend_queue_capacity: 4,497 decode_queue_depth: 2,498 decode_queue_capacity: 2,499 presenter_queue_depth: 1,500 presenter_queue_capacity: 2,501 transport_delivery_p95_ms: Some(17),502 decode_p95_ms: Some(33),503 presentation_p95_ms: None,504 }505 }506507 fn decision() -> AdaptiveBitrateDecisionV1 {508 AdaptiveBitrateDecisionV1 {509 decision_id: 0x0102_0304_0506_0708,510 report_id: 0x1112_1314_1516_1718,511 target_kbps: 12_000,512 floor_kbps: 4_000,513 ceiling_kbps: 20_000,514 state: AdaptiveBitrateStateV1::Decrease,515 reasons: AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION516 .union(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG),517 applied: true,518 }519 }520521 522 fn feedback_report_golden_vector() {523 assert_eq!(524 report().encode().unwrap(),525 [526 0x53, 0x47, 0x52, 0x31, // SGR1527 0x00, 0x01, 0x00, 0x40, // version, fixed length528 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, // report ID529 0x03, 0xe8, 0x01, 0x00, // interval, flags, reserved530 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, // last sequence531 0x01, 0x02, 0x03, 0x04, // transport drops532 0x11, 0x12, 0x13, 0x14, // frontend drops533 0x21, 0x22, 0x23, 0x24, // decoder drops534 0x31, 0x32, 0x33, 0x34, // presenter drops535 0x01, 0x04, 0x02, 0x02, 0x01, 0x02, // queue pairs536 0x00, 0x11, 0x00, 0x21, 0xff, 0xff, // p95 latencies537 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // reserved538 ]539 );540 }541542 543 fn adaptive_bitrate_decision_golden_vector() {544 assert_eq!(545 decision().encode().unwrap(),546 [547 0x53, 0x47, 0x44, 0x31, // SGD1548 0x00, 0x01, 0x00, 0x28, // version, fixed length549 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, // decision ID550 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, // report ID551 0x00, 0x00, 0x2e, 0xe0, // target: 12000552 0x00, 0x00, 0x0f, 0xa0, // floor: 4000553 0x00, 0x00, 0x4e, 0x20, // ceiling: 20000554 0x01, 0x01, 0x00, 0x12, // decrease, applied, reasons555 ]556 );557 }558559 560 fn records_round_trip() {561 let report = report();562 assert_eq!(563 MediaFeedbackReportV1::decode(&report.encode().unwrap()).unwrap(),564 report565 );566 let decision = decision();567 assert_eq!(568 AdaptiveBitrateDecisionV1::decode(&decision.encode().unwrap()).unwrap(),569 decision570 );571 }572573 574 fn feedback_report_rejects_out_of_bounds_fields() {575 let mut value = report();576 value.report_id = 0;577 assert!(value.validate().is_err());578 value = report();579 value.interval_ms = 249;580 assert!(value.validate().is_err());581 value = report();582 value.interval_ms = 5_001;583 assert!(value.validate().is_err());584 value = report();585 value.last_sequence = Some(u64::MAX);586 assert!(value.validate().is_err());587 value = report();588 value.frontend_queue_capacity = 0;589 assert!(value.validate().is_err());590 value = report();591 value.decode_queue_capacity = 17;592 assert!(value.validate().is_err());593 value = report();594 value.presenter_queue_depth = 3;595 assert!(value.validate().is_err());596 value = report();597 value.decode_p95_ms = Some(60_001);598 assert!(value.validate().is_err());599 }600601 602 fn feedback_report_rejects_malformed_wire_fields() {603 let valid = report().encode().unwrap();604 for (offset, value) in [605 (0, 0), // magic606 (5, 2), // version607 (7, 63), // length608 (18, 2), // flags609 (19, 1), // reserved610 (63, 1), // trailing reserved611 ] {612 let mut malformed = valid;613 malformed[offset] = value;614 assert!(615 MediaFeedbackReportV1::decode(&malformed).is_err(),616 "offset {offset}"617 );618 }619620 let mut invalid_latency = valid;621 invalid_latency[52..54].copy_from_slice(&60_001_u16.to_be_bytes());622 assert!(MediaFeedbackReportV1::decode(&invalid_latency).is_err());623 }624625 626 fn adaptive_bitrate_decision_rejects_out_of_bounds_fields() {627 let mut value = decision();628 value.decision_id = 0;629 assert!(value.validate().is_err());630 value = decision();631 value.report_id = 0;632 assert!(value.validate().is_err());633 value = decision();634 value.floor_kbps = 999;635 assert!(value.validate().is_err());636 value = decision();637 value.ceiling_kbps = 100_001;638 assert!(value.validate().is_err());639 value = decision();640 value.target_kbps = value.floor_kbps - 1;641 assert!(value.validate().is_err());642 value = decision();643 value.target_kbps = value.ceiling_kbps + 1;644 assert!(value.validate().is_err());645 }646647 648 fn adaptive_bitrate_decision_rejects_malformed_wire_fields() {649 let valid = decision().encode().unwrap();650 for (offset, value) in [651 (0, 0), // magic652 (5, 2), // version653 (7, 39), // length654 (36, 3), // state655 (37, 2), // applied656 (38, 1), // unknown high reason bit657 ] {658 let mut malformed = valid;659 malformed[offset] = value;660 assert!(661 AdaptiveBitrateDecisionV1::decode(&malformed).is_err(),662 "offset {offset}"663 );664 }665 }666667 668 async fn async_report_records_round_trip_and_clean_eof() {669 let first = report();670 let mut second = report();671 second.report_id += 1;672 let (mut writer, mut reader) = duplex(256);673 write_media_feedback_report_v1(&mut writer, &first)674 .await675 .unwrap();676 write_media_feedback_report_v1(&mut writer, &second)677 .await678 .unwrap();679 writer.shutdown().await.unwrap();680681 assert_eq!(682 read_media_feedback_report_v1(&mut reader).await.unwrap(),683 Some(first)684 );685 assert_eq!(686 read_media_feedback_report_v1(&mut reader).await.unwrap(),687 Some(second)688 );689 assert_eq!(690 read_media_feedback_report_v1(&mut reader).await.unwrap(),691 None692 );693 }694695 696 async fn async_decision_records_round_trip_and_clean_eof() {697 let expected = decision();698 let (mut writer, mut reader) = duplex(128);699 write_adaptive_bitrate_decision_v1(&mut writer, &expected)700 .await701 .unwrap();702 writer.shutdown().await.unwrap();703704 assert_eq!(705 read_adaptive_bitrate_decision_v1(&mut reader)706 .await707 .unwrap(),708 Some(expected)709 );710 assert_eq!(711 read_adaptive_bitrate_decision_v1(&mut reader)712 .await713 .unwrap(),714 None715 );716 }717718 719 async fn truncated_fixed_records_are_io_errors() {720 let (mut report_writer, mut report_reader) = duplex(64);721 report_writer722 .write_all(&report().encode().unwrap()[..63])723 .await724 .unwrap();725 report_writer.shutdown().await.unwrap();726 let error = read_media_feedback_report_v1(&mut report_reader)727 .await728 .unwrap_err();729 assert!(730 matches!(error, ProtocolError::Io(error) if error.kind() == ErrorKind::UnexpectedEof)731 );732733 let (mut decision_writer, mut decision_reader) = duplex(40);734 decision_writer735 .write_all(&decision().encode().unwrap()[..39])736 .await737 .unwrap();738 decision_writer.shutdown().await.unwrap();739 let error = read_adaptive_bitrate_decision_v1(&mut decision_reader)740 .await741 .unwrap_err();742 assert!(743 matches!(error, ProtocolError::Io(error) if error.kind() == ErrorKind::UnexpectedEof)744 );745 }746}