1use std::collections::BTreeMap;2use std::sync::atomic::{AtomicU64, Ordering};3use std::sync::{Arc, Mutex as StdMutex, MutexGuard as StdMutexGuard};4use std::time::{Duration, Instant};56use iroh::Endpoint;7use sigil_protocol::{AUDIO_ALPN_V1, AudioFlags, AudioPacket, AudioPacketHeader, Capability};8use tauri::{9 AppHandle, Emitter, State,10 ipc::{Channel, Response},11};1213use super::transport::negotiate_v1;14use crate::commands::state::{AUDIO_DELIVERY_CAPACITY, AppState, AudioDeliveryState};15use crate::media::moq_receiver::MoqAudioReceiver;1617// Three 20 ms Opus packets cap the Rust→webview handoff at 60 ms. The18// AudioWorklet owns a separate fixed ring and never feeds back into transport.19const AUDIO_CHANNEL_MAGIC: [u8; 4] = *b"SGAC";20const AUDIO_CHANNEL_VERSION: u16 = 1;21const AUDIO_CHANNEL_HEADER_LEN: usize = 24;2223pub(crate) fn lock_audio_deliveries(24 deliveries: &StdMutex<AudioDeliveryState>,25) -> StdMutexGuard<'_, AudioDeliveryState> {26 deliveries27 .lock()28 .unwrap_or_else(std::sync::PoisonError::into_inner)29}3031pub(crate) fn next_audio_generation(counter: &AtomicU64) -> Result<u64, String> {32 counter33 .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |current| {34 current.checked_add(1)35 })36 .map(|previous| previous + 1)37 .map_err(|_| "Audio connection generation overflowed".to_string())38}3940fn audio_generation_is_current(counter: &AtomicU64, generation: u64) -> bool {41 counter.load(Ordering::SeqCst) == generation42}4344fn emit_audio_event_if_current(45 app: &AppHandle,46 event: &str,47 generation_counter: &AtomicU64,48 deliveries: &StdMutex<AudioDeliveryState>,49 generation: u64,50 payload: impl FnOnce(&AudioDeliveryState) -> serde_json::Value,51) -> bool {52 let deliveries = lock_audio_deliveries(deliveries);53 if generation_counter.load(Ordering::SeqCst) != generation54 || deliveries.generation() != Some(generation)55 {56 return false;57 }58 let _ = app.emit(event, payload(&deliveries));59 true60}616263struct AudioReorderBuffer {64 expected_sequence: Option<u64>,65 packets: BTreeMap<u64, AudioPacket>,66}676869struct OrderedAudioPacket {70 packet: AudioPacket,71 discontinuity: bool,72}7374impl AudioReorderBuffer {75 const CAPACITY: usize = 3;7677 fn insert(&mut self, packet: AudioPacket) -> Result<(Vec<OrderedAudioPacket>, u64), String> {78 let sequence = packet.header.sequence;79 let expected = self.expected_sequence.get_or_insert(sequence);80 if sequence < *expected || self.packets.contains_key(&sequence) {81 return Ok((Vec::new(), 0));82 }83 self.packets.insert(sequence, packet);8485 let mut dropped = 0_u64;86 let mut discontinuity = false;87 if !self.packets.contains_key(expected) && self.packets.len() >= Self::CAPACITY {88 let next = *self89 .packets90 .first_key_value()91 .expect("capacity check guarantees one packet")92 .0;93 dropped = next.saturating_sub(*expected);94 *expected = next;95 discontinuity = true;96 }9798 let mut ordered = Vec::with_capacity(Self::CAPACITY);99 while let Some(packet) = self.packets.remove(expected) {100 ordered.push(OrderedAudioPacket {101 packet,102 discontinuity,103 });104 discontinuity = false;105 *expected = expected106 .checked_add(1)107 .ok_or_else(|| "Audio sequence overflowed".to_string())?;108 }109 Ok((ordered, dropped))110 }111}112113fn encode_audio_channel_packet(114 generation: u64,115 delivery_id: u64,116 packet: AudioPacket,117 force_discontinuity: bool,118) -> Result<Vec<u8>, String> {119 let protocol_packet = if force_discontinuity {120 let header = AudioPacketHeader::opus(121 packet.payload.len(),122 packet.header.sequence,123 packet.header.capture_timestamp_us,124 packet.header.pts_us,125 AudioFlags::DISCONTINUITY,126 )127 .map_err(|error| error.to_string())?;128 AudioPacket::new(header, packet.payload).map_err(|error| error.to_string())?129 } else {130 packet131 };132 let datagram = protocol_packet133 .encode_datagram()134 .map_err(|error| error.to_string())?;135 let mut envelope = Vec::with_capacity(AUDIO_CHANNEL_HEADER_LEN + datagram.len());136 envelope.extend_from_slice(&AUDIO_CHANNEL_MAGIC);137 envelope.extend_from_slice(&AUDIO_CHANNEL_VERSION.to_be_bytes());138 envelope.extend_from_slice(&(AUDIO_CHANNEL_HEADER_LEN as u16).to_be_bytes());139 envelope.extend_from_slice(&generation.to_be_bytes());140 envelope.extend_from_slice(&delivery_id.to_be_bytes());141 envelope.extend_from_slice(&datagram);142 debug_assert_eq!(envelope.len(), AUDIO_CHANNEL_HEADER_LEN + datagram.len());143 Ok(envelope)144}145146pub(crate) struct AudioStartRequest {147 pub(crate) address: iroh::EndpointAddr,148 pub(crate) handshake_nonce: [u8; 16],149 pub(crate) media_session_id: u64,150 pub(crate) audio_supported: bool,151 pub(crate) audio_channel: Channel<Response>,152 pub(crate) audio_deliveries: Arc<StdMutex<AudioDeliveryState>>,153 pub(crate) connection_generation: Arc<AtomicU64>,154 pub(crate) generation: u64,155}156157pub(crate) async fn try_start_audio(158 app: AppHandle,159 endpoint: &Endpoint,160 request: AudioStartRequest,161) -> Result<iroh::endpoint::Connection, String> {162 if !request.audio_supported {163 return Err("WebCodecs Opus AudioDecoder is unavailable".to_string());164 }165 let media_session_id = request.media_session_id;166 let audio_connection = tokio::time::timeout(167 Duration::from_secs(3),168 endpoint.connect(request.address, AUDIO_ALPN_V1),169 )170 .await171 .map_err(|_| "Timed out connecting optional audio".to_string())?172 .map_err(|error| format!("Audio connection unavailable: {error}"))?;173 let (mut send, mut recv) = audio_connection174 .open_bi()175 .await176 .map_err(|error| format!("Failed to open audio handshake: {error}"))?;177 let negotiation = negotiate_v1(178 &mut send,179 &mut recv,180 request.handshake_nonce,181 vec![Capability::AudioOpus],182 Some(Capability::AudioOpus),183 "audio",184 None,185 )186 .await?;187 if negotiation.session_id != media_session_id {188 audio_connection.close(1_u32.into(), b"audio session mismatch");189 return Err("Host returned mismatched media and audio sessions".to_string());190 }191 send.finish()192 .map_err(|error| format!("Failed to finish audio handshake: {error}"))?;193194 let task_audio_connection = audio_connection.clone();195 tokio::spawn(async move {196 let audio_channel = request.audio_channel;197 let audio_deliveries = request.audio_deliveries;198 let connection_generation = request.connection_generation;199 let generation = request.generation;200 let mut reorder = AudioReorderBuffer::default();201 let mut transport_received_total = 0_u64;202 let mut sequence_dropped_total = 0_u64;203 let mut frontend_dropped_total = 0_u64;204 let mut frontend_sent_total = 0_u64;205 let mut pending_discontinuity = false;206 let mut last_stats = Instant::now();207208 loop {209 let datagram = match task_audio_connection.read_datagram().await {210 Ok(datagram) => datagram,211 Err(error) => {212 emit_audio_event_if_current(213 &app,214 "audio-state",215 &connection_generation,216 &audio_deliveries,217 generation,218 |_| {219 serde_json::json!({220 "generation": generation,221 "available": false,222 "error": format!("Audio connection ended: {error}")223 })224 },225 );226 break;227 }228 };229 if !audio_generation_is_current(&connection_generation, generation) {230 break;231 }232 let packet = match AudioPacket::decode_datagram(&datagram) {233 Ok(packet) => packet,234 Err(error) => {235 emit_audio_event_if_current(236 &app,237 "audio-state",238 &connection_generation,239 &audio_deliveries,240 generation,241 |_| {242 serde_json::json!({243 "generation": generation,244 "available": false,245 "error": format!("Invalid audio packet: {error}")246 })247 },248 );249 break;250 }251 };252 transport_received_total = transport_received_total.saturating_add(1);253 let (packets, dropped) = match reorder.insert(packet) {254 Ok(result) => result,255 Err(error) => {256 emit_audio_event_if_current(257 &app,258 "audio-state",259 &connection_generation,260 &audio_deliveries,261 generation,262 |_| {263 serde_json::json!({264 "generation": generation,265 "available": false,266 "error": error267 })268 },269 );270 break;271 }272 };273 sequence_dropped_total = sequence_dropped_total.saturating_add(dropped);274 if dropped > 0 {275 pending_discontinuity = true;276 }277278 for ordered in packets {279 let delivery_id = match lock_audio_deliveries(&audio_deliveries).reserve(generation)280 {281 Ok(Some(delivery_id)) => delivery_id,282 Ok(None) => {283 frontend_dropped_total = frontend_dropped_total.saturating_add(1);284 pending_discontinuity = true;285 continue;286 }287 Err(_) => return,288 };289 let envelope = match encode_audio_channel_packet(290 generation,291 delivery_id,292 ordered.packet,293 ordered.discontinuity || pending_discontinuity,294 ) {295 Ok(envelope) => envelope,296 Err(error) => {297 lock_audio_deliveries(&audio_deliveries)298 .release_failed_delivery(generation, delivery_id);299 emit_audio_event_if_current(300 &app,301 "audio-state",302 &connection_generation,303 &audio_deliveries,304 generation,305 |_| {306 serde_json::json!({307 "generation": generation,308 "available": false,309 "error": error310 })311 },312 );313 return;314 }315 };316 if !audio_generation_is_current(&connection_generation, generation) {317 lock_audio_deliveries(&audio_deliveries)318 .release_failed_delivery(generation, delivery_id);319 return;320 }321 if audio_channel.send(Response::new(envelope)).is_err() {322 lock_audio_deliveries(&audio_deliveries)323 .release_failed_delivery(generation, delivery_id);324 emit_audio_event_if_current(325 &app,326 "audio-state",327 &connection_generation,328 &audio_deliveries,329 generation,330 |_| {331 serde_json::json!({332 "generation": generation,333 "available": false,334 "error": "Audio webview channel closed"335 })336 },337 );338 return;339 }340 frontend_sent_total = frontend_sent_total.saturating_add(1);341 pending_discontinuity = false;342 }343344 if last_stats.elapsed() >= Duration::from_millis(250) {345 if !emit_audio_event_if_current(346 &app,347 "audio-stats",348 &connection_generation,349 &audio_deliveries,350 generation,351 |deliveries| {352 serde_json::json!({353 "generation": generation,354 "transport_received_total": transport_received_total,355 "sequence_dropped_total": sequence_dropped_total,356 "frontend_dropped_total": frontend_dropped_total,357 "frontend_sent_total": frontend_sent_total,358 "frontend_queue_depth": deliveries.depth(generation).unwrap_or(0),359 "frontend_queue_capacity": AUDIO_DELIVERY_CAPACITY,360 })361 },362 ) {363 return;364 }365 last_stats = Instant::now();366 }367 }368 });369 Ok(audio_connection)370}371372pub(crate) struct MoqAudioStartRequest {373 pub(crate) audio_supported: bool,374 pub(crate) audio_channel: Channel<Response>,375 pub(crate) audio_deliveries: Arc<StdMutex<AudioDeliveryState>>,376 pub(crate) connection_generation: Arc<AtomicU64>,377 pub(crate) generation: u64,378}379380pub(crate) fn try_start_moq_audio(381 app: AppHandle,382 mut receiver: MoqAudioReceiver,383 request: MoqAudioStartRequest,384) -> Result<(), String> {385 if !request.audio_supported {386 return Err("WebCodecs Opus AudioDecoder is unavailable".to_string());387 }388 tokio::spawn(async move {389 let mut reorder = AudioReorderBuffer::default();390 let mut transport_received_total = 0_u64;391 let mut sequence_dropped_total = 0_u64;392 let mut frontend_dropped_total = 0_u64;393 let mut frontend_sent_total = 0_u64;394 let mut pending_discontinuity = false;395 let mut last_stats = Instant::now();396397 loop {398 let packet = match receiver.next().await {399 Ok(Some(packet)) => packet,400 Ok(None) => {401 emit_audio_event_if_current(402 &app,403 "audio-state",404 &request.connection_generation,405 &request.audio_deliveries,406 request.generation,407 |_| {408 serde_json::json!({409 "generation": request.generation,410 "available": false,411 "error": "Authenticated MoQ audio track ended"412 })413 },414 );415 break;416 }417 Err(error) => {418 emit_audio_event_if_current(419 &app,420 "audio-state",421 &request.connection_generation,422 &request.audio_deliveries,423 request.generation,424 |_| {425 serde_json::json!({426 "generation": request.generation,427 "available": false,428 "error": error429 })430 },431 );432 break;433 }434 };435 if !audio_generation_is_current(&request.connection_generation, request.generation) {436 break;437 }438 transport_received_total = transport_received_total.saturating_add(1);439 let (packets, dropped) = match reorder.insert(packet) {440 Ok(result) => result,441 Err(error) => {442 emit_audio_event_if_current(443 &app,444 "audio-state",445 &request.connection_generation,446 &request.audio_deliveries,447 request.generation,448 |_| {449 serde_json::json!({450 "generation": request.generation,451 "available": false,452 "error": error453 })454 },455 );456 break;457 }458 };459 sequence_dropped_total = sequence_dropped_total.saturating_add(dropped);460 pending_discontinuity |= dropped > 0;461 for ordered in packets {462 let delivery_id = match lock_audio_deliveries(&request.audio_deliveries)463 .reserve(request.generation)464 {465 Ok(Some(delivery_id)) => delivery_id,466 Ok(None) => {467 frontend_dropped_total = frontend_dropped_total.saturating_add(1);468 pending_discontinuity = true;469 continue;470 }471 Err(_) => return,472 };473 let envelope = match encode_audio_channel_packet(474 request.generation,475 delivery_id,476 ordered.packet,477 ordered.discontinuity || pending_discontinuity,478 ) {479 Ok(envelope) => envelope,480 Err(error) => {481 lock_audio_deliveries(&request.audio_deliveries)482 .release_failed_delivery(request.generation, delivery_id);483 emit_audio_event_if_current(484 &app,485 "audio-state",486 &request.connection_generation,487 &request.audio_deliveries,488 request.generation,489 |_| {490 serde_json::json!({491 "generation": request.generation,492 "available": false,493 "error": error494 })495 },496 );497 return;498 }499 };500 if request.audio_channel.send(Response::new(envelope)).is_err() {501 lock_audio_deliveries(&request.audio_deliveries)502 .release_failed_delivery(request.generation, delivery_id);503 emit_audio_event_if_current(504 &app,505 "audio-state",506 &request.connection_generation,507 &request.audio_deliveries,508 request.generation,509 |_| {510 serde_json::json!({511 "generation": request.generation,512 "available": false,513 "error": "Audio webview channel closed"514 })515 },516 );517 return;518 }519 frontend_sent_total = frontend_sent_total.saturating_add(1);520 pending_discontinuity = false;521 }522 if last_stats.elapsed() >= Duration::from_millis(250) {523 if !emit_audio_event_if_current(524 &app,525 "audio-stats",526 &request.connection_generation,527 &request.audio_deliveries,528 request.generation,529 |deliveries| {530 serde_json::json!({531 "generation": request.generation,532 "transport_received_total": transport_received_total,533 "sequence_dropped_total": sequence_dropped_total,534 "frontend_dropped_total": frontend_dropped_total,535 "frontend_sent_total": frontend_sent_total,536 "frontend_queue_depth": deliveries.depth(request.generation).unwrap_or(0),537 "frontend_queue_capacity": AUDIO_DELIVERY_CAPACITY,538 })539 },540 ) {541 return;542 }543 last_stats = Instant::now();544 }545 }546 });547 Ok(())548}549550pub(crate) fn iroh_client_ack_audio(551 state: State<'_, AppState>,552 generation: u64,553 delivery_id: u64,554) -> Result<bool, String> {555 lock_audio_deliveries(&state.audio_deliveries).acknowledge(generation, delivery_id)?;556 Ok(true)557}558559pub(crate) fn cancel_audio_generation(560 generation_counter: &AtomicU64,561 deliveries: &StdMutex<AudioDeliveryState>,562 expected_generation: u64,563) -> Result<bool, String> {564 let replacement_generation = expected_generation565 .checked_add(1)566 .ok_or_else(|| "Audio connection generation overflowed".to_string())?;567 let mut deliveries = lock_audio_deliveries(deliveries);568 if deliveries.generation() != Some(expected_generation)569 || generation_counter.load(Ordering::SeqCst) != expected_generation570 {571 return Ok(false);572 }573 if generation_counter574 .compare_exchange(575 expected_generation,576 replacement_generation,577 Ordering::SeqCst,578 Ordering::SeqCst,579 )580 .is_err()581 {582 return Ok(false);583 }584 if !deliveries.cancel_generation(expected_generation) {585 return Err("Audio delivery generation changed during cancellation".to_string());586 }587 Ok(true)588}589590pub(crate) async fn iroh_client_stop_audio(591 state: State<'_, AppState>,592 expected_generation: u64,593) -> Result<bool, String> {594 if !cancel_audio_generation(595 &state.audio_connection_generation,596 &state.audio_deliveries,597 expected_generation,598 )? {599 return Ok(false);600 }601602 let connection = {603 let mut audio_connection = state.audio_connection.lock().await;604 if audio_connection605 .as_ref()606 .is_some_and(|(generation, _)| *generation == expected_generation)607 {608 audio_connection.take().map(|(_, connection)| connection)609 } else {610 None611 }612 };613 if let Some(connection) = connection {614 connection.close(0_u32.into(), b"audio stopped by client");615 }616 Ok(true)617}618619620mod tests {621 use super::*;622623 fn audio_packet(sequence: u64) -> AudioPacket {624 let payload = vec![0xf8, 0xff, 0xfe];625 AudioPacket::new(626 AudioPacketHeader::opus(627 payload.len(),628 sequence,629 sequence * 20_000,630 sequence as i64 * 20_000,631 AudioFlags::NONE,632 )633 .unwrap(),634 payload,635 )636 .unwrap()637 }638639 640 fn audio_delivery_tokens_are_bounded_unique_and_acknowledged_exactly_once() {641 let mut deliveries = AudioDeliveryState::default();642 deliveries.begin_generation(7).unwrap();643644 assert_eq!(deliveries.reserve(7).unwrap(), Some(1));645 assert_eq!(deliveries.reserve(7).unwrap(), Some(2));646 assert_eq!(deliveries.reserve(7).unwrap(), Some(3));647 assert_eq!(deliveries.depth(7), Some(AUDIO_DELIVERY_CAPACITY));648 assert_eq!(deliveries.reserve(7).unwrap(), None);649650 deliveries.acknowledge(7, 2).unwrap();651 assert!(deliveries.acknowledge(7, 2).is_err());652 assert_eq!(deliveries.reserve(7).unwrap(), Some(4));653 assert!(deliveries.acknowledge(6, 1).is_err());654 assert!(deliveries.acknowledge(7, 99).is_err());655 assert_eq!(deliveries.depth(7), Some(AUDIO_DELIVERY_CAPACITY));656 }657658 659 fn audio_generation_cancellation_rejects_stale_and_clears_current_tokens() {660 let counter = AtomicU64::new(11);661 let deliveries = StdMutex::new(AudioDeliveryState::default());662 lock_audio_deliveries(&deliveries)663 .begin_generation(11)664 .unwrap();665 assert_eq!(666 lock_audio_deliveries(&deliveries).reserve(11).unwrap(),667 Some(1)668 );669670 assert!(!cancel_audio_generation(&counter, &deliveries, 10).unwrap());671 assert_eq!(counter.load(Ordering::SeqCst), 11);672 assert_eq!(lock_audio_deliveries(&deliveries).depth(11), Some(1));673674 assert!(cancel_audio_generation(&counter, &deliveries, 11).unwrap());675 assert_eq!(counter.load(Ordering::SeqCst), 12);676 assert_eq!(lock_audio_deliveries(&deliveries).generation(), None);677 assert!(!cancel_audio_generation(&counter, &deliveries, 11).unwrap());678 }679680 681 fn audio_reorder_window_is_bounded_and_marks_skipped_packets() {682 let mut reorder = AudioReorderBuffer::default();683 let (first, dropped) = reorder.insert(audio_packet(10)).unwrap();684 assert_eq!(first.len(), 1);685 assert_eq!(dropped, 0);686687 assert!(reorder.insert(audio_packet(12)).unwrap().0.is_empty());688 assert!(reorder.insert(audio_packet(13)).unwrap().0.is_empty());689 let (ordered, dropped) = reorder.insert(audio_packet(14)).unwrap();690 assert_eq!(dropped, 1);691 assert_eq!(692 ordered693 .iter()694 .map(|packet| packet.packet.header.sequence)695 .collect::<Vec<_>>(),696 vec![12, 13, 14]697 );698 assert!(ordered[0].discontinuity);699 assert!(!ordered[1].discontinuity);700 assert!(reorder.packets.len() <= AudioReorderBuffer::CAPACITY);701702 assert!(reorder.insert(audio_packet(16)).unwrap().0.is_empty());703 let (ordered, dropped) = reorder.insert(audio_packet(15)).unwrap();704 assert_eq!(dropped, 0);705 assert_eq!(706 ordered707 .iter()708 .map(|packet| packet.packet.header.sequence)709 .collect::<Vec<_>>(),710 vec![15, 16]711 );712 assert!(reorder.insert(audio_packet(16)).unwrap().0.is_empty());713 }714715 716 fn audio_channel_envelope_is_protocol_strict_and_can_force_discontinuity() {717 let encoded = encode_audio_channel_packet(9, 42, audio_packet(7), true).unwrap();718 assert_eq!(&encoded[0..4], b"SGAC");719 assert_eq!(&encoded[4..6], &1_u16.to_be_bytes());720 assert_eq!(&encoded[6..8], &24_u16.to_be_bytes());721 assert_eq!(&encoded[8..16], &9_u64.to_be_bytes());722 assert_eq!(&encoded[16..24], &42_u64.to_be_bytes());723 let decoded = AudioPacket::decode_datagram(&encoded[AUDIO_CHANNEL_HEADER_LEN..]).unwrap();724 assert_eq!(decoded.header.sequence, 7);725 assert!(decoded.header.flags.contains(AudioFlags::DISCONTINUITY));726 assert_eq!(decoded.payload, vec![0xf8, 0xff, 0xfe]);727 }728}