1use std::sync::Arc;2use std::sync::atomic::{AtomicU64, Ordering};3use std::time::Duration;45use iroh::Endpoint;6use iroh_moq::{Moq, MoqSession};78use moq_net::Track;9use moq_net::{BroadcastConsumer, GroupConsumer, TrackConsumer};1011use sigil_protocol::MOQ_VIDEO_H264_TRACK;12use sigil_protocol::{13 AudioFlags, AudioPacket, AudioPacketHeader, AuthenticatedMediaObject, FrameFlags,14 KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,15 MAX_MEDIA_OBJECT_ID_V3, MediaCodec, MediaFrame, MediaObjectCoordinates, MediaTrack,16 SignedMediaGenerationCertificate, SignedSubscriptionCapability, SubscriptionTracks,17 decode_media_frame_object, media_moq_broadcast_name,18};19use tauri::{AppHandle, Manager};2021use crate::commands::state::AppState;22use crate::media::audio_delivery::cancel_audio_generation;23use crate::media::frame_channel::{take_generation_owned, take_generation_owned_triple};24use crate::media::moq_catalog::{25 subscribe_authenticated_goq_video_track, subscribe_goq_video_track,26};2728use crate::media::object_receiver::media_object_frame;29use crate::media::transport::CLIENT_ENDPOINT_CLOSE_TIMEOUT;3031const CLIENT_MOQ_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);32const CLIENT_MOQ_SUBSCRIBE_TIMEOUT: Duration = Duration::from_secs(10);33// Match the protocol's maximum single-object delivery horizon. A publisher may34// never hold a partially delivered object open indefinitely.35const CLIENT_MOQ_OBJECT_READ_TIMEOUT: Duration =36 Duration::from_millis(MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS as u64);37// Sigil's external encoder can take 500 ms to reach its next configured IDR.38// Allow another 500 ms for a relay path to deliver the superseding group.39const CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT: Duration = Duration::from_secs(1);4041pub(crate) async fn retire_upstream_moq_generation(42 app: &AppHandle,43 media_generation: u64,44 audio_generation: Option<u64>,45) -> bool {46 let state = app.state::<AppState>();47 let endpoint = {48 // Selection and retirement are one generation-checked transaction.49 // A stale reader must never close a replacement session whose task was50 // installed after an explicit disconnect/reconnect.51 let _connection_serial = state.client_connection_serial.lock().await;52 let media_connection = {53 let mut slot = state.media_connection.lock().await;54 take_generation_owned(&mut slot, media_generation)55 };56 let Some(media_connection) = media_connection else {57 return false;58 };5960 {61 let mut control = state.media_control.lock().await;62 let _ = take_generation_owned(&mut control, media_generation);63 }64 {65 let mut delivery = state.frame_delivery.lock().await;66 let _ = take_generation_owned(&mut delivery, media_generation);67 }68 let feedback_connection = {69 let mut feedback = state.media_feedback.lock().await;70 take_generation_owned_triple(&mut feedback, media_generation)71 .map(|(connection, _)| connection)72 };73 *state.input_send.lock().await = None;7475 let audio_connection = if let Some(audio_generation) = audio_generation {76 if let Err(error) = cancel_audio_generation(77 &state.audio_connection_generation,78 &state.audio_deliveries,79 audio_generation,80 ) {81 eprintln!(82 "[client] failed to retire audio generation after upstream MoQ ended: {error}"83 );84 }85 let mut slot = state.audio_connection.lock().await;86 take_generation_owned(&mut slot, audio_generation)87 } else {88 None89 };9091 let endpoint = state.client_endpoint.lock().await.take();92 media_connection.close(0_u32.into(), b"upstream MoQ media ended");93 if let Some(feedback_connection) = feedback_connection {94 feedback_connection.close(0_u32.into(), b"upstream MoQ media ended");95 }96 if let Some(audio_connection) = audio_connection {97 audio_connection.close(0_u32.into(), b"upstream MoQ media ended");98 }99 state100 .client_connection_active101 .store(false, Ordering::SeqCst);102 endpoint103 };104105 if let Some(endpoint) = endpoint106 && tokio::time::timeout(CLIENT_ENDPOINT_CLOSE_TIMEOUT, endpoint.close())107 .await108 .is_err()109 {110 eprintln!("[client] timed out retiring endpoint after upstream MoQ media ended");111 }112 true113}114115116pub(crate) enum MoqMediaReadOutcome {117 Frame {118 frame: MediaFrame,119 discontinuity: bool,120 },121 Dropped {122 reason: KeyframeRequestReasonV3,123 },124 Malformed(String),125}126127128enum MoqGroupRecovery {129 /// Sigil deliberately cancels the old GOP when a configured IDR starts a130 /// replacement. This is normal live-edge supersession, not evidence that131 /// another keyframe request is needed.132 ExpectedSupersession,133 RecoverableGap(KeyframeRequestReasonV3),134}135136struct MoqGroupCursor {137 group: GroupConsumer,138 sequence: u64,139 object_count: usize,140 object_bytes: usize,141 group_gap: bool,142 replacement_for_cancelled_group: bool,143}144145/// Own every upstream handle for as long as the Portal frame task is alive.146/// In particular, dropping `Moq` aborts its actor and dropping the consumers147/// cancels their subscriptions, so none of these may be scoped only to setup.148struct MoqMediaLifetime {149 _moq: Moq,150 _session: MoqSession,151 _broadcast: BroadcastConsumer,152}153154pub(crate) struct MoqMediaReceiver {155 _lifetime: Option<MoqMediaLifetime>,156 track: TrackConsumer,157 current_group: Option<MoqGroupCursor>,158 last_group_sequence: Option<u64>,159 last_frame_sequence: Option<u64>,160 waiting_for_keyframe: bool,161 pending_group_recovery: Option<MoqGroupRecovery>,162 certificate: Option<SignedMediaGenerationCertificate>,163 verification_failures: Option<Arc<AtomicU64>>,164}165166struct MoqAudioGroupCursor {167 group: GroupConsumer,168 sequence: u64,169 object_count: usize,170 object_bytes: usize,171}172173pub(crate) struct MoqAudioReceiver {174 track: TrackConsumer,175 current_group: Option<MoqAudioGroupCursor>,176 last_group_sequence: Option<u64>,177 last_packet_sequence: Option<u64>,178 discontinuity_pending: bool,179 certificate: SignedMediaGenerationCertificate,180 verification_failures: Arc<AtomicU64>,181}182183impl MoqAudioReceiver {184 fn new(185 track: TrackConsumer,186 certificate: SignedMediaGenerationCertificate,187 verification_failures: Arc<AtomicU64>,188 ) -> Self {189 Self {190 track,191 current_group: None,192 last_group_sequence: None,193 last_packet_sequence: None,194 discontinuity_pending: true,195 certificate,196 verification_failures,197 }198 }199200 pub(crate) async fn next(&mut self) -> Result<Option<AudioPacket>, String> {201 loop {202 if self.current_group.is_none() {203 let Some(group) = self204 .track205 .next_group()206 .await207 .map_err(|error| format!("Upstream MoQ audio track failed: {error}"))?208 else {209 return Ok(None);210 };211 if self212 .last_group_sequence213 .is_some_and(|previous| previous.checked_add(1) != Some(group.sequence))214 {215 self.discontinuity_pending = true;216 }217 if self218 .last_group_sequence219 .is_some_and(|previous| group.sequence <= previous)220 {221 return Err("Upstream MoQ audio group sequence did not increase".to_string());222 }223 self.current_group = Some(MoqAudioGroupCursor {224 sequence: group.sequence,225 group,226 object_count: 0,227 object_bytes: 0,228 });229 }230231 let cursor = self232 .current_group233 .as_mut()234 .expect("audio group initialized");235 let object = match tokio::time::timeout(236 CLIENT_MOQ_OBJECT_READ_TIMEOUT,237 cursor.group.read_frame(),238 )239 .await240 {241 Err(_) => {242 self.last_group_sequence = Some(cursor.sequence);243 self.current_group = None;244 self.discontinuity_pending = true;245 continue;246 }247 Ok(Err(error)) if moq_group_error_is_recoverable(&error) => {248 self.last_group_sequence = Some(cursor.sequence);249 self.current_group = None;250 self.discontinuity_pending = true;251 continue;252 }253 Ok(Err(error)) => {254 return Err(format!(255 "Upstream MoQ audio group {} failed: {error}",256 cursor.sequence257 ));258 }259 Ok(Ok(None)) => {260 self.last_group_sequence = Some(cursor.sequence);261 self.current_group = None;262 continue;263 }264 Ok(Ok(Some(object))) => object,265 };266 if cursor.object_count >= 5 {267 return Err("Upstream MoQ audio group exceeded five packets".to_string());268 }269 cursor.object_bytes = cursor270 .object_bytes271 .checked_add(object.len())272 .filter(|bytes| {273 *bytes274 <= 5 * (sigil_protocol::AUDIO_HEADER_LEN275 + sigil_protocol::MAX_AUDIO_PAYLOAD_LEN276 + 256)277 })278 .ok_or_else(|| "Upstream MoQ audio group exceeded its byte bound".to_string())?;279 let coordinates = AuthenticatedMediaObject::coordinates(&object).map_err(|error| {280 self.verification_failures.fetch_add(1, Ordering::Relaxed);281 format!("Invalid authenticated MoQ audio header: {error}")282 })?;283 let expected = MediaObjectCoordinates {284 generation_id: self.certificate.claims.generation_id,285 track: MediaTrack::AudioOpus,286 group_id: cursor.sequence,287 object_id: u32::try_from(cursor.object_count)288 .map_err(|_| "MoQ audio object index overflowed".to_string())?,289 flags: coordinates.flags,290 };291 let payload = AuthenticatedMediaObject::verify(&object, &self.certificate, expected)292 .map_err(|error| {293 self.verification_failures.fetch_add(1, Ordering::Relaxed);294 format!("Authenticated upstream MoQ audio verification failed: {error}")295 })?;296 let mut packet = AudioPacket::decode_datagram(payload)297 .map_err(|error| format!("Invalid upstream MoQ Opus packet: {error}"))?;298 if coordinates.flags != u16::from(packet.header.flags.bits()) {299 self.verification_failures.fetch_add(1, Ordering::Relaxed);300 return Err("Authenticated MoQ audio flags do not match the packet".to_string());301 }302 if self303 .last_packet_sequence304 .is_some_and(|previous| previous.checked_add(1) != Some(packet.header.sequence))305 {306 self.discontinuity_pending = true;307 }308 if self.discontinuity_pending309 && !packet.header.flags.contains(AudioFlags::DISCONTINUITY)310 {311 packet.header = AudioPacketHeader::opus(312 packet.payload.len(),313 packet.header.sequence,314 packet.header.capture_timestamp_us,315 packet.header.pts_us,316 AudioFlags::DISCONTINUITY,317 )318 .map_err(|error| error.to_string())?;319 }320 cursor.object_count += 1;321 self.last_packet_sequence = Some(packet.header.sequence);322 self.discontinuity_pending = false;323 return Ok(Some(packet));324 }325 }326}327328impl MoqMediaReceiver {329 fn new(330 moq: Moq,331 session: MoqSession,332 broadcast: BroadcastConsumer,333 track: TrackConsumer,334 certificate: Option<SignedMediaGenerationCertificate>,335 verification_failures: Option<Arc<AtomicU64>>,336 ) -> Self {337 Self {338 _lifetime: Some(MoqMediaLifetime {339 _moq: moq,340 _session: session,341 _broadcast: broadcast,342 }),343 track,344 current_group: None,345 last_group_sequence: None,346 last_frame_sequence: None,347 waiting_for_keyframe: true,348 pending_group_recovery: None,349 certificate,350 verification_failures,351 }352 }353354 pub(crate) fn last_frame_sequence(&self) -> Option<u64> {355 self.last_frame_sequence356 }357358 359 fn for_test(track: TrackConsumer) -> Self {360 Self {361 _lifetime: None,362 track,363 current_group: None,364 last_group_sequence: None,365 last_frame_sequence: None,366 waiting_for_keyframe: true,367 pending_group_recovery: None,368 certificate: None,369 verification_failures: None,370 }371 }372373 374 fn authenticated_for_test(375 track: TrackConsumer,376 certificate: SignedMediaGenerationCertificate,377 verification_failures: Arc<AtomicU64>,378 ) -> Self {379 Self {380 _lifetime: None,381 track,382 current_group: None,383 last_group_sequence: None,384 last_frame_sequence: None,385 waiting_for_keyframe: true,386 pending_group_recovery: None,387 certificate: Some(certificate),388 verification_failures: Some(verification_failures),389 }390 }391392 pub(crate) async fn next(&mut self) -> Result<Option<MoqMediaReadOutcome>, String> {393 self.next_with_timeouts(394 CLIENT_MOQ_OBJECT_READ_TIMEOUT,395 CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT,396 )397 .await398 }399400 async fn next_with_timeouts(401 &mut self,402 object_read_timeout: Duration,403 group_recovery_timeout: Duration,404 ) -> Result<Option<MoqMediaReadOutcome>, String> {405 loop {406 if self.current_group.is_none() {407 let replacement_for_cancelled_group = self.pending_group_recovery.is_some();408 let group = if replacement_for_cancelled_group {409 match tokio::time::timeout(410 group_recovery_timeout,411 self.track.next_group(),412 )413 .await414 {415 Ok(group) => group,416 Err(_) => {417 let recovery = self418 .pending_group_recovery419 .take()420 .expect("pending MoQ recovery reason was present");421 return match recovery {422 MoqGroupRecovery::ExpectedSupersession => Err(format!(423 "Timed out after {} ms waiting for the MoQ group that supersedes an expected GOP cancellation",424 group_recovery_timeout.as_millis()425 )),426 MoqGroupRecovery::RecoverableGap(reason) => {427 Ok(Some(MoqMediaReadOutcome::Dropped { reason }))428 }429 };430 }431 }432 } else {433 self.track.next_group().await434 }435 .map_err(|error| format!("Upstream MoQ video track failed: {error}"))?;436 let Some(group) = group else {437 return Ok(None);438 };439 self.pending_group_recovery = None;440 let sequence = group.sequence;441 let group_gap =442 match classify_moq_group_sequence(self.last_group_sequence, sequence) {443 Ok(group_gap) => group_gap,444 Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),445 };446 if group_gap {447 self.waiting_for_keyframe = true;448 }449 self.current_group = Some(MoqGroupCursor {450 group,451 sequence,452 object_count: 0,453 object_bytes: 0,454 group_gap,455 replacement_for_cancelled_group,456 });457 }458459 let cursor = self460 .current_group461 .as_mut()462 .expect("MoQ group cursor was initialized");463 let object =464 match tokio::time::timeout(object_read_timeout, cursor.group.read_frame()).await {465 Err(_) => {466 let sequence = cursor.sequence;467 self.last_group_sequence = Some(sequence);468 self.current_group = None;469 self.waiting_for_keyframe = true;470 return Ok(Some(MoqMediaReadOutcome::Dropped {471 reason: KeyframeRequestReasonV3::DeliveryTimeout,472 }));473 }474 Ok(Ok(object)) => object,475 Ok(Err(error)) if moq_group_error_is_recoverable(&error) => {476 let sequence = cursor.sequence;477 self.last_group_sequence = Some(sequence);478 self.current_group = None;479 self.waiting_for_keyframe = true;480 let reason = moq_group_error_reason(&error);481 self.pending_group_recovery =482 Some(if matches!(error, moq_net::Error::Cancel) {483 MoqGroupRecovery::ExpectedSupersession484 } else {485 MoqGroupRecovery::RecoverableGap(reason)486 });487 continue;488 }489 Ok(Err(error)) => {490 return Ok(Some(MoqMediaReadOutcome::Malformed(format!(491 "Upstream MoQ group {} failed: {error}",492 cursor.sequence493 ))));494 }495 };496 let Some(object) = object else {497 if cursor.object_count == 0 {498 if cursor.replacement_for_cancelled_group {499 self.last_group_sequence = Some(cursor.sequence);500 self.current_group = None;501 self.waiting_for_keyframe = true;502 return Ok(Some(MoqMediaReadOutcome::Dropped {503 reason: KeyframeRequestReasonV3::TransportGap,504 }));505 }506 return Ok(Some(MoqMediaReadOutcome::Malformed(format!(507 "Upstream MoQ group {} was empty",508 cursor.sequence509 ))));510 }511 self.last_group_sequence = Some(cursor.sequence);512 self.current_group = None;513 continue;514 };515516 let next_group_bytes = match validate_moq_object_bounds(517 cursor.sequence,518 cursor.object_count,519 cursor.object_bytes,520 object.len(),521 ) {522 Ok(next_group_bytes) => next_group_bytes,523 Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),524 };525 let authenticated_coordinates = if self.certificate.is_some() {526 match AuthenticatedMediaObject::coordinates(&object) {527 Ok(coordinates) => Some(coordinates),528 Err(error) => {529 record_verification_failure(&self.verification_failures);530 return Ok(Some(MoqMediaReadOutcome::Malformed(format!(531 "Invalid authenticated MoQ media header in group {} object {}: {error}",532 cursor.sequence, cursor.object_count533 ))));534 }535 }536 } else {537 None538 };539 let payload = if let (Some(certificate), Some(coordinates)) =540 (&self.certificate, authenticated_coordinates)541 {542 let expected_object_id = u32::try_from(cursor.object_count).map_err(|_| {543 "Authenticated MoQ object index exceeded the protocol range".to_string()544 })?;545 let expected = MediaObjectCoordinates {546 generation_id: certificate.claims.generation_id,547 track: MediaTrack::VideoH264,548 group_id: cursor.sequence,549 object_id: expected_object_id,550 flags: coordinates.flags,551 };552 match AuthenticatedMediaObject::verify(&object, certificate, expected) {553 Ok(payload) => payload,554 Err(error) => {555 record_verification_failure(&self.verification_failures);556 return Ok(Some(MoqMediaReadOutcome::Malformed(format!(557 "Authenticated upstream MoQ media verification failed in group {} object {}: {error}",558 cursor.sequence, cursor.object_count559 ))));560 }561 }562 } else {563 object.as_ref()564 };565 let frame = match decode_media_frame_object(payload) {566 Ok(frame) => frame,567 Err(error) => {568 return Ok(Some(MoqMediaReadOutcome::Malformed(format!(569 "Invalid upstream MoQ media object in group {} object {}: {error}",570 cursor.sequence, cursor.object_count571 ))));572 }573 };574 if authenticated_coordinates.is_some_and(|coordinates| {575 coordinates.flags != u16::from(frame.header.flags.bits())576 }) {577 record_verification_failure(&self.verification_failures);578 return Ok(Some(MoqMediaReadOutcome::Malformed(format!(579 "Authenticated upstream MoQ flags do not match the media payload in group {} object {}",580 cursor.sequence, cursor.object_count581 ))));582 }583584 let first_object = cursor.object_count == 0;585 let frame_contiguous = match validate_moq_group_frame(586 cursor.sequence,587 first_object,588 self.last_frame_sequence,589 &frame,590 ) {591 Ok(frame_contiguous) => frame_contiguous,592 Err(_error) if first_object && cursor.replacement_for_cancelled_group => {593 self.last_group_sequence = Some(cursor.sequence);594 self.current_group = None;595 self.waiting_for_keyframe = true;596 return Ok(Some(MoqMediaReadOutcome::Dropped {597 reason: KeyframeRequestReasonV3::TransportGap,598 }));599 }600 Err(_error) if !first_object => {601 self.last_group_sequence = Some(cursor.sequence);602 self.current_group = None;603 self.waiting_for_keyframe = true;604 return Ok(Some(MoqMediaReadOutcome::Dropped {605 reason: KeyframeRequestReasonV3::TransportGap,606 }));607 }608 Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),609 };610 let discontinuity = self.waiting_for_keyframe611 || (first_object && cursor.group_gap)612 || (first_object && !frame_contiguous)613 || frame.header.flags.contains(FrameFlags::DISCONTINUITY);614 cursor.object_count += 1;615 cursor.object_bytes = next_group_bytes;616 self.last_frame_sequence = Some(frame.header.sequence);617 self.waiting_for_keyframe = false;618 return Ok(Some(MoqMediaReadOutcome::Frame {619 frame,620 discontinuity,621 }));622 }623 }624}625626fn classify_moq_group_sequence(previous: Option<u64>, current: u64) -> Result<bool, String> {627 let Some(previous) = previous else {628 return Ok(false);629 };630 if current <= previous {631 return Err(format!(632 "Upstream MoQ group sequence did not increase: previous={previous}, current={current}"633 ));634 }635 Ok(previous.checked_add(1) != Some(current))636}637638fn validate_moq_object_bounds(639 group_sequence: u64,640 object_count: usize,641 object_bytes: usize,642 object_len: usize,643) -> Result<usize, String> {644 let max_objects = MAX_MEDIA_OBJECT_ID_V3 as usize + 1;645 if object_count >= max_objects {646 return Err(format!(647 "Upstream MoQ group {group_sequence} exceeded {max_objects} media objects"648 ));649 }650 let next_group_bytes = object_bytes651 .checked_add(object_len)652 .ok_or_else(|| "Upstream MoQ group byte count overflowed".to_string())?;653 if next_group_bytes > MAX_MEDIA_GROUP_BYTES_V3 {654 return Err(format!(655 "Upstream MoQ group {group_sequence} exceeded the {MAX_MEDIA_GROUP_BYTES_V3} byte limit"656 ));657 }658 Ok(next_group_bytes)659}660661fn validate_moq_group_frame(662 group_sequence: u64,663 first_object: bool,664 last_frame_sequence: Option<u64>,665 frame: &MediaFrame,666) -> Result<bool, String> {667 if first_object668 && !(frame.header.codec == MediaCodec::H264669 && frame.header.flags.contains(FrameFlags::KEYFRAME)670 && frame.header.flags.contains(FrameFlags::CODEC_CONFIG))671 {672 return Err(format!(673 "Upstream MoQ group {group_sequence} did not begin with a configured H.264 keyframe"674 ));675 }676 let contiguous = last_frame_sequence677 .is_none_or(|previous| previous.checked_add(1) == Some(frame.header.sequence));678 if !first_object && !contiguous {679 return Err(format!(680 "Upstream MoQ group {group_sequence} contains a non-contiguous access-unit sequence"681 ));682 }683 Ok(contiguous)684}685686fn moq_group_error_is_recoverable(error: &moq_net::Error) -> bool {687 matches!(688 error,689 moq_net::Error::Cancel690 | moq_net::Error::Old691 | moq_net::Error::Timeout692 | moq_net::Error::Dropped693 | moq_net::Error::CacheFull694 | moq_net::Error::Remote(0 | 2 | 3 | 24 | 26)695 )696}697698fn moq_group_error_reason(error: &moq_net::Error) -> KeyframeRequestReasonV3 {699 if matches!(error, moq_net::Error::Timeout | moq_net::Error::Remote(3)) {700 KeyframeRequestReasonV3::DeliveryTimeout701 } else {702 KeyframeRequestReasonV3::TransportGap703 }704}705706pub(crate) async fn open_upstream_moq_media(707 endpoint: &Endpoint,708 address: &iroh::EndpointAddr,709 session_id: u64,710 advertised_generation_id: Option<u64>,711 advertised_broadcast_name: Option<&str>,712 authentication: Option<MoqAuthenticationExpectation>,713 verification_failures: Arc<AtomicU64>,714) -> Result<715 (716 MoqMediaReceiver,717 Option<MoqAudioReceiver>,718 iroh::endpoint::Connection,719 ),720 String,721> {722 let generation_id = advertised_generation_id.unwrap_or(session_id);723 let broadcast_name = match advertised_broadcast_name {724 Some(name) => name.to_owned(),725 None => media_moq_broadcast_name(session_id)726 .map_err(|error| format!("Invalid MoQ media session name: {error}"))?,727 };728 let moq = Moq::new(endpoint.clone());729 let mut session =730 tokio::time::timeout(CLIENT_MOQ_CONNECT_TIMEOUT, moq.connect(address.clone()))731 .await732 .map_err(|_| "Timed out connecting upstream MoQ media session".to_string())?733 .map_err(|error| format!("Failed to connect upstream MoQ media session: {error:#}"))?;734 if session.remote_id() != address.id {735 session.close(1, b"remote identity mismatch");736 return Err(format!(737 "Upstream MoQ connected to unexpected peer {}; expected {}",738 session.remote_id(),739 address.id740 ));741 }742 let diagnostics_connection = session.conn().clone();743 let broadcast = tokio::time::timeout(744 CLIENT_MOQ_SUBSCRIBE_TIMEOUT,745 session.subscribe(&broadcast_name),746 )747 .await748 .map_err(|_| format!("Timed out waiting for upstream MoQ broadcast {broadcast_name}"))?749 .map_err(|error| {750 format!("Failed to subscribe to upstream MoQ broadcast {broadcast_name}: {error}")751 })?;752 let catalog = if let Some(authentication) = authentication {753 let capability =754 SignedSubscriptionCapability::decode(&authentication.subscription_capability)755 .map_err(|error| format!("Invalid subscription capability: {error}"))?;756 capability757 .verify_binding(758 authentication.expected_host,759 generation_id,760 *endpoint.id().as_bytes(),761 SubscriptionTracks::VIDEO_H264,762 authentication.authorization_revision,763 authentication.now_unix,764 )765 .map_err(|error| format!("Subscription capability rejected: {error}"))?;766 subscribe_authenticated_goq_video_track(767 &broadcast,768 CLIENT_MOQ_SUBSCRIBE_TIMEOUT,769 authentication.expected_host,770 generation_id,771 authentication.now_unix,772 )773 .await?774 } else {775 subscribe_goq_video_track(&broadcast, CLIENT_MOQ_SUBSCRIBE_TIMEOUT).await?776 };777 eprintln!("[client] moq catalog: {}", catalog.mode.label());778 let audio = match (catalog.audio_track, catalog.certificate.clone()) {779 (Some(track), Some(certificate)) => Some(MoqAudioReceiver::new(780 track,781 certificate,782 Arc::clone(&verification_failures),783 )),784 _ => None,785 };786 Ok((787 MoqMediaReceiver::new(788 moq,789 session,790 broadcast,791 catalog.track,792 catalog.certificate,793 Some(verification_failures),794 ),795 audio,796 diagnostics_connection,797 ))798}799800fn record_verification_failure(counter: &Option<Arc<AtomicU64>>) {801 if let Some(counter) = counter {802 counter.fetch_add(1, Ordering::Relaxed);803 }804}805806pub(crate) struct MoqAuthenticationExpectation {807 pub(crate) expected_host: [u8; 32],808 pub(crate) subscription_capability: String,809 pub(crate) authorization_revision: u64,810 pub(crate) now_unix: u64,811}812813814mod tests {815 use super::*;816817 fn authentication_fixture() -> (818 sigil_protocol::MediaGenerationSigningKey,819 SignedMediaGenerationCertificate,820 ) {821 let host = ed25519_dalek::SigningKey::from_bytes(&[7; 32]);822 let generation = sigil_protocol::MediaGenerationSigningKey::from_bytes(&[9; 32]);823 let certificate = generation824 .certify(825 host.verifying_key().to_bytes(),826 &[7; 32],827 42,828 1_700_000_000,829 1_700_000_600,830 )831 .unwrap();832 (generation, certificate)833 }834835 836 fn upstream_moq_group_ids_detect_only_real_transport_gaps() {837 assert!(!classify_moq_group_sequence(None, 41).unwrap());838 assert!(!classify_moq_group_sequence(Some(41), 42).unwrap());839 assert!(classify_moq_group_sequence(Some(42), 44).unwrap());840 assert!(classify_moq_group_sequence(Some(44), 44).is_err());841 assert!(classify_moq_group_sequence(Some(44), 43).is_err());842 }843844 845 async fn authenticated_receiver_verifies_before_codec_validation_and_counts_tampering() {846 let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();847 let (generation, certificate) = authentication_fixture();848 let failures = Arc::new(AtomicU64::new(0));849 let mut receiver = MoqMediaReceiver::authenticated_for_test(850 producer.consume(),851 certificate,852 Arc::clone(&failures),853 );854 let frame = media_object_frame(1, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));855 let payload = sigil_protocol::encode_media_frame_object(&frame).unwrap();856 let mut group = producer.append_group().unwrap();857 let object = generation858 .authenticate(859 MediaObjectCoordinates {860 generation_id: 42,861 track: MediaTrack::VideoH264,862 group_id: group.sequence,863 object_id: 0,864 flags: u16::from(frame.header.flags.bits()),865 },866 &payload,867 )868 .unwrap();869 group.write_frame(object.into_bytes()).unwrap();870 group.finish().unwrap();871 assert!(matches!(872 receiver.next().await.unwrap(),873 Some(MoqMediaReadOutcome::Frame { .. })874 ));875 assert_eq!(failures.load(Ordering::Relaxed), 0);876877 let mut tampered_group = producer.append_group().unwrap();878 let tampered_frame =879 media_object_frame(2, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));880 let payload = sigil_protocol::encode_media_frame_object(&tampered_frame).unwrap();881 let mut object = generation882 .authenticate(883 MediaObjectCoordinates {884 generation_id: 42,885 track: MediaTrack::VideoH264,886 group_id: tampered_group.sequence,887 object_id: 0,888 flags: u16::from(tampered_frame.header.flags.bits()),889 },890 &payload,891 )892 .unwrap()893 .into_bytes();894 *object.last_mut().unwrap() ^= 1;895 tampered_group.write_frame(object).unwrap();896 tampered_group.finish().unwrap();897 assert!(matches!(898 receiver.next().await.unwrap(),899 Some(MoqMediaReadOutcome::Malformed(_))900 ));901 assert_eq!(failures.load(Ordering::Relaxed), 1);902 }903904 905 async fn authenticated_audio_receiver_preserves_verified_discontinuity() {906 let mut producer = Track::new(sigil_protocol::MOQ_AUDIO_OPUS_TRACK).produce();907 let (generation, certificate) = authentication_fixture();908 let failures = Arc::new(AtomicU64::new(0));909 let mut receiver =910 MoqAudioReceiver::new(producer.consume(), certificate, Arc::clone(&failures));911 let packet = AudioPacket::new(912 AudioPacketHeader::opus(3, 7, 140_000, 140_000, AudioFlags::NONE).unwrap(),913 vec![0xf8, 0xff, 0xfe],914 )915 .unwrap();916 let payload = packet.encode_datagram().unwrap();917 let mut group = producer.append_group().unwrap();918 let object = generation919 .authenticate(920 MediaObjectCoordinates {921 generation_id: 42,922 track: MediaTrack::AudioOpus,923 group_id: group.sequence,924 object_id: 0,925 flags: 0,926 },927 &payload,928 )929 .unwrap();930 group.write_frame(object.into_bytes()).unwrap();931 group.finish().unwrap();932933 let received = receiver.next().await.unwrap().unwrap();934 assert_eq!(received.header.sequence, 7);935 assert!(received.header.flags.contains(AudioFlags::DISCONTINUITY));936 assert_eq!(failures.load(Ordering::Relaxed), 0);937 }938939 940 fn upstream_moq_group_gap_keyframe_is_the_recovery_barrier() {941 let keyframe = media_object_frame(80, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));942 let contiguous = validate_moq_group_frame(9, true, Some(42), &keyframe).unwrap();943 assert!(!contiguous);944945 // A native group-id gap puts the decoder into resync, but this exact946 // configured frame 0 exits it immediately. There is deliberately no947 // keyframe-request action on MoqMediaReadOutcome::Frame: requesting a948 // replacement here would recreate the grouped-v3 feedback loop.949 let group_gap = classify_moq_group_sequence(Some(7), 9).unwrap();950 let discontinuity = group_gap || !contiguous;951 assert!(discontinuity);952 }953954 955 async fn upstream_moq_idr_abort_to_next_frame_zero_has_no_feedback_loop() {956 let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();957 let mut receiver = MoqMediaReceiver::for_test(producer.consume());958959 let mut prior_group = producer.append_group().unwrap();960 let prior_keyframe =961 media_object_frame(40, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));962 prior_group963 .write_frame(sigil_protocol::encode_media_frame_object(&prior_keyframe).unwrap())964 .unwrap();965 assert!(matches!(966 receiver.next().await.unwrap(),967 Some(MoqMediaReadOutcome::Frame { .. })968 ));969970 prior_group.abort(moq_net::Error::Cancel).unwrap();971 let mut replacement = producer.append_group().unwrap();972 let replacement_keyframe =973 media_object_frame(80, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));974 replacement975 .write_frame(sigil_protocol::encode_media_frame_object(&replacement_keyframe).unwrap())976 .unwrap();977 replacement.finish().unwrap();978979 // `next` consumes the expected Cancel internally and returns the980 // replacement frame directly. A Dropped outcome here would enqueue a981 // needless keyframe request and recreate the recovery feedback loop.982 assert!(matches!(983 receiver.next().await.unwrap(),984 Some(MoqMediaReadOutcome::Frame {985 discontinuity: true,986 ..987 })988 ));989 }990991 992 async fn upstream_moq_expected_cancel_without_replacement_is_terminal_without_request() {993 let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();994 let mut receiver = MoqMediaReceiver::for_test(producer.consume());995 let mut group = producer.append_group().unwrap();996 let keyframe = media_object_frame(1, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));997 group998 .write_frame(sigil_protocol::encode_media_frame_object(&keyframe).unwrap())999 .unwrap();1000 assert!(matches!(1001 receiver.next().await.unwrap(),1002 Some(MoqMediaReadOutcome::Frame { .. })1003 ));1004 group.abort(moq_net::Error::Cancel).unwrap();10051006 let error = receiver1007 .next_with_timeouts(Duration::from_millis(25), Duration::from_millis(25))1008 .await1009 .unwrap_err();1010 assert!(error.contains("expected GOP cancellation"));1011 }10121013 1014 async fn upstream_moq_partial_object_read_has_protocol_bounded_deadline() {1015 let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();1016 let mut receiver = MoqMediaReceiver::for_test(producer.consume());1017 let _stalled_group = producer.append_group().unwrap();10181019 assert!(matches!(1020 receiver1021 .next_with_timeouts(Duration::from_millis(25), Duration::from_millis(25))1022 .await1023 .unwrap(),1024 Some(MoqMediaReadOutcome::Dropped {1025 reason: KeyframeRequestReasonV3::DeliveryTimeout1026 })1027 ));1028 assert_eq!(1029 CLIENT_MOQ_OBJECT_READ_TIMEOUT,1030 Duration::from_millis(MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS as u64)1031 );1032 assert!(CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT >= Duration::from_millis(1_000));1033 }10341035 1036 async fn upstream_moq_invalid_cancel_replacement_requests_recovery() {1037 let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();1038 let mut receiver = MoqMediaReceiver::for_test(producer.consume());1039 let mut prior = producer.append_group().unwrap();1040 let keyframe = media_object_frame(10, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));1041 prior1042 .write_frame(sigil_protocol::encode_media_frame_object(&keyframe).unwrap())1043 .unwrap();1044 assert!(matches!(1045 receiver.next().await.unwrap(),1046 Some(MoqMediaReadOutcome::Frame { .. })1047 ));1048 prior.abort(moq_net::Error::Cancel).unwrap();10491050 let mut invalid_replacement = producer.append_group().unwrap();1051 let delta = media_object_frame(11, FrameFlags::NONE);1052 invalid_replacement1053 .write_frame(sigil_protocol::encode_media_frame_object(&delta).unwrap())1054 .unwrap();1055 invalid_replacement.finish().unwrap();1056 assert!(matches!(1057 receiver.next().await.unwrap(),1058 Some(MoqMediaReadOutcome::Dropped {1059 reason: KeyframeRequestReasonV3::TransportGap1060 })1061 ));1062 }10631064 1065 fn upstream_moq_requires_configured_keyframe_zero_and_contiguous_deltas() {1066 let delta = media_object_frame(10, FrameFlags::NONE);1067 assert!(validate_moq_group_frame(3, true, None, &delta).is_err());10681069 let unconfigured_keyframe = media_object_frame(10, FrameFlags::KEYFRAME);1070 assert!(validate_moq_group_frame(3, true, None, &unconfigured_keyframe).is_err());10711072 let configured =1073 media_object_frame(10, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));1074 assert!(validate_moq_group_frame(3, true, None, &configured).is_ok());1075 assert!(validate_moq_group_frame(3, false, Some(8), &delta).is_err());1076 assert!(validate_moq_group_frame(3, false, Some(9), &delta).is_ok());1077 }10781079 1080 fn upstream_moq_bounds_objects_and_group_bytes_before_growth() {1081 assert_eq!(validate_moq_object_bounds(1, 0, 0, 40).unwrap(), 40);1082 assert!(validate_moq_object_bounds(1, MAX_MEDIA_OBJECT_ID_V3 as usize + 1, 0, 1,).is_err());1083 assert!(validate_moq_object_bounds(1, 0, MAX_MEDIA_GROUP_BYTES_V3, 1).is_err());1084 assert!(validate_moq_object_bounds(1, 0, usize::MAX, 1).is_err());1085 }10861087 1088 fn upstream_moq_cancellation_is_resync_but_protocol_errors_are_terminal() {1089 for recoverable in [1090 moq_net::Error::Cancel,1091 moq_net::Error::Old,1092 moq_net::Error::Timeout,1093 moq_net::Error::Dropped,1094 moq_net::Error::CacheFull,1095 moq_net::Error::Remote(3),1096 ] {1097 assert!(moq_group_error_is_recoverable(&recoverable));1098 }1099 assert!(!moq_group_error_is_recoverable(1100 &moq_net::Error::ProtocolViolation1101 ));1102 assert!(!moq_group_error_is_recoverable(&moq_net::Error::WrongSize));1103 assert_eq!(1104 moq_group_error_reason(&moq_net::Error::Timeout),1105 KeyframeRequestReasonV3::DeliveryTimeout1106 );1107 assert_eq!(1108 moq_group_error_reason(&moq_net::Error::Cancel),1109 KeyframeRequestReasonV3::TransportGap1110 );1111 }1112}