- commit
- c1bdfb51e7fc89e111c6f24d138c0002e076ad03
- parent
- 4fad948dc6781ce889b51c9d0fb66d9b126e1f9b
- author
- Dami <dami.etoile@gmail.com>
| M | src-tauri/ | 548 | +----------------------------- |
| A | src-tauri/ | 551 | ++++++++++++++++++++++++++++++ |
| M | src-tauri/ | 1 | + |
@@ -1,9 +1,12 @@
11 use super::auth::derive_iroh_secret_from_key;22 use super::enrollment::{connection_enrollment, mark_invitation_redeemed};33 use super::state::{4- AUDIO_DELIVERY_CAPACITY, AccumulatedMediaFeedback, AppState, AudioDeliveryState,5- CLIENT_INPUT_QUEUE_CAPACITY, FRAME_ALPN, INPUT_ALPN, MediaFeedbackSender,6- development_direct_node_available,4+ AccumulatedMediaFeedback, AppState, CLIENT_INPUT_QUEUE_CAPACITY, FRAME_ALPN, INPUT_ALPN,5+ MediaFeedbackSender, development_direct_node_available,6+};7+use crate::media::audio_delivery::{8+ self, AudioStartRequest, cancel_audio_generation, lock_audio_deliveries, next_audio_generation,9+ try_start_audio,710 };811 pub use crate::media::frame_channel::FramePayload;912 use crate::media::frame_channel::{@@ -30,16 +33,15 @@ use serde::{Deserialize, Serialize};
3033 3134 use sigil_protocol::MOQ_VIDEO_H264_TRACK;3235 use sigil_protocol::{33- AUDIO_ALPN_V1, AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1, AdaptiveBitrateStateV1,34- AudioFlags, AudioPacket, AudioPacketHeader, Capability, FrameFlags, INPUT_ALPN_V1, InputEvent,35- InvitationGrants, KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3,36- MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MAX_MEDIA_OBJECT_ID_V3, MEDIA_FEEDBACK_ALPN_V1,37- MediaCodec, MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1, MediaFrame,38- MediaObjectV3, PointerPosition, PointerSurfaceDimensions, ProtocolError,39- RELATIVE_POINTER_DELTA_MAX, RELATIVE_POINTER_DELTA_MIN, decode_media_frame_object,40- media_moq_broadcast_name, read_adaptive_bitrate_decision_v1, read_input_ack, read_media_frame,41- read_media_object, read_media_object_v3, write_input_event, write_media_control_request_v3,42- write_media_feedback_report_v1,36+ AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1, AdaptiveBitrateStateV1, Capability,37+ FrameFlags, INPUT_ALPN_V1, InputEvent, InvitationGrants, KeyframeRequestReasonV3,38+ MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MAX_MEDIA_OBJECT_ID_V3,39+ MEDIA_FEEDBACK_ALPN_V1, MediaCodec, MediaControlRequestV3, MediaFeedbackFlags,40+ MediaFeedbackReportV1, MediaFrame, MediaObjectV3, PointerPosition, PointerSurfaceDimensions,41+ ProtocolError, RELATIVE_POINTER_DELTA_MAX, RELATIVE_POINTER_DELTA_MIN,42+ decode_media_frame_object, media_moq_broadcast_name, read_adaptive_bitrate_decision_v1,43+ read_input_ack, read_media_frame, read_media_object, read_media_object_v3, write_input_event,44+ write_media_control_request_v3, write_media_feedback_report_v1,4345 };4446 use std::collections::BTreeMap;4547 use std::io::Cursor;@@ -65,16 +67,11 @@ const CLIENT_MOQ_OBJECT_READ_TIMEOUT: Duration =
6567 // Sigil's external encoder can take 500 ms to reach its next configured IDR.6668 // Allow another 500 ms for a relay path to deliver the superseding group.6769 const CLIENT_MOQ_GROUP_RECOVERY_TIMEOUT: Duration = Duration::from_secs(1);68-// Three 20 ms Opus packets cap the Rust→webview handoff at 60 ms. The69-// AudioWorklet owns a separate fixed ring and never feeds back into transport.7070 const CLIENT_FRAME_STATS_INTERVAL: Duration = Duration::from_millis(250);7171 const CLIENT_ENDPOINT_CLOSE_TIMEOUT: Duration = Duration::from_secs(2);7272 const CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);7373 const CLIENT_MEDIA_FEEDBACK_IO_TIMEOUT: Duration = Duration::from_secs(5);7474 const CLIENT_ADAPTIVE_DECISION_DELIVERY_INTERVAL: Duration = Duration::from_secs(1);75-const AUDIO_CHANNEL_MAGIC: [u8; 4] = *b"SGAC";76-const AUDIO_CHANNEL_VERSION: u16 = 1;77-const AUDIO_CHANNEL_HEADER_LEN: usize = 24;7875 7976 fn lock_network_diagnostics(8077 diagnostics: &StdMutex<NetworkSessionDiagnostics>,@@ -172,45 +169,6 @@ async fn retire_upstream_moq_generation(
172169 true173170 }174171 175-fn lock_audio_deliveries(176- deliveries: &StdMutex<AudioDeliveryState>,177-) -> StdMutexGuard<'_, AudioDeliveryState> {178- deliveries179- .lock()180- .unwrap_or_else(std::sync::PoisonError::into_inner)181-}182-183-fn next_audio_generation(counter: &AtomicU64) -> Result<u64, String> {184- counter185- .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |current| {186- current.checked_add(1)187- })188- .map(|previous| previous + 1)189- .map_err(|_| "Audio connection generation overflowed".to_string())190-}191-192-fn audio_generation_is_current(counter: &AtomicU64, generation: u64) -> bool {193- counter.load(Ordering::SeqCst) == generation194-}195-196-fn emit_audio_event_if_current(197- app: &AppHandle,198- event: &str,199- generation_counter: &AtomicU64,200- deliveries: &StdMutex<AudioDeliveryState>,201- generation: u64,202- payload: impl FnOnce(&AudioDeliveryState) -> serde_json::Value,203-) -> bool {204- let deliveries = lock_audio_deliveries(deliveries);205- if generation_counter.load(Ordering::SeqCst) != generation206- || deliveries.generation() != Some(generation)207- {208- return false;209- }210- let _ = app.emit(event, payload(&deliveries));211- true212-}213-214172 struct ClientConnectGuard {215173 active: Arc<AtomicBool>,216174 committed: bool,@@ -1748,318 +1706,6 @@ fn observe_input_ack_if_negotiated(
17481706 lock_network_diagnostics(diagnostics).observe_input_ack(sequence, now)17491707 }17501708 1751-1752-struct AudioReorderBuffer {1753- expected_sequence: Option<u64>,1754- packets: BTreeMap<u64, AudioPacket>,1755-}1756-1757-1758-struct OrderedAudioPacket {1759- packet: AudioPacket,1760- discontinuity: bool,1761-}1762-1763-impl AudioReorderBuffer {1764- const CAPACITY: usize = 3;1765-1766- fn insert(&mut self, packet: AudioPacket) -> Result<(Vec<OrderedAudioPacket>, u64), String> {1767- let sequence = packet.header.sequence;1768- let expected = self.expected_sequence.get_or_insert(sequence);1769- if sequence < *expected || self.packets.contains_key(&sequence) {1770- return Ok((Vec::new(), 0));1771- }1772- self.packets.insert(sequence, packet);1773-1774- let mut dropped = 0_u64;1775- let mut discontinuity = false;1776- if !self.packets.contains_key(expected) && self.packets.len() >= Self::CAPACITY {1777- let next = *self1778- .packets1779- .first_key_value()1780- .expect("capacity check guarantees one packet")1781- .0;1782- dropped = next.saturating_sub(*expected);1783- *expected = next;1784- discontinuity = true;1785- }1786-1787- let mut ordered = Vec::with_capacity(Self::CAPACITY);1788- while let Some(packet) = self.packets.remove(expected) {1789- ordered.push(OrderedAudioPacket {1790- packet,1791- discontinuity,1792- });1793- discontinuity = false;1794- *expected = expected1795- .checked_add(1)1796- .ok_or_else(|| "Audio sequence overflowed".to_string())?;1797- }1798- Ok((ordered, dropped))1799- }1800-}1801-1802-fn encode_audio_channel_packet(1803- generation: u64,1804- delivery_id: u64,1805- packet: AudioPacket,1806- force_discontinuity: bool,1807-) -> Result<Vec<u8>, String> {1808- let protocol_packet = if force_discontinuity {1809- let header = AudioPacketHeader::opus(1810- packet.payload.len(),1811- packet.header.sequence,1812- packet.header.capture_timestamp_us,1813- packet.header.pts_us,1814- AudioFlags::DISCONTINUITY,1815- )1816- .map_err(|error| error.to_string())?;1817- AudioPacket::new(header, packet.payload).map_err(|error| error.to_string())?1818- } else {1819- packet1820- };1821- let datagram = protocol_packet1822- .encode_datagram()1823- .map_err(|error| error.to_string())?;1824- let mut envelope = Vec::with_capacity(AUDIO_CHANNEL_HEADER_LEN + datagram.len());1825- envelope.extend_from_slice(&AUDIO_CHANNEL_MAGIC);1826- envelope.extend_from_slice(&AUDIO_CHANNEL_VERSION.to_be_bytes());1827- envelope.extend_from_slice(&(AUDIO_CHANNEL_HEADER_LEN as u16).to_be_bytes());1828- envelope.extend_from_slice(&generation.to_be_bytes());1829- envelope.extend_from_slice(&delivery_id.to_be_bytes());1830- envelope.extend_from_slice(&datagram);1831- debug_assert_eq!(envelope.len(), AUDIO_CHANNEL_HEADER_LEN + datagram.len());1832- Ok(envelope)1833-}1834-1835-struct AudioStartRequest {1836- address: iroh::EndpointAddr,1837- handshake_nonce: [u8; 16],1838- media_session_id: Option<u64>,1839- audio_supported: bool,1840- audio_channel: Channel<Response>,1841- audio_deliveries: Arc<StdMutex<AudioDeliveryState>>,1842- connection_generation: Arc<AtomicU64>,1843- generation: u64,1844-}1845-1846-async fn try_start_audio(1847- app: AppHandle,1848- endpoint: &Endpoint,1849- request: AudioStartRequest,1850-) -> Result<iroh::endpoint::Connection, String> {1851- if !request.audio_supported {1852- return Err("WebCodecs Opus AudioDecoder is unavailable".to_string());1853- }1854- let media_session_id = request1855- .media_session_id1856- .ok_or_else(|| "The connected host protocol does not negotiate audio".to_string())?;1857- let audio_connection = tokio::time::timeout(1858- Duration::from_secs(3),1859- endpoint.connect(request.address, AUDIO_ALPN_V1),1860- )1861- .await1862- .map_err(|_| "Timed out connecting optional audio".to_string())?1863- .map_err(|error| format!("Audio connection unavailable: {error}"))?;1864- let (mut send, mut recv) = audio_connection1865- .open_bi()1866- .await1867- .map_err(|error| format!("Failed to open audio handshake: {error}"))?;1868- let negotiation = negotiate_v1(1869- &mut send,1870- &mut recv,1871- request.handshake_nonce,1872- vec![Capability::AudioOpus],1873- Some(Capability::AudioOpus),1874- "audio",1875- None,1876- )1877- .await?;1878- if negotiation.session_id != media_session_id {1879- audio_connection.close(1_u32.into(), b"audio session mismatch");1880- return Err("Host returned mismatched media and audio sessions".to_string());1881- }1882- send.finish()1883- .map_err(|error| format!("Failed to finish audio handshake: {error}"))?;1884-1885- let task_audio_connection = audio_connection.clone();1886- tokio::spawn(async move {1887- let audio_channel = request.audio_channel;1888- let audio_deliveries = request.audio_deliveries;1889- let connection_generation = request.connection_generation;1890- let generation = request.generation;1891- let mut reorder = AudioReorderBuffer::default();1892- let mut transport_received_total = 0_u64;1893- let mut sequence_dropped_total = 0_u64;1894- let mut frontend_dropped_total = 0_u64;1895- let mut frontend_sent_total = 0_u64;1896- let mut pending_discontinuity = false;1897- let mut last_stats = Instant::now();1898-1899- loop {1900- let datagram = match task_audio_connection.read_datagram().await {1901- Ok(datagram) => datagram,1902- Err(error) => {1903- emit_audio_event_if_current(1904- &app,1905- "audio-state",1906- &connection_generation,1907- &audio_deliveries,1908- generation,1909- |_| {1910- serde_json::json!({1911- "generation": generation,1912- "available": false,1913- "error": format!("Audio connection ended: {error}")1914- })1915- },1916- );1917- break;1918- }1919- };1920- if !audio_generation_is_current(&connection_generation, generation) {1921- break;1922- }1923- let packet = match AudioPacket::decode_datagram(&datagram) {1924- Ok(packet) => packet,1925- Err(error) => {1926- emit_audio_event_if_current(1927- &app,1928- "audio-state",1929- &connection_generation,1930- &audio_deliveries,1931- generation,1932- |_| {1933- serde_json::json!({1934- "generation": generation,1935- "available": false,1936- "error": format!("Invalid audio packet: {error}")1937- })1938- },1939- );1940- break;1941- }1942- };1943- transport_received_total = transport_received_total.saturating_add(1);1944- let (packets, dropped) = match reorder.insert(packet) {1945- Ok(result) => result,1946- Err(error) => {1947- emit_audio_event_if_current(1948- &app,1949- "audio-state",1950- &connection_generation,1951- &audio_deliveries,1952- generation,1953- |_| {1954- serde_json::json!({1955- "generation": generation,1956- "available": false,1957- "error": error1958- })1959- },1960- );1961- break;1962- }1963- };1964- sequence_dropped_total = sequence_dropped_total.saturating_add(dropped);1965- if dropped > 0 {1966- pending_discontinuity = true;1967- }1968-1969- for ordered in packets {1970- let delivery_id = match lock_audio_deliveries(&audio_deliveries).reserve(generation)1971- {1972- Ok(Some(delivery_id)) => delivery_id,1973- Ok(None) => {1974- frontend_dropped_total = frontend_dropped_total.saturating_add(1);1975- pending_discontinuity = true;1976- continue;1977- }1978- Err(_) => return,1979- };1980- let envelope = match encode_audio_channel_packet(1981- generation,1982- delivery_id,1983- ordered.packet,1984- ordered.discontinuity || pending_discontinuity,1985- ) {1986- Ok(envelope) => envelope,1987- Err(error) => {1988- lock_audio_deliveries(&audio_deliveries)1989- .release_failed_delivery(generation, delivery_id);1990- emit_audio_event_if_current(1991- &app,1992- "audio-state",1993- &connection_generation,1994- &audio_deliveries,1995- generation,1996- |_| {1997- serde_json::json!({1998- "generation": generation,1999- "available": false,2000- "error": error2001- })2002- },2003- );2004- return;2005- }2006- };2007- if !audio_generation_is_current(&connection_generation, generation) {2008- lock_audio_deliveries(&audio_deliveries)2009- .release_failed_delivery(generation, delivery_id);2010- return;2011- }2012- if audio_channel.send(Response::new(envelope)).is_err() {2013- lock_audio_deliveries(&audio_deliveries)2014- .release_failed_delivery(generation, delivery_id);2015- emit_audio_event_if_current(2016- &app,2017- "audio-state",2018- &connection_generation,2019- &audio_deliveries,2020- generation,2021- |_| {2022- serde_json::json!({2023- "generation": generation,2024- "available": false,2025- "error": "Audio webview channel closed"2026- })2027- },2028- );2029- return;2030- }2031- frontend_sent_total = frontend_sent_total.saturating_add(1);2032- pending_discontinuity = false;2033- }2034-2035- if last_stats.elapsed() >= Duration::from_millis(250) {2036- if !emit_audio_event_if_current(2037- &app,2038- "audio-stats",2039- &connection_generation,2040- &audio_deliveries,2041- generation,2042- |deliveries| {2043- serde_json::json!({2044- "generation": generation,2045- "transport_received_total": transport_received_total,2046- "sequence_dropped_total": sequence_dropped_total,2047- "frontend_dropped_total": frontend_dropped_total,2048- "frontend_sent_total": frontend_sent_total,2049- "frontend_queue_depth": deliveries.depth(generation).unwrap_or(0),2050- "frontend_queue_capacity": AUDIO_DELIVERY_CAPACITY,2051- })2052- },2053- ) {2054- return;2055- }2056- last_stats = Instant::now();2057- }2058- }2059- });2060- Ok(audio_connection)2061-}2062-20631709 20641710 pub async fn iroh_client_connect(20651711 app: AppHandle,@@ -3701,63 +3347,6 @@ mod tests {
37013347 assert!(parse_keyframe_request_reason("please").is_err());37023348 }37033349 3704- fn audio_packet(sequence: u64) -> AudioPacket {3705- let payload = vec![0xf8, 0xff, 0xfe];3706- AudioPacket::new(3707- AudioPacketHeader::opus(3708- payload.len(),3709- sequence,3710- sequence * 20_000,3711- sequence as i64 * 20_000,3712- AudioFlags::NONE,3713- )3714- .unwrap(),3715- payload,3716- )3717- .unwrap()3718- }3719-3720- 3721- fn audio_delivery_tokens_are_bounded_unique_and_acknowledged_exactly_once() {3722- let mut deliveries = AudioDeliveryState::default();3723- deliveries.begin_generation(7).unwrap();3724-3725- assert_eq!(deliveries.reserve(7).unwrap(), Some(1));3726- assert_eq!(deliveries.reserve(7).unwrap(), Some(2));3727- assert_eq!(deliveries.reserve(7).unwrap(), Some(3));3728- assert_eq!(deliveries.depth(7), Some(AUDIO_DELIVERY_CAPACITY));3729- assert_eq!(deliveries.reserve(7).unwrap(), None);3730-3731- deliveries.acknowledge(7, 2).unwrap();3732- assert!(deliveries.acknowledge(7, 2).is_err());3733- assert_eq!(deliveries.reserve(7).unwrap(), Some(4));3734- assert!(deliveries.acknowledge(6, 1).is_err());3735- assert!(deliveries.acknowledge(7, 99).is_err());3736- assert_eq!(deliveries.depth(7), Some(AUDIO_DELIVERY_CAPACITY));3737- }3738-3739- 3740- fn audio_generation_cancellation_rejects_stale_and_clears_current_tokens() {3741- let counter = AtomicU64::new(11);3742- let deliveries = StdMutex::new(AudioDeliveryState::default());3743- lock_audio_deliveries(&deliveries)3744- .begin_generation(11)3745- .unwrap();3746- assert_eq!(3747- lock_audio_deliveries(&deliveries).reserve(11).unwrap(),3748- Some(1)3749- );3750-3751- assert!(!cancel_audio_generation(&counter, &deliveries, 10).unwrap());3752- assert_eq!(counter.load(Ordering::SeqCst), 11);3753- assert_eq!(lock_audio_deliveries(&deliveries).depth(11), Some(1));3754-3755- assert!(cancel_audio_generation(&counter, &deliveries, 11).unwrap());3756- assert_eq!(counter.load(Ordering::SeqCst), 12);3757- assert_eq!(lock_audio_deliveries(&deliveries).generation(), None);3758- assert!(!cancel_audio_generation(&counter, &deliveries, 11).unwrap());3759- }3760-37613350 37623351 fn media_generations_are_nonzero_monotonic_and_checked_for_overflow() {37633352 let counter = AtomicU64::new(0);@@ -3937,55 +3526,6 @@ mod tests {
39373526 assert_eq!(error, "webview channel closed");39383527 }39393528 3940- 3941- fn audio_reorder_window_is_bounded_and_marks_skipped_packets() {3942- let mut reorder = AudioReorderBuffer::default();3943- let (first, dropped) = reorder.insert(audio_packet(10)).unwrap();3944- assert_eq!(first.len(), 1);3945- assert_eq!(dropped, 0);3946-3947- assert!(reorder.insert(audio_packet(12)).unwrap().0.is_empty());3948- assert!(reorder.insert(audio_packet(13)).unwrap().0.is_empty());3949- let (ordered, dropped) = reorder.insert(audio_packet(14)).unwrap();3950- assert_eq!(dropped, 1);3951- assert_eq!(3952- ordered3953- .iter()3954- .map(|packet| packet.packet.header.sequence)3955- .collect::<Vec<_>>(),3956- vec![12, 13, 14]3957- );3958- assert!(ordered[0].discontinuity);3959- assert!(!ordered[1].discontinuity);3960- assert!(reorder.packets.len() <= AudioReorderBuffer::CAPACITY);3961-3962- assert!(reorder.insert(audio_packet(16)).unwrap().0.is_empty());3963- let (ordered, dropped) = reorder.insert(audio_packet(15)).unwrap();3964- assert_eq!(dropped, 0);3965- assert_eq!(3966- ordered3967- .iter()3968- .map(|packet| packet.packet.header.sequence)3969- .collect::<Vec<_>>(),3970- vec![15, 16]3971- );3972- assert!(reorder.insert(audio_packet(16)).unwrap().0.is_empty());3973- }3974-3975- 3976- fn audio_channel_envelope_is_protocol_strict_and_can_force_discontinuity() {3977- let encoded = encode_audio_channel_packet(9, 42, audio_packet(7), true).unwrap();3978- assert_eq!(&encoded[0..4], b"SGAC");3979- assert_eq!(&encoded[4..6], &1_u16.to_be_bytes());3980- assert_eq!(&encoded[6..8], &24_u16.to_be_bytes());3981- assert_eq!(&encoded[8..16], &9_u64.to_be_bytes());3982- assert_eq!(&encoded[16..24], &42_u64.to_be_bytes());3983- let decoded = AudioPacket::decode_datagram(&encoded[AUDIO_CHANNEL_HEADER_LEN..]).unwrap();3984- assert_eq!(decoded.header.sequence, 7);3985- assert!(decoded.header.flags.contains(AudioFlags::DISCONTINUITY));3986- assert_eq!(decoded.payload, vec![0xf8, 0xff, 0xfe]);3987- }3988-39893529 39903530 fn client_connect_guard_rejects_overlap_and_resets_only_failed_attempts() {39913531 let active = Arc::new(AtomicBool::new(false));@@ -4533,39 +4073,7 @@ pub fn iroh_client_ack_audio(
45334073 generation: u64,45344074 delivery_id: u64,45354075 ) -> Result<bool, String> {4536- lock_audio_deliveries(&state.audio_deliveries).acknowledge(generation, delivery_id)?;4537- Ok(true)4538-}4539-4540-fn cancel_audio_generation(4541- generation_counter: &AtomicU64,4542- deliveries: &StdMutex<AudioDeliveryState>,4543- expected_generation: u64,4544-) -> Result<bool, String> {4545- let replacement_generation = expected_generation4546- .checked_add(1)4547- .ok_or_else(|| "Audio connection generation overflowed".to_string())?;4548- let mut deliveries = lock_audio_deliveries(deliveries);4549- if deliveries.generation() != Some(expected_generation)4550- || generation_counter.load(Ordering::SeqCst) != expected_generation4551- {4552- return Ok(false);4553- }4554- if generation_counter4555- .compare_exchange(4556- expected_generation,4557- replacement_generation,4558- Ordering::SeqCst,4559- Ordering::SeqCst,4560- )4561- .is_err()4562- {4563- return Ok(false);4564- }4565- if !deliveries.cancel_generation(expected_generation) {4566- return Err("Audio delivery generation changed during cancellation".to_string());4567- }4568- Ok(true)4076+ audio_delivery::iroh_client_ack_audio(state, generation, delivery_id)45694077 }45704078 45714079 @@ -4573,29 +4081,7 @@ pub async fn iroh_client_stop_audio(
45734081 state: State<'_, AppState>,45744082 expected_generation: u64,45754083 ) -> Result<bool, String> {4576- if !cancel_audio_generation(4577- &state.audio_connection_generation,4578- &state.audio_deliveries,4579- expected_generation,4580- )? {4581- return Ok(false);4582- }4583-4584- let connection = {4585- let mut audio_connection = state.audio_connection.lock().await;4586- if audio_connection4587- .as_ref()4588- .is_some_and(|(generation, _)| *generation == expected_generation)4589- {4590- audio_connection.take().map(|(_, connection)| connection)4591- } else {4592- None4593- }4594- };4595- if let Some(connection) = connection {4596- connection.close(0_u32.into(), b"audio stopped by client");4597- }4598- Ok(true)4084+ audio_delivery::iroh_client_stop_audio(state, expected_generation).await45994085 }46004086 46014087 @@ -0,0 +1,551 @@
1+use std::collections::BTreeMap;2+use std::sync::atomic::{AtomicU64, Ordering};3+use std::sync::{Arc, Mutex as StdMutex, MutexGuard as StdMutexGuard};4+use std::time::{Duration, Instant};5+6+use iroh::Endpoint;7+use sigil_protocol::{AUDIO_ALPN_V1, AudioFlags, AudioPacket, AudioPacketHeader, Capability};8+use tauri::{9+ AppHandle, Emitter, State,10+ ipc::{Channel, Response},11+};12+13+use super::transport::negotiate_v1;14+use crate::commands::state::{AUDIO_DELIVERY_CAPACITY, AppState, AudioDeliveryState};15+16+// 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.18+const AUDIO_CHANNEL_MAGIC: [u8; 4] = *b"SGAC";19+const AUDIO_CHANNEL_VERSION: u16 = 1;20+const AUDIO_CHANNEL_HEADER_LEN: usize = 24;21+22+pub(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+}29+30+pub(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+}38+39+fn audio_generation_is_current(counter: &AtomicU64, generation: u64) -> bool {40+ counter.load(Ordering::SeqCst) == generation41+}42+43+fn 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+}60+61+62+struct AudioReorderBuffer {63+ expected_sequence: Option<u64>,64+ packets: BTreeMap<u64, AudioPacket>,65+}66+67+68+struct OrderedAudioPacket {69+ packet: AudioPacket,70+ discontinuity: bool,71+}72+73+impl AudioReorderBuffer {74+ const CAPACITY: usize = 3;75+76+ 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);83+84+ 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+ }96+97+ 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+}111+112+fn 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+}144+145+pub(crate) struct AudioStartRequest {146+ pub(crate) address: iroh::EndpointAddr,147+ pub(crate) handshake_nonce: [u8; 16],148+ pub(crate) media_session_id: Option<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+}155+156+pub(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 = request165+ .media_session_id166+ .ok_or_else(|| "The connected host protocol does not negotiate audio".to_string())?;167+ let audio_connection = tokio::time::timeout(168+ Duration::from_secs(3),169+ endpoint.connect(request.address, AUDIO_ALPN_V1),170+ )171+ .await172+ .map_err(|_| "Timed out connecting optional audio".to_string())?173+ .map_err(|error| format!("Audio connection unavailable: {error}"))?;174+ let (mut send, mut recv) = audio_connection175+ .open_bi()176+ .await177+ .map_err(|error| format!("Failed to open audio handshake: {error}"))?;178+ let negotiation = negotiate_v1(179+ &mut send,180+ &mut recv,181+ request.handshake_nonce,182+ vec![Capability::AudioOpus],183+ Some(Capability::AudioOpus),184+ "audio",185+ None,186+ )187+ .await?;188+ if negotiation.session_id != media_session_id {189+ audio_connection.close(1_u32.into(), b"audio session mismatch");190+ return Err("Host returned mismatched media and audio sessions".to_string());191+ }192+ send.finish()193+ .map_err(|error| format!("Failed to finish audio handshake: {error}"))?;194+195+ let task_audio_connection = audio_connection.clone();196+ tokio::spawn(async move {197+ let audio_channel = request.audio_channel;198+ let audio_deliveries = request.audio_deliveries;199+ let connection_generation = request.connection_generation;200+ let generation = request.generation;201+ let mut reorder = AudioReorderBuffer::default();202+ let mut transport_received_total = 0_u64;203+ let mut sequence_dropped_total = 0_u64;204+ let mut frontend_dropped_total = 0_u64;205+ let mut frontend_sent_total = 0_u64;206+ let mut pending_discontinuity = false;207+ let mut last_stats = Instant::now();208+209+ loop {210+ let datagram = match task_audio_connection.read_datagram().await {211+ Ok(datagram) => datagram,212+ Err(error) => {213+ emit_audio_event_if_current(214+ &app,215+ "audio-state",216+ &connection_generation,217+ &audio_deliveries,218+ generation,219+ |_| {220+ serde_json::json!({221+ "generation": generation,222+ "available": false,223+ "error": format!("Audio connection ended: {error}")224+ })225+ },226+ );227+ break;228+ }229+ };230+ if !audio_generation_is_current(&connection_generation, generation) {231+ break;232+ }233+ let packet = match AudioPacket::decode_datagram(&datagram) {234+ Ok(packet) => packet,235+ Err(error) => {236+ emit_audio_event_if_current(237+ &app,238+ "audio-state",239+ &connection_generation,240+ &audio_deliveries,241+ generation,242+ |_| {243+ serde_json::json!({244+ "generation": generation,245+ "available": false,246+ "error": format!("Invalid audio packet: {error}")247+ })248+ },249+ );250+ break;251+ }252+ };253+ transport_received_total = transport_received_total.saturating_add(1);254+ let (packets, dropped) = match reorder.insert(packet) {255+ Ok(result) => result,256+ Err(error) => {257+ emit_audio_event_if_current(258+ &app,259+ "audio-state",260+ &connection_generation,261+ &audio_deliveries,262+ generation,263+ |_| {264+ serde_json::json!({265+ "generation": generation,266+ "available": false,267+ "error": error268+ })269+ },270+ );271+ break;272+ }273+ };274+ sequence_dropped_total = sequence_dropped_total.saturating_add(dropped);275+ if dropped > 0 {276+ pending_discontinuity = true;277+ }278+279+ for ordered in packets {280+ let delivery_id = match lock_audio_deliveries(&audio_deliveries).reserve(generation)281+ {282+ Ok(Some(delivery_id)) => delivery_id,283+ Ok(None) => {284+ frontend_dropped_total = frontend_dropped_total.saturating_add(1);285+ pending_discontinuity = true;286+ continue;287+ }288+ Err(_) => return,289+ };290+ let envelope = match encode_audio_channel_packet(291+ generation,292+ delivery_id,293+ ordered.packet,294+ ordered.discontinuity || pending_discontinuity,295+ ) {296+ Ok(envelope) => envelope,297+ Err(error) => {298+ lock_audio_deliveries(&audio_deliveries)299+ .release_failed_delivery(generation, delivery_id);300+ emit_audio_event_if_current(301+ &app,302+ "audio-state",303+ &connection_generation,304+ &audio_deliveries,305+ generation,306+ |_| {307+ serde_json::json!({308+ "generation": generation,309+ "available": false,310+ "error": error311+ })312+ },313+ );314+ return;315+ }316+ };317+ if !audio_generation_is_current(&connection_generation, generation) {318+ lock_audio_deliveries(&audio_deliveries)319+ .release_failed_delivery(generation, delivery_id);320+ return;321+ }322+ if audio_channel.send(Response::new(envelope)).is_err() {323+ lock_audio_deliveries(&audio_deliveries)324+ .release_failed_delivery(generation, delivery_id);325+ emit_audio_event_if_current(326+ &app,327+ "audio-state",328+ &connection_generation,329+ &audio_deliveries,330+ generation,331+ |_| {332+ serde_json::json!({333+ "generation": generation,334+ "available": false,335+ "error": "Audio webview channel closed"336+ })337+ },338+ );339+ return;340+ }341+ frontend_sent_total = frontend_sent_total.saturating_add(1);342+ pending_discontinuity = false;343+ }344+345+ if last_stats.elapsed() >= Duration::from_millis(250) {346+ if !emit_audio_event_if_current(347+ &app,348+ "audio-stats",349+ &connection_generation,350+ &audio_deliveries,351+ generation,352+ |deliveries| {353+ serde_json::json!({354+ "generation": generation,355+ "transport_received_total": transport_received_total,356+ "sequence_dropped_total": sequence_dropped_total,357+ "frontend_dropped_total": frontend_dropped_total,358+ "frontend_sent_total": frontend_sent_total,359+ "frontend_queue_depth": deliveries.depth(generation).unwrap_or(0),360+ "frontend_queue_capacity": AUDIO_DELIVERY_CAPACITY,361+ })362+ },363+ ) {364+ return;365+ }366+ last_stats = Instant::now();367+ }368+ }369+ });370+ Ok(audio_connection)371+}372+373+pub(crate) fn iroh_client_ack_audio(374+ state: State<'_, AppState>,375+ generation: u64,376+ delivery_id: u64,377+) -> Result<bool, String> {378+ lock_audio_deliveries(&state.audio_deliveries).acknowledge(generation, delivery_id)?;379+ Ok(true)380+}381+382+pub(crate) fn cancel_audio_generation(383+ generation_counter: &AtomicU64,384+ deliveries: &StdMutex<AudioDeliveryState>,385+ expected_generation: u64,386+) -> Result<bool, String> {387+ let replacement_generation = expected_generation388+ .checked_add(1)389+ .ok_or_else(|| "Audio connection generation overflowed".to_string())?;390+ let mut deliveries = lock_audio_deliveries(deliveries);391+ if deliveries.generation() != Some(expected_generation)392+ || generation_counter.load(Ordering::SeqCst) != expected_generation393+ {394+ return Ok(false);395+ }396+ if generation_counter397+ .compare_exchange(398+ expected_generation,399+ replacement_generation,400+ Ordering::SeqCst,401+ Ordering::SeqCst,402+ )403+ .is_err()404+ {405+ return Ok(false);406+ }407+ if !deliveries.cancel_generation(expected_generation) {408+ return Err("Audio delivery generation changed during cancellation".to_string());409+ }410+ Ok(true)411+}412+413+pub(crate) async fn iroh_client_stop_audio(414+ state: State<'_, AppState>,415+ expected_generation: u64,416+) -> Result<bool, String> {417+ if !cancel_audio_generation(418+ &state.audio_connection_generation,419+ &state.audio_deliveries,420+ expected_generation,421+ )? {422+ return Ok(false);423+ }424+425+ let connection = {426+ let mut audio_connection = state.audio_connection.lock().await;427+ if audio_connection428+ .as_ref()429+ .is_some_and(|(generation, _)| *generation == expected_generation)430+ {431+ audio_connection.take().map(|(_, connection)| connection)432+ } else {433+ None434+ }435+ };436+ if let Some(connection) = connection {437+ connection.close(0_u32.into(), b"audio stopped by client");438+ }439+ Ok(true)440+}441+442+443+mod tests {444+ use super::*;445+446+ fn audio_packet(sequence: u64) -> AudioPacket {447+ let payload = vec![0xf8, 0xff, 0xfe];448+ AudioPacket::new(449+ AudioPacketHeader::opus(450+ payload.len(),451+ sequence,452+ sequence * 20_000,453+ sequence as i64 * 20_000,454+ AudioFlags::NONE,455+ )456+ .unwrap(),457+ payload,458+ )459+ .unwrap()460+ }461+462+ 463+ fn audio_delivery_tokens_are_bounded_unique_and_acknowledged_exactly_once() {464+ let mut deliveries = AudioDeliveryState::default();465+ deliveries.begin_generation(7).unwrap();466+467+ assert_eq!(deliveries.reserve(7).unwrap(), Some(1));468+ assert_eq!(deliveries.reserve(7).unwrap(), Some(2));469+ assert_eq!(deliveries.reserve(7).unwrap(), Some(3));470+ assert_eq!(deliveries.depth(7), Some(AUDIO_DELIVERY_CAPACITY));471+ assert_eq!(deliveries.reserve(7).unwrap(), None);472+473+ deliveries.acknowledge(7, 2).unwrap();474+ assert!(deliveries.acknowledge(7, 2).is_err());475+ assert_eq!(deliveries.reserve(7).unwrap(), Some(4));476+ assert!(deliveries.acknowledge(6, 1).is_err());477+ assert!(deliveries.acknowledge(7, 99).is_err());478+ assert_eq!(deliveries.depth(7), Some(AUDIO_DELIVERY_CAPACITY));479+ }480+481+ 482+ fn audio_generation_cancellation_rejects_stale_and_clears_current_tokens() {483+ let counter = AtomicU64::new(11);484+ let deliveries = StdMutex::new(AudioDeliveryState::default());485+ lock_audio_deliveries(&deliveries)486+ .begin_generation(11)487+ .unwrap();488+ assert_eq!(489+ lock_audio_deliveries(&deliveries).reserve(11).unwrap(),490+ Some(1)491+ );492+493+ assert!(!cancel_audio_generation(&counter, &deliveries, 10).unwrap());494+ assert_eq!(counter.load(Ordering::SeqCst), 11);495+ assert_eq!(lock_audio_deliveries(&deliveries).depth(11), Some(1));496+497+ assert!(cancel_audio_generation(&counter, &deliveries, 11).unwrap());498+ assert_eq!(counter.load(Ordering::SeqCst), 12);499+ assert_eq!(lock_audio_deliveries(&deliveries).generation(), None);500+ assert!(!cancel_audio_generation(&counter, &deliveries, 11).unwrap());501+ }502+503+ 504+ fn audio_reorder_window_is_bounded_and_marks_skipped_packets() {505+ let mut reorder = AudioReorderBuffer::default();506+ let (first, dropped) = reorder.insert(audio_packet(10)).unwrap();507+ assert_eq!(first.len(), 1);508+ assert_eq!(dropped, 0);509+510+ assert!(reorder.insert(audio_packet(12)).unwrap().0.is_empty());511+ assert!(reorder.insert(audio_packet(13)).unwrap().0.is_empty());512+ let (ordered, dropped) = reorder.insert(audio_packet(14)).unwrap();513+ assert_eq!(dropped, 1);514+ assert_eq!(515+ ordered516+ .iter()517+ .map(|packet| packet.packet.header.sequence)518+ .collect::<Vec<_>>(),519+ vec![12, 13, 14]520+ );521+ assert!(ordered[0].discontinuity);522+ assert!(!ordered[1].discontinuity);523+ assert!(reorder.packets.len() <= AudioReorderBuffer::CAPACITY);524+525+ assert!(reorder.insert(audio_packet(16)).unwrap().0.is_empty());526+ let (ordered, dropped) = reorder.insert(audio_packet(15)).unwrap();527+ assert_eq!(dropped, 0);528+ assert_eq!(529+ ordered530+ .iter()531+ .map(|packet| packet.packet.header.sequence)532+ .collect::<Vec<_>>(),533+ vec![15, 16]534+ );535+ assert!(reorder.insert(audio_packet(16)).unwrap().0.is_empty());536+ }537+538+ 539+ fn audio_channel_envelope_is_protocol_strict_and_can_force_discontinuity() {540+ let encoded = encode_audio_channel_packet(9, 42, audio_packet(7), true).unwrap();541+ assert_eq!(&encoded[0..4], b"SGAC");542+ assert_eq!(&encoded[4..6], &1_u16.to_be_bytes());543+ assert_eq!(&encoded[6..8], &24_u16.to_be_bytes());544+ assert_eq!(&encoded[8..16], &9_u64.to_be_bytes());545+ assert_eq!(&encoded[16..24], &42_u64.to_be_bytes());546+ let decoded = AudioPacket::decode_datagram(&encoded[AUDIO_CHANNEL_HEADER_LEN..]).unwrap();547+ assert_eq!(decoded.header.sequence, 7);548+ assert!(decoded.header.flags.contains(AudioFlags::DISCONTINUITY));549+ assert_eq!(decoded.payload, vec![0xf8, 0xff, 0xfe]);550+ }551+}