diff --git a/Cargo.lock b/Cargo.lock index 1b78997..0806b27 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1898,7 +1898,7 @@ checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" [[package]] name = "voice" -version = "0.1.2" +version = "0.1.3" dependencies = [ "aes-gcm", "bytes", diff --git a/Cargo.toml b/Cargo.toml index 3546543..3a32988 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/connection/mod.rs b/src/connection/mod.rs index 33caf4f..0c3f9f4 100644 --- a/src/connection/mod.rs +++ b/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::(); // 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()); diff --git a/src/connection/supervisor.rs b/src/connection/supervisor.rs index 8142cdc..73161ec 100644 --- a/src/connection/supervisor.rs +++ b/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()) }