Bound v2 receiver buffering to four accepted objects, restore accept order before sequence checks, fast-forward only at explicit discontinuity keyframes, and add an enforceable live FPS probe gate.
- commit
- 635bb9474e46219ad0d4c40ca18b7f8990b85dd4
- parent
- 95491259db85e512f703de9ac5012fbf7a538da7
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 227 | +++++++++++++++++++++++++++++- |
| M | src-tauri/ | 179 | +++++++++++++++++++++++- |
@@ -1,4 +1,7 @@
1-use std::time::{Duration, Instant};1+use std::{2+ collections::BTreeMap,3+ time::{Duration, Instant},4+};25 36 use anyhow::{Context, Result, bail, ensure};47 use clap::Parser;@@ -21,6 +24,10 @@ struct Args {
2124 frames: u32,2225 2326 timeout_seconds: u64,27+ /// Fail unless accepted media frames sustain at least this first-to-last28+ /// cadence. Intended for live appliance performance gates.29+ 30+ minimum_fps: Option<f64>,2431 2532 expect_size: (u16, u16),2633 /// Exercise the reliable ordered v1 media stream instead of independent@@ -66,9 +73,83 @@ enum MediaObjectOutcome {
6673 Malformed { index: u64, error: ProtocolError },6774 }6875 76+impl MediaObjectOutcome {77+ fn index(&self) -> u64 {78+ match self {79+ Self::Frame { index, .. } | Self::Dropped { index } | Self::Malformed { index, .. } => {80+ *index81+ }82+ }83+ }84+85+ fn is_fast_forward_barrier(&self) -> bool {86+ let Self::Frame { frame, .. } = self else {87+ return false;88+ };89+ frame.header.flags.contains(FrameFlags::KEYFRAME)90+ && frame.header.flags.contains(FrameFlags::CODEC_CONFIG)91+ && frame.header.flags.contains(FrameFlags::DISCONTINUITY)92+ }93+}94+95+96+struct MediaObjectReorder {97+ next_index: u64,98+ completed: BTreeMap<u64, MediaObjectOutcome>,99+}100+101+impl MediaObjectReorder {102+ fn new(first_index: u64) -> Self {103+ Self {104+ next_index: first_index,105+ completed: BTreeMap::new(),106+ }107+ }108+109+ fn pending_len(&self) -> usize {110+ self.completed.len()111+ }112+113+ fn push(&mut self, outcome: MediaObjectOutcome) -> Result<Option<MediaObjectOutcome>> {114+ if matches!(outcome, MediaObjectOutcome::Malformed { .. }) {115+ // Malformed objects remain terminal as soon as their read completes.116+ return Ok(Some(outcome));117+ }118+ let index = outcome.index();119+ if index < self.next_index {120+ return Ok(Some(outcome));121+ }122+ if outcome.is_fast_forward_barrier() {123+ self.completed124+ .retain(|completed_index, _| *completed_index > index);125+ self.next_index = index126+ .checked_add(1)127+ .context("media object reorder overflowed")?;128+ return Ok(Some(outcome));129+ }130+ ensure!(131+ self.completed.insert(index, outcome).is_none(),132+ "media object {index} completed more than once"133+ );134+ self.take_next()135+ }136+137+ fn take_next(&mut self) -> Result<Option<MediaObjectOutcome>> {138+ let Some(outcome) = self.completed.remove(&self.next_index) else {139+ return Ok(None);140+ };141+ self.next_index = self142+ .next_index143+ .checked_add(1)144+ .context("media object reorder overflowed")?;145+ Ok(Some(outcome))146+ }147+}148+69149 struct MediaObjectReceiver {70150 connection: iroh::endpoint::Connection,71151 reads: tokio::task::JoinSet<MediaObjectOutcome>,152+ reorder: MediaObjectReorder,72153 next_index: u64,73154 accepting: bool,74155 read_timeout: Duration,@@ -79,6 +160,7 @@ impl MediaObjectReceiver {
79160 Self {80161 connection,81162 reads: tokio::task::JoinSet::new(),163+ reorder: MediaObjectReorder::new(0),82164 next_index: 0,83165 accepting: true,84166 read_timeout,@@ -87,13 +169,22 @@ impl MediaObjectReceiver {
87169 88170 async fn next(&mut self) -> Result<Option<MediaObjectOutcome>> {89171 loop {172+ if let Some(completed) = self.reorder.take_next()? {173+ return Ok(Some(completed));174+ }90175 if !self.accepting && self.reads.is_empty() {176+ ensure!(177+ self.reorder.pending_len() == 0,178+ "media connection closed with an incomplete object order"179+ );91180 return Ok(None);92181 }93182 94183 tokio::select! {95184 accepted = self.connection.accept_uni(),96- if self.accepting && self.reads.len() < MEDIA_OBJECT_CAPACITY => {185+ if self.accepting186+ && self.reads.len() + self.reorder.pending_len()187+ < MEDIA_OBJECT_CAPACITY => {97188 match accepted {98189 Ok(mut stream) => {99190 let index = self.next_index;@@ -115,11 +206,15 @@ impl MediaObjectReceiver {
115206 }116207 }117208 completed = self.reads.join_next(), if !self.reads.is_empty() => {118- return match completed.expect("guarded by non-empty media object task set") {119- Ok(outcome) => Ok(Some(outcome)),209+ match completed.expect("guarded by non-empty media object task set") {210+ Ok(outcome) => {211+ if let Some(completed) = self.reorder.push(outcome)? {212+ return Ok(Some(completed));213+ }214+ }120215 Err(error) if error.is_cancelled() => continue,121- Err(error) => Err(error).context("media object read task failed"),122- };216+ Err(error) => return Err(error).context("media object read task failed"),217+ }123218 }124219 }125220 }@@ -360,6 +455,8 @@ async fn main() -> Result<()> {
360455 let mut media_objects_late = 0_u64;361456 let mut last_sequence: Option<u64> = None;362457 let mut dimensions = None;458+ let mut first_accepted_at = None;459+ let mut last_accepted_at = None;363460 let mut object_receiver = (media_transport == MediaTransport::IndependentV2).then(|| {364461 MediaObjectReceiver::new(365462 media_connection.clone(),@@ -419,6 +516,10 @@ async fn main() -> Result<()> {
419516 },420517 };421518 519+ let accepted_at = Instant::now();520+ first_accepted_at.get_or_insert(accepted_at);521+ last_accepted_at = Some(accepted_at);522+422523 if received == 0 {423524 ensure!(424525 frame.header.flags.contains(FrameFlags::KEYFRAME)@@ -456,6 +557,21 @@ async fn main() -> Result<()> {
456557 );457558 ensure!(keyframes > 0, "probe received no H.264 keyframe");458559 ensure!(gaps == 0, "probe observed {gaps} media sequence gaps");560+ let accepted_span = first_accepted_at561+ .zip(last_accepted_at)562+ .map_or(Duration::ZERO, |(first, last)| {563+ last.saturating_duration_since(first)564+ });565+ let accepted_fps = accepted_frame_rate(received, accepted_span);566+ if let Some(minimum_fps) = args.minimum_fps {567+ let accepted_fps = accepted_fps.context(568+ "--minimum-fps requires at least two accepted frames with measurable elapsed time",569+ )?;570+ ensure!(571+ accepted_fps >= minimum_fps,572+ "probe sustained only {accepted_fps:.3} accepted fps; required at least {minimum_fps:.3} fps"573+ );574+ }459575 let (path_mode, path_rtt_ms) = selected_path_diagnostics(&media_connection);460576 input_send.finish().context("finishing input stream")?;461577 if let Some(mut media_send) = media_send {@@ -481,6 +597,10 @@ async fn main() -> Result<()> {
481597 println!("media_objects_dropped={media_objects_dropped}");482598 println!("media_objects_late={media_objects_late}");483599 println!("transport={}", media_transport.label());600+ match accepted_fps {601+ Some(fps) => println!("accepted_fps={fps:.3}"),602+ None => println!("accepted_fps=unknown"),603+ }484604 println!("input_ack_micros={input_ack_micros}");485605 println!(486606 "pointer_smoke={}",@@ -534,6 +654,13 @@ fn sequence_gap(previous: u64, current: u64) -> Result<u64> {
534654 Ok(current - expected)535655 }536656 657+fn accepted_frame_rate(frame_count: u32, elapsed: Duration) -> Option<f64> {658+ if frame_count < 2 || elapsed.is_zero() {659+ return None;660+ }661+ Some(f64::from(frame_count - 1) / elapsed.as_secs_f64())662+}663+537664 fn selected_path_diagnostics(538665 connection: &iroh::endpoint::Connection,539666 ) -> (&'static str, Option<f64>) {@@ -563,6 +690,16 @@ fn parse_size(value: &str) -> std::result::Result<(u16, u16), String> {
563690 Ok((width, height))564691 }565692 693+fn parse_minimum_fps(value: &str) -> std::result::Result<f64, String> {694+ let minimum_fps = value695+ .parse::<f64>()696+ .map_err(|_| "minimum fps must be a number".to_owned())?;697+ if !minimum_fps.is_finite() || minimum_fps <= 0.0 || minimum_fps > 240.0 {698+ return Err("minimum fps must be finite and within (0, 240]".to_owned());699+ }700+ Ok(minimum_fps)701+}702+566703 async fn negotiate(567704 send: &mut iroh::endpoint::SendStream,568705 recv: &mut iroh::endpoint::RecvStream,@@ -683,6 +820,84 @@ mod tests {
683820 .unwrap()684821 }685822 823+ fn media_outcome(index: u64, sequence: u64, flags: FrameFlags) -> MediaObjectOutcome {824+ MediaObjectOutcome::Frame {825+ index,826+ frame: media_frame(sequence, flags),827+ }828+ }829+830+ 831+ fn media_object_reorder_restores_accept_order_before_sequence_checks() {832+ let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);833+ let mut reorder = MediaObjectReorder::new(0);834+835+ assert!(836+ reorder837+ .push(media_outcome(1, 11, FrameFlags::NONE))838+ .unwrap()839+ .is_none()840+ );841+ assert_eq!(reorder.pending_len(), 1);842+ assert_eq!(843+ reorder844+ .push(media_outcome(0, 10, keyframe))845+ .unwrap()846+ .unwrap()847+ .index(),848+ 0849+ );850+ assert_eq!(reorder.take_next().unwrap().unwrap().index(), 1);851+ assert_eq!(reorder.pending_len(), 0);852+ }853+854+ 855+ fn discontinuity_keyframe_is_an_explicit_latest_frame_barrier() {856+ let barrier = FrameFlags::KEYFRAME857+ .union(FrameFlags::CODEC_CONFIG)858+ .union(FrameFlags::DISCONTINUITY);859+ let mut reorder = MediaObjectReorder::new(0);860+861+ assert!(862+ reorder863+ .push(media_outcome(1, 11, FrameFlags::NONE))864+ .unwrap()865+ .is_none()866+ );867+ assert_eq!(868+ reorder869+ .push(media_outcome(2, 20, barrier))870+ .unwrap()871+ .unwrap()872+ .index(),873+ 2874+ );875+ assert_eq!(reorder.pending_len(), 0);876+ assert_eq!(877+ reorder878+ .push(media_outcome(0, 10, FrameFlags::NONE))879+ .unwrap()880+ .unwrap()881+ .index(),882+ 0883+ );884+ }885+886+ 887+ fn minimum_fps_parser_and_first_to_last_evaluation_are_bounded() {888+ assert_eq!(parse_minimum_fps("55").unwrap(), 55.0);889+ assert_eq!(parse_minimum_fps("240").unwrap(), 240.0);890+ for invalid in ["0", "-1", "240.1", "NaN", "inf", "not-a-number"] {891+ assert!(parse_minimum_fps(invalid).is_err(), "{invalid}");892+ }893+894+ assert_eq!(accepted_frame_rate(56, Duration::from_secs(1)), Some(55.0));895+ assert_eq!(accepted_frame_rate(1, Duration::from_secs(1)), None);896+ assert_eq!(accepted_frame_rate(2, Duration::ZERO), None);897+ assert!(accepted_frame_rate(56, Duration::from_secs(1)).unwrap() >= 55.0);898+ assert!(accepted_frame_rate(55, Duration::from_secs(1)).unwrap() < 55.0);899+ }900+686901 687902 fn media_objects_wait_for_a_decodable_keyframe_then_deliver_contiguous_frames() {688903 let mut sequence = MediaObjectSequence::new();@@ -679,9 +679,87 @@ enum MediaObjectReadOutcome {
679679 Malformed(String),680680 }681681 682+impl MediaObjectReadOutcome {683+ fn object_index(&self) -> Option<u64> {684+ match self {685+ Self::Frame { object_index, .. } | Self::Dropped { object_index } => {686+ Some(*object_index)687+ }688+ Self::Malformed(_) => None,689+ }690+ }691+692+ fn is_fast_forward_barrier(&self) -> bool {693+ let Self::Frame { frame, .. } = self else {694+ return false;695+ };696+ frame.header.flags.contains(FrameFlags::KEYFRAME)697+ && frame.header.flags.contains(FrameFlags::CODEC_CONFIG)698+ && frame.header.flags.contains(FrameFlags::DISCONTINUITY)699+ }700+}701+702+703+struct MediaObjectReorder {704+ next_object_index: u64,705+ completed: BTreeMap<u64, MediaObjectReadOutcome>,706+}707+708+impl MediaObjectReorder {709+ fn new(first_object_index: u64) -> Self {710+ Self {711+ next_object_index: first_object_index,712+ completed: BTreeMap::new(),713+ }714+ }715+716+ fn pending_len(&self) -> usize {717+ self.completed.len()718+ }719+720+ fn push(721+ &mut self,722+ outcome: MediaObjectReadOutcome,723+ ) -> Result<Option<MediaObjectReadOutcome>, String> {724+ let Some(object_index) = outcome.object_index() else {725+ // Malformed objects remain terminal as soon as their read completes.726+ return Ok(Some(outcome));727+ };728+ if object_index < self.next_object_index {729+ return Ok(Some(outcome));730+ }731+ if outcome.is_fast_forward_barrier() {732+ self.completed733+ .retain(|completed_index, _| *completed_index > object_index);734+ self.next_object_index = object_index735+ .checked_add(1)736+ .ok_or_else(|| "Media object reorder index overflowed".to_string())?;737+ return Ok(Some(outcome));738+ }739+ if self.completed.insert(object_index, outcome).is_some() {740+ return Err(format!(741+ "Media object {object_index} completed more than once"742+ ));743+ }744+ self.take_next()745+ }746+747+ fn take_next(&mut self) -> Result<Option<MediaObjectReadOutcome>, String> {748+ let Some(outcome) = self.completed.remove(&self.next_object_index) else {749+ return Ok(None);750+ };751+ self.next_object_index = self752+ .next_object_index753+ .checked_add(1)754+ .ok_or_else(|| "Media object reorder index overflowed".to_string())?;755+ Ok(Some(outcome))756+ }757+}758+682759 struct MediaObjectReceiver {683760 connection: iroh::endpoint::Connection,684761 reads: tokio::task::JoinSet<MediaObjectReadOutcome>,762+ reorder: MediaObjectReorder,685763 next_object_index: u64,686764 connection_closed: bool,687765 }@@ -691,6 +769,7 @@ impl MediaObjectReceiver {
691769 Self {692770 connection,693771 reads: tokio::task::JoinSet::new(),772+ reorder: MediaObjectReorder::new(1),694773 next_object_index: 0,695774 connection_closed: false,696775 }@@ -698,7 +777,13 @@ impl MediaObjectReceiver {
698777 699778 async fn next(&mut self) -> Result<Option<MediaObjectReadOutcome>, String> {700779 loop {780+ if let Some(completed) = self.reorder.take_next()? {781+ return Ok(Some(completed));782+ }701783 if self.connection_closed && self.reads.is_empty() {784+ if self.reorder.pending_len() != 0 {785+ return Err("Media connection closed with an incomplete object order".into());786+ }702787 return Ok(None);703788 }704789 @@ -708,10 +793,13 @@ impl MediaObjectReceiver {
708793 let completed = completed709794 .ok_or_else(|| "Media object reader ended unexpectedly".to_string())?710795 .map_err(|error| format!("Media object reader task failed: {error}"))?;711- return Ok(Some(completed));796+ if let Some(completed) = self.reorder.push(completed)? {797+ return Ok(Some(completed));798+ }712799 }713800 accepted = self.connection.accept_uni(), if !self.connection_closed714- && self.reads.len() < CLIENT_MEDIA_OBJECT_CAPACITY => {801+ && self.reads.len() + self.reorder.pending_len()802+ < CLIENT_MEDIA_OBJECT_CAPACITY => {715803 let mut stream = match accepted {716804 Ok(stream) => stream,717805 Err(_) => {@@ -2273,6 +2361,93 @@ mod tests {
22732361 MediaFrame::new(header, payload).unwrap()22742362 }22752363 2364+ fn media_object_outcome(2365+ object_index: u64,2366+ sequence: u64,2367+ flags: FrameFlags,2368+ ) -> MediaObjectReadOutcome {2369+ MediaObjectReadOutcome::Frame {2370+ object_index,2371+ frame: media_object_frame(sequence, flags),2372+ }2373+ }2374+2375+ fn completed_object_index(outcome: &MediaObjectReadOutcome) -> Option<u64> {2376+ outcome.object_index()2377+ }2378+2379+ 2380+ fn media_object_reorder_restores_accept_order_without_false_resync() {2381+ let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);2382+ let mut reorder = MediaObjectReorder::new(1);2383+2384+ assert!(2385+ reorder2386+ .push(media_object_outcome(2, 11, FrameFlags::NONE))2387+ .unwrap()2388+ .is_none()2389+ );2390+ assert_eq!(reorder.pending_len(), 1);2391+ let first = reorder2392+ .push(media_object_outcome(1, 10, keyframe))2393+ .unwrap()2394+ .unwrap();2395+ assert_eq!(completed_object_index(&first), Some(1));2396+ assert_eq!(2397+ completed_object_index(&reorder.take_next().unwrap().unwrap()),2398+ Some(2)2399+ );2400+ assert_eq!(reorder.pending_len(), 0);2401+ }2402+2403+ 2404+ fn explicit_discontinuity_keyframe_fast_forwards_bounded_reorder() {2405+ let barrier = FrameFlags::KEYFRAME2406+ .union(FrameFlags::CODEC_CONFIG)2407+ .union(FrameFlags::DISCONTINUITY);2408+ let mut reorder = MediaObjectReorder::new(1);2409+2410+ assert!(2411+ reorder2412+ .push(media_object_outcome(2, 11, FrameFlags::NONE))2413+ .unwrap()2414+ .is_none()2415+ );2416+ let recovered = reorder2417+ .push(media_object_outcome(3, 20, barrier))2418+ .unwrap()2419+ .unwrap();2420+ assert_eq!(completed_object_index(&recovered), Some(3));2421+ assert_eq!(reorder.pending_len(), 0);2422+ assert_eq!(2423+ completed_object_index(2424+ &reorder2425+ .push(media_object_outcome(1, 10, FrameFlags::NONE))2426+ .unwrap()2427+ .unwrap()2428+ ),2429+ Some(1)2430+ );2431+ }2432+2433+ 2434+ fn malformed_media_object_bypasses_reorder_and_remains_terminal() {2435+ let mut reorder = MediaObjectReorder::new(1);2436+ assert!(2437+ reorder2438+ .push(media_object_outcome(2, 11, FrameFlags::NONE))2439+ .unwrap()2440+ .is_none()2441+ );2442+ assert!(matches!(2443+ reorder2444+ .push(MediaObjectReadOutcome::Malformed("bad object".into()))2445+ .unwrap(),2446+ Some(MediaObjectReadOutcome::Malformed(_))2447+ ));2448+ assert_eq!(reorder.pending_len(), 1);2449+ }2450+22762451 22772452 fn media_objects_begin_on_a_configured_keyframe_then_deliver_contiguously() {22782453 let keyframe_flags = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);