diff --git a/Cargo.lock b/Cargo.lock index 0806b27..371ac65 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -84,9 +84,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cc" -version = "1.5.1" +version = "1.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f360145194ee8e21db5ee7f3fcd4fe52210864c75c985dae33218202c8bbe040" +checksum = "50a649af8a827553c29fb0cb4bd4a6f1a0dd695bd3232b9bc98bd9c8a3ffbb8b" dependencies = [ "find-msvc-tools", "shlex", @@ -340,9 +340,9 @@ dependencies = [ [[package]] name = "either" -version = "1.18.0" +version = "1.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" +checksum = "0e9c71c2167ca323c882b99918929403426e2373ea17242ff5653e0d5e1058be" [[package]] name = "elliptic-curve" @@ -458,7 +458,7 @@ checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.7", ] [[package]] @@ -755,9 +755,9 @@ dependencies = [ [[package]] name = "libc" -version = "0.2.189" +version = "0.2.190" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" +checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78" [[package]] name = "libcrux-aead" @@ -998,9 +998,9 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" [[package]] name = "mio" -version = "1.2.3" +version = "1.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8" +checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" dependencies = [ "libc", "wasi", @@ -1060,9 +1060,9 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "openmls" -version = "0.8.1" +version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dcb512bfe6a55777518853ea535c6241f069cb0e8984678c117151d2a1e7e903" +checksum = "13c61b7be3e702ecf32a3937f4e99d1189543e37cd78f769700b8bc9dd3cbcf3" dependencies = [ "fluvio-wasm-timer", "getrandom 0.3.4", @@ -1218,9 +1218,9 @@ checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" [[package]] name = "pin-utils" -version = "0.1.0" +version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +checksum = "13bee6c73da26345c729282832b60b0363cf3dd9f4bfd81d8551b7a1c889a113" [[package]] name = "pkcs8" @@ -1594,14 +1594,14 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.7", ] [[package]] name = "serde_json" -version = "1.0.151" +version = "1.0.152" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +checksum = "1741ab7a6cc54a03a89b5d563ed60075c277d9e3cfa73ad0c1f23f23974703c6" dependencies = [ "itoa", "memchr", @@ -1656,9 +1656,9 @@ checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" [[package]] name = "smallvec" -version = "1.16.2" +version = "1.16.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f9395f0f0eee849a9b707b2f06bb92a6a422090e2123bb2ef8e87a0e61892a8e" +checksum = "5b3dc8af474f516a851ff4bd12db780f948b9250ad37211e4eec0bccea54e01b" [[package]] name = "socket2" @@ -1699,9 +1699,9 @@ dependencies = [ [[package]] name = "syn" -version = "3.0.6" +version = "3.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee" +checksum = "d62a2e0561533f2ca2561d0cf27fd9fedb640a1bf2616ff5d5c80d99017faadc" dependencies = [ "proc-macro2", "quote", @@ -1725,7 +1725,7 @@ checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.7", ] [[package]] @@ -1752,9 +1752,9 @@ dependencies = [ [[package]] name = "tokio" -version = "1.53.1" +version = "1.53.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +checksum = "e95f91fcc7a621e8b030f6aa23c71fe9838ae2fb4d8118b75602a328f5144044" dependencies = [ "bytes", "libc", @@ -1773,14 +1773,14 @@ checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" dependencies = [ "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.7", ] [[package]] name = "tokio-rustls" -version = "0.26.5" +version = "0.26.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b0c85f2c3ef0b1cd58b36682f4b17aaa995f0e5db534d85692b4903abce21f67" +checksum = "c9cc2678c2cdd569ef8215e2afd7954ada2ae20b4fdd2c5fe6139a3b02d105db" dependencies = [ "rustls", "tokio", @@ -1881,9 +1881,9 @@ checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" [[package]] name = "uuid" -version = "1.26.1" +version = "1.28.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2ef6dac1e96601b4fb3acccccff2139741fcb757cb9a36089bf5be91cfb285ce" +checksum = "7cc1186384beb7dd8eedea376413fd654937285ea6c9cfbb928dc3043ea4b606" dependencies = [ "getrandom 0.4.3", "js-sys", @@ -1969,7 +1969,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn 3.0.6", + "syn 3.0.7", "wasm-bindgen-shared", ] @@ -2140,18 +2140,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.59" +version = "0.8.62" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6df92bf3d9227be3d53173901ddbffac2babc27ae50f397776ffd6dc33f800cb" +checksum = "86502bf56ac7c77571a32e2647bb2a15894565e981fb2a48d7bde2d91c965a9d" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.59" +version = "0.8.62" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac4f328cf2f05d084e496c3e9c3f33ed0a183656a16e1fcec4d464d8373aec82" +checksum = "5457206954b06561e2608c7e19cf58b1926586d999c246eebe4502f7e2039d1a" dependencies = [ "proc-macro2", "quote", @@ -2160,9 +2160,9 @@ dependencies = [ [[package]] name = "zeroize" -version = "1.9.0" +version = "1.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" +checksum = "e13084392c5e4bc371903e2935a5eaeed24905a7511356b883835e18a78f6879" dependencies = [ "zeroize_derive", ] diff --git a/src/connection/builders.rs b/src/connection/builders.rs index b468aa6..9e1650f 100644 --- a/src/connection/builders.rs +++ b/src/connection/builders.rs @@ -68,6 +68,12 @@ pub(super) fn speaking_message(speaking: bool, ssrc: u32) -> Message { ) } +// only announce starts: Discord infers the stop from RTP, and an explicit stop (mask 0) makes +// the gateway churn begin/end speaking, which some clients render as a click +pub(super) fn should_send_speaking(speaking: bool) -> bool { + speaking +} + /// Op 15 `MEDIA_SINK_WANTS` with `any: 0`. There is no receive path here, so stop the SFU /// forwarding other members' media that we would only drop. pub(super) fn media_sink_wants_message() -> Message { @@ -82,3 +88,14 @@ pub(super) fn close_message() -> Message { reason: Default::default(), })) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn only_a_speaking_start_is_announced() { + assert!(should_send_speaking(true)); + assert!(!should_send_speaking(false), "no explicit op-5 stop"); + } +} diff --git a/src/connection/mod.rs b/src/connection/mod.rs index 0c3f9f4..047f129 100644 --- a/src/connection/mod.rs +++ b/src/connection/mod.rs @@ -405,6 +405,7 @@ impl VoiceConnection { dave_rx, dispatcher.clone(), speaking.clone(), + dave_version > 0, state.clone(), )); diff --git a/src/connection/send_loop.rs b/src/connection/send_loop.rs index 3747b95..f66e5f4 100644 --- a/src/connection/send_loop.rs +++ b/src/connection/send_loop.rs @@ -1,4 +1,4 @@ -use super::builders::speaking_message; +use super::builders::{should_send_speaking, speaking_message}; use super::dave::handle_dave_event; use super::*; @@ -14,16 +14,20 @@ pub(super) async fn send_loop

( mut dave_rx: mpsc::UnboundedReceiver, dispatcher: EventDispatcher, speaking_flag: Arc, + dave_enabled: bool, state: Arc, ) where P: OpusFrameProvider, { let speaking_tx = ws_tx.clone(); let mut pacer = FramePacer::with_transport(provider, udp, dave, transport, ssrc); + pacer.set_dave_enabled(dave_enabled); pacer.on_speaking(move |speaking| { // Keep the shared flag in sync so the gateway task can re-announce it on op 11/12. speaking_flag.store(speaking, Ordering::Relaxed); - let _ = speaking_tx.send(speaking_message(speaking, ssrc)); + if should_send_speaking(speaking) { + let _ = speaking_tx.send(speaking_message(speaking, ssrc)); + } }); // Pending protocol-version transitions, keyed by transition id (set on prepare, applied on diff --git a/src/connection/supervisor.rs b/src/connection/supervisor.rs index 73161ec..adb0c3b 100644 --- a/src/connection/supervisor.rs +++ b/src/connection/supervisor.rs @@ -320,6 +320,9 @@ fn handle_inbound( tracing::info!("voice: gateway session resumed (RESUMED)"); state.store(ConnectionState::Connected as u8, Ordering::SeqCst); } + // Op 4 `SESSION_DESCRIPTION` again after the handshake (region change): re-announce + // so the new session maps our SSRC. + 4 => reannounce.announce_if_speaking(), // Op 11 `CLIENT_CONNECT`: a batch of `user_ids` that joined. The voice gateway does // not carry their audio SSRCs here, so report 0. 11 => { @@ -432,3 +435,74 @@ pub(super) async fn reconnect(url: &str) -> Result<(WsSink, WsSource), ConnectEr }; Ok(ws.split()) } + +#[cfg(test)] +mod tests { + use super::*; + + fn reannounce(speaking: bool) -> (SpeakingReannounce, mpsc::UnboundedReceiver) { + let (ws_tx, ws_rx) = mpsc::unbounded_channel(); + let reannounce = SpeakingReannounce { + ws_tx, + speaking: Arc::new(AtomicBool::new(speaking)), + ssrc: 0xABCD, + }; + (reannounce, ws_rx) + } + + fn session_description() -> Message { + Message::text(json!({ "op": 4, "d": {} }).to_string()) + } + + #[test] + fn speaking_is_reannounced_on_session_description() { + let (reannounce, mut ws_rx) = reannounce(true); + let (dave_tx, _dave_rx) = mpsc::unbounded_channel(); + let state = Arc::new(AtomicU8::new(ConnectionState::Connected as u8)); + let ping = Arc::new(AtomicU64::new(u64::MAX)); + let mut last_seq = 0; + let mut last_ack = Instant::now(); + + handle_inbound( + session_description(), + &dave_tx, + &EventDispatcher::new(), + &reannounce, + &state, + &mut last_seq, + &ping, + None, + &mut last_ack, + ); + + let sent = ws_rx.try_recv().expect("a speaking re-announce"); + let value: Value = serde_json::from_str(sent.into_text().unwrap().as_str()).unwrap(); + assert_eq!(value["op"], 5); + assert_eq!(value["d"]["speaking"], 1); + assert_eq!(value["d"]["ssrc"], 0xABCD); + } + + #[test] + fn nothing_is_reannounced_when_not_speaking() { + let (reannounce, mut ws_rx) = reannounce(false); + let (dave_tx, _dave_rx) = mpsc::unbounded_channel(); + let state = Arc::new(AtomicU8::new(ConnectionState::Connected as u8)); + let ping = Arc::new(AtomicU64::new(u64::MAX)); + let mut last_seq = 0; + let mut last_ack = Instant::now(); + + handle_inbound( + session_description(), + &dave_tx, + &EventDispatcher::new(), + &reannounce, + &state, + &mut last_seq, + &ping, + None, + &mut last_ack, + ); + + assert!(ws_rx.try_recv().is_err()); + } +} diff --git a/src/pacer.rs b/src/pacer.rs index dede453..0641acb 100644 --- a/src/pacer.rs +++ b/src/pacer.rs @@ -94,6 +94,8 @@ pub struct FramePacer { packet: Vec, silence_left: u8, speaking: bool, + // DAVE negotiated for this connection; while it's on and the ratchet isn't ready, hold audio + dave_enabled: bool, on_speaking: Option>, } @@ -122,6 +124,7 @@ impl FramePacer { packet: Vec::with_capacity(MAX_PACKET_BYTES), silence_left: 0, speaking: false, + dave_enabled: false, on_speaking: None, } } @@ -132,6 +135,11 @@ impl FramePacer { self.on_speaking = Some(Box::new(handler)); } + // set when DAVE is negotiated, so frames are held until the key ratchet is ready + pub fn set_dave_enabled(&mut self, enabled: bool) { + self.dave_enabled = enabled; + } + /// Whether the pacer currently considers itself speaking. pub fn is_speaking(&self) -> bool { self.speaking @@ -145,28 +153,37 @@ impl FramePacer { /// Process exactly one frame slot (no internal sleep). Callers that want to control timing /// themselves can drive this; most should use [`run`](Self::run). pub async fn tick(&mut self) -> io::Result { - if let Some(frame) = self.provider.provide() { - self.set_speaking(true); + // no ratchet yet: hold the frame (don't poll) and send silence instead of plaintext audio + if self.dave_enabled && !self.dave.is_ready() { + self.silence_slot().await?; + return Ok(PacerStatus::Silence); + } + + let provided = self.provider.provide(); + // never send a zero-length payload; a stalled provider counts as the end of speech + if let Some(frame) = provided.as_deref().filter(|frame| !frame.is_empty()) { self.silence_left = SILENCE_FRAME_COUNT; // `false` => active-group encryption failed; drop the frame rather than leak plaintext. - if self.build_packet(&frame) { + if self.build_packet(frame) { self.send().await?; + // announce only once the frame is actually on the wire + self.set_speaking(true); } else { // Dropped frames still occupy their 20 ms slot: keep the RTP clock on wall time. self.header.advance_timestamp(SAMPLES_PER_FRAME); } - Ok(PacerStatus::Sent) - } else if self.silence_left > 0 { + return Ok(PacerStatus::Sent); + } + if provided.is_some() { + self.silence_left = SILENCE_FRAME_COUNT; + } + + if self.silence_left > 0 { // Stop speaking as soon as the silence flush begins; the change-detection in // `set_speaking` makes this fire once. self.set_speaking(false); self.silence_left -= 1; - if self.build_packet(&OPUS_SILENCE_FRAME) { - self.send().await?; - } else { - // Dropped frames still occupy their 20 ms slot: keep the RTP clock on wall time. - self.header.advance_timestamp(SAMPLES_PER_FRAME); - } + self.silence_slot().await?; Ok(PacerStatus::Silence) } else { self.set_speaking(false); @@ -174,6 +191,16 @@ impl FramePacer { } } + // send one silence frame; if it can't be built, just advance the clock over this slot + async fn silence_slot(&mut self) -> io::Result<()> { + if self.build_packet(&OPUS_SILENCE_FRAME) { + self.send().await + } else { + self.header.advance_timestamp(SAMPLES_PER_FRAME); + Ok(()) + } + } + /// Self-paced loop: ticks every 20 ms forever. Spawn this as a task and drop/abort it to /// stop. Returns only on a sink I/O error. pub async fn run(&mut self) -> io::Result<()> { diff --git a/tests/pacer.rs b/tests/pacer.rs index 9a68dae..708a22c 100644 --- a/tests/pacer.rs +++ b/tests/pacer.rs @@ -1,11 +1,24 @@ //! Tests the 20 ms pacer: RTP framing, silence draining, and sequence/timestamp progression. +use std::io; +use std::sync::{Arc, Mutex}; + use bytes::Bytes; use voice::rtp::RTP_HEADER_LEN; use voice::{ - DaveEncryptor, FramePacer, PacerStatus, VecSink, SAMPLES_PER_FRAME, SILENCE_FRAME_COUNT, + DaveEncryptor, FramePacer, FrameSink, PacerStatus, VecSink, OPUS_SILENCE_FRAME, + SAMPLES_PER_FRAME, SILENCE_FRAME_COUNT, }; +// a sink that always fails, to exercise the send-error path +struct FailingSink; + +impl FrameSink for FailingSink { + async fn send(&mut self, _packet: &[u8]) -> io::Result<()> { + Err(io::Error::new(io::ErrorKind::Other, "send failed")) + } +} + #[tokio::test] async fn paces_audio_then_silence_then_idle() { let mut remaining = vec![ @@ -58,3 +71,63 @@ async fn paces_audio_then_silence_then_idle() { assert_eq!(&packets[0][RTP_HEADER_LEN..], b"frame-a"); } + +#[tokio::test] +async fn frames_are_kept_until_the_key_ratchet_is_ready() { + let queued = Arc::new(Mutex::new(vec![Bytes::from_static(b"audio")])); + let provider = { + let queued = Arc::clone(&queued); + move || queued.lock().unwrap().pop() + }; + + let sink = VecSink::new(); + // fresh session: DAVE on, but the key ratchet isn't installed yet + let dave = DaveEncryptor::new(1, 2).expect("create DAVE session"); + let mut pacer = FramePacer::new(provider, sink.clone(), dave, 0x1234); + pacer.set_dave_enabled(true); + + assert_eq!(pacer.tick().await.unwrap(), PacerStatus::Silence); + assert_eq!(pacer.tick().await.unwrap(), PacerStatus::Silence); + + // the frame is still queued, and only silence went out (never plaintext audio) + assert_eq!(queued.lock().unwrap().len(), 1); + for packet in sink.packets() { + assert_eq!(&packet[RTP_HEADER_LEN..], &OPUS_SILENCE_FRAME[..]); + } +} + +#[tokio::test] +async fn an_empty_frame_is_not_sent() { + let mut calls = 0; + let provider = move || { + calls += 1; + (calls == 1).then(Bytes::new) + }; + + let sink = VecSink::new(); + let dave = DaveEncryptor::new(1, 2).expect("create DAVE session"); + let mut pacer = FramePacer::new(provider, sink.clone(), dave, 0x1234); + + assert_eq!(pacer.tick().await.unwrap(), PacerStatus::Silence); + let packets = sink.packets(); + assert_eq!(packets.len(), 1); + assert_eq!(&packets[0][RTP_HEADER_LEN..], &OPUS_SILENCE_FRAME[..]); +} + +#[tokio::test] +async fn a_failed_send_does_not_announce_speaking() { + let mut frames = vec![Bytes::from_static(b"audio")]; + let provider = move || frames.pop(); + let dave = DaveEncryptor::new(1, 2).expect("create DAVE session"); + let mut pacer = FramePacer::new(provider, FailingSink, dave, 0x1234); + + let announced = Arc::new(Mutex::new(Vec::new())); + let seen = Arc::clone(&announced); + pacer.on_speaking(move |speaking| seen.lock().unwrap().push(speaking)); + + assert!(pacer.tick().await.is_err()); + assert!( + announced.lock().unwrap().is_empty(), + "a frame that never sent must not announce speaking" + ); +}