add connect timeouts, flush close on disconnect, offload dave keygen

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "voice"
-version = "0.1.2"
+version = "0.1.3"
 edition = "2021"
 description = "Discord voice sending with Gateway v8, UDP/RTP, and DAVE (MLS) end-to-end encryption"
 license = "MIT"

src/connection/mod.rs

@@ -207,7 +207,10 @@ impl VoiceConnection {
     {
         let state = Arc::new(AtomicU8::new(ConnectionState::Connecting as u8));
         let url = info.ws_url();
-        let (ws, _response) = connect_async(url.as_str()).await?;
+        let (ws, _response) = match timeout(HANDSHAKE_TIMEOUT, connect_async(url.as_str())).await {
+            Ok(result) => result?,
+            Err(_) => return Err("voice gateway connect timed out".into()),
+        };
         let (mut sink, mut stream) = ws.split();
 
         sink.send(Message::text(identify_message(&info))).await?;
@@ -363,14 +366,23 @@ impl VoiceConnection {
             dave_protocol_version: dave_version,
         });
         let transport = cipher_for_mode(mode, &secret_key).map_err(ConnectError::from)?;
-        let mut dave = DaveEncryptor::new(info.user_id, info.channel_id)?;
+        // MLS keygen (Ed25519 + HPKE) is CPU-bound; keep it off the async worker.
+        let user_id = info.user_id;
+        let channel_id = info.channel_id;
+        let mut dave =
+            tokio::task::spawn_blocking(move || DaveEncryptor::new(user_id, channel_id)).await??;
 
         let (ws_tx, ws_rx) = mpsc::unbounded_channel::<Message>();
 
         // Queue our MLS key package up front, so the gateway has it before it issues the add
         // proposals that put us in the group.
         if dave_version > 0 {
-            send_key_package(&mut dave, &ws_tx);
+            let ws_tx_kp = ws_tx.clone();
+            dave = tokio::task::spawn_blocking(move || {
+                send_key_package(&mut dave, &ws_tx_kp);
+                dave
+            })
+            .await?;
         }
         // Send-only connection: ask the SFU for no inbound media at all.
         let _ = ws_tx.send(media_sink_wants_message());

src/connection/supervisor.rs

@@ -58,14 +58,37 @@ pub(super) async fn gateway_loop(
                 }
                 let mut recovered = false;
                 if is_resumable(code) {
+                    // Only resume if still Connected; a concurrent `disconnect` sets Closed, and
+                    // resuming would revive a connection the caller asked to drop.
+                    let claimed = state
+                        .compare_exchange(
+                            ConnectionState::Connected as u8,
+                            ConnectionState::Reconnecting as u8,
+                            Ordering::SeqCst,
+                            Ordering::SeqCst,
+                        )
+                        .is_ok();
+                    if !claimed {
+                        // `disconnect` raced in after the drop; flush its queued close and stop.
+                        let _ = sink.send(Message::Close(None)).await;
+                        return;
+                    }
                     tracing::warn!(
                         code = code.unwrap_or(0),
                         "voice: gateway dropped, attempting resume"
                     );
-                    state.store(ConnectionState::Reconnecting as u8, Ordering::SeqCst);
                     // Backoff is jittered because Discord cycling a voice server drops every
                     // connection on the host at once, and a fixed schedule reconnects them in step.
                     for attempt in 0..5u32 {
+                        // Nothing pumps `ws_rx` during this loop, so a `disconnect` queued there is
+                        // invisible; watch the state and flush the close frame ourselves instead of
+                        // leaking the session until Discord times it out.
+                        if ConnectionState::from_u8(state.load(Ordering::SeqCst))
+                            == ConnectionState::Closed
+                        {
+                            let _ = sink.send(Message::Close(None)).await;
+                            return;
+                        }
                         sleep(backoff_delay(attempt)).await;
                         if let Ok((new_sink, new_stream)) = reconnect(&resume.url).await {
                             sink = new_sink;
@@ -403,6 +426,9 @@ pub(super) fn forward_dave_json(op: u64, d: &Value, dave_tx: &mpsc::UnboundedSen
 }
 
 pub(super) async fn reconnect(url: &str) -> Result<(WsSink, WsSource), ConnectError> {
-    let (ws, _response) = connect_async(url).await?;
+    let (ws, _response) = match timeout(HANDSHAKE_TIMEOUT, connect_async(url)).await {
+        Ok(result) => result?,
+        Err(_) => return Err("voice gateway reconnect timed out".into()),
+    };
     Ok(ws.split())
 }