1use std::sync::atomic::{AtomicUsize, Ordering};2use std::sync::{Arc, Mutex as StdMutex};3use std::time::{Duration, Instant};45use sigil_protocol::{FrameFlags, KeyframeRequestReasonV3, MediaCodec};6use tauri::{7 AppHandle, Emitter,8 ipc::{Channel, Response},9};1011use crate::commands::state::MediaControlRequestSender;12use crate::media::frame_channel::{13 FrameEnvelopeMetadata, emit_frame_error, encode_frame_envelope, release_frame_channel_slot,14 try_reserve_frame_channel_slot,15};16use crate::media::media_control::try_queue_media_keyframe_request;17use crate::media::metrics::{ClientMediaMetrics, lock_client_media_metrics};18use crate::media::moq_receiver::{19 MoqMediaReadOutcome, MoqMediaReceiver, retire_upstream_moq_generation,20};21use crate::media::network_diagnostics::{22 NetworkLeg, NetworkSessionDiagnostics, lock_network_diagnostics,23};24use crate::media::object_receiver::{25 MediaObjectReadOutcomeV3, MediaObjectReceiverV3, MediaObjectSequenceDecisionV3,26 MediaObjectSequenceV3,27};28use crate::media::transport::MediaTransport;2930const CLIENT_FRAME_STATS_INTERVAL: Duration = Duration::from_millis(250);3132pub(crate) struct VideoDeliveryRequest {33 pub(crate) app: AppHandle,34 pub(crate) endpoint: iroh::Endpoint,35 pub(crate) frame_connection: iroh::endpoint::Connection,36 pub(crate) input_connection: Option<iroh::endpoint::Connection>,37 pub(crate) audio_connection: Option<iroh::endpoint::Connection>,38 pub(crate) network_diagnostics: Arc<StdMutex<NetworkSessionDiagnostics>>,39 pub(crate) media_control_requests: Option<MediaControlRequestSender>,40 pub(crate) upstream_moq_media: Option<MoqMediaReceiver>,41 pub(crate) frame_events_in_flight: Arc<AtomicUsize>,42 pub(crate) frame_channel: Channel<Response>,43 pub(crate) media_transport: MediaTransport,44 pub(crate) media_generation: u64,45 pub(crate) connected_audio_generation: Option<u64>,46}4748pub(crate) async fn run_video_delivery(request: VideoDeliveryRequest) {49 let VideoDeliveryRequest {50 app,51 endpoint,52 frame_connection: frame_connection_for_stats,53 input_connection,54 audio_connection: audio_connection_for_stats,55 network_diagnostics,56 media_control_requests,57 upstream_moq_media,58 frame_events_in_flight,59 frame_channel,60 media_transport,61 media_generation,62 connected_audio_generation,63 } = request;6465 let metrics_started = Instant::now();66 let mut initial_metrics = ClientMediaMetrics::default();67 // Joining a running encoder commonly starts in the middle of a GOP.68 // The initial keyframe wait is a real resync episode, just like a wait69 // entered after frontend backpressure, so account for it from t=0.70 initial_metrics.begin_frontend_resync(Duration::ZERO);71 let metrics = Arc::new(StdMutex::new(initial_metrics));72 let mut previous_sequence: Option<u64> = None;73 let mut frontend_waiting_for_keyframe = true;74 let mut media_objects_v3 = (media_transport == MediaTransport::GroupedObjectsV3)75 .then(|| MediaObjectReceiverV3::new(frame_connection_for_stats.clone()));76 let mut media_object_sequence_v3 = MediaObjectSequenceV3::new();77 let mut upstream_moq_media = upstream_moq_media;7879 let (stats_stop, mut stats_stop_rx) = tokio::sync::watch::channel(false);80 let stats_app = app.clone();81 let stats_metrics = Arc::clone(&metrics);82 let stats_in_flight = Arc::clone(&frame_events_in_flight);83 let stats_connection = frame_connection_for_stats.clone();84 let stats_input_connection = input_connection.clone();85 let stats_audio_connection = audio_connection_for_stats.clone();86 let stats_network_diagnostics = Arc::clone(&network_diagnostics);87 let stats_task = tokio::spawn(async move {88 let mut interval = tokio::time::interval(CLIENT_FRAME_STATS_INTERVAL);89 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);90 // Tokio intervals tick immediately once. Consume that tick so the91 // first payload represents a full diagnostics interval.92 interval.tick().await;93 let mut last_path_sample = Instant::now() - Duration::from_secs(1);9495 loop {96 tokio::select! {97 _ = interval.tick() => {98 if last_path_sample.elapsed() >= Duration::from_secs(1) {99 let mut diagnostics = lock_network_diagnostics(&stats_network_diagnostics);100 diagnostics.observe_connection(NetworkLeg::Media, &stats_connection);101 if let Some(input_connection) = stats_input_connection.as_ref() {102 diagnostics.observe_connection(NetworkLeg::Input, input_connection);103 }104 if let Some(audio_connection) = stats_audio_connection.as_ref() {105 diagnostics.observe_connection(NetworkLeg::Audio, audio_connection);106 }107 last_path_sample = Instant::now();108 }109 let queue_depth = stats_in_flight.load(Ordering::SeqCst);110 let network_snapshot = lock_network_diagnostics(&stats_network_diagnostics)111 .snapshot(Instant::now());112 let payload = lock_client_media_metrics(&stats_metrics).snapshot(113 metrics_started.elapsed(),114 queue_depth,115 network_snapshot,116 media_generation,117 );118 let _ = stats_app.emit("frame-stats", payload);119 }120 changed = stats_stop_rx.changed() => {121 if changed.is_err() || *stats_stop_rx.borrow() {122 break;123 }124 }125 }126 }127 });128129 'frames: loop {130 let (131 w,132 h,133 frame_buf,134 is_keyframe,135 codec,136 sequence,137 capture_timestamp_micros,138 pts_micros,139 discontinuity,140 codec_config,141 ) = match media_transport {142 MediaTransport::UpstreamMoq => {143 let receiver = upstream_moq_media144 .as_mut()145 .expect("upstream MoQ receiver must exist for MoQ transport");146 loop {147 let outcome = match receiver.next().await {148 Ok(Some(outcome)) => outcome,149 Ok(None) => {150 emit_frame_error(151 &app,152 media_generation,153 "Upstream MoQ video track closed",154 );155 break 'frames;156 }157 Err(error) => {158 emit_frame_error(&app, media_generation, error);159 break 'frames;160 }161 };162 match outcome {163 MoqMediaReadOutcome::Dropped { reason } => {164 frontend_waiting_for_keyframe = true;165 let mut metrics = lock_client_media_metrics(&metrics);166 metrics.observe_transport_object_drop(false);167 metrics.begin_frontend_resync(metrics_started.elapsed());168 try_queue_media_keyframe_request(169 media_control_requests.as_ref(),170 reason,171 receiver.last_frame_sequence(),172 );173 }174 MoqMediaReadOutcome::Malformed(error) => {175 emit_frame_error(&app, media_generation, error);176 break 'frames;177 }178 MoqMediaReadOutcome::Frame {179 frame,180 discontinuity,181 } => {182 let codec = match frame.header.codec {183 MediaCodec::H264 => "h264".to_string(),184 };185 break (186 u32::from(frame.header.width),187 u32::from(frame.header.height),188 frame.payload,189 frame.header.flags.contains(FrameFlags::KEYFRAME),190 codec,191 Some(frame.header.sequence),192 Some(frame.header.capture_timestamp_us),193 Some(frame.header.pts_us),194 discontinuity,195 frame.header.flags.contains(FrameFlags::CODEC_CONFIG),196 );197 }198 }199 }200 }201 MediaTransport::GroupedObjectsV3 => {202 let receiver = media_objects_v3203 .as_mut()204 .expect("media v3 receiver must exist for media v3 transport");205 loop {206 let outcome = match receiver.next().await {207 Ok(Some(outcome)) => outcome,208 Ok(None) => {209 emit_frame_error(&app, media_generation, "Connection closed");210 break 'frames;211 }212 Err(error) => {213 emit_frame_error(&app, media_generation, error);214 break 'frames;215 }216 };217 match outcome {218 MediaObjectReadOutcomeV3::Dropped { reason, .. } => {219 let begins_resync = media_object_sequence_v3.note_dropped_object();220 let mut metrics = lock_client_media_metrics(&metrics);221 metrics.observe_transport_object_drop(!begins_resync);222 if begins_resync {223 frontend_waiting_for_keyframe = true;224 metrics.begin_frontend_resync(metrics_started.elapsed());225 try_queue_media_keyframe_request(226 media_control_requests.as_ref(),227 reason,228 media_object_sequence_v3.last_sequence(),229 );230 }231 }232 MediaObjectReadOutcomeV3::Malformed(error) => {233 emit_frame_error(&app, media_generation, error);234 break 'frames;235 }236 MediaObjectReadOutcomeV3::Object { object, .. } => {237 let was_waiting = media_object_sequence_v3.waiting_for_keyframe();238 let discontinuity = match media_object_sequence_v3.classify(&object) {239 MediaObjectSequenceDecisionV3::Deliver { discontinuity } => {240 discontinuity241 }242 MediaObjectSequenceDecisionV3::DropLate => {243 lock_client_media_metrics(&metrics)244 .observe_transport_object_drop(true);245 continue;246 }247 MediaObjectSequenceDecisionV3::DropUntilKeyframe => {248 frontend_waiting_for_keyframe = true;249 let mut metrics = lock_client_media_metrics(&metrics);250 metrics.observe_transport_object_drop(false);251 metrics.begin_frontend_resync(metrics_started.elapsed());252 if !was_waiting {253 try_queue_media_keyframe_request(254 media_control_requests.as_ref(),255 KeyframeRequestReasonV3::TransportGap,256 media_object_sequence_v3.last_sequence(),257 );258 }259 continue;260 }261 };262 let codec = match object.header.codec {263 MediaCodec::H264 => "h264".to_string(),264 };265 break (266 u32::from(object.header.width),267 u32::from(object.header.height),268 object.payload,269 object.header.flags.contains(FrameFlags::KEYFRAME),270 codec,271 Some(object.header.sequence),272 Some(object.header.capture_timestamp_us),273 Some(object.header.pts_us),274 discontinuity,275 object.header.flags.contains(FrameFlags::CODEC_CONFIG),276 );277 }278 }279 }280 }281 };282 lock_client_media_metrics(&metrics).observe_transport_receive(283 metrics_started.elapsed(),284 sequence,285 is_keyframe,286 );287288 let sequence_gap = match sequence.zip(previous_sequence) {289 Some((current, previous)) => match sequence_gap(previous, current) {290 Ok(gap) => gap,291 Err(error) => {292 emit_frame_error(&app, media_generation, error);293 break;294 }295 },296 None => 0,297 };298 if sequence.is_some() {299 previous_sequence = sequence;300 }301 lock_client_media_metrics(&metrics).observe_sequence_drop(sequence_gap);302303 if frontend_waiting_for_keyframe && !is_keyframe {304 lock_client_media_metrics(&metrics).observe_frontend_resync_drop();305 continue;306 }307 if !try_reserve_frame_channel_slot(&frame_events_in_flight) {308 let begins_resync = !frontend_waiting_for_keyframe;309 frontend_waiting_for_keyframe = true;310 let mut metrics = lock_client_media_metrics(&metrics);311 metrics.observe_frontend_queue_drop();312 metrics.begin_frontend_resync(metrics_started.elapsed());313 if begins_resync {314 try_queue_media_keyframe_request(315 media_control_requests.as_ref(),316 KeyframeRequestReasonV3::FrontendBackpressure,317 sequence,318 );319 }320 continue;321 }322 lock_client_media_metrics(&metrics)323 .observe_frontend_queue_depth(frame_events_in_flight.load(Ordering::SeqCst));324325 let envelope = match encode_frame_envelope(326 FrameEnvelopeMetadata {327 width: w,328 height: h,329 codec: &codec,330 keyframe: is_keyframe,331 discontinuity,332 codec_config,333 sequence,334 capture_timestamp_micros,335 pts_micros,336 },337 &frame_buf,338 ) {339 Ok(envelope) => envelope,340 Err(error) => {341 emit_frame_error(&app, media_generation, error);342 release_frame_channel_slot(&frame_events_in_flight);343 break;344 }345 };346 let ipc_send_started = Instant::now();347 let send_result = frame_channel.send(Response::new(envelope));348 lock_client_media_metrics(&metrics).observe_frontend_ipc_send_duration(349 metrics_started.elapsed(),350 ipc_send_started.elapsed(),351 );352 if send_result.is_err() {353 release_frame_channel_slot(&frame_events_in_flight);354 break;355 }356357 frontend_waiting_for_keyframe = false;358 {359 let mut metrics = lock_client_media_metrics(&metrics);360 let elapsed = metrics_started.elapsed();361 metrics.observe_frontend_send(elapsed);362 metrics.finish_frontend_resync(elapsed);363 }364365 tokio::task::yield_now().await;366 }367368 let _ = stats_stop.send(true);369 let _ = stats_task.await;370 if media_transport == MediaTransport::UpstreamMoq {371 retire_upstream_moq_generation(&app, media_generation, connected_audio_generation).await;372 }373 drop(endpoint);374}375376fn sequence_gap(previous: u64, current: u64) -> Result<u64, String> {377 let expected = previous378 .checked_add(1)379 .ok_or_else(|| format!("Media sequence overflowed after {previous}"))?;380 if current < expected {381 return Err(format!(382 "Non-monotonic media sequence: previous={previous}, current={current}"383 ));384 }385 Ok(current - expected)386}387388389mod tests {390 use super::*;391392 393 fn sequence_checks_reject_duplicates_regressions_and_overflow() {394 assert_eq!(sequence_gap(41, 42).unwrap(), 0);395 assert_eq!(sequence_gap(41, 45).unwrap(), 3);396 assert!(sequence_gap(41, 41).is_err());397 assert!(sequence_gap(41, 40).is_err());398 assert!(sequence_gap(u64::MAX, 0).is_err());399 }400}