1use super::*;23const MOQ_ATTACHMENT_TIMEOUT: Duration = Duration::from_secs(10);4pub(super) const MOQ_REJECT_CODE: u32 = 0x534d;567enum MoqGroupDecision {8 Published {9 group_id: u64,10 frame_id: u32,11 cancelled_previous: bool,12 },13 SkipUntilKeyframe,14 EnterResync,15}1617/// Owns the single bounded live MoQ track. One configured H.264 GOP maps to18/// one native MoQ group; its application frame sequence remains inside the19/// encoded object envelope and is never reused as the transport group id.20struct MoqGroupPublisher {21 track: TrackProducer,22 current_group: Option<GroupProducer>,23 cursor: MediaV3GroupCursor,24 object_bytes: usize,25}2627impl MoqGroupPublisher {28 fn new(track: TrackProducer) -> Self {29 Self {30 track,31 current_group: None,32 cursor: MediaV3GroupCursor::default(),33 object_bytes: 0,34 }35 }3637 fn publish(38 &mut self,39 config: &HostConfig,40 frame: &EncodedFrame,41 replay_discontinuity: bool,42 ) -> Result<MoqGroupDecision> {43 let position = match self.cursor.classify(frame) {44 MediaV3GroupDecision::Send(position) => position,45 MediaV3GroupDecision::SkipUntilKeyframe => {46 return Ok(MoqGroupDecision::SkipUntilKeyframe);47 }48 MediaV3GroupDecision::EnterResync => {49 self.abort_current();50 return Ok(MoqGroupDecision::EnterResync);51 }52 };53 let object = encode_media_frame_object(&media_frame_for_encoded(54 config,55 frame,56 replay_discontinuity || position.discontinuity,57 )?)?;58 let next_object_bytes = if position.object_id == 0 {59 Some(object.len())60 } else {61 self.object_bytes.checked_add(object.len())62 };63 let Some(next_object_bytes) =64 next_object_bytes.filter(|bytes| *bytes <= MAX_MEDIA_GROUP_BYTES_V3)65 else {66 self.cursor.request_keyframe();67 self.abort_current();68 return Ok(MoqGroupDecision::EnterResync);69 };7071 if position.object_id == 0 {72 // A new independently-decodable GOP supersedes the previous one.73 // Actively aborting it cancels a slow subscriber rather than74 // retaining a playable history behind the live edge.75 let cancelled_previous = self.abort_current().is_some();76 let mut group = self77 .track78 .append_group()79 .context("creating sequential MoQ video group")?;80 let group_id = group.sequence;81 group82 .write_frame(object)83 .context("writing configured keyframe to MoQ group")?;84 self.object_bytes = next_object_bytes;85 self.current_group = Some(group);86 return Ok(MoqGroupDecision::Published {87 group_id,88 frame_id: 0,89 cancelled_previous,90 });91 }9293 let group = self94 .current_group95 .as_mut()96 .context("MoQ delta frame has no active configured-keyframe group")?;97 let group_id = group.sequence;98 group99 .write_frame(object)100 .context("writing delta access unit to MoQ group")?;101 self.object_bytes = next_object_bytes;102 Ok(MoqGroupDecision::Published {103 group_id,104 frame_id: position.object_id,105 cancelled_previous: false,106 })107 }108109 fn request_keyframe(&mut self) -> Option<u64> {110 self.cursor.request_keyframe();111 self.abort_current()112 }113114 fn abort_current(&mut self) -> Option<u64> {115 self.object_bytes = 0;116 let mut group = self.current_group.take()?;117 let group_id = group.sequence;118 let _ = group.abort(MoqError::Cancel);119 Some(group_id)120 }121122 fn abort(mut self) {123 self.abort_current();124 let _ = self.track.abort(MoqError::Cancel);125 }126}127128fn apply_moq_keyframe_request(129 publisher: &mut MoqGroupPublisher,130 replay_cursor: &mut MediaReplayCursor,131 through_sequence: Option<u64>,132 reason: KeyframeRequestReasonV3,133) -> Option<u64> {134 // The bounded current group is already the late joiner's decodable replay.135 // Aborting it on Join can strand a static source until its next natural IDR.136 if reason == KeyframeRequestReasonV3::Join {137 return None;138 }139 let cancelled_group = publisher.request_keyframe();140 replay_cursor.enter_resync_through(through_sequence);141 cancelled_group142}143144pub(super) async fn serve_authorized_moq(145 connection: Connection,146 origin: Origin,147 attachment: ClaimedMoqAttachment,148) -> Result<()> {149 let ClaimedMoqAttachment {150 session_id,151 broadcast_name,152 broadcast,153 attached,154 closed,155 telemetry,156 } = attachment;157 let result: Result<()> = async {158 let web_transport = web_transport_iroh::Session::raw(connection);159 let session = tokio::time::timeout(160 MOQ_ATTACHMENT_TIMEOUT,161 iroh_moq::MoqSession::session_accept(web_transport, origin),162 )163 .await164 .context("timed out completing authorized MoQ handshake")?165 .context("completing authorized MoQ handshake")?;166 let broadcast_closed = broadcast.clone();167 session.publish(&broadcast_name, broadcast);168 ensure!(169 attached.send(()).is_ok(),170 "control session ended before MoQ attachment completed"171 );172 info!(173 remote = %session.remote_id(),174 session_id,175 %broadcast_name,176 track = MOQ_VIDEO_H264_TRACK,177 "authorized MoQ media attachment accepted"178 );179 let mut telemetry_interval = tokio::time::interval(Duration::from_secs(1));180 telemetry_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);181 loop {182 tokio::select! {183 reason = session.closed() => {184 debug!(remote = %session.remote_id(), ?reason, "MoQ media session closed");185 break;186 }187 reason = broadcast_closed.closed() => {188 debug!(remote = %session.remote_id(), ?reason, "control-owned MoQ broadcast closed");189 session.close(0, b"control session ended");190 break;191 }192 _ = telemetry_interval.tick() => {193 telemetry.record_selected_path(session.conn());194 }195 }196 }197 Ok(())198 }199 .await;200 let _ = closed.send(());201 result202}203204pub(super) async fn serve_control_moq(205 connection: Connection,206 config: HostConfig,207 sessions: &Arc<SessionRegistry>,208 authorization: &AuthorizationPolicy,209) -> Result<()> {210 let remote = connection.remote_id();211 let handshake_permit = sessions212 .pending_handshakes213 .try_acquire()214 .context("too many pending handshakes")?;215 let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())216 .await217 .context("timed out accepting MoQ control stream")?218 .context("accepting MoQ control stream")?;219 let hello = receive_hello(&mut recv, Capability::VideoH264).await?;220 drop(handshake_permit);221 debug!(%remote, agent = %hello.agent, "MoQ control hello received");222223 let grants = match authorization.authorize_or_redeem(224 remote,225 hello.invitation.as_deref(),226 unix_timestamp_now()?,227 ) {228 Ok(grants) => grants,229 Err(error) => {230 send_rejection(&mut send, "Portal peer is not authorized").await?;231 return Err(error.context("authorizing MoQ control peer"));232 }233 };234 ensure!(235 grants.contains(InvitationGrants::VIEW),236 "authorized MoQ control peer lacks view permission"237 );238 let lease = match sessions.claim(remote, hello.nonce, grants) {239 Ok(lease) => lease,240 Err(error) => {241 send_rejection(&mut send, "host already has an active client").await?;242 return Err(error);243 }244 };245246 let source = match config.source {247 VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),248 VideoSource::GamescopePipewire => {249 let primary = spawn_gamescope_pipewire_after_static_preflight(250 config.clone(),251 lease.session_clock,252 )253 .await?;254 select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await255 }256 };257 let EncodedSource {258 frames: frame_receiver,259 current_gop: mut current_gop_receiver,260 task: source_task,261 pointer_surface_dimensions,262 encoder_control,263 } = match source {264 Ok(source) => source,265 Err(error) => {266 send_rejection(&mut send, "video source is unavailable").await?;267 return Err(error);268 }269 };270 let source_task = SourceTaskGuard::new(source_task);271 sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;272273 let mut broadcast = Broadcast::new().produce();274 let track = broadcast275 .create_track(Track {276 name: MOQ_VIDEO_H264_TRACK.to_owned(),277 priority: MOQ_VIDEO_TRACK_PRIORITY,278 })279 .context("creating static MoQ H.264 track")?;280 let catalog = publish_goq_catalog(&mut broadcast)?;281 let broadcast_name = media_moq_broadcast_name(lease.session_id)?;282 let attachment = sessions.expect_moq(283 remote,284 lease.session_id,285 broadcast_name.clone(),286 broadcast.consume(),287 )?;288289 let mut control_hello = HostHello::accepted(290 lease.session_id,291 negotiated_capabilities(&hello, MEDIA_CAPABILITIES),292 );293 if let Some(dimensions) = pointer_surface_dimensions {294 control_hello = control_hello.with_pointer_surface_dimensions(dimensions);295 }296 write_host_hello(&mut send, &control_hello).await?;297 send.finish().context("finishing MoQ control response")?;298 drop(send);299 info!(300 %remote,301 session_id = lease.session_id,302 %broadcast_name,303 "MoQ control client accepted; awaiting authorized media attachment"304 );305306 let MoqAttachmentWait {307 mut attached,308 closed,309 } = attachment;310 tokio::time::timeout(MOQ_ATTACHMENT_TIMEOUT, async {311 tokio::select! {312 result = &mut attached => {313 result.context("authorized MoQ handler ended before attachment")314 }315 reason = connection.closed() => {316 Err(anyhow::anyhow!("control connection closed before MoQ attachment: {reason:?}"))317 }318 }319 })320 .await321 .context("timed out waiting for authorized MoQ attachment")??;322323 let session_result = run_control_moq_session(324 &connection,325 &config,326 &mut current_gop_receiver,327 recv,328 remote,329 closed,330 track,331 &mut broadcast,332 encoder_control,333 Arc::clone(&lease.media_v3_telemetry),334 )335 .await;336 let catalog_result = catalog.finish();337338 drop(current_gop_receiver);339 drop(frame_receiver);340 source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;341 drop(lease);342 info!(%remote, "MoQ control client released");343 match session_result {344 Err(error) => Err(error),345 Ok(()) => catalog_result,346 }347}348349350async fn run_control_moq_session(351 connection: &Connection,352 config: &HostConfig,353 current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,354 control_recv: iroh::endpoint::RecvStream,355 remote: EndpointId,356 mut moq_closed: tokio::sync::oneshot::Receiver<()>,357 track: TrackProducer,358 broadcast: &mut BroadcastProducer,359 encoder_control: Option<EncoderControl>,360 telemetry: Arc<MediaV3Telemetry>,361) -> Result<()> {362 let maximum_replay_age = maximum_media_replay_age(config.framerate);363 let mut replay_cursor = MediaReplayCursor::default();364 let mut publisher = MoqGroupPublisher::new(track);365 let (control_sender, mut control_requests) = tokio::sync::watch::channel(None);366 let mut control_task = tokio::spawn(forward_media_v3_control_requests(367 control_recv,368 control_sender,369 ));370 let mut control_task_finished = false;371 let mut control_receiver_open = true;372 let mut forced_idr = ForcedIdrCoordinator::new(encoder_control, Arc::clone(&telemetry));373374 let result = async {375 loop {376 tokio::select! {377 biased;378 control_result = &mut control_task, if !control_task_finished => {379 control_task_finished = true;380 match control_result {381 Ok(Ok(())) => {382 debug!(%remote, "MoQ keyframe-control stream closed");383 }384 Ok(Err(error)) => {385 return Err(error).context("reading MoQ keyframe-control stream");386 }387 Err(error) => {388 return Err(error).context("MoQ keyframe-control task failed");389 }390 }391 }392 changed = control_requests.changed(), if control_receiver_open => {393 if changed.is_err() {394 control_receiver_open = false;395 continue;396 }397 let Some(request) = *control_requests.borrow_and_update() else {398 continue;399 };400 let through_sequence = current_gop_receiver401 .borrow()402 .as_ref()403 .and_then(|gop| gop.frames.last())404 .map(|frame| frame.sequence);405 let cancelled_group = apply_moq_keyframe_request(406 &mut publisher,407 &mut replay_cursor,408 through_sequence,409 request.reason,410 );411 if cancelled_group.is_some() {412 telemetry413 .scheduler_cancellations414 .fetch_add(1, Ordering::Relaxed);415 }416 let forced_idr_disposition = forced_idr.request(request.reason);417 if let ForcedIdrDisposition::Failed { error } = &forced_idr_disposition {418 warn!(419 %remote,420 request_id = request.request_id,421 ?request.reason,422 %error,423 "forced-IDR request failed; retaining natural-IDR fallback"424 );425 }426 debug!(427 %remote,428 request_id = request.request_id,429 ?request.reason,430 advisory_last_sequence = ?request.last_sequence,431 coalesced = cancelled_group.is_none(),432 ?cancelled_group,433 ?forced_idr_disposition,434 "accepted MoQ keyframe request"435 );436 }437 acknowledgement = forced_idr.acknowledgements.join_next(),438 if forced_idr.pending_revision.is_some() =>439 {440 forced_idr.complete(acknowledgement, remote, "iroh-moq");441 }442 reason = connection.closed() => {443 debug!(%remote, ?reason, "MoQ control connection closed");444 return Ok(());445 }446 result = &mut moq_closed => {447 debug!(%remote, ?result, "authorized MoQ media attachment closed");448 return Ok(());449 }450 changed = current_gop_receiver.changed() => {451 if let Err(error) = changed {452 return Err(error).context("encoded source stopped");453 }454 let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {455 continue;456 };457 let initial_replay_started_at =458 replay_cursor.last_sequence.is_none().then(Instant::now);459 let replay_through_sequence = current_gop460 .frames461 .last()462 .map(|frame| frame.sequence)463 .context("current GOP snapshot is empty")?;464 for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {465 let replay_discontinuity = match replay_cursor.classify(466 &frame,467 replay_through_sequence,468 initial_replay_started_at,469 Instant::now(),470 maximum_replay_age,471 ) {472 MediaReplayDecision::Send { discontinuity } => discontinuity,473 MediaReplayDecision::SkipUntilKeyframe => {474 if publisher.request_keyframe().is_some() {475 telemetry476 .scheduler_cancellations477 .fetch_add(1, Ordering::Relaxed);478 }479 replay_cursor.enter_resync_through(Some(replay_through_sequence));480 break;481 }482 MediaReplayDecision::DiscardStaleSuffix { through_sequence } => {483 let cancelled_group = publisher.request_keyframe();484 if cancelled_group.is_some() {485 telemetry486 .scheduler_cancellations487 .fetch_add(1, Ordering::Relaxed);488 }489 debug!(490 %remote,491 through_sequence,492 ?cancelled_group,493 "cancelled stale MoQ media suffix"494 );495 break;496 }497 };498 let decision = publisher499 .publish(config, &frame, replay_discontinuity)500 .inspect_err(|_error| {501 telemetry.send_failures.fetch_add(1, Ordering::Relaxed);502 })?;503 match decision {504 MoqGroupDecision::Published {505 group_id,506 frame_id,507 cancelled_previous,508 } => {509 if cancelled_previous {510 telemetry511 .scheduler_cancellations512 .fetch_add(1, Ordering::Relaxed);513 }514 debug!(515 sequence = frame.sequence,516 group_id,517 frame_id,518 cancelled_previous,519 "published upstream MoQ video frame"520 );521 replay_cursor.commit_sent(&frame);522 }523 MoqGroupDecision::SkipUntilKeyframe => {524 replay_cursor.enter_resync_through(Some(replay_through_sequence));525 break;526 }527 MoqGroupDecision::EnterResync => {528 replay_cursor.enter_resync_through(Some(replay_through_sequence));529 break;530 }531 }532 }533 }534 }535 }536 }537 .await;538539 forced_idr.abort_and_drain(remote, "iroh-moq").await;540 publisher.abort();541 let _ = broadcast.abort(MoqError::Cancel);542 if !control_task_finished {543 control_task.abort();544 let _ = control_task.await;545 }546 result547}548549pub(super) fn media_frame_for_encoded(550 _config: &HostConfig,551 frame: &EncodedFrame,552 discontinuity: bool,553) -> Result<MediaFrame> {554 let mut flags = FrameFlags::NONE;555 if frame.keyframe {556 flags = flags.union(FrameFlags::KEYFRAME);557 }558 if frame.codec_config {559 flags = flags.union(FrameFlags::CODEC_CONFIG);560 }561 if discontinuity || frame.discontinuity {562 flags = flags.union(FrameFlags::DISCONTINUITY);563 }564 let header = MediaFrameHeader::h264(565 frame.width,566 frame.height,567 frame.data.len(),568 frame.sequence,569 frame.capture_timestamp_micros,570 frame.presentation_timestamp_micros,571 flags,572 )?;573 MediaFrame::new(header, frame.data.clone()).map_err(Into::into)574}575576577mod tests {578 use super::super::{media_v3_encoded_frame, moq_test_config};579 use super::*;580581 582 async fn upstream_moq_groups_are_sequential_and_cancel_the_superseded_gop() {583 let track_info = Track {584 name: MOQ_VIDEO_H264_TRACK.to_owned(),585 priority: MOQ_VIDEO_TRACK_PRIORITY,586 };587 let mut broadcast = Broadcast::new().produce();588 let track = broadcast.create_track(track_info.clone()).unwrap();589 let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();590 let mut publisher = MoqGroupPublisher::new(track);591 let config = moq_test_config();592593 assert_eq!(594 publisher595 .publish(&config, &media_v3_encoded_frame(100, true, true, 4), false)596 .unwrap(),597 MoqGroupDecision::Published {598 group_id: 0,599 frame_id: 0,600 cancelled_previous: false,601 }602 );603 let mut first_group = consumer.recv_group().await.unwrap().unwrap();604 assert_eq!(first_group.sequence, 0);605 assert!(first_group.read_frame().await.unwrap().is_some());606607 assert_eq!(608 publisher609 .publish(610 &config,611 &media_v3_encoded_frame(101, false, false, 4),612 false,613 )614 .unwrap(),615 MoqGroupDecision::Published {616 group_id: 0,617 frame_id: 1,618 cancelled_previous: false,619 }620 );621 assert!(first_group.read_frame().await.unwrap().is_some());622623 assert_eq!(624 publisher625 .publish(&config, &media_v3_encoded_frame(200, true, true, 4), false)626 .unwrap(),627 MoqGroupDecision::Published {628 group_id: 1,629 frame_id: 0,630 cancelled_previous: true,631 }632 );633 assert!(first_group.finished().await.is_err());634 let mut second_group = consumer.recv_group().await.unwrap().unwrap();635 assert_eq!(second_group.sequence, 1);636 let object = second_group.read_frame().await.unwrap().unwrap();637 let frame = sigil_protocol::decode_media_frame_object(&object).unwrap();638 assert_eq!(frame.header.sequence, 200);639 assert!(frame.header.flags.contains(FrameFlags::DISCONTINUITY));640 }641642 643 async fn upstream_moq_late_join_preserves_active_static_group() {644 let track_info = Track {645 name: MOQ_VIDEO_H264_TRACK.to_owned(),646 priority: MOQ_VIDEO_TRACK_PRIORITY,647 };648 let mut broadcast = Broadcast::new().produce();649 let track = broadcast.create_track(track_info.clone()).unwrap();650 let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();651 let mut publisher = MoqGroupPublisher::new(track);652 let mut replay_cursor = MediaReplayCursor::default();653 let config = moq_test_config();654 let keyframe = media_v3_encoded_frame(10, true, true, 1);655 let first_delta = media_v3_encoded_frame(11, false, false, 1);656 let next_delta = media_v3_encoded_frame(12, false, false, 1);657658 publisher.publish(&config, &keyframe, false).unwrap();659 replay_cursor.commit_sent(&keyframe);660 let mut active_group = consumer.recv_group().await.unwrap().unwrap();661 assert!(active_group.read_frame().await.unwrap().is_some());662663 publisher.publish(&config, &first_delta, false).unwrap();664 replay_cursor.commit_sent(&first_delta);665 assert!(active_group.read_frame().await.unwrap().is_some());666667 assert_eq!(668 apply_moq_keyframe_request(669 &mut publisher,670 &mut replay_cursor,671 Some(first_delta.sequence),672 KeyframeRequestReasonV3::Join,673 ),674 None675 );676 assert_eq!(replay_cursor.last_sequence, Some(first_delta.sequence));677 assert!(!replay_cursor.waiting_for_keyframe);678 assert_eq!(679 publisher.publish(&config, &next_delta, false).unwrap(),680 MoqGroupDecision::Published {681 group_id: 0,682 frame_id: 2,683 cancelled_previous: false,684 }685 );686 assert!(active_group.read_frame().await.unwrap().is_some());687 }688689 690 async fn upstream_moq_resync_aborts_current_group_and_waits_for_configured_idr() {691 let track_info = Track {692 name: MOQ_VIDEO_H264_TRACK.to_owned(),693 priority: MOQ_VIDEO_TRACK_PRIORITY,694 };695 let mut broadcast = Broadcast::new().produce();696 let track = broadcast.create_track(track_info.clone()).unwrap();697 let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();698 let mut publisher = MoqGroupPublisher::new(track);699 let config = moq_test_config();700701 publisher702 .publish(&config, &media_v3_encoded_frame(10, true, true, 1), false)703 .unwrap();704 let mut cancelled = consumer.recv_group().await.unwrap().unwrap();705 assert_eq!(publisher.request_keyframe(), Some(0));706 assert!(cancelled.finished().await.is_err());707 assert_eq!(708 publisher709 .publish(&config, &media_v3_encoded_frame(11, false, false, 1), false,)710 .unwrap(),711 MoqGroupDecision::SkipUntilKeyframe712 );713 assert_eq!(714 publisher715 .publish(&config, &media_v3_encoded_frame(20, true, true, 1), false)716 .unwrap(),717 MoqGroupDecision::Published {718 group_id: 1,719 frame_id: 0,720 cancelled_previous: false,721 }722 );723 }724725 726 async fn upstream_moq_group_counts_envelope_bytes_before_upstream_cache_eviction() {727 let track_info = Track {728 name: MOQ_VIDEO_H264_TRACK.to_owned(),729 priority: MOQ_VIDEO_TRACK_PRIORITY,730 };731 let mut broadcast = Broadcast::new().produce();732 let track = broadcast.create_track(track_info.clone()).unwrap();733 let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();734 let mut publisher = MoqGroupPublisher::new(track);735 let config = moq_test_config();736 publisher737 .publish(&config, &media_v3_encoded_frame(10, true, true, 1), false)738 .unwrap();739 let mut cancelled = consumer.recv_group().await.unwrap().unwrap();740741 // Payload-only accounting would accept this next one-byte access unit,742 // but its fixed application envelope would overflow moq-net's 32 MiB743 // group cache and silently evict the keyframe.744 publisher.object_bytes = MAX_MEDIA_GROUP_BYTES_V3 - 1;745 assert_eq!(746 publisher747 .publish(&config, &media_v3_encoded_frame(11, false, false, 1), false,)748 .unwrap(),749 MoqGroupDecision::EnterResync750 );751 assert!(cancelled.finished().await.is_err());752 assert_eq!(publisher.object_bytes, 0);753 }754}