- commit
- c948dc88beba773d85fa6e2a17f83e58df966419
- parent
- 3f5d54cf81a046b8c880643b98cfd10c8fa1cce8
- author
- Dami <dami.etoile@gmail.com>
| M | src-tauri/ | 429 | +----------------------------- |
| M | src-tauri/ | 428 | ++++++++++++++++++++++++++++++- |
@@ -30,8 +30,9 @@ use crate::media::network_diagnostics::{
3030 NetworkLeg, NetworkSessionDiagnostics, lock_network_diagnostics,3131 };3232 use crate::media::object_receiver::{33- CLIENT_MEDIA_OBJECT_CAPACITY, CLIENT_MEDIA_OBJECT_READ_TIMEOUT, MediaObjectReadOutcome,34- MediaObjectReceiver, MediaObjectSequence, MediaObjectSequenceDecision,33+ MediaObjectReadOutcome, MediaObjectReadOutcomeV3, MediaObjectReceiver, MediaObjectReceiverV3,34+ MediaObjectSequence, MediaObjectSequenceDecision, MediaObjectSequenceDecisionV3,35+ MediaObjectSequenceV3,3536 };3637 use crate::media::transport::{3738 CLIENT_ENDPOINT_CLOSE_TIMEOUT, MediaTransport, open_negotiated_media_stream,@@ -41,11 +42,10 @@ use iroh::{Endpoint, SecretKey, endpoint::presets};
4142 use openh264::{formats::YUVSource, nal_units};4243 use serde::Serialize;4344 use sigil_protocol::{44- FrameFlags, InputEvent, InvitationGrants, KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3,45- MediaCodec, MediaControlRequestV3, MediaObjectV3, PointerSurfaceDimensions, ProtocolError,46- read_media_frame, read_media_object_v3, write_media_control_request_v3,45+ FrameFlags, InputEvent, InvitationGrants, KeyframeRequestReasonV3, MediaCodec,46+ MediaControlRequestV3, PointerSurfaceDimensions, read_media_frame,47+ write_media_control_request_v3,4748 };48-use std::collections::BTreeMap;4949 use std::io::Cursor;5050 5151 use std::sync::atomic::AtomicU64;@@ -114,276 +114,6 @@ pub struct ConnectResult {
114114 pub error: Option<String>,115115 }116116 117-118-enum MediaObjectReadOutcomeV3 {119- Object {120- accept_index: u64,121- object: MediaObjectV3,122- },123- Dropped {124- accept_index: u64,125- reason: KeyframeRequestReasonV3,126- },127- Malformed(String),128-}129-130-impl MediaObjectReadOutcomeV3 {131- fn accept_index(&self) -> Option<u64> {132- match self {133- Self::Object { accept_index, .. } | Self::Dropped { accept_index, .. } => {134- Some(*accept_index)135- }136- Self::Malformed(_) => None,137- }138- }139-140- fn is_fast_forward_barrier(&self) -> bool {141- let Self::Object { object, .. } = self else {142- return false;143- };144- object.header.object_id == 0145- && object.header.flags.contains(FrameFlags::KEYFRAME)146- && object.header.flags.contains(FrameFlags::CODEC_CONFIG)147- && object.header.flags.contains(FrameFlags::DISCONTINUITY)148- }149-}150-151-152-struct MediaObjectReorderV3 {153- next_accept_index: u64,154- completed: BTreeMap<u64, MediaObjectReadOutcomeV3>,155-}156-157-impl MediaObjectReorderV3 {158- fn new(first_accept_index: u64) -> Self {159- Self {160- next_accept_index: first_accept_index,161- completed: BTreeMap::new(),162- }163- }164-165- fn pending_len(&self) -> usize {166- self.completed.len()167- }168-169- fn push(170- &mut self,171- outcome: MediaObjectReadOutcomeV3,172- ) -> Result<Option<MediaObjectReadOutcomeV3>, String> {173- let Some(accept_index) = outcome.accept_index() else {174- return Ok(Some(outcome));175- };176- if accept_index < self.next_accept_index {177- // A discontinuity barrier may advance beyond older in-flight178- // reads. Their eventual timeout/reset outcomes belong to the179- // superseded GOP and must not poison the recovered sequence.180- return Ok(None);181- }182- if self.completed.insert(accept_index, outcome).is_some() {183- return Err(format!(184- "Duplicate media v3 accept index {accept_index} completed"185- ));186- }187- if self188- .completed189- .get(&accept_index)190- .is_some_and(MediaObjectReadOutcomeV3::is_fast_forward_barrier)191- {192- self.completed.retain(|index, _| *index >= accept_index);193- self.next_accept_index = accept_index;194- }195- self.take_next()196- }197-198- fn take_next(&mut self) -> Result<Option<MediaObjectReadOutcomeV3>, String> {199- let Some(outcome) = self.completed.remove(&self.next_accept_index) else {200- return Ok(None);201- };202- self.next_accept_index = self203- .next_accept_index204- .checked_add(1)205- .ok_or_else(|| "Media v3 accept index overflowed".to_string())?;206- Ok(Some(outcome))207- }208-}209-210-struct MediaObjectReceiverV3 {211- connection: iroh::endpoint::Connection,212- reads: tokio::task::JoinSet<MediaObjectReadOutcomeV3>,213- reorder: MediaObjectReorderV3,214- next_accept_index: u64,215- connection_closed: bool,216-}217-218-impl MediaObjectReceiverV3 {219- fn new(connection: iroh::endpoint::Connection) -> Self {220- Self {221- connection,222- reads: tokio::task::JoinSet::new(),223- reorder: MediaObjectReorderV3::new(1),224- next_accept_index: 0,225- connection_closed: false,226- }227- }228-229- async fn next(&mut self) -> Result<Option<MediaObjectReadOutcomeV3>, String> {230- loop {231- if let Some(completed) = self.reorder.take_next()? {232- return Ok(Some(completed));233- }234- if self.connection_closed && self.reads.is_empty() {235- if self.reorder.pending_len() != 0 {236- return Err("Media v3 connection closed with incomplete object order".into());237- }238- return Ok(None);239- }240-241- tokio::select! {242- biased;243- completed = self.reads.join_next(), if !self.reads.is_empty() => {244- let completed = completed245- .ok_or_else(|| "Media v3 object reader ended unexpectedly".to_string())?246- .map_err(|error| format!("Media v3 object reader task failed: {error}"))?;247- if let Some(completed) = self.reorder.push(completed)? {248- return Ok(Some(completed));249- }250- }251- accepted = self.connection.accept_uni(), if !self.connection_closed252- && self.reads.len() + self.reorder.pending_len()253- < CLIENT_MEDIA_OBJECT_CAPACITY => {254- let mut stream = match accepted {255- Ok(stream) => stream,256- Err(_) => {257- self.connection_closed = true;258- continue;259- }260- };261- self.next_accept_index = self.next_accept_index.checked_add(1)262- .ok_or_else(|| "Media v3 accept index overflowed".to_string())?;263- let accept_index = self.next_accept_index;264- self.reads.spawn(async move {265- match tokio::time::timeout(266- CLIENT_MEDIA_OBJECT_READ_TIMEOUT,267- read_media_object_v3(&mut stream),268- )269- .await270- {271- Err(_) => MediaObjectReadOutcomeV3::Dropped {272- accept_index,273- reason: KeyframeRequestReasonV3::DeliveryTimeout,274- },275- Ok(Err(ProtocolError::Io(_))) => MediaObjectReadOutcomeV3::Dropped {276- accept_index,277- reason: KeyframeRequestReasonV3::TransportGap,278- },279- Ok(Err(error)) => MediaObjectReadOutcomeV3::Malformed(format!(280- "Invalid media v3 object: {error}"281- )),282- Ok(Ok(object)) => MediaObjectReadOutcomeV3::Object {283- accept_index,284- object,285- },286- }287- });288- }289- }290- }291- }292-}293-294-impl Drop for MediaObjectReceiverV3 {295- fn drop(&mut self) {296- self.reads.abort_all();297- }298-}299-300-301-enum MediaObjectSequenceDecisionV3 {302- Deliver { discontinuity: bool },303- DropLate,304- DropUntilKeyframe,305-}306-307-308-struct MediaObjectSequenceV3 {309- group_id: Option<u64>,310- last_object_id: Option<u32>,311- last_sequence: Option<u64>,312- group_payload_bytes: usize,313- waiting_for_keyframe: bool,314-}315-316-impl MediaObjectSequenceV3 {317- fn new() -> Self {318- Self {319- waiting_for_keyframe: true,320- ..Self::default()321- }322- }323-324- fn note_dropped_object(&mut self) -> bool {325- let entered = !self.waiting_for_keyframe;326- self.waiting_for_keyframe = true;327- entered328- }329-330- fn classify(&mut self, object: &MediaObjectV3) -> MediaObjectSequenceDecisionV3 {331- let header = &object.header;332- if self333- .group_id334- .is_some_and(|group_id| header.group_id < group_id)335- || self336- .last_sequence337- .is_some_and(|sequence| header.sequence <= sequence)338- {339- return MediaObjectSequenceDecisionV3::DropLate;340- }341-342- let new_group = self.group_id != Some(header.group_id);343- let recovery_keyframe = header.object_id == 0344- && header.flags.contains(FrameFlags::KEYFRAME)345- && header.flags.contains(FrameFlags::CODEC_CONFIG);346- if new_group && !recovery_keyframe {347- self.waiting_for_keyframe = true;348- return MediaObjectSequenceDecisionV3::DropUntilKeyframe;349- }350- if !new_group && self.waiting_for_keyframe {351- return MediaObjectSequenceDecisionV3::DropUntilKeyframe;352- }353-354- let sequence_contiguous = self355- .last_sequence356- .is_none_or(|sequence| sequence.checked_add(1) == Some(header.sequence));357- let object_contiguous = new_group358- || self359- .last_object_id360- .is_some_and(|object_id| object_id.checked_add(1) == Some(header.object_id));361- let next_group_bytes = if new_group {362- object.payload.len()363- } else {364- self.group_payload_bytes365- .saturating_add(object.payload.len())366- };367- if (!sequence_contiguous && !new_group)368- || !object_contiguous369- || next_group_bytes > MAX_MEDIA_GROUP_BYTES_V3370- {371- self.waiting_for_keyframe = true;372- return MediaObjectSequenceDecisionV3::DropUntilKeyframe;373- }374-375- let discontinuity = header.flags.contains(FrameFlags::DISCONTINUITY)376- || self.waiting_for_keyframe377- || !sequence_contiguous;378- self.group_id = Some(header.group_id);379- self.last_object_id = Some(header.object_id);380- self.last_sequence = Some(header.sequence);381- self.group_payload_bytes = next_group_bytes;382- self.waiting_for_keyframe = false;383- MediaObjectSequenceDecisionV3::Deliver { discontinuity }384- }385-}386-387117 async fn run_media_control_writer_v3(388118 mut stream: iroh::endpoint::SendStream,389119 mut requests: tokio::sync::mpsc::Receiver<(KeyframeRequestReasonV3, Option<u64>)>,@@ -945,7 +675,7 @@ pub async fn iroh_client_connect(
945675 try_queue_media_keyframe_request(946676 media_control_requests.as_ref(),947677 reason,948- media_object_sequence_v3.last_sequence,678+ media_object_sequence_v3.last_sequence(),949679 );950680 }951681 }@@ -954,7 +684,7 @@ pub async fn iroh_client_connect(
954684 break 'frames;955685 }956686 MediaObjectReadOutcomeV3::Object { object, .. } => {957- let was_waiting = media_object_sequence_v3.waiting_for_keyframe;687+ let was_waiting = media_object_sequence_v3.waiting_for_keyframe();958688 let discontinuity = match media_object_sequence_v3.classify(&object)959689 {960690 MediaObjectSequenceDecisionV3::Deliver { discontinuity } => {@@ -974,7 +704,7 @@ pub async fn iroh_client_connect(
974704 try_queue_media_keyframe_request(975705 media_control_requests.as_ref(),976706 KeyframeRequestReasonV3::TransportGap,977- media_object_sequence_v3.last_sequence,707+ media_object_sequence_v3.last_sequence(),978708 );979709 }980710 continue;@@ -1365,147 +1095,6 @@ fn sequence_gap(previous: u64, current: u64) -> Result<u64, String> {
13651095 mod tests {13661096 use super::*;13671097 1368- fn media_object_v3(1369- group_id: u64,1370- object_id: u32,1371- sequence: u64,1372- flags: FrameFlags,1373- ) -> MediaObjectV3 {1374- let payload = vec![sequence as u8];1375- let header = sigil_protocol::MediaObjectHeaderV3::h264(1376- 1280,1377- 800,1378- payload.len(),1379- if object_id == 0 { 0 } else { 128 },1380- flags,1381- object_id,1382- group_id,1383- sequence,1384- sequence * 1_000,1385- sequence as i64 * 1_000,1386- 100,1387- )1388- .unwrap();1389- MediaObjectV3::new(header, payload).unwrap()1390- }1391-1392- fn media_object_outcome_v3(1393- accept_index: u64,1394- group_id: u64,1395- object_id: u32,1396- sequence: u64,1397- flags: FrameFlags,1398- ) -> MediaObjectReadOutcomeV3 {1399- MediaObjectReadOutcomeV3::Object {1400- accept_index,1401- object: media_object_v3(group_id, object_id, sequence, flags),1402- }1403- }1404-1405- 1406- fn media_v3_groups_use_wire_object_identity_and_recover_on_new_group_zero() {1407- let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);1408- let mut sequence = MediaObjectSequenceV3::new();1409-1410- assert_eq!(1411- sequence.classify(&media_object_v3(10, 0, 10, keyframe)),1412- MediaObjectSequenceDecisionV3::Deliver {1413- discontinuity: true1414- }1415- );1416- assert_eq!(1417- sequence.classify(&media_object_v3(10, 1, 11, FrameFlags::NONE)),1418- MediaObjectSequenceDecisionV3::Deliver {1419- discontinuity: false1420- }1421- );1422- assert_eq!(1423- sequence.classify(&media_object_v3(10, 3, 13, FrameFlags::NONE)),1424- MediaObjectSequenceDecisionV3::DropUntilKeyframe1425- );1426- assert_eq!(1427- sequence.classify(&media_object_v3(10, 2, 12, FrameFlags::NONE)),1428- MediaObjectSequenceDecisionV3::DropUntilKeyframe1429- );1430- assert_eq!(1431- sequence.classify(&media_object_v3(1432- 20,1433- 0,1434- 20,1435- keyframe.union(FrameFlags::DISCONTINUITY),1436- )),1437- MediaObjectSequenceDecisionV3::Deliver {1438- discontinuity: true1439- }1440- );1441- assert_eq!(1442- sequence.classify(&media_object_v3(10, 2, 12, FrameFlags::NONE)),1443- MediaObjectSequenceDecisionV3::DropLate1444- );1445- }1446-1447- 1448- fn late_dropped_completion_after_v3_barrier_cannot_poison_recovered_group() {1449- let barrier = FrameFlags::KEYFRAME1450- .union(FrameFlags::CODEC_CONFIG)1451- .union(FrameFlags::DISCONTINUITY);1452- let mut reorder = MediaObjectReorderV3::new(1);1453- let mut sequence = MediaObjectSequenceV3::new();1454-1455- let recovered = reorder1456- .push(media_object_outcome_v3(5, 20, 0, 20, barrier))1457- .unwrap()1458- .unwrap();1459- let MediaObjectReadOutcomeV3::Object { object, .. } = recovered else {1460- panic!("recovery barrier must remain an object");1461- };1462- assert_eq!(1463- sequence.classify(&object),1464- MediaObjectSequenceDecisionV3::Deliver {1465- discontinuity: true1466- }1467- );1468-1469- assert!(1470- reorder1471- .push(MediaObjectReadOutcomeV3::Dropped {1472- accept_index: 1,1473- reason: KeyframeRequestReasonV3::DeliveryTimeout,1474- })1475- .unwrap()1476- .is_none()1477- );1478-1479- let delta = reorder1480- .push(media_object_outcome_v3(6, 20, 1, 21, FrameFlags::NONE))1481- .unwrap()1482- .unwrap();1483- let MediaObjectReadOutcomeV3::Object { object, .. } = delta else {1484- panic!("new group delta must remain an object");1485- };1486- assert_eq!(1487- sequence.classify(&object),1488- MediaObjectSequenceDecisionV3::Deliver {1489- discontinuity: false1490- }1491- );1492- }1493-1494- 1495- fn media_v3_receiver_rejects_group_payload_growth_beyond_the_shared_cap() {1496- let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);1497- let mut sequence = MediaObjectSequenceV3::new();1498- assert!(matches!(1499- sequence.classify(&media_object_v3(1, 0, 1, keyframe)),1500- MediaObjectSequenceDecisionV3::Deliver { .. }1501- ));1502- sequence.group_payload_bytes = MAX_MEDIA_GROUP_BYTES_V3;1503- assert_eq!(1504- sequence.classify(&media_object_v3(1, 1, 2, FrameFlags::NONE)),1505- MediaObjectSequenceDecisionV3::DropUntilKeyframe1506- );1507- }1508-15091098 15101099 fn keyframe_request_reason_mapping_is_strict_and_coalesces_decoder_reasons() {15111100 assert_eq!(@@ -1,10 +1,13 @@
11 use std::collections::BTreeMap;22 use std::time::Duration;33 4-use sigil_protocol::{FrameFlags, MediaFrame, ProtocolError, read_media_object};4+use sigil_protocol::{5+ FrameFlags, KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3, MediaFrame, MediaObjectV3,6+ ProtocolError, read_media_object, read_media_object_v3,7+};58 6-pub(crate) const CLIENT_MEDIA_OBJECT_CAPACITY: usize = 4;7-pub(crate) const CLIENT_MEDIA_OBJECT_READ_TIMEOUT: Duration = Duration::from_secs(1);9+const CLIENT_MEDIA_OBJECT_CAPACITY: usize = 4;10+const CLIENT_MEDIA_OBJECT_READ_TIMEOUT: Duration = Duration::from_secs(1);811 912 1013 pub(crate) enum MediaObjectReadOutcome {@@ -246,6 +249,284 @@ impl MediaObjectSequence {
246249 }247250 }248251 252+253+pub(crate) enum MediaObjectReadOutcomeV3 {254+ Object {255+ accept_index: u64,256+ object: MediaObjectV3,257+ },258+ Dropped {259+ accept_index: u64,260+ reason: KeyframeRequestReasonV3,261+ },262+ Malformed(String),263+}264+265+impl MediaObjectReadOutcomeV3 {266+ fn accept_index(&self) -> Option<u64> {267+ match self {268+ Self::Object { accept_index, .. } | Self::Dropped { accept_index, .. } => {269+ Some(*accept_index)270+ }271+ Self::Malformed(_) => None,272+ }273+ }274+275+ fn is_fast_forward_barrier(&self) -> bool {276+ let Self::Object { object, .. } = self else {277+ return false;278+ };279+ object.header.object_id == 0280+ && object.header.flags.contains(FrameFlags::KEYFRAME)281+ && object.header.flags.contains(FrameFlags::CODEC_CONFIG)282+ && object.header.flags.contains(FrameFlags::DISCONTINUITY)283+ }284+}285+286+287+struct MediaObjectReorderV3 {288+ next_accept_index: u64,289+ completed: BTreeMap<u64, MediaObjectReadOutcomeV3>,290+}291+292+impl MediaObjectReorderV3 {293+ fn new(first_accept_index: u64) -> Self {294+ Self {295+ next_accept_index: first_accept_index,296+ completed: BTreeMap::new(),297+ }298+ }299+300+ fn pending_len(&self) -> usize {301+ self.completed.len()302+ }303+304+ fn push(305+ &mut self,306+ outcome: MediaObjectReadOutcomeV3,307+ ) -> Result<Option<MediaObjectReadOutcomeV3>, String> {308+ let Some(accept_index) = outcome.accept_index() else {309+ return Ok(Some(outcome));310+ };311+ if accept_index < self.next_accept_index {312+ // A discontinuity barrier may advance beyond older in-flight313+ // reads. Their eventual timeout/reset outcomes belong to the314+ // superseded GOP and must not poison the recovered sequence.315+ return Ok(None);316+ }317+ if self.completed.insert(accept_index, outcome).is_some() {318+ return Err(format!(319+ "Duplicate media v3 accept index {accept_index} completed"320+ ));321+ }322+ if self323+ .completed324+ .get(&accept_index)325+ .is_some_and(MediaObjectReadOutcomeV3::is_fast_forward_barrier)326+ {327+ self.completed.retain(|index, _| *index >= accept_index);328+ self.next_accept_index = accept_index;329+ }330+ self.take_next()331+ }332+333+ fn take_next(&mut self) -> Result<Option<MediaObjectReadOutcomeV3>, String> {334+ let Some(outcome) = self.completed.remove(&self.next_accept_index) else {335+ return Ok(None);336+ };337+ self.next_accept_index = self338+ .next_accept_index339+ .checked_add(1)340+ .ok_or_else(|| "Media v3 accept index overflowed".to_string())?;341+ Ok(Some(outcome))342+ }343+}344+345+pub(crate) struct MediaObjectReceiverV3 {346+ connection: iroh::endpoint::Connection,347+ reads: tokio::task::JoinSet<MediaObjectReadOutcomeV3>,348+ reorder: MediaObjectReorderV3,349+ next_accept_index: u64,350+ connection_closed: bool,351+}352+353+impl MediaObjectReceiverV3 {354+ pub(crate) fn new(connection: iroh::endpoint::Connection) -> Self {355+ Self {356+ connection,357+ reads: tokio::task::JoinSet::new(),358+ reorder: MediaObjectReorderV3::new(1),359+ next_accept_index: 0,360+ connection_closed: false,361+ }362+ }363+364+ pub(crate) async fn next(&mut self) -> Result<Option<MediaObjectReadOutcomeV3>, String> {365+ loop {366+ if let Some(completed) = self.reorder.take_next()? {367+ return Ok(Some(completed));368+ }369+ if self.connection_closed && self.reads.is_empty() {370+ if self.reorder.pending_len() != 0 {371+ return Err("Media v3 connection closed with incomplete object order".into());372+ }373+ return Ok(None);374+ }375+376+ tokio::select! {377+ biased;378+ completed = self.reads.join_next(), if !self.reads.is_empty() => {379+ let completed = completed380+ .ok_or_else(|| "Media v3 object reader ended unexpectedly".to_string())?381+ .map_err(|error| format!("Media v3 object reader task failed: {error}"))?;382+ if let Some(completed) = self.reorder.push(completed)? {383+ return Ok(Some(completed));384+ }385+ }386+ accepted = self.connection.accept_uni(), if !self.connection_closed387+ && self.reads.len() + self.reorder.pending_len()388+ < CLIENT_MEDIA_OBJECT_CAPACITY => {389+ let mut stream = match accepted {390+ Ok(stream) => stream,391+ Err(_) => {392+ self.connection_closed = true;393+ continue;394+ }395+ };396+ self.next_accept_index = self.next_accept_index.checked_add(1)397+ .ok_or_else(|| "Media v3 accept index overflowed".to_string())?;398+ let accept_index = self.next_accept_index;399+ self.reads.spawn(async move {400+ match tokio::time::timeout(401+ CLIENT_MEDIA_OBJECT_READ_TIMEOUT,402+ read_media_object_v3(&mut stream),403+ )404+ .await405+ {406+ Err(_) => MediaObjectReadOutcomeV3::Dropped {407+ accept_index,408+ reason: KeyframeRequestReasonV3::DeliveryTimeout,409+ },410+ Ok(Err(ProtocolError::Io(_))) => MediaObjectReadOutcomeV3::Dropped {411+ accept_index,412+ reason: KeyframeRequestReasonV3::TransportGap,413+ },414+ Ok(Err(error)) => MediaObjectReadOutcomeV3::Malformed(format!(415+ "Invalid media v3 object: {error}"416+ )),417+ Ok(Ok(object)) => MediaObjectReadOutcomeV3::Object {418+ accept_index,419+ object,420+ },421+ }422+ });423+ }424+ }425+ }426+ }427+}428+429+impl Drop for MediaObjectReceiverV3 {430+ fn drop(&mut self) {431+ self.reads.abort_all();432+ }433+}434+435+436+pub(crate) enum MediaObjectSequenceDecisionV3 {437+ Deliver { discontinuity: bool },438+ DropLate,439+ DropUntilKeyframe,440+}441+442+443+pub(crate) struct MediaObjectSequenceV3 {444+ group_id: Option<u64>,445+ last_object_id: Option<u32>,446+ last_sequence: Option<u64>,447+ group_payload_bytes: usize,448+ waiting_for_keyframe: bool,449+}450+451+impl MediaObjectSequenceV3 {452+ pub(crate) fn new() -> Self {453+ Self {454+ waiting_for_keyframe: true,455+ ..Self::default()456+ }457+ }458+459+ pub(crate) fn last_sequence(&self) -> Option<u64> {460+ self.last_sequence461+ }462+463+ pub(crate) fn waiting_for_keyframe(&self) -> bool {464+ self.waiting_for_keyframe465+ }466+467+ pub(crate) fn note_dropped_object(&mut self) -> bool {468+ let entered = !self.waiting_for_keyframe;469+ self.waiting_for_keyframe = true;470+ entered471+ }472+473+ pub(crate) fn classify(&mut self, object: &MediaObjectV3) -> MediaObjectSequenceDecisionV3 {474+ let header = &object.header;475+ if self476+ .group_id477+ .is_some_and(|group_id| header.group_id < group_id)478+ || self479+ .last_sequence480+ .is_some_and(|sequence| header.sequence <= sequence)481+ {482+ return MediaObjectSequenceDecisionV3::DropLate;483+ }484+485+ let new_group = self.group_id != Some(header.group_id);486+ let recovery_keyframe = header.object_id == 0487+ && header.flags.contains(FrameFlags::KEYFRAME)488+ && header.flags.contains(FrameFlags::CODEC_CONFIG);489+ if new_group && !recovery_keyframe {490+ self.waiting_for_keyframe = true;491+ return MediaObjectSequenceDecisionV3::DropUntilKeyframe;492+ }493+ if !new_group && self.waiting_for_keyframe {494+ return MediaObjectSequenceDecisionV3::DropUntilKeyframe;495+ }496+497+ let sequence_contiguous = self498+ .last_sequence499+ .is_none_or(|sequence| sequence.checked_add(1) == Some(header.sequence));500+ let object_contiguous = new_group501+ || self502+ .last_object_id503+ .is_some_and(|object_id| object_id.checked_add(1) == Some(header.object_id));504+ let next_group_bytes = if new_group {505+ object.payload.len()506+ } else {507+ self.group_payload_bytes508+ .saturating_add(object.payload.len())509+ };510+ if (!sequence_contiguous && !new_group)511+ || !object_contiguous512+ || next_group_bytes > MAX_MEDIA_GROUP_BYTES_V3513+ {514+ self.waiting_for_keyframe = true;515+ return MediaObjectSequenceDecisionV3::DropUntilKeyframe;516+ }517+518+ let discontinuity = header.flags.contains(FrameFlags::DISCONTINUITY)519+ || self.waiting_for_keyframe520+ || !sequence_contiguous;521+ self.group_id = Some(header.group_id);522+ self.last_object_id = Some(header.object_id);523+ self.last_sequence = Some(header.sequence);524+ self.group_payload_bytes = next_group_bytes;525+ self.waiting_for_keyframe = false;526+ MediaObjectSequenceDecisionV3::Deliver { discontinuity }527+ }528+}529+249530 250531 pub(crate) fn media_object_frame(sequence: u64, flags: FrameFlags) -> MediaFrame {251532 let payload = vec![sequence as u8];@@ -266,6 +547,147 @@ pub(crate) fn media_object_frame(sequence: u64, flags: FrameFlags) -> MediaFrame
266547 mod tests {267548 use super::*;268549 550+ fn media_object_v3(551+ group_id: u64,552+ object_id: u32,553+ sequence: u64,554+ flags: FrameFlags,555+ ) -> MediaObjectV3 {556+ let payload = vec![sequence as u8];557+ let header = sigil_protocol::MediaObjectHeaderV3::h264(558+ 1280,559+ 800,560+ payload.len(),561+ if object_id == 0 { 0 } else { 128 },562+ flags,563+ object_id,564+ group_id,565+ sequence,566+ sequence * 1_000,567+ sequence as i64 * 1_000,568+ 100,569+ )570+ .unwrap();571+ MediaObjectV3::new(header, payload).unwrap()572+ }573+574+ fn media_object_outcome_v3(575+ accept_index: u64,576+ group_id: u64,577+ object_id: u32,578+ sequence: u64,579+ flags: FrameFlags,580+ ) -> MediaObjectReadOutcomeV3 {581+ MediaObjectReadOutcomeV3::Object {582+ accept_index,583+ object: media_object_v3(group_id, object_id, sequence, flags),584+ }585+ }586+587+ 588+ fn media_v3_groups_use_wire_object_identity_and_recover_on_new_group_zero() {589+ let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);590+ let mut sequence = MediaObjectSequenceV3::new();591+592+ assert_eq!(593+ sequence.classify(&media_object_v3(10, 0, 10, keyframe)),594+ MediaObjectSequenceDecisionV3::Deliver {595+ discontinuity: true596+ }597+ );598+ assert_eq!(599+ sequence.classify(&media_object_v3(10, 1, 11, FrameFlags::NONE)),600+ MediaObjectSequenceDecisionV3::Deliver {601+ discontinuity: false602+ }603+ );604+ assert_eq!(605+ sequence.classify(&media_object_v3(10, 3, 13, FrameFlags::NONE)),606+ MediaObjectSequenceDecisionV3::DropUntilKeyframe607+ );608+ assert_eq!(609+ sequence.classify(&media_object_v3(10, 2, 12, FrameFlags::NONE)),610+ MediaObjectSequenceDecisionV3::DropUntilKeyframe611+ );612+ assert_eq!(613+ sequence.classify(&media_object_v3(614+ 20,615+ 0,616+ 20,617+ keyframe.union(FrameFlags::DISCONTINUITY),618+ )),619+ MediaObjectSequenceDecisionV3::Deliver {620+ discontinuity: true621+ }622+ );623+ assert_eq!(624+ sequence.classify(&media_object_v3(10, 2, 12, FrameFlags::NONE)),625+ MediaObjectSequenceDecisionV3::DropLate626+ );627+ }628+629+ 630+ fn late_dropped_completion_after_v3_barrier_cannot_poison_recovered_group() {631+ let barrier = FrameFlags::KEYFRAME632+ .union(FrameFlags::CODEC_CONFIG)633+ .union(FrameFlags::DISCONTINUITY);634+ let mut reorder = MediaObjectReorderV3::new(1);635+ let mut sequence = MediaObjectSequenceV3::new();636+637+ let recovered = reorder638+ .push(media_object_outcome_v3(5, 20, 0, 20, barrier))639+ .unwrap()640+ .unwrap();641+ let MediaObjectReadOutcomeV3::Object { object, .. } = recovered else {642+ panic!("recovery barrier must remain an object");643+ };644+ assert_eq!(645+ sequence.classify(&object),646+ MediaObjectSequenceDecisionV3::Deliver {647+ discontinuity: true648+ }649+ );650+651+ assert!(652+ reorder653+ .push(MediaObjectReadOutcomeV3::Dropped {654+ accept_index: 1,655+ reason: KeyframeRequestReasonV3::DeliveryTimeout,656+ })657+ .unwrap()658+ .is_none()659+ );660+661+ let delta = reorder662+ .push(media_object_outcome_v3(6, 20, 1, 21, FrameFlags::NONE))663+ .unwrap()664+ .unwrap();665+ let MediaObjectReadOutcomeV3::Object { object, .. } = delta else {666+ panic!("new group delta must remain an object");667+ };668+ assert_eq!(669+ sequence.classify(&object),670+ MediaObjectSequenceDecisionV3::Deliver {671+ discontinuity: false672+ }673+ );674+ }675+676+ 677+ fn media_v3_receiver_rejects_group_payload_growth_beyond_the_shared_cap() {678+ let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);679+ let mut sequence = MediaObjectSequenceV3::new();680+ assert!(matches!(681+ sequence.classify(&media_object_v3(1, 0, 1, keyframe)),682+ MediaObjectSequenceDecisionV3::Deliver { .. }683+ ));684+ sequence.group_payload_bytes = MAX_MEDIA_GROUP_BYTES_V3;685+ assert_eq!(686+ sequence.classify(&media_object_v3(1, 1, 2, FrameFlags::NONE)),687+ MediaObjectSequenceDecisionV3::DropUntilKeyframe688+ );689+ }690+269691 fn media_object_outcome(270692 object_index: u64,271693 sequence: u64,