- commit
- 77aacb288d396ce1fc4c85e650d20c02cc31bfcd
- parent
- b3ed6c07989654b6f5a5885b5f6c09694a68471c
- author
- Dami <dami.etoile@gmail.com>
| M | src-tauri/ | 691 | +----------------------------- |
| M | src-tauri/ | 1 | + |
| A | src-tauri/ | 688 | ++++++++++++++++++++++++++++++ |
| M | src-tauri/ | 2 | + |
@@ -5,49 +5,41 @@ use super::state::{
55 MediaFeedbackSender, development_direct_node_available,66 };77 use crate::media::audio_delivery::{8- self, AudioStartRequest, cancel_audio_generation, lock_audio_deliveries, next_audio_generation,9- try_start_audio,8+ self, AudioStartRequest, lock_audio_deliveries, next_audio_generation, try_start_audio,109 };1110 pub use crate::media::frame_channel::FramePayload;1211 use crate::media::frame_channel::{1312 FrameEnvelopeMetadata, byte_to_codec, close_generation_connection, emit_frame_error,1413 encode_frame_envelope, next_media_generation, release_frame_channel_slot,15- release_frame_channel_slot_for_generation, take_generation_owned, take_generation_owned_triple,14+ release_frame_channel_slot_for_generation, take_generation_owned_triple,1615 try_reserve_frame_channel_slot, validate_legacy_media_header,1716 };1817 use crate::media::metrics::{ClientMediaMetrics, lock_client_media_metrics};19-use crate::media::moq_catalog::subscribe_goq_video_track;18+use crate::media::moq_receiver::{19+ MoqMediaReadOutcome, open_upstream_moq_media, retire_upstream_moq_generation,20+};2021 use crate::media::network_diagnostics::{NetworkLeg, NetworkSessionDiagnostics};21-22-use crate::media::object_receiver::media_object_frame;2322 use crate::media::object_receiver::{2423 CLIENT_MEDIA_OBJECT_CAPACITY, CLIENT_MEDIA_OBJECT_READ_TIMEOUT, MediaObjectReadOutcome,2524 MediaObjectReceiver, MediaObjectSequence, MediaObjectSequenceDecision,2625 };2726 use crate::media::transport::{28- MediaTransport, NegotiatedV1Stream, connect_error_is_unsupported_alpn, negotiate_v1,29- open_negotiated_media_stream,27+ CLIENT_ENDPOINT_CLOSE_TIMEOUT, MediaTransport, NegotiatedV1Stream,28+ connect_error_is_unsupported_alpn, negotiate_v1, open_negotiated_media_stream,3029 };3130 use base64::Engine;3231 use iroh::{Endpoint, SecretKey, endpoint::presets};33-use iroh_moq::{Moq, MoqSession};34-35-use moq_net::Track;36-use moq_net::{BroadcastConsumer, GroupConsumer, TrackConsumer};3732 use openh264::{formats::YUVSource, nal_units};3833 use serde::{Deserialize, Serialize};39-40-use sigil_protocol::MOQ_VIDEO_H264_TRACK;4134 use sigil_protocol::{4235 AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1, AdaptiveBitrateStateV1, Capability,4336 FrameFlags, INPUT_ALPN_V1, InputEvent, InvitationGrants, KeyframeRequestReasonV3,44- MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MAX_MEDIA_OBJECT_ID_V3,45- MEDIA_FEEDBACK_ALPN_V1, MediaCodec, MediaControlRequestV3, MediaFeedbackFlags,46- MediaFeedbackReportV1, MediaFrame, MediaObjectV3, PointerPosition, PointerSurfaceDimensions,47- ProtocolError, RELATIVE_POINTER_DELTA_MAX, RELATIVE_POINTER_DELTA_MIN,48- decode_media_frame_object, media_moq_broadcast_name, read_adaptive_bitrate_decision_v1,49- read_input_ack, read_media_frame, read_media_object_v3, write_input_event,50- write_media_control_request_v3, write_media_feedback_report_v1,37+ MAX_MEDIA_GROUP_BYTES_V3, MEDIA_FEEDBACK_ALPN_V1, MediaCodec, MediaControlRequestV3,38+ MediaFeedbackFlags, MediaFeedbackReportV1, MediaObjectV3, PointerPosition,39+ PointerSurfaceDimensions, ProtocolError, RELATIVE_POINTER_DELTA_MAX,40+ RELATIVE_POINTER_DELTA_MIN, read_adaptive_bitrate_decision_v1, read_input_ack,41+ read_media_frame, read_media_object_v3, write_input_event, write_media_control_request_v3,42+ write_media_feedback_report_v1,5143 };5244 use std::collections::BTreeMap;5345 use std::io::Cursor;@@ -62,17 +54,7 @@ use tauri::{
6254 // ─── Client commands ──────────────────────────────────────────────────────────6355 6456 const LEGACY_MEDIA_IDLE_TIMEOUT: Duration = Duration::from_secs(10);65-const CLIENT_MOQ_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);66-const CLIENT_MOQ_SUBSCRIBE_TIMEOUT: Duration = Duration::from_secs(10);67-// Match the protocol's maximum single-object delivery horizon. A publisher may68-// never hold a partially delivered object open indefinitely.69-const CLIENT_MOQ_OBJECT_READ_TIMEOUT: Duration =70- Duration::from_millis(MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS as u64);71-// Sigil's external encoder can take 500 ms to reach its next configured IDR.72-// Allow another 500 ms for a relay path to deliver the superseding group.73-const CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT: Duration = Duration::from_secs(1);7457 const CLIENT_FRAME_STATS_INTERVAL: Duration = Duration::from_millis(250);75-const CLIENT_ENDPOINT_CLOSE_TIMEOUT: Duration = Duration::from_secs(2);7658 const CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);7759 const CLIENT_MEDIA_FEEDBACK_IO_TIMEOUT: Duration = Duration::from_secs(5);7860 const CLIENT_ADAPTIVE_DECISION_DELIVERY_INTERVAL: Duration = Duration::from_secs(1);@@ -99,80 +81,6 @@ async fn retire_media_feedback_generation(app: &AppHandle, generation: u64) -> b
9981 true10082 }10183 102-async fn retire_upstream_moq_generation(103- app: &AppHandle,104- media_generation: u64,105- audio_generation: Option<u64>,106-) -> bool {107- let state = app.state::<AppState>();108- let endpoint = {109- // Selection and retirement are one generation-checked transaction.110- // A stale reader must never close a replacement session whose task was111- // installed after an explicit disconnect/reconnect.112- let _connection_serial = state.client_connection_serial.lock().await;113- let media_connection = {114- let mut slot = state.media_connection.lock().await;115- take_generation_owned(&mut slot, media_generation)116- };117- let Some(media_connection) = media_connection else {118- return false;119- };120-121- {122- let mut control = state.media_control.lock().await;123- let _ = take_generation_owned(&mut control, media_generation);124- }125- {126- let mut delivery = state.frame_delivery.lock().await;127- let _ = take_generation_owned(&mut delivery, media_generation);128- }129- let feedback_connection = {130- let mut feedback = state.media_feedback.lock().await;131- take_generation_owned_triple(&mut feedback, media_generation)132- .map(|(connection, _)| connection)133- };134- *state.input_send.lock().await = None;135-136- let audio_connection = if let Some(audio_generation) = audio_generation {137- if let Err(error) = cancel_audio_generation(138- &state.audio_connection_generation,139- &state.audio_deliveries,140- audio_generation,141- ) {142- eprintln!(143- "[client] failed to retire audio generation after upstream MoQ ended: {error}"144- );145- }146- let mut slot = state.audio_connection.lock().await;147- take_generation_owned(&mut slot, audio_generation)148- } else {149- None150- };151-152- let endpoint = state.client_endpoint.lock().await.take();153- media_connection.close(0_u32.into(), b"upstream MoQ media ended");154- if let Some(feedback_connection) = feedback_connection {155- feedback_connection.close(0_u32.into(), b"upstream MoQ media ended");156- }157- if let Some(audio_connection) = audio_connection {158- audio_connection.close(0_u32.into(), b"upstream MoQ media ended");159- }160- state161- .client_connection_active162- .store(false, Ordering::SeqCst);163- endpoint164- };165-166- if let Some(endpoint) = endpoint167- && tokio::time::timeout(CLIENT_ENDPOINT_CLOSE_TIMEOUT, endpoint.close())168- .await169- .is_err()170- {171- eprintln!("[client] timed out retiring endpoint after upstream MoQ media ended");172- }173- true174-}175-17684 struct ClientConnectGuard {17785 active: Arc<AtomicBool>,17886 committed: bool,@@ -263,355 +171,6 @@ struct AdaptiveBitrateDecisionDiagnostic {
263171 applied: bool,264172 }265173 266-267-enum MoqMediaReadOutcome {268- Frame {269- frame: MediaFrame,270- discontinuity: bool,271- },272- Dropped {273- reason: KeyframeRequestReasonV3,274- },275- Malformed(String),276-}277-278-279-enum MoqGroupRecovery {280- /// Sigil deliberately cancels the old GOP when a configured IDR starts a281- /// replacement. This is normal live-edge supersession, not evidence that282- /// another keyframe request is needed.283- ExpectedSupersession,284- RecoverableGap(KeyframeRequestReasonV3),285-}286-287-struct MoqGroupCursor {288- group: GroupConsumer,289- sequence: u64,290- object_count: usize,291- object_bytes: usize,292- group_gap: bool,293- replacement_for_cancelled_group: bool,294-}295-296-/// Own every upstream handle for as long as the Portal frame task is alive.297-/// In particular, dropping `Moq` aborts its actor and dropping the consumers298-/// cancels their subscriptions, so none of these may be scoped only to setup.299-struct MoqMediaLifetime {300- _moq: Moq,301- _session: MoqSession,302- _broadcast: BroadcastConsumer,303-}304-305-struct MoqMediaReceiver {306- _lifetime: Option<MoqMediaLifetime>,307- track: TrackConsumer,308- current_group: Option<MoqGroupCursor>,309- last_group_sequence: Option<u64>,310- last_frame_sequence: Option<u64>,311- waiting_for_keyframe: bool,312- pending_group_recovery: Option<MoqGroupRecovery>,313-}314-315-impl MoqMediaReceiver {316- fn new(317- moq: Moq,318- session: MoqSession,319- broadcast: BroadcastConsumer,320- track: TrackConsumer,321- ) -> Self {322- Self {323- _lifetime: Some(MoqMediaLifetime {324- _moq: moq,325- _session: session,326- _broadcast: broadcast,327- }),328- track,329- current_group: None,330- last_group_sequence: None,331- last_frame_sequence: None,332- waiting_for_keyframe: true,333- pending_group_recovery: None,334- }335- }336-337- 338- fn for_test(track: TrackConsumer) -> Self {339- Self {340- _lifetime: None,341- track,342- current_group: None,343- last_group_sequence: None,344- last_frame_sequence: None,345- waiting_for_keyframe: true,346- pending_group_recovery: None,347- }348- }349-350- async fn next(&mut self) -> Result<Option<MoqMediaReadOutcome>, String> {351- self.next_with_timeouts(352- CLIENT_MOQ_OBJECT_READ_TIMEOUT,353- CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT,354- )355- .await356- }357-358- async fn next_with_timeouts(359- &mut self,360- object_read_timeout: Duration,361- group_recovery_timeout: Duration,362- ) -> Result<Option<MoqMediaReadOutcome>, String> {363- loop {364- if self.current_group.is_none() {365- let replacement_for_cancelled_group = self.pending_group_recovery.is_some();366- let group = if replacement_for_cancelled_group {367- match tokio::time::timeout(368- group_recovery_timeout,369- self.track.next_group(),370- )371- .await372- {373- Ok(group) => group,374- Err(_) => {375- let recovery = self376- .pending_group_recovery377- .take()378- .expect("pending MoQ recovery reason was present");379- return match recovery {380- MoqGroupRecovery::ExpectedSupersession => Err(format!(381- "Timed out after {} ms waiting for the MoQ group that supersedes an expected GOP cancellation",382- group_recovery_timeout.as_millis()383- )),384- MoqGroupRecovery::RecoverableGap(reason) => {385- Ok(Some(MoqMediaReadOutcome::Dropped { reason }))386- }387- };388- }389- }390- } else {391- self.track.next_group().await392- }393- .map_err(|error| format!("Upstream MoQ video track failed: {error}"))?;394- let Some(group) = group else {395- return Ok(None);396- };397- self.pending_group_recovery = None;398- let sequence = group.sequence;399- let group_gap =400- match classify_moq_group_sequence(self.last_group_sequence, sequence) {401- Ok(group_gap) => group_gap,402- Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),403- };404- if group_gap {405- self.waiting_for_keyframe = true;406- }407- self.current_group = Some(MoqGroupCursor {408- group,409- sequence,410- object_count: 0,411- object_bytes: 0,412- group_gap,413- replacement_for_cancelled_group,414- });415- }416-417- let cursor = self418- .current_group419- .as_mut()420- .expect("MoQ group cursor was initialized");421- let object =422- match tokio::time::timeout(object_read_timeout, cursor.group.read_frame()).await {423- Err(_) => {424- let sequence = cursor.sequence;425- self.last_group_sequence = Some(sequence);426- self.current_group = None;427- self.waiting_for_keyframe = true;428- return Ok(Some(MoqMediaReadOutcome::Dropped {429- reason: KeyframeRequestReasonV3::DeliveryTimeout,430- }));431- }432- Ok(Ok(object)) => object,433- Ok(Err(error)) if moq_group_error_is_recoverable(&error) => {434- let sequence = cursor.sequence;435- self.last_group_sequence = Some(sequence);436- self.current_group = None;437- self.waiting_for_keyframe = true;438- let reason = moq_group_error_reason(&error);439- self.pending_group_recovery =440- Some(if matches!(error, moq_net::Error::Cancel) {441- MoqGroupRecovery::ExpectedSupersession442- } else {443- MoqGroupRecovery::RecoverableGap(reason)444- });445- continue;446- }447- Ok(Err(error)) => {448- return Ok(Some(MoqMediaReadOutcome::Malformed(format!(449- "Upstream MoQ group {} failed: {error}",450- cursor.sequence451- ))));452- }453- };454- let Some(object) = object else {455- if cursor.object_count == 0 {456- if cursor.replacement_for_cancelled_group {457- self.last_group_sequence = Some(cursor.sequence);458- self.current_group = None;459- self.waiting_for_keyframe = true;460- return Ok(Some(MoqMediaReadOutcome::Dropped {461- reason: KeyframeRequestReasonV3::TransportGap,462- }));463- }464- return Ok(Some(MoqMediaReadOutcome::Malformed(format!(465- "Upstream MoQ group {} was empty",466- cursor.sequence467- ))));468- }469- self.last_group_sequence = Some(cursor.sequence);470- self.current_group = None;471- continue;472- };473-474- let next_group_bytes = match validate_moq_object_bounds(475- cursor.sequence,476- cursor.object_count,477- cursor.object_bytes,478- object.len(),479- ) {480- Ok(next_group_bytes) => next_group_bytes,481- Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),482- };483- let frame = match decode_media_frame_object(&object) {484- Ok(frame) => frame,485- Err(error) => {486- return Ok(Some(MoqMediaReadOutcome::Malformed(format!(487- "Invalid upstream MoQ media object in group {} object {}: {error}",488- cursor.sequence, cursor.object_count489- ))));490- }491- };492-493- let first_object = cursor.object_count == 0;494- let frame_contiguous = match validate_moq_group_frame(495- cursor.sequence,496- first_object,497- self.last_frame_sequence,498- &frame,499- ) {500- Ok(frame_contiguous) => frame_contiguous,501- Err(_error) if first_object && cursor.replacement_for_cancelled_group => {502- self.last_group_sequence = Some(cursor.sequence);503- self.current_group = None;504- self.waiting_for_keyframe = true;505- return Ok(Some(MoqMediaReadOutcome::Dropped {506- reason: KeyframeRequestReasonV3::TransportGap,507- }));508- }509- Err(_error) if !first_object => {510- self.last_group_sequence = Some(cursor.sequence);511- self.current_group = None;512- self.waiting_for_keyframe = true;513- return Ok(Some(MoqMediaReadOutcome::Dropped {514- reason: KeyframeRequestReasonV3::TransportGap,515- }));516- }517- Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),518- };519- let discontinuity = self.waiting_for_keyframe520- || (first_object && cursor.group_gap)521- || (first_object && !frame_contiguous)522- || frame.header.flags.contains(FrameFlags::DISCONTINUITY);523- cursor.object_count += 1;524- cursor.object_bytes = next_group_bytes;525- self.last_frame_sequence = Some(frame.header.sequence);526- self.waiting_for_keyframe = false;527- return Ok(Some(MoqMediaReadOutcome::Frame {528- frame,529- discontinuity,530- }));531- }532- }533-}534-535-fn classify_moq_group_sequence(previous: Option<u64>, current: u64) -> Result<bool, String> {536- let Some(previous) = previous else {537- return Ok(false);538- };539- if current <= previous {540- return Err(format!(541- "Upstream MoQ group sequence did not increase: previous={previous}, current={current}"542- ));543- }544- Ok(previous.checked_add(1) != Some(current))545-}546-547-fn validate_moq_object_bounds(548- group_sequence: u64,549- object_count: usize,550- object_bytes: usize,551- object_len: usize,552-) -> Result<usize, String> {553- let max_objects = MAX_MEDIA_OBJECT_ID_V3 as usize + 1;554- if object_count >= max_objects {555- return Err(format!(556- "Upstream MoQ group {group_sequence} exceeded {max_objects} media objects"557- ));558- }559- let next_group_bytes = object_bytes560- .checked_add(object_len)561- .ok_or_else(|| "Upstream MoQ group byte count overflowed".to_string())?;562- if next_group_bytes > MAX_MEDIA_GROUP_BYTES_V3 {563- return Err(format!(564- "Upstream MoQ group {group_sequence} exceeded the {MAX_MEDIA_GROUP_BYTES_V3} byte limit"565- ));566- }567- Ok(next_group_bytes)568-}569-570-fn validate_moq_group_frame(571- group_sequence: u64,572- first_object: bool,573- last_frame_sequence: Option<u64>,574- frame: &MediaFrame,575-) -> Result<bool, String> {576- if first_object577- && !(frame.header.codec == MediaCodec::H264578- && frame.header.flags.contains(FrameFlags::KEYFRAME)579- && frame.header.flags.contains(FrameFlags::CODEC_CONFIG))580- {581- return Err(format!(582- "Upstream MoQ group {group_sequence} did not begin with a configured H.264 keyframe"583- ));584- }585- let contiguous = last_frame_sequence586- .is_none_or(|previous| previous.checked_add(1) == Some(frame.header.sequence));587- if !first_object && !contiguous {588- return Err(format!(589- "Upstream MoQ group {group_sequence} contains a non-contiguous access-unit sequence"590- ));591- }592- Ok(contiguous)593-}594-595-fn moq_group_error_is_recoverable(error: &moq_net::Error) -> bool {596- matches!(597- error,598- moq_net::Error::Cancel599- | moq_net::Error::Old600- | moq_net::Error::Timeout601- | moq_net::Error::Dropped602- | moq_net::Error::CacheFull603- | moq_net::Error::Remote(0 | 2 | 3 | 24 | 26)604- )605-}606-607-fn moq_group_error_reason(error: &moq_net::Error) -> KeyframeRequestReasonV3 {608- if matches!(error, moq_net::Error::Timeout | moq_net::Error::Remote(3)) {609- KeyframeRequestReasonV3::DeliveryTimeout610- } else {611- KeyframeRequestReasonV3::TransportGap612- }613-}614-615174 616175 enum MediaObjectReadOutcomeV3 {617176 Object {@@ -1263,45 +822,6 @@ async fn run_media_feedback_session(
1263822 retire_media_feedback_generation(&app, generation).await;1264823 }1265824 1266-async fn open_upstream_moq_media(1267- endpoint: &Endpoint,1268- address: &iroh::EndpointAddr,1269- session_id: u64,1270-) -> Result<(MoqMediaReceiver, iroh::endpoint::Connection), String> {1271- let broadcast_name = media_moq_broadcast_name(session_id)1272- .map_err(|error| format!("Invalid MoQ media session name: {error}"))?;1273- let moq = Moq::new(endpoint.clone());1274- let mut session =1275- tokio::time::timeout(CLIENT_MOQ_CONNECT_TIMEOUT, moq.connect(address.clone()))1276- .await1277- .map_err(|_| "Timed out connecting upstream MoQ media session".to_string())?1278- .map_err(|error| format!("Failed to connect upstream MoQ media session: {error:#}"))?;1279- if session.remote_id() != address.id {1280- session.close(1, b"remote identity mismatch");1281- return Err(format!(1282- "Upstream MoQ connected to unexpected peer {}; expected {}",1283- session.remote_id(),1284- address.id1285- ));1286- }1287- let diagnostics_connection = session.conn().clone();1288- let broadcast = tokio::time::timeout(1289- CLIENT_MOQ_SUBSCRIBE_TIMEOUT,1290- session.subscribe(&broadcast_name),1291- )1292- .await1293- .map_err(|_| format!("Timed out waiting for upstream MoQ broadcast {broadcast_name}"))?1294- .map_err(|error| {1295- format!("Failed to subscribe to upstream MoQ broadcast {broadcast_name}: {error}")1296- })?;1297- let catalog = subscribe_goq_video_track(&broadcast, CLIENT_MOQ_SUBSCRIBE_TIMEOUT).await?;1298- eprintln!("[client] moq catalog: {}", catalog.mode.label());1299- Ok((1300- MoqMediaReceiver::new(moq, session, broadcast, catalog.track),1301- diagnostics_connection,1302- ))1303-}1304-1305825 async fn run_media_control_writer_v3(1306826 mut stream: iroh::endpoint::SendStream,1307827 mut requests: tokio::sync::mpsc::Receiver<(KeyframeRequestReasonV3, Option<u64>)>,@@ -2118,7 +1638,7 @@ pub async fn iroh_client_connect(
21181638 try_queue_media_keyframe_request(21191639 media_control_requests.as_ref(),21201640 reason,2121- receiver.last_frame_sequence,1641+ receiver.last_frame_sequence(),21221642 );21231643 }21241644 MoqMediaReadOutcome::Malformed(error) => {@@ -2632,189 +2152,6 @@ mod tests {
26322152 }26332153 }26342154 2635- 2636- fn upstream_moq_group_ids_detect_only_real_transport_gaps() {2637- assert!(!classify_moq_group_sequence(None, 41).unwrap());2638- assert!(!classify_moq_group_sequence(Some(41), 42).unwrap());2639- assert!(classify_moq_group_sequence(Some(42), 44).unwrap());2640- assert!(classify_moq_group_sequence(Some(44), 44).is_err());2641- assert!(classify_moq_group_sequence(Some(44), 43).is_err());2642- }2643-2644- 2645- fn upstream_moq_group_gap_keyframe_is_the_recovery_barrier() {2646- let keyframe = media_object_frame(80, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));2647- let contiguous = validate_moq_group_frame(9, true, Some(42), &keyframe).unwrap();2648- assert!(!contiguous);2649-2650- // A native group-id gap puts the decoder into resync, but this exact2651- // configured frame 0 exits it immediately. There is deliberately no2652- // keyframe-request action on MoqMediaReadOutcome::Frame: requesting a2653- // replacement here would recreate the grouped-v3 feedback loop.2654- let group_gap = classify_moq_group_sequence(Some(7), 9).unwrap();2655- let discontinuity = group_gap || !contiguous;2656- assert!(discontinuity);2657- }2658-2659- 2660- async fn upstream_moq_idr_abort_to_next_frame_zero_has_no_feedback_loop() {2661- let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();2662- let mut receiver = MoqMediaReceiver::for_test(producer.consume());2663-2664- let mut prior_group = producer.append_group().unwrap();2665- let prior_keyframe =2666- media_object_frame(40, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));2667- prior_group2668- .write_frame(sigil_protocol::encode_media_frame_object(&prior_keyframe).unwrap())2669- .unwrap();2670- assert!(matches!(2671- receiver.next().await.unwrap(),2672- Some(MoqMediaReadOutcome::Frame { .. })2673- ));2674-2675- prior_group.abort(moq_net::Error::Cancel).unwrap();2676- let mut replacement = producer.append_group().unwrap();2677- let replacement_keyframe =2678- media_object_frame(80, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));2679- replacement2680- .write_frame(sigil_protocol::encode_media_frame_object(&replacement_keyframe).unwrap())2681- .unwrap();2682- replacement.finish().unwrap();2683-2684- // `next` consumes the expected Cancel internally and returns the2685- // replacement frame directly. A Dropped outcome here would enqueue a2686- // needless keyframe request and recreate the recovery feedback loop.2687- assert!(matches!(2688- receiver.next().await.unwrap(),2689- Some(MoqMediaReadOutcome::Frame {2690- discontinuity: true,2691- ..2692- })2693- ));2694- }2695-2696- 2697- async fn upstream_moq_expected_cancel_without_replacement_is_terminal_without_request() {2698- let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();2699- let mut receiver = MoqMediaReceiver::for_test(producer.consume());2700- let mut group = producer.append_group().unwrap();2701- let keyframe = media_object_frame(1, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));2702- group2703- .write_frame(sigil_protocol::encode_media_frame_object(&keyframe).unwrap())2704- .unwrap();2705- assert!(matches!(2706- receiver.next().await.unwrap(),2707- Some(MoqMediaReadOutcome::Frame { .. })2708- ));2709- group.abort(moq_net::Error::Cancel).unwrap();2710-2711- let error = receiver2712- .next_with_timeouts(Duration::from_millis(25), Duration::from_millis(25))2713- .await2714- .unwrap_err();2715- assert!(error.contains("expected GOP cancellation"));2716- }2717-2718- 2719- async fn upstream_moq_partial_object_read_has_protocol_bounded_deadline() {2720- let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();2721- let mut receiver = MoqMediaReceiver::for_test(producer.consume());2722- let _stalled_group = producer.append_group().unwrap();2723-2724- assert!(matches!(2725- receiver2726- .next_with_timeouts(Duration::from_millis(25), Duration::from_millis(25))2727- .await2728- .unwrap(),2729- Some(MoqMediaReadOutcome::Dropped {2730- reason: KeyframeRequestReasonV3::DeliveryTimeout2731- })2732- ));2733- assert_eq!(2734- CLIENT_MOQ_OBJECT_READ_TIMEOUT,2735- Duration::from_millis(MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS as u64)2736- );2737- assert!(CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT >= Duration::from_millis(1_000));2738- }2739-2740- 2741- async fn upstream_moq_invalid_cancel_replacement_requests_recovery() {2742- let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();2743- let mut receiver = MoqMediaReceiver::for_test(producer.consume());2744- let mut prior = producer.append_group().unwrap();2745- let keyframe = media_object_frame(10, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));2746- prior2747- .write_frame(sigil_protocol::encode_media_frame_object(&keyframe).unwrap())2748- .unwrap();2749- assert!(matches!(2750- receiver.next().await.unwrap(),2751- Some(MoqMediaReadOutcome::Frame { .. })2752- ));2753- prior.abort(moq_net::Error::Cancel).unwrap();2754-2755- let mut invalid_replacement = producer.append_group().unwrap();2756- let delta = media_object_frame(11, FrameFlags::NONE);2757- invalid_replacement2758- .write_frame(sigil_protocol::encode_media_frame_object(&delta).unwrap())2759- .unwrap();2760- invalid_replacement.finish().unwrap();2761- assert!(matches!(2762- receiver.next().await.unwrap(),2763- Some(MoqMediaReadOutcome::Dropped {2764- reason: KeyframeRequestReasonV3::TransportGap2765- })2766- ));2767- }2768-2769- 2770- fn upstream_moq_requires_configured_keyframe_zero_and_contiguous_deltas() {2771- let delta = media_object_frame(10, FrameFlags::NONE);2772- assert!(validate_moq_group_frame(3, true, None, &delta).is_err());2773-2774- let unconfigured_keyframe = media_object_frame(10, FrameFlags::KEYFRAME);2775- assert!(validate_moq_group_frame(3, true, None, &unconfigured_keyframe).is_err());2776-2777- let configured =2778- media_object_frame(10, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));2779- assert!(validate_moq_group_frame(3, true, None, &configured).is_ok());2780- assert!(validate_moq_group_frame(3, false, Some(8), &delta).is_err());2781- assert!(validate_moq_group_frame(3, false, Some(9), &delta).is_ok());2782- }2783-2784- 2785- fn upstream_moq_bounds_objects_and_group_bytes_before_growth() {2786- assert_eq!(validate_moq_object_bounds(1, 0, 0, 40).unwrap(), 40);2787- assert!(validate_moq_object_bounds(1, MAX_MEDIA_OBJECT_ID_V3 as usize + 1, 0, 1,).is_err());2788- assert!(validate_moq_object_bounds(1, 0, MAX_MEDIA_GROUP_BYTES_V3, 1).is_err());2789- assert!(validate_moq_object_bounds(1, 0, usize::MAX, 1).is_err());2790- }2791-2792- 2793- fn upstream_moq_cancellation_is_resync_but_protocol_errors_are_terminal() {2794- for recoverable in [2795- moq_net::Error::Cancel,2796- moq_net::Error::Old,2797- moq_net::Error::Timeout,2798- moq_net::Error::Dropped,2799- moq_net::Error::CacheFull,2800- moq_net::Error::Remote(3),2801- ] {2802- assert!(moq_group_error_is_recoverable(&recoverable));2803- }2804- assert!(!moq_group_error_is_recoverable(2805- &moq_net::Error::ProtocolViolation2806- ));2807- assert!(!moq_group_error_is_recoverable(&moq_net::Error::WrongSize));2808- assert_eq!(2809- moq_group_error_reason(&moq_net::Error::Timeout),2810- KeyframeRequestReasonV3::DeliveryTimeout2811- );2812- assert_eq!(2813- moq_group_error_reason(&moq_net::Error::Cancel),2814- KeyframeRequestReasonV3::TransportGap2815- );2816- }2817-28182155 28192156 fn media_v3_groups_use_wire_object_identity_and_recover_on_new_group_zero() {28202157 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);@@ -0,0 +1,688 @@
1+use std::sync::atomic::Ordering;2+use std::time::Duration;3+4+use iroh::Endpoint;5+use iroh_moq::{Moq, MoqSession};6+7+use moq_net::Track;8+use moq_net::{BroadcastConsumer, GroupConsumer, TrackConsumer};9+10+use sigil_protocol::MOQ_VIDEO_H264_TRACK;11+use sigil_protocol::{12+ FrameFlags, KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3,13+ MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MAX_MEDIA_OBJECT_ID_V3, MediaCodec, MediaFrame,14+ decode_media_frame_object, media_moq_broadcast_name,15+};16+use tauri::{AppHandle, Manager};17+18+use crate::commands::state::AppState;19+use crate::media::audio_delivery::cancel_audio_generation;20+use crate::media::frame_channel::{take_generation_owned, take_generation_owned_triple};21+use crate::media::moq_catalog::subscribe_goq_video_track;22+23+use crate::media::object_receiver::media_object_frame;24+use crate::media::transport::CLIENT_ENDPOINT_CLOSE_TIMEOUT;25+26+const CLIENT_MOQ_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);27+const CLIENT_MOQ_SUBSCRIBE_TIMEOUT: Duration = Duration::from_secs(10);28+// Match the protocol's maximum single-object delivery horizon. A publisher may29+// never hold a partially delivered object open indefinitely.30+const CLIENT_MOQ_OBJECT_READ_TIMEOUT: Duration =31+ Duration::from_millis(MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS as u64);32+// Sigil's external encoder can take 500 ms to reach its next configured IDR.33+// Allow another 500 ms for a relay path to deliver the superseding group.34+const CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT: Duration = Duration::from_secs(1);35+36+pub(crate) async fn retire_upstream_moq_generation(37+ app: &AppHandle,38+ media_generation: u64,39+ audio_generation: Option<u64>,40+) -> bool {41+ let state = app.state::<AppState>();42+ let endpoint = {43+ // Selection and retirement are one generation-checked transaction.44+ // A stale reader must never close a replacement session whose task was45+ // installed after an explicit disconnect/reconnect.46+ let _connection_serial = state.client_connection_serial.lock().await;47+ let media_connection = {48+ let mut slot = state.media_connection.lock().await;49+ take_generation_owned(&mut slot, media_generation)50+ };51+ let Some(media_connection) = media_connection else {52+ return false;53+ };54+55+ {56+ let mut control = state.media_control.lock().await;57+ let _ = take_generation_owned(&mut control, media_generation);58+ }59+ {60+ let mut delivery = state.frame_delivery.lock().await;61+ let _ = take_generation_owned(&mut delivery, media_generation);62+ }63+ let feedback_connection = {64+ let mut feedback = state.media_feedback.lock().await;65+ take_generation_owned_triple(&mut feedback, media_generation)66+ .map(|(connection, _)| connection)67+ };68+ *state.input_send.lock().await = None;69+70+ let audio_connection = if let Some(audio_generation) = audio_generation {71+ if let Err(error) = cancel_audio_generation(72+ &state.audio_connection_generation,73+ &state.audio_deliveries,74+ audio_generation,75+ ) {76+ eprintln!(77+ "[client] failed to retire audio generation after upstream MoQ ended: {error}"78+ );79+ }80+ let mut slot = state.audio_connection.lock().await;81+ take_generation_owned(&mut slot, audio_generation)82+ } else {83+ None84+ };85+86+ let endpoint = state.client_endpoint.lock().await.take();87+ media_connection.close(0_u32.into(), b"upstream MoQ media ended");88+ if let Some(feedback_connection) = feedback_connection {89+ feedback_connection.close(0_u32.into(), b"upstream MoQ media ended");90+ }91+ if let Some(audio_connection) = audio_connection {92+ audio_connection.close(0_u32.into(), b"upstream MoQ media ended");93+ }94+ state95+ .client_connection_active96+ .store(false, Ordering::SeqCst);97+ endpoint98+ };99+100+ if let Some(endpoint) = endpoint101+ && tokio::time::timeout(CLIENT_ENDPOINT_CLOSE_TIMEOUT, endpoint.close())102+ .await103+ .is_err()104+ {105+ eprintln!("[client] timed out retiring endpoint after upstream MoQ media ended");106+ }107+ true108+}109+110+111+pub(crate) enum MoqMediaReadOutcome {112+ Frame {113+ frame: MediaFrame,114+ discontinuity: bool,115+ },116+ Dropped {117+ reason: KeyframeRequestReasonV3,118+ },119+ Malformed(String),120+}121+122+123+enum MoqGroupRecovery {124+ /// Sigil deliberately cancels the old GOP when a configured IDR starts a125+ /// replacement. This is normal live-edge supersession, not evidence that126+ /// another keyframe request is needed.127+ ExpectedSupersession,128+ RecoverableGap(KeyframeRequestReasonV3),129+}130+131+struct MoqGroupCursor {132+ group: GroupConsumer,133+ sequence: u64,134+ object_count: usize,135+ object_bytes: usize,136+ group_gap: bool,137+ replacement_for_cancelled_group: bool,138+}139+140+/// Own every upstream handle for as long as the Portal frame task is alive.141+/// In particular, dropping `Moq` aborts its actor and dropping the consumers142+/// cancels their subscriptions, so none of these may be scoped only to setup.143+struct MoqMediaLifetime {144+ _moq: Moq,145+ _session: MoqSession,146+ _broadcast: BroadcastConsumer,147+}148+149+pub(crate) struct MoqMediaReceiver {150+ _lifetime: Option<MoqMediaLifetime>,151+ track: TrackConsumer,152+ current_group: Option<MoqGroupCursor>,153+ last_group_sequence: Option<u64>,154+ last_frame_sequence: Option<u64>,155+ waiting_for_keyframe: bool,156+ pending_group_recovery: Option<MoqGroupRecovery>,157+}158+159+impl MoqMediaReceiver {160+ fn new(161+ moq: Moq,162+ session: MoqSession,163+ broadcast: BroadcastConsumer,164+ track: TrackConsumer,165+ ) -> Self {166+ Self {167+ _lifetime: Some(MoqMediaLifetime {168+ _moq: moq,169+ _session: session,170+ _broadcast: broadcast,171+ }),172+ track,173+ current_group: None,174+ last_group_sequence: None,175+ last_frame_sequence: None,176+ waiting_for_keyframe: true,177+ pending_group_recovery: None,178+ }179+ }180+181+ pub(crate) fn last_frame_sequence(&self) -> Option<u64> {182+ self.last_frame_sequence183+ }184+185+ 186+ fn for_test(track: TrackConsumer) -> Self {187+ Self {188+ _lifetime: None,189+ track,190+ current_group: None,191+ last_group_sequence: None,192+ last_frame_sequence: None,193+ waiting_for_keyframe: true,194+ pending_group_recovery: None,195+ }196+ }197+198+ pub(crate) async fn next(&mut self) -> Result<Option<MoqMediaReadOutcome>, String> {199+ self.next_with_timeouts(200+ CLIENT_MOQ_OBJECT_READ_TIMEOUT,201+ CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT,202+ )203+ .await204+ }205+206+ async fn next_with_timeouts(207+ &mut self,208+ object_read_timeout: Duration,209+ group_recovery_timeout: Duration,210+ ) -> Result<Option<MoqMediaReadOutcome>, String> {211+ loop {212+ if self.current_group.is_none() {213+ let replacement_for_cancelled_group = self.pending_group_recovery.is_some();214+ let group = if replacement_for_cancelled_group {215+ match tokio::time::timeout(216+ group_recovery_timeout,217+ self.track.next_group(),218+ )219+ .await220+ {221+ Ok(group) => group,222+ Err(_) => {223+ let recovery = self224+ .pending_group_recovery225+ .take()226+ .expect("pending MoQ recovery reason was present");227+ return match recovery {228+ MoqGroupRecovery::ExpectedSupersession => Err(format!(229+ "Timed out after {} ms waiting for the MoQ group that supersedes an expected GOP cancellation",230+ group_recovery_timeout.as_millis()231+ )),232+ MoqGroupRecovery::RecoverableGap(reason) => {233+ Ok(Some(MoqMediaReadOutcome::Dropped { reason }))234+ }235+ };236+ }237+ }238+ } else {239+ self.track.next_group().await240+ }241+ .map_err(|error| format!("Upstream MoQ video track failed: {error}"))?;242+ let Some(group) = group else {243+ return Ok(None);244+ };245+ self.pending_group_recovery = None;246+ let sequence = group.sequence;247+ let group_gap =248+ match classify_moq_group_sequence(self.last_group_sequence, sequence) {249+ Ok(group_gap) => group_gap,250+ Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),251+ };252+ if group_gap {253+ self.waiting_for_keyframe = true;254+ }255+ self.current_group = Some(MoqGroupCursor {256+ group,257+ sequence,258+ object_count: 0,259+ object_bytes: 0,260+ group_gap,261+ replacement_for_cancelled_group,262+ });263+ }264+265+ let cursor = self266+ .current_group267+ .as_mut()268+ .expect("MoQ group cursor was initialized");269+ let object =270+ match tokio::time::timeout(object_read_timeout, cursor.group.read_frame()).await {271+ Err(_) => {272+ let sequence = cursor.sequence;273+ self.last_group_sequence = Some(sequence);274+ self.current_group = None;275+ self.waiting_for_keyframe = true;276+ return Ok(Some(MoqMediaReadOutcome::Dropped {277+ reason: KeyframeRequestReasonV3::DeliveryTimeout,278+ }));279+ }280+ Ok(Ok(object)) => object,281+ Ok(Err(error)) if moq_group_error_is_recoverable(&error) => {282+ let sequence = cursor.sequence;283+ self.last_group_sequence = Some(sequence);284+ self.current_group = None;285+ self.waiting_for_keyframe = true;286+ let reason = moq_group_error_reason(&error);287+ self.pending_group_recovery =288+ Some(if matches!(error, moq_net::Error::Cancel) {289+ MoqGroupRecovery::ExpectedSupersession290+ } else {291+ MoqGroupRecovery::RecoverableGap(reason)292+ });293+ continue;294+ }295+ Ok(Err(error)) => {296+ return Ok(Some(MoqMediaReadOutcome::Malformed(format!(297+ "Upstream MoQ group {} failed: {error}",298+ cursor.sequence299+ ))));300+ }301+ };302+ let Some(object) = object else {303+ if cursor.object_count == 0 {304+ if cursor.replacement_for_cancelled_group {305+ self.last_group_sequence = Some(cursor.sequence);306+ self.current_group = None;307+ self.waiting_for_keyframe = true;308+ return Ok(Some(MoqMediaReadOutcome::Dropped {309+ reason: KeyframeRequestReasonV3::TransportGap,310+ }));311+ }312+ return Ok(Some(MoqMediaReadOutcome::Malformed(format!(313+ "Upstream MoQ group {} was empty",314+ cursor.sequence315+ ))));316+ }317+ self.last_group_sequence = Some(cursor.sequence);318+ self.current_group = None;319+ continue;320+ };321+322+ let next_group_bytes = match validate_moq_object_bounds(323+ cursor.sequence,324+ cursor.object_count,325+ cursor.object_bytes,326+ object.len(),327+ ) {328+ Ok(next_group_bytes) => next_group_bytes,329+ Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),330+ };331+ let frame = match decode_media_frame_object(&object) {332+ Ok(frame) => frame,333+ Err(error) => {334+ return Ok(Some(MoqMediaReadOutcome::Malformed(format!(335+ "Invalid upstream MoQ media object in group {} object {}: {error}",336+ cursor.sequence, cursor.object_count337+ ))));338+ }339+ };340+341+ let first_object = cursor.object_count == 0;342+ let frame_contiguous = match validate_moq_group_frame(343+ cursor.sequence,344+ first_object,345+ self.last_frame_sequence,346+ &frame,347+ ) {348+ Ok(frame_contiguous) => frame_contiguous,349+ Err(_error) if first_object && cursor.replacement_for_cancelled_group => {350+ self.last_group_sequence = Some(cursor.sequence);351+ self.current_group = None;352+ self.waiting_for_keyframe = true;353+ return Ok(Some(MoqMediaReadOutcome::Dropped {354+ reason: KeyframeRequestReasonV3::TransportGap,355+ }));356+ }357+ Err(_error) if !first_object => {358+ self.last_group_sequence = Some(cursor.sequence);359+ self.current_group = None;360+ self.waiting_for_keyframe = true;361+ return Ok(Some(MoqMediaReadOutcome::Dropped {362+ reason: KeyframeRequestReasonV3::TransportGap,363+ }));364+ }365+ Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),366+ };367+ let discontinuity = self.waiting_for_keyframe368+ || (first_object && cursor.group_gap)369+ || (first_object && !frame_contiguous)370+ || frame.header.flags.contains(FrameFlags::DISCONTINUITY);371+ cursor.object_count += 1;372+ cursor.object_bytes = next_group_bytes;373+ self.last_frame_sequence = Some(frame.header.sequence);374+ self.waiting_for_keyframe = false;375+ return Ok(Some(MoqMediaReadOutcome::Frame {376+ frame,377+ discontinuity,378+ }));379+ }380+ }381+}382+383+fn classify_moq_group_sequence(previous: Option<u64>, current: u64) -> Result<bool, String> {384+ let Some(previous) = previous else {385+ return Ok(false);386+ };387+ if current <= previous {388+ return Err(format!(389+ "Upstream MoQ group sequence did not increase: previous={previous}, current={current}"390+ ));391+ }392+ Ok(previous.checked_add(1) != Some(current))393+}394+395+fn validate_moq_object_bounds(396+ group_sequence: u64,397+ object_count: usize,398+ object_bytes: usize,399+ object_len: usize,400+) -> Result<usize, String> {401+ let max_objects = MAX_MEDIA_OBJECT_ID_V3 as usize + 1;402+ if object_count >= max_objects {403+ return Err(format!(404+ "Upstream MoQ group {group_sequence} exceeded {max_objects} media objects"405+ ));406+ }407+ let next_group_bytes = object_bytes408+ .checked_add(object_len)409+ .ok_or_else(|| "Upstream MoQ group byte count overflowed".to_string())?;410+ if next_group_bytes > MAX_MEDIA_GROUP_BYTES_V3 {411+ return Err(format!(412+ "Upstream MoQ group {group_sequence} exceeded the {MAX_MEDIA_GROUP_BYTES_V3} byte limit"413+ ));414+ }415+ Ok(next_group_bytes)416+}417+418+fn validate_moq_group_frame(419+ group_sequence: u64,420+ first_object: bool,421+ last_frame_sequence: Option<u64>,422+ frame: &MediaFrame,423+) -> Result<bool, String> {424+ if first_object425+ && !(frame.header.codec == MediaCodec::H264426+ && frame.header.flags.contains(FrameFlags::KEYFRAME)427+ && frame.header.flags.contains(FrameFlags::CODEC_CONFIG))428+ {429+ return Err(format!(430+ "Upstream MoQ group {group_sequence} did not begin with a configured H.264 keyframe"431+ ));432+ }433+ let contiguous = last_frame_sequence434+ .is_none_or(|previous| previous.checked_add(1) == Some(frame.header.sequence));435+ if !first_object && !contiguous {436+ return Err(format!(437+ "Upstream MoQ group {group_sequence} contains a non-contiguous access-unit sequence"438+ ));439+ }440+ Ok(contiguous)441+}442+443+fn moq_group_error_is_recoverable(error: &moq_net::Error) -> bool {444+ matches!(445+ error,446+ moq_net::Error::Cancel447+ | moq_net::Error::Old448+ | moq_net::Error::Timeout449+ | moq_net::Error::Dropped450+ | moq_net::Error::CacheFull451+ | moq_net::Error::Remote(0 | 2 | 3 | 24 | 26)452+ )453+}454+455+fn moq_group_error_reason(error: &moq_net::Error) -> KeyframeRequestReasonV3 {456+ if matches!(error, moq_net::Error::Timeout | moq_net::Error::Remote(3)) {457+ KeyframeRequestReasonV3::DeliveryTimeout458+ } else {459+ KeyframeRequestReasonV3::TransportGap460+ }461+}462+463+pub(crate) async fn open_upstream_moq_media(464+ endpoint: &Endpoint,465+ address: &iroh::EndpointAddr,466+ session_id: u64,467+) -> Result<(MoqMediaReceiver, iroh::endpoint::Connection), String> {468+ let broadcast_name = media_moq_broadcast_name(session_id)469+ .map_err(|error| format!("Invalid MoQ media session name: {error}"))?;470+ let moq = Moq::new(endpoint.clone());471+ let mut session =472+ tokio::time::timeout(CLIENT_MOQ_CONNECT_TIMEOUT, moq.connect(address.clone()))473+ .await474+ .map_err(|_| "Timed out connecting upstream MoQ media session".to_string())?475+ .map_err(|error| format!("Failed to connect upstream MoQ media session: {error:#}"))?;476+ if session.remote_id() != address.id {477+ session.close(1, b"remote identity mismatch");478+ return Err(format!(479+ "Upstream MoQ connected to unexpected peer {}; expected {}",480+ session.remote_id(),481+ address.id482+ ));483+ }484+ let diagnostics_connection = session.conn().clone();485+ let broadcast = tokio::time::timeout(486+ CLIENT_MOQ_SUBSCRIBE_TIMEOUT,487+ session.subscribe(&broadcast_name),488+ )489+ .await490+ .map_err(|_| format!("Timed out waiting for upstream MoQ broadcast {broadcast_name}"))?491+ .map_err(|error| {492+ format!("Failed to subscribe to upstream MoQ broadcast {broadcast_name}: {error}")493+ })?;494+ let catalog = subscribe_goq_video_track(&broadcast, CLIENT_MOQ_SUBSCRIBE_TIMEOUT).await?;495+ eprintln!("[client] moq catalog: {}", catalog.mode.label());496+ Ok((497+ MoqMediaReceiver::new(moq, session, broadcast, catalog.track),498+ diagnostics_connection,499+ ))500+}501+502+503+mod tests {504+ use super::*;505+506+ 507+ fn upstream_moq_group_ids_detect_only_real_transport_gaps() {508+ assert!(!classify_moq_group_sequence(None, 41).unwrap());509+ assert!(!classify_moq_group_sequence(Some(41), 42).unwrap());510+ assert!(classify_moq_group_sequence(Some(42), 44).unwrap());511+ assert!(classify_moq_group_sequence(Some(44), 44).is_err());512+ assert!(classify_moq_group_sequence(Some(44), 43).is_err());513+ }514+515+ 516+ fn upstream_moq_group_gap_keyframe_is_the_recovery_barrier() {517+ let keyframe = media_object_frame(80, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));518+ let contiguous = validate_moq_group_frame(9, true, Some(42), &keyframe).unwrap();519+ assert!(!contiguous);520+521+ // A native group-id gap puts the decoder into resync, but this exact522+ // configured frame 0 exits it immediately. There is deliberately no523+ // keyframe-request action on MoqMediaReadOutcome::Frame: requesting a524+ // replacement here would recreate the grouped-v3 feedback loop.525+ let group_gap = classify_moq_group_sequence(Some(7), 9).unwrap();526+ let discontinuity = group_gap || !contiguous;527+ assert!(discontinuity);528+ }529+530+ 531+ async fn upstream_moq_idr_abort_to_next_frame_zero_has_no_feedback_loop() {532+ let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();533+ let mut receiver = MoqMediaReceiver::for_test(producer.consume());534+535+ let mut prior_group = producer.append_group().unwrap();536+ let prior_keyframe =537+ media_object_frame(40, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));538+ prior_group539+ .write_frame(sigil_protocol::encode_media_frame_object(&prior_keyframe).unwrap())540+ .unwrap();541+ assert!(matches!(542+ receiver.next().await.unwrap(),543+ Some(MoqMediaReadOutcome::Frame { .. })544+ ));545+546+ prior_group.abort(moq_net::Error::Cancel).unwrap();547+ let mut replacement = producer.append_group().unwrap();548+ let replacement_keyframe =549+ media_object_frame(80, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));550+ replacement551+ .write_frame(sigil_protocol::encode_media_frame_object(&replacement_keyframe).unwrap())552+ .unwrap();553+ replacement.finish().unwrap();554+555+ // `next` consumes the expected Cancel internally and returns the556+ // replacement frame directly. A Dropped outcome here would enqueue a557+ // needless keyframe request and recreate the recovery feedback loop.558+ assert!(matches!(559+ receiver.next().await.unwrap(),560+ Some(MoqMediaReadOutcome::Frame {561+ discontinuity: true,562+ ..563+ })564+ ));565+ }566+567+ 568+ async fn upstream_moq_expected_cancel_without_replacement_is_terminal_without_request() {569+ let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();570+ let mut receiver = MoqMediaReceiver::for_test(producer.consume());571+ let mut group = producer.append_group().unwrap();572+ let keyframe = media_object_frame(1, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));573+ group574+ .write_frame(sigil_protocol::encode_media_frame_object(&keyframe).unwrap())575+ .unwrap();576+ assert!(matches!(577+ receiver.next().await.unwrap(),578+ Some(MoqMediaReadOutcome::Frame { .. })579+ ));580+ group.abort(moq_net::Error::Cancel).unwrap();581+582+ let error = receiver583+ .next_with_timeouts(Duration::from_millis(25), Duration::from_millis(25))584+ .await585+ .unwrap_err();586+ assert!(error.contains("expected GOP cancellation"));587+ }588+589+ 590+ async fn upstream_moq_partial_object_read_has_protocol_bounded_deadline() {591+ let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();592+ let mut receiver = MoqMediaReceiver::for_test(producer.consume());593+ let _stalled_group = producer.append_group().unwrap();594+595+ assert!(matches!(596+ receiver597+ .next_with_timeouts(Duration::from_millis(25), Duration::from_millis(25))598+ .await599+ .unwrap(),600+ Some(MoqMediaReadOutcome::Dropped {601+ reason: KeyframeRequestReasonV3::DeliveryTimeout602+ })603+ ));604+ assert_eq!(605+ CLIENT_MOQ_OBJECT_READ_TIMEOUT,606+ Duration::from_millis(MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS as u64)607+ );608+ assert!(CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT >= Duration::from_millis(1_000));609+ }610+611+ 612+ async fn upstream_moq_invalid_cancel_replacement_requests_recovery() {613+ let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();614+ let mut receiver = MoqMediaReceiver::for_test(producer.consume());615+ let mut prior = producer.append_group().unwrap();616+ let keyframe = media_object_frame(10, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));617+ prior618+ .write_frame(sigil_protocol::encode_media_frame_object(&keyframe).unwrap())619+ .unwrap();620+ assert!(matches!(621+ receiver.next().await.unwrap(),622+ Some(MoqMediaReadOutcome::Frame { .. })623+ ));624+ prior.abort(moq_net::Error::Cancel).unwrap();625+626+ let mut invalid_replacement = producer.append_group().unwrap();627+ let delta = media_object_frame(11, FrameFlags::NONE);628+ invalid_replacement629+ .write_frame(sigil_protocol::encode_media_frame_object(&delta).unwrap())630+ .unwrap();631+ invalid_replacement.finish().unwrap();632+ assert!(matches!(633+ receiver.next().await.unwrap(),634+ Some(MoqMediaReadOutcome::Dropped {635+ reason: KeyframeRequestReasonV3::TransportGap636+ })637+ ));638+ }639+640+ 641+ fn upstream_moq_requires_configured_keyframe_zero_and_contiguous_deltas() {642+ let delta = media_object_frame(10, FrameFlags::NONE);643+ assert!(validate_moq_group_frame(3, true, None, &delta).is_err());644+645+ let unconfigured_keyframe = media_object_frame(10, FrameFlags::KEYFRAME);646+ assert!(validate_moq_group_frame(3, true, None, &unconfigured_keyframe).is_err());647+648+ let configured =649+ media_object_frame(10, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));650+ assert!(validate_moq_group_frame(3, true, None, &configured).is_ok());651+ assert!(validate_moq_group_frame(3, false, Some(8), &delta).is_err());652+ assert!(validate_moq_group_frame(3, false, Some(9), &delta).is_ok());653+ }654+655+ 656+ fn upstream_moq_bounds_objects_and_group_bytes_before_growth() {657+ assert_eq!(validate_moq_object_bounds(1, 0, 0, 40).unwrap(), 40);658+ assert!(validate_moq_object_bounds(1, MAX_MEDIA_OBJECT_ID_V3 as usize + 1, 0, 1,).is_err());659+ assert!(validate_moq_object_bounds(1, 0, MAX_MEDIA_GROUP_BYTES_V3, 1).is_err());660+ assert!(validate_moq_object_bounds(1, 0, usize::MAX, 1).is_err());661+ }662+663+ 664+ fn upstream_moq_cancellation_is_resync_but_protocol_errors_are_terminal() {665+ for recoverable in [666+ moq_net::Error::Cancel,667+ moq_net::Error::Old,668+ moq_net::Error::Timeout,669+ moq_net::Error::Dropped,670+ moq_net::Error::CacheFull,671+ moq_net::Error::Remote(3),672+ ] {673+ assert!(moq_group_error_is_recoverable(&recoverable));674+ }675+ assert!(!moq_group_error_is_recoverable(676+ &moq_net::Error::ProtocolViolation677+ ));678+ assert!(!moq_group_error_is_recoverable(&moq_net::Error::WrongSize));679+ assert_eq!(680+ moq_group_error_reason(&moq_net::Error::Timeout),681+ KeyframeRequestReasonV3::DeliveryTimeout682+ );683+ assert_eq!(684+ moq_group_error_reason(&moq_net::Error::Cancel),685+ KeyframeRequestReasonV3::TransportGap686+ );687+ }688+}@@ -16,6 +16,8 @@ pub const MEDIA_TRANSPORT_NAMES: [&str; 5] = [
1616 "grouped-v3",1717 ];1818 19+pub(crate) const CLIENT_ENDPOINT_CLOSE_TIMEOUT: Duration = Duration::from_secs(2);20+1921 pub(crate) struct NegotiatedV1Stream {2022 pub(crate) session_id: u64,2123 pub(crate) capabilities: Vec<Capability>,