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};1516// Three 20 ms Opus packets cap the Rust→webview handoff at 60 ms. The17// AudioWorklet owns a separate fixed ring and never feeds back into transport.18const AUDIO_CHANNEL_MAGIC: [u8; 4] = *b"SGAC";19const AUDIO_CHANNEL_VERSION: u16 = 1;20const AUDIO_CHANNEL_HEADER_LEN: usize = 24;2122pub(crate) fn lock_audio_deliveries(23 deliveries: &StdMutex<AudioDeliveryState>,24) -> StdMutexGuard<'_, AudioDeliveryState> {25 deliveries26 .lock()27 .unwrap_or_else(std::sync::PoisonError::into_inner)28}2930pub(crate) fn next_audio_generation(counter: &AtomicU64) -> Result<u64, String> {31 counter32 .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |current| {33 current.checked_add(1)34 })35 .map(|previous| previous + 1)36 .map_err(|_| "Audio connection generation overflowed".to_string())37}3839fn audio_generation_is_current(counter: &AtomicU64, generation: u64) -> bool {40 counter.load(Ordering::SeqCst) == generation41}4243fn emit_audio_event_if_current(44 app: &AppHandle,45 event: &str,46 generation_counter: &AtomicU64,47 deliveries: &StdMutex<AudioDeliveryState>,48 generation: u64,49 payload: impl FnOnce(&AudioDeliveryState) -> serde_json::Value,50) -> bool {51 let deliveries = lock_audio_deliveries(deliveries);52 if generation_counter.load(Ordering::SeqCst) != generation53 || deliveries.generation() != Some(generation)54 {55 return false;56 }57 let _ = app.emit(event, payload(&deliveries));58 true59}606162struct AudioReorderBuffer {63 expected_sequence: Option<u64>,64 packets: BTreeMap<u64, AudioPacket>,65}666768struct OrderedAudioPacket {69 packet: AudioPacket,70 discontinuity: bool,71}7273impl AudioReorderBuffer {74 const CAPACITY: usize = 3;7576 fn insert(&mut self, packet: AudioPacket) -> Result<(Vec<OrderedAudioPacket>, u64), String> {77 let sequence = packet.header.sequence;78 let expected = self.expected_sequence.get_or_insert(sequence);79 if sequence < *expected || self.packets.contains_key(&sequence) {80 return Ok((Vec::new(), 0));81 }82 self.packets.insert(sequence, packet);8384 let mut dropped = 0_u64;85 let mut discontinuity = false;86 if !self.packets.contains_key(expected) && self.packets.len() >= Self::CAPACITY {87 let next = *self88 .packets89 .first_key_value()90 .expect("capacity check guarantees one packet")91 .0;92 dropped = next.saturating_sub(*expected);93 *expected = next;94 discontinuity = true;95 }9697 let mut ordered = Vec::with_capacity(Self::CAPACITY);98 while let Some(packet) = self.packets.remove(expected) {99 ordered.push(OrderedAudioPacket {100 packet,101 discontinuity,102 });103 discontinuity = false;104 *expected = expected105 .checked_add(1)106 .ok_or_else(|| "Audio sequence overflowed".to_string())?;107 }108 Ok((ordered, dropped))109 }110}111112fn encode_audio_channel_packet(113 generation: u64,114 delivery_id: u64,115 packet: AudioPacket,116 force_discontinuity: bool,117) -> Result<Vec<u8>, String> {118 let protocol_packet = if force_discontinuity {119 let header = AudioPacketHeader::opus(120 packet.payload.len(),121 packet.header.sequence,122 packet.header.capture_timestamp_us,123 packet.header.pts_us,124 AudioFlags::DISCONTINUITY,125 )126 .map_err(|error| error.to_string())?;127 AudioPacket::new(header, packet.payload).map_err(|error| error.to_string())?128 } else {129 packet130 };131 let datagram = protocol_packet132 .encode_datagram()133 .map_err(|error| error.to_string())?;134 let mut envelope = Vec::with_capacity(AUDIO_CHANNEL_HEADER_LEN + datagram.len());135 envelope.extend_from_slice(&AUDIO_CHANNEL_MAGIC);136 envelope.extend_from_slice(&AUDIO_CHANNEL_VERSION.to_be_bytes());137 envelope.extend_from_slice(&(AUDIO_CHANNEL_HEADER_LEN as u16).to_be_bytes());138 envelope.extend_from_slice(&generation.to_be_bytes());139 envelope.extend_from_slice(&delivery_id.to_be_bytes());140 envelope.extend_from_slice(&datagram);141 debug_assert_eq!(envelope.len(), AUDIO_CHANNEL_HEADER_LEN + datagram.len());142 Ok(envelope)143}144145pub(crate) struct AudioStartRequest {146 pub(crate) address: iroh::EndpointAddr,147 pub(crate) handshake_nonce: [u8; 16],148 pub(crate) media_session_id: u64,149 pub(crate) audio_supported: bool,150 pub(crate) audio_channel: Channel<Response>,151 pub(crate) audio_deliveries: Arc<StdMutex<AudioDeliveryState>>,152 pub(crate) connection_generation: Arc<AtomicU64>,153 pub(crate) generation: u64,154}155156pub(crate) async fn try_start_audio(157 app: AppHandle,158 endpoint: &Endpoint,159 request: AudioStartRequest,160) -> Result<iroh::endpoint::Connection, String> {161 if !request.audio_supported {162 return Err("WebCodecs Opus AudioDecoder is unavailable".to_string());163 }164 let media_session_id = request.media_session_id;165 let audio_connection = tokio::time::timeout(166 Duration::from_secs(3),167 endpoint.connect(request.address, AUDIO_ALPN_V1),168 )169 .await170 .map_err(|_| "Timed out connecting optional audio".to_string())?171 .map_err(|error| format!("Audio connection unavailable: {error}"))?;172 let (mut send, mut recv) = audio_connection173 .open_bi()174 .await175 .map_err(|error| format!("Failed to open audio handshake: {error}"))?;176 let negotiation = negotiate_v1(177 &mut send,178 &mut recv,179 request.handshake_nonce,180 vec![Capability::AudioOpus],181 Some(Capability::AudioOpus),182 "audio",183 None,184 )185 .await?;186 if negotiation.session_id != media_session_id {187 audio_connection.close(1_u32.into(), b"audio session mismatch");188 return Err("Host returned mismatched media and audio sessions".to_string());189 }190 send.finish()191 .map_err(|error| format!("Failed to finish audio handshake: {error}"))?;192193 let task_audio_connection = audio_connection.clone();194 tokio::spawn(async move {195 let audio_channel = request.audio_channel;196 let audio_deliveries = request.audio_deliveries;197 let connection_generation = request.connection_generation;198 let generation = request.generation;199 let mut reorder = AudioReorderBuffer::default();200 let mut transport_received_total = 0_u64;201 let mut sequence_dropped_total = 0_u64;202 let mut frontend_dropped_total = 0_u64;203 let mut frontend_sent_total = 0_u64;204 let mut pending_discontinuity = false;205 let mut last_stats = Instant::now();206207 loop {208 let datagram = match task_audio_connection.read_datagram().await {209 Ok(datagram) => datagram,210 Err(error) => {211 emit_audio_event_if_current(212 &app,213 "audio-state",214 &connection_generation,215 &audio_deliveries,216 generation,217 |_| {218 serde_json::json!({219 "generation": generation,220 "available": false,221 "error": format!("Audio connection ended: {error}")222 })223 },224 );225 break;226 }227 };228 if !audio_generation_is_current(&connection_generation, generation) {229 break;230 }231 let packet = match AudioPacket::decode_datagram(&datagram) {232 Ok(packet) => packet,233 Err(error) => {234 emit_audio_event_if_current(235 &app,236 "audio-state",237 &connection_generation,238 &audio_deliveries,239 generation,240 |_| {241 serde_json::json!({242 "generation": generation,243 "available": false,244 "error": format!("Invalid audio packet: {error}")245 })246 },247 );248 break;249 }250 };251 transport_received_total = transport_received_total.saturating_add(1);252 let (packets, dropped) = match reorder.insert(packet) {253 Ok(result) => result,254 Err(error) => {255 emit_audio_event_if_current(256 &app,257 "audio-state",258 &connection_generation,259 &audio_deliveries,260 generation,261 |_| {262 serde_json::json!({263 "generation": generation,264 "available": false,265 "error": error266 })267 },268 );269 break;270 }271 };272 sequence_dropped_total = sequence_dropped_total.saturating_add(dropped);273 if dropped > 0 {274 pending_discontinuity = true;275 }276277 for ordered in packets {278 let delivery_id = match lock_audio_deliveries(&audio_deliveries).reserve(generation)279 {280 Ok(Some(delivery_id)) => delivery_id,281 Ok(None) => {282 frontend_dropped_total = frontend_dropped_total.saturating_add(1);283 pending_discontinuity = true;284 continue;285 }286 Err(_) => return,287 };288 let envelope = match encode_audio_channel_packet(289 generation,290 delivery_id,291 ordered.packet,292 ordered.discontinuity || pending_discontinuity,293 ) {294 Ok(envelope) => envelope,295 Err(error) => {296 lock_audio_deliveries(&audio_deliveries)297 .release_failed_delivery(generation, delivery_id);298 emit_audio_event_if_current(299 &app,300 "audio-state",301 &connection_generation,302 &audio_deliveries,303 generation,304 |_| {305 serde_json::json!({306 "generation": generation,307 "available": false,308 "error": error309 })310 },311 );312 return;313 }314 };315 if !audio_generation_is_current(&connection_generation, generation) {316 lock_audio_deliveries(&audio_deliveries)317 .release_failed_delivery(generation, delivery_id);318 return;319 }320 if audio_channel.send(Response::new(envelope)).is_err() {321 lock_audio_deliveries(&audio_deliveries)322 .release_failed_delivery(generation, delivery_id);323 emit_audio_event_if_current(324 &app,325 "audio-state",326 &connection_generation,327 &audio_deliveries,328 generation,329 |_| {330 serde_json::json!({331 "generation": generation,332 "available": false,333 "error": "Audio webview channel closed"334 })335 },336 );337 return;338 }339 frontend_sent_total = frontend_sent_total.saturating_add(1);340 pending_discontinuity = false;341 }342343 if last_stats.elapsed() >= Duration::from_millis(250) {344 if !emit_audio_event_if_current(345 &app,346 "audio-stats",347 &connection_generation,348 &audio_deliveries,349 generation,350 |deliveries| {351 serde_json::json!({352 "generation": generation,353 "transport_received_total": transport_received_total,354 "sequence_dropped_total": sequence_dropped_total,355 "frontend_dropped_total": frontend_dropped_total,356 "frontend_sent_total": frontend_sent_total,357 "frontend_queue_depth": deliveries.depth(generation).unwrap_or(0),358 "frontend_queue_capacity": AUDIO_DELIVERY_CAPACITY,359 })360 },361 ) {362 return;363 }364 last_stats = Instant::now();365 }366 }367 });368 Ok(audio_connection)369}370371pub(crate) fn iroh_client_ack_audio(372 state: State<'_, AppState>,373 generation: u64,374 delivery_id: u64,375) -> Result<bool, String> {376 lock_audio_deliveries(&state.audio_deliveries).acknowledge(generation, delivery_id)?;377 Ok(true)378}379380pub(crate) fn cancel_audio_generation(381 generation_counter: &AtomicU64,382 deliveries: &StdMutex<AudioDeliveryState>,383 expected_generation: u64,384) -> Result<bool, String> {385 let replacement_generation = expected_generation386 .checked_add(1)387 .ok_or_else(|| "Audio connection generation overflowed".to_string())?;388 let mut deliveries = lock_audio_deliveries(deliveries);389 if deliveries.generation() != Some(expected_generation)390 || generation_counter.load(Ordering::SeqCst) != expected_generation391 {392 return Ok(false);393 }394 if generation_counter395 .compare_exchange(396 expected_generation,397 replacement_generation,398 Ordering::SeqCst,399 Ordering::SeqCst,400 )401 .is_err()402 {403 return Ok(false);404 }405 if !deliveries.cancel_generation(expected_generation) {406 return Err("Audio delivery generation changed during cancellation".to_string());407 }408 Ok(true)409}410411pub(crate) async fn iroh_client_stop_audio(412 state: State<'_, AppState>,413 expected_generation: u64,414) -> Result<bool, String> {415 if !cancel_audio_generation(416 &state.audio_connection_generation,417 &state.audio_deliveries,418 expected_generation,419 )? {420 return Ok(false);421 }422423 let connection = {424 let mut audio_connection = state.audio_connection.lock().await;425 if audio_connection426 .as_ref()427 .is_some_and(|(generation, _)| *generation == expected_generation)428 {429 audio_connection.take().map(|(_, connection)| connection)430 } else {431 None432 }433 };434 if let Some(connection) = connection {435 connection.close(0_u32.into(), b"audio stopped by client");436 }437 Ok(true)438}439440441mod tests {442 use super::*;443444 fn audio_packet(sequence: u64) -> AudioPacket {445 let payload = vec![0xf8, 0xff, 0xfe];446 AudioPacket::new(447 AudioPacketHeader::opus(448 payload.len(),449 sequence,450 sequence * 20_000,451 sequence as i64 * 20_000,452 AudioFlags::NONE,453 )454 .unwrap(),455 payload,456 )457 .unwrap()458 }459460 461 fn audio_delivery_tokens_are_bounded_unique_and_acknowledged_exactly_once() {462 let mut deliveries = AudioDeliveryState::default();463 deliveries.begin_generation(7).unwrap();464465 assert_eq!(deliveries.reserve(7).unwrap(), Some(1));466 assert_eq!(deliveries.reserve(7).unwrap(), Some(2));467 assert_eq!(deliveries.reserve(7).unwrap(), Some(3));468 assert_eq!(deliveries.depth(7), Some(AUDIO_DELIVERY_CAPACITY));469 assert_eq!(deliveries.reserve(7).unwrap(), None);470471 deliveries.acknowledge(7, 2).unwrap();472 assert!(deliveries.acknowledge(7, 2).is_err());473 assert_eq!(deliveries.reserve(7).unwrap(), Some(4));474 assert!(deliveries.acknowledge(6, 1).is_err());475 assert!(deliveries.acknowledge(7, 99).is_err());476 assert_eq!(deliveries.depth(7), Some(AUDIO_DELIVERY_CAPACITY));477 }478479 480 fn audio_generation_cancellation_rejects_stale_and_clears_current_tokens() {481 let counter = AtomicU64::new(11);482 let deliveries = StdMutex::new(AudioDeliveryState::default());483 lock_audio_deliveries(&deliveries)484 .begin_generation(11)485 .unwrap();486 assert_eq!(487 lock_audio_deliveries(&deliveries).reserve(11).unwrap(),488 Some(1)489 );490491 assert!(!cancel_audio_generation(&counter, &deliveries, 10).unwrap());492 assert_eq!(counter.load(Ordering::SeqCst), 11);493 assert_eq!(lock_audio_deliveries(&deliveries).depth(11), Some(1));494495 assert!(cancel_audio_generation(&counter, &deliveries, 11).unwrap());496 assert_eq!(counter.load(Ordering::SeqCst), 12);497 assert_eq!(lock_audio_deliveries(&deliveries).generation(), None);498 assert!(!cancel_audio_generation(&counter, &deliveries, 11).unwrap());499 }500501 502 fn audio_reorder_window_is_bounded_and_marks_skipped_packets() {503 let mut reorder = AudioReorderBuffer::default();504 let (first, dropped) = reorder.insert(audio_packet(10)).unwrap();505 assert_eq!(first.len(), 1);506 assert_eq!(dropped, 0);507508 assert!(reorder.insert(audio_packet(12)).unwrap().0.is_empty());509 assert!(reorder.insert(audio_packet(13)).unwrap().0.is_empty());510 let (ordered, dropped) = reorder.insert(audio_packet(14)).unwrap();511 assert_eq!(dropped, 1);512 assert_eq!(513 ordered514 .iter()515 .map(|packet| packet.packet.header.sequence)516 .collect::<Vec<_>>(),517 vec![12, 13, 14]518 );519 assert!(ordered[0].discontinuity);520 assert!(!ordered[1].discontinuity);521 assert!(reorder.packets.len() <= AudioReorderBuffer::CAPACITY);522523 assert!(reorder.insert(audio_packet(16)).unwrap().0.is_empty());524 let (ordered, dropped) = reorder.insert(audio_packet(15)).unwrap();525 assert_eq!(dropped, 0);526 assert_eq!(527 ordered528 .iter()529 .map(|packet| packet.packet.header.sequence)530 .collect::<Vec<_>>(),531 vec![15, 16]532 );533 assert!(reorder.insert(audio_packet(16)).unwrap().0.is_empty());534 }535536 537 fn audio_channel_envelope_is_protocol_strict_and_can_force_discontinuity() {538 let encoded = encode_audio_channel_packet(9, 42, audio_packet(7), true).unwrap();539 assert_eq!(&encoded[0..4], b"SGAC");540 assert_eq!(&encoded[4..6], &1_u16.to_be_bytes());541 assert_eq!(&encoded[6..8], &24_u16.to_be_bytes());542 assert_eq!(&encoded[8..16], &9_u64.to_be_bytes());543 assert_eq!(&encoded[16..24], &42_u64.to_be_bytes());544 let decoded = AudioPacket::decode_datagram(&encoded[AUDIO_CHANNEL_HEADER_LEN..]).unwrap();545 assert_eq!(decoded.header.sequence, 7);546 assert!(decoded.header.flags.contains(AudioFlags::DISCONTINUITY));547 assert_eq!(decoded.payload, vec![0xf8, 0xff, 0xfe]);548 }549}