- commit
- b3ed6c07989654b6f5a5885b5f6c09694a68471c
- parent
- c1bdfb51e7fc89e111c6f24d138c0002e076ad03
- author
- Dami <dami.etoile@gmail.com>
| M | src-tauri/ | 425 | +----------------------------- |
| M | src-tauri/ | 1 | + |
| A | src-tauri/ | 432 | ++++++++++++++++++++++++++++++ |
@@ -18,6 +18,12 @@ use crate::media::frame_channel::{
1818 use crate::media::metrics::{ClientMediaMetrics, lock_client_media_metrics};1919 use crate::media::moq_catalog::subscribe_goq_video_track;2020 use crate::media::network_diagnostics::{NetworkLeg, NetworkSessionDiagnostics};21+22+use crate::media::object_receiver::media_object_frame;23+use crate::media::object_receiver::{24+ CLIENT_MEDIA_OBJECT_CAPACITY, CLIENT_MEDIA_OBJECT_READ_TIMEOUT, MediaObjectReadOutcome,25+ MediaObjectReceiver, MediaObjectSequence, MediaObjectSequenceDecision,26+};2127 use crate::media::transport::{2228 MediaTransport, NegotiatedV1Stream, connect_error_is_unsupported_alpn, negotiate_v1,2329 open_negotiated_media_stream,@@ -40,7 +46,7 @@ use sigil_protocol::{
4046 MediaFeedbackReportV1, MediaFrame, MediaObjectV3, PointerPosition, PointerSurfaceDimensions,4147 ProtocolError, RELATIVE_POINTER_DELTA_MAX, RELATIVE_POINTER_DELTA_MIN,4248 decode_media_frame_object, media_moq_broadcast_name, read_adaptive_bitrate_decision_v1,43- read_input_ack, read_media_frame, read_media_object, read_media_object_v3, write_input_event,49+ read_input_ack, read_media_frame, read_media_object_v3, write_input_event,4450 write_media_control_request_v3, write_media_feedback_report_v1,4551 };4652 use std::collections::BTreeMap;@@ -56,8 +62,6 @@ use tauri::{
5662 // ─── Client commands ──────────────────────────────────────────────────────────5763 5864 const LEGACY_MEDIA_IDLE_TIMEOUT: Duration = Duration::from_secs(10);59-const CLIENT_MEDIA_OBJECT_CAPACITY: usize = 4;60-const CLIENT_MEDIA_OBJECT_READ_TIMEOUT: Duration = Duration::from_secs(1);6165 const CLIENT_MOQ_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);6266 const CLIENT_MOQ_SUBSCRIBE_TIMEOUT: Duration = Duration::from_secs(10);6367 // Match the protocol's maximum single-object delivery horizon. A publisher may@@ -608,183 +612,6 @@ fn moq_group_error_reason(error: &moq_net::Error) -> KeyframeRequestReasonV3 {
608612 }609613 }610614 611-612-enum MediaObjectReadOutcome {613- Frame {614- object_index: u64,615- frame: MediaFrame,616- },617- Dropped {618- object_index: u64,619- },620- Malformed(String),621-}622-623-impl MediaObjectReadOutcome {624- fn object_index(&self) -> Option<u64> {625- match self {626- Self::Frame { object_index, .. } | Self::Dropped { object_index } => {627- Some(*object_index)628- }629- Self::Malformed(_) => None,630- }631- }632-633- fn is_fast_forward_barrier(&self) -> bool {634- let Self::Frame { frame, .. } = self else {635- return false;636- };637- frame.header.flags.contains(FrameFlags::KEYFRAME)638- && frame.header.flags.contains(FrameFlags::CODEC_CONFIG)639- && frame.header.flags.contains(FrameFlags::DISCONTINUITY)640- }641-}642-643-644-struct MediaObjectReorder {645- next_object_index: u64,646- completed: BTreeMap<u64, MediaObjectReadOutcome>,647-}648-649-impl MediaObjectReorder {650- fn new(first_object_index: u64) -> Self {651- Self {652- next_object_index: first_object_index,653- completed: BTreeMap::new(),654- }655- }656-657- fn pending_len(&self) -> usize {658- self.completed.len()659- }660-661- fn push(662- &mut self,663- outcome: MediaObjectReadOutcome,664- ) -> Result<Option<MediaObjectReadOutcome>, String> {665- let Some(object_index) = outcome.object_index() else {666- // Malformed objects remain terminal as soon as their read completes.667- return Ok(Some(outcome));668- };669- if object_index < self.next_object_index {670- return Ok(Some(outcome));671- }672- if outcome.is_fast_forward_barrier() {673- self.completed674- .retain(|completed_index, _| *completed_index > object_index);675- self.next_object_index = object_index676- .checked_add(1)677- .ok_or_else(|| "Media object reorder index overflowed".to_string())?;678- return Ok(Some(outcome));679- }680- if self.completed.insert(object_index, outcome).is_some() {681- return Err(format!(682- "Media object {object_index} completed more than once"683- ));684- }685- self.take_next()686- }687-688- fn take_next(&mut self) -> Result<Option<MediaObjectReadOutcome>, String> {689- let Some(outcome) = self.completed.remove(&self.next_object_index) else {690- return Ok(None);691- };692- self.next_object_index = self693- .next_object_index694- .checked_add(1)695- .ok_or_else(|| "Media object reorder index overflowed".to_string())?;696- Ok(Some(outcome))697- }698-}699-700-struct MediaObjectReceiver {701- connection: iroh::endpoint::Connection,702- reads: tokio::task::JoinSet<MediaObjectReadOutcome>,703- reorder: MediaObjectReorder,704- next_object_index: u64,705- connection_closed: bool,706-}707-708-impl MediaObjectReceiver {709- fn new(connection: iroh::endpoint::Connection) -> Self {710- Self {711- connection,712- reads: tokio::task::JoinSet::new(),713- reorder: MediaObjectReorder::new(1),714- next_object_index: 0,715- connection_closed: false,716- }717- }718-719- async fn next(&mut self) -> Result<Option<MediaObjectReadOutcome>, String> {720- loop {721- if let Some(completed) = self.reorder.take_next()? {722- return Ok(Some(completed));723- }724- if self.connection_closed && self.reads.is_empty() {725- if self.reorder.pending_len() != 0 {726- return Err("Media connection closed with an incomplete object order".into());727- }728- return Ok(None);729- }730-731- tokio::select! {732- biased;733- completed = self.reads.join_next(), if !self.reads.is_empty() => {734- let completed = completed735- .ok_or_else(|| "Media object reader ended unexpectedly".to_string())?736- .map_err(|error| format!("Media object reader task failed: {error}"))?;737- if let Some(completed) = self.reorder.push(completed)? {738- return Ok(Some(completed));739- }740- }741- accepted = self.connection.accept_uni(), if !self.connection_closed742- && self.reads.len() + self.reorder.pending_len()743- < CLIENT_MEDIA_OBJECT_CAPACITY => {744- let mut stream = match accepted {745- Ok(stream) => stream,746- Err(_) => {747- self.connection_closed = true;748- continue;749- }750- };751- self.next_object_index = self.next_object_index.checked_add(1)752- .ok_or_else(|| "Media object index overflowed".to_string())?;753- let object_index = self.next_object_index;754- self.reads.spawn(async move {755- match tokio::time::timeout(756- CLIENT_MEDIA_OBJECT_READ_TIMEOUT,757- read_media_object(&mut stream),758- )759- .await760- {761- Err(_) => MediaObjectReadOutcome::Dropped { object_index },762- Ok(Err(ProtocolError::Io(_))) => {763- MediaObjectReadOutcome::Dropped { object_index }764- }765- Ok(Err(error)) => {766- MediaObjectReadOutcome::Malformed(format!(767- "Invalid media object: {error}"768- ))769- }770- Ok(Ok(frame)) => MediaObjectReadOutcome::Frame {771- object_index,772- frame,773- },774- }775- });776- }777- }778- }779- }780-}781-782-impl Drop for MediaObjectReceiver {783- fn drop(&mut self) {784- self.reads.abort_all();785- }786-}787-788615 789616 enum MediaObjectReadOutcomeV3 {790617 Object {@@ -968,65 +795,6 @@ impl Drop for MediaObjectReceiverV3 {
968795 }969796 }970797 971-972-enum MediaObjectSequenceDecision {973- Deliver { discontinuity: bool },974- DropLate,975- DropUntilKeyframe,976-}977-978-979-struct MediaObjectSequence {980- last_sequence: Option<u64>,981- last_object_index: u64,982- waiting_for_keyframe: bool,983-}984-985-impl MediaObjectSequence {986- fn new() -> Self {987- Self {988- waiting_for_keyframe: true,989- ..Self::default()990- }991- }992-993- fn note_dropped_object(&mut self, object_index: u64) -> bool {994- if object_index <= self.last_object_index {995- return false;996- }997- self.waiting_for_keyframe = true;998- true999- }1000-1001- fn classify(&mut self, object_index: u64, frame: &MediaFrame) -> MediaObjectSequenceDecision {1002- if object_index <= self.last_object_index1003- || self1004- .last_sequence1005- .is_some_and(|sequence| frame.header.sequence <= sequence)1006- {1007- return MediaObjectSequenceDecision::DropLate;1008- }1009-1010- let keyframe = frame.header.flags.contains(FrameFlags::KEYFRAME)1011- && frame.header.flags.contains(FrameFlags::CODEC_CONFIG);1012- let sequence_contiguous = self1013- .last_sequence1014- .is_none_or(|sequence| sequence.checked_add(1) == Some(frame.header.sequence));1015- if !keyframe && (self.waiting_for_keyframe || !sequence_contiguous) {1016- self.waiting_for_keyframe = true;1017- return MediaObjectSequenceDecision::DropUntilKeyframe;1018- }1019-1020- let discontinuity = frame.header.flags.contains(FrameFlags::DISCONTINUITY)1021- || self.waiting_for_keyframe1022- || !sequence_contiguous;1023- self.last_sequence = Some(frame.header.sequence);1024- self.last_object_index = object_index;1025- self.waiting_for_keyframe = false;1026- MediaObjectSequenceDecision::Deliver { discontinuity }1027- }1028-}1029-1030798 1031799 enum MediaObjectSequenceDecisionV3 {1032800 Deliver { discontinuity: bool },@@ -2827,32 +2595,6 @@ fn sequence_gap(previous: u64, current: u64) -> Result<u64, String> {
28272595 mod tests {28282596 use super::*;28292597 2830- fn media_object_frame(sequence: u64, flags: FrameFlags) -> MediaFrame {2831- let payload = vec![sequence as u8];2832- let header = sigil_protocol::MediaFrameHeader::h264(2833- 1280,2834- 800,2835- payload.len(),2836- sequence,2837- sequence * 1_000,2838- sequence as i64 * 1_000,2839- flags,2840- )2841- .unwrap();2842- MediaFrame::new(header, payload).unwrap()2843- }2844-2845- fn media_object_outcome(2846- object_index: u64,2847- sequence: u64,2848- flags: FrameFlags,2849- ) -> MediaObjectReadOutcome {2850- MediaObjectReadOutcome::Frame {2851- object_index,2852- frame: media_object_frame(sequence, flags),2853- }2854- }2855-28562598 fn media_object_v3(28572599 group_id: u64,28582600 object_id: u32,@@ -3073,159 +2815,6 @@ mod tests {
30732815 );30742816 }30752817 3076- fn completed_object_index(outcome: &MediaObjectReadOutcome) -> Option<u64> {3077- outcome.object_index()3078- }3079-3080- 3081- fn media_object_reorder_restores_accept_order_without_false_resync() {3082- let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);3083- let mut reorder = MediaObjectReorder::new(1);3084-3085- assert!(3086- reorder3087- .push(media_object_outcome(2, 11, FrameFlags::NONE))3088- .unwrap()3089- .is_none()3090- );3091- assert_eq!(reorder.pending_len(), 1);3092- let first = reorder3093- .push(media_object_outcome(1, 10, keyframe))3094- .unwrap()3095- .unwrap();3096- assert_eq!(completed_object_index(&first), Some(1));3097- assert_eq!(3098- completed_object_index(&reorder.take_next().unwrap().unwrap()),3099- Some(2)3100- );3101- assert_eq!(reorder.pending_len(), 0);3102- }3103-3104- 3105- fn explicit_discontinuity_keyframe_fast_forwards_bounded_reorder() {3106- let barrier = FrameFlags::KEYFRAME3107- .union(FrameFlags::CODEC_CONFIG)3108- .union(FrameFlags::DISCONTINUITY);3109- let mut reorder = MediaObjectReorder::new(1);3110-3111- assert!(3112- reorder3113- .push(media_object_outcome(2, 11, FrameFlags::NONE))3114- .unwrap()3115- .is_none()3116- );3117- let recovered = reorder3118- .push(media_object_outcome(3, 20, barrier))3119- .unwrap()3120- .unwrap();3121- assert_eq!(completed_object_index(&recovered), Some(3));3122- assert_eq!(reorder.pending_len(), 0);3123- assert_eq!(3124- completed_object_index(3125- &reorder3126- .push(media_object_outcome(1, 10, FrameFlags::NONE))3127- .unwrap()3128- .unwrap()3129- ),3130- Some(1)3131- );3132- }3133-3134- 3135- fn malformed_media_object_bypasses_reorder_and_remains_terminal() {3136- let mut reorder = MediaObjectReorder::new(1);3137- assert!(3138- reorder3139- .push(media_object_outcome(2, 11, FrameFlags::NONE))3140- .unwrap()3141- .is_none()3142- );3143- assert!(matches!(3144- reorder3145- .push(MediaObjectReadOutcome::Malformed("bad object".into()))3146- .unwrap(),3147- Some(MediaObjectReadOutcome::Malformed(_))3148- ));3149- assert_eq!(reorder.pending_len(), 1);3150- }3151-3152- 3153- fn media_objects_begin_on_a_configured_keyframe_then_deliver_contiguously() {3154- let keyframe_flags = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);3155- let mut sequence = MediaObjectSequence::new();3156-3157- assert_eq!(3158- sequence.classify(1, &media_object_frame(1, FrameFlags::NONE)),3159- MediaObjectSequenceDecision::DropUntilKeyframe3160- );3161- assert_eq!(3162- sequence.classify(2, &media_object_frame(2, keyframe_flags)),3163- MediaObjectSequenceDecision::Deliver {3164- discontinuity: true3165- }3166- );3167- assert_eq!(3168- sequence.classify(3, &media_object_frame(3, FrameFlags::NONE)),3169- MediaObjectSequenceDecision::Deliver {3170- discontinuity: false3171- }3172- );3173- }3174-3175- 3176- fn media_object_sequence_gaps_drop_deltas_until_a_new_keyframe() {3177- let keyframe_flags = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);3178- let mut sequence = MediaObjectSequence::new();3179-3180- assert!(matches!(3181- sequence.classify(1, &media_object_frame(10, keyframe_flags)),3182- MediaObjectSequenceDecision::Deliver { .. }3183- ));3184- assert_eq!(3185- sequence.classify(3, &media_object_frame(12, FrameFlags::NONE)),3186- MediaObjectSequenceDecision::DropUntilKeyframe3187- );3188- assert_eq!(3189- sequence.classify(4, &media_object_frame(13, FrameFlags::NONE)),3190- MediaObjectSequenceDecision::DropUntilKeyframe3191- );3192- assert_eq!(3193- sequence.classify(5, &media_object_frame(14, keyframe_flags)),3194- MediaObjectSequenceDecision::Deliver {3195- discontinuity: true3196- }3197- );3198- }3199-3200- 3201- fn late_media_object_completion_cannot_rewind_a_recovered_stream() {3202- let keyframe_flags = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);3203- let mut sequence = MediaObjectSequence::new();3204-3205- assert!(matches!(3206- sequence.classify(10, &media_object_frame(10, keyframe_flags)),3207- MediaObjectSequenceDecision::Deliver { .. }3208- ));3209- assert!(sequence.note_dropped_object(12));3210- assert_eq!(3211- sequence.classify(13, &media_object_frame(13, keyframe_flags)),3212- MediaObjectSequenceDecision::Deliver {3213- discontinuity: true3214- }3215- );3216- assert!(!sequence.note_dropped_object(11));3217- assert_eq!(3218- sequence.classify(12, &media_object_frame(12, FrameFlags::NONE)),3219- MediaObjectSequenceDecision::DropLate3220- );3221- assert_eq!(3222- sequence.classify(14, &media_object_frame(14, FrameFlags::NONE)),3223- MediaObjectSequenceDecision::Deliver {3224- discontinuity: false3225- }3226- );3227- }3228-32292818 32302819 fn media_v3_groups_use_wire_object_identity_and_recover_on_new_group_zero() {32312820 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);@@ -3,4 +3,5 @@ pub(crate) mod frame_channel;
33 pub(crate) mod metrics;44 pub(crate) mod moq_catalog;55 pub(crate) mod network_diagnostics;6+pub(crate) mod object_receiver;67 pub(crate) mod transport;@@ -0,0 +1,432 @@
1+use std::collections::BTreeMap;2+use std::time::Duration;3+4+use sigil_protocol::{FrameFlags, MediaFrame, ProtocolError, read_media_object};5+6+pub(crate) const CLIENT_MEDIA_OBJECT_CAPACITY: usize = 4;7+pub(crate) const CLIENT_MEDIA_OBJECT_READ_TIMEOUT: Duration = Duration::from_secs(1);8+9+10+pub(crate) enum MediaObjectReadOutcome {11+ Frame {12+ object_index: u64,13+ frame: MediaFrame,14+ },15+ Dropped {16+ object_index: u64,17+ },18+ Malformed(String),19+}20+21+impl MediaObjectReadOutcome {22+ fn object_index(&self) -> Option<u64> {23+ match self {24+ Self::Frame { object_index, .. } | Self::Dropped { object_index } => {25+ Some(*object_index)26+ }27+ Self::Malformed(_) => None,28+ }29+ }30+31+ fn is_fast_forward_barrier(&self) -> bool {32+ let Self::Frame { frame, .. } = self else {33+ return false;34+ };35+ frame.header.flags.contains(FrameFlags::KEYFRAME)36+ && frame.header.flags.contains(FrameFlags::CODEC_CONFIG)37+ && frame.header.flags.contains(FrameFlags::DISCONTINUITY)38+ }39+}40+41+42+struct MediaObjectReorder {43+ next_object_index: u64,44+ completed: BTreeMap<u64, MediaObjectReadOutcome>,45+}46+47+impl MediaObjectReorder {48+ fn new(first_object_index: u64) -> Self {49+ Self {50+ next_object_index: first_object_index,51+ completed: BTreeMap::new(),52+ }53+ }54+55+ fn pending_len(&self) -> usize {56+ self.completed.len()57+ }58+59+ fn push(60+ &mut self,61+ outcome: MediaObjectReadOutcome,62+ ) -> Result<Option<MediaObjectReadOutcome>, String> {63+ let Some(object_index) = outcome.object_index() else {64+ // Malformed objects remain terminal as soon as their read completes.65+ return Ok(Some(outcome));66+ };67+ if object_index < self.next_object_index {68+ return Ok(Some(outcome));69+ }70+ if outcome.is_fast_forward_barrier() {71+ self.completed72+ .retain(|completed_index, _| *completed_index > object_index);73+ self.next_object_index = object_index74+ .checked_add(1)75+ .ok_or_else(|| "Media object reorder index overflowed".to_string())?;76+ return Ok(Some(outcome));77+ }78+ if self.completed.insert(object_index, outcome).is_some() {79+ return Err(format!(80+ "Media object {object_index} completed more than once"81+ ));82+ }83+ self.take_next()84+ }85+86+ fn take_next(&mut self) -> Result<Option<MediaObjectReadOutcome>, String> {87+ let Some(outcome) = self.completed.remove(&self.next_object_index) else {88+ return Ok(None);89+ };90+ self.next_object_index = self91+ .next_object_index92+ .checked_add(1)93+ .ok_or_else(|| "Media object reorder index overflowed".to_string())?;94+ Ok(Some(outcome))95+ }96+}97+98+pub(crate) struct MediaObjectReceiver {99+ connection: iroh::endpoint::Connection,100+ reads: tokio::task::JoinSet<MediaObjectReadOutcome>,101+ reorder: MediaObjectReorder,102+ next_object_index: u64,103+ connection_closed: bool,104+}105+106+impl MediaObjectReceiver {107+ pub(crate) fn new(connection: iroh::endpoint::Connection) -> Self {108+ Self {109+ connection,110+ reads: tokio::task::JoinSet::new(),111+ reorder: MediaObjectReorder::new(1),112+ next_object_index: 0,113+ connection_closed: false,114+ }115+ }116+117+ pub(crate) async fn next(&mut self) -> Result<Option<MediaObjectReadOutcome>, String> {118+ loop {119+ if let Some(completed) = self.reorder.take_next()? {120+ return Ok(Some(completed));121+ }122+ if self.connection_closed && self.reads.is_empty() {123+ if self.reorder.pending_len() != 0 {124+ return Err("Media connection closed with an incomplete object order".into());125+ }126+ return Ok(None);127+ }128+129+ tokio::select! {130+ biased;131+ completed = self.reads.join_next(), if !self.reads.is_empty() => {132+ let completed = completed133+ .ok_or_else(|| "Media object reader ended unexpectedly".to_string())?134+ .map_err(|error| format!("Media object reader task failed: {error}"))?;135+ if let Some(completed) = self.reorder.push(completed)? {136+ return Ok(Some(completed));137+ }138+ }139+ accepted = self.connection.accept_uni(), if !self.connection_closed140+ && self.reads.len() + self.reorder.pending_len()141+ < CLIENT_MEDIA_OBJECT_CAPACITY => {142+ let mut stream = match accepted {143+ Ok(stream) => stream,144+ Err(_) => {145+ self.connection_closed = true;146+ continue;147+ }148+ };149+ self.next_object_index = self.next_object_index.checked_add(1)150+ .ok_or_else(|| "Media object index overflowed".to_string())?;151+ let object_index = self.next_object_index;152+ self.reads.spawn(async move {153+ match tokio::time::timeout(154+ CLIENT_MEDIA_OBJECT_READ_TIMEOUT,155+ read_media_object(&mut stream),156+ )157+ .await158+ {159+ Err(_) => MediaObjectReadOutcome::Dropped { object_index },160+ Ok(Err(ProtocolError::Io(_))) => {161+ MediaObjectReadOutcome::Dropped { object_index }162+ }163+ Ok(Err(error)) => {164+ MediaObjectReadOutcome::Malformed(format!(165+ "Invalid media object: {error}"166+ ))167+ }168+ Ok(Ok(frame)) => MediaObjectReadOutcome::Frame {169+ object_index,170+ frame,171+ },172+ }173+ });174+ }175+ }176+ }177+ }178+}179+180+impl Drop for MediaObjectReceiver {181+ fn drop(&mut self) {182+ self.reads.abort_all();183+ }184+}185+186+187+pub(crate) enum MediaObjectSequenceDecision {188+ Deliver { discontinuity: bool },189+ DropLate,190+ DropUntilKeyframe,191+}192+193+194+pub(crate) struct MediaObjectSequence {195+ last_sequence: Option<u64>,196+ last_object_index: u64,197+ waiting_for_keyframe: bool,198+}199+200+impl MediaObjectSequence {201+ pub(crate) fn new() -> Self {202+ Self {203+ waiting_for_keyframe: true,204+ ..Self::default()205+ }206+ }207+208+ pub(crate) fn note_dropped_object(&mut self, object_index: u64) -> bool {209+ if object_index <= self.last_object_index {210+ return false;211+ }212+ self.waiting_for_keyframe = true;213+ true214+ }215+216+ pub(crate) fn classify(217+ &mut self,218+ object_index: u64,219+ frame: &MediaFrame,220+ ) -> MediaObjectSequenceDecision {221+ if object_index <= self.last_object_index222+ || self223+ .last_sequence224+ .is_some_and(|sequence| frame.header.sequence <= sequence)225+ {226+ return MediaObjectSequenceDecision::DropLate;227+ }228+229+ let keyframe = frame.header.flags.contains(FrameFlags::KEYFRAME)230+ && frame.header.flags.contains(FrameFlags::CODEC_CONFIG);231+ let sequence_contiguous = self232+ .last_sequence233+ .is_none_or(|sequence| sequence.checked_add(1) == Some(frame.header.sequence));234+ if !keyframe && (self.waiting_for_keyframe || !sequence_contiguous) {235+ self.waiting_for_keyframe = true;236+ return MediaObjectSequenceDecision::DropUntilKeyframe;237+ }238+239+ let discontinuity = frame.header.flags.contains(FrameFlags::DISCONTINUITY)240+ || self.waiting_for_keyframe241+ || !sequence_contiguous;242+ self.last_sequence = Some(frame.header.sequence);243+ self.last_object_index = object_index;244+ self.waiting_for_keyframe = false;245+ MediaObjectSequenceDecision::Deliver { discontinuity }246+ }247+}248+249+250+pub(crate) fn media_object_frame(sequence: u64, flags: FrameFlags) -> MediaFrame {251+ let payload = vec![sequence as u8];252+ let header = sigil_protocol::MediaFrameHeader::h264(253+ 1280,254+ 800,255+ payload.len(),256+ sequence,257+ sequence * 1_000,258+ sequence as i64 * 1_000,259+ flags,260+ )261+ .unwrap();262+ MediaFrame::new(header, payload).unwrap()263+}264+265+266+mod tests {267+ use super::*;268+269+ fn media_object_outcome(270+ object_index: u64,271+ sequence: u64,272+ flags: FrameFlags,273+ ) -> MediaObjectReadOutcome {274+ MediaObjectReadOutcome::Frame {275+ object_index,276+ frame: media_object_frame(sequence, flags),277+ }278+ }279+280+ fn completed_object_index(outcome: &MediaObjectReadOutcome) -> Option<u64> {281+ outcome.object_index()282+ }283+284+ 285+ fn media_object_reorder_restores_accept_order_without_false_resync() {286+ let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);287+ let mut reorder = MediaObjectReorder::new(1);288+289+ assert!(290+ reorder291+ .push(media_object_outcome(2, 11, FrameFlags::NONE))292+ .unwrap()293+ .is_none()294+ );295+ assert_eq!(reorder.pending_len(), 1);296+ let first = reorder297+ .push(media_object_outcome(1, 10, keyframe))298+ .unwrap()299+ .unwrap();300+ assert_eq!(completed_object_index(&first), Some(1));301+ assert_eq!(302+ completed_object_index(&reorder.take_next().unwrap().unwrap()),303+ Some(2)304+ );305+ assert_eq!(reorder.pending_len(), 0);306+ }307+308+ 309+ fn explicit_discontinuity_keyframe_fast_forwards_bounded_reorder() {310+ let barrier = FrameFlags::KEYFRAME311+ .union(FrameFlags::CODEC_CONFIG)312+ .union(FrameFlags::DISCONTINUITY);313+ let mut reorder = MediaObjectReorder::new(1);314+315+ assert!(316+ reorder317+ .push(media_object_outcome(2, 11, FrameFlags::NONE))318+ .unwrap()319+ .is_none()320+ );321+ let recovered = reorder322+ .push(media_object_outcome(3, 20, barrier))323+ .unwrap()324+ .unwrap();325+ assert_eq!(completed_object_index(&recovered), Some(3));326+ assert_eq!(reorder.pending_len(), 0);327+ assert_eq!(328+ completed_object_index(329+ &reorder330+ .push(media_object_outcome(1, 10, FrameFlags::NONE))331+ .unwrap()332+ .unwrap()333+ ),334+ Some(1)335+ );336+ }337+338+ 339+ fn malformed_media_object_bypasses_reorder_and_remains_terminal() {340+ let mut reorder = MediaObjectReorder::new(1);341+ assert!(342+ reorder343+ .push(media_object_outcome(2, 11, FrameFlags::NONE))344+ .unwrap()345+ .is_none()346+ );347+ assert!(matches!(348+ reorder349+ .push(MediaObjectReadOutcome::Malformed("bad object".into()))350+ .unwrap(),351+ Some(MediaObjectReadOutcome::Malformed(_))352+ ));353+ assert_eq!(reorder.pending_len(), 1);354+ }355+356+ 357+ fn media_objects_begin_on_a_configured_keyframe_then_deliver_contiguously() {358+ let keyframe_flags = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);359+ let mut sequence = MediaObjectSequence::new();360+361+ assert_eq!(362+ sequence.classify(1, &media_object_frame(1, FrameFlags::NONE)),363+ MediaObjectSequenceDecision::DropUntilKeyframe364+ );365+ assert_eq!(366+ sequence.classify(2, &media_object_frame(2, keyframe_flags)),367+ MediaObjectSequenceDecision::Deliver {368+ discontinuity: true369+ }370+ );371+ assert_eq!(372+ sequence.classify(3, &media_object_frame(3, FrameFlags::NONE)),373+ MediaObjectSequenceDecision::Deliver {374+ discontinuity: false375+ }376+ );377+ }378+379+ 380+ fn media_object_sequence_gaps_drop_deltas_until_a_new_keyframe() {381+ let keyframe_flags = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);382+ let mut sequence = MediaObjectSequence::new();383+384+ assert!(matches!(385+ sequence.classify(1, &media_object_frame(10, keyframe_flags)),386+ MediaObjectSequenceDecision::Deliver { .. }387+ ));388+ assert_eq!(389+ sequence.classify(3, &media_object_frame(12, FrameFlags::NONE)),390+ MediaObjectSequenceDecision::DropUntilKeyframe391+ );392+ assert_eq!(393+ sequence.classify(4, &media_object_frame(13, FrameFlags::NONE)),394+ MediaObjectSequenceDecision::DropUntilKeyframe395+ );396+ assert_eq!(397+ sequence.classify(5, &media_object_frame(14, keyframe_flags)),398+ MediaObjectSequenceDecision::Deliver {399+ discontinuity: true400+ }401+ );402+ }403+404+ 405+ fn late_media_object_completion_cannot_rewind_a_recovered_stream() {406+ let keyframe_flags = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);407+ let mut sequence = MediaObjectSequence::new();408+409+ assert!(matches!(410+ sequence.classify(10, &media_object_frame(10, keyframe_flags)),411+ MediaObjectSequenceDecision::Deliver { .. }412+ ));413+ assert!(sequence.note_dropped_object(12));414+ assert_eq!(415+ sequence.classify(13, &media_object_frame(13, keyframe_flags)),416+ MediaObjectSequenceDecision::Deliver {417+ discontinuity: true418+ }419+ );420+ assert!(!sequence.note_dropped_object(11));421+ assert_eq!(422+ sequence.classify(12, &media_object_frame(12, FrameFlags::NONE)),423+ MediaObjectSequenceDecision::DropLate424+ );425+ assert_eq!(426+ sequence.classify(14, &media_object_frame(14, FrameFlags::NONE)),427+ MediaObjectSequenceDecision::Deliver {428+ discontinuity: false429+ }430+ );431+ }432+}