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_recv: iroh::endpoint::RecvStream,36 pub(crate) frame_connection: iroh::endpoint::Connection,37 pub(crate) input_connection: iroh::endpoint::Connection,38 pub(crate) audio_connection: Option<iroh::endpoint::Connection>,39 pub(crate) network_diagnostics: Arc<StdMutex<NetworkSessionDiagnostics>>,40 pub(crate) media_control_requests: Option<MediaControlRequestSender>,41 pub(crate) upstream_moq_media: Option<MoqMediaReceiver>,42 pub(crate) frame_events_in_flight: Arc<AtomicUsize>,43 pub(crate) frame_channel: Channel<Response>,44 pub(crate) media_transport: MediaTransport,45 pub(crate) media_generation: u64,46 pub(crate) connected_audio_generation: Option<u64>,47}4849pub(crate) async fn run_video_delivery(request: VideoDeliveryRequest) {50 let VideoDeliveryRequest {51 app,52 endpoint,53 // The negotiated control/v3 handshake recv leg carries no media but54 // must stay open for the session's lifetime.55 frame_recv: _control_recv_keepalive,56 frame_connection: frame_connection_for_stats,57 input_connection,58 audio_connection: audio_connection_for_stats,59 network_diagnostics,60 media_control_requests,61 upstream_moq_media,62 frame_events_in_flight,63 frame_channel,64 media_transport,65 media_generation,66 connected_audio_generation,67 } = request;6869 let metrics_started = Instant::now();70 let mut initial_metrics = ClientMediaMetrics::default();71 // Joining a running encoder commonly starts in the middle of a GOP.72 // The initial keyframe wait is a real resync episode, just like a wait73 // entered after frontend backpressure, so account for it from t=0.74 initial_metrics.begin_frontend_resync(Duration::ZERO);75 let metrics = Arc::new(StdMutex::new(initial_metrics));76 let mut previous_sequence: Option<u64> = None;77 let mut frontend_waiting_for_keyframe = true;78 let mut media_objects_v3 = (media_transport == MediaTransport::GroupedObjectsV3)79 .then(|| MediaObjectReceiverV3::new(frame_connection_for_stats.clone()));80 let mut media_object_sequence_v3 = MediaObjectSequenceV3::new();81 let mut upstream_moq_media = upstream_moq_media;8283 let (stats_stop, mut stats_stop_rx) = tokio::sync::watch::channel(false);84 let stats_app = app.clone();85 let stats_metrics = Arc::clone(&metrics);86 let stats_in_flight = Arc::clone(&frame_events_in_flight);87 let stats_connection = frame_connection_for_stats.clone();88 let stats_input_connection = input_connection.clone();89 let stats_audio_connection = audio_connection_for_stats.clone();90 let stats_network_diagnostics = Arc::clone(&network_diagnostics);91 let stats_task = tokio::spawn(async move {92 let mut interval = tokio::time::interval(CLIENT_FRAME_STATS_INTERVAL);93 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);94 // Tokio intervals tick immediately once. Consume that tick so the95 // first payload represents a full diagnostics interval.96 interval.tick().await;97 let mut last_path_sample = Instant::now() - Duration::from_secs(1);9899 loop {100 tokio::select! {101 _ = interval.tick() => {102 if last_path_sample.elapsed() >= Duration::from_secs(1) {103 let mut diagnostics = lock_network_diagnostics(&stats_network_diagnostics);104 diagnostics.observe_connection(NetworkLeg::Media, &stats_connection);105 diagnostics.observe_connection(NetworkLeg::Input, &stats_input_connection);106 if let Some(audio_connection) = stats_audio_connection.as_ref() {107 diagnostics.observe_connection(NetworkLeg::Audio, audio_connection);108 }109 last_path_sample = Instant::now();110 }111 let queue_depth = stats_in_flight.load(Ordering::SeqCst);112 let network_snapshot = lock_network_diagnostics(&stats_network_diagnostics)113 .snapshot(Instant::now());114 let payload = lock_client_media_metrics(&stats_metrics).snapshot(115 metrics_started.elapsed(),116 queue_depth,117 network_snapshot,118 media_generation,119 );120 let _ = stats_app.emit("frame-stats", payload);121 }122 changed = stats_stop_rx.changed() => {123 if changed.is_err() || *stats_stop_rx.borrow() {124 break;125 }126 }127 }128 }129 });130131 'frames: loop {132 let (133 w,134 h,135 frame_buf,136 is_keyframe,137 codec,138 sequence,139 capture_timestamp_micros,140 pts_micros,141 discontinuity,142 codec_config,143 ) = match media_transport {144 MediaTransport::UpstreamMoq => {145 let receiver = upstream_moq_media146 .as_mut()147 .expect("upstream MoQ receiver must exist for MoQ transport");148 loop {149 let outcome = match receiver.next().await {150 Ok(Some(outcome)) => outcome,151 Ok(None) => {152 emit_frame_error(153 &app,154 media_generation,155 "Upstream MoQ video track closed",156 );157 break 'frames;158 }159 Err(error) => {160 emit_frame_error(&app, media_generation, error);161 break 'frames;162 }163 };164 match outcome {165 MoqMediaReadOutcome::Dropped { reason } => {166 frontend_waiting_for_keyframe = true;167 let mut metrics = lock_client_media_metrics(&metrics);168 metrics.observe_transport_object_drop(false);169 metrics.begin_frontend_resync(metrics_started.elapsed());170 try_queue_media_keyframe_request(171 media_control_requests.as_ref(),172 reason,173 receiver.last_frame_sequence(),174 );175 }176 MoqMediaReadOutcome::Malformed(error) => {177 emit_frame_error(&app, media_generation, error);178 break 'frames;179 }180 MoqMediaReadOutcome::Frame {181 frame,182 discontinuity,183 } => {184 let codec = match frame.header.codec {185 MediaCodec::H264 => "h264".to_string(),186 };187 break (188 u32::from(frame.header.width),189 u32::from(frame.header.height),190 frame.payload,191 frame.header.flags.contains(FrameFlags::KEYFRAME),192 codec,193 Some(frame.header.sequence),194 Some(frame.header.capture_timestamp_us),195 Some(frame.header.pts_us),196 discontinuity,197 frame.header.flags.contains(FrameFlags::CODEC_CONFIG),198 );199 }200 }201 }202 }203 MediaTransport::GroupedObjectsV3 => {204 let receiver = media_objects_v3205 .as_mut()206 .expect("media v3 receiver must exist for media v3 transport");207 loop {208 let outcome = match receiver.next().await {209 Ok(Some(outcome)) => outcome,210 Ok(None) => {211 emit_frame_error(&app, media_generation, "Connection closed");212 break 'frames;213 }214 Err(error) => {215 emit_frame_error(&app, media_generation, error);216 break 'frames;217 }218 };219 match outcome {220 MediaObjectReadOutcomeV3::Dropped { reason, .. } => {221 let begins_resync = media_object_sequence_v3.note_dropped_object();222 let mut metrics = lock_client_media_metrics(&metrics);223 metrics.observe_transport_object_drop(!begins_resync);224 if begins_resync {225 frontend_waiting_for_keyframe = true;226 metrics.begin_frontend_resync(metrics_started.elapsed());227 try_queue_media_keyframe_request(228 media_control_requests.as_ref(),229 reason,230 media_object_sequence_v3.last_sequence(),231 );232 }233 }234 MediaObjectReadOutcomeV3::Malformed(error) => {235 emit_frame_error(&app, media_generation, error);236 break 'frames;237 }238 MediaObjectReadOutcomeV3::Object { object, .. } => {239 let was_waiting = media_object_sequence_v3.waiting_for_keyframe();240 let discontinuity = match media_object_sequence_v3.classify(&object) {241 MediaObjectSequenceDecisionV3::Deliver { discontinuity } => {242 discontinuity243 }244 MediaObjectSequenceDecisionV3::DropLate => {245 lock_client_media_metrics(&metrics)246 .observe_transport_object_drop(true);247 continue;248 }249 MediaObjectSequenceDecisionV3::DropUntilKeyframe => {250 frontend_waiting_for_keyframe = true;251 let mut metrics = lock_client_media_metrics(&metrics);252 metrics.observe_transport_object_drop(false);253 metrics.begin_frontend_resync(metrics_started.elapsed());254 if !was_waiting {255 try_queue_media_keyframe_request(256 media_control_requests.as_ref(),257 KeyframeRequestReasonV3::TransportGap,258 media_object_sequence_v3.last_sequence(),259 );260 }261 continue;262 }263 };264 let codec = match object.header.codec {265 MediaCodec::H264 => "h264".to_string(),266 };267 break (268 u32::from(object.header.width),269 u32::from(object.header.height),270 object.payload,271 object.header.flags.contains(FrameFlags::KEYFRAME),272 codec,273 Some(object.header.sequence),274 Some(object.header.capture_timestamp_us),275 Some(object.header.pts_us),276 discontinuity,277 object.header.flags.contains(FrameFlags::CODEC_CONFIG),278 );279 }280 }281 }282 }283 };284 lock_client_media_metrics(&metrics).observe_transport_receive(285 metrics_started.elapsed(),286 sequence,287 is_keyframe,288 );289290 let sequence_gap = match sequence.zip(previous_sequence) {291 Some((current, previous)) => match sequence_gap(previous, current) {292 Ok(gap) => gap,293 Err(error) => {294 emit_frame_error(&app, media_generation, error);295 break;296 }297 },298 None => 0,299 };300 if sequence.is_some() {301 previous_sequence = sequence;302 }303 lock_client_media_metrics(&metrics).observe_sequence_drop(sequence_gap);304305 if frontend_waiting_for_keyframe && !is_keyframe {306 lock_client_media_metrics(&metrics).observe_frontend_resync_drop();307 continue;308 }309 if !try_reserve_frame_channel_slot(&frame_events_in_flight) {310 let begins_resync = !frontend_waiting_for_keyframe;311 frontend_waiting_for_keyframe = true;312 let mut metrics = lock_client_media_metrics(&metrics);313 metrics.observe_frontend_queue_drop();314 metrics.begin_frontend_resync(metrics_started.elapsed());315 if begins_resync {316 try_queue_media_keyframe_request(317 media_control_requests.as_ref(),318 KeyframeRequestReasonV3::FrontendBackpressure,319 sequence,320 );321 }322 continue;323 }324 lock_client_media_metrics(&metrics)325 .observe_frontend_queue_depth(frame_events_in_flight.load(Ordering::SeqCst));326327 let envelope = match encode_frame_envelope(328 FrameEnvelopeMetadata {329 width: w,330 height: h,331 codec: &codec,332 keyframe: is_keyframe,333 discontinuity,334 codec_config,335 sequence,336 capture_timestamp_micros,337 pts_micros,338 },339 &frame_buf,340 ) {341 Ok(envelope) => envelope,342 Err(error) => {343 emit_frame_error(&app, media_generation, error);344 release_frame_channel_slot(&frame_events_in_flight);345 break;346 }347 };348 let ipc_send_started = Instant::now();349 let send_result = frame_channel.send(Response::new(envelope));350 lock_client_media_metrics(&metrics).observe_frontend_ipc_send_duration(351 metrics_started.elapsed(),352 ipc_send_started.elapsed(),353 );354 if send_result.is_err() {355 release_frame_channel_slot(&frame_events_in_flight);356 break;357 }358359 frontend_waiting_for_keyframe = false;360 {361 let mut metrics = lock_client_media_metrics(&metrics);362 let elapsed = metrics_started.elapsed();363 metrics.observe_frontend_send(elapsed);364 metrics.finish_frontend_resync(elapsed);365 }366367 tokio::task::yield_now().await;368 }369370 let _ = stats_stop.send(true);371 let _ = stats_task.await;372 if media_transport == MediaTransport::UpstreamMoq {373 retire_upstream_moq_generation(&app, media_generation, connected_audio_generation).await;374 }375 drop(endpoint);376}377378fn sequence_gap(previous: u64, current: u64) -> Result<u64, String> {379 let expected = previous380 .checked_add(1)381 .ok_or_else(|| format!("Media sequence overflowed after {previous}"))?;382 if current < expected {383 return Err(format!(384 "Non-monotonic media sequence: previous={previous}, current={current}"385 ));386 }387 Ok(current - expected)388}389390391mod tests {392 use super::*;393394 395 fn sequence_checks_reject_duplicates_regressions_and_overflow() {396 assert_eq!(sequence_gap(41, 42).unwrap(), 0);397 assert_eq!(sequence_gap(41, 45).unwrap(), 3);398 assert!(sequence_gap(41, 41).is_err());399 assert!(sequence_gap(41, 40).is_err());400 assert!(sequence_gap(u64::MAX, 0).is_err());401 }402}