diff --git a/crates/muninn-daemon/src/main.rs b/crates/muninn-daemon/src/main.rs index 7857ad3..0b69938 100644 --- a/crates/muninn-daemon/src/main.rs +++ b/crates/muninn-daemon/src/main.rs @@ -1,10 +1,11 @@ mod idunn_presence; mod media_packetizer; +mod media_send; -use crate::media_packetizer::{ - AudioPcmStreamSendConfig, AudioPcmStreamSendState, GameCultMediaWireRecord, - MuninnMediaSendPayload, VideoAnnexBStreamSendConfig, VideoAnnexBStreamSendState, - decode_media_wire_record, +use crate::media_packetizer::{AudioPcmStreamSendConfig, VideoAnnexBStreamSendConfig}; +use crate::media_send::{ + MediaSendCore, MediaSendPolicy, media_intake_channel, spawn_audio_group_reader, + spawn_video_group_reader, }; use anyhow::{Context, Result, anyhow}; use cultmesh_rs::{ @@ -12,12 +13,11 @@ use cultmesh_rs::{ CultMeshSharedMemoryFrameRing, CultMeshStreamBodyTransport, CultMeshStreamCatalog, CultMeshStreamClock, CultMeshStreamDescriptor, CultMeshStreamKind, }; -use cultnet_rs::GameCultMediaReceiverFeedbackRecord; use cultnet_rs::{ - CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, CultNetRudpServerEvent, - CultNetRudpServerHub, CultNetRudpServerHubOptions, CultNetRudpServerSessionContext, - CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, CultNetTransportFrame, - CultNetWireContract, GAMECULT_MEDIA_STREAM_ADVERTISEMENT_SCHEMA, + CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, CultNetRudpServerHub, + CultNetRudpServerHubOptions, CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportOptions, CultNetWireContract, + GAMECULT_MEDIA_STREAM_ADVERTISEMENT_SCHEMA, GAMECULT_MEDIA_STREAM_REQUEST_SCHEMA, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, GameCultMediaStreamAdvertisementRecord, GameCultMediaStreamRequestRecord, GameCultRuntimePresenceHealthRecord, decode_cultnet_message_from_slice, @@ -37,7 +37,7 @@ use odin_core::{ }; use serde::{Deserialize, Serialize}; use serde_json::json; -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{HashMap, HashSet}; use std::env; use std::fs; #[cfg(feature = "psmoveapi-tracker")] @@ -52,8 +52,9 @@ use std::process::{Child, Command, Stdio}; use std::sync::{ Arc, Mutex, atomic::{AtomicU64, Ordering}, - mpsc, }; +#[cfg(any(windows, feature = "psmoveapi-tracker"))] +use std::sync::mpsc; use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -90,25 +91,14 @@ const MUNINN_RUDP_MEDIA_MAX_FRAGMENT_BYTES: usize = MUNINN_RUDP_IPV4_UDP_PAYLOAD /// nearly every audio fragment was resent before its ack arrived and the /// producer's send loop drowned in retransmits, starving video (measured /// 2026-09-10: 7 frames in 9 s with audio on, 266 without). Audio expires at -/// the assembly deadline anyway; 30 ms costs nothing it can feel. +/// the latency budget anyway; 30 ms costs nothing it can feel. const MUNINN_RUDP_MEDIA_RESEND_DELAY_MS: u64 = 30; -const MUNINN_RUDP_MEDIA_RECEIVER_ASSEMBLY_DEADLINE_MS: u64 = 2_000; +/// What a request that names no latency budget gets, and what the +/// advertisement offers as `default_latency_budget_ms`. It is the request, +/// not this constant, that owns the deadline once one is named. +const MUNINN_RUDP_MEDIA_DEFAULT_LATENCY_BUDGET_MS: u64 = 250; const MUNINN_RUDP_MEDIA_RECEIVER_GAP_WAIT_MS: u64 = 16; -const MUNINN_RUDP_MEDIA_REPAIR_CACHE_CHUNKS: usize = 16_384; -const MUNINN_RUDP_MEDIA_REPAIR_BURST_CHUNKS: usize = 2_048; -const MUNINN_RUDP_MEDIA_REPAIR_INITIAL_CHUNKS_PER_SECOND: usize = 4_096; -const MUNINN_RUDP_MEDIA_REPAIR_MIN_CHUNKS_PER_SECOND: usize = 8; -const MUNINN_RUDP_MEDIA_REPAIR_MAX_CHUNKS_PER_SECOND: usize = 16_384; -const MUNINN_RUDP_MEDIA_REPAIR_ADD_CHUNKS_PER_SECOND: usize = 2_048; -const MUNINN_RUDP_MEDIA_REPAIR_RECOVERY_INTERVAL_MS: u64 = 2_000; -const MUNINN_RUDP_MEDIA_REPAIR_MAX_FEEDBACK_PER_POLL: usize = 32; -const MUNINN_RUDP_MEDIA_REPAIR_MAX_CHUNKS_PER_POLL: usize = 256; const MUNINN_RUDP_MEDIA_SOCKET_BUFFER_BYTES: usize = 16 * 1024 * 1024; -/// A receiver that has said nothing for this long (Ratatoskr pings every -/// second or so) is dropped so its payloads stop being sent into the void. -const MUNINN_RUDP_MEDIA_RECEIVER_TIMEOUT_MS: u64 = 6_000; -const MUNINN_RUDP_MEDIA_SEND_PACE_EVERY_PAYLOADS: usize = 4; -const MUNINN_RUDP_MEDIA_SEND_PACE_SLEEP_US: u64 = 250; const MUNINN_RUDP_ACTIVE_CATALOG_REPUBLISH_MS: u64 = 2_000; const MUNINN_ODIN_PROVIDER_LEASE_REFRESH_SECONDS: u64 = 30; const PS_MOVE_LED_REPORT_LEN: usize = 49; @@ -275,12 +265,7 @@ struct MuninnRudpMediaProfile { max_fragment_bytes: usize, video_b_frames: u8, video_rc_lookahead: u8, - sender_queue_deadline_ms: u64, sender_resend_delay_ms: u64, - sender_reliable_expire_after_ms: u64, - sender_pace_every_payloads: usize, - sender_pace_sleep_us: u64, - receiver_assembly_deadline_ms: u64, receiver_gap_wait_ms: u64, } @@ -1488,7 +1473,7 @@ fn command_rudp_video_bitrate_kbps(command: &MuninnCaptureStreamCommandRecord) - fn command_rudp_latency_budget_ms(command: &MuninnCaptureStreamCommandRecord) -> u32 { if command.rudp_latency_budget_ms == 0 { - MUNINN_RUDP_MEDIA_RECEIVER_ASSEMBLY_DEADLINE_MS as u32 + MUNINN_RUDP_MEDIA_DEFAULT_LATENCY_BUDGET_MS as u32 } else { command.rudp_latency_budget_ms } @@ -2415,7 +2400,8 @@ fn run_rudp_mux_once( let mut audio_sender = None; let result = (|| -> Result { let media_profile = muninn_rudp_media_profile_for_options(options); - let mut hub = open_media_rudp_hub(options, &media_profile)?; + let policy = MediaSendPolicy::from_request(options.rudp_latency_budget_ms); + let hub = open_media_rudp_hub(options, &media_profile, &policy)?; publish_stream( node, options, @@ -2431,10 +2417,10 @@ fn run_rudp_mux_once( )?; let mut last_stream_publish_at = Instant::now(); - let (payload_tx, payload_rx) = mpsc::channel::>(); + let (intake_tx, intake_rx) = media_intake_channel(); video_sender = if let Some(stdout) = video_ffmpeg_stdout { - Some(video_rudp_payload_reader( - payload_tx.clone(), + Some(spawn_video_group_reader( + intake_tx.clone(), stdout, VideoAnnexBStreamSendConfig { stream_id: options.stream_id.clone(), @@ -2445,19 +2431,20 @@ fn run_rudp_mux_once( frame_duration_ticks: video_frame_duration_ticks(options)?, timebase_num: 1, timebase_den: 90_000, - deadline_delay_ticks: rudp_media_deadline_delay_ticks(&media_profile), + deadline_delay_ticks: policy.video_deadline_delay_ticks(), max_payload_bytes: options.media_packet_bytes.max(256), max_pending_bytes: options.media_packet_bytes.max(256) * 4096, source_runtime_id: options.host_id.clone(), source_role: "muninn.rudp.video".to_string(), }, + policy.clone(), )) } else { None }; audio_sender = if let Some(stdout) = audio_ffmpeg_stdout { - Some(audio_rudp_payload_reader( - payload_tx.clone(), + Some(spawn_audio_group_reader( + intake_tx.clone(), stdout, AudioPcmStreamSendConfig { stream_id: options.stream_id.clone(), @@ -2468,7 +2455,7 @@ fn run_rudp_mux_once( packet_duration_ticks: 480, timebase_num: 1, timebase_den: options.audio_sample_rate, - deadline_delay_ticks: rudp_audio_deadline_delay_ticks(options, &media_profile), + deadline_delay_ticks: policy.audio_deadline_delay_ticks(options.audio_sample_rate), channels: options.audio_channels, bytes_per_sample: 4, max_pending_bytes: 480usize @@ -2478,193 +2465,39 @@ fn run_rudp_mux_once( source_runtime_id: options.host_id.clone(), source_role: "muninn.rudp.audio".to_string(), }, + policy.clone(), )) } else { None }; - drop(payload_tx); + drop(intake_tx); - let mut payloads_sent = 0_u64; - let mut payloads_queue_expired = 0_u64; - let mut payloads_send_expired = 0_u64; - let mut receiver_feedback = MuninnRudpReceiverFeedbackStats::default(); - let mut handled_keyframe_requests = 0_u64; - let mut repair_cache = - RecentVideoChunkRepairCache::new(MUNINN_RUDP_MEDIA_REPAIR_CACHE_CHUNKS); - let mut repair_budget = MuninnRudpRepairBudget::new( - MUNINN_RUDP_MEDIA_REPAIR_INITIAL_CHUNKS_PER_SECOND, - MUNINN_RUDP_MEDIA_REPAIR_BURST_CHUNKS, - ); - let mut video_send_pacer = MuninnRudpMediaSendPacer::new( - media_profile.sender_pace_every_payloads, - Duration::from_micros(media_profile.sender_pace_sleep_us), - ); - let mut audio_send_pacer = MuninnRudpMediaSendPacer::new(0, Duration::ZERO); - let mut pending_payloads = PendingMuninnMediaSendQueues::default(); - let mut payload_channel_disconnected = false; + let mut core = MediaSendCore::new(hub, &policy); loop { - if pending_payloads.is_empty() && !payload_channel_disconnected { - payload_channel_disconnected = receive_pending_media_payloads( - &payload_rx, - &mut pending_payloads, - Duration::from_millis(5), - )?; - } else if !payload_channel_disconnected { - payload_channel_disconnected = - drain_available_media_payloads(&payload_rx, &mut pending_payloads)?; - } - - if let Some(queued) = pending_payloads.pop_next() { - if media_payload_queue_age_exceeded( - queued.queued_at, - Instant::now(), - Duration::from_millis(media_profile.sender_queue_deadline_ms), - ) { - payloads_queue_expired += 1; - let payloads_dropped = payloads_queue_expired + payloads_send_expired; - // A stale payload is already unsendable. Do not make - // queue recovery slower by polling feedback, sending - // repairs, and polling retransmits once for every stale - // item. Normal live-send and idle paths service those - // controls as soon as the expired backlog is drained. - if payloads_dropped == 1 || payloads_dropped % 300 == 0 { - let expired = reliable_packets_expired(&hub); - eprintln!( - "{}", - rudp_media_progress_detail( - payloads_sent, - payloads_dropped, - payloads_queue_expired, - payloads_send_expired, - expired, - hub.sessions().len(), - &receiver_feedback - ) - ); - } - continue; - } - - let payload_len = queued.payload.payload.len(); - let send_pacer = match queued.kind { - QueuedMuninnMediaKind::Video => &mut video_send_pacer, - QueuedMuninnMediaKind::Audio => &mut audio_send_pacer, - }; - let sent = send_rudp_media_payload_with_backpressure( - &mut hub, - None, - queued.payload.clone(), - queued.queued_at, - Duration::from_millis(media_profile.sender_queue_deadline_ms), - send_pacer, - )?; - if !sent { - payloads_send_expired += 1; - continue; - } - let payloads_dropped = payloads_queue_expired + payloads_send_expired; - if queued.kind == QueuedMuninnMediaKind::Video { - repair_cache.remember(&queued.payload)?; - } - poll_rudp_media_receiver_feedback( - &mut hub, - &mut receiver_feedback, - &repair_cache, - &mut repair_budget, - &media_profile, - &mut video_send_pacer, - payloads_dropped, - )?; - record_receiver_keyframe_pressure( - &receiver_feedback, - &mut handled_keyframe_requests, - ); - poll_rudp_resends_with_backpressure(&mut hub)?; - republish_running_stream_if_due( - node, - options, - plan, - supervisor_pid, - video_ffmpeg - .as_ref() - .map(Child::id) - .or_else(|| audio_ffmpeg.as_ref().map(Child::id)) - .unwrap_or(supervisor_pid), - restart_count, - &mut last_stream_publish_at, - )?; - payloads_sent += 1; - if payloads_sent == 1 || payloads_sent % 900 == 0 { - let expired = reliable_packets_expired(&hub); - eprintln!( - "{}; pending_audio={} pending_video={}; latest {:?} payload was {payload_len} bytes.", - rudp_media_progress_detail( - payloads_sent, - payloads_dropped, - payloads_queue_expired, - payloads_send_expired, - expired, - hub.sessions().len(), - &receiver_feedback - ), - pending_payloads.audio_len(), - pending_payloads.video_len(), - queued.kind - ); - } - continue; - } - - if payload_channel_disconnected { - let expired = reliable_packets_expired(&hub); - let payloads_dropped = payloads_queue_expired + payloads_send_expired; + core.intake(&intake_rx, Duration::from_millis(5))?; + if core.send_next(Instant::now())? { + core.service_control(Instant::now())?; + } else if core.finished() { break Ok(RudpMuxRestart { - detail: format!( - "encoder stdout ended; {}", - rudp_media_progress_detail( - payloads_sent, - payloads_dropped, - payloads_queue_expired, - payloads_send_expired, - expired, - hub.sessions().len(), - &receiver_feedback - ) - ), + detail: format!("encoder stdout ended; {}", core.progress_detail()), delay: default_rudp_mux_restart_delay(restart_count), }); + } else { + core.service_control(Instant::now())?; } - - { - let payloads_dropped = payloads_queue_expired + payloads_send_expired; - poll_rudp_media_receiver_feedback( - &mut hub, - &mut receiver_feedback, - &repair_cache, - &mut repair_budget, - &media_profile, - &mut video_send_pacer, - payloads_dropped, - )?; - record_receiver_keyframe_pressure( - &receiver_feedback, - &mut handled_keyframe_requests, - ); - poll_rudp_resends_with_backpressure(&mut hub)?; - republish_running_stream_if_due( - node, - options, - plan, - supervisor_pid, - video_ffmpeg - .as_ref() - .map(Child::id) - .or_else(|| audio_ffmpeg.as_ref().map(Child::id)) - .unwrap_or(supervisor_pid), - restart_count, - &mut last_stream_publish_at, - )?; - } + republish_running_stream_if_due( + node, + options, + plan, + supervisor_pid, + video_ffmpeg + .as_ref() + .map(Child::id) + .or_else(|| audio_ffmpeg.as_ref().map(Child::id)) + .unwrap_or(supervisor_pid), + restart_count, + &mut last_stream_publish_at, + )?; } })(); @@ -2692,506 +2525,17 @@ fn run_rudp_mux_once( result } -struct QueuedMuninnMediaSendPayload { - payload: MuninnMediaSendPayload, - queued_at: Instant, - kind: QueuedMuninnMediaKind, -} - -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum QueuedMuninnMediaKind { - Video, - Audio, -} - -#[derive(Default)] -struct PendingMuninnMediaSendQueues { - audio: VecDeque, - video: VecDeque, -} - -impl PendingMuninnMediaSendQueues { - fn push(&mut self, payload: QueuedMuninnMediaSendPayload) { - match payload.kind { - QueuedMuninnMediaKind::Audio => self.audio.push_back(payload), - QueuedMuninnMediaKind::Video => self.video.push_back(payload), - } - } - - fn pop_next(&mut self) -> Option { - self.audio.pop_front().or_else(|| self.video.pop_front()) - } - - fn is_empty(&self) -> bool { - self.audio.is_empty() && self.video.is_empty() - } - - fn audio_len(&self) -> usize { - self.audio.len() - } - - fn video_len(&self) -> usize { - self.video.len() - } -} - -#[derive(Clone, Debug, Default, PartialEq, Eq)] -struct MuninnRudpReceiverFeedbackStats { - feedback_records: u64, - requested_keyframes: u64, - late_frames: u64, - missing_video_chunks: u64, - repaired_video_chunks: u64, - deferred_repair_chunks: u64, - repair_chunks_per_second: usize, - highest_decodable_frame_id: Option, -} - -#[derive(Debug)] -struct MuninnRudpRepairBudget { - chunks_per_second: usize, - min_chunks_per_second: usize, - max_chunks_per_second: usize, - add_chunks_per_second: usize, - recovery_interval: Duration, - max_available_chunks: usize, - available_chunks: usize, - last_refill_at: Instant, - last_rate_adjust_at: Instant, - last_queue_dropped: u64, -} - -impl MuninnRudpRepairBudget { - fn new(chunks_per_second: usize, max_available_chunks: usize) -> Self { - let now = Instant::now(); - let max_available_chunks = max_available_chunks.max(1); - Self { - chunks_per_second: chunks_per_second.max(1), - min_chunks_per_second: MUNINN_RUDP_MEDIA_REPAIR_MIN_CHUNKS_PER_SECOND, - max_chunks_per_second: MUNINN_RUDP_MEDIA_REPAIR_MAX_CHUNKS_PER_SECOND, - add_chunks_per_second: MUNINN_RUDP_MEDIA_REPAIR_ADD_CHUNKS_PER_SECOND, - recovery_interval: Duration::from_millis(MUNINN_RUDP_MEDIA_REPAIR_RECOVERY_INTERVAL_MS), - max_available_chunks, - available_chunks: max_available_chunks, - last_refill_at: now, - last_rate_adjust_at: now, - last_queue_dropped: 0, - } - } - - fn chunks_per_second(&self) -> usize { - self.chunks_per_second - } - - fn take(&mut self, requested_chunks: usize, now: Instant, queue_dropped: u64) -> usize { - self.adjust_rate(now, queue_dropped, requested_chunks); - self.refill(now); - let allowed = requested_chunks.min(self.available_chunks); - self.available_chunks -= allowed; - allowed - } - - fn adjust_rate(&mut self, now: Instant, queue_dropped: u64, requested_chunks: usize) { - if queue_dropped > self.last_queue_dropped { - self.chunks_per_second = (self.chunks_per_second / 2).max(self.min_chunks_per_second); - self.available_chunks = self.available_chunks.min(self.chunks_per_second); - self.last_queue_dropped = queue_dropped; - self.last_rate_adjust_at = now; - return; - } - self.last_queue_dropped = queue_dropped; - if requested_chunks == 0 - || now.saturating_duration_since(self.last_rate_adjust_at) < self.recovery_interval - { - return; - } - if self.chunks_per_second < self.max_chunks_per_second { - self.chunks_per_second = self - .chunks_per_second - .saturating_add(self.add_chunks_per_second) - .min(self.max_chunks_per_second); - self.last_rate_adjust_at = now; - } - } - - fn refill(&mut self, now: Instant) { - let elapsed_ms = now - .saturating_duration_since(self.last_refill_at) - .as_millis() as usize; - if elapsed_ms == 0 { - return; - } - let refill_chunks = elapsed_ms.saturating_mul(self.chunks_per_second) / 1_000; - if refill_chunks == 0 { - return; - } - self.available_chunks = self - .available_chunks - .saturating_add(refill_chunks) - .min(self.max_available_chunks); - self.last_refill_at = now; - } -} - -#[derive(Debug)] -struct RecentVideoChunkRepairCache { - max_entries: usize, - order: VecDeque, - entries: HashMap, -} - -impl RecentVideoChunkRepairCache { - fn new(max_entries: usize) -> Self { - Self { - max_entries, - order: VecDeque::new(), - entries: HashMap::new(), - } - } - - fn remember(&mut self, payload: &MuninnMediaSendPayload) -> Result<()> { - let Some(key) = video_repair_cache_key_from_payload(payload)? else { - return Ok(()); - }; - if self.entries.contains_key(&key) { - self.entries.insert(key, payload.clone()); - return Ok(()); - } - self.order.push_back(key.clone()); - self.entries.insert(key, payload.clone()); - while self.entries.len() > self.max_entries { - let Some(expired) = self.order.pop_front() else { - break; - }; - self.entries.remove(&expired); - } - Ok(()) - } - - fn repair_payloads_for_feedback( - &self, - feedback: &GameCultMediaReceiverFeedbackRecord, - ) -> Vec { - feedback - .missing_video_chunk_keys - .iter() - .filter_map(|chunk_key| { - self.entries - .get(&video_repair_cache_key( - &feedback.stream_id, - &feedback.session_id, - chunk_key, - )) - .cloned() - }) - .collect() - } -} - -fn video_repair_cache_key(stream_id: &str, session_id: &str, chunk_key: &str) -> String { - format!("{stream_id}:{session_id}:video:{chunk_key}") -} - -fn video_repair_cache_key_from_payload(payload: &MuninnMediaSendPayload) -> Result> { - if payload.channel_id != crate::media_packetizer::MUNINN_MEDIA_RUDP_CHANNEL { - return Ok(None); - } - let GameCultMediaWireRecord::Video(video) = decode_media_wire_record(&payload.payload)? else { - return Ok(None); - }; - Ok(Some(video_repair_cache_key( - &video.stream_id, - &video.session_id, - &crate::media_packetizer::video_chunk_feedback_key(video.frame_id, video.chunk_index), - ))) -} - -fn queue_muninn_media_payload( - tx: &mpsc::Sender>, - payload: MuninnMediaSendPayload, - kind: QueuedMuninnMediaKind, -) -> Result<()> { - tx.send(Ok(QueuedMuninnMediaSendPayload { - payload, - queued_at: Instant::now(), - kind, - })) - .context("queueing typed Muninn media payload") -} - -fn receive_pending_media_payloads( - rx: &mpsc::Receiver>, - pending: &mut PendingMuninnMediaSendQueues, - timeout: Duration, -) -> Result { - match rx.recv_timeout(timeout) { - Ok(Ok(payload)) => { - pending.push(payload); - drain_available_media_payloads(rx, pending) - } - Ok(Err(error)) => Err(error), - Err(mpsc::RecvTimeoutError::Timeout) => Ok(false), - Err(mpsc::RecvTimeoutError::Disconnected) => Ok(true), - } -} - -fn drain_available_media_payloads( - rx: &mpsc::Receiver>, - pending: &mut PendingMuninnMediaSendQueues, -) -> Result { - loop { - match rx.try_recv() { - Ok(Ok(payload)) => pending.push(payload), - Ok(Err(error)) => return Err(error), - Err(mpsc::TryRecvError::Empty) => return Ok(false), - Err(mpsc::TryRecvError::Disconnected) => return Ok(true), - } - } -} - -fn media_payload_queue_age_exceeded(queued_at: Instant, now: Instant, max_age: Duration) -> bool { - now.saturating_duration_since(queued_at) > max_age -} - -fn poll_rudp_resends_with_backpressure(hub: &mut CultNetRudpServerHub) -> Result<()> { - match hub.poll_resends() { - Ok(()) => Ok(()), - Err(error) if is_would_block_error(&error) => Ok(()), - Err(error) => Err(error).context("polling Muninn RUDP media resends"), - } -} - -/// To one receiver, or with `None` to every receiver attached right now. A -/// payload with nobody to send it to is not a failure; the progress line -/// says how many receivers there are. -fn send_rudp_media_payload_with_backpressure( - hub: &mut CultNetRudpServerHub, - receiver: Option<&CultNetRudpServerSessionContext>, - payload: MuninnMediaSendPayload, - queued_at: Instant, - max_age: Duration, - send_pacer: &mut MuninnRudpMediaSendPacer, -) -> Result { - let receivers = match receiver { - Some(receiver) => vec![receiver.clone()], - None => hub.sessions(), - }; - for receiver in &receivers { - loop { - match hub.send(receiver, payload.channel_id, payload.payload.clone()) { - Ok(()) => break, - Err(error) if is_would_block_error(&error) => { - if media_payload_queue_age_exceeded(queued_at, Instant::now(), max_age) { - return Ok(false); - } - poll_rudp_resends_with_backpressure(hub)?; - thread::sleep(Duration::from_millis(1)); - } - Err(error) => { - return Err(error) - .context("sending typed Muninn media payload over RUDP media"); - } - } - } - } - send_pacer.observe_sent_payload(); - Ok(true) -} - -#[derive(Debug)] -struct MuninnRudpMediaSendPacer { - payloads_since_pause: usize, - every_payloads: usize, - sleep_for: Duration, -} - -impl MuninnRudpMediaSendPacer { - fn new(every_payloads: usize, sleep_for: Duration) -> Self { - Self { - payloads_since_pause: 0, - every_payloads: every_payloads.max(1), - sleep_for, - } - } - - fn observe_sent_payload(&mut self) { - if self.sleep_for.is_zero() { - return; - } - self.payloads_since_pause = self.payloads_since_pause.saturating_add(1); - if self.payloads_since_pause < self.every_payloads { - return; - } - self.payloads_since_pause = 0; - thread::sleep(self.sleep_for); - } -} - -fn is_would_block_error(error: &anyhow::Error) -> bool { - error.chain().any(|cause| { - cause - .downcast_ref::() - .is_some_and(|io| io.kind() == ErrorKind::WouldBlock) - || cause.to_string().contains("os error 10035") - }) -} - fn default_rudp_mux_restart_delay(restart_count: u32) -> Duration { Duration::from_secs((2 + restart_count).min(30) as u64) } -fn record_receiver_keyframe_pressure( - receiver_feedback: &MuninnRudpReceiverFeedbackStats, - handled_keyframe_requests: &mut u64, -) { - if receiver_feedback.requested_keyframes <= *handled_keyframe_requests { - return; - } - *handled_keyframe_requests = receiver_feedback.requested_keyframes; - eprintln!( - "Muninn RUDP receiver requested a fresh keyframe; continuing current low-latency encoder session until explicit encoder control exists." - ); -} - -fn rudp_media_progress_detail( - sent: u64, - queue_dropped: u64, - queue_expired: u64, - send_expired: u64, - reliable_expired: u64, - receivers: usize, - receiver_feedback: &MuninnRudpReceiverFeedbackStats, -) -> String { - format!( - "Muninn RUDP media progress: receivers={receivers} sent={sent} queue_dropped={queue_dropped} queue_expired={queue_expired} send_expired={send_expired} reliable_expired={reliable_expired} receiver_feedback={} receiver_keyframes={} receiver_late_frames={} receiver_missing_chunks={} receiver_repaired_chunks={} receiver_deferred_repairs={} repair_rate={} receiver_highest_decodable={}", - receiver_feedback.feedback_records, - receiver_feedback.requested_keyframes, - receiver_feedback.late_frames, - receiver_feedback.missing_video_chunks, - receiver_feedback.repaired_video_chunks, - receiver_feedback.deferred_repair_chunks, - receiver_feedback.repair_chunks_per_second, - receiver_feedback - .highest_decodable_frame_id - .map(|frame_id| frame_id.to_string()) - .unwrap_or_else(|| "none".to_string()) - ) -} - -fn poll_rudp_media_receiver_feedback( - hub: &mut CultNetRudpServerHub, - stats: &mut MuninnRudpReceiverFeedbackStats, - repair_cache: &RecentVideoChunkRepairCache, - repair_budget: &mut MuninnRudpRepairBudget, - media_profile: &MuninnRudpMediaProfile, - send_pacer: &mut MuninnRudpMediaSendPacer, - queue_dropped: u64, -) -> Result<()> { - for gone in hub.remove_timed_out_sessions(MUNINN_RUDP_MEDIA_RECEIVER_TIMEOUT_MS) { - eprintln!( - "Muninn media receiver {} at {} went silent and was dropped.", - String::from_utf8_lossy(&gone.connect_payload), - gone.remote_addr - ); - } - let mut feedback_processed = 0_usize; - loop { - match hub.receive_event_once() { - Ok(Some(CultNetRudpServerEvent::Connected { session })) => { - eprintln!( - "Muninn media receiver {} attached from {}.", - String::from_utf8_lossy(&session.connect_payload), - session.remote_addr - ); - } - Ok(Some(CultNetRudpServerEvent::Disconnected { session, reason })) => { - eprintln!( - "Muninn media receiver {} at {} disconnected: {}.", - String::from_utf8_lossy(&session.connect_payload), - session.remote_addr, - String::from_utf8_lossy(&reason) - ); - } - Ok(Some(CultNetRudpServerEvent::Frame { session, frame })) => { - if feedback_processed >= MUNINN_RUDP_MEDIA_REPAIR_MAX_FEEDBACK_PER_POLL { - return Ok(()); - } - feedback_processed += 1; - let repair_payloads = - record_rudp_media_receiver_feedback(&frame, stats, repair_cache)?; - let requested_repairs = repair_payloads.len(); - let poll_limited_repairs = - requested_repairs.min(MUNINN_RUDP_MEDIA_REPAIR_MAX_CHUNKS_PER_POLL); - let allowed_repairs = - repair_budget.take(poll_limited_repairs, Instant::now(), queue_dropped); - stats.repair_chunks_per_second = repair_budget.chunks_per_second(); - stats.deferred_repair_chunks = stats - .deferred_repair_chunks - .saturating_add(requested_repairs.saturating_sub(allowed_repairs) as u64); - // Repairs go back to the receiver that asked, not to everyone. - for payload in repair_payloads.into_iter().take(allowed_repairs) { - if send_rudp_media_payload_with_backpressure( - hub, - Some(&session), - payload, - Instant::now(), - Duration::from_millis(media_profile.sender_queue_deadline_ms), - send_pacer, - )? { - stats.repaired_video_chunks = stats.repaired_video_chunks.saturating_add(1); - } - } - } - Ok(Some(CultNetRudpServerEvent::Pong { .. })) => {} - Ok(None) => return Ok(()), - Err(error) if is_would_block_error(&error) => return Ok(()), - Err(error) => return Err(error).context("polling Muninn RUDP media feedback"), - } - } -} - -fn record_rudp_media_receiver_feedback( - frame: &CultNetTransportFrame, - stats: &mut MuninnRudpReceiverFeedbackStats, - repair_cache: &RecentVideoChunkRepairCache, -) -> Result> { - if frame.channel_id != crate::media_packetizer::MUNINN_MEDIA_RUDP_CHANNEL { - return Ok(Vec::new()); - } - - let GameCultMediaWireRecord::Feedback(feedback) = decode_media_wire_record(&frame.payload)? - else { - return Ok(Vec::new()); - }; - - let repair_payloads = repair_cache.repair_payloads_for_feedback(&feedback); - stats.feedback_records = stats.feedback_records.saturating_add(1); - if feedback.requested_keyframe { - stats.requested_keyframes = stats.requested_keyframes.saturating_add(1); - } - stats.late_frames = stats - .late_frames - .saturating_add(feedback.late_frame_ids.len() as u64); - stats.missing_video_chunks = stats - .missing_video_chunks - .saturating_add(feedback.missing_video_chunk_keys.len() as u64); - if let Some(frame_id) = feedback.highest_decodable_frame_id { - stats.highest_decodable_frame_id = Some( - stats - .highest_decodable_frame_id - .map(|current| current.max(frame_id)) - .unwrap_or(frame_id), - ); - } - Ok(repair_payloads) -} - /// Muninn listens for receivers here. It used to dial a receiver that had /// to open a port for it; that put a firewall rule on every viewer's /// machine. The producer is the host that serves, so the producer listens. fn open_media_rudp_hub( options: &Options, media_profile: &MuninnRudpMediaProfile, + policy: &MediaSendPolicy, ) -> Result { let socket = UdpSocket::bind(options.media_rudp_bind).with_context(|| { format!( @@ -3204,7 +2548,11 @@ fn open_media_rudp_hub( socket .set_nonblocking(true) .context("setting Muninn media RUDP listener nonblocking")?; - let hub = CultNetRudpServerHub::new(muninn_media_rudp_hub_options(socket, media_profile))?; + let hub = CultNetRudpServerHub::new(muninn_media_rudp_hub_options( + socket, + media_profile, + policy, + ))?; eprintln!( "Muninn media RUDP listening at {} for receivers dialling connection {:#x}.", hub.local_addr()?, @@ -3226,110 +2574,21 @@ fn configure_media_rudp_socket_buffers(socket: &UdpSocket) -> Result<()> { /// One session per receiver carries both legs: video lossy under parity on /// the media channel, audio reliable on its own channel and given up on -/// past the receiver's assembly deadline. +/// past the request's latency budget. fn muninn_media_rudp_hub_options( socket: UdpSocket, media_profile: &MuninnRudpMediaProfile, + policy: &MediaSendPolicy, ) -> CultNetRudpServerHubOptions { let mut options = CultNetRudpServerHubOptions::new("muninn-media", socket, MUNINN_MEDIA_RUDP_CONNECTION_ID); options.resend_delay_ms = media_profile.sender_resend_delay_ms; options.max_fragment_bytes = Some(media_profile.max_fragment_bytes as u32); options.media_delivery = Some(cultnet_rs::CultNetTransportDelivery::Unreliable); - options.media_reliable_expire_after_ms = Some(media_profile.receiver_assembly_deadline_ms); + options.media_reliable_expire_after_ms = Some(policy.latency_budget_ms()); options } -fn reliable_packets_expired(hub: &CultNetRudpServerHub) -> u64 { - hub.stats().reliable_packets_expired -} - -fn video_rudp_payload_reader( - tx: mpsc::Sender>, - mut reader: R, - config: VideoAnnexBStreamSendConfig, -) -> thread::JoinHandle<()> -where - R: Read + Send + 'static, -{ - thread::spawn(move || { - if let Err(error) = read_video_rudp_payloads(&tx, &mut reader, config) { - let _ = tx.send(Err(error)); - } - }) -} - -fn audio_rudp_payload_reader( - tx: mpsc::Sender>, - mut reader: R, - config: AudioPcmStreamSendConfig, -) -> thread::JoinHandle<()> -where - R: Read + Send + 'static, -{ - thread::spawn(move || { - if let Err(error) = read_audio_rudp_payloads(&tx, &mut reader, config) { - let _ = tx.send(Err(error)); - } - }) -} - -fn read_video_rudp_payloads( - tx: &mpsc::Sender>, - reader: &mut R, - config: VideoAnnexBStreamSendConfig, -) -> Result<()> -where - R: Read, -{ - let mut sender = VideoAnnexBStreamSendState::new(config)?; - let mut buffer = vec![0_u8; 64 * 1024]; - loop { - let read = reader - .read(&mut buffer) - .context("reading encoded Annex B video from ffmpeg stdout")?; - if read == 0 { - for payload in sender.finish(×tamp()?)? { - queue_muninn_media_payload(tx, payload, QueuedMuninnMediaKind::Video) - .context("queueing final typed video media payload")?; - } - return Ok(()); - } - for payload in sender.push(×tamp()?, &buffer[..read])? { - queue_muninn_media_payload(tx, payload, QueuedMuninnMediaKind::Video) - .context("queueing typed video media payload")?; - } - } -} - -fn read_audio_rudp_payloads( - tx: &mpsc::Sender>, - reader: &mut R, - config: AudioPcmStreamSendConfig, -) -> Result<()> -where - R: Read, -{ - let mut sender = AudioPcmStreamSendState::new(config)?; - let mut buffer = vec![0_u8; 16 * 1024]; - loop { - let read = reader - .read(&mut buffer) - .context("reading PCM audio from ffmpeg stdout")?; - if read == 0 { - for payload in sender.finish(×tamp()?)? { - queue_muninn_media_payload(tx, payload, QueuedMuninnMediaKind::Audio) - .context("queueing final typed audio media payload")?; - } - return Ok(()); - } - for payload in sender.push(×tamp()?, &buffer[..read])? { - queue_muninn_media_payload(tx, payload, QueuedMuninnMediaKind::Audio) - .context("queueing typed audio media payload")?; - } - } -} - fn video_frame_duration_ticks(options: &Options) -> Result { if options.framerate == 0 { return Err(anyhow!("framerate must be greater than zero")); @@ -3337,32 +2596,6 @@ fn video_frame_duration_ticks(options: &Options) -> Result { Ok((90_000_u32 / options.framerate).max(1)) } -fn rudp_media_deadline_delay_ticks(media_profile: &MuninnRudpMediaProfile) -> i64 { - i64::try_from( - media_profile - .receiver_assembly_deadline_ms - .saturating_mul(90), - ) - .unwrap_or(i64::MAX) -} - -fn rudp_audio_deadline_delay_ticks( - options: &Options, - media_profile: &MuninnRudpMediaProfile, -) -> i64 { - if options.audio_sample_rate == 0 { - return i64::from(1_024); - } - i64::try_from( - media_profile - .receiver_assembly_deadline_ms - .saturating_mul(u64::from(options.audio_sample_rate)) - / 1_000, - ) - .unwrap_or(i64::MAX) - .max(i64::from(1_024)) -} - fn publish_surface( node: &mut cultmesh_rs::CultMeshNode, options: &Options, @@ -3845,10 +3078,8 @@ fn publish_runtime_boundary_records( "video_gop_frames": muninn_rudp_video_gop_frames(options), "video_rate_control": "cbr", "video_rc_lookahead": media_profile.video_rc_lookahead, - "sender_queue_deadline_ms": media_profile.sender_queue_deadline_ms, "sender_resend_delay_ms": media_profile.sender_resend_delay_ms, - "sender_reliable_expire_after_ms": media_profile.sender_reliable_expire_after_ms, - "receiver_assembly_deadline_ms": media_profile.receiver_assembly_deadline_ms, + "latency_budget_ms": options.rudp_latency_budget_ms, "receiver_gap_wait_ms": media_profile.receiver_gap_wait_ms, "late_media_policy": "drop expired queued media; do not repair frames outside the latency budget", "recovery": "fixed quarter-second IDR budget with receiver feedback pressure telemetry" @@ -10261,24 +9492,12 @@ fn muninn_rudp_media_profile() -> MuninnRudpMediaProfile { } fn muninn_rudp_media_profile_for_options(options: &Options) -> MuninnRudpMediaProfile { - muninn_rudp_media_profile_for_bitrate_and_latency( - options.rudp_video_bitrate_kbps, - options.rudp_latency_budget_ms, - ) + muninn_rudp_media_profile_for_bitrate(options.rudp_video_bitrate_kbps) } +/// Encoder and wire shape. It carries no deadline: the request's latency +/// budget reaches the sender only through `MediaSendPolicy`. fn muninn_rudp_media_profile_for_bitrate(video_bitrate_kbps: u32) -> MuninnRudpMediaProfile { - muninn_rudp_media_profile_for_bitrate_and_latency( - video_bitrate_kbps, - MUNINN_RUDP_MEDIA_RECEIVER_ASSEMBLY_DEADLINE_MS as u32, - ) -} - -fn muninn_rudp_media_profile_for_bitrate_and_latency( - video_bitrate_kbps: u32, - latency_budget_ms: u32, -) -> MuninnRudpMediaProfile { - let latency_budget_ms = u64::from(latency_budget_ms.max(1)); MuninnRudpMediaProfile { profile_id: MUNINN_RUDP_MEDIA_PROFILE_ID, video_codec: "h264", @@ -10290,12 +9509,7 @@ fn muninn_rudp_media_profile_for_bitrate_and_latency( max_fragment_bytes: MUNINN_RUDP_MEDIA_MAX_FRAGMENT_BYTES, video_b_frames: 0, video_rc_lookahead: 0, - sender_queue_deadline_ms: latency_budget_ms, sender_resend_delay_ms: MUNINN_RUDP_MEDIA_RESEND_DELAY_MS, - sender_reliable_expire_after_ms: latency_budget_ms, - sender_pace_every_payloads: MUNINN_RUDP_MEDIA_SEND_PACE_EVERY_PAYLOADS, - sender_pace_sleep_us: MUNINN_RUDP_MEDIA_SEND_PACE_SLEEP_US, - receiver_assembly_deadline_ms: latency_budget_ms, receiver_gap_wait_ms: MUNINN_RUDP_MEDIA_RECEIVER_GAP_WAIT_MS, } } @@ -10571,7 +9785,7 @@ impl Options { media_transport: MediaTransport::Rudp, media_packet_bytes: MUNINN_RUDP_MEDIA_PACKET_BYTES, rudp_video_bitrate_kbps: MUNINN_RUDP_MEDIA_VIDEO_BITRATE_KBPS, - rudp_latency_budget_ms: MUNINN_RUDP_MEDIA_RECEIVER_ASSEMBLY_DEADLINE_MS as u32, + rudp_latency_budget_ms: MUNINN_RUDP_MEDIA_DEFAULT_LATENCY_BUDGET_MS as u32, width: 1920, height: 1080, framerate: 30, @@ -11215,6 +10429,7 @@ fn parse_move_state_source(value: &str) -> Result { #[cfg(test)] mod tests { use super::*; + use crate::media_send::{MediaKind, QueuedMediaGroup}; fn media_stream_request() -> GameCultMediaStreamRequestRecord { GameCultMediaStreamRequestRecord { @@ -12356,9 +11571,9 @@ mod tests { access_unit: &[u8], ) -> Vec { let mut sender = crate::media_packetizer::VideoAnnexBStreamSendState::new(config).unwrap(); - let mut payloads = sender.push("2026-06-18T00:00:00Z", access_unit).unwrap(); - payloads.extend(sender.finish("2026-06-18T00:00:00Z").unwrap()); - payloads + let mut groups = sender.push("2026-06-18T00:00:00Z", access_unit).unwrap(); + groups.extend(sender.finish("2026-06-18T00:00:00Z").unwrap()); + groups.into_iter().flatten().collect() } #[test] @@ -12441,7 +11656,8 @@ mod tests { fn rudp_media_hub_serves_lossy_video_and_reliable_audio_on_one_session() { let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); let profile = muninn_rudp_media_profile(); - let options = muninn_media_rudp_hub_options(socket, &profile); + let policy = MediaSendPolicy::from_request(137); + let options = muninn_media_rudp_hub_options(socket, &profile, &policy); assert_eq!(options.runtime_id, "muninn-media"); assert_eq!(options.connection_id, MUNINN_MEDIA_RUDP_CONNECTION_ID); assert_eq!(options.resend_delay_ms, MUNINN_RUDP_MEDIA_RESEND_DELAY_MS); @@ -12468,7 +11684,7 @@ mod tests { ); assert_eq!( channel("audio").reliable_expire_after_ms, - Some(profile.receiver_assembly_deadline_ms), + Some(policy.latency_budget_ms()), "audio older than the receiver's deadline is not worth a retransmit" ); } @@ -12503,59 +11719,8 @@ mod tests { assert_eq!(args.last().map(String::as_str), Some("pipe:1")); } - #[test] - fn media_payload_queue_deadline_is_strictly_bounded() { - let queued_at = Instant::now(); - let deadline = Duration::from_millis(muninn_rudp_media_profile().sender_queue_deadline_ms); - - assert!(!media_payload_queue_age_exceeded( - queued_at, - queued_at + deadline, - deadline - )); - assert!(media_payload_queue_age_exceeded( - queued_at, - queued_at + deadline + Duration::from_millis(1), - deadline - )); - } - - #[test] - fn rudp_media_sender_prioritizes_pending_audio_over_video() { - let queued_at = Instant::now(); - let mut pending = PendingMuninnMediaSendQueues::default(); - pending.push(QueuedMuninnMediaSendPayload { - payload: MuninnMediaSendPayload { - channel_id: crate::media_packetizer::MUNINN_MEDIA_RUDP_CHANNEL, - payload: vec![0x76], - }, - queued_at, - kind: QueuedMuninnMediaKind::Video, - }); - pending.push(QueuedMuninnMediaSendPayload { - payload: MuninnMediaSendPayload { - channel_id: crate::media_packetizer::MUNINN_MEDIA_RUDP_CHANNEL, - payload: vec![0x61], - }, - queued_at, - kind: QueuedMuninnMediaKind::Audio, - }); - - let first = pending.pop_next().expect("audio payload should be queued"); - let second = pending - .pop_next() - .expect("video payload should remain queued"); - - assert_eq!(first.kind, QueuedMuninnMediaKind::Audio); - assert_eq!(first.payload.payload, vec![0x61]); - assert_eq!(second.kind, QueuedMuninnMediaKind::Video); - assert_eq!(second.payload.payload, vec![0x76]); - assert!(pending.is_empty()); - } - - #[test] - fn rudp_latency_budget_owns_sender_and_record_deadlines() { - let options = Options::parse( + fn options_with_latency_budget(budget_ms: &str) -> Options { + Options::parse( [ "activate", "--media-transport", @@ -12567,210 +11732,72 @@ mod tests { "--audio-sample-rate", "48000", "--rudp-latency-budget-ms", - "2000", + budget_ms, ] .into_iter() .map(String::from), ) - .unwrap(); - let profile = muninn_rudp_media_profile_for_options(&options); + .unwrap() + } - assert_eq!(profile.sender_queue_deadline_ms, 2000); - assert_eq!(profile.sender_reliable_expire_after_ms, 2000); + #[test] + fn the_requests_latency_budget_owns_queue_hub_expiry_and_record_deadlines() { + // Two budgets, both different from the default and from each other, so + // a clamp, an offset or a fallback to the default is visible at one of + // them. + for (budget_ms, video_ticks, audio_ticks) in [(137_u32, 12_330_i64, 6_576_i64), (613, 55_170, 29_424)] + { + let options = options_with_latency_budget(&budget_ms.to_string()); + let policy = MediaSendPolicy::from_request(options.rudp_latency_budget_ms); + let produced_at = Instant::now(); + let group = QueuedMediaGroup::new(MediaKind::Video, produced_at, &policy, Vec::new()); + + assert_eq!( + group.deadline - produced_at, + Duration::from_millis(u64::from(budget_ms)) + ); + assert_eq!(policy.video_deadline_delay_ticks(), video_ticks); + assert_eq!( + policy.audio_deadline_delay_ticks(options.audio_sample_rate), + audio_ticks + ); + let hub_options = muninn_media_rudp_hub_options( + UdpSocket::bind("127.0.0.1:0").unwrap(), + &muninn_rudp_media_profile_for_options(&options), + &policy, + ); + assert_eq!( + hub_options.media_reliable_expire_after_ms, + Some(u64::from(budget_ms)) + ); + } assert_eq!( - profile.sender_pace_every_payloads, - MUNINN_RUDP_MEDIA_SEND_PACE_EVERY_PAYLOADS - ); - assert_eq!( - profile.sender_pace_sleep_us, - MUNINN_RUDP_MEDIA_SEND_PACE_SLEEP_US - ); - assert_eq!(profile.receiver_assembly_deadline_ms, 2000); - assert_eq!(rudp_media_deadline_delay_ticks(&profile), 180_000); - assert_eq!(rudp_audio_deadline_delay_ticks(&options, &profile), 96_000); - assert_eq!( - rudp_endpoint_for_options(&options), + rudp_endpoint_for_options(&options_with_latency_budget("137")), "cultmesh://odin/media/muninn-raven-av#muninn.raven.av.rudp" ); } #[test] - fn rudp_media_progress_detail_reports_queue_and_transport_pressure() { - let receiver_feedback = MuninnRudpReceiverFeedbackStats { - feedback_records: 2, - requested_keyframes: 1, - late_frames: 3, - missing_video_chunks: 4, - repaired_video_chunks: 0, - deferred_repair_chunks: 5, - repair_chunks_per_second: 64, - highest_decodable_frame_id: Some(88), - }; - - assert_eq!( - rudp_media_progress_detail(120, 3, 2, 1, 9, 2, &receiver_feedback), - "Muninn RUDP media progress: receivers=2 sent=120 queue_dropped=3 queue_expired=2 send_expired=1 reliable_expired=9 receiver_feedback=2 receiver_keyframes=1 receiver_late_frames=3 receiver_missing_chunks=4 receiver_repaired_chunks=0 receiver_deferred_repairs=5 repair_rate=64 receiver_highest_decodable=88" - ); - } - - #[test] - fn rudp_repair_budget_backs_off_on_media_drops_and_recovers_when_stable() { - let start = Instant::now(); - let mut budget = MuninnRudpRepairBudget { - chunks_per_second: 64, - min_chunks_per_second: 8, - max_chunks_per_second: 128, - add_chunks_per_second: 8, - recovery_interval: Duration::from_secs(2), - max_available_chunks: 4, - available_chunks: 4, - last_refill_at: start, - last_rate_adjust_at: start, - last_queue_dropped: 0, - }; - - assert_eq!(budget.take(4, start, 0), 4); - assert_eq!(budget.chunks_per_second(), 64); - assert_eq!(budget.take(3, start + Duration::from_millis(10), 1), 0); - assert_eq!(budget.chunks_per_second(), 32); - assert_eq!(budget.take(3, start + Duration::from_secs(1), 1), 3); - assert_eq!(budget.chunks_per_second(), 32); - assert_eq!(budget.take(3, start + Duration::from_secs(3), 1), 3); - assert_eq!(budget.chunks_per_second(), 40); - } - - #[test] - fn default_rudp_repair_budget_has_lan_stream_headroom() { - let mut budget = MuninnRudpRepairBudget::new( - MUNINN_RUDP_MEDIA_REPAIR_INITIAL_CHUNKS_PER_SECOND, - MUNINN_RUDP_MEDIA_REPAIR_BURST_CHUNKS, - ); - let start = budget.last_refill_at; - - assert_eq!(budget.chunks_per_second(), 4_096); - assert_eq!(budget.take(2_048, start, 0), 2_048); - assert_eq!(budget.take(8_192, start + Duration::from_secs(1), 0), 2_048); - assert_eq!(budget.take(8_192, start + Duration::from_secs(3), 0), 2_048); - assert_eq!(budget.chunks_per_second(), 6_144); - assert_eq!(budget.take(512, start + Duration::from_secs(4), 1), 512); - assert_eq!(budget.chunks_per_second(), 3_072); - } - - #[test] - fn receiver_feedback_keyframe_requests_are_recorded_without_encoder_restart() { - let mut handled = 0; - let mut receiver_feedback = MuninnRudpReceiverFeedbackStats::default(); - - record_receiver_keyframe_pressure(&receiver_feedback, &mut handled); - assert_eq!(handled, 0); - - receiver_feedback.requested_keyframes = 1; - record_receiver_keyframe_pressure(&receiver_feedback, &mut handled); - assert_eq!(handled, 1); - record_receiver_keyframe_pressure(&receiver_feedback, &mut handled); - assert_eq!(handled, 1); - - receiver_feedback.requested_keyframes = 2; - record_receiver_keyframe_pressure(&receiver_feedback, &mut handled); - assert_eq!(handled, 2); - } - - #[test] - fn rudp_media_receiver_feedback_updates_sender_pressure_stats() { - let feedback = cultnet_rs::build_receiver_feedback(cultnet_rs::ReceiverFeedbackOptions { - stream_id: "muninn.raven.av.rudp", - session_id: "raven:session:video", - receiver_id: "starfire.obs", - highest_decodable_frame_id: Some(41), - missing_frame_ids: Vec::new(), - missing_video_chunk_keys: vec!["42:1".to_string(), "42:3".to_string()], - late_frame_ids: vec![42, 43], - requested_keyframe: true, - jitter_us: 500, - decode_queue_us: 2_000, - observed_at: "unix:1000", - }) - .unwrap(); - let payload = crate::media_packetizer::encode_media_wire_record( - &crate::media_packetizer::GameCultMediaWireRecord::Feedback(feedback), - crate::media_packetizer::MediaWireProvenance { - stored_at: "unix:1000", - runtime_id: "starfire", - role: "mimir.obs", - producer: "mimir", - }, + fn a_request_naming_no_latency_budget_gets_250_ms_and_the_advertisement_says_so() { + assert_eq!(MUNINN_RUDP_MEDIA_DEFAULT_LATENCY_BUDGET_MS, 250); + let options = Options::parse([].into_iter()).unwrap(); + assert_eq!(options.rudp_latency_budget_ms, 250); + let advertised = Options::parse( + ["serve", "--media-rudp-advertise", "127.0.0.1:5220"] + .into_iter() + .map(String::from), ) .unwrap(); - let frame = CultNetTransportFrame { - channel_id: crate::media_packetizer::MUNINN_MEDIA_RUDP_CHANNEL.to_string(), - payload, - }; - let mut stats = MuninnRudpReceiverFeedbackStats::default(); - let repair_cache = RecentVideoChunkRepairCache::new(16); + let advertisement = media_stream_advertisement(&advertised, false).unwrap(); + assert_eq!(advertisement.default_latency_budget_ms, 250); - let repairs = - record_rudp_media_receiver_feedback(&frame, &mut stats, &repair_cache).unwrap(); - - assert_eq!(stats.feedback_records, 1); - assert_eq!(stats.requested_keyframes, 1); - assert_eq!(stats.late_frames, 2); - assert_eq!(stats.missing_video_chunks, 2); - assert_eq!(stats.repaired_video_chunks, 0); - assert_eq!(stats.highest_decodable_frame_id, Some(41)); - assert!(repairs.is_empty()); - } - - #[test] - fn repair_cache_returns_recent_missing_video_chunks_from_feedback() { - let video = cultnet_rs::GameCultMediaVideoAccessUnitRecord { - stream_id: "muninn.raven.av.rudp".to_string(), - session_id: "raven:session:video".to_string(), - frame_id: 42, - codec: "h264".to_string(), - pts_ticks: 126_000, - duration_ticks: 3_000, - timebase_num: 1, - timebase_den: 90_000, - keyframe: false, - dependency_frame_id: Some(41), - deadline_ticks: 127_800, - chunk_index: 3, - chunk_count: 5, - payload: vec![1, 2, 3, 4], - }; - let payload = MuninnMediaSendPayload { - channel_id: crate::media_packetizer::MUNINN_MEDIA_RUDP_CHANNEL, - payload: crate::media_packetizer::encode_media_wire_record( - &crate::media_packetizer::GameCultMediaWireRecord::Video(video), - crate::media_packetizer::MediaWireProvenance { - stored_at: "unix:1000", - runtime_id: "muninn-test", - role: "repair-cache-test", - producer: "mimir", - }, - ) - .unwrap(), - }; - let feedback = cultnet_rs::build_receiver_feedback(cultnet_rs::ReceiverFeedbackOptions { - stream_id: "muninn.raven.av.rudp", - session_id: "raven:session:video", - receiver_id: "starfire.obs", - highest_decodable_frame_id: Some(41), - missing_frame_ids: Vec::new(), - missing_video_chunk_keys: vec!["42:3".to_string(), "42:4".to_string()], - late_frame_ids: vec![42], - requested_keyframe: true, - jitter_us: 500, - decode_queue_us: 2_000, - observed_at: "unix:1000", - }) - .unwrap(); - - let mut cache = RecentVideoChunkRepairCache::new(16); - cache.remember(&payload).unwrap(); - let repairs = cache.repair_payloads_for_feedback(&feedback); - - assert_eq!(repairs, vec![payload]); + let mut request = media_stream_request(); + request.latency_budget_ms = 0; + let command = command_from_media_stream_request(&options, &request).unwrap(); + assert_eq!(command_rudp_latency_budget_ms(&command), 250); + request.latency_budget_ms = 40; + let command = command_from_media_stream_request(&options, &request).unwrap(); + assert_eq!(command_rudp_latency_budget_ms(&command), 40); } #[test] @@ -14037,10 +13064,20 @@ Device 00:07:04:A8:00:D0 (public) ); assert_eq!( media_profile - .get("sender_reliable_expire_after_ms") + .get("latency_budget_ms") .and_then(|value| value.as_u64()), - Some(MUNINN_RUDP_MEDIA_RECEIVER_ASSEMBLY_DEADLINE_MS) + Some(MUNINN_RUDP_MEDIA_DEFAULT_LATENCY_BUDGET_MS) ); + for retired in [ + "sender_queue_deadline_ms", + "sender_reliable_expire_after_ms", + "receiver_assembly_deadline_ms", + ] { + assert!( + media_profile.get(retired).is_none(), + "{retired} was a second name for the request budget" + ); + } assert_eq!( media_profile .get("sender_resend_delay_ms") diff --git a/crates/muninn-daemon/src/media_packetizer.rs b/crates/muninn-daemon/src/media_packetizer.rs index d814944..bcf2423 100644 --- a/crates/muninn-daemon/src/media_packetizer.rs +++ b/crates/muninn-daemon/src/media_packetizer.rs @@ -213,11 +213,17 @@ pub struct MuninnMediaSendPayload { pub payload: Vec, } +/// Everything one video access unit puts on the wire: its chunks and, until +/// the FEC cut replaces it, their parity. A group is sent whole or not at +/// all, because parity for a frame that never completes is wasted bandwidth. +pub type MuninnMediaPayloadGroup = Vec; + pub struct VideoAnnexBStreamSendState { config: VideoAnnexBStreamSendConfig, pending: Vec, next_frame_id: u64, next_pts_ticks: i64, + dropped_access_units: u64, } impl VideoAnnexBStreamSendState { @@ -260,26 +266,35 @@ impl VideoAnnexBStreamSendState { next_pts_ticks: config.first_pts_ticks, config, pending: Vec::new(), + dropped_access_units: 0, }) } - pub fn push(&mut self, stored_at: &str, bytes: &[u8]) -> Result> { + /// One group per completed access unit. Whatever is still pending after + /// that is a single unfinished access unit; when it outgrows + /// `max_pending_bytes` it is dropped and counted, never an error that + /// would restart the stream. + pub fn push(&mut self, stored_at: &str, bytes: &[u8]) -> Result> { if !bytes.is_empty() { self.pending.extend_from_slice(bytes); } + let groups = self.emit_available(stored_at, false)?; if self.pending.len() > self.config.max_pending_bytes { - return Err(anyhow!( - "Annex B stream sender pending buffer exceeded {} bytes without a complete frame", - self.config.max_pending_bytes - )); + self.pending.clear(); + self.dropped_access_units += 1; } - self.emit_available(stored_at, false) + Ok(groups) } - pub fn finish(&mut self, stored_at: &str) -> Result> { + pub fn finish(&mut self, stored_at: &str) -> Result> { self.emit_available(stored_at, true) } + /// Access units dropped for outgrowing `max_pending_bytes`, cumulative. + pub fn dropped_access_units(&self) -> u64 { + self.dropped_access_units + } + pub fn pending_bytes(&self) -> usize { self.pending.len() } @@ -292,10 +307,17 @@ impl VideoAnnexBStreamSendState { &mut self, stored_at: &str, flush_last: bool, - ) -> Result> { + ) -> Result> { if stored_at.is_empty() { return Err(anyhow!("stored_at must be non-empty")); } + // After a drop the tail of the dropped access unit arrives with no + // start code in front of it; skip to the next one. + if let Some((start, _)) = find_annex_b_start_code(&self.pending, 0) + && start > 0 + { + self.pending.drain(..start); + } if self.pending.is_empty() { return Ok(Vec::new()); } @@ -319,7 +341,7 @@ impl VideoAnnexBStreamSendState { return Ok(Vec::new()); } - let mut send_payloads = Vec::new(); + let mut groups = Vec::with_capacity(emit_count); let mut emitted_bytes = 0_usize; for access_unit in access_units.iter().take(emit_count) { emitted_bytes = emitted_bytes @@ -345,7 +367,7 @@ impl VideoAnnexBStreamSendState { access_unit, )?; let wire_records = video_wire_records_with_parity(&records)?; - send_payloads.extend(wire_payloads_to_media_send(encode_media_wire_records( + groups.push(wire_payloads_to_media_send(encode_media_wire_records( &wire_records, stored_at, &self.config.source_runtime_id, @@ -355,7 +377,7 @@ impl VideoAnnexBStreamSendState { } self.pending.drain(..emitted_bytes); - Ok(send_payloads) + Ok(groups) } fn advance_frame_clock(&mut self) -> Result<()> { @@ -1525,8 +1547,8 @@ mod tests { stream.extend_from_slice(&start_code()); stream.extend_from_slice(&[second_slice_header, 0x80]); let mut sender = VideoAnnexBStreamSendState::new(config)?; - let mut payloads = sender.push("2026-06-18T00:00:00Z", &stream)?; - payloads.extend(sender.finish("2026-06-18T00:00:00Z")?); + let mut payloads = flat(sender.push("2026-06-18T00:00:00Z", &stream)?); + payloads.extend(flat(sender.finish("2026-06-18T00:00:00Z")?)); let mut records = Vec::new(); for payload in &payloads { if let GameCultMediaWireRecord::Video(record) = @@ -1595,8 +1617,8 @@ mod tests { stream.extend_from_slice(&start_code()); stream.extend_from_slice(&[0x41, 0x80]); let mut sender = VideoAnnexBStreamSendState::new(stream_send_config())?; - let mut payloads = sender.push("2026-06-18T00:00:00Z", &stream)?; - payloads.extend(sender.finish("2026-06-18T00:00:00Z")?); + let mut payloads = flat(sender.push("2026-06-18T00:00:00Z", &stream)?); + payloads.extend(flat(sender.finish("2026-06-18T00:00:00Z")?)); assert!(payloads.len() >= 2); let wire_records = payloads .iter() @@ -1636,9 +1658,10 @@ mod tests { let second_push = sender.push("2026-06-18T00:00:00Z", &second_frame)?; assert_eq!(second_push.len(), 1); - assert_eq!(second_push[0].channel_id, MUNINN_MEDIA_RUDP_CHANNEL); + assert_eq!(second_push[0].len(), 1); + assert_eq!(second_push[0][0].channel_id, MUNINN_MEDIA_RUDP_CHANNEL); let GameCultMediaWireRecord::Video(first) = - decode_media_wire_record(&second_push[0].payload)? + decode_media_wire_record(&second_push[0][0].payload)? else { panic!("expected video media record"); }; @@ -1651,7 +1674,8 @@ mod tests { let tail = sender.finish("2026-06-18T00:00:00Z")?; assert_eq!(tail.len(), 1); - let GameCultMediaWireRecord::Video(second) = decode_media_wire_record(&tail[0].payload)? + assert_eq!(tail[0].len(), 1); + let GameCultMediaWireRecord::Video(second) = decode_media_wire_record(&tail[0][0].payload)? else { panic!("expected video media record"); }; @@ -1680,21 +1704,84 @@ mod tests { ); } + fn flat(groups: Vec) -> Vec { + groups.into_iter().flatten().collect() + } + #[test] - fn annex_b_stream_send_state_rejects_unbounded_pending_bytes() -> Result<()> { + fn annex_b_stream_send_state_drops_and_counts_an_oversize_access_unit_instead_of_erroring() + -> Result<()> { let mut config = stream_send_config(); - config.max_pending_bytes = 4; + config.max_pending_bytes = 8; let mut sender = VideoAnnexBStreamSendState::new(config)?; - let error = sender - .push("2026-06-18T00:00:00Z", &[0x47, 0x40, 0x00, 0x10, 0x00]) - .unwrap_err(); + // One access unit that never completes and is larger than the ceiling. + let mut oversize = start_code().to_vec(); + oversize.extend_from_slice(&[0x65, 0x80, 1, 2, 3, 4, 5, 6, 7, 8]); + let groups = sender.push("2026-06-18T00:00:00Z", &oversize)?; - assert!( - error - .to_string() - .contains("pending buffer exceeded 4 bytes") - ); + assert!(groups.is_empty()); + assert_eq!(sender.dropped_access_units(), 1); + assert_eq!(sender.pending_bytes(), 0); + + // The tail of the dropped unit has no start code and is skipped; the + // stream then carries on with the next access units. + let mut next = vec![9, 9, 9]; + next.extend_from_slice(&start_code()); + next.extend_from_slice(&[0x65, 0x80]); + next.extend_from_slice(&start_code()); + next.extend_from_slice(&[0x41, 0x80]); + let groups = sender.push("2026-06-18T00:00:00Z", &next)?; + + assert_eq!(groups.len(), 1); + assert_eq!(sender.dropped_access_units(), 1); + let GameCultMediaWireRecord::Video(record) = + decode_media_wire_record(&groups[0][0].payload)? + else { + panic!("expected video media record"); + }; + assert!(record.keyframe); + Ok(()) + } + + #[test] + fn annex_b_stream_send_state_returns_one_group_per_access_unit_with_all_its_chunks_and_parity() + -> Result<()> { + // Three chunks of 16 bytes plus a short fourth: one access unit. + let mut stream = start_code().to_vec(); + stream.extend_from_slice(&[0x65, 0x80]); + stream.resize(52, 0xAB); + stream.extend_from_slice(&start_code()); + stream.extend_from_slice(&[0x41, 0x80]); + let mut sender = VideoAnnexBStreamSendState::new(stream_send_config())?; + + let mut groups = sender.push("2026-06-18T00:00:00Z", &stream)?; + groups.extend(sender.finish("2026-06-18T00:00:00Z")?); + + assert_eq!(groups.len(), 2); + let mut frame_ids_per_group = Vec::new(); + for group in &groups { + let mut ids = std::collections::BTreeSet::new(); + let mut data = 0; + let mut parity = 0; + for payload in group { + match decode_media_wire_record(&payload.payload)? { + GameCultMediaWireRecord::Video(record) => { + ids.insert(record.frame_id); + data += 1; + } + GameCultMediaWireRecord::VideoParity(record) => { + ids.insert(record.frame_id); + parity += 1; + } + other => panic!("unexpected record {other:?}"), + } + } + assert_eq!(ids.len(), 1, "a group carries exactly one frame"); + frame_ids_per_group.push((ids.into_iter().next().unwrap(), data, parity)); + } + assert_eq!(frame_ids_per_group[0], (9, 4, 4)); + assert_eq!(frame_ids_per_group[1], (10, 1, 0)); Ok(()) } diff --git a/crates/muninn-daemon/src/media_send.rs b/crates/muninn-daemon/src/media_send.rs new file mode 100644 index 0000000..e84855f --- /dev/null +++ b/crates/muninn-daemon/src/media_send.rs @@ -0,0 +1,960 @@ +//! The media send path: the one place that decides when a frame is too old. +//! +//! **Owner.** `MediaSendCore` owns the bounded group queues, the hub, the +//! receiver-feedback intake, the repair cache and its budget, and the progress +//! counters. `run_rudp_mux_once` spawns the encoders and publishes; it feeds +//! this module bytes and never decides a deadline. +//! +//! **One deadline owner.** `MediaSendPolicy::latency_budget` comes from the +//! request's `latency_budget_ms` and is the only source of a deadline. A group +//! is stamped `produced_at + latency_budget` when it leaves the packetizer; a +//! repair inherits the deadline of the frame it repairs; the hub's reliable +//! expiry and the record deadline ticks derive from the same budget. +//! +//! **Bounds.** Between the encoder and the socket every queue has a count +//! bound (`MEDIA_GROUP_CHANNEL_BOUND`, the per-kind group caps) and a time +//! bound (the group deadline). Crossing either drops whole groups, oldest +//! first, and counts them. The video cap is 512 groups of at most one access +//! unit each, and an access unit is at most `max_pending_bytes` (848 * 4096 +//! bytes at the default packet size), so the count cap alone implies a worst +//! case of about 1.7 GiB; in practice the deadline empties the queue long +//! before that. + +use crate::media_packetizer::{ + AudioPcmStreamSendConfig, AudioPcmStreamSendState, GameCultMediaWireRecord, + MUNINN_MEDIA_RUDP_CHANNEL, MuninnMediaPayloadGroup, MuninnMediaSendPayload, + VideoAnnexBStreamSendConfig, VideoAnnexBStreamSendState, decode_media_wire_record, + video_chunk_feedback_key, +}; +use anyhow::{Context, Result}; +use cultnet_rs::{ + CultNetRudpServerEvent, CultNetRudpServerHub, CultNetRudpServerSessionContext, + CultNetTransportFrame, GameCultMediaReceiverFeedbackRecord, +}; +use std::collections::{HashMap, VecDeque}; +use std::io::{ErrorKind, Read}; +use std::sync::mpsc; +use std::thread; +use std::time::{Duration, Instant}; + +pub const MEDIA_SEND_AUDIO_GROUP_CAP: usize = 256; +pub const MEDIA_SEND_VIDEO_GROUP_CAP: usize = 512; +/// Groups the reader threads may have handed over and the send loop not yet +/// taken. Small on purpose: the caps above are where a backlog lives. +pub const MEDIA_GROUP_CHANNEL_BOUND: usize = 8; + +const REPAIR_CACHE_CHUNKS: usize = 16_384; +pub const REPAIR_BURST_CHUNKS: usize = 2_048; +pub const REPAIR_INITIAL_CHUNKS_PER_SECOND: usize = 4_096; +const REPAIR_MIN_CHUNKS_PER_SECOND: usize = 8; +const REPAIR_MAX_CHUNKS_PER_SECOND: usize = 16_384; +const REPAIR_ADD_CHUNKS_PER_SECOND: usize = 2_048; +const REPAIR_RECOVERY_INTERVAL_MS: u64 = 2_000; +const REPAIR_MAX_FEEDBACK_PER_POLL: usize = 32; +const REPAIR_MAX_CHUNKS_PER_POLL: usize = 256; +/// A receiver that has said nothing for this long (Ratatoskr pings every +/// second or so) is dropped so its payloads stop being sent into the void. +const RECEIVER_TIMEOUT_MS: u64 = 6_000; +pub const SEND_PACE_EVERY_PAYLOADS: usize = 4; +pub const SEND_PACE_SLEEP_US: u64 = 250; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum MediaKind { + Video, + Audio, +} + +/// What the request decided about sending. Nothing else mints a deadline. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct MediaSendPolicy { + pub latency_budget: Duration, + pub video_pace_every_payloads: usize, + pub video_pace_sleep: Duration, +} + +impl MediaSendPolicy { + pub fn from_request(latency_budget_ms: u32) -> Self { + Self { + latency_budget: Duration::from_millis(u64::from(latency_budget_ms.max(1))), + video_pace_every_payloads: SEND_PACE_EVERY_PAYLOADS, + video_pace_sleep: Duration::from_micros(SEND_PACE_SLEEP_US), + } + } + + pub fn latency_budget_ms(&self) -> u64 { + self.latency_budget.as_millis() as u64 + } + + /// The video record deadline, in the 90 kHz video timebase. + pub fn video_deadline_delay_ticks(&self) -> i64 { + i64::try_from(self.latency_budget_ms().saturating_mul(90)).unwrap_or(i64::MAX) + } + + /// The audio record deadline, in samples at `sample_rate`. + pub fn audio_deadline_delay_ticks(&self, sample_rate: u32) -> i64 { + if sample_rate == 0 { + return 1_024; + } + i64::try_from( + self.latency_budget_ms() + .saturating_mul(u64::from(sample_rate)) + / 1_000, + ) + .unwrap_or(i64::MAX) + .max(1_024) + } +} + +/// One access unit with its parity, or one audio packet. It is sent whole or +/// not at all, and it dies at `deadline`. +pub struct QueuedMediaGroup { + pub kind: MediaKind, + pub deadline: Instant, + pub payloads: MuninnMediaPayloadGroup, +} + +impl QueuedMediaGroup { + pub fn new( + kind: MediaKind, + produced_at: Instant, + policy: &MediaSendPolicy, + payloads: MuninnMediaPayloadGroup, + ) -> Self { + Self { + kind, + deadline: produced_at + policy.latency_budget, + payloads, + } + } + + /// Sendable at the deadline itself, expired one instant after. + pub fn expired(&self, now: Instant) -> bool { + now > self.deadline + } +} + +/// What a reader thread hands the send loop. +pub enum MediaIntake { + Group(QueuedMediaGroup), + /// Video access units the packetizer dropped for outgrowing its ceiling. + VideoDropped(u64), +} + +pub type MediaIntakeSender = mpsc::SyncSender>; +pub type MediaIntakeReceiver = mpsc::Receiver>; + +pub fn media_intake_channel() -> (MediaIntakeSender, MediaIntakeReceiver) { + mpsc::sync_channel(MEDIA_GROUP_CHANNEL_BOUND) +} + +#[derive(Default)] +struct PendingMediaGroups { + audio: VecDeque, + video: VecDeque, +} + +impl PendingMediaGroups { + /// `true` when the queue was at its cap and its oldest group was dropped + /// to make room. + fn push(&mut self, group: QueuedMediaGroup) -> bool { + let (queue, cap) = match group.kind { + MediaKind::Audio => (&mut self.audio, MEDIA_SEND_AUDIO_GROUP_CAP), + MediaKind::Video => (&mut self.video, MEDIA_SEND_VIDEO_GROUP_CAP), + }; + let dropped = queue.len() >= cap; + if dropped { + queue.pop_front(); + } + queue.push_back(group); + dropped + } + + /// Audio leaves before video: a late audio packet is a click, a late + /// frame is one dropped frame. + fn pop_next(&mut self) -> Option { + self.audio.pop_front().or_else(|| self.video.pop_front()) + } + + fn is_empty(&self) -> bool { + self.audio.is_empty() && self.video.is_empty() + } +} + +/// Counted where it happens. `groups_dropped_*` is a cap crossing; +/// `groups_expired` is a group whose deadline passed with none of it sent; +/// `groups_cut_short` is one whose deadline passed with part of it already on +/// the wire, which is the only way a partial frame can leave. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct MediaSendStats { + pub groups_sent: u64, + pub payloads_sent: u64, + pub groups_dropped_audio: u64, + pub groups_dropped_video: u64, + pub groups_expired: u64, + pub groups_cut_short: u64, +} + +impl MediaSendStats { + pub fn groups_lost(&self) -> u64 { + self.groups_dropped_audio + + self.groups_dropped_video + + self.groups_expired + + self.groups_cut_short + } + + /// Accounts for one group the send loop has finished with. + fn record_group(&mut self, outcome: &GroupSend, group_payloads: usize) { + match outcome { + GroupSend::Complete => { + self.groups_sent += 1; + self.payloads_sent += group_payloads as u64; + } + GroupSend::DeadlinePassed { payloads_out: 0 } => self.groups_expired += 1, + GroupSend::DeadlinePassed { payloads_out } => { + self.payloads_sent += *payloads_out as u64; + self.groups_cut_short += 1; + } + } + } +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct MuninnRudpReceiverFeedbackStats { + pub feedback_records: u64, + pub requested_keyframes: u64, + pub late_frames: u64, + pub missing_video_chunks: u64, + pub repaired_video_chunks: u64, + pub deferred_repair_chunks: u64, + pub repair_chunks_per_second: usize, + pub highest_decodable_frame_id: Option, +} + +#[derive(Debug)] +pub struct MuninnRudpRepairBudget { + chunks_per_second: usize, + min_chunks_per_second: usize, + max_chunks_per_second: usize, + add_chunks_per_second: usize, + recovery_interval: Duration, + max_available_chunks: usize, + available_chunks: usize, + last_refill_at: Instant, + last_rate_adjust_at: Instant, + last_queue_dropped: u64, +} + +impl MuninnRudpRepairBudget { + pub fn new(chunks_per_second: usize, max_available_chunks: usize) -> Self { + let now = Instant::now(); + let max_available_chunks = max_available_chunks.max(1); + Self { + chunks_per_second: chunks_per_second.max(1), + min_chunks_per_second: REPAIR_MIN_CHUNKS_PER_SECOND, + max_chunks_per_second: REPAIR_MAX_CHUNKS_PER_SECOND, + add_chunks_per_second: REPAIR_ADD_CHUNKS_PER_SECOND, + recovery_interval: Duration::from_millis(REPAIR_RECOVERY_INTERVAL_MS), + max_available_chunks, + available_chunks: max_available_chunks, + last_refill_at: now, + last_rate_adjust_at: now, + last_queue_dropped: 0, + } + } + + pub fn chunks_per_second(&self) -> usize { + self.chunks_per_second + } + + pub fn take(&mut self, requested_chunks: usize, now: Instant, queue_dropped: u64) -> usize { + self.adjust_rate(now, queue_dropped, requested_chunks); + self.refill(now); + let allowed = requested_chunks.min(self.available_chunks); + self.available_chunks -= allowed; + allowed + } + + fn adjust_rate(&mut self, now: Instant, queue_dropped: u64, requested_chunks: usize) { + if queue_dropped > self.last_queue_dropped { + self.chunks_per_second = (self.chunks_per_second / 2).max(self.min_chunks_per_second); + self.available_chunks = self.available_chunks.min(self.chunks_per_second); + self.last_queue_dropped = queue_dropped; + self.last_rate_adjust_at = now; + return; + } + self.last_queue_dropped = queue_dropped; + if requested_chunks == 0 + || now.saturating_duration_since(self.last_rate_adjust_at) < self.recovery_interval + { + return; + } + if self.chunks_per_second < self.max_chunks_per_second { + self.chunks_per_second = self + .chunks_per_second + .saturating_add(self.add_chunks_per_second) + .min(self.max_chunks_per_second); + self.last_rate_adjust_at = now; + } + } + + fn refill(&mut self, now: Instant) { + let elapsed_ms = now + .saturating_duration_since(self.last_refill_at) + .as_millis() as usize; + if elapsed_ms == 0 { + return; + } + let refill_chunks = elapsed_ms.saturating_mul(self.chunks_per_second) / 1_000; + if refill_chunks == 0 { + return; + } + self.available_chunks = self + .available_chunks + .saturating_add(refill_chunks) + .min(self.max_available_chunks); + self.last_refill_at = now; + } +} + +#[derive(Debug)] +struct RepairEntry { + payload: MuninnMediaSendPayload, + /// The deadline of the frame group the chunk left in. A repair never + /// gets a deadline of its own. + deadline: Instant, +} + +#[derive(Debug)] +pub struct RecentVideoChunkRepairCache { + max_entries: usize, + order: VecDeque, + entries: HashMap, +} + +impl RecentVideoChunkRepairCache { + pub fn new(max_entries: usize) -> Self { + Self { + max_entries, + order: VecDeque::new(), + entries: HashMap::new(), + } + } + + pub fn remember(&mut self, payload: &MuninnMediaSendPayload, deadline: Instant) -> Result<()> { + let Some(key) = video_repair_cache_key_from_payload(payload)? else { + return Ok(()); + }; + let entry = RepairEntry { + payload: payload.clone(), + deadline, + }; + if self.entries.contains_key(&key) { + self.entries.insert(key, entry); + return Ok(()); + } + self.order.push_back(key.clone()); + self.entries.insert(key, entry); + while self.entries.len() > self.max_entries { + let Some(expired) = self.order.pop_front() else { + break; + }; + self.entries.remove(&expired); + } + Ok(()) + } + + /// The chunks the receiver is missing that are still inside their frame's + /// deadline at `now`, each with that deadline. + pub fn repair_payloads_for_feedback( + &self, + feedback: &GameCultMediaReceiverFeedbackRecord, + now: Instant, + ) -> Vec<(MuninnMediaSendPayload, Instant)> { + feedback + .missing_video_chunk_keys + .iter() + .filter_map(|chunk_key| { + self.entries.get(&video_repair_cache_key( + &feedback.stream_id, + &feedback.session_id, + chunk_key, + )) + }) + .filter(|entry| now <= entry.deadline) + .map(|entry| (entry.payload.clone(), entry.deadline)) + .collect() + } +} + +fn video_repair_cache_key(stream_id: &str, session_id: &str, chunk_key: &str) -> String { + format!("{stream_id}:{session_id}:video:{chunk_key}") +} + +fn video_repair_cache_key_from_payload(payload: &MuninnMediaSendPayload) -> Result> { + if payload.channel_id != MUNINN_MEDIA_RUDP_CHANNEL { + return Ok(None); + } + let GameCultMediaWireRecord::Video(video) = decode_media_wire_record(&payload.payload)? else { + return Ok(None); + }; + Ok(Some(video_repair_cache_key( + &video.stream_id, + &video.session_id, + &video_chunk_feedback_key(video.frame_id, video.chunk_index), + ))) +} + +#[derive(Debug)] +pub struct MuninnRudpMediaSendPacer { + payloads_since_pause: usize, + every_payloads: usize, + sleep_for: Duration, +} + +impl MuninnRudpMediaSendPacer { + pub fn new(every_payloads: usize, sleep_for: Duration) -> Self { + Self { + payloads_since_pause: 0, + every_payloads: every_payloads.max(1), + sleep_for, + } + } + + fn observe_sent_payload(&mut self) { + if self.sleep_for.is_zero() { + return; + } + self.payloads_since_pause = self.payloads_since_pause.saturating_add(1); + if self.payloads_since_pause < self.every_payloads { + return; + } + self.payloads_since_pause = 0; + thread::sleep(self.sleep_for); + } +} + +pub fn is_would_block_error(error: &anyhow::Error) -> bool { + error.chain().any(|cause| { + cause + .downcast_ref::() + .is_some_and(|io| io.kind() == ErrorKind::WouldBlock) + || cause.to_string().contains("os error 10035") + }) +} + +/// Tries `attempt` until it goes through or `deadline` passes. While the +/// socket is full it polls resends and tries again; it never sleeps and never +/// outlives the deadline. `false` means the deadline won. +fn send_until( + host: &mut H, + deadline: Instant, + mut now: impl FnMut() -> Instant, + mut attempt: impl FnMut(&mut H) -> Result<()>, + mut poll: impl FnMut(&mut H) -> Result<()>, +) -> Result { + loop { + match attempt(host) { + Ok(()) => return Ok(true), + Err(error) if is_would_block_error(&error) => { + if now() > deadline { + return Ok(false); + } + poll(host)?; + thread::yield_now(); + } + Err(error) => return Err(error).context("sending typed Muninn media payload"), + } + } +} + +fn poll_resends(hub: &mut CultNetRudpServerHub) -> Result<()> { + match hub.poll_resends() { + Ok(()) => Ok(()), + Err(error) if is_would_block_error(&error) => Ok(()), + Err(error) => Err(error).context("polling Muninn RUDP media resends"), + } +} + +#[derive(Debug)] +enum GroupSend { + Complete, + /// The deadline passed after `payloads_out` payloads had gone out. + DeadlinePassed { payloads_out: usize }, +} + +/// Every payload to every receiver, in order, or as many as the deadline +/// allows. A payload with nobody to send it to is not a failure; the progress +/// line says how many receivers there are. +fn send_payloads( + hub: &mut CultNetRudpServerHub, + receivers: &[CultNetRudpServerSessionContext], + payloads: &[MuninnMediaSendPayload], + deadline: Instant, + pacer: &mut MuninnRudpMediaSendPacer, +) -> Result { + for (payloads_out, payload) in payloads.iter().enumerate() { + for receiver in receivers { + let sent = send_until( + hub, + deadline, + Instant::now, + |hub| hub.send(receiver, payload.channel_id, payload.payload.clone()), + poll_resends, + )?; + if !sent { + return Ok(GroupSend::DeadlinePassed { payloads_out }); + } + } + pacer.observe_sent_payload(); + } + Ok(GroupSend::Complete) +} + +pub struct MediaSendCore { + hub: CultNetRudpServerHub, + queues: PendingMediaGroups, + repair_cache: RecentVideoChunkRepairCache, + repair_budget: MuninnRudpRepairBudget, + video_pacer: MuninnRudpMediaSendPacer, + audio_pacer: MuninnRudpMediaSendPacer, + stats: MediaSendStats, + feedback: MuninnRudpReceiverFeedbackStats, + handled_keyframe_requests: u64, + disconnected: bool, +} + +impl MediaSendCore { + pub fn new(hub: CultNetRudpServerHub, policy: &MediaSendPolicy) -> Self { + Self { + hub, + queues: PendingMediaGroups::default(), + repair_cache: RecentVideoChunkRepairCache::new(REPAIR_CACHE_CHUNKS), + repair_budget: MuninnRudpRepairBudget::new( + REPAIR_INITIAL_CHUNKS_PER_SECOND, + REPAIR_BURST_CHUNKS, + ), + video_pacer: MuninnRudpMediaSendPacer::new( + policy.video_pace_every_payloads, + policy.video_pace_sleep, + ), + audio_pacer: MuninnRudpMediaSendPacer::new(0, Duration::ZERO), + stats: MediaSendStats::default(), + feedback: MuninnRudpReceiverFeedbackStats::default(), + handled_keyframe_requests: 0, + disconnected: false, + } + } + + #[cfg(test)] + pub fn stats(&self) -> &MediaSendStats { + &self.stats + } + + #[cfg(test)] + pub fn feedback(&self) -> &MuninnRudpReceiverFeedbackStats { + &self.feedback + } + + #[cfg(test)] + pub fn hub(&self) -> &CultNetRudpServerHub { + &self.hub + } + + /// Nothing queued and every reader gone: the encoders have ended. + pub fn finished(&self) -> bool { + self.disconnected && self.queues.is_empty() + } + + /// Takes what the reader threads have produced. Waits up to `wait` for the + /// first group only when nothing is queued to send. + pub fn intake(&mut self, rx: &MediaIntakeReceiver, wait: Duration) -> Result<()> { + if self.disconnected { + return Ok(()); + } + if self.queues.is_empty() { + match rx.recv_timeout(wait) { + Ok(item) => self.accept(item?), + Err(mpsc::RecvTimeoutError::Timeout) => return Ok(()), + Err(mpsc::RecvTimeoutError::Disconnected) => { + self.disconnected = true; + return Ok(()); + } + } + } + loop { + match rx.try_recv() { + Ok(item) => self.accept(item?), + Err(mpsc::TryRecvError::Empty) => return Ok(()), + Err(mpsc::TryRecvError::Disconnected) => { + self.disconnected = true; + return Ok(()); + } + } + } + } + + fn accept(&mut self, item: MediaIntake) { + match item { + MediaIntake::Group(group) => self.enqueue(group), + MediaIntake::VideoDropped(count) => { + self.stats.groups_dropped_video += count; + self.log_loss_milestone(); + } + } + } + + pub fn enqueue(&mut self, group: QueuedMediaGroup) { + let kind = group.kind; + if self.queues.push(group) { + match kind { + MediaKind::Audio => self.stats.groups_dropped_audio += 1, + MediaKind::Video => self.stats.groups_dropped_video += 1, + } + self.log_loss_milestone(); + } + } + + /// Sends the next group that is still inside its deadline at `now`, + /// discarding expired ones whole on the way. `true` when a group went out + /// (all of it, or the part the deadline allowed). + pub fn send_next(&mut self, now: Instant) -> Result { + while let Some(group) = self.queues.pop_next() { + if group.expired(now) { + self.stats.groups_expired += 1; + self.log_loss_milestone(); + continue; + } + let receivers = self.hub.sessions(); + let pacer = match group.kind { + MediaKind::Video => &mut self.video_pacer, + MediaKind::Audio => &mut self.audio_pacer, + }; + let outcome = send_payloads( + &mut self.hub, + &receivers, + &group.payloads, + group.deadline, + pacer, + )?; + self.stats.record_group(&outcome, group.payloads.len()); + match outcome { + GroupSend::Complete => { + if group.kind == MediaKind::Video { + for payload in &group.payloads { + self.repair_cache.remember(payload, group.deadline)?; + } + } + if self.stats.groups_sent == 1 || self.stats.groups_sent % 900 == 0 { + eprintln!("{}", self.progress_detail()); + } + } + GroupSend::DeadlinePassed { .. } => self.log_loss_milestone(), + } + return Ok(true); + } + Ok(false) + } + + /// Receiver feedback, repairs, keyframe pressure and resends. + pub fn service_control(&mut self, now: Instant) -> Result<()> { + self.poll_feedback(now)?; + if self.feedback.requested_keyframes > self.handled_keyframe_requests { + self.handled_keyframe_requests = self.feedback.requested_keyframes; + eprintln!( + "Muninn RUDP receiver requested a fresh keyframe; continuing current low-latency encoder session until explicit encoder control exists." + ); + } + poll_resends(&mut self.hub) + } + + fn poll_feedback(&mut self, now: Instant) -> Result<()> { + for gone in self.hub.remove_timed_out_sessions(RECEIVER_TIMEOUT_MS) { + eprintln!( + "Muninn media receiver {} at {} went silent and was dropped.", + String::from_utf8_lossy(&gone.connect_payload), + gone.remote_addr + ); + } + let mut feedback_processed = 0_usize; + loop { + match self.hub.receive_event_once() { + Ok(Some(CultNetRudpServerEvent::Connected { session })) => { + eprintln!( + "Muninn media receiver {} attached from {}.", + String::from_utf8_lossy(&session.connect_payload), + session.remote_addr + ); + } + Ok(Some(CultNetRudpServerEvent::Disconnected { session, reason })) => { + eprintln!( + "Muninn media receiver {} at {} disconnected: {}.", + String::from_utf8_lossy(&session.connect_payload), + session.remote_addr, + String::from_utf8_lossy(&reason) + ); + } + Ok(Some(CultNetRudpServerEvent::Frame { session, frame })) => { + if feedback_processed >= REPAIR_MAX_FEEDBACK_PER_POLL { + return Ok(()); + } + feedback_processed += 1; + let repairs = record_receiver_feedback( + &frame, + &mut self.feedback, + &self.repair_cache, + now, + )?; + let requested = repairs.len(); + let allowed = self.repair_budget.take( + requested.min(REPAIR_MAX_CHUNKS_PER_POLL), + now, + self.stats.groups_lost(), + ); + self.feedback.repair_chunks_per_second = self.repair_budget.chunks_per_second(); + self.feedback.deferred_repair_chunks = self + .feedback + .deferred_repair_chunks + .saturating_add(requested.saturating_sub(allowed) as u64); + // Repairs go back to the receiver that asked, not to everyone, + // and each one keeps the deadline of its own frame. + for (payload, deadline) in repairs.into_iter().take(allowed) { + let outcome = send_payloads( + &mut self.hub, + std::slice::from_ref(&session), + std::slice::from_ref(&payload), + deadline, + &mut self.video_pacer, + )?; + if matches!(outcome, GroupSend::Complete) { + self.feedback.repaired_video_chunks = + self.feedback.repaired_video_chunks.saturating_add(1); + } + } + } + Ok(Some(CultNetRudpServerEvent::Pong { .. })) => {} + Ok(None) => return Ok(()), + Err(error) if is_would_block_error(&error) => return Ok(()), + Err(error) => return Err(error).context("polling Muninn RUDP media feedback"), + } + } + } + + fn log_loss_milestone(&self) { + let lost = self.stats.groups_lost(); + // Loss can arrive in the hundreds per second; the first and every + // 300th is enough to see it without flooding the log. + if lost == 1 || lost % 300 == 0 { + eprintln!("{}", self.progress_detail()); + } + } + + pub fn progress_detail(&self) -> String { + format!( + "Muninn RUDP media progress: receivers={} groups_sent={} payloads_sent={} groups_dropped_audio={} groups_dropped_video={} groups_expired={} groups_cut_short={} reliable_expired={} receiver_feedback={} receiver_keyframes={} receiver_late_frames={} receiver_missing_chunks={} receiver_repaired_chunks={} receiver_deferred_repairs={} repair_rate={} receiver_highest_decodable={}; pending_audio={} pending_video={}", + self.hub.sessions().len(), + self.stats.groups_sent, + self.stats.payloads_sent, + self.stats.groups_dropped_audio, + self.stats.groups_dropped_video, + self.stats.groups_expired, + self.stats.groups_cut_short, + self.hub.stats().reliable_packets_expired, + self.feedback.feedback_records, + self.feedback.requested_keyframes, + self.feedback.late_frames, + self.feedback.missing_video_chunks, + self.feedback.repaired_video_chunks, + self.feedback.deferred_repair_chunks, + self.feedback.repair_chunks_per_second, + self.feedback + .highest_decodable_frame_id + .map(|frame_id| frame_id.to_string()) + .unwrap_or_else(|| "none".to_string()), + self.queues.audio.len(), + self.queues.video.len(), + ) + } + + #[cfg(test)] + pub(crate) fn queued_groups(&self, kind: MediaKind) -> usize { + match kind { + MediaKind::Audio => self.queues.audio.len(), + MediaKind::Video => self.queues.video.len(), + } + } + + #[cfg(test)] + pub(crate) fn queued_payload_bytes(&self, kind: MediaKind) -> usize { + match kind { + MediaKind::Audio => &self.queues.audio, + MediaKind::Video => &self.queues.video, + } + .iter() + .flat_map(|group| &group.payloads) + .map(|payload| payload.payload.len()) + .sum() + } + + #[cfg(test)] + pub(crate) fn oldest_deadline(&self, kind: MediaKind) -> Option { + match kind { + MediaKind::Audio => self.queues.audio.front(), + MediaKind::Video => self.queues.video.front(), + } + .map(|group| group.deadline) + } +} + +fn record_receiver_feedback( + frame: &CultNetTransportFrame, + stats: &mut MuninnRudpReceiverFeedbackStats, + repair_cache: &RecentVideoChunkRepairCache, + now: Instant, +) -> Result> { + if frame.channel_id != MUNINN_MEDIA_RUDP_CHANNEL { + return Ok(Vec::new()); + } + + let GameCultMediaWireRecord::Feedback(feedback) = decode_media_wire_record(&frame.payload)? + else { + return Ok(Vec::new()); + }; + + let repairs = repair_cache.repair_payloads_for_feedback(&feedback, now); + stats.feedback_records = stats.feedback_records.saturating_add(1); + if feedback.requested_keyframe { + stats.requested_keyframes = stats.requested_keyframes.saturating_add(1); + } + stats.late_frames = stats + .late_frames + .saturating_add(feedback.late_frame_ids.len() as u64); + stats.missing_video_chunks = stats + .missing_video_chunks + .saturating_add(feedback.missing_video_chunk_keys.len() as u64); + if let Some(frame_id) = feedback.highest_decodable_frame_id { + stats.highest_decodable_frame_id = Some( + stats + .highest_decodable_frame_id + .map(|current| current.max(frame_id)) + .unwrap_or(frame_id), + ); + } + Ok(repairs) +} + +/// Reads Annex B from `reader`, packetizes it, and hands one group per access +/// unit to the send loop, stamped with the request's deadline. +pub fn spawn_video_group_reader( + tx: MediaIntakeSender, + mut reader: R, + config: VideoAnnexBStreamSendConfig, + policy: MediaSendPolicy, +) -> thread::JoinHandle<()> +where + R: Read + Send + 'static, +{ + thread::spawn(move || { + if let Err(error) = read_video_groups(&tx, &mut reader, config, &policy) { + let _ = tx.send(Err(error)); + } + }) +} + +pub fn spawn_audio_group_reader( + tx: MediaIntakeSender, + mut reader: R, + config: AudioPcmStreamSendConfig, + policy: MediaSendPolicy, +) -> thread::JoinHandle<()> +where + R: Read + Send + 'static, +{ + thread::spawn(move || { + if let Err(error) = read_audio_groups(&tx, &mut reader, config, &policy) { + let _ = tx.send(Err(error)); + } + }) +} + +fn send_intake(tx: &MediaIntakeSender, item: MediaIntake) -> Result<()> { + tx.send(Ok(item)).context("queueing typed Muninn media") +} + +fn read_video_groups( + tx: &MediaIntakeSender, + reader: &mut R, + config: VideoAnnexBStreamSendConfig, + policy: &MediaSendPolicy, +) -> Result<()> { + let mut sender = VideoAnnexBStreamSendState::new(config)?; + let mut buffer = vec![0_u8; 64 * 1024]; + let mut dropped_reported = 0_u64; + loop { + let read = reader + .read(&mut buffer) + .context("reading encoded Annex B video from ffmpeg stdout")?; + let stored_at = crate::timestamp()?; + let (groups, done) = if read == 0 { + (sender.finish(&stored_at)?, true) + } else { + (sender.push(&stored_at, &buffer[..read])?, false) + }; + for payloads in groups { + send_intake( + tx, + MediaIntake::Group(QueuedMediaGroup::new( + MediaKind::Video, + Instant::now(), + policy, + payloads, + )), + )?; + } + let dropped = sender.dropped_access_units(); + if dropped > dropped_reported { + send_intake(tx, MediaIntake::VideoDropped(dropped - dropped_reported))?; + dropped_reported = dropped; + } + if done { + return Ok(()); + } + } +} + +fn read_audio_groups( + tx: &MediaIntakeSender, + reader: &mut R, + config: AudioPcmStreamSendConfig, + policy: &MediaSendPolicy, +) -> Result<()> { + let mut sender = AudioPcmStreamSendState::new(config)?; + let mut buffer = vec![0_u8; 16 * 1024]; + loop { + let read = reader + .read(&mut buffer) + .context("reading PCM audio from ffmpeg stdout")?; + let stored_at = crate::timestamp()?; + let (payloads, done) = if read == 0 { + (sender.finish(&stored_at)?, true) + } else { + (sender.push(&stored_at, &buffer[..read])?, false) + }; + for payload in payloads { + send_intake( + tx, + MediaIntake::Group(QueuedMediaGroup::new( + MediaKind::Audio, + Instant::now(), + policy, + vec![payload], + )), + )?; + } + if done { + return Ok(()); + } + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/muninn-daemon/src/media_send/tests.rs b/crates/muninn-daemon/src/media_send/tests.rs new file mode 100644 index 0000000..7f39e6d --- /dev/null +++ b/crates/muninn-daemon/src/media_send/tests.rs @@ -0,0 +1,845 @@ +use super::*; +use crate::media_packetizer::{MediaWireProvenance, encode_media_wire_record}; +use cultnet_rs::{ + CultNetRudpPacketType, CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, + CultNetTransportDelivery, ReceiverFeedbackOptions, build_receiver_feedback, decode_rudp_packet, +}; +use std::collections::{BTreeMap, BTreeSet}; +use std::io::Cursor; +use std::net::{SocketAddr, UdpSocket}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; + +const CLIP: &[u8] = include_bytes!("../../testdata/clip.h264"); +const STREAM_ID: &str = "muninn.harness.rudp"; +const VIDEO_SESSION_ID: &str = "harness:video"; +const AUDIO_SESSION_ID: &str = "harness:audio"; +const WAIT: Duration = Duration::from_secs(5); + +fn wait_until(timeout: Duration, mut done: impl FnMut() -> bool) -> bool { + let start = Instant::now(); + while start.elapsed() < timeout { + if done() { + return true; + } + thread::sleep(Duration::from_millis(1)); + } + done() +} + +fn video_config() -> VideoAnnexBStreamSendConfig { + VideoAnnexBStreamSendConfig { + stream_id: STREAM_ID.to_string(), + session_id: VIDEO_SESSION_ID.to_string(), + codec: "h264".to_string(), + first_frame_id: 0, + first_pts_ticks: 0, + frame_duration_ticks: 3_000, + timebase_num: 1, + timebase_den: 90_000, + deadline_delay_ticks: 22_500, + max_payload_bytes: 848, + max_pending_bytes: 848 * 4096, + source_runtime_id: "harness".to_string(), + source_role: "muninn.rudp.video".to_string(), + } +} + +fn audio_config() -> AudioPcmStreamSendConfig { + AudioPcmStreamSendConfig { + stream_id: STREAM_ID.to_string(), + session_id: AUDIO_SESSION_ID.to_string(), + codec: "pcm-f32le-interleaved".to_string(), + first_packet_id: 0, + first_pts_ticks: 0, + packet_duration_ticks: 480, + timebase_num: 1, + timebase_den: 48_000, + deadline_delay_ticks: 24_000, + channels: 2, + bytes_per_sample: 4, + max_pending_bytes: 480 * 2 * 4 * 128, + source_runtime_id: "harness".to_string(), + source_role: "muninn.rudp.audio".to_string(), + } +} + +/// Ten audio packets of an ascending f32 ramp: every byte position is +/// distinct, so a reordered or dropped packet cannot pass for the ramp. +fn pcm_ramp() -> Vec { + let samples = 480 * 2 * 10; + (0..samples) + .flat_map(|index| (index as f32 / samples as f32).to_le_bytes()) + .collect() +} + +/// The clip's access units as the packetizer sends them: one group each. +fn clip_groups() -> Vec { + let mut sender = VideoAnnexBStreamSendState::new(video_config()).unwrap(); + let mut groups = sender.push("unix-0", CLIP).unwrap(); + groups.extend(sender.finish("unix-0").unwrap()); + groups +} + +fn group_at( + kind: MediaKind, + produced_at: Instant, + policy: &MediaSendPolicy, + payloads: MuninnMediaPayloadGroup, +) -> QueuedMediaGroup { + QueuedMediaGroup::new(kind, produced_at, policy, payloads) +} + +fn marker_payload(marker: u8) -> MuninnMediaSendPayload { + MuninnMediaSendPayload { + channel_id: MUNINN_MEDIA_RUDP_CHANNEL, + payload: vec![marker], + } +} + +/// `(frame_id, is_parity, chunk or parity index)` of a video wire record. +fn frame_of(payload: &[u8]) -> Option<(u64, bool, u16)> { + match decode_media_wire_record(payload).ok()? { + GameCultMediaWireRecord::Video(record) => Some((record.frame_id, false, record.chunk_index)), + GameCultMediaWireRecord::VideoParity(record) => { + Some((record.frame_id, true, record.parity_index)) + } + _ => None, + } +} + +fn test_hub(policy: &MediaSendPolicy) -> CultNetRudpServerHub { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket.set_nonblocking(true).unwrap(); + CultNetRudpServerHub::new(crate::muninn_media_rudp_hub_options( + socket, + &crate::muninn_rudp_media_profile(), + policy, + )) + .unwrap() +} + +/// The request's budget and no pacing sleeps: the harness is about what the +/// core decides, not about how fast a Windows sender is allowed to go. +fn quiet_policy(latency_budget_ms: u32) -> MediaSendPolicy { + let mut policy = MediaSendPolicy::from_request(latency_budget_ms); + policy.video_pace_sleep = Duration::ZERO; + policy +} + +/// One datagram the hub sent toward the receiver, as the relay saw it. +#[derive(Clone, Debug)] +struct Sent { + packet_type: CultNetRudpPacketType, + channel: String, + payload: Vec, + dropped: bool, +} + +/// A UDP relay between the receiver and the hub. The hub's data packets are +/// dropped by a seeded schedule. Everything is logged before the drop, so +/// tests assert on what the sender put on the wire. +struct DropRelay { + addr: SocketAddr, + log: Arc>>, + stop: Arc, + thread: Option>, +} + +impl DropRelay { + fn start(hub_addr: SocketAddr, drop_percent: u64, seed: u64) -> Self { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_millis(1))) + .unwrap(); + let addr = socket.local_addr().unwrap(); + let log = Arc::new(Mutex::new(Vec::new())); + let stop = Arc::new(AtomicBool::new(false)); + let thread = { + let log = Arc::clone(&log); + let stop = Arc::clone(&stop); + thread::spawn(move || { + let mut client: Option = None; + let mut state = seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1; + let mut buffer = vec![0_u8; 65_535]; + while !stop.load(Ordering::Relaxed) { + let Ok((received, from)) = socket.recv_from(&mut buffer) else { + continue; + }; + if from != hub_addr { + client = Some(from); + let _ = socket.send_to(&buffer[..received], hub_addr); + continue; + } + let Some(client) = client else { continue }; + let Ok(packet) = decode_rudp_packet(&buffer[..received]) else { + continue; + }; + let is_data = packet.packet_type == CultNetRudpPacketType::Data; + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + let dropped = is_data && state % 100 < drop_percent; + log.lock().unwrap().push(Sent { + packet_type: packet.packet_type, + channel: packet.channel_id, + payload: packet.payload, + dropped, + }); + if !dropped { + let _ = socket.send_to(&buffer[..received], client); + } + } + }) + }; + Self { + addr, + log, + stop, + thread: Some(thread), + } + } + + fn data(&self) -> Vec { + self.log + .lock() + .unwrap() + .iter() + .filter(|sent| sent.packet_type == CultNetRudpPacketType::Data) + .cloned() + .collect() + } + + fn video_frames_on_wire(&self) -> Vec<(u64, bool, u16)> { + self.data() + .iter() + .filter(|sent| sent.channel == MUNINN_MEDIA_RUDP_CHANNEL) + .filter_map(|sent| frame_of(&sent.payload)) + .collect() + } +} + +impl Drop for DropRelay { + fn drop(&mut self) { + self.stop.store(true, Ordering::Relaxed); + if let Some(thread) = self.thread.take() { + let _ = thread.join(); + } + } +} + +struct Rig { + core: MediaSendCore, + client: CultNetRudpSocketTransportConnection, + relay: DropRelay, + policy: MediaSendPolicy, +} + +fn rig(latency_budget_ms: u32, drop_percent: u64, seed: u64) -> Rig { + let policy = quiet_policy(latency_budget_ms); + let hub = test_hub(&policy); + let relay = DropRelay::start(hub.local_addr().unwrap(), drop_percent, seed); + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket.set_nonblocking(true).unwrap(); + let mut options = CultNetRudpSocketTransportOptions::client( + "harness-receiver", + socket, + relay.addr, + crate::MUNINN_MEDIA_RUDP_CONNECTION_ID, + ); + options.media_delivery = Some(CultNetTransportDelivery::Unreliable); + options.max_fragment_bytes = Some(crate::MUNINN_RUDP_MEDIA_MAX_FRAGMENT_BYTES as u32); + let mut client = CultNetRudpSocketTransportConnection::new(options).unwrap(); + let mut core = MediaSendCore::new(hub, &policy); + client.connect(b"harness-receiver".to_vec()).unwrap(); + assert!( + wait_until(WAIT, || { + core.service_control(Instant::now()).unwrap(); + let _ = client.receive_once().unwrap(); + client.connected() && core.hub().sessions().len() == 1 + }), + "receiver never attached through the relay" + ); + Rig { + core, + client, + relay, + policy, + } +} + +// ---- policy, groups and queues ----------------------------------------------------- + +#[test] +fn a_group_is_expired_only_after_its_deadline_and_never_at_it() { + let policy = quiet_policy(250); + let produced_at = Instant::now(); + let group = group_at(MediaKind::Video, produced_at, &policy, Vec::new()); + + assert!(!group.expired(group.deadline)); + assert!(!group.expired(group.deadline - Duration::from_millis(1))); + assert!(group.expired(group.deadline + Duration::from_millis(1))); +} + +#[test] +fn a_zero_request_budget_still_leaves_the_stream_a_millisecond() { + assert_eq!( + MediaSendPolicy::from_request(0).latency_budget, + Duration::from_millis(1) + ); +} + +#[test] +fn the_queue_caps_groups_per_kind_and_drops_the_oldest_first() { + let policy = quiet_policy(250); + let mut core = MediaSendCore::new(test_hub(&policy), &policy); + let t0 = Instant::now(); + let over = 37; + for index in 0..(MEDIA_SEND_VIDEO_GROUP_CAP + over) { + core.enqueue(group_at( + MediaKind::Video, + t0 + Duration::from_millis(index as u64), + &policy, + vec![marker_payload(1)], + )); + } + for index in 0..(MEDIA_SEND_AUDIO_GROUP_CAP + 5) { + core.enqueue(group_at( + MediaKind::Audio, + t0 + Duration::from_millis(index as u64), + &policy, + vec![marker_payload(2)], + )); + } + + assert_eq!( + core.queued_groups(MediaKind::Video), + MEDIA_SEND_VIDEO_GROUP_CAP + ); + assert_eq!( + core.queued_groups(MediaKind::Audio), + MEDIA_SEND_AUDIO_GROUP_CAP + ); + // One byte per group: resident bytes are bounded by the cap times the group size. + assert_eq!( + core.queued_payload_bytes(MediaKind::Video), + MEDIA_SEND_VIDEO_GROUP_CAP + ); + assert_eq!( + core.queued_payload_bytes(MediaKind::Audio), + MEDIA_SEND_AUDIO_GROUP_CAP + ); + assert_eq!(core.stats().groups_dropped_video, over as u64); + assert_eq!(core.stats().groups_dropped_audio, 5); + // The survivors are the newest: the oldest deadline left is the 38th group's. + assert_eq!( + core.oldest_deadline(MediaKind::Video), + Some(t0 + Duration::from_millis(over as u64) + policy.latency_budget) + ); + assert_eq!( + core.oldest_deadline(MediaKind::Audio), + Some(t0 + Duration::from_millis(5) + policy.latency_budget) + ); +} + +#[test] +fn the_intake_channel_holds_the_reader_at_its_bound_instead_of_growing() { + let policy = quiet_policy(5_000); + let (tx, rx) = media_intake_channel(); + let frames = clip_groups().len(); + assert!(frames > MEDIA_GROUP_CHANNEL_BOUND + 1); + let reader = spawn_video_group_reader(tx, Cursor::new(CLIP.to_vec()), video_config(), policy); + + // Nothing drains, so the reader parks with the channel full. + thread::sleep(Duration::from_millis(300)); + assert!( + !reader.is_finished(), + "an unbounded channel let the reader run to the end" + ); + + let mut received = 0; + wait_until(WAIT, || { + while rx.try_recv().is_ok() { + received += 1; + } + received == frames + }); + assert_eq!(received, frames); + assert!(wait_until(WAIT, || reader.is_finished())); +} + +#[test] +fn a_blocked_send_gives_up_at_the_group_deadline_and_not_before() { + let start = Instant::now(); + let deadline = start + Duration::from_millis(3); + let ticks = std::cell::Cell::new(0_u64); + let clock = || { + ticks.set(ticks.get() + 1); + start + Duration::from_millis(ticks.get()) + }; + let mut polls = 0_u32; + + // Blocked forever: the deadline ends it, after polling resends in between. + let sent = send_until( + &mut polls, + deadline, + clock, + |_| Err(std::io::Error::from(ErrorKind::WouldBlock).into()), + |polls| { + *polls += 1; + Ok(()) + }, + ) + .unwrap(); + assert!(!sent); + assert_eq!(polls, 3, "one poll per blocked attempt inside the deadline"); + + // Blocked twice, then through: sent, without waiting for the deadline. + ticks.set(0); + let mut attempts = 0_u32; + let sent = send_until( + &mut attempts, + start + Duration::from_millis(100), + clock, + |attempts| { + *attempts += 1; + if *attempts < 3 { + Err(std::io::Error::from(ErrorKind::WouldBlock).into()) + } else { + Ok(()) + } + }, + |_| Ok(()), + ) + .unwrap(); + assert!(sent); + assert_eq!(attempts, 3); +} + +#[test] +fn a_send_error_that_is_not_backpressure_is_an_error_not_a_dropped_group() { + let error = send_until( + &mut (), + Instant::now() + Duration::from_secs(1), + Instant::now, + |_| Err(anyhow::anyhow!("socket closed")), + |_| Ok(()), + ) + .unwrap_err(); + assert!(format!("{error:#}").contains("socket closed")); +} + +// ---- moved from the mux: repair budget, feedback intake, repair cache --------------- + +#[test] +fn rudp_repair_budget_backs_off_on_media_drops_and_recovers_when_stable() { + let start = Instant::now(); + let mut budget = MuninnRudpRepairBudget { + chunks_per_second: 64, + min_chunks_per_second: 8, + max_chunks_per_second: 128, + add_chunks_per_second: 8, + recovery_interval: Duration::from_secs(2), + max_available_chunks: 4, + available_chunks: 4, + last_refill_at: start, + last_rate_adjust_at: start, + last_queue_dropped: 0, + }; + + assert_eq!(budget.take(4, start, 0), 4); + assert_eq!(budget.chunks_per_second(), 64); + assert_eq!(budget.take(3, start + Duration::from_millis(10), 1), 0); + assert_eq!(budget.chunks_per_second(), 32); + assert_eq!(budget.take(3, start + Duration::from_secs(1), 1), 3); + assert_eq!(budget.chunks_per_second(), 32); + assert_eq!(budget.take(3, start + Duration::from_secs(3), 1), 3); + assert_eq!(budget.chunks_per_second(), 40); +} + +#[test] +fn default_rudp_repair_budget_has_lan_stream_headroom() { + let mut budget = + MuninnRudpRepairBudget::new(REPAIR_INITIAL_CHUNKS_PER_SECOND, REPAIR_BURST_CHUNKS); + let start = budget.last_refill_at; + + assert_eq!(budget.chunks_per_second(), 4_096); + assert_eq!(budget.take(2_048, start, 0), 2_048); + assert_eq!(budget.take(8_192, start + Duration::from_secs(1), 0), 2_048); + assert_eq!(budget.take(8_192, start + Duration::from_secs(3), 0), 2_048); + assert_eq!(budget.chunks_per_second(), 6_144); + assert_eq!(budget.take(512, start + Duration::from_secs(4), 1), 512); + assert_eq!(budget.chunks_per_second(), 3_072); +} + +fn feedback_for( + missing_chunk_keys: Vec, + requested_keyframe: bool, +) -> GameCultMediaReceiverFeedbackRecord { + build_receiver_feedback(ReceiverFeedbackOptions { + stream_id: STREAM_ID, + session_id: VIDEO_SESSION_ID, + receiver_id: "harness-receiver", + highest_decodable_frame_id: Some(41), + missing_frame_ids: Vec::new(), + missing_video_chunk_keys: missing_chunk_keys, + late_frame_ids: vec![42, 43], + requested_keyframe, + jitter_us: 500, + decode_queue_us: 2_000, + observed_at: "unix:1000", + }) + .unwrap() +} + +fn feedback_wire(feedback: GameCultMediaReceiverFeedbackRecord) -> Vec { + encode_media_wire_record( + &GameCultMediaWireRecord::Feedback(feedback), + MediaWireProvenance { + stored_at: "unix:1000", + runtime_id: "harness", + role: "harness.receiver", + producer: "harness", + }, + ) + .unwrap() +} + +#[test] +fn rudp_media_receiver_feedback_updates_sender_pressure_stats() { + let frame = CultNetTransportFrame { + channel_id: MUNINN_MEDIA_RUDP_CHANNEL.to_string(), + payload: feedback_wire(feedback_for(vec!["42:1".into(), "42:3".into()], true)), + }; + let mut stats = MuninnRudpReceiverFeedbackStats::default(); + let repair_cache = RecentVideoChunkRepairCache::new(16); + + let repairs = + record_receiver_feedback(&frame, &mut stats, &repair_cache, Instant::now()).unwrap(); + + assert_eq!(stats.feedback_records, 1); + assert_eq!(stats.requested_keyframes, 1); + assert_eq!(stats.late_frames, 2); + assert_eq!(stats.missing_video_chunks, 2); + assert_eq!(stats.repaired_video_chunks, 0); + assert_eq!(stats.highest_decodable_frame_id, Some(41)); + assert!(repairs.is_empty()); +} + +#[test] +fn repair_cache_returns_recent_missing_video_chunks_with_their_frames_deadline() { + let groups = clip_groups(); + let payload = groups[0][0].clone(); + let (frame_id, _, chunk_index) = frame_of(&payload.payload).unwrap(); + let feedback = feedback_for( + vec![ + video_chunk_feedback_key(frame_id, chunk_index), + "999:0".to_string(), + ], + false, + ); + let deadline = Instant::now() + Duration::from_millis(250); + + let mut cache = RecentVideoChunkRepairCache::new(16); + cache.remember(&payload, deadline).unwrap(); + let repairs = cache.repair_payloads_for_feedback(&feedback, deadline); + + assert_eq!(repairs, vec![(payload, deadline)]); +} + +// ---- the wire: whole groups, repairs inside their frame's deadline ------------------ + +#[test] +fn a_group_past_its_deadline_puts_nothing_on_the_wire_and_a_live_one_puts_all_of_it() { + let mut rig = rig(250, 0, 1); + let groups = clip_groups(); + let stale = groups[0].clone(); + let live = groups + .iter() + .skip(1) + .find(|group| group.len() > 2) + .expect("the clip has a later multi-chunk frame") + .clone(); + let live_frame = frame_of(&live[0].payload).unwrap().0; + let t0 = Instant::now(); + let budget = rig.policy.latency_budget; + rig.core + .enqueue(group_at(MediaKind::Video, t0, &rig.policy, stale)); + rig.core.enqueue(group_at( + MediaKind::Video, + t0 + 5 * budget, + &rig.policy, + live.clone(), + )); + + // Two budgets after the stale group was made, inside the live group's. + assert!(rig.core.send_next(t0 + 2 * budget).unwrap()); + + assert_eq!(rig.core.stats().groups_expired, 1); + assert_eq!(rig.core.stats().groups_sent, 1); + assert!(wait_until(WAIT, || { + rig.relay.video_frames_on_wire().len() == live.len() + })); + thread::sleep(Duration::from_millis(100)); + let on_wire = rig.relay.video_frames_on_wire(); + assert_eq!( + on_wire.len(), + live.len(), + "every payload of the live group, and no other" + ); + assert!(on_wire.iter().all(|(frame, _, _)| *frame == live_frame)); + assert!(rig.core.progress_detail().contains("groups_sent=1")); + assert!(rig.core.progress_detail().contains("groups_expired=1")); +} + +#[test] +fn audio_leaves_before_video_on_the_wire() { + let mut rig = rig(250, 0, 1); + let t0 = Instant::now(); + let video = clip_groups().remove(1); + let mut audio_sender = AudioPcmStreamSendState::new(audio_config()).unwrap(); + let audio = audio_sender.push("unix-0", &pcm_ramp()).unwrap().remove(0); + rig.core + .enqueue(group_at(MediaKind::Video, t0, &rig.policy, video)); + rig.core + .enqueue(group_at(MediaKind::Audio, t0, &rig.policy, vec![audio])); + + assert!(rig.core.send_next(t0).unwrap()); + assert!(rig.core.send_next(t0).unwrap()); + + assert!(wait_until(WAIT, || !rig.relay.data().is_empty())); + assert_eq!(rig.relay.data()[0].channel, "audio"); +} + +/// Sends the clip's first multi-chunk frame at `t0` and returns its deadline, +/// its frame id and how many payloads it put on the wire. +fn send_repairable_frame(rig: &mut Rig, t0: Instant) -> (Instant, u64, usize) { + let group = clip_groups() + .into_iter() + .find(|group| group.len() > 2) + .unwrap(); + let (frame_id, _, _) = frame_of(&group[0].payload).unwrap(); + let queued = group_at(MediaKind::Video, t0, &rig.policy, group.clone()); + let deadline = queued.deadline; + rig.core.enqueue(queued); + assert!(rig.core.send_next(t0).unwrap()); + assert!(wait_until(WAIT, || rig.relay.video_frames_on_wire().len() + == group.len())); + (deadline, frame_id, group.len()) +} + +fn ask_for_chunk(rig: &mut Rig, frame_id: u64, at: Instant, records_seen: u64) { + rig.client + .send( + MUNINN_MEDIA_RUDP_CHANNEL, + feedback_wire(feedback_for( + vec![video_chunk_feedback_key(frame_id, 0)], + false, + )), + ) + .unwrap(); + assert!( + wait_until(WAIT, || { + rig.core.service_control(at).unwrap(); + rig.core.feedback().feedback_records == records_seen + }), + "the hub never received the feedback" + ); +} + +#[test] +fn a_repair_is_sent_inside_its_frames_deadline_and_not_after_it() { + // Taken before the handshake, so real time is well past it by the time + // anything is sent: a repair that took its deadline from the clock + // instead of its frame would still be alive at the probe below. + let t0 = Instant::now(); + let mut rig = rig(250, 0, 1); + let (deadline, frame_id, frame_payloads) = send_repairable_frame(&mut rig, t0); + let epsilon = Duration::from_millis(2); + + ask_for_chunk(&mut rig, frame_id, deadline - epsilon, 1); + assert!( + wait_until(WAIT, || rig.relay.video_frames_on_wire().len() + == frame_payloads + 1), + "a repair asked for just before the deadline is sent" + ); + assert_eq!(rig.core.feedback().repaired_video_chunks, 1); + + ask_for_chunk(&mut rig, frame_id, deadline + epsilon, 2); + thread::sleep(Duration::from_millis(150)); + assert_eq!( + rig.relay.video_frames_on_wire().len(), + frame_payloads + 1, + "a repair asked for just after the deadline is not sent" + ); + assert_eq!(rig.core.feedback().repaired_video_chunks, 1); + assert_eq!(rig.core.feedback().missing_video_chunks, 2); +} + +// ---- the harness end to end: the fixture clip and the PCM ramp --------------------- + +#[derive(Default)] +struct Received { + chunks: BTreeMap>, + chunk_counts: BTreeMap, + audio: BTreeMap>, +} + +impl Received { + fn take(&mut self, rig: &mut Rig) { + while let Some(frame) = rig.client.receive_once().unwrap() { + match decode_media_wire_record(&frame.payload).unwrap() { + GameCultMediaWireRecord::Video(record) => { + self.chunks + .entry(record.frame_id) + .or_default() + .insert(record.chunk_index); + self.chunk_counts + .insert(record.frame_id, record.chunk_count); + } + GameCultMediaWireRecord::Audio(record) => { + self.audio.insert(record.packet_id, record.payload); + } + _ => {} + } + } + } + + fn complete_frames(&self) -> BTreeSet { + self.chunks + .iter() + .filter(|(frame, chunks)| chunks.len() == self.chunk_counts[frame] as usize) + .map(|(frame, _)| *frame) + .collect() + } +} + +/// Feeds the fixture through the real readers and the send loop, over the +/// relay, and returns what the receiver assembled. +fn run_clip(rig: &mut Rig, with_audio: bool) -> Received { + let (tx, rx) = media_intake_channel(); + let video = spawn_video_group_reader( + tx.clone(), + Cursor::new(CLIP.to_vec()), + video_config(), + rig.policy.clone(), + ); + let audio = with_audio.then(|| { + spawn_audio_group_reader( + tx.clone(), + Cursor::new(pcm_ramp()), + audio_config(), + rig.policy.clone(), + ) + }); + drop(tx); + let mut received = Received::default(); + let start = Instant::now(); + while !rig.core.finished() && start.elapsed() < WAIT { + rig.core.intake(&rx, Duration::from_millis(2)).unwrap(); + rig.core.send_next(Instant::now()).unwrap(); + rig.core.service_control(Instant::now()).unwrap(); + received.take(rig); + } + assert!(rig.core.finished(), "the send loop never finished the clip"); + let settle = Instant::now(); + while settle.elapsed() < Duration::from_millis(300) { + rig.core.service_control(Instant::now()).unwrap(); + received.take(rig); + thread::sleep(Duration::from_millis(1)); + } + video.join().unwrap(); + if let Some(audio) = audio { + audio.join().unwrap(); + } + received +} + +#[test] +fn the_clip_and_the_pcm_ramp_arrive_whole_through_a_lossless_relay() { + // A budget far above the run time: nothing here may expire. + let mut rig = rig(5_000, 0, 1); + let frames = clip_groups().len() as u64; + + let received = run_clip(&mut rig, true); + + assert_eq!( + received.complete_frames(), + (0..frames).collect::>() + ); + assert_eq!(rig.core.stats().groups_lost(), 0); + let heard: Vec = received.audio.values().flatten().copied().collect(); + assert_eq!(received.audio.len(), 10); + assert_eq!(heard, pcm_ramp(), "the ramp arrived complete and in order"); +} + +#[test] +fn the_seeded_relay_drops_the_same_datagrams_every_run_and_the_sender_carries_on() { + let run = |seed| { + let mut rig = rig(5_000, 30, seed); + let received = run_clip(&mut rig, false); + let dropped: Vec = rig.relay.data().iter().map(|sent| sent.dropped).collect(); + (received.complete_frames(), dropped, rig.core.stats().clone()) + }; + + let (complete_a, dropped_a, stats_a) = run(7); + let (complete_b, dropped_b, _) = run(7); + let (_, dropped_other, _) = run(8); + let frames = clip_groups().len(); + + assert_eq!(dropped_a, dropped_b, "same seed, same drop schedule"); + assert_ne!( + dropped_a, dropped_other, + "a different seed drops different datagrams" + ); + assert_eq!(complete_a, complete_b); + assert!(dropped_a.iter().any(|dropped| *dropped)); + assert!( + complete_a.len() < frames, + "30% loss without parity or repair leaves incomplete frames" + ); + assert_eq!(stats_a.groups_sent, frames as u64); + assert_eq!( + stats_a.groups_lost(), + 0, + "loss on the wire is not a sender-side drop" + ); +} + +#[test] +fn a_group_is_counted_by_what_the_deadline_let_reach_the_wire() { + let mut stats = MediaSendStats::default(); + + stats.record_group(&GroupSend::Complete, 6); + assert_eq!((stats.groups_sent, stats.payloads_sent), (1, 6)); + + // Deadline passed before the first payload: nothing left, so nothing was cut. + stats.record_group(&GroupSend::DeadlinePassed { payloads_out: 0 }, 6); + assert_eq!((stats.groups_expired, stats.groups_cut_short), (1, 0)); + assert_eq!(stats.payloads_sent, 6); + + // Deadline passed with part of the group already out: a partial frame. + stats.record_group(&GroupSend::DeadlinePassed { payloads_out: 2 }, 6); + assert_eq!((stats.groups_expired, stats.groups_cut_short), (1, 1)); + assert_eq!(stats.payloads_sent, 8); + assert_eq!(stats.groups_lost(), 2); +} + +#[test] +fn an_oversize_access_unit_reaches_the_sender_as_a_dropped_video_group_not_an_error() { + let policy = quiet_policy(250); + let mut core = MediaSendCore::new(test_hub(&policy), &policy); + let (tx, rx) = media_intake_channel(); + let mut config = video_config(); + config.max_pending_bytes = 8; + let mut oversize = vec![0, 0, 0, 1, 0x65, 0x80]; + oversize.extend_from_slice(&[7; 20]); + let reader = spawn_video_group_reader(tx, Cursor::new(oversize), config, policy); + + assert!(wait_until(WAIT, || reader.is_finished())); + core.intake(&rx, Duration::from_millis(50)).unwrap(); + core.intake(&rx, Duration::from_millis(50)).unwrap(); + + assert_eq!(core.stats().groups_dropped_video, 1); + assert!(core.finished(), "the stream ended cleanly, it was not restarted by an error"); +} diff --git a/crates/muninn-daemon/testdata/clip.h264 b/crates/muninn-daemon/testdata/clip.h264 new file mode 100644 index 0000000..860d9e0 Binary files /dev/null and b/crates/muninn-daemon/testdata/clip.h264 differ diff --git a/crates/muninn-daemon/testdata/clip.h264.provenance.txt b/crates/muninn-daemon/testdata/clip.h264.provenance.txt new file mode 100644 index 0000000..3c0029b --- /dev/null +++ b/crates/muninn-daemon/testdata/clip.h264.provenance.txt @@ -0,0 +1,7 @@ +clip.h264: 15 frames of the ffmpeg testsrc2 pattern, 160x120 at 30 fps, H.264 Constrained +Baseline in Annex B with access unit delimiters, one IDR every 8 frames, no B-frames. +Generated once and committed so the media send harness needs no ffmpeg. + +Command (ffmpeg 8.1.1 full build, libx264, 2026-09-29): + +ffmpeg -f lavfi -i "testsrc2=size=160x120:rate=30:duration=0.5" -c:v libx264 -profile:v baseline -preset veryslow -crf 28 -bf 0 -g 8 -x264-params "threads=1:aud=1:repeat-headers=1:keyint=8:min-keyint=8:scenecut=0" -f h264 clip.h264