1use super::*;23const MEDIA_V3_IN_FLIGHT_CAPACITY: usize = 4;4const MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY: u8 = 0;5const MEDIA_V3_DELTA_PUBLISHER_PRIORITY: u8 = 128;6const MEDIA_V3_KEYFRAME_TRANSPORT_PRIORITY: i32 = 10;7const MEDIA_V3_DELTA_TRANSPORT_PRIORITY: i32 = 0;8// Wire-visible application reset code. Keep stable across compatible v3 releases.9const MEDIA_V3_RESET_CODE: u32 = 0x5357;10const MEDIA_V3_DELIVERY_FRAME_PERIODS: u64 = 4;11const MEDIA_V3_MAX_CONTROL_REQUESTS_PER_SECOND: u32 = 10;12const MEDIA_V3_CONTROL_REQUEST_INTERVAL: Duration =13 Duration::from_millis(1_000 / MEDIA_V3_MAX_CONTROL_REQUESTS_PER_SECOND as u64);141516pub(super) struct MediaV3GroupCursor {17 group_id: Option<u64>,18 last_sequence: Option<u64>,19 last_object_id: Option<u32>,20 payload_bytes: usize,21 waiting_for_keyframe: bool,22 discontinuity_pending: bool,23}2425impl Default for MediaV3GroupCursor {26 fn default() -> Self {27 Self {28 group_id: None,29 last_sequence: None,30 last_object_id: None,31 payload_bytes: 0,32 waiting_for_keyframe: true,33 discontinuity_pending: false,34 }35 }36}3738impl MediaV3GroupCursor {39 pub(super) fn classify(&mut self, frame: &EncodedFrame) -> MediaV3GroupDecision {40 let independently_decodable = frame.keyframe && frame.codec_config;41 if independently_decodable {42 if frame.data.len() > MAX_MEDIA_GROUP_BYTES_V3 {43 self.enter_resync();44 return MediaV3GroupDecision::EnterResync;45 }46 let sequence_discontinuity = self47 .last_sequence48 .is_some_and(|last| last.checked_add(1) != Some(frame.sequence));49 let position = MediaV3ObjectPosition {50 group_id: frame.sequence,51 object_id: 0,52 discontinuity: self.discontinuity_pending || sequence_discontinuity,53 };54 self.group_id = Some(frame.sequence);55 self.last_sequence = Some(frame.sequence);56 self.last_object_id = Some(0);57 self.payload_bytes = frame.data.len();58 self.waiting_for_keyframe = false;59 self.discontinuity_pending = false;60 return MediaV3GroupDecision::Send(position);61 }6263 if self.waiting_for_keyframe {64 return MediaV3GroupDecision::SkipUntilKeyframe;65 }66 if frame.keyframe || frame.codec_config {67 self.enter_resync();68 return MediaV3GroupDecision::EnterResync;69 }70 let Some(group_id) = self.group_id else {71 self.enter_resync();72 return MediaV3GroupDecision::EnterResync;73 };74 let contiguous =75 self.last_sequence.and_then(|last| last.checked_add(1)) == Some(frame.sequence);76 let object_id = self.last_object_id.and_then(|last| last.checked_add(1));77 let payload_bytes = self.payload_bytes.checked_add(frame.data.len());78 let (Some(object_id), Some(payload_bytes)) = (object_id, payload_bytes) else {79 self.enter_resync();80 return MediaV3GroupDecision::EnterResync;81 };82 if !contiguous83 || object_id > MAX_MEDIA_OBJECT_ID_V384 || payload_bytes > MAX_MEDIA_GROUP_BYTES_V385 {86 self.enter_resync();87 return MediaV3GroupDecision::EnterResync;88 }8990 self.last_sequence = Some(frame.sequence);91 self.last_object_id = Some(object_id);92 self.payload_bytes = payload_bytes;93 MediaV3GroupDecision::Send(MediaV3ObjectPosition {94 group_id,95 object_id,96 discontinuity: false,97 })98 }99100 pub(super) fn request_keyframe(&mut self) {101 self.enter_resync();102 }103104 fn enter_resync(&mut self) {105 self.group_id = None;106 self.last_object_id = None;107 self.payload_bytes = 0;108 self.waiting_for_keyframe = true;109 self.discontinuity_pending = true;110 }111}112113114enum MediaV3ScheduleDecision {115 Send {116 discontinuity: bool,117 cancel_sequences: Vec<u64>,118 },119 SkipUntilKeyframe,120 EnterResync {121 cancel_sequences: Vec<u64>,122 },123}124125126struct MediaV3Scheduler {127 in_flight: Vec<u64>,128 last_scheduled_sequence: Option<u64>,129 waiting_for_keyframe: bool,130 discontinuity_pending: bool,131}132133impl Default for MediaV3Scheduler {134 fn default() -> Self {135 Self {136 in_flight: Vec::with_capacity(MEDIA_V3_IN_FLIGHT_CAPACITY),137 last_scheduled_sequence: None,138 waiting_for_keyframe: true,139 discontinuity_pending: false,140 }141 }142}143144impl MediaV3Scheduler {145 fn schedule(146 &mut self,147 sequence: u64,148 independently_decodable: bool,149 ) -> MediaV3ScheduleDecision {150 if self151 .last_scheduled_sequence152 .is_some_and(|last| sequence <= last)153 {154 return MediaV3ScheduleDecision::SkipUntilKeyframe;155 }156157 let sequence_discontinuity = self158 .last_scheduled_sequence159 .is_some_and(|last| last.checked_add(1) != Some(sequence));160 if independently_decodable {161 let cancel_sequences = std::mem::take(&mut self.in_flight);162 let discontinuity = self.discontinuity_pending163 || sequence_discontinuity164 || !cancel_sequences.is_empty();165 self.in_flight.push(sequence);166 self.last_scheduled_sequence = Some(sequence);167 self.waiting_for_keyframe = false;168 self.discontinuity_pending = false;169 return MediaV3ScheduleDecision::Send {170 discontinuity,171 cancel_sequences,172 };173 }174175 if self.waiting_for_keyframe {176 return MediaV3ScheduleDecision::SkipUntilKeyframe;177 }178 if sequence_discontinuity || self.in_flight.len() == MEDIA_V3_IN_FLIGHT_CAPACITY {179 return MediaV3ScheduleDecision::EnterResync {180 cancel_sequences: self.enter_resync(),181 };182 }183184 self.in_flight.push(sequence);185 self.last_scheduled_sequence = Some(sequence);186 MediaV3ScheduleDecision::Send {187 discontinuity: false,188 cancel_sequences: Vec::new(),189 }190 }191192 fn complete(&mut self, sequence: u64) {193 if let Some(index) = self.in_flight.iter().position(|value| *value == sequence) {194 self.in_flight.swap_remove(index);195 }196 }197198 fn fail(&mut self, sequence: u64) -> Option<Vec<u64>> {199 let index = self.in_flight.iter().position(|value| *value == sequence)?;200 self.in_flight.swap_remove(index);201 Some(self.enter_resync())202 }203204 fn fail_all(&mut self) -> Vec<u64> {205 self.enter_resync()206 }207208 fn request_keyframe(&mut self) -> Vec<u64> {209 if self.waiting_for_keyframe {210 self.discontinuity_pending = true;211 return Vec::new();212 }213 self.enter_resync()214 }215216 fn enter_resync(&mut self) -> Vec<u64> {217 self.waiting_for_keyframe = true;218 self.discontinuity_pending = true;219 std::mem::take(&mut self.in_flight)220 }221}222223fn apply_media_v3_keyframe_request(224 scheduler: &mut MediaV3Scheduler,225 group_cursor: &mut MediaV3GroupCursor,226 replay_cursor: &mut MediaReplayCursor,227 through_sequence: Option<u64>,228 reason: KeyframeRequestReasonV3,229) -> (bool, Vec<u64>) {230 // Every v3 session already begins by replaying the bounded current GOP231 // from object zero. A Join arriving after that replay was scheduled must232 // not cancel the only decodable image on a damage-driven static source.233 if reason == KeyframeRequestReasonV3::Join || scheduler.waiting_for_keyframe {234 return (false, Vec::new());235 }236 let cancel_sequences = scheduler.request_keyframe();237 group_cursor.request_keyframe();238 replay_cursor.enter_resync_through(through_sequence);239 (true, cancel_sequences)240}241242fn apply_media_v3_send_failure(243 scheduler: &mut MediaV3Scheduler,244 group_cursor: &mut MediaV3GroupCursor,245 replay_cursor: &mut MediaReplayCursor,246 sequence: u64,247 through_sequence: Option<u64>,248) -> Option<Vec<u64>> {249 let cancel_sequences = scheduler.fail(sequence)?;250 group_cursor.request_keyframe();251 replay_cursor.enter_resync_through(through_sequence);252 Some(cancel_sequences)253}254255256enum MediaV3ControlDecision {257 Accept,258 Pace(Duration),259}260261262struct MediaV3ControlGate {263 last_request_id: Option<u64>,264 last_accepted_at: Option<Instant>,265}266267impl MediaV3ControlGate {268 fn accept(269 &mut self,270 request: MediaControlRequestV3,271 now: Instant,272 ) -> Result<MediaV3ControlDecision> {273 ensure!(274 self.last_request_id275 .is_none_or(|last| request.request_id > last),276 "v3 media control request IDs must be strictly increasing"277 );278 self.last_request_id = Some(request.request_id);279 if let Some(last) = self.last_accepted_at {280 let elapsed = now.saturating_duration_since(last);281 if elapsed < MEDIA_V3_CONTROL_REQUEST_INTERVAL {282 return Ok(MediaV3ControlDecision::Pace(283 MEDIA_V3_CONTROL_REQUEST_INTERVAL - elapsed,284 ));285 }286 }287 self.last_accepted_at = Some(now);288 Ok(MediaV3ControlDecision::Accept)289 }290}291292pub(super) async fn forward_media_v3_control_requests<R>(293 mut reader: R,294 sender: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,295) -> Result<()>296where297 R: tokio::io::AsyncRead + Unpin,298{299 let mut gate = MediaV3ControlGate::default();300 while let Some(request) = read_media_control_request_v3(&mut reader).await? {301 match gate.accept(request, Instant::now())? {302 MediaV3ControlDecision::Accept => {}303 MediaV3ControlDecision::Pace(retry_after) => {304 // Stop reading during the rejection interval so QUIC flow305 // control, rather than this task, absorbs an abusive burst.306 // This also bounds rejection logging to the configured rate.307 debug!(308 request_id = request.request_id,309 retry_after_ms = retry_after.as_millis(),310 "paced rate-limited v3 keyframe request"311 );312 tokio::time::sleep(retry_after).await;313 continue;314 }315 }316 sender.send_replace(Some(request));317 if sender.is_closed() {318 return Ok(());319 }320 }321 Ok(())322}323324struct ResetOnDropSendStreamV3(Option<SendStream>);325326impl ResetOnDropSendStreamV3 {327 fn new(stream: SendStream) -> Self {328 Self(Some(stream))329 }330331 fn stream(&self) -> &SendStream {332 self.0.as_ref().expect("send stream guard is armed")333 }334335 fn stream_mut(&mut self) -> &mut SendStream {336 self.0.as_mut().expect("send stream guard is armed")337 }338339 fn disarm(&mut self) {340 self.0.take();341 }342}343344impl Drop for ResetOnDropSendStreamV3 {345 fn drop(&mut self) {346 if let Some(stream) = self.0.as_mut() {347 let _ = stream.reset(MEDIA_V3_RESET_CODE.into());348 }349 }350}351352pub(super) async fn serve_media_v3(353 connection: Connection,354 config: HostConfig,355 sessions: &Arc<SessionRegistry>,356 authorization: &AuthorizationPolicy,357) -> Result<()> {358 let remote = connection.remote_id();359 let handshake_permit = sessions360 .pending_handshakes361 .try_acquire()362 .context("too many pending handshakes")?;363 let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())364 .await365 .context("timed out accepting media v3 handshake stream")?366 .context("accepting media v3 handshake stream")?;367 let hello = receive_hello(&mut recv, Capability::VideoH264).await?;368 drop(handshake_permit);369 debug!(%remote, agent = %hello.agent, "media v3 hello received");370371 let grants = match authorization.authorize_or_redeem(372 remote,373 hello.invitation.as_deref(),374 unix_timestamp_now()?,375 ) {376 Ok(grants) => grants,377 Err(error) => {378 send_rejection(&mut send, "Portal peer is not authorized").await?;379 return Err(error.context("authorizing media v3 peer"));380 }381 };382 ensure!(383 grants.contains(InvitationGrants::VIEW),384 "authorized media peer lacks view permission"385 );386387 let lease = match sessions.claim(remote, hello.nonce, grants) {388 Ok(lease) => lease,389 Err(error) => {390 send_rejection(&mut send, "host already has an active client").await?;391 return Err(error);392 }393 };394395 let source = match config.source {396 VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),397 VideoSource::GamescopePipewire => {398 let primary = spawn_gamescope_pipewire_after_static_preflight(399 config.clone(),400 lease.session_clock,401 )402 .await?;403 select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await404 }405 };406 let EncodedSource {407 frames: frame_receiver,408 current_gop: mut current_gop_receiver,409 task: source_task,410 pointer_surface_dimensions,411 encoder_control,412 } = match source {413 Ok(source) => source,414 Err(error) => {415 send_rejection(&mut send, "video source is unavailable").await?;416 return Err(error);417 }418 };419 let source_task = SourceTaskGuard::new(source_task);420 sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;421422 let mut media_hello = HostHello::accepted(423 lease.session_id,424 negotiated_capabilities(&hello, MEDIA_CAPABILITIES),425 );426 if let Some(dimensions) = pointer_surface_dimensions {427 media_hello = media_hello.with_pointer_surface_dimensions(dimensions);428 }429 write_host_hello(&mut send, &media_hello).await?;430 send.finish()431 .context("finishing media v3 handshake response")?;432 drop(send);433 info!(%remote, session_id = lease.session_id, "media v3 client accepted");434435 let session_result = run_media_v3_session(436 &connection,437 &config,438 &mut current_gop_receiver,439 recv,440 remote,441 encoder_control,442 Arc::clone(&lease.media_v3_telemetry),443 )444 .await;445446 drop(current_gop_receiver);447 drop(frame_receiver);448 source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;449 drop(lease);450 info!(%remote, "media v3 client released");451 session_result452}453454async fn run_media_v3_session(455 connection: &Connection,456 config: &HostConfig,457 current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,458 control_recv: iroh::endpoint::RecvStream,459 remote: EndpointId,460 encoder_control: Option<EncoderControl>,461 telemetry: Arc<MediaV3Telemetry>,462) -> Result<()> {463 let maximum_replay_age = maximum_media_replay_age(config.framerate);464 let delivery_timeout_ms = media_v3_delivery_timeout_ms(config.framerate);465 let mut replay_cursor = MediaReplayCursor::default();466 let mut group_cursor = MediaV3GroupCursor::default();467 let mut scheduler = MediaV3Scheduler::default();468 let mut send_tasks = tokio::task::JoinSet::new();469 let (control_sender, mut control_requests) = tokio::sync::watch::channel(None);470 let mut control_task = tokio::spawn(forward_media_v3_control_requests(471 control_recv,472 control_sender,473 ));474 let mut control_task_finished = false;475 let mut control_receiver_open = true;476 let mut forced_idr = ForcedIdrCoordinator::new(encoder_control, Arc::clone(&telemetry));477478 let result = loop {479 telemetry.record_selected_path(connection);480 tokio::select! {481 biased;482 control_result = &mut control_task, if !control_task_finished => {483 control_task_finished = true;484 match control_result {485 Ok(Ok(())) => {486 // Keep polling the watch receiver after clean EOF so a487 // final request published immediately before the488 // sender closed cannot be lost.489 debug!(%remote, "media v3 control stream closed");490 }491 Ok(Err(error)) => {492 break Err(error).context("reading media v3 control stream");493 }494 Err(error) => {495 break Err(error).context("media v3 control task failed");496 }497 }498 }499 changed = control_requests.changed(), if control_receiver_open => {500 if changed.is_err() {501 control_receiver_open = false;502 continue;503 }504 let request = *control_requests.borrow_and_update();505 let Some(request) = request else {506 continue;507 };508 let through_sequence = current_gop_receiver509 .borrow()510 .as_ref()511 .and_then(|gop| gop.frames.last())512 .map(|frame| frame.sequence);513 let (transitioned, cancel_sequences) = apply_media_v3_keyframe_request(514 &mut scheduler,515 &mut group_cursor,516 &mut replay_cursor,517 through_sequence,518 request.reason,519 );520 if !cancel_sequences.is_empty() {521 telemetry.scheduler_cancellations.fetch_add(522 u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),523 Ordering::Relaxed,524 );525 send_tasks.abort_all();526 }527 let forced_idr_disposition = forced_idr.request(request.reason);528 if let ForcedIdrDisposition::Failed { error } = &forced_idr_disposition {529 warn!(530 %remote,531 request_id = request.request_id,532 ?request.reason,533 %error,534 "forced-IDR request failed; retaining natural-IDR fallback"535 );536 }537 debug!(538 %remote,539 request_id = request.request_id,540 ?request.reason,541 advisory_last_sequence = ?request.last_sequence,542 coalesced = !transitioned,543 ?cancel_sequences,544 ?forced_idr_disposition,545 "accepted media v3 keyframe request"546 );547 }548 acknowledgement = forced_idr.acknowledgements.join_next(),549 if forced_idr.pending_revision.is_some() =>550 {551 forced_idr.complete(acknowledgement, remote, "grouped-v3");552 }553 closed = connection.closed() => {554 debug!(%remote, ?closed, "media v3 connection closed");555 break Ok(());556 }557 task = send_tasks.join_next(), if !send_tasks.is_empty() => {558 match task.expect("guarded by non-empty send task set") {559 Ok((sequence, Ok(()))) => scheduler.complete(sequence),560 Ok((sequence, Err(error))) => {561 let through_sequence = current_gop_receiver562 .borrow()563 .as_ref()564 .and_then(|gop| gop.frames.last())565 .map(|frame| frame.sequence);566 if let Some(cancel_sequences) = apply_media_v3_send_failure(567 &mut scheduler,568 &mut group_cursor,569 &mut replay_cursor,570 sequence,571 through_sequence,572 ) {573 telemetry.send_failures.fetch_add(1, Ordering::Relaxed);574 telemetry.scheduler_cancellations.fetch_add(575 u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),576 Ordering::Relaxed,577 );578 warn!(579 sequence,580 %error,581 "media v3 object send failed; waiting for keyframe"582 );583 if !cancel_sequences.is_empty() {584 send_tasks.abort_all();585 }586 } else {587 debug!(588 sequence,589 %error,590 "ignored stale media v3 object failure from a superseded group"591 );592 }593 }594 Err(error) if error.is_cancelled() => {}595 Err(error) => {596 warn!(%error, "media v3 object task failed; waiting for keyframe");597 let cancel_sequences = scheduler.fail_all();598 telemetry.send_failures.fetch_add(1, Ordering::Relaxed);599 telemetry.scheduler_cancellations.fetch_add(600 u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),601 Ordering::Relaxed,602 );603 group_cursor.request_keyframe();604 let through_sequence = current_gop_receiver605 .borrow()606 .as_ref()607 .and_then(|gop| gop.frames.last())608 .map(|frame| frame.sequence);609 replay_cursor.enter_resync_through(through_sequence);610 if !cancel_sequences.is_empty() {611 send_tasks.abort_all();612 }613 }614 }615 }616 changed = current_gop_receiver.changed() => {617 if let Err(error) = changed {618 break Err(error).context("encoded source stopped");619 }620 let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {621 continue;622 };623 let initial_replay_started_at =624 replay_cursor.last_sequence.is_none().then(Instant::now);625 let replay_through_sequence = current_gop626 .frames627 .last()628 .map(|frame| frame.sequence)629 .context("current GOP snapshot is empty")?;630631 for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {632 let replay_discontinuity = match replay_cursor.classify(633 &frame,634 replay_through_sequence,635 initial_replay_started_at,636 Instant::now(),637 maximum_replay_age,638 ) {639 MediaReplayDecision::Send { discontinuity } => discontinuity,640 MediaReplayDecision::SkipUntilKeyframe => {641 replay_cursor.enter_resync_through(Some(replay_through_sequence));642 break;643 }644 MediaReplayDecision::DiscardStaleSuffix { .. } => {645 group_cursor.request_keyframe();646 let cancel_sequences = scheduler.fail_all();647 telemetry.scheduler_cancellations.fetch_add(648 u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),649 Ordering::Relaxed,650 );651 if !cancel_sequences.is_empty() {652 send_tasks.abort_all();653 }654 break;655 }656 };657 let position = match group_cursor.classify(&frame) {658 MediaV3GroupDecision::Send(position) => position,659 MediaV3GroupDecision::SkipUntilKeyframe => {660 replay_cursor.enter_resync_through(Some(replay_through_sequence));661 break;662 }663 MediaV3GroupDecision::EnterResync => {664 replay_cursor.enter_resync_through(Some(replay_through_sequence));665 let cancel_sequences = scheduler.fail_all();666 telemetry.scheduler_cancellations.fetch_add(667 u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),668 Ordering::Relaxed,669 );670 if !cancel_sequences.is_empty() {671 send_tasks.abort_all();672 }673 break;674 }675 };676 let independently_decodable = position.object_id == 0;677 let (scheduler_discontinuity, cancel_sequences) =678 match scheduler.schedule(frame.sequence, independently_decodable) {679 MediaV3ScheduleDecision::Send {680 discontinuity,681 cancel_sequences,682 } => (discontinuity, cancel_sequences),683 MediaV3ScheduleDecision::SkipUntilKeyframe => continue,684 MediaV3ScheduleDecision::EnterResync { cancel_sequences } => {685 group_cursor.request_keyframe();686 replay_cursor.enter_resync_through(Some(replay_through_sequence));687 telemetry.scheduler_cancellations.fetch_add(688 u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),689 Ordering::Relaxed,690 );691 if !cancel_sequences.is_empty() {692 send_tasks.abort_all();693 }694 break;695 }696 };697 if !cancel_sequences.is_empty() {698 telemetry.scheduler_cancellations.fetch_add(699 u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),700 Ordering::Relaxed,701 );702 debug!(703 sequence = frame.sequence,704 group_id = position.group_id,705 ?cancel_sequences,706 "configured keyframe superseding media v3 objects"707 );708 send_tasks.abort_all();709 }710711 let media_object = media_v3_object_for_encoded(712 config,713 &frame,714 position,715 replay_discontinuity716 || position.discontinuity717 || scheduler_discontinuity,718 delivery_timeout_ms,719 )?;720 let sequence = frame.sequence;721 let stream = match tokio::time::timeout(722 Duration::from_millis(u64::from(delivery_timeout_ms)),723 connection.open_uni(),724 )725 .await726 {727 Ok(Ok(stream)) => stream,728 Ok(Err(error)) => {729 telemetry.send_failures.fetch_add(1, Ordering::Relaxed);730 warn!(sequence, %error, "opening media v3 object stream failed");731 if let Some(cancel_sequences) = apply_media_v3_send_failure(732 &mut scheduler,733 &mut group_cursor,734 &mut replay_cursor,735 sequence,736 Some(replay_through_sequence),737 ) && !cancel_sequences.is_empty()738 {739 send_tasks.abort_all();740 }741 break;742 }743 Err(_) => {744 telemetry.send_failures.fetch_add(1, Ordering::Relaxed);745 warn!(sequence, "opening media v3 object stream timed out");746 if let Some(cancel_sequences) = apply_media_v3_send_failure(747 &mut scheduler,748 &mut group_cursor,749 &mut replay_cursor,750 sequence,751 Some(replay_through_sequence),752 ) && !cancel_sequences.is_empty()753 {754 send_tasks.abort_all();755 }756 break;757 }758 };759 let stream = ResetOnDropSendStreamV3::new(stream);760 let transport_priority =761 media_v3_transport_priority(media_object.header.publisher_priority)?;762 if let Err(error) = stream.stream().set_priority(transport_priority) {763 telemetry.send_failures.fetch_add(1, Ordering::Relaxed);764 warn!(sequence, %error, "setting media v3 object priority failed");765 if let Some(cancel_sequences) = apply_media_v3_send_failure(766 &mut scheduler,767 &mut group_cursor,768 &mut replay_cursor,769 sequence,770 Some(replay_through_sequence),771 ) && !cancel_sequences.is_empty()772 {773 send_tasks.abort_all();774 }775 break;776 }777 send_tasks.spawn(async move {778 (779 sequence,780 send_media_v3_object(stream, media_object).await,781 )782 });783 replay_cursor.commit_sent(&frame);784 }785 }786 }787 };788789 forced_idr.abort_and_drain(remote, "grouped-v3").await;790 send_tasks.abort_all();791 while send_tasks.join_next().await.is_some() {}792 if !control_task_finished {793 control_task.abort();794 let _ = control_task.await;795 }796 result797}798799fn media_v3_delivery_timeout_ms(framerate: u32) -> u32 {800 debug_assert!(framerate > 0);801 let timeout_ms = 1_000_u64802 .saturating_mul(MEDIA_V3_DELIVERY_FRAME_PERIODS)803 .div_ceil(u64::from(framerate.max(1)));804 u32::try_from(timeout_ms).unwrap_or(u32::MAX).clamp(805 MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,806 MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,807 )808}809810fn media_v3_object_for_encoded(811 _config: &HostConfig,812 frame: &EncodedFrame,813 position: MediaV3ObjectPosition,814 discontinuity: bool,815 delivery_timeout_ms: u32,816) -> Result<MediaObjectV3> {817 let mut flags = FrameFlags::NONE;818 if frame.keyframe {819 flags = flags.union(FrameFlags::KEYFRAME);820 }821 if frame.codec_config {822 flags = flags.union(FrameFlags::CODEC_CONFIG);823 }824 if discontinuity || frame.discontinuity {825 flags = flags.union(FrameFlags::DISCONTINUITY);826 }827 let publisher_priority = if position.object_id == 0 {828 MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY829 } else {830 MEDIA_V3_DELTA_PUBLISHER_PRIORITY831 };832 let header = MediaObjectHeaderV3::h264(833 frame.width,834 frame.height,835 frame.data.len(),836 publisher_priority,837 flags,838 position.object_id,839 position.group_id,840 frame.sequence,841 frame.capture_timestamp_micros,842 frame.presentation_timestamp_micros,843 delivery_timeout_ms,844 )?;845 MediaObjectV3::new(header, frame.data.clone()).map_err(Into::into)846}847848async fn send_media_v3_object(849 mut stream: ResetOnDropSendStreamV3,850 object: MediaObjectV3,851) -> Result<()> {852 let delivery_timeout = Duration::from_millis(u64::from(object.header.delivery_timeout_ms));853 tokio::time::timeout(delivery_timeout, async {854 write_media_object_v3(stream.stream_mut(), &object)855 .await856 .context("writing media v3 object")?;857 stream858 .stream_mut()859 .finish()860 .context("finishing media v3 object stream")?;861 match stream862 .stream()863 .stopped()864 .await865 .context("waiting for media v3 object acknowledgement")?866 {867 None => {868 stream.disarm();869 Ok(())870 }871 Some(code) => bail!("peer stopped media v3 object stream with code {code}"),872 }873 })874 .await875 .context("media v3 object exceeded its delivery timeout")?876}877878fn media_v3_transport_priority(publisher_priority: u8) -> Result<i32> {879 // V3 follows MoQ's lower-is-higher publisher priority, while Iroh/QUIC880 // uses larger integers for more important streams. Keep the inversion881 // explicit rather than leaking either convention across the boundary.882 match publisher_priority {883 MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY => Ok(MEDIA_V3_KEYFRAME_TRANSPORT_PRIORITY),884 MEDIA_V3_DELTA_PUBLISHER_PRIORITY => Ok(MEDIA_V3_DELTA_TRANSPORT_PRIORITY),885 _ => bail!("unsupported media v3 publisher priority {publisher_priority}"),886 }887}888889pub(super) fn new_current_gop_frames(890 current_gop: EncodedGop,891 last_sequence: Option<u64>,892) -> impl Iterator<Item = EncodedFrame> {893 current_gop894 .frames895 .into_iter()896 .skip_while(move |frame| last_sequence.is_some_and(|last| frame.sequence <= last))897}898899900mod tests {901 use super::super::media_v3_encoded_frame;902 use super::*;903904 905 fn media_v3_groups_begin_at_configured_idr_and_assign_contiguous_objects() {906 let mut cursor = MediaV3GroupCursor::default();907 assert_eq!(908 cursor.classify(&media_v3_encoded_frame(10, true, true, 4)),909 MediaV3GroupDecision::Send(MediaV3ObjectPosition {910 group_id: 10,911 object_id: 0,912 discontinuity: false,913 })914 );915 assert_eq!(916 cursor.classify(&media_v3_encoded_frame(11, false, false, 3)),917 MediaV3GroupDecision::Send(MediaV3ObjectPosition {918 group_id: 10,919 object_id: 1,920 discontinuity: false,921 })922 );923924 cursor.request_keyframe();925 assert_eq!(926 cursor.classify(&media_v3_encoded_frame(12, false, false, 3)),927 MediaV3GroupDecision::SkipUntilKeyframe928 );929 assert_eq!(930 cursor.classify(&media_v3_encoded_frame(20, true, true, 4)),931 MediaV3GroupDecision::Send(MediaV3ObjectPosition {932 group_id: 20,933 object_id: 0,934 discontinuity: true,935 })936 );937 }938939 940 fn media_v3_group_rejects_noncontiguous_or_unconfigured_frames() {941 let mut cursor = MediaV3GroupCursor::default();942 assert_eq!(943 cursor.classify(&media_v3_encoded_frame(10, true, false, 1)),944 MediaV3GroupDecision::SkipUntilKeyframe945 );946 assert!(matches!(947 cursor.classify(&media_v3_encoded_frame(20, true, true, 1)),948 MediaV3GroupDecision::Send(_)949 ));950 assert_eq!(951 cursor.classify(&media_v3_encoded_frame(22, false, false, 1)),952 MediaV3GroupDecision::EnterResync953 );954 assert!(cursor.waiting_for_keyframe);955 }956957 958 fn media_v3_group_accepts_exact_limits_and_rejects_overflow() {959 let mut object_cursor = MediaV3GroupCursor::default();960 assert!(matches!(961 object_cursor.classify(&media_v3_encoded_frame(0, true, true, 1)),962 MediaV3GroupDecision::Send(_)963 ));964 for sequence in 1..=u64::from(MAX_MEDIA_OBJECT_ID_V3) {965 assert!(matches!(966 object_cursor.classify(&media_v3_encoded_frame(sequence, false, false, 1)),967 MediaV3GroupDecision::Send(MediaV3ObjectPosition { object_id, .. })968 if u64::from(object_id) == sequence969 ));970 }971 assert_eq!(972 object_cursor.classify(&media_v3_encoded_frame(973 u64::from(MAX_MEDIA_OBJECT_ID_V3) + 1,974 false,975 false,976 1,977 )),978 MediaV3GroupDecision::EnterResync979 );980981 let mut byte_cursor = MediaV3GroupCursor::default();982 assert!(matches!(983 byte_cursor.classify(&media_v3_encoded_frame(10, true, true, 1)),984 MediaV3GroupDecision::Send(_)985 ));986 byte_cursor.payload_bytes = MAX_MEDIA_GROUP_BYTES_V3 - 1;987 assert!(matches!(988 byte_cursor.classify(&media_v3_encoded_frame(11, false, false, 1)),989 MediaV3GroupDecision::Send(_)990 ));991 assert_eq!(byte_cursor.payload_bytes, MAX_MEDIA_GROUP_BYTES_V3);992 assert_eq!(993 byte_cursor.classify(&media_v3_encoded_frame(12, false, false, 1)),994 MediaV3GroupDecision::EnterResync995 );996 }997998 999 fn media_v3_keyframe_requests_cancel_once_and_recover_discontinuously() {1000 let mut scheduler = MediaV3Scheduler::default();1001 assert!(matches!(1002 scheduler.schedule(10, true),1003 MediaV3ScheduleDecision::Send { .. }1004 ));1005 assert!(matches!(1006 scheduler.schedule(11, false),1007 MediaV3ScheduleDecision::Send { .. }1008 ));1009 assert_eq!(scheduler.request_keyframe(), vec![10, 11]);1010 assert!(scheduler.request_keyframe().is_empty());1011 assert_eq!(1012 scheduler.schedule(12, false),1013 MediaV3ScheduleDecision::SkipUntilKeyframe1014 );1015 assert_eq!(1016 scheduler.schedule(20, true),1017 MediaV3ScheduleDecision::Send {1018 discontinuity: true,1019 cancel_sequences: vec![],1020 }1021 );1022 }10231024 1025 fn media_v3_join_request_preserves_initial_current_gop_replay() {1026 let mut scheduler = MediaV3Scheduler::default();1027 let mut group_cursor = MediaV3GroupCursor::default();1028 let mut replay_cursor = MediaReplayCursor::default();10291030 let (transitioned, cancel_sequences) = apply_media_v3_keyframe_request(1031 &mut scheduler,1032 &mut group_cursor,1033 &mut replay_cursor,1034 Some(42),1035 KeyframeRequestReasonV3::Join,1036 );10371038 assert!(!transitioned);1039 assert!(cancel_sequences.is_empty());1040 assert_eq!(replay_cursor.last_sequence, None);1041 assert!(replay_cursor.waiting_for_keyframe);1042 assert!(!replay_cursor.discontinuity_pending);1043 assert!(group_cursor.waiting_for_keyframe);1044 assert!(!group_cursor.discontinuity_pending);1045 assert!(!scheduler.discontinuity_pending);10461047 assert!(matches!(1048 group_cursor.classify(&media_v3_encoded_frame(10, true, true, 1)),1049 MediaV3GroupDecision::Send(MediaV3ObjectPosition {1050 discontinuity: false,1051 ..1052 })1053 ));1054 assert!(matches!(1055 scheduler.schedule(10, true),1056 MediaV3ScheduleDecision::Send {1057 discontinuity: false,1058 ..1059 }1060 ));1061 }10621063 1064 fn media_v3_late_join_preserves_active_static_current_gop() {1065 let mut scheduler = MediaV3Scheduler::default();1066 let mut group_cursor = MediaV3GroupCursor::default();1067 let mut replay_cursor = MediaReplayCursor::default();1068 let keyframe = media_v3_encoded_frame(10, true, true, 1);1069 let delta = media_v3_encoded_frame(11, false, false, 1);10701071 assert!(matches!(1072 group_cursor.classify(&keyframe),1073 MediaV3GroupDecision::Send(_)1074 ));1075 assert!(matches!(1076 scheduler.schedule(keyframe.sequence, true),1077 MediaV3ScheduleDecision::Send { .. }1078 ));1079 replay_cursor.commit_sent(&keyframe);1080 assert!(matches!(1081 group_cursor.classify(&delta),1082 MediaV3GroupDecision::Send(_)1083 ));1084 assert!(matches!(1085 scheduler.schedule(delta.sequence, false),1086 MediaV3ScheduleDecision::Send { .. }1087 ));1088 replay_cursor.commit_sent(&delta);10891090 let (transitioned, cancel_sequences) = apply_media_v3_keyframe_request(1091 &mut scheduler,1092 &mut group_cursor,1093 &mut replay_cursor,1094 Some(delta.sequence),1095 KeyframeRequestReasonV3::Join,1096 );10971098 assert!(!transitioned);1099 assert!(cancel_sequences.is_empty());1100 assert_eq!(scheduler.in_flight, vec![10, 11]);1101 assert!(!scheduler.waiting_for_keyframe);1102 assert!(!scheduler.discontinuity_pending);1103 assert_eq!(group_cursor.group_id, Some(10));1104 assert_eq!(group_cursor.last_sequence, Some(11));1105 assert!(!group_cursor.waiting_for_keyframe);1106 assert_eq!(replay_cursor.last_sequence, Some(11));1107 assert!(!replay_cursor.waiting_for_keyframe);1108 }11091110 1111 fn media_v3_stale_old_group_failure_preserves_replacement_group() {1112 let mut scheduler = MediaV3Scheduler::default();1113 let mut group_cursor = MediaV3GroupCursor::default();1114 let mut replay_cursor = MediaReplayCursor::default();1115 let old_keyframe = media_v3_encoded_frame(60, true, true, 1);1116 let old_delta = media_v3_encoded_frame(61, false, false, 1);1117 let replacement = media_v3_encoded_frame(70, true, true, 1);11181119 for frame in [&old_keyframe, &old_delta] {1120 assert!(matches!(1121 group_cursor.classify(frame),1122 MediaV3GroupDecision::Send(_)1123 ));1124 assert!(matches!(1125 scheduler.schedule(frame.sequence, frame.keyframe && frame.codec_config),1126 MediaV3ScheduleDecision::Send { .. }1127 ));1128 replay_cursor.commit_sent(frame);1129 }1130 assert!(matches!(1131 group_cursor.classify(&replacement),1132 MediaV3GroupDecision::Send(_)1133 ));1134 assert_eq!(1135 scheduler.schedule(replacement.sequence, true),1136 MediaV3ScheduleDecision::Send {1137 discontinuity: true,1138 cancel_sequences: vec![60, 61],1139 }1140 );1141 replay_cursor.commit_sent(&replacement);11421143 assert_eq!(1144 apply_media_v3_send_failure(1145 &mut scheduler,1146 &mut group_cursor,1147 &mut replay_cursor,1148 old_keyframe.sequence,1149 Some(replacement.sequence),1150 ),1151 None1152 );1153 assert_eq!(scheduler.in_flight, vec![70]);1154 assert!(!scheduler.waiting_for_keyframe);1155 assert_eq!(group_cursor.group_id, Some(70));1156 assert!(!group_cursor.waiting_for_keyframe);1157 assert_eq!(replay_cursor.last_sequence, Some(70));1158 assert!(!replay_cursor.waiting_for_keyframe);11591160 let next_delta = media_v3_encoded_frame(71, false, false, 1);1161 assert!(matches!(1162 group_cursor.classify(&next_delta),1163 MediaV3GroupDecision::Send(MediaV3ObjectPosition {1164 group_id: 70,1165 object_id: 1,1166 ..1167 })1168 ));1169 assert!(matches!(1170 scheduler.schedule(next_delta.sequence, false),1171 MediaV3ScheduleDecision::Send {1172 discontinuity: false,1173 ..1174 }1175 ));1176 }11771178 1179 fn media_v3_control_gate_is_monotonic_and_accepts_at_most_ten_per_second() {1180 let started = Instant::now();1181 let request = |request_id| {1182 MediaControlRequestV3::request_keyframe(1183 request_id,1184 None,1185 sigil_protocol::KeyframeRequestReasonV3::TransportGap,1186 )1187 };1188 let mut gate = MediaV3ControlGate::default();1189 assert_eq!(1190 gate.accept(request(1), started).unwrap(),1191 MediaV3ControlDecision::Accept1192 );1193 assert_eq!(1194 gate.accept(1195 request(2),1196 started + MEDIA_V3_CONTROL_REQUEST_INTERVAL - Duration::from_nanos(1),1197 )1198 .unwrap(),1199 MediaV3ControlDecision::Pace(Duration::from_nanos(1))1200 );1201 assert_eq!(1202 gate.accept(request(3), started + MEDIA_V3_CONTROL_REQUEST_INTERVAL)1203 .unwrap(),1204 MediaV3ControlDecision::Accept1205 );1206 assert!(1207 gate.accept(request(3), started + Duration::from_secs(1))1208 .is_err()1209 );1210 }12111212 1213 async fn media_v3_rejected_control_requests_apply_read_side_pacing() {1214 use tokio::io::AsyncWriteExt as _;12151216 let requests = [1, 2, 3].map(|request_id| {1217 MediaControlRequestV3::request_keyframe(1218 request_id,1219 None,1220 KeyframeRequestReasonV3::TransportGap,1221 )1222 });1223 let (mut writer, reader) = tokio::io::duplex(128);1224 for request in &requests {1225 sigil_protocol::write_media_control_request_v3(&mut writer, request)1226 .await1227 .unwrap();1228 }1229 writer.shutdown().await.unwrap();1230 let (sender, mut receiver) = tokio::sync::watch::channel(None);12311232 let started = Instant::now();1233 forward_media_v3_control_requests(reader, sender)1234 .await1235 .unwrap();12361237 assert!(started.elapsed() >= MEDIA_V3_CONTROL_REQUEST_INTERVAL);1238 receiver.changed().await.unwrap();1239 assert_eq!(*receiver.borrow_and_update(), Some(requests[2]));1240 }12411242 1243 async fn media_v3_control_eof_is_clean_and_malformed_input_is_terminal() {1244 use tokio::io::AsyncWriteExt as _;12451246 let (writer, reader) = tokio::io::duplex(64);1247 let (sender, mut receiver) = tokio::sync::watch::channel(None);1248 drop(writer);1249 forward_media_v3_control_requests(reader, sender)1250 .await1251 .unwrap();1252 assert!(receiver.changed().await.is_err());12531254 let final_request = MediaControlRequestV3::request_keyframe(1255 1,1256 Some(9),1257 sigil_protocol::KeyframeRequestReasonV3::DecoderReset,1258 );1259 let (mut writer, reader) = tokio::io::duplex(64);1260 let (sender, mut receiver) = tokio::sync::watch::channel(None);1261 sigil_protocol::write_media_control_request_v3(&mut writer, &final_request)1262 .await1263 .unwrap();1264 writer.shutdown().await.unwrap();1265 forward_media_v3_control_requests(reader, sender)1266 .await1267 .unwrap();1268 receiver.changed().await.unwrap();1269 assert_eq!(*receiver.borrow_and_update(), Some(final_request));12701271 let (mut writer, reader) = tokio::io::duplex(64);1272 let (sender, _receiver) = tokio::sync::watch::channel(None);1273 writer1274 .write_all(&[0; sigil_protocol::MEDIA_CONTROL_REQUEST_V3_LEN])1275 .await1276 .unwrap();1277 writer.shutdown().await.unwrap();1278 assert!(1279 forward_media_v3_control_requests(reader, sender)1280 .await1281 .is_err()1282 );1283 }12841285 1286 fn media_v3_deadline_ceil_clamps_four_frame_periods() {1287 assert_eq!(media_v3_delivery_timeout_ms(60), 67);1288 assert_eq!(media_v3_delivery_timeout_ms(240), 17);1289 assert_eq!(1290 media_v3_delivery_timeout_ms(1_000),1291 MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS1292 );1293 assert_eq!(1294 media_v3_delivery_timeout_ms(1),1295 MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS1296 );1297 }12981299 1300 fn media_v3_publisher_priority_maps_to_inverse_transport_priority() {1301 let keyframe_transport =1302 media_v3_transport_priority(MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY).unwrap();1303 let delta_transport =1304 media_v3_transport_priority(MEDIA_V3_DELTA_PUBLISHER_PRIORITY).unwrap();1305 assert_eq!(keyframe_transport, MEDIA_V3_KEYFRAME_TRANSPORT_PRIORITY);1306 assert_eq!(delta_transport, MEDIA_V3_DELTA_TRANSPORT_PRIORITY);1307 assert!(keyframe_transport > delta_transport);1308 assert!(media_v3_transport_priority(1).is_err());1309 }13101311 1312 fn current_gop_replay_is_complete_and_skips_already_sent_frames() {1313 let frame = |sequence, keyframe| EncodedFrame {1314 sequence,1315 width: 1_280,1316 height: 800,1317 capture_timestamp_micros: sequence,1318 presentation_timestamp_micros: sequence as i64,1319 observed_at: std::time::Instant::now(),1320 keyframe,1321 codec_config: keyframe,1322 discontinuity: false,1323 data: Arc::from([sequence as u8]),1324 };13251326 let gop = || EncodedGop {1327 frames: vec![frame(10, true), frame(11, false), frame(12, false)],1328 payload_bytes: 3,1329 };13301331 let initial = new_current_gop_frames(gop(), None).collect::<Vec<_>>();1332 assert_eq!(1333 initial1334 .iter()1335 .map(|frame| frame.sequence)1336 .collect::<Vec<_>>(),1337 vec![10, 11, 12]1338 );1339 assert!(initial[0].keyframe && initial[0].codec_config);13401341 let resumed = new_current_gop_frames(gop(), Some(10)).collect::<Vec<_>>();1342 assert_eq!(1343 resumed1344 .iter()1345 .map(|frame| frame.sequence)1346 .collect::<Vec<_>>(),1347 vec![11, 12]1348 );13491350 let behind_current_gop = new_current_gop_frames(gop(), Some(8)).collect::<Vec<_>>();1351 assert_eq!(behind_current_gop[0].sequence, 10);1352 assert!(behind_current_gop[0].keyframe && behind_current_gop[0].codec_config);13531354 assert!(new_current_gop_frames(gop(), Some(12)).next().is_none());1355 }1356}