Fix speaking and DAVE handling in the voice pacer

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");
+    }
+}

src/connection/mod.rs

@@ -405,6 +405,7 @@ impl VoiceConnection {
             dave_rx,
             dispatcher.clone(),
             speaking.clone(),
+            dave_version > 0,
             state.clone(),
         ));
 

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<P>(
     mut dave_rx: mpsc::UnboundedReceiver<DaveEvent>,
     dispatcher: EventDispatcher,
     speaking_flag: Arc<AtomicBool>,
+    dave_enabled: bool,
     state: Arc<AtomicU8>,
 ) 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

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<Message>) {
+        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());
+    }
+}