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.20pub(super) struct MoqGroupPublisher {21 track: TrackProducer,22 current_group: Option<GroupProducer>,23 cursor: MediaV3GroupCursor,24 object_bytes: usize,25 authentication: Option<MoqObjectAuthentication>,26}2728struct MoqObjectAuthentication {29 generation_id: u64,30 signing_key: MediaGenerationSigningKey,31}3233impl MoqGroupPublisher {34 fn new(track: TrackProducer) -> Self {35 Self {36 track,37 current_group: None,38 cursor: MediaV3GroupCursor::default(),39 object_bytes: 0,40 authentication: None,41 }42 }4344 fn authenticated(45 track: TrackProducer,46 generation_id: u64,47 signing_key: MediaGenerationSigningKey,48 ) -> Self {49 Self {50 track,51 current_group: None,52 cursor: MediaV3GroupCursor::default(),53 object_bytes: 0,54 authentication: Some(MoqObjectAuthentication {55 generation_id,56 signing_key,57 }),58 }59 }6061 fn publish(62 &mut self,63 config: &HostConfig,64 frame: &EncodedFrame,65 replay_discontinuity: bool,66 ) -> Result<MoqGroupDecision> {67 let position = match self.cursor.classify(frame) {68 MediaV3GroupDecision::Send(position) => position,69 MediaV3GroupDecision::SkipUntilKeyframe => {70 return Ok(MoqGroupDecision::SkipUntilKeyframe);71 }72 MediaV3GroupDecision::EnterResync => {73 self.abort_current();74 return Ok(MoqGroupDecision::EnterResync);75 }76 };77 let media_frame = media_frame_for_encoded(78 config,79 frame,80 replay_discontinuity || position.discontinuity,81 )?;82 let payload = encode_media_frame_object(&media_frame)?;8384 if position.object_id == 0 {85 // A new independently-decodable GOP supersedes the previous one.86 // Actively aborting it cancels a slow subscriber rather than87 // retaining a playable history behind the live edge.88 let cancelled_previous = self.abort_current().is_some();89 let mut group = self90 .track91 .append_group()92 .context("creating sequential MoQ video group")?;93 let group_id = group.sequence;94 let object = self.authenticate_object(95 group_id,96 position.object_id,97 u16::from(media_frame.header.flags.bits()),98 payload,99 )?;100 let object_len = object.len();101 if object.len() > MAX_MEDIA_GROUP_BYTES_V3 {102 let _ = group.abort(MoqError::Cancel);103 self.cursor.request_keyframe();104 return Ok(MoqGroupDecision::EnterResync);105 }106 group107 .write_frame(object)108 .context("writing configured keyframe to MoQ group")?;109 self.object_bytes = object_len;110 self.current_group = Some(group);111 return Ok(MoqGroupDecision::Published {112 group_id,113 frame_id: 0,114 cancelled_previous,115 });116 }117118 let group_id = self119 .current_group120 .as_ref()121 .map(|group| group.sequence)122 .context("MoQ delta frame has no active configured-keyframe group")?;123 let object = self.authenticate_object(124 group_id,125 position.object_id,126 u16::from(media_frame.header.flags.bits()),127 payload,128 )?;129 let Some(next_object_bytes) = self130 .object_bytes131 .checked_add(object.len())132 .filter(|bytes| *bytes <= MAX_MEDIA_GROUP_BYTES_V3)133 else {134 self.cursor.request_keyframe();135 self.abort_current();136 return Ok(MoqGroupDecision::EnterResync);137 };138 let group = self139 .current_group140 .as_mut()141 .context("MoQ delta frame lost its configured-keyframe group")?;142 group143 .write_frame(object)144 .context("writing delta access unit to MoQ group")?;145 self.object_bytes = next_object_bytes;146 Ok(MoqGroupDecision::Published {147 group_id,148 frame_id: position.object_id,149 cancelled_previous: false,150 })151 }152153 fn authenticate_object(154 &self,155 group_id: u64,156 object_id: u32,157 flags: u16,158 payload: Vec<u8>,159 ) -> Result<Vec<u8>> {160 let Some(authentication) = &self.authentication else {161 return Ok(payload);162 };163 authentication164 .signing_key165 .authenticate(166 MediaObjectCoordinates {167 generation_id: authentication.generation_id,168 track: MediaTrack::VideoH264,169 group_id,170 object_id,171 flags,172 },173 &payload,174 )175 .map(|object| object.into_bytes())176 .map_err(Into::into)177 }178179 fn request_keyframe(&mut self) -> Option<u64> {180 self.cursor.request_keyframe();181 self.abort_current()182 }183184 fn abort_current(&mut self) -> Option<u64> {185 self.object_bytes = 0;186 let mut group = self.current_group.take()?;187 let group_id = group.sequence;188 let _ = group.abort(MoqError::Cancel);189 Some(group_id)190 }191192 fn abort(mut self) {193 self.abort_current();194 let _ = self.track.abort(MoqError::Cancel);195 }196}197198const MOQ_AUDIO_PACKETS_PER_GROUP: usize = 5;199const MAX_MOQ_AUDIO_GROUP_BYTES: usize =200 MOQ_AUDIO_PACKETS_PER_GROUP * (AUDIO_HEADER_LEN + MAX_AUDIO_PAYLOAD_LEN + 256);201202struct MoqAudioPublisher {203 track: TrackProducer,204 current_group: Option<GroupProducer>,205 object_count: usize,206 object_bytes: usize,207 generation_id: u64,208 signing_key: MediaGenerationSigningKey,209}210211impl MoqAudioPublisher {212 fn new(213 track: TrackProducer,214 generation_id: u64,215 signing_key: MediaGenerationSigningKey,216 ) -> Self {217 Self {218 track,219 current_group: None,220 object_count: 0,221 object_bytes: 0,222 generation_id,223 signing_key,224 }225 }226227 fn publish(&mut self, packet: crate::audio::EncodedAudioPacket) -> Result<()> {228 let flags = if packet.discontinuity {229 AudioFlags::DISCONTINUITY230 } else {231 AudioFlags::NONE232 };233 let payload = AudioPacket::new(234 AudioPacketHeader::opus(235 packet.payload.len(),236 packet.sequence,237 packet.capture_timestamp_us,238 packet.pts_us,239 flags,240 )?,241 packet.payload.as_ref().to_vec(),242 )?243 .encode_datagram()?;244245 if self.current_group.is_none()246 || self.object_count >= MOQ_AUDIO_PACKETS_PER_GROUP247 || packet.discontinuity248 {249 self.finish_group()?;250 self.current_group = Some(251 self.track252 .append_group()253 .context("creating bounded MoQ Opus group")?,254 );255 }256 let group = self257 .current_group258 .as_mut()259 .context("Opus packet has no active MoQ group")?;260 let object_id = u32::try_from(self.object_count).context("audio object id overflowed")?;261 let object = self.signing_key.authenticate(262 MediaObjectCoordinates {263 generation_id: self.generation_id,264 track: MediaTrack::AudioOpus,265 group_id: group.sequence,266 object_id,267 flags: u16::from(flags.bits()),268 },269 &payload,270 )?;271 let next_bytes = self272 .object_bytes273 .checked_add(object.as_bytes().len())274 .filter(|bytes| *bytes <= MAX_MOQ_AUDIO_GROUP_BYTES)275 .context("bounded MoQ Opus group exceeded its byte limit")?;276 group277 .write_frame(object.into_bytes())278 .context("writing authenticated Opus object")?;279 self.object_count += 1;280 self.object_bytes = next_bytes;281 if self.object_count == MOQ_AUDIO_PACKETS_PER_GROUP {282 self.finish_group()?;283 }284 Ok(())285 }286287 fn finish_group(&mut self) -> Result<()> {288 if let Some(mut group) = self.current_group.take() {289 group.finish().context("finishing bounded MoQ Opus group")?;290 }291 self.object_count = 0;292 self.object_bytes = 0;293 Ok(())294 }295296 fn abort(mut self) {297 if let Some(mut group) = self.current_group.take() {298 let _ = group.abort(MoqError::Cancel);299 }300 let _ = self.track.abort(MoqError::Cancel);301 }302}303304305pub(super) async fn run_generation_video_publisher(306 config: HostConfig,307 mut current_gop_receiver: tokio::sync::watch::Receiver<Option<EncodedGop>>,308 mut control_requests: tokio::sync::watch::Receiver<Option<MediaControlRequestV3>>,309 track: TrackProducer,310 generation_id: u64,311 signing_key: MediaGenerationSigningKey,312 encoder_control: Option<EncoderControl>,313 telemetry: Arc<MediaV3Telemetry>,314 mut shutdown: tokio::sync::watch::Receiver<bool>,315) -> Result<()> {316 let maximum_replay_age = maximum_media_replay_age(config.framerate);317 let mut replay_cursor = MediaReplayCursor::default();318 let mut publisher = MoqGroupPublisher::authenticated(track, generation_id, signing_key);319 let mut forced_idr = ForcedIdrCoordinator::new(encoder_control, Arc::clone(&telemetry));320 let log_identity = iroh::SecretKey::from_bytes(&[1; 32]).public();321322 let result = async {323 loop {324 tokio::select! {325 biased;326 changed = shutdown.changed() => {327 if changed.is_err() || *shutdown.borrow_and_update() {328 return Ok(());329 }330 }331 changed = control_requests.changed() => {332 if changed.is_err() {333 continue;334 }335 let Some(request) = *control_requests.borrow_and_update() else {336 continue;337 };338 let through_sequence = current_gop_receiver339 .borrow()340 .as_ref()341 .and_then(|gop| gop.frames.last())342 .map(|frame| frame.sequence);343 let cancelled_group = apply_moq_keyframe_request(344 &mut publisher,345 &mut replay_cursor,346 through_sequence,347 request.reason,348 );349 if cancelled_group.is_some() {350 telemetry.scheduler_cancellations.fetch_add(1, Ordering::Relaxed);351 }352 let disposition = forced_idr.request(request.reason);353 debug!(generation_id, request_id = request.request_id, ?disposition, ?cancelled_group, "shared generation accepted keyframe request");354 }355 acknowledgement = forced_idr.acknowledgements.join_next(),356 if forced_idr.pending_revision.is_some() =>357 {358 forced_idr.complete(acknowledgement, log_identity, "shared-iroh-moq");359 }360 changed = current_gop_receiver.changed() => {361 changed.context("shared encoded video source stopped")?;362 let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {363 continue;364 };365 publish_generation_gop(366 &config,367 &mut publisher,368 &mut replay_cursor,369 current_gop,370 maximum_replay_age,371 &telemetry,372 )?;373 }374 }375 }376 }377 .await;378 forced_idr379 .abort_and_drain(log_identity, "shared-iroh-moq")380 .await;381 publisher.abort();382 result383}384385fn publish_generation_gop(386 config: &HostConfig,387 publisher: &mut MoqGroupPublisher,388 replay_cursor: &mut MediaReplayCursor,389 current_gop: EncodedGop,390 maximum_replay_age: Duration,391 telemetry: &MediaV3Telemetry,392) -> Result<()> {393 let initial_replay_started_at = replay_cursor.last_sequence.is_none().then(Instant::now);394 let replay_through_sequence = current_gop395 .frames396 .last()397 .map(|frame| frame.sequence)398 .context("shared current GOP snapshot is empty")?;399 for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {400 let replay_discontinuity = match replay_cursor.classify(401 &frame,402 replay_through_sequence,403 initial_replay_started_at,404 Instant::now(),405 maximum_replay_age,406 ) {407 MediaReplayDecision::Send { discontinuity } => discontinuity,408 MediaReplayDecision::SkipUntilKeyframe409 | MediaReplayDecision::DiscardStaleSuffix { .. } => {410 if publisher.request_keyframe().is_some() {411 telemetry412 .scheduler_cancellations413 .fetch_add(1, Ordering::Relaxed);414 }415 replay_cursor.enter_resync_through(Some(replay_through_sequence));416 break;417 }418 };419 match publisher420 .publish(config, &frame, replay_discontinuity)421 .inspect_err(|_| {422 telemetry.send_failures.fetch_add(1, Ordering::Relaxed);423 })? {424 MoqGroupDecision::Published {425 cancelled_previous, ..426 } => {427 if cancelled_previous {428 telemetry429 .scheduler_cancellations430 .fetch_add(1, Ordering::Relaxed);431 }432 replay_cursor.commit_sent(&frame);433 }434 MoqGroupDecision::SkipUntilKeyframe | MoqGroupDecision::EnterResync => {435 replay_cursor.enter_resync_through(Some(replay_through_sequence));436 break;437 }438 }439 }440 Ok(())441}442443pub(super) async fn run_generation_audio_publisher(444 mut packets: tokio::sync::mpsc::Receiver<crate::audio::EncodedAudioPacket>,445 track: TrackProducer,446 generation_id: u64,447 signing_key: MediaGenerationSigningKey,448 mut shutdown: tokio::sync::watch::Receiver<bool>,449) -> Result<()> {450 let mut publisher = MoqAudioPublisher::new(track, generation_id, signing_key);451 let result = async {452 loop {453 tokio::select! {454 changed = shutdown.changed() => {455 if changed.is_err() || *shutdown.borrow_and_update() {456 return Ok(());457 }458 }459 packet = packets.recv() => {460 let packet = packet.context("shared encoded Opus source stopped")?;461 publisher.publish(packet)?;462 }463 }464 }465 }466 .await;467 publisher.abort();468 result469}470471fn apply_moq_keyframe_request(472 publisher: &mut MoqGroupPublisher,473 replay_cursor: &mut MediaReplayCursor,474 through_sequence: Option<u64>,475 reason: KeyframeRequestReasonV3,476) -> Option<u64> {477 // The bounded current group is already the late joiner's decodable replay.478 // Aborting it on Join can strand a static source until its next natural IDR.479 if reason == KeyframeRequestReasonV3::Join {480 return None;481 }482 let cancelled_group = publisher.request_keyframe();483 replay_cursor.enter_resync_through(through_sequence);484 cancelled_group485}486487pub(super) async fn serve_authorized_moq(488 connection: Connection,489 origin: Origin,490 attachment: ClaimedMoqAttachment,491) -> Result<()> {492 let ClaimedMoqAttachment {493 session_id,494 broadcast_name,495 broadcast,496 attached,497 closed,498 telemetry,499 } = attachment;500 let result: Result<()> = async {501 let web_transport = web_transport_iroh::Session::raw(connection);502 let session = tokio::time::timeout(503 MOQ_ATTACHMENT_TIMEOUT,504 iroh_moq::MoqSession::session_accept(web_transport, origin),505 )506 .await507 .context("timed out completing authorized MoQ handshake")?508 .context("completing authorized MoQ handshake")?;509 let broadcast_closed = broadcast.clone();510 session.publish(&broadcast_name, broadcast);511 ensure!(512 attached.send(()).is_ok(),513 "control session ended before MoQ attachment completed"514 );515 info!(516 remote = %session.remote_id(),517 session_id,518 %broadcast_name,519 track = MOQ_VIDEO_H264_TRACK,520 "authorized MoQ media attachment accepted"521 );522 let mut telemetry_interval = tokio::time::interval(Duration::from_secs(1));523 telemetry_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);524 loop {525 tokio::select! {526 reason = session.closed() => {527 debug!(remote = %session.remote_id(), ?reason, "MoQ media session closed");528 break;529 }530 reason = broadcast_closed.closed() => {531 debug!(remote = %session.remote_id(), ?reason, "control-owned MoQ broadcast closed");532 session.close(0, b"control session ended");533 break;534 }535 _ = telemetry_interval.tick() => {536 telemetry.record_selected_path(session.conn());537 }538 }539 }540 Ok(())541 }542 .await;543 let _ = closed.send(());544 result545}546547pub(super) async fn serve_control_moq(548 connection: Connection,549 config: HostConfig,550 sessions: &Arc<SessionRegistry>,551 authorization: &AuthorizationPolicy,552) -> Result<()> {553 let remote = connection.remote_id();554 let handshake_permit = sessions555 .pending_handshakes556 .try_acquire()557 .context("too many pending handshakes")?;558 let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())559 .await560 .context("timed out accepting MoQ control stream")?561 .context("accepting MoQ control stream")?;562 let hello = receive_hello(&mut recv, Capability::VideoH264).await?;563 drop(handshake_permit);564 debug!(%remote, agent = %hello.agent, "MoQ control hello received");565566 let grants = match authorization.authorize_or_redeem(567 remote,568 hello.invitation.as_deref(),569 unix_timestamp_now()?,570 ) {571 Ok(grants) => grants,572 Err(error) => {573 send_rejection(&mut send, "Portal peer is not authorized").await?;574 return Err(error.context("authorizing MoQ control peer"));575 }576 };577 ensure!(578 grants.contains(InvitationGrants::VIEW),579 "authorized MoQ control peer lacks view permission"580 );581 let lease = match sessions.claim(remote, hello.nonce, grants) {582 Ok(lease) => lease,583 Err(error) => {584 send_rejection(&mut send, "host already has an active client").await?;585 return Err(error);586 }587 };588589 let source = match config.source {590 VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),591 VideoSource::GamescopePipewire => {592 let primary = spawn_gamescope_pipewire_after_static_preflight(593 config.clone(),594 lease.session_clock,595 )596 .await?;597 select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await598 }599 };600 let EncodedSource {601 frames: frame_receiver,602 current_gop: mut current_gop_receiver,603 task: source_task,604 pointer_surface_dimensions,605 encoder_control,606 } = match source {607 Ok(source) => source,608 Err(error) => {609 send_rejection(&mut send, "video source is unavailable").await?;610 return Err(error);611 }612 };613 let source_task = SourceTaskGuard::new(source_task);614 sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;615616 let mut broadcast = Broadcast::new().produce();617 let track = broadcast618 .create_track(Track {619 name: MOQ_VIDEO_H264_TRACK.to_owned(),620 priority: MOQ_VIDEO_TRACK_PRIORITY,621 })622 .context("creating static MoQ H.264 track")?;623 let catalog = publish_goq_catalog(&mut broadcast)?;624 let broadcast_name = media_moq_broadcast_name(lease.session_id)?;625 let attachment = sessions.expect_moq(626 remote,627 lease.session_id,628 broadcast_name.clone(),629 broadcast.consume(),630 )?;631632 let mut control_hello = HostHello::accepted(633 lease.session_id,634 negotiated_capabilities(&hello, MEDIA_CAPABILITIES),635 );636 if let Some(dimensions) = pointer_surface_dimensions {637 control_hello = control_hello.with_pointer_surface_dimensions(dimensions);638 }639 write_host_hello(&mut send, &control_hello).await?;640 send.finish().context("finishing MoQ control response")?;641 drop(send);642 info!(643 %remote,644 session_id = lease.session_id,645 %broadcast_name,646 "MoQ control client accepted; awaiting authorized media attachment"647 );648649 let MoqAttachmentWait {650 mut attached,651 closed,652 } = attachment;653 tokio::time::timeout(MOQ_ATTACHMENT_TIMEOUT, async {654 tokio::select! {655 result = &mut attached => {656 result.context("authorized MoQ handler ended before attachment")657 }658 reason = connection.closed() => {659 Err(anyhow::anyhow!("control connection closed before MoQ attachment: {reason:?}"))660 }661 }662 })663 .await664 .context("timed out waiting for authorized MoQ attachment")??;665666 let session_result = run_control_moq_session(667 &connection,668 &config,669 &mut current_gop_receiver,670 MoqControlReader::V1(recv),671 remote,672 closed,673 track,674 &mut broadcast,675 encoder_control,676 Arc::clone(&lease.media_v3_telemetry),677 None,678 )679 .await;680 let catalog_result = catalog.finish();681682 drop(current_gop_receiver);683 drop(frame_receiver);684 source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;685 drop(lease);686 info!(%remote, "MoQ control client released");687 match session_result {688 Err(error) => Err(error),689 Ok(()) => catalog_result,690 }691}692693pub(super) async fn serve_control_moq_v2(694 connection: Connection,695 sessions: &Arc<SessionRegistry>,696 generations: &Arc<MediaGenerationManager>,697 authorization: &AuthorizationPolicy,698 input_operations: &Arc<InputOperations>,699 host_secret: [u8; 32],700) -> Result<()> {701 let remote = connection.remote_id();702 let handshake_permit = sessions703 .pending_handshakes704 .try_acquire()705 .context("too many pending handshakes")?;706 let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())707 .await708 .context("timed out accepting MoQ control v2 stream")?709 .context("accepting MoQ control v2 stream")?;710 let hello = tokio::time::timeout(HANDSHAKE_TIMEOUT, read_client_hello_v2(&mut recv))711 .await712 .context("timed out waiting for control v2 hello")??713 .context("client closed before control v2 hello")?;714 drop(handshake_permit);715 ensure!(716 hello.capabilities.contains(&Capability::VideoH264),717 "client did not offer required capability VideoH264"718 );719 debug!(%remote, agent = %hello.agent, "MoQ control v2 hello received");720721 let authorized = match authorization.authorize_or_redeem_viewer(722 remote,723 hello.invitation.as_deref(),724 unix_timestamp_now()?,725 ) {726 Ok(authorized) => authorized,727 Err(error) => {728 send_rejection_v2(&mut send, "Portal peer is not authorized").await?;729 return Err(error.context("authorizing MoQ control v2 peer"));730 }731 };732 ensure!(733 authorized.grants.contains(InvitationGrants::VIEW),734 "authorized MoQ control v2 peer lacks view permission"735 );736 let mut lease = match sessions.claim_v2_authorized(remote, hello.nonce, authorized) {737 Ok(lease) => lease,738 Err(error) => {739 send_rejection_v2(&mut send, error.to_string()).await?;740 return Err(error);741 }742 };743744 let generation = match generations.acquire().await {745 Ok(generation) => generation,746 Err(error) => {747 send_rejection_v2(&mut send, "shared media generation is unavailable").await?;748 return Err(error);749 }750 };751 let shared = Arc::clone(&generation.shared);752 let mut initial_snapshot = sessions.bind_v2_generation(753 remote,754 lease.session_id,755 shared.generation_id,756 shared.session_clock,757 Arc::clone(&shared.telemetry),758 shared.encoder_control.clone(),759 )?;760 if let Some(transition) = lease.replacement_transition {761 input_operations.neutralize_focus_transition(762 sessions,763 transition.transition_id,764 FocusTransitionReasonV2::Replaced,765 )?;766 initial_snapshot = sessions767 .subscribe_v2_snapshots(remote, lease.session_id)?768 .borrow()769 .clone()770 .context("replacement completion did not publish a session snapshot")?;771 }772 // Keep the predecessor's generation lease alive until this viewer has773 // acquired the shared generation, then retire only that replaced control774 // stream. Survivors and the producer remain untouched.775 lease.retire_replaced_viewer();776 let issued_at_unix = unix_timestamp_now()?;777 let mut subscription_nonce = [0_u8; 32];778 getrandom::fill(&mut subscription_nonce).context("generating subscription capability nonce")?;779 let subscription_tracks = if shared.audio_enabled {780 SubscriptionTracks::ALL781 } else {782 SubscriptionTracks::VIDEO_H264783 };784 let subscription_capability = SignedSubscriptionCapability::issue(785 SubscriptionClaims::new(786 shared.certificate.claims.host_node_id,787 shared.generation_id,788 *remote.as_bytes(),789 subscription_tracks,790 lease.authorization_revision,791 issued_at_unix,792 issued_at_unix.saturating_add(15 * 60),793 subscription_nonce,794 1,795 )?,796 &host_secret,797 )?;798 let subscription_token = subscription_capability.encode();799800 let broadcast_name = shared.broadcast_name.clone();801 let attachment = sessions.expect_moq_v2(802 remote,803 lease.session_id,804 broadcast_name.clone(),805 shared.consumer()?,806 subscription_capability,807 shared.certificate.claims.host_node_id,808 )?;809810 let mut supported = vec![Capability::VideoH264];811 if shared.audio_enabled {812 supported.push(Capability::AudioOpus);813 }814 let negotiated = supported815 .iter()816 .copied()817 .filter(|capability| hello.capabilities.contains(capability))818 .collect();819 let mut control_hello = HostHelloV2::accepted(lease.session_id, negotiated, initial_snapshot)820 .with_media_subscription_capability(subscription_token, lease.authorization_revision);821 if let Some(dimensions) = shared.pointer_surface_dimensions {822 control_hello = control_hello.with_pointer_surface_dimensions(dimensions);823 }824 write_host_hello_v2(&mut send, &control_hello).await?;825 info!(826 %remote,827 session_id = lease.session_id,828 authorization_revision = lease.authorization_revision,829 authorization_committed_revision = lease.authorization_committed_revision,830 %broadcast_name,831 "MoQ control v2 client accepted; awaiting authorized media attachment"832 );833834 let MoqAttachmentWait {835 mut attached,836 closed,837 } = attachment;838 tokio::time::timeout(MOQ_ATTACHMENT_TIMEOUT, async {839 tokio::select! {840 result = &mut attached => result.context("authorized MoQ handler ended before v2 attachment"),841 reason = connection.closed() => Err(anyhow::anyhow!("control v2 connection closed before MoQ attachment: {reason:?}")),842 }843 })844 .await845 .context("timed out waiting for authorized MoQ v2 attachment")??;846847 let session_result = run_generation_control_session(848 &connection,849 send,850 recv,851 Arc::clone(sessions),852 Arc::clone(input_operations),853 remote,854 lease.session_id,855 closed,856 shared.keyframe_requests.clone(),857 )858 .await;859860 if let Some(transition) = sessions.invalidate_v2_focus(861 remote,862 lease.session_id,863 FocusTransitionReasonV2::Disconnected,864 )? {865 input_operations.neutralize_focus_transition(866 sessions,867 transition.transition_id,868 FocusTransitionReasonV2::Disconnected,869 )?;870 }871 drop(lease);872 generation.release().await;873 info!(%remote, "MoQ control v2 client released");874 session_result875}876877878async fn run_generation_control_session(879 connection: &Connection,880 send: iroh::endpoint::SendStream,881 recv: iroh::endpoint::RecvStream,882 sessions: Arc<SessionRegistry>,883 input_operations: Arc<InputOperations>,884 remote: EndpointId,885 session_id: u64,886 mut moq_closed: tokio::sync::oneshot::Receiver<()>,887 keyframes: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,888) -> Result<()> {889 let mut control = tokio::spawn(forward_control_v2_requests(890 send,891 recv,892 sessions,893 input_operations,894 remote,895 session_id,896 keyframes,897 ));898 tokio::select! {899 result = &mut control => result.context("shared-generation control task failed")?,900 reason = connection.closed() => {901 debug!(%remote, ?reason, "shared-generation control connection closed");902 Ok(())903 }904 result = &mut moq_closed => {905 debug!(%remote, ?result, "shared-generation MoQ attachment closed");906 Ok(())907 }908 }909}910911912enum MoqControlReader {913 V1(iroh::endpoint::RecvStream),914 V2 {915 send: iroh::endpoint::SendStream,916 recv: iroh::endpoint::RecvStream,917 sessions: Arc<SessionRegistry>,918 input_operations: Arc<InputOperations>,919 session_id: u64,920 },921}922923async fn forward_control_v2_requests(924 mut send: iroh::endpoint::SendStream,925 mut recv: iroh::endpoint::RecvStream,926 sessions: Arc<SessionRegistry>,927 input_operations: Arc<InputOperations>,928 remote: EndpointId,929 session_id: u64,930 keyframes: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,931) -> Result<()> {932 let mut snapshots = sessions.subscribe_v2_snapshots(remote, session_id)?;933 let mut focus_deadlines = tokio::time::interval(Duration::from_millis(250));934 focus_deadlines.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);935 loop {936 tokio::select! {937 _ = focus_deadlines.tick() => {938 if let Some(transition) = sessions.expire_v2_focus()? {939 input_operations.neutralize_focus_transition(940 &sessions,941 transition.transition_id,942 FocusTransitionReasonV2::ActivationExpired,943 )?;944 }945 }946 changed = snapshots.changed() => {947 changed.context("v2 session snapshot publisher stopped")?;948 let Some(snapshot) = snapshots.borrow_and_update().clone() else {949 break;950 };951 write_server_control_v2(952 &mut send,953 &ServerControlEnvelopeV2::Snapshot { snapshot },954 )955 .await?;956 }957 command = read_client_control_v2(&mut recv) => {958 let Some(command) = command? else { break; };959 match command {960 ClientControlEnvelopeV2::Focus { command } => {961 let request_id = command.request_id;962 let outcome = sessions.apply_focus_command(remote, session_id, &command);963 let result = match outcome {964 Ok(effect) => {965 if let Some(transition) = effect.neutralization {966 input_operations.neutralize_focus_transition(967 &sessions,968 transition.transition_id,969 effect.snapshot.transition_reason,970 )?;971 }972 FocusCommandResultV2 {973 request_id,974 accepted: true,975 revision: sessions.v2_revision(remote, session_id)?,976 message: None,977 }978 }979 Err(error) => FocusCommandResultV2 {980 request_id,981 accepted: false,982 revision: sessions.v2_revision(remote, session_id)?,983 message: Some(error.to_string()),984 },985 };986 write_server_control_v2(987 &mut send,988 &ServerControlEnvelopeV2::FocusResult { result },989 )990 .await?;991 }992 ClientControlEnvelopeV2::Keyframe {993 request_id,994 last_sequence,995 reason,996 } => {997 sessions.admit_v2_keyframe_request(remote, session_id)?;998 keyframes.send_replace(Some(MediaControlRequestV3::request_keyframe(999 request_id,1000 last_sequence,1001 reason.into(),1002 )));1003 }1004 }1005 }1006 }1007 }1008 Ok(())1009}101010111012async fn run_control_moq_session(1013 connection: &Connection,1014 config: &HostConfig,1015 current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,1016 control_reader: MoqControlReader,1017 remote: EndpointId,1018 mut moq_closed: tokio::sync::oneshot::Receiver<()>,1019 track: TrackProducer,1020 broadcast: &mut BroadcastProducer,1021 encoder_control: Option<EncoderControl>,1022 telemetry: Arc<MediaV3Telemetry>,1023 authentication: Option<(u64, MediaGenerationSigningKey)>,1024) -> Result<()> {1025 let maximum_replay_age = maximum_media_replay_age(config.framerate);1026 let mut replay_cursor = MediaReplayCursor::default();1027 let mut publisher = match authentication {1028 Some((generation_id, signing_key)) => {1029 MoqGroupPublisher::authenticated(track, generation_id, signing_key)1030 }1031 None => MoqGroupPublisher::new(track),1032 };1033 let (control_sender, mut control_requests) = tokio::sync::watch::channel(None);1034 let terminate_when_control_ends = matches!(&control_reader, MoqControlReader::V2 { .. });1035 let mut control_task = match control_reader {1036 MoqControlReader::V1(control_recv) => tokio::spawn(forward_media_v3_control_requests(1037 control_recv,1038 control_sender,1039 )),1040 MoqControlReader::V2 {1041 send,1042 recv,1043 sessions,1044 input_operations,1045 session_id,1046 } => tokio::spawn(forward_control_v2_requests(1047 send,1048 recv,1049 sessions,1050 input_operations,1051 remote,1052 session_id,1053 control_sender,1054 )),1055 };1056 let mut control_task_finished = false;1057 let mut control_receiver_open = true;1058 let mut forced_idr = ForcedIdrCoordinator::new(encoder_control, Arc::clone(&telemetry));10591060 let result = async {1061 loop {1062 tokio::select! {1063 biased;1064 control_result = &mut control_task, if !control_task_finished => {1065 control_task_finished = true;1066 match control_result {1067 Ok(Ok(())) => {1068 debug!(%remote, "MoQ keyframe-control stream closed");1069 if terminate_when_control_ends {1070 return Ok(());1071 }1072 }1073 Ok(Err(error)) => {1074 return Err(error).context("reading MoQ keyframe-control stream");1075 }1076 Err(error) => {1077 return Err(error).context("MoQ keyframe-control task failed");1078 }1079 }1080 }1081 changed = control_requests.changed(), if control_receiver_open => {1082 if changed.is_err() {1083 control_receiver_open = false;1084 continue;1085 }1086 let Some(request) = *control_requests.borrow_and_update() else {1087 continue;1088 };1089 let through_sequence = current_gop_receiver1090 .borrow()1091 .as_ref()1092 .and_then(|gop| gop.frames.last())1093 .map(|frame| frame.sequence);1094 let cancelled_group = apply_moq_keyframe_request(1095 &mut publisher,1096 &mut replay_cursor,1097 through_sequence,1098 request.reason,1099 );1100 if cancelled_group.is_some() {1101 telemetry1102 .scheduler_cancellations1103 .fetch_add(1, Ordering::Relaxed);1104 }1105 let forced_idr_disposition = forced_idr.request(request.reason);1106 if let ForcedIdrDisposition::Failed { error } = &forced_idr_disposition {1107 warn!(1108 %remote,1109 request_id = request.request_id,1110 ?request.reason,1111 %error,1112 "forced-IDR request failed; retaining natural-IDR fallback"1113 );1114 }1115 debug!(1116 %remote,1117 request_id = request.request_id,1118 ?request.reason,1119 advisory_last_sequence = ?request.last_sequence,1120 coalesced = cancelled_group.is_none(),1121 ?cancelled_group,1122 ?forced_idr_disposition,1123 "accepted MoQ keyframe request"1124 );1125 }1126 acknowledgement = forced_idr.acknowledgements.join_next(),1127 if forced_idr.pending_revision.is_some() =>1128 {1129 forced_idr.complete(acknowledgement, remote, "iroh-moq");1130 }1131 reason = connection.closed() => {1132 debug!(%remote, ?reason, "MoQ control connection closed");1133 return Ok(());1134 }1135 result = &mut moq_closed => {1136 debug!(%remote, ?result, "authorized MoQ media attachment closed");1137 return Ok(());1138 }1139 changed = current_gop_receiver.changed() => {1140 if let Err(error) = changed {1141 return Err(error).context("encoded source stopped");1142 }1143 let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {1144 continue;1145 };1146 let initial_replay_started_at =1147 replay_cursor.last_sequence.is_none().then(Instant::now);1148 let replay_through_sequence = current_gop1149 .frames1150 .last()1151 .map(|frame| frame.sequence)1152 .context("current GOP snapshot is empty")?;1153 for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {1154 let replay_discontinuity = match replay_cursor.classify(1155 &frame,1156 replay_through_sequence,1157 initial_replay_started_at,1158 Instant::now(),1159 maximum_replay_age,1160 ) {1161 MediaReplayDecision::Send { discontinuity } => discontinuity,1162 MediaReplayDecision::SkipUntilKeyframe => {1163 if publisher.request_keyframe().is_some() {1164 telemetry1165 .scheduler_cancellations1166 .fetch_add(1, Ordering::Relaxed);1167 }1168 replay_cursor.enter_resync_through(Some(replay_through_sequence));1169 break;1170 }1171 MediaReplayDecision::DiscardStaleSuffix { through_sequence } => {1172 let cancelled_group = publisher.request_keyframe();1173 if cancelled_group.is_some() {1174 telemetry1175 .scheduler_cancellations1176 .fetch_add(1, Ordering::Relaxed);1177 }1178 debug!(1179 %remote,1180 through_sequence,1181 ?cancelled_group,1182 "cancelled stale MoQ media suffix"1183 );1184 break;1185 }1186 };1187 let decision = publisher1188 .publish(config, &frame, replay_discontinuity)1189 .inspect_err(|_error| {1190 telemetry.send_failures.fetch_add(1, Ordering::Relaxed);1191 })?;1192 match decision {1193 MoqGroupDecision::Published {1194 group_id,1195 frame_id,1196 cancelled_previous,1197 } => {1198 if cancelled_previous {1199 telemetry1200 .scheduler_cancellations1201 .fetch_add(1, Ordering::Relaxed);1202 }1203 debug!(1204 sequence = frame.sequence,1205 group_id,1206 frame_id,1207 cancelled_previous,1208 "published upstream MoQ video frame"1209 );1210 replay_cursor.commit_sent(&frame);1211 }1212 MoqGroupDecision::SkipUntilKeyframe => {1213 replay_cursor.enter_resync_through(Some(replay_through_sequence));1214 break;1215 }1216 MoqGroupDecision::EnterResync => {1217 replay_cursor.enter_resync_through(Some(replay_through_sequence));1218 break;1219 }1220 }1221 }1222 }1223 }1224 }1225 }1226 .await;12271228 forced_idr.abort_and_drain(remote, "iroh-moq").await;1229 publisher.abort();1230 let _ = broadcast.abort(MoqError::Cancel);1231 if !control_task_finished {1232 control_task.abort();1233 let _ = control_task.await;1234 }1235 result1236}12371238pub(super) fn media_frame_for_encoded(1239 _config: &HostConfig,1240 frame: &EncodedFrame,1241 discontinuity: bool,1242) -> Result<MediaFrame> {1243 let mut flags = FrameFlags::NONE;1244 if frame.keyframe {1245 flags = flags.union(FrameFlags::KEYFRAME);1246 }1247 if frame.codec_config {1248 flags = flags.union(FrameFlags::CODEC_CONFIG);1249 }1250 if discontinuity || frame.discontinuity {1251 flags = flags.union(FrameFlags::DISCONTINUITY);1252 }1253 let header = MediaFrameHeader::h264(1254 frame.width,1255 frame.height,1256 frame.data.len(),1257 frame.sequence,1258 frame.capture_timestamp_micros,1259 frame.presentation_timestamp_micros,1260 flags,1261 )?;1262 MediaFrame::new(header, frame.data.clone()).map_err(Into::into)1263}126412651266mod tests {1267 use super::super::{media_v3_encoded_frame, moq_test_config};1268 use super::*;12691270 fn authenticated_publisher(1271 track: TrackProducer,1272 ) -> (1273 MoqGroupPublisher,1274 sigil_protocol::SignedMediaGenerationCertificate,1275 ) {1276 let host = iroh::SecretKey::from_bytes(&[7; 32]);1277 let signing_key = MediaGenerationSigningKey::from_bytes(&[9; 32]);1278 let certificate = signing_key1279 .certify(1280 *host.public().as_bytes(),1281 &host.to_bytes(),1282 42,1283 1_700_000_000,1284 1_700_000_600,1285 )1286 .unwrap();1287 (1288 MoqGroupPublisher::authenticated(track, 42, signing_key),1289 certificate,1290 )1291 }12921293 1294 async fn upstream_moq_groups_are_sequential_and_cancel_the_superseded_gop() {1295 let track_info = Track {1296 name: MOQ_VIDEO_H264_TRACK.to_owned(),1297 priority: MOQ_VIDEO_TRACK_PRIORITY,1298 };1299 let mut broadcast = Broadcast::new().produce();1300 let track = broadcast.create_track(track_info.clone()).unwrap();1301 let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();1302 let mut publisher = MoqGroupPublisher::new(track);1303 let config = moq_test_config();13041305 assert_eq!(1306 publisher1307 .publish(&config, &media_v3_encoded_frame(100, true, true, 4), false)1308 .unwrap(),1309 MoqGroupDecision::Published {1310 group_id: 0,1311 frame_id: 0,1312 cancelled_previous: false,1313 }1314 );1315 let mut first_group = consumer.recv_group().await.unwrap().unwrap();1316 assert_eq!(first_group.sequence, 0);1317 assert!(first_group.read_frame().await.unwrap().is_some());13181319 assert_eq!(1320 publisher1321 .publish(1322 &config,1323 &media_v3_encoded_frame(101, false, false, 4),1324 false,1325 )1326 .unwrap(),1327 MoqGroupDecision::Published {1328 group_id: 0,1329 frame_id: 1,1330 cancelled_previous: false,1331 }1332 );1333 assert!(first_group.read_frame().await.unwrap().is_some());13341335 assert_eq!(1336 publisher1337 .publish(&config, &media_v3_encoded_frame(200, true, true, 4), false)1338 .unwrap(),1339 MoqGroupDecision::Published {1340 group_id: 1,1341 frame_id: 0,1342 cancelled_previous: true,1343 }1344 );1345 assert!(first_group.finished().await.is_err());1346 let mut second_group = consumer.recv_group().await.unwrap().unwrap();1347 assert_eq!(second_group.sequence, 1);1348 let object = second_group.read_frame().await.unwrap().unwrap();1349 let frame = sigil_protocol::decode_media_frame_object(&object).unwrap();1350 assert_eq!(frame.header.sequence, 200);1351 assert!(frame.header.flags.contains(FrameFlags::DISCONTINUITY));1352 }13531354 1355 async fn control_v2_objects_are_signed_after_final_moq_coordinates_are_known() {1356 let track_info = Track {1357 name: MOQ_VIDEO_H264_TRACK.to_owned(),1358 priority: MOQ_VIDEO_TRACK_PRIORITY,1359 };1360 let mut broadcast = Broadcast::new().produce();1361 let track = broadcast.create_track(track_info.clone()).unwrap();1362 let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();1363 let (mut publisher, certificate) = authenticated_publisher(track);1364 let config = moq_test_config();1365 let encoded = media_v3_encoded_frame(100, true, true, 4);1366 publisher.publish(&config, &encoded, false).unwrap();1367 let mut group = consumer.recv_group().await.unwrap().unwrap();1368 let object = group.read_frame().await.unwrap().unwrap();1369 let payload = sigil_protocol::AuthenticatedMediaObject::verify(1370 &object,1371 &certificate,1372 MediaObjectCoordinates {1373 generation_id: 42,1374 track: MediaTrack::VideoH264,1375 group_id: group.sequence,1376 object_id: 0,1377 flags: u16::from(FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG).bits()),1378 },1379 )1380 .unwrap();1381 assert_eq!(1382 sigil_protocol::decode_media_frame_object(payload)1383 .unwrap()1384 .header1385 .sequence,1386 1001387 );1388 }13891390 1391 async fn upstream_moq_late_join_preserves_active_static_group() {1392 let track_info = Track {1393 name: MOQ_VIDEO_H264_TRACK.to_owned(),1394 priority: MOQ_VIDEO_TRACK_PRIORITY,1395 };1396 let mut broadcast = Broadcast::new().produce();1397 let track = broadcast.create_track(track_info.clone()).unwrap();1398 let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();1399 let mut publisher = MoqGroupPublisher::new(track);1400 let mut replay_cursor = MediaReplayCursor::default();1401 let config = moq_test_config();1402 let keyframe = media_v3_encoded_frame(10, true, true, 1);1403 let first_delta = media_v3_encoded_frame(11, false, false, 1);1404 let next_delta = media_v3_encoded_frame(12, false, false, 1);14051406 publisher.publish(&config, &keyframe, false).unwrap();1407 replay_cursor.commit_sent(&keyframe);1408 let mut active_group = consumer.recv_group().await.unwrap().unwrap();1409 assert!(active_group.read_frame().await.unwrap().is_some());14101411 publisher.publish(&config, &first_delta, false).unwrap();1412 replay_cursor.commit_sent(&first_delta);1413 assert!(active_group.read_frame().await.unwrap().is_some());14141415 assert_eq!(1416 apply_moq_keyframe_request(1417 &mut publisher,1418 &mut replay_cursor,1419 Some(first_delta.sequence),1420 KeyframeRequestReasonV3::Join,1421 ),1422 None1423 );1424 assert_eq!(replay_cursor.last_sequence, Some(first_delta.sequence));1425 assert!(!replay_cursor.waiting_for_keyframe);1426 assert_eq!(1427 publisher.publish(&config, &next_delta, false).unwrap(),1428 MoqGroupDecision::Published {1429 group_id: 0,1430 frame_id: 2,1431 cancelled_previous: false,1432 }1433 );1434 assert!(active_group.read_frame().await.unwrap().is_some());1435 }14361437 1438 async fn upstream_moq_resync_aborts_current_group_and_waits_for_configured_idr() {1439 let track_info = Track {1440 name: MOQ_VIDEO_H264_TRACK.to_owned(),1441 priority: MOQ_VIDEO_TRACK_PRIORITY,1442 };1443 let mut broadcast = Broadcast::new().produce();1444 let track = broadcast.create_track(track_info.clone()).unwrap();1445 let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();1446 let mut publisher = MoqGroupPublisher::new(track);1447 let config = moq_test_config();14481449 publisher1450 .publish(&config, &media_v3_encoded_frame(10, true, true, 1), false)1451 .unwrap();1452 let mut cancelled = consumer.recv_group().await.unwrap().unwrap();1453 assert_eq!(publisher.request_keyframe(), Some(0));1454 assert!(cancelled.finished().await.is_err());1455 assert_eq!(1456 publisher1457 .publish(&config, &media_v3_encoded_frame(11, false, false, 1), false,)1458 .unwrap(),1459 MoqGroupDecision::SkipUntilKeyframe1460 );1461 assert_eq!(1462 publisher1463 .publish(&config, &media_v3_encoded_frame(20, true, true, 1), false)1464 .unwrap(),1465 MoqGroupDecision::Published {1466 group_id: 1,1467 frame_id: 0,1468 cancelled_previous: false,1469 }1470 );1471 }14721473 1474 async fn upstream_moq_group_counts_envelope_bytes_before_upstream_cache_eviction() {1475 let track_info = Track {1476 name: MOQ_VIDEO_H264_TRACK.to_owned(),1477 priority: MOQ_VIDEO_TRACK_PRIORITY,1478 };1479 let mut broadcast = Broadcast::new().produce();1480 let track = broadcast.create_track(track_info.clone()).unwrap();1481 let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();1482 let mut publisher = MoqGroupPublisher::new(track);1483 let config = moq_test_config();1484 publisher1485 .publish(&config, &media_v3_encoded_frame(10, true, true, 1), false)1486 .unwrap();1487 let mut cancelled = consumer.recv_group().await.unwrap().unwrap();14881489 // Payload-only accounting would accept this next one-byte access unit,1490 // but its fixed application envelope would overflow moq-net's 32 MiB1491 // group cache and silently evict the keyframe.1492 publisher.object_bytes = MAX_MEDIA_GROUP_BYTES_V3 - 1;1493 assert_eq!(1494 publisher1495 .publish(&config, &media_v3_encoded_frame(11, false, false, 1), false,)1496 .unwrap(),1497 MoqGroupDecision::EnterResync1498 );1499 assert!(cancelled.finished().await.is_err());1500 assert_eq!(publisher.object_bytes, 0);1501 }1502}